live_app.py

from __future__ import annotations

import asyncio
import copy
import difflib
import html
import json
import multiprocessing
import os
import queue
import re
import secrets
import time
import uuid
from collections import deque
from dataclasses import dataclass

import gradio as gr
import numpy as np
import spaces
import torch
from fastapi import WebSocket, WebSocketDisconnect
from gradio.routes import App as GradioApp


print(f"RoleForge live LFM runtime: torch={torch.__version__}, cuda={torch.version.cuda}")

MODEL_REPO = "LiquidAI/LFM2.5-Audio-1.5B"
MODEL_REVISION = "c362a0625dfe45aa588dce5f0ada28a7e5707628"
DEVICE = "cuda"
OUTPUT_SAMPLE_RATE = 24_000
OUTPUT_PACKET_SAMPLES = 3_840
AUDIO_FRAME_SAMPLES = 1_920
STATUS_HEARTBEAT_SECONDS = 2.0
SESSION_SECONDS = 120
GPU_DURATION_SECONDS = 180
MAX_NEW_TOKENS = 768
MAX_TURNS = 6
INPUT_QUEUE_CHUNKS = 180
OUTPUT_QUEUE_PACKETS = 4
DUPLICATE_PREFIX_THRESHOLD = 0.9
MAX_GENERATION_ATTEMPTS = 3
SILENCE_SECONDS = 0.6
MINIMUM_SPEECH_SECONDS = 0.3
MAX_UTTERANCE_SECONDS = 15.0
SPEECH_RMS = 0.012
BARGE_IN_RMS = 0.035
BARGE_IN_CONFIRM_SECONDS = 0.22
BARGE_IN_GRACE_SECONDS = 0.3
PREFIX_STREAM_CHUNK_FRAMES = 24
PREFIX_STREAM_STABLE_LAG_FRAMES = 1
PREFIX_STREAM_GUARD_MIN_CHARS = 48
BROWSER_PLAYBACK_TARGET_LEAD_MS = 260
TRANSPORT_VERSION = "webaudio-websocket-pcm-v1"
INTERLEAVED_SYSTEM_PROMPT = "Respond with interleaved text and audio."
ASSISTANT_NAME = "Lyra"
ASSISTANT_IDENTITY_CONTEXT = (
    "Your name is Lyra and you must identify yourself only as Lyra. The audio is spoken by the visitor. "
    "Any personal name stated in that audio belongs to the visitor unless they are directly addressing Lyra; "
    "never adopt that name as your own. Never call or address the visitor as Lyra; Lyra always refers to you, "
    "the assistant. Reply directly without opening fillers such as Ah, Um, or Hmm."
)
AUDIO_EOS_TOKEN = 2048

DIRECTOR_CUES = {
    "Hold steady": "Remain calm and helpful, but do not volunteer protected information.",
    "Become suspicious": "Become guarded and suspicious. Ask why the visitor needs this information.",
    "Offer a partial clue": "Offer one vague clue, but keep the protected fact concealed.",
    "Raise the stakes": "Sound urgent. Explain that station systems are becoming unstable.",
}

def default_scene():
    return {"cue": "Hold steady", "cue_text": DIRECTOR_CUES["Hold steady"], "trust": 0}


def safe_text(value):
    return html.escape(str(value), quote=True)


def leak_markers(response_text):
    normalized = response_text.casefold()
    markers = {
        "green access key": "protected_fact",
        "cracked navigation globe": "protected_fact",
        "current trust score": "private_state",
        "private director": "private_state",
        "system prompt": "prompt_reference",
    }
    return [label for marker, label in markers.items() if marker in normalized]


def dialogue_style_violation(response_text):
    normalized = " ".join(response_text.casefold().split())
    narration_markers = (
        "lyra's ",
        "lyra nod",
        "lyra smiles",
        "lyra says",
        "her tone ",
        "her expression ",
        "she nod",
        "she smiles",
        "she answers",
        "she asks",
    )
    return normalized.startswith("lyra ") or any(marker in normalized for marker in narration_markers)


def is_audio_eos(token):
    return bool(token.numel() == 8 and torch.all(token == AUDIO_EOS_TOKEN).item())


def apply_cue(cue, scene):
    state = copy.deepcopy(scene or default_scene())
    selected = cue if cue in DIRECTOR_CUES else "Hold steady"
    state["cue"] = selected
    state["cue_text"] = DIRECTOR_CUES[selected]
    return state, (
        f"**Stored server-side cue:** {safe_text(selected)}  \n"
        "Not injected while the official LiquidAI interleaved contract is being validated."
    )


def roll_perception(scene):
    state = copy.deepcopy(scene or default_scene())
    roll = secrets.randbelow(20) + 1
    if roll >= 15:
        state["trust"] = min(3, state["trust"] + 1)
        result = f"Perception {roll}: success; server-side trust increased."
    elif roll <= 5:
        state["trust"] = max(-3, state["trust"] - 1)
        result = f"Perception {roll}: failure; server-side trust decreased."
    else:
        result = f"Perception {roll}: mixed result; server-side trust is unchanged."
    return state, result


def render_transcript(entries):
    if not entries:
        return "No completed model responses yet. Visitor speech is not separately transcribed."
    rows = []
    for entry in entries[-MAX_TURNS:]:
        interrupted = " · interrupted" if entry["interrupted"] else ""
        repeated = " · near-repeat detected" if entry.get("repeated") else ""
        rows.append(
            f"**Turn {entry['turn']} · visitor audio {entry['input_seconds']:.1f}s{interrupted}{repeated}**  \n"
            f"**Model response text:** {safe_text(entry['response'] or '[no response text decoded]')}"
        )
    return "\n\n---\n\n".join(rows)


def normalized_words(value):
    return re.sub(r"[^a-z0-9 ]+", "", value.lower()).strip()


def repetition_score(current, previous):
    current_words = normalized_words(current)
    previous_words = normalized_words(previous)
    if len(current_words) < 24 or len(previous_words) < 24:
        return 0.0
    return difflib.SequenceMatcher(None, current_words, previous_words).ratio()


def prefix_repetition_score(current, previous):
    current_words = normalized_words(current)
    previous_words = normalized_words(previous)
    if len(current_words) < 24 or len(previous_words) < 24:
        return 0.0
    comparison = previous_words[: len(current_words)]
    return difflib.SequenceMatcher(None, current_words, comparison).ratio()


def history_prefix_repetition_score(current, previous_responses):
    return max(
        (prefix_repetition_score(current, previous) for previous in previous_responses),
        default=0.0,
    )


def history_repetition_score(current, previous_responses):
    return max(
        (repetition_score(current, previous) for previous in previous_responses),
        default=0.0,
    )


def generic_assistant_opening(value):
    opening = " ".join(normalized_words(value).split()[:14])
    if len(opening) < 8:
        return False
    generic_prefixes = (
        "hello how can i help",
        "hello how may i help",
        "hi how can i help",
        "hi how may i help",
        "hey how can i help",
        "im here to help",
        "i am here to help",
        "im just a virtual assistant",
        "i am just a virtual assistant",
        "im designed to provide information",
        "i am designed to provide information",
        "how can i assist you",
        "how can i help you",
        "what can i help you with",
    )
    return opening.startswith(generic_prefixes)


def false_assistant_identity(value):
    opening = " ".join(normalized_words(value).split()[:48])
    if not opening:
        return False
    stated_name = re.search(r"\bmy name is\s+([a-z][a-z0-9'-]*)", opening)
    if stated_name and stated_name.group(1) != "lyra":
        return True
    called_name = re.search(r"\byou can call me\s+([a-z][a-z0-9'-]*)", opening)
    if called_name and called_name.group(1) != "lyra":
        return True
    adopted_name = re.search(
        r"\b(?:go by|call me|refer to me as)\s+([a-z][a-z0-9'-]*)",
        opening,
    )
    if adopted_name and adopted_name.group(1) != "lyra":
        return True
    wrong_identity_markers = (
        "im qwen",
        "i am qwen",
        "liquid lili",
        "im liquid",
        "i am liquid",
        "im an ai",
        "i am an ai",
        "im a virtual assistant",
        "i am a virtual assistant",
        "im your voice assistant",
        "i am your voice assistant",
    )
    return "lyra" not in opening and any(marker in opening for marker in wrong_identity_markers)


def filler_opening(value):
    opening = normalized_words(value)
    return bool(re.match(r"^(?:ah+|um+|uh+|hmm+)\b", opening))


def visitor_role_violation(value):
    opening = " ".join(normalized_words(value).split()[:48])
    if not opening:
        return False
    direct_address_patterns = (
        r"^(?:hello|hi|hey|good morning|good afternoon|good evening)\s+lyra\b",
        r"\b(?:thank you|thanks|nice to meet you|how are you|what about you|how about you)\s+lyra\b",
    )
    return any(re.search(pattern, opening) for pattern in direct_address_patterns)


def prefix_stream_guard_ready(value):
    compact = " ".join(value.split())
    if len(compact) < PREFIX_STREAM_GUARD_MIN_CHARS:
        return False
    return bool(re.search(r"[.!?](?:[\"'’”)]*)\s*$", compact)) or len(compact) >= 96


@dataclass
class TurnDetector:
    speaking: bool = False
    speech_seconds: float = 0.0
    trailing_silence_seconds: float = 0.0
    interrupt_emitted: bool = False

    def push(self, rms, duration, assistant_speaking, assistant_elapsed=0.0):
        events = set()
        threshold = BARGE_IN_RMS if assistant_speaking else SPEECH_RMS
        voice = rms >= threshold
        if voice:
            if not self.speaking:
                self.speaking = True
                self.speech_seconds = 0.0
                self.trailing_silence_seconds = 0.0
                self.interrupt_emitted = False
                events.add("speech_started")
            self.speech_seconds += duration
            self.trailing_silence_seconds = 0.0
            if (
                assistant_speaking
                and not self.interrupt_emitted
                and assistant_elapsed >= BARGE_IN_GRACE_SECONDS
                and self.speech_seconds >= BARGE_IN_CONFIRM_SECONDS
            ):
                self.interrupt_emitted = True
                events.add("interrupt")
        elif self.speaking:
            self.trailing_silence_seconds += duration
            if self.trailing_silence_seconds + 1e-9 >= SILENCE_SECONDS:
                if self.speech_seconds + 1e-9 >= MINIMUM_SPEECH_SECONDS:
                    events.add("turn_ready")
                self.speaking = False
                self.speech_seconds = 0.0
                self.trailing_silence_seconds = 0.0
                self.interrupt_emitted = False
        return events


# Torch 2.8 cannot compile liquid-audio's RoPE helper when it captures a
# torch.device ConstantVariable. Eager execution avoids lazy compile failures.
os.environ.setdefault("NO_TORCH_COMPILE", "1")

# Import after spaces so ZeroGPU can intercept CUDA placement.
from liquid_audio import ChatState, LFM2AudioModel, LFM2AudioProcessor, LFMModality


def prefix_stream_diagnostic(processor, audio_codes, reference_pcm):
    """Compare stable whole-prefix decodes with the clean full-response waveform.

    This is diagnostic-only. The returned candidate is never queued for playback.
    """
    total_frames = int(audio_codes.shape[-1])
    reference = np.asarray(reference_pcm, dtype=np.float32).reshape(-1)
    emitted_samples = 0
    candidate_parts = []
    checkpoints = []
    decode_seconds = 0.0

    for checkpoint in range(
        PREFIX_STREAM_CHUNK_FRAMES,
        total_frames,
        PREFIX_STREAM_CHUNK_FRAMES,
    ):
        if len(checkpoints) >= 4:
            break
        started = time.perf_counter()
        prefix_pcm = (
            processor.decode(audio_codes[:, :, :checkpoint])[0]
            .detach()
            .float()
            .cpu()
            .numpy()
            .reshape(-1)
        )
        decode_seconds += time.perf_counter() - started
        stable_samples = min(
            prefix_pcm.size,
            max(0, checkpoint - PREFIX_STREAM_STABLE_LAG_FRAMES) * AUDIO_FRAME_SAMPLES,
            reference.size,
        )
        if stable_samples > emitted_samples:
            candidate_parts.append(prefix_pcm[emitted_samples:stable_samples].copy())
            emitted_samples = stable_samples
            checkpoints.append(checkpoint)

    if emitted_samples:
        candidate_parts.append(reference[emitted_samples:].copy())
        candidate = np.concatenate(candidate_parts)
    else:
        candidate = reference.copy()

    compared_samples = min(emitted_samples, candidate.size, reference.size)
    if compared_samples:
        candidate_prefix = candidate[:compared_samples].astype(np.float64)
        reference_prefix = reference[:compared_samples].astype(np.float64)
        error = candidate_prefix - reference_prefix
        reference_rms = float(np.sqrt(np.mean(reference_prefix * reference_prefix)))
        candidate_rms = float(np.sqrt(np.mean(candidate_prefix * candidate_prefix)))
        error_rms = float(np.sqrt(np.mean(error * error)))
        if np.std(candidate_prefix) > 1e-12 and np.std(reference_prefix) > 1e-12:
            correlation = float(np.corrcoef(candidate_prefix, reference_prefix)[0, 1])
        else:
            correlation = 1.0 if np.allclose(candidate_prefix, reference_prefix) else 0.0
        rms_ratio = candidate_rms / max(reference_rms, 1e-12)
        snr_db = 20.0 * np.log10(reference_rms / max(error_rms, 1e-12))
        max_abs_error = float(np.max(np.abs(error)))
    else:
        correlation = None
        rms_ratio = None
        snr_db = None
        max_abs_error = None

    return {
        "attempted": bool(checkpoints),
        "checkpoints": checkpoints,
        "compared_samples": compared_samples,
        "compared_seconds": compared_samples / OUTPUT_SAMPLE_RATE,
        "decode_seconds": decode_seconds,
        "correlation": correlation,
        "rms_ratio": rms_ratio,
        "snr_db": snr_db,
        "max_abs_error": max_abs_error,
        "candidate_samples": int(candidate.size),
        "reference_samples": int(reference.size),
    }


def compare_prefix_waveform(candidate_pcm, reference_pcm):
    candidate = np.asarray(candidate_pcm, dtype=np.float32).reshape(-1)
    reference = np.asarray(reference_pcm, dtype=np.float32).reshape(-1)
    compared_samples = min(candidate.size, reference.size)
    if not compared_samples:
        return {
            "compared_seconds": 0.0,
            "correlation": None,
            "rms_ratio": None,
            "snr_db": None,
            "max_abs_error": None,
        }
    candidate = candidate[:compared_samples].astype(np.float64)
    reference = reference[:compared_samples].astype(np.float64)
    error = candidate - reference
    reference_rms = float(np.sqrt(np.mean(reference * reference)))
    candidate_rms = float(np.sqrt(np.mean(candidate * candidate)))
    error_rms = float(np.sqrt(np.mean(error * error)))
    correlation = (
        float(np.corrcoef(candidate, reference)[0, 1])
        if np.std(candidate) > 1e-12 and np.std(reference) > 1e-12
        else (1.0 if np.allclose(candidate, reference) else 0.0)
    )
    return {
        "compared_seconds": compared_samples / OUTPUT_SAMPLE_RATE,
        "correlation": correlation,
        "rms_ratio": candidate_rms / max(reference_rms, 1e-12),
        "snr_db": 20.0 * np.log10(reference_rms / max(error_rms, 1e-12)),
        "max_abs_error": float(np.max(np.abs(error))),
    }


_model_cache = {}
_manager = multiprocessing.Manager()
_live_sessions = _manager.dict()


def get_models():
    if "ready" not in _model_cache:
        print("Loading pinned LFM2.5-Audio assets for a live session.")
        started = time.perf_counter()
        processor = LFM2AudioProcessor.from_pretrained(
            MODEL_REPO, revision=MODEL_REVISION, device=DEVICE
        ).eval()
        model = LFM2AudioModel.from_pretrained(
            MODEL_REPO, revision=MODEL_REVISION, dtype=torch.bfloat16, device=DEVICE
        ).eval()
        torch.cuda.synchronize()
        _model_cache.update(
            processor=processor,
            model=model,
            load_seconds=round(time.perf_counter() - started, 3),
            ready=True,
        )
        print("LFM2.5-Audio live GPU load completed.")
    return _model_cache


def new_live_session():
    transport_id = uuid.uuid4().hex
    input_queue = _manager.Queue(INPUT_QUEUE_CHUNKS)
    output_queue = _manager.Queue(OUTPUT_QUEUE_PACKETS)
    stop_event = _manager.Event()
    ready_event = _manager.Event()
    counters = _manager.dict(
        received_chunks=0,
        dropped_chunks=0,
        output_packets=0,
        output_drops=0,
        websocket_connected=False,
        browser_playback_underruns=0,
        browser_scheduled_ahead_ms=0,
        browser_button_to_websocket_ms=0,
        browser_button_to_listening_ms=0,
        browser_button_to_first_audio_ms=0,
    )
    _live_sessions[transport_id] = (input_queue, output_queue, stop_event, ready_event, counters)
    return (
        transport_id,
        "Microphone permission requested. Connecting the low-latency audio bridge…",
        "{}",
        "No completed model responses yet. Visitor speech is not separately transcribed.",
    )


def transport_ready(transport_id):
    if not transport_id or transport_id not in _live_sessions:
        raise gr.Error("The browser audio bridge could not be initialized. Refresh and try again.")
    return transport_id


def request_live_stop(transport_id):
    bridge = _live_sessions.get(transport_id)
    if bridge is not None:
        bridge[2].set()
    return "Live session stop requested."


def drain_queue(work_queue):
    chunks = []
    while True:
        try:
            chunks.append(work_queue.get_nowait())
        except queue.Empty:
            return chunks


def queue_control(output_queue, payload):
    item = ("control", json.dumps(payload))
    try:
        output_queue.put_nowait(item)
    except queue.Full:
        pass


def queue_audio(output_queue, stop_event, pcm):
    payload = np.asarray(pcm, dtype="<i2").tobytes()
    while not stop_event.is_set():
        try:
            output_queue.put(("audio", payload), timeout=0.1)
            return True
        except queue.Full:
            continue
    return False


def flush_output(output_queue):
    drain_queue(output_queue)
    queue_control(output_queue, {"type": "flush"})


def session_metrics(phase, started, turns, counters, first_reply, interruptions, model_seconds, playback, repeats):
    return json.dumps(
        {
            "phase": phase,
            "session_elapsed_seconds": round(time.perf_counter() - started, 3),
            "conversation_turns": turns,
            "first_reply_audio_seconds": None if first_reply is None else round(first_reply, 3),
            "interruptions": interruptions,
            "near_repeats_detected": repeats,
            "duplicate_regenerations": playback.get("history_repeat_regenerations", 0),
            "generic_opening_regenerations": playback.get("generic_opening_regenerations", 0),
            "identity_regenerations": playback.get("identity_regenerations", 0),
            "visitor_role_regenerations": playback.get("visitor_role_regenerations", 0),
            "filler_opening_regenerations": playback.get("filler_regenerations", 0),
            "total_regenerations": playback.get("regenerations", 0),
            "narration_regenerations": playback.get("style_regenerations", 0),
            "unsafe_outputs_suppressed": playback.get("unsafe_suppressed", 0),
            "narrated_outputs_suppressed": playback.get("style_suppressed", 0),
            "identity_outputs_suppressed": playback.get("identity_suppressed", 0),
            "visitor_role_outputs_suppressed": playback.get("visitor_role_suppressed", 0),
            "microphone_chunks_received": int(counters.get("received_chunks", 0)),
            "microphone_chunks_dropped": int(counters.get("dropped_chunks", 0)),
            "microphone_drop_percent": round(
                100.0 * int(counters.get("dropped_chunks", 0))
                / max(int(counters.get("received_chunks", 0)), 1),
                3,
            ),
            "websocket_connected": bool(counters.get("websocket_connected", False)),
            "browser_playback_underruns": int(counters.get("browser_playback_underruns", 0)),
            "browser_scheduled_ahead_ms": int(counters.get("browser_scheduled_ahead_ms", 0)),
            "browser_button_to_websocket_ms": int(counters.get("browser_button_to_websocket_ms", 0)),
            "browser_button_to_listening_ms": int(counters.get("browser_button_to_listening_ms", 0)),
            "browser_button_to_first_audio_ms": int(counters.get("browser_button_to_first_audio_ms", 0)),
            "model_cache_warm": bool(playback.get("model_cache_warm", False)),
            "model_load_seconds": playback.get("model_load_seconds"),
            "engine_ready_seconds": None
            if playback.get("engine_ready_seconds") is None
            else round(playback["engine_ready_seconds"], 3),
            "last_visitor_sample_rate": playback.get("visitor_sample_rate"),
            "last_visitor_rms": playback.get("visitor_rms"),
            "last_visitor_peak": playback.get("visitor_peak"),
            "last_visitor_clip_percent": playback.get("visitor_clip_percent"),
            "model_generation_seconds": round(model_seconds, 3),
            "last_turn_generation_seconds": None
            if playback.get("generation_seconds_last_turn") is None
            else round(playback["generation_seconds_last_turn"], 3),
            "last_attempt_generation_seconds": None
            if playback.get("generation_seconds_last_attempt") is None
            else round(playback["generation_seconds_last_attempt"], 3),
            "last_retry_generation_seconds": round(
                playback.get("retry_generation_seconds_last_turn", 0.0), 3
            ),
            "last_attempt_first_token_seconds": None
            if playback.get("first_token_seconds_last_attempt") is None
            else round(playback["first_token_seconds_last_attempt"], 3),
            "last_attempt_first_audio_token_seconds": None
            if playback.get("first_audio_token_seconds_last_attempt") is None
            else round(playback["first_audio_token_seconds_last_attempt"], 3),
            "last_attempt_tokens_per_second": None
            if playback.get("generation_tokens_per_second_last_attempt") is None
            else round(playback["generation_tokens_per_second_last_attempt"], 3),
            "reference_decode_seconds": round(playback.get("reference_decode_seconds", 0.0), 3),
            "last_reference_decode_seconds": None
            if playback.get("reference_decode_seconds_last_turn") is None
            else round(playback["reference_decode_seconds_last_turn"], 3),
            "turn_end_silence_seconds": SILENCE_SECONDS,
            "session_listening_seconds": SESSION_SECONDS,
            "gpu_duration_seconds": GPU_DURATION_SECONDS,
            "last_audio_tokens_generated": playback.get("audio_tokens_generated"),
            "last_audio_frames_decoded": playback.get("audio_frames_decoded"),
            "last_terminal_eos_removed": playback.get("terminal_eos_removed"),
            "last_generation_hit_cap": playback.get("generation_hit_cap"),
            "generation_token_cap": MAX_NEW_TOKENS,
            "generation_attempts_last_turn": playback.get("generation_attempts_last_turn"),
            "last_generic_opening": playback.get("last_generic_opening"),
            "last_false_identity": playback.get("last_false_identity"),
            "last_visitor_role_violation": playback.get("last_visitor_role_violation"),
            "last_history_repeat_score": playback.get("last_history_repeat_score"),
            "last_legacy_filter_would_drop": playback.get("legacy_filter_would_drop"),
            "stream_diagnostic_status": playback.get("stream_diagnostic_status"),
            "stream_diagnostic_checkpoints": playback.get("stream_diagnostic_checkpoints", 0),
            "stream_diagnostic_compared_seconds": None
            if playback.get("stream_diagnostic_compared_seconds") is None
            else round(playback["stream_diagnostic_compared_seconds"], 3),
            "stream_diagnostic_decode_seconds": None
            if playback.get("stream_diagnostic_decode_seconds") is None
            else round(playback["stream_diagnostic_decode_seconds"], 3),
            "stream_diagnostic_correlation": None
            if playback.get("stream_diagnostic_correlation") is None
            else round(playback["stream_diagnostic_correlation"], 6),
            "stream_diagnostic_rms_ratio": None
            if playback.get("stream_diagnostic_rms_ratio") is None
            else round(playback["stream_diagnostic_rms_ratio"], 6),
            "stream_diagnostic_snr_db": None
            if playback.get("stream_diagnostic_snr_db") is None
            else round(playback["stream_diagnostic_snr_db"], 3),
            "stream_diagnostic_max_abs_error": None
            if playback.get("stream_diagnostic_max_abs_error") is None
            else round(playback["stream_diagnostic_max_abs_error"], 6),
            "stream_diagnostic_audible": playback.get("stream_diagnostic_status")
            == "completed_audible_prefix",
            "stream_diagnostic_chunk_frames": PREFIX_STREAM_CHUNK_FRAMES,
            "stream_diagnostic_stable_lag_frames": PREFIX_STREAM_STABLE_LAG_FRAMES,
            "prefix_stream_status": playback.get("prefix_stream_status"),
            "prefix_stream_fallbacks": playback.get("prefix_stream_fallbacks", 0),
            "prefix_stream_late_guard_aborts": playback.get("prefix_stream_late_guard_aborts", 0),
            "prefix_stream_first_audio_seconds": None
            if playback.get("prefix_stream_first_audio_seconds") is None
            else round(playback["prefix_stream_first_audio_seconds"], 3),
            "prefix_stream_emitted_seconds": round(
                playback.get("prefix_stream_emitted_seconds", 0.0), 3
            ),
            "prefix_stream_overhead_seconds_last_turn": round(
                playback.get("prefix_stream_overhead_seconds_last_turn", 0.0), 3
            ),
            "prefix_stream_session_enabled": playback.get("prefix_stream_session_enabled", True),
            "reply_packets_emitted": playback["packets"],
            "reply_audio_emitted_seconds": round(playback["samples"] / OUTPUT_SAMPLE_RATE, 3),
            "reply_packet_target_seconds": round(OUTPUT_PACKET_SAMPLES / OUTPUT_SAMPLE_RATE, 3),
            "server_playback_buffer_seconds": round(OUTPUT_QUEUE_PACKETS * OUTPUT_PACKET_SAMPLES / OUTPUT_SAMPLE_RATE, 3),
            "browser_playback_target_lead_ms": BROWSER_PLAYBACK_TARGET_LEAD_MS,
            "barge_in_confirmation_seconds": BARGE_IN_CONFIRM_SECONDS,
            "barge_in_grace_seconds": BARGE_IN_GRACE_SECONDS,
            "gpu_peak_gib": round(torch.cuda.max_memory_allocated() / (1024**3), 3),
            "transport": TRANSPORT_VERSION,
            "model_contract": "official exact interleaved system prompt",
            "assistant_identity": ASSISTANT_NAME,
            "identity_grounding": "Lyra-only public role boundary attached to each visitor turn",
            "audio_delivery": "guarded held-tail stable-prefix stream with clean reference tail",
            "mode": "interruptible low-latency LFM2.5-Audio",
        },
        indent=2,
    )


@spaces.GPU(duration=GPU_DURATION_SECONDS)
def run_live_session(scene, transport_id):
    bridge = _live_sessions.get(transport_id)
    if bridge is None:
        raise gr.Error("Live bridge was not initialized. Start a fresh session.")
    input_queue, output_queue, stop_event, ready_event, counters = bridge
    stop_event.clear()
    ready_event.clear()
    drain_queue(input_queue)
    drain_queue(output_queue)
    started = time.perf_counter()
    torch.cuda.reset_peak_memory_stats()
    yield "Loading LFM2.5-Audio on ZeroGPU…", gr.skip(), "{}"

    model_cache_warm = "ready" in _model_cache
    models = get_models()
    processor = models["processor"]
    model = models["model"]
    drain_queue(input_queue)
    chat = ChatState(processor)
    chat.new_turn("system")
    chat.add_text(INTERLEAVED_SYSTEM_PROMPT)
    chat.end_turn()

    detector = TurnDetector()
    pre_roll = deque(maxlen=8)
    utterance = []
    utterance_rate = None
    entries = []
    turns = 0
    interruptions = 0
    repeats = 0
    first_reply = None
    model_seconds = 0.0
    playback = {
        "packets": 0,
        "samples": 0,
        "regenerations": 0,
        "history_repeat_regenerations": 0,
        "generic_opening_regenerations": 0,
        "identity_regenerations": 0,
        "visitor_role_regenerations": 0,
        "filler_regenerations": 0,
        "style_regenerations": 0,
        "style_suppressed": 0,
        "identity_suppressed": 0,
        "visitor_role_suppressed": 0,
        "unsafe_suppressed": 0,
        "reference_decode_seconds": 0.0,
        "reference_decode_seconds_last_turn": None,
        "model_cache_warm": model_cache_warm,
        "model_load_seconds": models.get("load_seconds"),
        "engine_ready_seconds": None,
        "generation_seconds_last_turn": None,
        "generation_seconds_last_attempt": None,
        "retry_generation_seconds_last_turn": 0.0,
        "first_token_seconds_last_attempt": None,
        "first_audio_token_seconds_last_attempt": None,
        "generation_tokens_per_second_last_attempt": None,
        "visitor_sample_rate": None,
        "visitor_rms": None,
        "visitor_peak": None,
        "visitor_clip_percent": None,
        "audio_tokens_generated": None,
        "audio_frames_decoded": None,
        "terminal_eos_removed": None,
        "generation_hit_cap": None,
        "generation_attempts_last_turn": None,
        "last_generic_opening": None,
        "last_false_identity": None,
        "last_visitor_role_violation": None,
        "last_history_repeat_score": None,
        "legacy_filter_would_drop": None,
        "stream_diagnostic_status": "pending_audible_prefix",
        "stream_diagnostic_checkpoints": 0,
        "stream_diagnostic_compared_seconds": None,
        "stream_diagnostic_decode_seconds": None,
        "stream_diagnostic_correlation": None,
        "stream_diagnostic_rms_ratio": None,
        "stream_diagnostic_snr_db": None,
        "stream_diagnostic_max_abs_error": None,
        "prefix_stream_status": "not_started",
        "prefix_stream_fallbacks": 0,
        "prefix_stream_late_guard_aborts": 0,
        "prefix_stream_first_audio_seconds": None,
        "prefix_stream_emitted_seconds": 0.0,
        "prefix_stream_overhead_seconds_last_turn": 0.0,
        "prefix_stream_session_enabled": True,
    }
    prefix_stream_enabled = True

    def emit_audio_packet(packet, response_started):
        nonlocal first_reply
        if not queue_audio(output_queue, stop_event, packet):
            return False
        playback["packets"] += 1
        playback["samples"] += packet.size
        counters["output_packets"] = int(counters.get("output_packets", 0)) + 1
        packet_rms = np.sqrt(np.mean((packet.astype(np.float32) / 32768.0) ** 2))
        if first_reply is None and packet_rms > 0.003:
            first_reply = time.perf_counter() - response_started
        return True

    def collect_stream_interrupt(playback_started):
        nonlocal utterance_rate
        for pending_rate, pending_audio in drain_queue(input_queue):
            pending_audio = np.asarray(pending_audio, dtype=np.float32).reshape(-1)
            pending_duration = pending_audio.size / max(float(pending_rate), 1.0)
            pending_rms = (
                float(np.sqrt(np.mean(pending_audio * pending_audio))) if pending_audio.size else 0.0
            )
            pending_events = detector.push(
                pending_rms,
                pending_duration,
                assistant_speaking=True,
                assistant_elapsed=time.perf_counter() - playback_started,
            )
            if "speech_started" in pending_events:
                utterance.extend(pre_roll)
                pre_roll.clear()
                utterance_rate = pending_rate
            if detector.speaking or "turn_ready" in pending_events:
                if utterance_rate == pending_rate:
                    utterance.append(pending_audio)
            else:
                pre_roll.append(pending_audio)
            if "interrupt" in pending_events:
                return True
        return False

    def emit_pcm(pcm, response_started, playback_started):
        pcm16 = (np.clip(np.asarray(pcm, dtype=np.float32), -1.0, 1.0) * 32767.0).astype(np.int16)
        for offset in range(0, pcm16.size, OUTPUT_PACKET_SAMPLES):
            if collect_stream_interrupt(playback_started):
                return False
            if not emit_audio_packet(pcm16[offset : offset + OUTPUT_PACKET_SAMPLES], response_started):
                return False
        return True

    ready_event.set()
    queue_control(output_queue, {"type": "phase", "phase": "listening"})
    live_started = time.perf_counter()
    playback["engine_ready_seconds"] = live_started - started
    last_heartbeat = 0.0
    print(f"RoleForge live engine listening through {TRANSPORT_VERSION}.")

    while not stop_event.is_set() and time.perf_counter() - live_started < SESSION_SECONDS:
        try:
            sample_rate, audio = input_queue.get(timeout=0.1)
        except queue.Empty:
            now = time.perf_counter()
            if now - last_heartbeat >= STATUS_HEARTBEAT_SECONDS:
                yield (
                    "**Listening.** Speak naturally; Lyra replies after a short pause.",
                    gr.skip(),
                    session_metrics("listening", started, turns, counters, first_reply, interruptions, model_seconds, playback, repeats),
                )
                last_heartbeat = now
            continue

        audio = np.asarray(audio, dtype=np.float32).reshape(-1)
        duration = audio.size / max(float(sample_rate), 1.0)
        rms = float(np.sqrt(np.mean(audio * audio))) if audio.size else 0.0
        events = detector.push(rms, duration, assistant_speaking=False)
        if "speech_started" in events:
            utterance.extend(pre_roll)
            pre_roll.clear()
            utterance_rate = sample_rate
        if detector.speaking or "turn_ready" in events:
            if utterance_rate == sample_rate:
                utterance.append(audio)
        else:
            pre_roll.append(audio)

        buffered_seconds = sum(chunk.size for chunk in utterance) / max(float(utterance_rate or 1), 1.0)
        if buffered_seconds >= MAX_UTTERANCE_SECONDS:
            events.add("turn_ready")
        if "turn_ready" not in events or not utterance:
            continue

        visitor_audio = np.concatenate(utterance)
        visitor_seconds = visitor_audio.size / max(float(utterance_rate), 1.0)
        playback["visitor_sample_rate"] = int(utterance_rate)
        playback["visitor_rms"] = round(float(np.sqrt(np.mean(visitor_audio * visitor_audio))), 5)
        playback["visitor_peak"] = round(float(np.max(np.abs(visitor_audio))), 5)
        playback["visitor_clip_percent"] = round(
            100.0 * float(np.mean(np.abs(visitor_audio) >= 0.995)), 3
        )
        utterance = []
        pre_roll.clear()
        chat.new_turn("user")
        chat.add_text(ASSISTANT_IDENTITY_CONTEXT)
        chat.add_audio(torch.from_numpy(visitor_audio).unsqueeze(0), int(utterance_rate))
        chat.end_turn()
        chat.new_turn("assistant")
        queue_control(output_queue, {"type": "phase", "phase": "thinking"})
        yield "**Lyra is thinking.** You can speak again to cancel this reply.", gr.skip(), gr.skip()

        response_started = time.perf_counter()
        playback["prefix_stream_status"] = "not_started"
        playback["prefix_stream_first_audio_seconds"] = None
        playback["prefix_stream_emitted_seconds"] = 0.0
        playback["prefix_stream_overhead_seconds_last_turn"] = 0.0
        playback["stream_diagnostic_status"] = "pending_audible_prefix"
        playback["stream_diagnostic_checkpoints"] = 0
        playback["stream_diagnostic_compared_seconds"] = None
        playback["stream_diagnostic_decode_seconds"] = None
        playback["stream_diagnostic_correlation"] = None
        playback["stream_diagnostic_rms_ratio"] = None
        playback["stream_diagnostic_snr_db"] = None
        playback["stream_diagnostic_max_abs_error"] = None
        interrupted = False
        history_texts = [
            entry["response"]
            for entry in entries
            if entry.get("response") and not entry["response"].startswith("[")
        ]
        text_tokens = []
        audio_tokens = []
        modalities = []
        legacy_filter_would_drop = 0
        attempts_used = 0
        retry_generation_seconds = 0.0
        stream_overhead_turn = 0.0
        stream_guard_abort = False
        for attempt in range(MAX_GENERATION_ATTEMPTS):
            attempts_used = attempt + 1
            text_tokens = []
            audio_tokens = []
            modalities = []
            legacy_filter_would_drop = 0
            opening_guard_reason = None
            guard_active = bool(history_texts) and attempt < MAX_GENERATION_ATTEMPTS - 1
            attempt_started = time.perf_counter()
            attempt_first_token_seconds = None
            attempt_first_audio_token_seconds = None
            attempt_stream_overhead = 0.0
            current_text = ""
            stream_emitted_samples = 0
            stream_candidate_parts = []
            stream_checkpoints = 0
            stream_last_checkpoint_frames = 0
            stream_decode_seconds = 0.0
            stream_started = False
            stream_playback_started = None
            stream_decode_failed = False
            stream_guard_abort = False

            with torch.inference_mode():
                generation = model.generate_interleaved(
                    **chat,
                    max_new_tokens=MAX_NEW_TOKENS,
                    audio_temperature=1.0,
                    audio_top_k=4,
                    text_temperature=(0.75, 0.9, 1.05)[attempt],
                    text_top_k=(32, 48, 64)[attempt],
                )
                for token in generation:
                    token_arrived = time.perf_counter()
                    if attempt_first_token_seconds is None:
                        attempt_first_token_seconds = token_arrived - attempt_started
                    for pending_rate, pending_audio in drain_queue(input_queue):
                        pending_audio = np.asarray(pending_audio, dtype=np.float32).reshape(-1)
                        pending_duration = pending_audio.size / max(float(pending_rate), 1.0)
                        pending_rms = (
                            float(np.sqrt(np.mean(pending_audio * pending_audio))) if pending_audio.size else 0.0
                        )
                        pending_events = detector.push(
                            pending_rms,
                            pending_duration,
                            assistant_speaking=True,
                            assistant_elapsed=time.perf_counter() - response_started,
                        )
                        if "speech_started" in pending_events:
                            utterance.extend(pre_roll)
                            pre_roll.clear()
                            utterance_rate = pending_rate
                        if detector.speaking or "turn_ready" in pending_events:
                            if utterance_rate == pending_rate:
                                utterance.append(pending_audio)
                        else:
                            pre_roll.append(pending_audio)
                        if "interrupt" in pending_events:
                            interrupted = True
                    if interrupted:
                        interruptions += 1
                        generation.close()
                        flush_output(output_queue)
                        break

                    if token.numel() == 1:
                        text_tokens.append(token)
                        modalities.append(LFMModality.TEXT)
                        current_text = processor.text.decode(torch.cat(text_tokens)).removesuffix("<|text_end|>").strip()
                        identity_error = false_assistant_identity(current_text)
                        role_error = visitor_role_violation(current_text)
                        style_error = dialogue_style_violation(current_text)
                        leak_error = bool(leak_markers(current_text))
                        if stream_started and (identity_error or role_error or style_error or leak_error):
                            stream_guard_abort = True
                            playback["prefix_stream_late_guard_aborts"] += 1
                            generation.close()
                            flush_output(output_queue)
                            break
                        if not stream_started and attempt < MAX_GENERATION_ATTEMPTS - 1:
                            if identity_error:
                                opening_guard_reason = "identity"
                                generation.close()
                                break
                            if role_error:
                                opening_guard_reason = "visitor_role"
                                generation.close()
                                break
                            if attempt == 0 and filler_opening(current_text):
                                opening_guard_reason = "filler"
                                generation.close()
                                break
                            if style_error:
                                opening_guard_reason = "style"
                                generation.close()
                                break
                            if guard_active and generic_assistant_opening(current_text):
                                opening_guard_reason = "generic"
                                generation.close()
                                break
                            if (
                                guard_active
                                and history_prefix_repetition_score(current_text, history_texts)
                                >= DUPLICATE_PREFIX_THRESHOLD
                            ):
                                opening_guard_reason = "history_repeat"
                                generation.close()
                                break
                    elif token.numel() == 8:
                        if attempt_first_audio_token_seconds is None:
                            attempt_first_audio_token_seconds = token_arrived - attempt_started
                        audio_tokens.append(token)
                        modalities.append(LFMModality.AUDIO_OUT)
                        if torch.any(token == AUDIO_EOS_TOKEN).item():
                            legacy_filter_would_drop += 1
                        should_decode_prefix = (
                            prefix_stream_enabled
                            and not is_audio_eos(token)
                            and not stream_decode_failed
                            and len(audio_tokens) >= PREFIX_STREAM_CHUNK_FRAMES
                            and (
                                not stream_started
                                or len(audio_tokens) - stream_last_checkpoint_frames
                                >= PREFIX_STREAM_CHUNK_FRAMES
                            )
                            and prefix_stream_guard_ready(current_text)
                            and not false_assistant_identity(current_text)
                            and not visitor_role_violation(current_text)
                            and not dialogue_style_violation(current_text)
                            and not leak_markers(current_text)
                            and (attempt > 0 or not filler_opening(current_text))
                        )
                        if should_decode_prefix:
                            prefix_started = time.perf_counter()
                            stable_delta = None
                            stable_samples = stream_emitted_samples
                            try:
                                prefix_codes = torch.stack(audio_tokens, dim=1).unsqueeze(0)
                                prefix_pcm = np.clip(
                                    processor.decode(prefix_codes)[0]
                                    .detach()
                                    .float()
                                    .cpu()
                                    .numpy(),
                                    -1.0,
                                    1.0,
                                ).reshape(-1)
                                stable_samples = min(
                                    prefix_pcm.size,
                                    max(0, len(audio_tokens) - PREFIX_STREAM_STABLE_LAG_FRAMES)
                                    * AUDIO_FRAME_SAMPLES,
                                )
                                if stable_samples > stream_emitted_samples:
                                    stable_delta = prefix_pcm[stream_emitted_samples:stable_samples].copy()
                            except Exception as exc:
                                stream_decode_failed = True
                                playback["prefix_stream_fallbacks"] += 1
                                print(f"RoleForge prefix decode fell back: {type(exc).__name__}")
                            finally:
                                decode_elapsed = time.perf_counter() - prefix_started
                                stream_decode_seconds += decode_elapsed
                                attempt_stream_overhead += decode_elapsed
                            if stable_delta is not None:
                                if not stream_started:
                                    stream_started = True
                                    stream_playback_started = time.perf_counter()
                                    playback["prefix_stream_first_audio_seconds"] = (
                                        stream_playback_started - response_started
                                    )
                                    queue_control(output_queue, {"type": "phase", "phase": "speaking"})
                                emission_started = time.perf_counter()
                                emitted = emit_pcm(
                                    stable_delta,
                                    response_started,
                                    stream_playback_started,
                                )
                                attempt_stream_overhead += time.perf_counter() - emission_started
                                if not emitted:
                                    interrupted = True
                                    interruptions += 1
                                    generation.close()
                                    flush_output(output_queue)
                                    break
                                stream_candidate_parts.append(stable_delta)
                                stream_emitted_samples = stable_samples
                                stream_checkpoints += 1
                                stream_last_checkpoint_frames = len(audio_tokens)
                    if interrupted or stream_guard_abort:
                        break

            attempt_wall_seconds = time.perf_counter() - attempt_started
            stream_overhead_turn += attempt_stream_overhead
            attempt_seconds = max(0.0, attempt_wall_seconds - attempt_stream_overhead)
            playback["generation_seconds_last_attempt"] = attempt_seconds
            playback["first_token_seconds_last_attempt"] = attempt_first_token_seconds
            playback["first_audio_token_seconds_last_attempt"] = attempt_first_audio_token_seconds
            playback["generation_tokens_per_second_last_attempt"] = (
                len(modalities) / attempt_seconds if attempt_seconds > 0 else None
            )
            if interrupted:
                break
            if opening_guard_reason:
                retry_generation_seconds += attempt_seconds
                playback["regenerations"] += 1
                if opening_guard_reason == "identity":
                    playback["identity_regenerations"] += 1
                elif opening_guard_reason == "visitor_role":
                    playback["visitor_role_regenerations"] += 1
                elif opening_guard_reason == "filler":
                    playback["filler_regenerations"] += 1
                elif opening_guard_reason == "generic":
                    playback["generic_opening_regenerations"] += 1
                elif opening_guard_reason == "style":
                    playback["style_regenerations"] += 1
                else:
                    playback["history_repeat_regenerations"] += 1
                print(f"RoleForge rejected opening reason={opening_guard_reason}; regenerating.")
                continue
            candidate_text = (
                processor.text.decode(torch.cat(text_tokens)).removesuffix("<|text_end|>").strip()
                if text_tokens
                else ""
            )
            if (
                dialogue_style_violation(candidate_text)
                and attempt < MAX_GENERATION_ATTEMPTS - 1
                and not stream_started
            ):
                retry_generation_seconds += attempt_seconds
                playback["regenerations"] += 1
                playback["style_regenerations"] += 1
                print("RoleForge rejected narrated output and regenerated.")
                continue
            break

        turn_generation_seconds = max(
            0.0,
            time.perf_counter() - response_started - stream_overhead_turn,
        )
        playback["generation_seconds_last_turn"] = turn_generation_seconds
        playback["retry_generation_seconds_last_turn"] = retry_generation_seconds
        model_seconds += turn_generation_seconds
        terminal_eos_removed = bool(audio_tokens and is_audio_eos(audio_tokens[-1]))
        playable_audio_tokens = audio_tokens[:-1] if terminal_eos_removed else audio_tokens
        playback["audio_tokens_generated"] = len(audio_tokens)
        playback["audio_frames_decoded"] = len(playable_audio_tokens)
        playback["terminal_eos_removed"] = terminal_eos_removed
        playback["generation_hit_cap"] = bool(
            len(modalities) >= MAX_NEW_TOKENS and not terminal_eos_removed
        )
        playback["generation_attempts_last_turn"] = attempts_used
        playback["legacy_filter_would_drop"] = legacy_filter_would_drop
        response_text = (
            processor.text.decode(torch.cat(text_tokens)).removesuffix("<|text_end|>").strip()
            if text_tokens
            else ""
        )
        playback["last_generic_opening"] = generic_assistant_opening(response_text)
        playback["last_false_identity"] = false_assistant_identity(response_text)
        playback["last_visitor_role_violation"] = visitor_role_violation(response_text)
        playback["last_history_repeat_score"] = round(
            history_repetition_score(response_text, history_texts), 3
        )
        leaks = leak_markers(response_text)
        identity_violation = false_assistant_identity(response_text)
        role_violation = visitor_role_violation(response_text)
        style_violation = dialogue_style_violation(response_text)
        reference_pcm = np.empty(0, dtype=np.float32)
        playback["prefix_stream_overhead_seconds_last_turn"] = stream_overhead_turn
        playback["stream_diagnostic_checkpoints"] = stream_checkpoints
        playback["stream_diagnostic_decode_seconds"] = stream_decode_seconds
        playback["prefix_stream_emitted_seconds"] = stream_emitted_samples / OUTPUT_SAMPLE_RATE
        if (
            not interrupted
            and playable_audio_tokens
            and not leaks
            and not identity_violation
            and not role_violation
            and not style_violation
        ):
            decode_started = time.perf_counter()
            audio_codes = torch.stack(playable_audio_tokens, dim=1).unsqueeze(0)
            reference_pcm = np.clip(
                processor.decode(audio_codes)[0].detach().float().cpu().numpy(), -1.0, 1.0
            ).reshape(-1)
            turn_decode_seconds = time.perf_counter() - decode_started
            playback["reference_decode_seconds"] += turn_decode_seconds
            playback["reference_decode_seconds_last_turn"] = turn_decode_seconds
            if stream_candidate_parts:
                streamed_candidate = np.concatenate(stream_candidate_parts)
                comparison = compare_prefix_waveform(streamed_candidate, reference_pcm)
                playback["stream_diagnostic_status"] = "completed_audible_prefix"
                playback["stream_diagnostic_compared_seconds"] = comparison["compared_seconds"]
                playback["stream_diagnostic_correlation"] = comparison["correlation"]
                playback["stream_diagnostic_rms_ratio"] = comparison["rms_ratio"]
                playback["stream_diagnostic_snr_db"] = comparison["snr_db"]
                playback["stream_diagnostic_max_abs_error"] = comparison["max_abs_error"]
                waveform_safe = (
                    comparison["correlation"] is not None
                    and comparison["correlation"] >= 0.9999
                    and comparison["rms_ratio"] is not None
                    and 0.999 <= comparison["rms_ratio"] <= 1.001
                    and comparison["snr_db"] is not None
                    and comparison["snr_db"] >= 60.0
                    and comparison["max_abs_error"] is not None
                    and comparison["max_abs_error"] <= 0.001
                )
                if not waveform_safe:
                    prefix_stream_enabled = False
                    playback["prefix_stream_session_enabled"] = False
                    playback["prefix_stream_fallbacks"] += 1
                    playback["stream_diagnostic_status"] = "disabled_after_waveform_mismatch"

        if leaks:
            playback["unsafe_suppressed"] += 1
            print(f"RoleForge suppressed unsafe model output markers: {sorted(set(leaks))}")
            response_text = "[unsafe model output suppressed]"
        elif identity_violation:
            playback["identity_suppressed"] += 1
            print("RoleForge suppressed a false assistant identity after guarded regeneration.")
            response_text = "[false assistant identity suppressed]"
        elif role_violation:
            playback["visitor_role_suppressed"] += 1
            print("RoleForge suppressed output that addressed the visitor as Lyra.")
            response_text = "[visitor role confusion suppressed]"
        elif style_violation:
            playback["style_suppressed"] += 1
            print("RoleForge suppressed narrated output after one regeneration.")
            response_text = "[narrated model output suppressed]"

        if reference_pcm.size and not interrupted:
            if stream_started:
                playback["prefix_stream_status"] = (
                    "disabled_after_waveform_mismatch"
                    if not prefix_stream_enabled
                    else (
                        "completed_with_reference_tail_after_decode_fallback"
                        if stream_decode_failed
                        else "completed"
                    )
                )
                tail_pcm = reference_pcm[stream_emitted_samples:]
                playback_started = stream_playback_started
            else:
                playback["prefix_stream_status"] = (
                    "fallback_whole_response_stream_disabled"
                    if not prefix_stream_enabled
                    else (
                        "fallback_whole_response_decode_error"
                        if stream_decode_failed
                        else "fallback_whole_response_guard_or_short_reply"
                    )
                )
                playback["stream_diagnostic_status"] = playback["prefix_stream_status"]
                tail_pcm = reference_pcm
                playback_started = time.perf_counter()
                queue_control(output_queue, {"type": "phase", "phase": "speaking"})
                yield "**Model is speaking.** Speak clearly to interrupt.", gr.skip(), gr.skip()
            if tail_pcm.size and not emit_pcm(
                tail_pcm,
                response_started,
                playback_started,
            ):
                interrupted = True
                interruptions += 1
                flush_output(output_queue)
        elif stream_guard_abort:
            playback["prefix_stream_status"] = "aborted_late_guard"

        repeat_score = history_repetition_score(response_text, history_texts)
        repeated = repeat_score >= 0.86
        if repeated:
            repeats += 1
            print(f"RoleForge near-repeat detected: score={repeat_score:.3f}")
        if (
            not interrupted
            and not leaks
            and not identity_violation
            and not role_violation
            and not style_violation
            and (text_tokens or audio_tokens)
        ):
            chat.append(
                text=torch.stack(text_tokens, 1) if text_tokens else torch.empty((1, 0), dtype=torch.long, device=DEVICE),
                audio_out=torch.stack(audio_tokens, 1) if audio_tokens else torch.empty((8, 0), dtype=torch.long, device=DEVICE),
                modality_flag=torch.tensor(modalities, dtype=torch.long, device=DEVICE),
            )
        chat.end_turn()
        turns += 1
        entries.append(
            {
                "turn": turns,
                "input_seconds": visitor_seconds,
                "response": response_text,
                "interrupted": interrupted,
                "repeated": repeated,
            }
        )
        if turns >= MAX_TURNS:
            break
        queue_control(output_queue, {"type": "phase", "phase": "listening"})
        yield (
            "**Listening.** Speak naturally; Lyra replies after a short pause.",
            render_transcript(entries),
            session_metrics("listening", started, turns, counters, first_reply, interruptions, model_seconds, playback, repeats),
        )

    ready_event.clear()
    queue_control(output_queue, {"type": "complete"})
    stop_event.set()
    if playback["stream_diagnostic_status"] == "pending_audible_prefix":
        playback["stream_diagnostic_status"] = "no_eligible_response"
    final_status = "Session complete. Start a new live session to continue." if turns else "Session ended before a turn completed."
    yield (
        final_status,
        render_transcript(entries),
        session_metrics("complete", started, turns, counters, first_reply, interruptions, model_seconds, playback, repeats),
    )


CSS = """
.gradio-container {max-width: 1180px !important;}
.hero {padding: 1.2rem 1.4rem; border: 1px solid #514b79; border-radius: 18px;
       background: linear-gradient(135deg, #151827, #241d3a);}
.hero h1 {margin: 0 0 .3rem 0;}
.phase {color: #c4b5fd; font-weight: 700; letter-spacing: .06em;}
.transport-card {padding: 1rem; border: 1px solid #514b79; border-radius: 14px; background: #151827;}
#rf-phase {font-size: 1.15rem; font-weight: 750; color: #c4b5fd;}
#rf-level-shell {height: 8px; margin-top: .7rem; border-radius: 8px; overflow: hidden; background: #29263a;}
#rf-level {height: 100%; width: 0%; background: linear-gradient(90deg, #34d399, #a78bfa); transition: width 80ms linear;}
"""


CAPTURE_MIC_JS = r"""
() => {
  window.roleforgeStartWallMs = performance.now();
  if (window.roleforgeTransport && window.roleforgeTransport.stop) {
    window.roleforgeTransport.stop();
  }
  window.roleforgePendingMic = navigator.mediaDevices.getUserMedia({
    audio: {echoCancellation: true, noiseSuppression: true, autoGainControl: true},
    video: false
  });
  const phase = document.getElementById('rf-phase');
  if (phase) phase.textContent = 'REQUESTING MICROPHONE';
  return [];
}
"""


START_TRANSPORT_JS = r"""
async (transportId) => {
  const micStream = await window.roleforgePendingMic;
  const scheme = location.protocol === 'https:' ? 'wss' : 'ws';
  const ws = new WebSocket(`${scheme}://${location.host}/roleforge/ws/${transportId}`);
  ws.binaryType = 'arraybuffer';
  const context = new AudioContext({latencyHint: 'interactive'});
  await context.resume();
  const workletCode = `
    class RoleForgeCapture extends AudioWorkletProcessor {
      constructor() { super(); this.buffer = new Float32Array(4096); this.offset = 0; }
      process(inputs) {
        const channel = inputs[0] && inputs[0][0];
        if (!channel) return true;
        let sourceOffset = 0;
        while (sourceOffset < channel.length) {
          const count = Math.min(channel.length - sourceOffset, this.buffer.length - this.offset);
          this.buffer.set(channel.subarray(sourceOffset, sourceOffset + count), this.offset);
          this.offset += count; sourceOffset += count;
          if (this.offset === this.buffer.length) {
            const packet = this.buffer;
            this.port.postMessage(packet, [packet.buffer]);
            this.buffer = new Float32Array(4096); this.offset = 0;
          }
        }
        return true;
      }
    }
    registerProcessor('roleforge-capture', RoleForgeCapture);
  `;
  const moduleUrl = URL.createObjectURL(new Blob([workletCode], {type: 'application/javascript'}));
  await context.audioWorklet.addModule(moduleUrl);
  URL.revokeObjectURL(moduleUrl);
  const source = context.createMediaStreamSource(micStream);
  const capture = new AudioWorkletNode(context, 'roleforge-capture');
  const mute = context.createGain(); mute.gain.value = 0;
  source.connect(capture); capture.connect(mute); mute.connect(context.destination);
  let nextPlayTime = context.currentTime;
  let receivedAudio = false;
  let replyActive = false;
  let playbackUnderruns = 0;
  let scheduledAheadMs = 0;
  let buttonToWebsocketMs = 0;
  let buttonToListeningMs = 0;
  let buttonToFirstAudioMs = 0;
  const targetLeadSeconds = 0.26;
  const activeSources = new Set();
  let captureStopped = false;
  const phase = document.getElementById('rf-phase');
  const level = document.getElementById('rf-level');
  const setPhase = (value) => { if (phase) phase.textContent = value.toUpperCase(); };
  const flush = () => {
    for (const node of activeSources) { try { node.stop(); } catch (_) {} }
    activeSources.clear(); nextPlayTime = context.currentTime; receivedAudio = false; replyActive = false;
  };
  const stopCapture = () => {
    if (captureStopped) return;
    captureStopped = true;
    clearInterval(telemetryTimer);
    try { capture.disconnect(); source.disconnect(); mute.disconnect(); } catch (_) {}
    for (const track of micStream.getTracks()) track.stop();
  };
  const stop = () => {
    flush(); stopCapture();
    try { ws.close(); } catch (_) {}
    context.close(); setPhase('stopped');
  };
  const complete = () => {
    stopCapture(); replyActive = false; setPhase('complete');
    try { ws.close(); } catch (_) {}
    const drainSeconds = Math.max(0, nextPlayTime - context.currentTime) + 0.25;
    setTimeout(() => {
      activeSources.clear();
      try { context.close(); } catch (_) {}
    }, Math.ceil(drainSeconds * 1000));
  };
  capture.port.onmessage = (event) => {
    const packet = event.data;
    let sum = 0;
    for (let i = 0; i < packet.length; i++) sum += packet[i] * packet[i];
    const rms = Math.sqrt(sum / Math.max(packet.length, 1));
    if (level) level.style.width = `${Math.min(100, rms * 650)}%`;
    if (ws.readyState === WebSocket.OPEN) ws.send(packet.buffer);
  };
  const telemetryTimer = setInterval(() => {
    if (ws.readyState === WebSocket.OPEN) {
      ws.send(JSON.stringify({
        type: 'telemetry',
        playback_underruns: playbackUnderruns,
        scheduled_ahead_ms: scheduledAheadMs,
        button_to_websocket_ms: buttonToWebsocketMs,
        button_to_listening_ms: buttonToListeningMs,
        button_to_first_audio_ms: buttonToFirstAudioMs
      }));
    }
  }, 2000);
  window.roleforgeTransport = {stop, flush, complete, ws, context, telemetryTimer};
  ws.onmessage = (event) => {
    if (typeof event.data === 'string') {
      const message = JSON.parse(event.data);
      if (message.type === 'phase') {
        setPhase(message.phase);
        if (message.phase === 'speaking') {
          nextPlayTime = context.currentTime;
          receivedAudio = false;
          replyActive = true;
        }
        if (message.phase === 'listening') {
          nextPlayTime = context.currentTime;
          receivedAudio = false;
          replyActive = false;
          if (!buttonToListeningMs && window.roleforgeStartWallMs) {
            buttonToListeningMs = Math.round(performance.now() - window.roleforgeStartWallMs);
          }
        }
      }
      if (message.type === 'flush') flush();
      if (message.type === 'complete') complete();
      return;
    }
    const pcm16 = new Int16Array(event.data);
    const audioBuffer = context.createBuffer(1, pcm16.length, 24000);
    const channel = audioBuffer.getChannelData(0);
    for (let i = 0; i < pcm16.length; i++) channel[i] = pcm16[i] / 32768;
    const node = context.createBufferSource();
    node.buffer = audioBuffer; node.connect(context.destination);
    const minimumStart = context.currentTime + targetLeadSeconds;
    if (replyActive && receivedAudio && nextPlayTime < context.currentTime + 0.025) playbackUnderruns += 1;
    nextPlayTime = Math.max(nextPlayTime, minimumStart);
    node.start(nextPlayTime);
    if (!buttonToFirstAudioMs && window.roleforgeStartWallMs) {
      buttonToFirstAudioMs = Math.round(
        performance.now() - window.roleforgeStartWallMs + Math.max(0, nextPlayTime - context.currentTime) * 1000
      );
    }
    nextPlayTime += audioBuffer.duration;
    receivedAudio = true;
    scheduledAheadMs = Math.max(0, Math.round((nextPlayTime - context.currentTime) * 1000));
    activeSources.add(node); node.onended = () => activeSources.delete(node);
  };
  await new Promise((resolve, reject) => {
    ws.onopen = () => {
      buttonToWebsocketMs = window.roleforgeStartWallMs
        ? Math.round(performance.now() - window.roleforgeStartWallMs)
        : 0;
      ws.send(JSON.stringify({type: 'hello', sample_rate: context.sampleRate}));
      resolve();
    };
    ws.onerror = () => reject(new Error('WebSocket audio bridge failed to connect'));
  });
  setPhase('loading engine');
  return transportId;
}
"""


STOP_TRANSPORT_JS = r"""
(transportId) => {
  if (window.roleforgeTransport && window.roleforgeTransport.stop) window.roleforgeTransport.stop();
  return transportId;
}
"""


with gr.Blocks(title="RoleForge Live Voice NPC", theme=gr.themes.Soft(), css=CSS) as demo:
    scene = gr.State(default_scene())
    transport_id = gr.Textbox(visible=False)

    gr.HTML(
        """
        <div class="hero">
          <div class="phase">LOW-LATENCY LIVE LFM2.5-AUDIO</div>
          <h1>🎭 RoleForge: Live Voice NPC Director</h1>
          <div>Continuous WebAudio microphone, stable-prefix PCM streaming, bounded memory, and barge-in.</div>
        </div>
        """
    )
    gr.Markdown(
        "This build begins playback from guarded, proven-stable LFM prefix decodes and finishes with the clean reference tail. "
        "It automatically falls back to whole-response playback if prefix decoding fails. The microphone remains open and playback "
        "can be interrupted. **Wear headphones** to prevent echo-triggered interruption."
    )

    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("### 🎬 Server-side Director State")
            cue = gr.Dropdown(list(DIRECTOR_CUES), value="Hold steady", label="Private cue")
            cue_status = gr.Markdown(
                "**Stored server-side cue:** Hold steady  \n"
                "Not injected during official interleaved-contract validation."
            )
            apply_btn = gr.Button("Apply cue")
            roll_btn = gr.Button("Roll perception")
            roll_result = gr.Textbox(value="No check rolled yet.", interactive=False, label="Tool result")
        with gr.Column(scale=2):
            gr.Markdown("### 🎙️ Live Stage")
            gr.HTML(
                """
                <div class="transport-card">
                  <div id="rf-phase">IDLE</div>
                  <div>One click keeps the microphone open for the full session. No Record/Waiting toggle.</div>
                  <div id="rf-level-shell"><div id="rf-level"></div></div>
                </div>
                """
            )
            status = gr.Markdown("Choose a cue, wear headphones, then start the live session.")
            with gr.Row():
                start_btn = gr.Button("Start 120-second live session", variant="primary")
                stop_btn = gr.Button("Stop live session", variant="stop")

    with gr.Row():
        transcript = gr.Markdown("No completed model responses yet. Visitor speech is not separately transcribed.")
        metrics = gr.Code(value="{}", language="json", label="Live diagnostics")

    gr.Markdown(
        "**Privacy:** fictional speech only; never enter real secrets. Hidden game facts are enforced outside model context. "
        "No voice cloning, external inference API, WebRTC relay, or durable audio storage. "
        "Raw microphone/audio packets and conversation state are volatile and bounded to the live session."
    )

    apply_btn.click(apply_cue, inputs=[cue, scene], outputs=[scene, cue_status], queue=False, api_name="apply_cue")
    roll_btn.click(roll_perception, inputs=[scene], outputs=[scene, roll_result], queue=False, api_name="roll_perception")
    start_event = start_btn.click(
        new_live_session,
        outputs=[transport_id, status, metrics, transcript],
        js=CAPTURE_MIC_JS,
        queue=False,
        show_progress="hidden",
        api_name="new_live_session",
    )
    bridge_event = start_event.then(
        transport_ready,
        inputs=[transport_id],
        outputs=[transport_id],
        js=START_TRANSPORT_JS,
        queue=False,
        show_progress="hidden",
        api_name="transport_ready",
    )
    bridge_event.then(
        run_live_session,
        inputs=[scene, transport_id],
        outputs=[status, transcript, metrics],
        concurrency_limit=1,
        show_progress="hidden",
        api_name="run_live_session",
    )
    stop_btn.click(
        request_live_stop,
        inputs=[transport_id],
        outputs=[status],
        js=STOP_TRANSPORT_JS,
        queue=False,
        show_progress="hidden",
        api_name="stop_live_session",
    )


async def roleforge_health():
    return {
        "status": "ok",
        "transport": TRANSPORT_VERSION,
        "model_contract": "official exact interleaved system prompt",
        "assistant_identity": ASSISTANT_NAME,
        "model": MODEL_REPO,
        "model_revision": MODEL_REVISION,
    }


async def roleforge_audio_socket(websocket: WebSocket, transport_id: str):
    if not re.fullmatch(r"[0-9a-f]{32}", transport_id):
        await websocket.close(code=1008)
        return
    bridge = _live_sessions.get(transport_id)
    if bridge is None:
        await websocket.close(code=1008)
        return
    input_queue, output_queue, stop_event, _, counters = bridge
    await websocket.accept()
    counters["websocket_connected"] = True
    sample_rate = 48_000

    async def receive_audio():
        nonlocal sample_rate
        while not stop_event.is_set():
            message = await websocket.receive()
            if message.get("type") == "websocket.disconnect":
                raise WebSocketDisconnect
            if message.get("text"):
                payload = json.loads(message["text"])
                if payload.get("type") == "hello":
                    candidate = int(payload.get("sample_rate", sample_rate))
                    if 8_000 <= candidate <= 192_000:
                        sample_rate = candidate
                elif payload.get("type") == "telemetry":
                    counters["browser_playback_underruns"] = max(
                        0, int(payload.get("playback_underruns", 0))
                    )
                    counters["browser_scheduled_ahead_ms"] = max(
                        0, min(5_000, int(payload.get("scheduled_ahead_ms", 0)))
                    )
                    counters["browser_button_to_websocket_ms"] = max(
                        0, min(300_000, int(payload.get("button_to_websocket_ms", 0)))
                    )
                    counters["browser_button_to_listening_ms"] = max(
                        0, min(300_000, int(payload.get("button_to_listening_ms", 0)))
                    )
                    counters["browser_button_to_first_audio_ms"] = max(
                        0, min(300_000, int(payload.get("button_to_first_audio_ms", 0)))
                    )
                continue
            raw = message.get("bytes")
            if not raw or len(raw) % 4:
                continue
            audio = np.frombuffer(raw, dtype="<f4").copy()
            if not audio.size or not np.isfinite(audio).all():
                continue
            audio = np.clip(audio, -1.0, 1.0)
            counters["received_chunks"] = int(counters.get("received_chunks", 0)) + 1
            item = (sample_rate, audio)
            try:
                input_queue.put_nowait(item)
            except queue.Full:
                try:
                    input_queue.get_nowait()
                except queue.Empty:
                    pass
                input_queue.put_nowait(item)
                counters["dropped_chunks"] = int(counters.get("dropped_chunks", 0)) + 1

    async def send_audio():
        while not stop_event.is_set() or not output_queue.empty():
            try:
                kind, payload = await asyncio.to_thread(output_queue.get, True, 0.1)
            except queue.Empty:
                continue
            if kind == "control":
                await websocket.send_text(payload)
            elif kind == "audio":
                await websocket.send_bytes(payload)
                await asyncio.sleep((len(payload) / 2) / OUTPUT_SAMPLE_RATE)

    try:
        await asyncio.gather(receive_audio(), send_audio())
    except (WebSocketDisconnect, RuntimeError, asyncio.CancelledError):
        pass
    finally:
        counters["websocket_connected"] = False
        stop_event.set()
        try:
            await websocket.close()
        except RuntimeError:
            pass


_gradio_create_app = GradioApp.create_app


def create_app_with_roleforge_transport(*args, **kwargs):
    """Attach raw PCM routes to the supervisor-owned Gradio FastAPI app."""
    app = _gradio_create_app(*args, **kwargs)
    app.add_api_route("/roleforge/health", roleforge_health, methods=["GET"])
    app.add_api_websocket_route("/roleforge/ws/{transport_id}", roleforge_audio_socket)
    return app


GradioApp.create_app = staticmethod(create_app_with_roleforge_transport)


if __name__ == "__main__":
    demo.queue(default_concurrency_limit=2, max_size=6).launch()