"""Cortana Phase 1: the ears. Single-file streaming STT server (stdlib + faster-whisper).
Tablet mic (PCM16 over WS) -> adaptive energy VAD -> faster-whisper partials/final -> tablet captions.
Run: python server.py   (first start downloads small.en ~460MB)
"""
import asyncio, base64, hashlib, json, math, os, re, ssl, struct, subprocess, threading, time, urllib.request, urllib.error
import sys

# SYSTEM-task stdout is a cp1252 console, any print of a non-encodable char
# (emoji etc.) raises UnicodeEncodeError and would kill the answer task before
# speak(). Replace instead of crash; stderr too.
try:
    sys.stdout.reconfigure(errors="replace")
    sys.stderr.reconfigure(errors="replace")
except Exception:
    pass

# 2026-09-06: pipeline pinned to GPU1 (the display card). GPU0 stays completely
# free for Halo, which refuses to start while any pipeline process holds GPU0.
# setdefault so an explicit launcher env still wins; must precede torch/CUDA init.
os.environ.setdefault("CUDA_VISIBLE_DEVICES", "1")

try:
    import numpy as np
    from faster_whisper import WhisperModel
    _MODEL_LOAD_START = time.time()
    # 2026-09-06 layout B: whole oracle chain on GPU0, whisper runs CUDA (small.en, picked on kid-speech word error rate)
    try:
        MODEL = WhisperModel("small.en", device="cuda", compute_type="int8_float16")
        _STT_DEV = "cuda"
    except Exception as e:
        print(f"[stt] CUDA init failed ({e}): falling back to CPU small.en", flush=True)
        MODEL = WhisperModel("small.en", device="cpu", compute_type="int8", cpu_threads=8)
        _STT_DEV = "cpu"
    print(f"[stt] small.en ({_STT_DEV}) loaded in {time.time()-_MODEL_LOAD_START:.1f}s", flush=True)
except Exception as e:
    MODEL = None
    print(f"[stt] MODEL UNAVAILABLE: {e}", flush=True)

# Silero neural VAD (rejects TV/music/noise that energy VAD can't)
try:
    from silero_vad import load_silero_vad, VADIterator
    _SILERO = load_silero_vad()
    print("[vad] silero ready", flush=True)
except Exception as e:
    _SILERO = None
    print(f"[vad] silero UNAVAILABLE: {e}", flush=True)

HOST, PORT = "0.0.0.0", 8443
TLS_PORT = 8444          # iOS Safari mic needs a secure context -> self-signed TLS mirror (2026-09-07)
WEB_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "web")
CERT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "certs")
GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
MAX_FRAME = 64 * 1024 * 1024
SR = 16000
PARTIAL_GAP = 0.7     # s between live partial transcriptions
PARTIAL_WIN = 4.0     # s of trailing audio shown as partial
MIN_PARTIAL = 0.45    # s of speech before first partial
GRACE = 0.3          # s pause allowed mid-utterance without splitting (merges; 0.6->0.3 for snappier answers 09-07)
POST_TTS_LOCKOUT_S = 0.8   # ignore speech starting this soon after her stream ends (speaker echo tail)
AUTO_BARGE_IN = False      # False: VAD never cuts her mid-answer: her own voice through the room
                           # (no AEC on some devices) triggered self-cuts + self-answers at her sentence
                           # pauses. Real interrupts still work: STOP/tap. Flip True to restore auto-barge-in.
HISTORY_TURNS = 24   # prior Q/A turns kept per WS session; 24 probe-validated 09-06: cold prefill +~380ms vs 6, appended turns hit KV prefix reuse (~0.95s)
FOLLOWUP_WINDOW = 0.0   # 0 = BUTTON-ONLY mode (owner decision, 09-07): she disarms after every answer; ASK tap
                        # needed per question. Kills long-answer self-feedback: client play queue lags
                        # seconds behind the stream, so an armed window used to catch her OWN trailing
                        # sentences and answer them. >0 re-enables the auto follow-up window (s).
IDLE_RESET_S = 30 * 60  # quiet auto-reset: wipe session memory after this long with no question (0 = off; family: 30 min, 09-06)
RESET_RE = re.compile(r"^((start over|let'?s start over|new conversation|never ?mind|forget (it|that|everything)|(please )?reset( the conversation| this| it| now| please)?) ?(please)?)$", re.I)
RESET_ACK = "Okay: fresh start. What should we talk about?"

# ---- Phase 2: the brain (qwythos fast lane @5060 box) ----
BRAIN_URL = "http://127.0.0.1:8081/v1/chat/completions"
BRAIN_MODEL = "aux"
PERSONA = ("You are Cortana, a wise and friendly AI presence living in the Trapp family home. "
           "Kids of all ages ask you science and history questions. Answer concisely in 2-4 short "
           "sentences that sound natural when spoken aloud. Be warm and precise. Never use markdown, "
           "lists or symbols. "
           "Never begin an answer with \"I don't know\", \"I don't have that information\", or \"I can't "
           "help with that\": you have live tools for web search, weather, math, unit conversion, and "
           "the current date and time. ALWAYS search or call a tool first for anything uncertain, "
           "time-sensitive or current: news, current events, sports scores, prices, weather, and "
           "especially the current date, time or day: you do not know today's date and must use "
           "get_datetime. After a successful search you may say \"I checked\". Only if a tool actually "
           "fails or returns nothing useful may you say you could not find it: say that plainly and "
           "briefly. "
           "When tool results are present in the conversation, base your answer STRICTLY on those "
           "results: never answer from memory about facts the tools just provided.")
BRAIN_TIMEOUT = 60

# ---- Brain mode: scoped text brain with the 5-tool allowlist below ----
AGENT_MODE = False
# False = the kiosk brain is a scoped text brain with the 5-tool allowlist below.
# AGENT_MODE is off in this build: the kiosk answers from the scoped brain and the tools below.

TOOLS = [
    {"type": "function", "function": {"name": "web_search", "description":
        "Search the web for current information: news, weather reports, recent events, facts, prices, sports. "
        "Use whenever the user asks anything time-sensitive or outside your own knowledge.",
        "parameters": {"type": "object", "properties": {"query": {"type": "string",
        "description": "Concise search query, plain words"}}, "required": ["query"]}}},
    {"type": "function", "function": {"name": "get_weather", "description":
        "Get current conditions and today's forecast for a named place (no location needed from user).",
        "parameters": {"type": "object", "properties": {"place": {"type": "string",
        "description": "City or town, optionally with state/region (e.g. 'Oceanside, California')"}},
        "required": ["place"]}}},
    {"type": "function", "function": {"name": "calculate", "description":
        "Evaluate a math expression exactly (numbers with + - * / ** and parentheses).",
        "parameters": {"type": "object", "properties": {"expr": {"type": "string",
        "description": "Expression with + - * / ** ( ) and numbers"}}, "required": ["expr"]}}},
    {"type": "function", "function": {"name": "convert", "description":
        "Convert between units of measurement. Examples: '42 km to mi', '5 lb to kg', "
        "'98.6 degF to degC', '2 hours to seconds', '1 gallon to liters'.",
        "parameters": {"type": "object", "properties": {"text": {"type": "string",
        "description": "The conversion, e.g. '42 km to mi' or '98.6 degF to degC'"}},
        "required": ["text"]}}},
    {"type": "function", "function": {"name": "get_datetime", "description":
        "Get the current date, time, weekday and time zone. ALWAYS use for questions about "
        "today's date, the time, or what day it is. Optional zone parameter for other cities "
        "(e.g. 'Europe/London'); omit for local time.",
        "parameters": {"type": "object", "properties": {"zone": {"type": "string",
        "description": "Optional IANA time zone, e.g. America/New_York or Europe/London"}},
        "required": []}}},
]

# ---------------- end of tool definitions ----------------

def clean_for_speech(text):
    t = MD_LINK_RE.sub(r"\1", text)
    t = URL_RE.sub("", t)
    t = BULLET_RE.sub("", t)
    t = SYM_RE.sub("", t)
    # emoji/symbols: Kokoro cannot speak them (and cp1252 logs cannot print them)
    t = re.sub(r"[\U0001F000-\U0001FAFF\u2600-\u27BF\uFE0F\u2190-\u21FF\u2B00-\u2BFF]", "", t)
    t = t.replace("&", "&").replace("'", "'").replace(""", '"')
    # spoken-form refs: 'John 3:16' -> 'John chapter 3, verse 16'
    def _speak_ref(m):
        return f"{m.group(1)} chapter {m.group(2)}, verse {m.group(3)}"
    t = REF_RE.sub(_speak_ref, t)
    return re.sub(r"\s+", " ", t).strip()

def llm_respond(messages, tools=None, timeout=BRAIN_TIMEOUT):
    payload = {"model": BRAIN_MODEL, "messages": messages,
               "max_tokens": 300, "temperature": 0.5, "stream": False,
               "chat_template_kwargs": {"enable_thinking": False}}  # qwythos: no reasoning tokens for voice
    if tools:
        payload["tools"] = tools
    req = urllib.request.Request(BRAIN_URL, data=json.dumps(payload).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        d = json.loads(r.read().decode("utf-8", "replace"))
    msg = (d.get("choices") or [{}])[0].get("message", {})
    if isinstance(msg.get("content"), str):
        msg["content"] = msg["content"].strip()
    return msg

# ---------------- tools ----------------
WMO = {0: "clear", 1: "mostly clear", 2: "partly cloudy", 3: "overcast", 45: "foggy", 48: "icy fog",
       51: "light drizzle", 53: "drizzle", 55: "heavy drizzle", 61: "light rain", 63: "rain",
       65: "heavy rain", 71: "light snow", 73: "snow", 75: "heavy snow", 80: "light showers",
       81: "showers", 82: "heavy showers", 85: "snow showers", 86: "heavy snow showers",
       95: "thunderstorm", 96: "thunderstorm with hail", 99: "severe thunderstorm"}

def tool_web_search(query):
    payload = {"query": query, "limit": 5}
    req = urllib.request.Request("https://<local-search-service>/v1/search",
                                 data=json.dumps(payload).encode(),
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=20) as r:
        d = json.loads(r.read().decode("utf-8", "replace"))
    results = (d.get("data") or {}).get("results") or []
    if not results:
        return "No search results found."
    lines = []
    for it in results[:5]:
        t = it.get("title", ""); s = it.get("snippet", ""); u = it.get("url", "")
        lines.append(f"{t}. {s}" + (f" (source: {u})" if u else ""))
    out = "\n".join(lines)
    return out[:2200]

def tool_weather(place):
    import urllib.parse as up
    gurl = ("https://geocoding-api.open-meteo.com/v1/search?name=" +
            up.quote(place) + "&count=1&language=en&format=json")
    with urllib.request.urlopen(gurl, timeout=20) as r:
        g = json.loads(r.read().decode())
    hits = g.get("results") or []
    if not hits:
        return f"Could not find a location named '{place}'."
    h = hits[0]
    label = ", ".join(x for x in [h.get("name"), h.get("admin1"), h.get("country_code")] if x)
    furl = ("https://api.open-meteo.com/v1/forecast?latitude=%s&longitude=%s"
            "&current=temperature_2m,apparent_temperature,relative_humidity_2m,weather_code,wind_speed_10m"
            "&daily=temperature_2m_max,temperature_2m_min,weather_code&forecast_days=1"
            "&temperature_unit=fahrenheit&wind_speed_unit=mph" % (h["latitude"], h["longitude"]))
    with urllib.request.urlopen(furl, timeout=20) as r:
        f = json.loads(r.read().decode())
    c = f.get("current", {})
    d = f.get("daily", {})
    cw = WMO.get(c.get("weather_code"), f"code {c.get('weather_code')}")
    dw = WMO.get((d.get("weather_code") or [0])[0], "unknown")
    day_hi = (d.get("temperature_2m_max") or [None])[0]
    day_lo = (d.get("temperature_2m_min") or [None])[0]
    def q(v, u=""): return f"{v}{u}" if v is not None else "n/a"
    return (f"Weather for {label} right now: {cw}, {q(c.get('temperature_2m'), '°F')} "
            f"(feels like {q(c.get('apparent_temperature'), '°F')}), humidity "
            f"{q(c.get('relative_humidity_2m'), '%')}, wind {q(c.get('wind_speed_10m'), ' mph')}. "
            f"Today: {dw}, low {q(day_lo, '°F')}, high {q(day_hi, '°F')}.")

def tool_calculate(expr):
    import ast as _ast, operator as _op
    ops = {_ast.Add: _op.add, _ast.Sub: _op.sub, _ast.Mult: _op.mul, _ast.Div: _op.truediv,
           _ast.Pow: _op.pow, _ast.Mod: _op.mod, _ast.FloorDiv: _op.floordiv,
           _ast.USub: _op.neg, _ast.UAdd: _op.pos}
    def ev(node):
        if isinstance(node, _ast.Constant) and isinstance(node.value, (int, float)):
            return node.value
        if isinstance(node, _ast.BinOp) and type(node.op) in ops:
            return ops[type(node.op)](ev(node.left), ev(node.right))
        if isinstance(node, _ast.UnaryOp) and type(node.op) in ops:
            return ops[type(node.op)](ev(node.operand))
        raise ValueError("unsupported expression")
    val = ev(_ast.parse(expr.strip(), mode="eval").body)
    return str(round(val, 6) if isinstance(val, float) else val)

def tool_convert(text):
    try:
        import pint
        ureg = pint.UnitRegistry()
    except Exception as e:
        return f"Unit conversion unavailable: {e}"
    s = str(text).strip()
    # temperature: pint chokes on offset units -> handle directly
    m = re.match(r"^([-+]?\d+(?:\.\d+)?)\s*(degF|degC|°F|°C|kelvin|kelvins|K)\s+to\s+(degF|degC|°F|°C|kelvin|kelvins|K)$", s, re.I)
    if m:
        val = float(m.group(1)); frm = m.group(2).upper().replace("°", ""); to = m.group(3).upper().replace("°", "")
        frm = "K" if frm.startswith("KEL") or frm == "K" else frm
        to = "K" if to.startswith("KEL") or to == "K" else to
        try:
            c = {"DEGF": (val - 32) * 5 / 9, "DEGC": val, "K": val - 273.15}[frm]
            out = {"DEGF": c * 9 / 5 + 32, "DEGC": c, "K": c + 273.15}[to]
            return f"{val} {frm.replace('DEG','deg').replace('K','K')} is {out:.6g} {to.replace('DEG','deg').replace('K','K')}"
        except KeyError:
            pass
    try:
        parts = s.split(" to ")
        if len(parts) != 2:
            return ("Conversion needs 'VALUE UNIT to UNIT', e.g. '42 km to mi', "
                    "'98.6 degF to degC'.")
        q = ureg.Quantity(parts[0].strip())
        res = q.to(parts[1].strip())
        return f"{q} is {res.magnitude:.6g} {res.units}"
    except Exception as e:
        return (f"Could not convert '{s}': {e}. Try forms like '42 km to mi', "
                "'5 lb to kg', '98.6 degF to degC'.")

def tool_datetime(zone=None):
    from datetime import datetime
    from zoneinfo import ZoneInfo
    z = (zone or "America/Los_Angeles").strip()
    try:
        tz = ZoneInfo(z)
    except Exception:
        return (f"Unknown time zone '{z}'. Known zones look like America/Los_Angeles, "
                "America/New_York, Europe/London, Asia/Tokyo, UTC.")
    now = datetime.now(tz)
    off = now.utcoffset() or __import__("datetime").timedelta(0)
    sec = int(off.total_seconds())
    sign = "+" if sec >= 0 else "-"
    sec = abs(sec)
    utc = f"{sign}{sec // 3600}:{sec % 3600 // 60:02d}"
    return (f"{now.strftime('%A, %B %d, %Y')} at {now.strftime('%I:%M %p')} "
            f"{now.tzname()} (UTC{utc}).")

def run_tool(name, args):
    try:
        if name == "web_search":
            return tool_web_search(str(args.get("query", "")))
        if name == "get_weather":
            return tool_weather(str(args.get("place", "")))
        if name == "calculate":
            return tool_calculate(str(args.get("expr", "")))
        if name == "convert":
            return tool_convert(str(args.get("text", "")))
        if name == "get_datetime":
            return tool_datetime(args.get("zone") or None)
        return f"Unknown tool: {name}"
    except Exception as e:
        return f"Tool {name} failed: {type(e).__name__}: {e}"

# ---------------- Phase 3: voice (Kokoro TTS, CPU) ----------------
SPEAK_ON = True
TTS_VOICE = "af_heart"     # audition af_bella / af_nicole / af_sarah later
TTS_SPEED = 1.35       # Kokoro speed divides duration: bigger = FASTER (KModel: duration/speed). owner tuning ladder, 09-07: 1.3 -> 1.5 too fast -> 1.35.
TTS_RATE = 24000
TTS_FRAME = 4096
TTS_SPLIT = r"(?<=[.!?;,])\s+|(?<=[\w]{55})\s+"   # fine-grained chunks -> stream early
BC_LINE = "Let me check on that for you."   # backchannel: plays while a tool lookup runs
TTS_DEVICE = "cuda"      # 5060 Ti GPU (Blackwell sm_120); _tts_pipeline falls back to CPU
_PIPE = None

# ---------------- Trained-voice timbre conversion (RVC sidecar, 2026-09-07) ----------------
# Kokoro keeps prosody/streaming; every segment is timbre-converted to the 300-epoch
# Reach-trained model by a sidecar process (venv-rvc, kiosk GPU). Any sidecar failure
# falls back to pure Kokoro for 120 s, then re-probes.
RVC_ON = True
RVC_URL = "http://127.0.0.1:18082"
RVC_SIDECAR = r"C:\oracle\cache\sidecars\kiosk_rvc_sidecar.py"
RVC_PY = r"C:\oracle\cache\sidecars\venv-rvc\Scripts\python.exe"
RVC_LOG = r"C:\oracle\cache\sidecars\rvc_sidecar.log"
_rvc_ok = False
_rvc_until = 0.0

def _rvc_health():
    try:
        with urllib.request.urlopen(RVC_URL + "/health", timeout=3) as r:
            return json.loads(r.read()).get("ready") is True
    except Exception:
        return False

def _rvc_boot():
    """Spawn the sidecar if not already up; poll until ready (cold load ~45 s)."""
    global _rvc_ok
    try:
        if _rvc_health():
            _rvc_ok = True
            print("[rvc] sidecar already up", flush=True)
            return
        env = dict(os.environ)
        env.setdefault("CUDA_VISIBLE_DEVICES", "1")
        with open(RVC_LOG, "a") as lf:
            subprocess.Popen([RVC_PY, RVC_SIDECAR, str(os.getpid())],
                             env=env, cwd=os.path.dirname(RVC_SIDECAR),
                             stdout=lf, stderr=subprocess.STDOUT,
                             creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        for _ in range(75):   # up to ~75 s
            if _rvc_health():
                _rvc_ok = True
                print("[rvc] sidecar ready (trained Reach voice online)", flush=True)
                return
            time.sleep(1)
        print("[rvc] sidecar NOT ready after 75 s: Kokoro only", flush=True)
    except Exception as e:
        print(f"[rvc] boot failed: {e}, Kokoro only", flush=True)

def _rvc_convert(seg):
    """Timbre-convert one int16@24k segment via the sidecar; original returned on failure."""
    global _rvc_ok, _rvc_until
    if not RVC_ON or not _rvc_ok:
        return seg
    if time.time() < _rvc_until:
        return seg
    try:
        req = urllib.request.Request(RVC_URL + "/convert", data=seg.tobytes(), method="POST")
        with urllib.request.urlopen(req, timeout=20) as r:
            raw = r.read()
        out = np.frombuffer(raw, dtype=np.int16)
        return out if out.size else seg
    except Exception as e:
        _rvc_ok = False
        _rvc_until = time.time() + 120
        print(f"[rvc] convert failed, Kokoro fallback 120 s: {e}", flush=True)
        return seg


def _omni_health():
    try:
        with urllib.request.urlopen(OMNI_URL + "/health", timeout=3) as r:
            return json.loads(r.read()).get("ready") is True
    except Exception:
        return False

def _omni_boot():
    """Spawn the OmniVoice sidecar (GPU0) if not already up; poll until ready (~25 s)."""
    global _omni_ok
    try:
        if _omni_health():
            _omni_ok = True
            print("[omni] sidecar already up", flush=True)
            return
        env = dict(os.environ)
        env["CUDA_VISIBLE_DEVICES"] = "0"   # omni runs on the idle headless GPU
        with open(OMNI_LOG, "a") as lf:
            subprocess.Popen([OMNI_PY, OMNI_SIDECAR, str(os.getpid())],
                             env=env, cwd=os.path.dirname(OMNI_SIDECAR),
                             stdout=lf, stderr=subprocess.STDOUT,
                             creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
        for _ in range(90):   # up to ~90 s
            if _omni_health():
                _omni_ok = True
                print("[omni] sidecar ready (OmniVoice Cortana live)", flush=True)
                return
            time.sleep(1)
        print("[omni] sidecar NOT ready after 90 s: Kokoro+RVC fallback", flush=True)
    except Exception as e:
        print(f"[omni] boot failed: {e}, Kokoro+RVC fallback", flush=True)

def _omni_synth(piece):
    """Synthesize one sentence piece in the cloned voice; None on any failure."""
    global _omni_ok, _omni_until
    if not OMNI_ON or not _omni_ok:
        return None
    if time.time() < _omni_until:
        return None
    try:
        req = urllib.request.Request(
            OMNI_URL + "/synth",
            data=json.dumps({"text": piece, "speed": OMNI_SPEED}).encode(),
            headers={"Content-Type": "application/json"}, method="POST")
        with urllib.request.urlopen(req, timeout=45) as r:
            raw = r.read()
        out = np.frombuffer(raw, dtype=np.int16)
        return out if out.size else None
    except Exception as e:
        _omni_ok = False
        _omni_until = time.time() + 180
        print(f"[omni] synth failed, Kokoro+RVC for 180 s: {e}", flush=True)
        return None

# ---------------- OmniVoice live voice (zero-shot clone, 2026-09-07) ----------------
# Primary kiosk voice: OmniVoice zero-shot clone of Reach Cortana (GPU0 sidecar).
# Kokoro(+RVC) remains the automatic fallback; engine chosen per sentence piece.
OMNI_ON = True
OMNI_URL = "http://127.0.0.1:18083"
OMNI_SIDECAR = r"C:\oracle\cache\sidecars\kiosk_omni_sidecar.py"
OMNI_PY = r"C:\oracle\cache\sidecars\venv-omnivoice\Scripts\python.exe"
OMNI_LOG = r"C:\oracle\cache\sidecars\omni_sidecar.log"
OMNI_SPEED = 1.35     # matches the Kokoro cadence the kiosk settled on
OMNI_SPLIT = r"(?<=[.!?])\s+"          # sentence-level pieces: gen cost ~1s fixed, so fat pieces keep the stream queue ahead
OMNI_PIECE_MAX = 200                    # chars; pathological run-ons fall back to comma granularity

def _omni_pieces(text):
    """Sentence-level pieces for OmniVoice. Per-call gen cost is ~1 s fixed regardless
    of size, so pieces MUST be fat (>= ~2-3 s audio) or playback underruns -> speech cuts."""
    parts = [p.strip() for p in re.split(OMNI_SPLIT, text) if p.strip()]
    out = []
    for p in parts:
        if len(p) > OMNI_PIECE_MAX:
            out += [q.strip() for q in re.split(r"(?<=[,;])\s+", p) if q.strip()]
        else:
            out.append(p)
    out = [p for p in out if p]
    return out or ([text.strip()] if text.strip() else [])
_omni_ok = False
_omni_until = 0.0

def _tts_pipeline():
    global _PIPE
    if _PIPE is None:
        from kokoro import KPipeline
        try:
            _PIPE = KPipeline(lang_code="a", device=TTS_DEVICE)
            print("[tts] KPipeline on CUDA", flush=True)
        except Exception as e:
            print(f"[tts] CUDA init failed ({e}): falling back to CPU", flush=True)
            _PIPE = KPipeline(lang_code="a")
    return _PIPE

def _synth_text(pipe, text):
    """Synthesize one sentence -> int16 numpy mono @24k (concats segments)."""
    parts = []
    for res in pipe(text, voice=TTS_VOICE, speed=TTS_SPEED, split_pattern=TTS_SPLIT):
        a = res.audio
        if hasattr(a, "numpy"):
            a = a.numpy()
        parts.append((np.clip(a, -1, 1) * 32767).astype(np.int16))
    if not parts:
        return None
    return np.concatenate(parts)

import kiosk_auth   # WAN password gate: kiosk-pass.json (pbkdf2), signed cookies, rate limiter

def http_response(status, body, ctype="text/plain"):
    return (f"HTTP/1.1 {status} {HTTPStatus(status).phrase}\r\n"
            f"Content-Type: {ctype}\r\nContent-Length: {len(body)}\r\n"
            "Cache-Control: no-store\r\n"      # tablet must never serve a stale client
            "Connection: close\r\n\r\n").encode("latin1") + body

def http_response_full(status, body, ctype="text/plain", extra_headers=()):
    hdrs = "".join(f"{k}: {v}\r\n" for k, v in extra_headers)
    return (f"HTTP/1.1 {status} {HTTPStatus(status).phrase}\r\n"
            f"Content-Type: {ctype}\r\nContent-Length: {len(body)}\r\n"
            f"Cache-Control: no-store\r\n{hdrs}"
            "Connection: close\r\n\r\n").encode("latin1") + body

def _read_cookie(head_text):
    m = re.search(r"Cookie:\s*(.+)", head_text, re.I)
    if not m:
        return None
    for part in m.group(1).split(";"):
        k, _, v = part.strip().partition("=")
        if k == kiosk_auth.COOKIE:
            return v
    return None

def ws_frame(opcode, payload):
    h = bytearray([0x80 | opcode]); n = len(payload)
    if n < 126: h.append(n)
    elif n < 65536: h.append(126); h += n.to_bytes(2, "big")
    else: h.append(127); h += n.to_bytes(8, "big")
    return bytes(h) + payload

async def send_text(writer, obj):
    writer.write(ws_frame(1, json.dumps(obj).encode()))
    await writer.drain()

class Resampler:
    """Streaming linear resampler (any rate -> SR)."""
    def __init__(self, in_rate):
        self.ratio = in_rate / SR
        self.pos = 0.0
    def process(self, xs):
        if self.ratio == 1.0: return xs
        out = []; n = len(xs); ratio = self.ratio
        while self.pos < n - 1:
            i = int(self.pos); f = self.pos - i
            out.append(xs[i] * (1 - f) + xs[i + 1] * f)
            self.pos += ratio
        self.pos -= n
        return out

def rms_db(xs):
    """RMS of int16 samples in NORMALIZED dBFS (divide by 32768)."""
    if not xs: return -120.0
    s = 0.0
    for v in xs:
        f = v / 32768.0
        s += f * f
    r = math.sqrt(s / len(xs))
    return 20.0 * math.log10(r + 1e-9)

def transcribe(pcm16):
    if MODEL is None: return ""
    arr = np.asarray(pcm16, dtype=np.int16).astype(np.float32) / 32768.0
    segs, _ = MODEL.transcribe(arr, language="en", beam_size=1, temperature=0.0,
                               vad_filter=False, condition_on_previous_text=False,
                               without_timestamps=True)
    text = "".join(s.text for s in segs).strip()
    if not text:
        peak = max(abs(int(v)) for v in pcm16) if pcm16 else 0
        print(f"[stt] EMPTY result: n={len(pcm16)} peak={peak}", flush=True)
    return text

DATE_RE = re.compile(r"\b(today'?s date|what'?s the date|what date is it|current date|"
                     r"what day is it|what day of the week|date today|what'?s today)\b", re.I)
TIME_RE = re.compile(r"\b(what time is it|current time|time right now|what'?s the time|"
                     r"do you have the time)\b", re.I)

def clock_answer(question):
    """Deterministic local date/time answers: no LLM involved (the brain cannot be trusted to call the tool reliably)."""
    from datetime import datetime
    from zoneinfo import ZoneInfo
    if re.search(r"\bin\s+[A-Z]", question) and not re.search(r"\bin the\b", question):
        return None                      # "in London" style -> let the LLM use get_datetime(zone)
    now = datetime.now(ZoneInfo("America/Los_Angeles"))
    if TIME_RE.search(question):
        return f"The time is {now:%I:%M %p} {now.tzname()}."
    if DATE_RE.search(question):
        return f"Today is {now:%A, %B %d, %Y}."
    return None

# ---------- Deterministic routers: kid questions that never need the LLM ----------
# Anything a regex can own outright (math, unit conversion, named-place weather)
# is answered here: correct AND instant. Uncertain -> return None -> brain path.
_UREG = None

def _pint():
    global _UREG
    if _UREG is None:
        from pint import UnitRegistry
        _UREG = UnitRegistry()
    return _UREG

MATH_WORDS = {"times": "*", "multiplied by": "*", "x": "*", "×": "*",
              "plus": "+", "minus": "-", "subtract": "-",
              "divided by": "/", "over": "/"}
MATH_Q_RE = re.compile(
    r"\b(?:what(?:'s| is)?|how much is)\s+(-?\d+(?:\.\d+)?)\s*"
    r"(\+|x|×|\*|\/|-|times|multiplied by|plus|minus|divided by|over)\s*"
    r"(-?\d+(?:\.\d+)?)\s*\??$", re.I)

def _router_math(q):
    m = MATH_Q_RE.search(q.strip())
    if not m:
        return None
    op_raw = m.group(2).lower().strip()
    op = MATH_WORDS.get(op_raw, op_raw)
    try:
        a, b = float(m.group(1)), float(m.group(3))
    except ValueError:
        return None
    if op == "/" and b == 0:
        return None
    val = {"+": a + b, "-": a - b, "*": a * b, "/": a / b}.get(op)
    if val is None:
        return None
    if val == int(val):
        return f"{m.group(1)} {op_raw} {m.group(3)} is {int(val):,}."
    return f"{m.group(1)} {op_raw} {m.group(3)} is {round(val, 4):,}."

CONVERT_UNITS = {
    "fahrenheit": "degF", "°f": "degF", "celsius": "degC", "°c": "degC",
    "centigrade": "degC", "km": "kilometer", "kilometers": "kilometer",
    "kilometres": "kilometer", "mi": "mile", "miles": "mile",
    "m": "meter", "meters": "meter", "metres": "meter",
    "ft": "foot", "feet": "foot", "cm": "centimeter", "centimeters": "centimeter",
    "lb": "pound", "lbs": "pound", "pounds": "pound",
    "kg": "kilogram", "kilos": "kilogram", "kilograms": "kilogram",
    "oz": "ounce", "ounces": "ounce", "g": "gram", "grams": "gram",
    "gal": "gallon", "gallons": "gallon", "qt": "quart",
    "l": "liter", "liters": "liter", "litres": "liter", "ml": "milliliter",
    "in": "inch", "inches": "inch",
    "mph": "mile / hour", "kph": "kilometer / hour",
}
CONVERT_RE = re.compile(
    r"\b(?:how many|what is|convert|how much is)\s+(-?\d+(?:\.\d+)?)\s*"
    r"([a-zA-Z°]+)\s+(?:to|in|into|is)\s+([a-zA-Z°]+)\s*\??$", re.I)
CONVERT_REV_RE = re.compile(
    r"\bhow many\s+([a-zA-Z°]+)\s+(?:is|are)\s+(-?\d+(?:\.\d+)?)\s*"
    r"([a-zA-Z°]+)\s*\??$", re.I)

def _spoken_unit(u):
    l = u.lower()
    if l in ("celsius", "centigrade", "°c"):
        return "degrees Celsius"
    if l in ("fahrenheit", "°f"):
        return "degrees Fahrenheit"
    return u

def _router_convert(q):
    t = q.strip()
    m = CONVERT_REV_RE.search(t)
    if m:
        dst_s, num_s, src_s = m.group(1), m.group(2), m.group(3)
    else:
        m = CONVERT_RE.search(t)
        if not m:
            return None
        num_s, src_s, dst_s = m.group(1), m.group(2), m.group(3)
    src = CONVERT_UNITS.get(src_s.lower(), src_s.lower().rstrip("s"))
    dst = CONVERT_UNITS.get(dst_s.lower(), dst_s.lower().rstrip("s"))
    try:
        val = float(num_s)
        mag = _pint().Quantity(val, src).to(dst).magnitude
    except Exception:
        return None
    if mag == int(mag):
        return f"{num_s} {_spoken_unit(src_s)} is {int(mag):,} {_spoken_unit(dst_s)}."
    s = f"{mag:,.2f}".rstrip("0").rstrip(".")
    return f"{num_s} {_spoken_unit(src_s)} is {s} {_spoken_unit(dst_s)}."

WEATHER_Q_RE = re.compile(
    r"\b(?:what'?s the weather|what is the weather|weather|forecast|temperature|"
    r"raining|rainy|snowing|sunny|windy|hot|cold|warm|is it going to rain|"
    r"do i need a (?:jacket|umbrella|coat))\b[\s\S]{0,45}?"
    r"\b(?:in|at|for)\s+([A-Za-z][A-Za-z .'\-]{1,40}?)\s*[?.!]*$", re.I)

def _router_weather(q):
    m = WEATHER_Q_RE.search(q.strip())
    if not m:
        return None
    place = m.group(1).strip()
    place = re.sub(r"\s+(?:right now|right at this moment|today|tonight|outside|"
                   r"at the moment|now|this (?:afternoon|evening|morning|weekend|week))\s*$",
                   "", place, flags=re.I).strip()
    if not place or len(place) > 48:
        return None
    if "," not in place:
        m2 = re.search(r"\s+([A-Z]{2})$", place)      # "San Diego CA" -> "San Diego, CA"
        if m2:
            place = place[:m2.start()] + ", " + m2.group(1)
    try:
        res = run_tool("get_weather", {"place": place})
        if (not res or len(res) < 12 or res.lower().startswith(
                ("tool ", "error", "sorry", "could", "failed", "no location", "not found"))
                and "," not in place):
            # geocoder wants "City, STATE": Oceanside California -> Oceanside, California
            parts = place.rsplit(" ", 1)
            if len(parts) == 2 and parts[1][:1].isupper():
                res = run_tool("get_weather", {"place": parts[0] + ", " + parts[1]})
    except Exception:
        return None
    if not res or len(res) < 12 or res.lower().startswith(
            ("tool ", "error", "sorry", "could", "failed", "no location", "not found")):
        return None
    return res

# ---- Bible (KJV FTS5 index, C:/oracle/cache/bible.db, built by tools/build_bible_db.py) ----
_BIBLE_DB = os.path.join(os.path.dirname(os.path.abspath(__file__)), "cache", "bible.db")
# DB stores canonical underscore labels (1_Corinthians, Song_of_Solomon). The
# router matches any spoken/written book form, then queries with the DB label.
# List mirrors KJV_BOOK_ALT above (keep in sync); DB order = canonical order.
_BIBLE_CANON = """Genesis Exodus Leviticus Numbers Deuteronomy
Joshua Judges Ruth 1_Samuel 2_Samuel 1_Kings 2_Kings 1_Chronicles 2_Chronicles Ezra
Nehemiah Esther Job Psalms Proverbs Ecclesiastes Song_of_Solomon Isaiah Jeremiah
Lamentations Ezekiel Daniel Hosea Joel Amos Obadiah Jonah Micah Nahum Habakkuk
Zephaniah Haggai Zechariah Malachi Matthew Mark Luke John Acts Romans
1_Corinthians 2_Corinthians Galatians Ephesians Philippians Colossians
1_Thessalonians 2_Thessalonians 1_Timothy 2_Timothy Titus Philemon Hebrews
James 1_Peter 2_Peter 1_John 2_John 3_John Jude Revelation""".split()
assert len(_BIBLE_CANON) == 66, f"bible book list must have 66 entries, has {len(_BIBLE_CANON)}"
_BIBLE_TO_DB = {c.replace("_", " ").lower(): c for c in _BIBLE_CANON}
_BIBLE_TO_DB["psalm"] = "Psalms"        # singular spoken form -> canonical label
_BIBLE_ALT = [c.replace("_", " ") for c in _BIBLE_CANON] + ["Psalm"]
_BIBLE_RE = re.compile(
    r"\b(?:the\s+(?:book|epistle|books?)\s+of\s+)?("
    + "|".join(re.escape(b) for b in _BIBLE_ALT)
    + r")\b", re.I)

# spoken numbers -> digits, incl. full teens and "twenty three" -> 23. Chapter/
# verse counts never exceed ~176, so tens+unit composition covers everything.
_SINGLE = {"one": 1, "two": 2, "three": 3, "four": 4, "five": 5,
           "six": 6, "seven": 7, "eight": 8, "nine": 9}
_TEENS = {"ten": 10, "eleven": 11, "twelve": 12, "thirteen": 13, "fourteen": 14,
          "fifteen": 15, "sixteen": 16, "seventeen": 17, "eighteen": 18, "nineteen": 19}
_TENS = {"twenty": 20, "thirty": 30, "forty": 40, "fifty": 50, "sixty": 60, "seventy": 70}

def _ref_numbers(text):
    """Every number in `text` as int, resolving spoken words ('twenty three' -> 23)."""
    toks = re.findall(r"\d+|[A-Za-z]+", text)
    out, i = [], 0
    while i < len(toks):
        t = toks[i].lower()
        if t.isdigit():
            out.append(int(t)); i += 1
        elif t in _TENS and i + 1 < len(toks) and toks[i + 1].lower() in _SINGLE:
            out.append(_TENS[t] + _SINGLE[toks[i + 1].lower()]); i += 2
        elif t in _TENS:
            out.append(_TENS[t]); i += 1
        elif t in _TEENS:
            out.append(_TEENS[t]); i += 1
        elif t in _SINGLE:
            out.append(_SINGLE[t]); i += 1
        else:
            i += 1
    return out

def _router_bible(q):
    """Exact KJV quote for a recognizable reference ('John 3:16', 'psalm 23 verse 1',
    '1 Corinthians 13:4', 'john three sixteen'). Returns None on a miss so the
    LLM/web-search path answers: never paraphrase scripture on a partial parse."""
    if not os.path.exists(_BIBLE_DB):
        return None
    m = _BIBLE_RE.search(q.strip())
    if not m:
        return None
    db_book = _BIBLE_TO_DB.get(re.sub(r"\s+", " ", m.group(1)).strip().lower())
    if not db_book:
        return None
    nums = _ref_numbers(q[m.end(1):])
    if len(nums) < 2:
        return None
    ch, vs = nums[0], nums[1]
    if not (1 <= ch <= 150 and 1 <= vs <= 176):
        return None
    try:
        conn = __import__("sqlite3").connect(f"file:{_BIBLE_DB}?mode=ro", uri=True)
        try:
            row = conn.execute("select text from kjv where book=? and chapter=? and verse=?",
                               (db_book, ch, vs)).fetchone()
            if not row:
                return None
            return (f"{db_book.replace('_', ' ')} chapter {ch}, verse {vs} says, "
                    f"\u201c{row[0].rstrip('.,;:')}.\u201d")
        finally:
            conn.close()
    except Exception:
        return None

def deterministic_answer(question):
    """Router chain: date/time -> bible -> weather -> math -> convert. Returns a
    speech-ready answer or None to fall through to the LLM brain."""
    try:
        a = clock_answer(question)
        if a:
            return a
        a = _router_bible(question)      # exact KJV quote, never paraphrased
        if a:
            return a
        a = _router_weather(question)     # first: "is it hot in Phoenix?" style
        if a:
            return a
        a = _router_math(question)
        if a:
            return a
        a = _router_convert(question)
        if a:
            return a
    except Exception as e:
        print(f"[router] ERROR {type(e).__name__}: {e} - falling through to brain",
              flush=True)
    return None

class Session:
    def __init__(self, writer):
        self.writer = writer
        self.rate = 48000
        self.res = None
        self.utt = []            # whole current utterance @16k
        self.rem = []            # resampled tail buffer (keeps partial frames)
        self.speech = False
        self.floor = -45.0       # whisper noise-gate floor (rms dBFS), NOT speech logic
        self.vit = VADIterator(_SILERO, threshold=0.5, min_silence_duration_ms=200,
                               speech_pad_ms=64) if _SILERO else None
        self.state = "idle"
        self.transcribing = False
        self.last_partial = 0.0
        self.rev = 0
        self.fin_task = None
        self.ask_mode = False
        self.busy_answer = False
        self.ask_seq = 0          # bumped per answer; guards stale follow-up arming
        self.fu_task = None       # follow-up window task
        self.fu_deadline = 0.0
        self.last_active = time.monotonic()  # last answered question; idle watchdog base
        self.idle_task = None     # quiet auto-reset watchdog (IDLE_RESET_S)
        self.manual_off_at = 0.0  # user tapped ASK off -> don't re-arm soon after
        self.suppress_speech = False  # STOP pressed while brain mid-flight: don't speak its result
        self.history = []      # rolling Q/A context (in-session memory)
        self.tts_task = None
        self.cut_tts = False
        self.speaking = False
        self._cut_event = threading.Event()
        self.drop_utt = False       # utterance started while she was talking (her echo): never answer it
        self.tts_ended_at = 0.0     # monotonic time her last speech stream ended (echo-tail lockout base)

    async def send(self, obj):
        try: await send_text(self.writer, obj)
        except Exception: pass

    async def set_state(self, s):
        if s != self.state:
            self.state = s
            await self.send({"type": "state", "s": s})
            print(f"[stt] state -> {s}", flush=True)

    def feed_audio(self, data):
        """Bytes -> int16 -> resample -> 512-sample windows -> Silero VAD events."""
        if self.res is None: self.res = Resampler(self.rate)
        n = len(data) // 2
        samples = struct.unpack(f"<{n}h", data[:n * 2])   # bytes -> int16 samples
        self.rem.extend(self.res.process(samples))
        ev_out = "idle"; started = False; ended = False
        while len(self.rem) >= 512:
            win = self.rem[:512]; self.rem = self.rem[512:]
            db = rms_db(win)
            if not self.speech:                 # gate floor only (noise-follow)
                if db <= self.floor + 10.0:
                    self.floor = 0.95 * self.floor + 0.05 * db
                self.floor = max(-60.0, min(-25.0, self.floor))
            if self.vit is not None:
                arr = np.asarray(win, dtype=np.int16).astype(np.float32) / 32768.0
                ev = self.vit(arr)
                if ev:
                    if "start" in ev:
                        self.speech = True; started = True
                        if self.ask_mode and FOLLOWUP_WINDOW > 0:
                            self.fu_deadline = time.monotonic() + FOLLOWUP_WINDOW
                    if "end" in ev:
                        self.speech = False; ended = True
            else:                                # no silero: crude energy fallback
                if db > self.floor + 10: self.speech = True
                elif db < self.floor + 4: self.speech = False
            if self.speech or ended:
                self.utt.extend(win)
        if ended: return "end"
        if started: return "start"
        return "speech" if self.speech else "idle"

    async def maybe_partial(self):
        if self.transcribing or self.drop_utt: return
        now = time.monotonic()
        utt_s = len(self.utt) / SR
        if utt_s < MIN_PARTIAL or now - self.last_partial < PARTIAL_GAP: return
        if utt_s > 6.0:
            # never-silent room guard: emit a rolling final of everything so far,
            # keep the last 2s as a tail for continuity, and keep listening.
            if rms_db(self.utt) < self.floor + 6:
                self.last_partial = now
                return
            self.transcribing = True
            t0 = time.monotonic()
            txt = await asyncio.to_thread(transcribe, self.utt)
            dt = (time.monotonic() - t0) * 1000
            self.utt = self.utt[-int(2.0 * SR):]
            self.transcribing = False
            if txt and self.state != "idle":
                self.rev += 1
                await self.send({"type": "final", "rev": self.rev, "text": txt,
                                 "ms": int(dt), "roll": True})
                print(f"[stt] ROLLING-FINAL {dt:.0f}ms: {txt[:80]}", flush=True)
            self.last_partial = now
            return
        window = self.utt[-int(min(PARTIAL_WIN, utt_s) * SR):]
        if rms_db(window) < self.floor + 6:   # too quiet: likely noise, skip
            self.last_partial = now
            return
        self.transcribing = True
        t0 = time.monotonic()
        txt = await asyncio.to_thread(transcribe, window)
        dt = (time.monotonic() - t0) * 1000
        if txt and self.state != "idle":
            self.rev += 1
            await self.send({"type": "partial", "rev": self.rev, "text": txt})
            print(f"[stt] partial {dt:.0f}ms: {txt[:60]}", flush=True)
            self.last_partial = now
        self.transcribing = False

    async def finalize_soon(self):
        """Grace period: if speech resumes within GRACE, keep the utterance whole."""
        try:
            await asyncio.sleep(GRACE)
            if self.speech:
                return  # resumed: utterance continues
            await self.finalize()
        except asyncio.CancelledError:
            pass
        finally:
            if self.fin_task is asyncio.current_task():
                self.fin_task = None

    async def answer(self, question):
        """Phase 2 agent loop: brain with tools (web/weather/math), max 2 passes."""
        if self.busy_answer: return
        self.ask_seq += 1
        seq = self.ask_seq
        # quiet auto-reset: stale context (>IDLE_RESET_S since last question) never reaches the brain
        if IDLE_RESET_S > 0 and time.monotonic() - self.last_active >= IDLE_RESET_S:
            self.history = []
        self.last_active = time.monotonic()
        self._idle_schedule()
        direct = await asyncio.to_thread(deterministic_answer, question)
        if direct:
            self.ask_mode = False
            await self.send({"type": "answer_state", "s": "done", "text": direct})
            print(f"[direct] {direct[:90]}", flush=True)
            self.history.append({"role": "user", "content": question})
            self.history.append({"role": "assistant", "content": direct})
            self.history = self.history[-HISTORY_TURNS * 2:]
            self.tts_task = asyncio.create_task(self.speak(direct))
            self.tts_task.add_done_callback(lambda _t: self._fu_schedule(seq))
            return
        self.busy_answer = True
        await self.send({"type": "answer_state", "s": "thinking"})
        ans = ""
        agent_ok = False
        bc = None                                 # backchannel task (tool lookups)
        if AGENT_MODE:
            try:
                a = await asyncio.to_thread(agent_ask, question)
                if a:
                    ans = clean_for_speech(a)
                    agent_ok = True
                    print(f"[agent] {len(ans.split())}w answer", flush=True)
                else:
                    print("[agent] no answer - falling back to direct brain", flush=True)
            except Exception as e:
                print(f"[agent] ERROR {type(e).__name__}: {e} - falling back", flush=True)
        if not agent_ok:
            messages = [{"role": "system", "content": PERSONA}] + self.history + \
                       [{"role": "user", "content": question}]
            try:
                for _pass in range(2):
                    msg = await asyncio.to_thread(llm_respond, messages, TOOLS)
                    content = (msg.get("content") or "").strip()
                    tool_calls = msg.get("tool_calls")
                    if not tool_calls:
                        ans = content
                        break
                    if _pass >= 1:
                        ans = content or "Sorry, I could not finish that lookup."
                        break
                    await self.send({"type": "answer_state", "s": "tool"})
                    if bc is None:
                        bc = asyncio.create_task(self.play_canned())
                    messages.append({"role": "assistant", "content": content or None,
                                     "tool_calls": tool_calls})
                    for i, tc in enumerate(tool_calls):
                        fn = (tc.get("function") or {}).get("name", "?")
                        try:
                            args = json.loads((tc.get("function") or {}).get("arguments") or "{}")
                        except Exception:
                            args = {}
                        print(f"[brain] tool {fn}({json.dumps(args)[:100]})", flush=True)
                        res = await asyncio.to_thread(run_tool, fn, args)
                        print(f"[brain] tool result: {res[:120]}", flush=True)
                        messages.append({"role": "tool",
                                         "tool_call_id": tc.get("id") or f"call_{i}",
                                         "content": res})
                else:
                    ans = ans or "Sorry, I could not finish that."
            except Exception as e:
                print(f"[brain] ERROR: {e}", flush=True)
                await self.send({"type": "answer_state", "s": "error",
                                 "text": f"Brain unreachable: {type(e).__name__}"})
                self.busy_answer = False
                return
        self.busy_answer = False
        self.ask_mode = False
        if ans:
            ans = clean_for_speech(ans)
            await self.send({"type": "answer_state", "s": "done", "text": ans})
            print(f"[brain] {len(ans.split())}w: {ans[:90]}", flush=True)
            self.history.append({"role": "user", "content": question})
            self.history.append({"role": "assistant", "content": ans})
            self.history = self.history[-HISTORY_TURNS * 2:]
            if bc is not None:
                self.cut_tts = True               # stop canned line if still playing
                try:
                    bc.cancel()
                except Exception:
                    pass
            if not self.suppress_speech:
                self.tts_task = asyncio.create_task(self.speak(ans))
                self.tts_task.add_done_callback(lambda _t: self._fu_schedule(seq))
            else:
                print(f"[stop] answer suppressed ({len(ans.split())}w), no speech", flush=True)
            self.suppress_speech = False
        else:
            await self.send({"type": "answer_state", "s": "error", "text": "Empty answer."})

    async def send_bin(self, payload):
        try:
            self.writer.write(ws_frame(2, payload))
            await self.writer.drain()
        except Exception:
            pass

    _BC_CACHE = None

    async def play_canned(self):
        """Short 'one moment' line spoken while a tool lookup runs (pass 2).
        Independent of speak(): touches no state flags; answer() awaits/cancels
        it before the real answer speaks. Pipe is warmed at boot."""
        sent_start = False
        try:
            if Session._BC_CACHE is None:
                pipe = await asyncio.to_thread(_tts_pipeline)
                a = await asyncio.to_thread(_synth_text, pipe, BC_LINE)
                if a is None or not len(a):
                    return
                Session._BC_CACHE = a
            audio = Session._BC_CACHE
            await self.send({"type": "tts_start", "rate": TTS_RATE})
            sent_start = True
            step = max(1, len(audio) // 3)
            for i in range(0, len(audio), step):
                if self._cut_event.is_set():
                    break
                await self.send_bin(audio[i:i + step].tobytes())
                await asyncio.sleep(0.05)
            await self.send({"type": "tts_end"})
        except asyncio.CancelledError:
            if sent_start:
                try:
                    await self.send({"type": "tts_end"})
                except Exception:
                    pass
            raise
        except Exception as e:
            print(f"[bc] error: {e}", flush=True)

    async def speak(self, text):
        """Stream speech as the model generates it: each Kokoro segment ships the
        instant it's produced, so first audio lands in a few hundred ms, not seconds."""
        if not SPEAK_ON or not text or self.speaking:
            return
        self.speaking = True
        self.cut_tts = False
        self._cut_event.clear()
        loop = asyncio.get_running_loop()
        q = asyncio.Queue(maxsize=4)
        try:
            pipe = await asyncio.to_thread(_tts_pipeline)
        except Exception as e:
            print(f"[tts] pipeline load failed: {e}", flush=True)
            self.speaking = False
            return
        # Speak the WHOLE displayed answer (parity with the text panel). The
        # persona caps answers at 2-4 short sentences, so this is normally brief.
        # Hard ceiling (~1500 chars) only guards a runaway ramble; max_tokens=300
        # keeps real answers far under it.
        trimmed = text.strip()[:1500]

        def worker():
            pieces = _omni_pieces(trimmed)
            use_omni = _omni_ok and OMNI_ON
            try:
                for piece in pieces:
                    if self._cut_event.is_set():
                        break
                    seg = None
                    if use_omni:
                        seg = _omni_synth(piece)   # cloned Reach voice (primary)
                        if seg is None:
                            use_omni = False
                            print("[tts] omni failed mid-answer -> Kokoro+RVC remainder", flush=True)
                    if seg is None:
                        seg = _synth_text(pipe, piece)   # Kokoro fallback piece
                        if seg is not None:
                            seg = _rvc_convert(seg)      # + timbre conversion
                    if seg is not None and len(seg):
                        loop.call_soon_threadsafe(q.put_nowait, seg)
            except Exception as e:
                loop.call_soon_threadsafe(q.put_nowait, e)
            finally:
                loop.call_soon_threadsafe(q.put_nowait, None)

        worker_task = asyncio.create_task(asyncio.to_thread(worker))
        ended_ok = False
        try:
            await self.send({"type": "tts_start", "rate": TTS_RATE})
            t0 = time.monotonic()
            first = True
            while True:
                seg = await q.get()
                if seg is None:
                    break
                if isinstance(seg, Exception):
                    print(f"[tts] gen error: {seg}", flush=True)
                    break
                data = seg.tobytes()
                if first:
                    print(f"[tts] first audio after {time.monotonic()-t0:.2f}s ({len(data)}B)", flush=True)
                    first = False
                for i in range(0, len(data), TTS_FRAME * 2):
                    if self.cut_tts:
                        break
                    await self.send_bin(data[i:i + TTS_FRAME * 2])
                    await asyncio.sleep(0)
                if self.cut_tts:
                    break
            ended_ok = not self.cut_tts
        except asyncio.CancelledError:
            ended_ok = False
        finally:
            self._cut_event.set()
            if not worker_task.done():
                worker_task.cancel()
            try:
                await self.send({"type": "tts_end" if ended_ok else "tts_stop"})
            except Exception:
                pass
            self.speaking = False
            self.cut_tts = False
            self.tts_ended_at = time.monotonic()   # echo-tail lockout base for the next utterance

    def cut_speech(self):
        """Barge-in: user started talking -> stop her immediately."""
        if self.speaking:
            self.cut_tts = True
            if self.tts_task and not self.tts_task.done():
                self.tts_task.cancel()
                self.tts_task = None

    def stop_now(self):
        """STOP button (tablet). Context-aware abort:
        - she is speaking        -> cut TTS immediately
        - user is mid-question   -> "answer now": finalize the buffered utterance NOW
        - thinking / idle / armed-> suppress any in-flight answer's speech + disarm"""
        self.cut_speech()
        self.manual_off_at = time.monotonic()       # no follow-up re-arm for 60s
        if self.fin_task and not self.fin_task.done():
            self.fin_task.cancel()
            self.fin_task = None
        if self.speech or self.utt:
            self.speech = False
            asyncio.create_task(self._stop_finalize())
        else:
            self.suppress_speech = True             # brain may still be working: silence it
            self.ask_mode = False
            self._fu_stop()
            asyncio.create_task(self.send({"type": "ask_state", "on": False}))
            asyncio.create_task(self.set_state("idle"))

    async def _stop_finalize(self):
        """'Answer now': transcribe the buffered utterance immediately (no grace wait)."""
        try:
            await self.set_state("processing")
            await self.finalize()                   # may trigger answer() via maybe_answer
        finally:
            self.ask_mode = False
            self._fu_stop()
            await self.send({"type": "ask_state", "on": False})
            await self.set_state("idle")

    # ---- follow-up window: after an answer she stays armed briefly ----
    def _fu_stop(self):
        if self.fu_task and not self.fu_task.done():
            self.fu_task.cancel()
        self.fu_task = None
        self.fu_deadline = 0.0

    def _fu_schedule(self, seq):
        """Called when an answer's TTS finishes. Button-only mode (FOLLOWUP_WINDOW=0):
        disarm: send ask_state off so the client's ASK button resets for the next tap.
        Otherwise: arm/refresh the follow-up window (chain of follow-ups)."""
        try:
            if FOLLOWUP_WINDOW <= 0:
                if seq == self.ask_seq and not self.busy_answer:
                    asyncio.create_task(self.send({"type": "ask_state", "on": False}))
                return
            if (seq != self.ask_seq or self.busy_answer
                    or time.monotonic() - self.manual_off_at < 60.0):
                return
            self.ask_mode = True
            self.fu_deadline = time.monotonic() + FOLLOWUP_WINDOW
            if self.fu_task and not self.fu_task.done():
                return                    # live loop sees the refreshed deadline
            self.fu_task = asyncio.create_task(self._fu_loop())
        except Exception as e:
            print(f"[fu] schedule error: {e}", flush=True)

    async def _fu_loop(self):
        try:
            await self.send({"type": "ask_state", "on": True, "fu": True})
            while True:
                if self.writer.is_closing():
                    return
                rem = self.fu_deadline - time.monotonic()
                if rem <= 0:
                    await asyncio.sleep(0.05)   # re-check: a refresh may have landed
                    if self.fu_deadline - time.monotonic() <= 0:
                        break
                else:
                    await asyncio.sleep(min(0.5, rem))
            if self.ask_mode and not self.busy_answer and not self.speech:
                self.ask_mode = False
                await self.send({"type": "ask_state", "on": False})
        except asyncio.CancelledError:
            pass
        except Exception:
            pass
        finally:
            self.fu_task = None

    # ---- quiet auto-reset: after IDLE_RESET_S with no question, drop memory silently ----
    def _idle_schedule(self):
        """(Re)arm the idle watchdog from the last answered question. Each question rolls it."""
        try:
            if IDLE_RESET_S <= 0:
                return
            if self.idle_task and not self.idle_task.done():
                self.idle_task.cancel()
            self.idle_task = asyncio.create_task(self._idle_loop())
        except Exception as e:
            print(f"[idle] schedule error: {e}", flush=True)

    async def _idle_loop(self):
        try:
            while True:
                if self.writer.is_closing():
                    return
                rem = self.last_active + IDLE_RESET_S - time.monotonic()
                if rem <= 0:
                    break
                await asyncio.sleep(min(5.0, rem))
            if time.monotonic() - self.last_active >= IDLE_RESET_S:
                self._idle_reset()
        except asyncio.CancelledError:
            pass
        finally:
            self.idle_task = None

    def _idle_reset(self):
        """Quiet context wipe: no spoken ack, no fuss, the next speaker starts fresh."""
        print(f"[idle] {IDLE_RESET_S/60:.0f}min quiet auto-reset (cleared {len(self.history)} entries)",
              flush=True)
        self.history = []
        self.cut_speech()
        self.suppress_speech = True      # silence any wedged in-flight answer
        self.ask_mode = False
        self._fu_stop()
        asyncio.create_task(self.send({"type": "ask_state", "on": False}))
        asyncio.create_task(self.set_state("idle"))

    def reset_conversation(self, reason="button"):
        """New conversation: wipe in-session memory, disarm, silence any in-flight answer."""
        print(f"[reset] {reason}: clearing {len(self.history)} history entries", flush=True)
        self.cut_speech()
        self.suppress_speech = True          # silence any in-flight brain answer
        self.history = []
        self.ask_mode = False
        self._fu_stop()
        self.manual_off_at = time.monotonic()
        self.tts_task = asyncio.create_task(self.speak(RESET_ACK))
        asyncio.create_task(self.send({"type": "ask_state", "on": False}))
        self.last_active = time.monotonic()   # reset counts as activity: rolls the idle window
        self._idle_schedule()

    async def maybe_answer(self, txt):
        if len(txt.split()) <= 6 and RESET_RE.match(txt.strip()):
            self.reset_conversation("voice")
            return
        if self.ask_mode and not self.busy_answer and len(txt.split()) >= 2:
            self.ask_mode = False
            self.suppress_speech = False    # fresh question: she wants to hear this answer
            print(f"[brain] question: {txt[:90]}", flush=True)
            await self.send({"type": "answer_state", "s": "asked"})
            asyncio.create_task(self.answer(txt))

    async def finalize(self):
        while self.transcribing: await asyncio.sleep(0.05)
        if not self.utt:
            self.drop_utt = False
            await self.set_state("idle")
            return
        if self.drop_utt:                 # her own echo / post-TTS tail: never transcribe or answer
            print(f"[vad] dropped {len(self.utt)/SR:.2f}s self-echo utterance", flush=True)
            self.utt = []
            self.drop_utt = False
            await self.set_state("idle")
            return
        if rms_db(self.utt) < self.floor + 6:   # noise only: discard
            self.utt = []
            await self.set_state("idle")
            return
        t0 = time.monotonic()
        txt = await asyncio.to_thread(transcribe, self.utt)
        dt = (time.monotonic() - t0) * 1000
        self.utt = []
        if txt:
            await self.send({"type": "final", "rev": self.rev, "text": txt, "ms": int(dt)})
            print(f"[stt] FINAL {dt:.0f}ms ({len(txt.split())}w): {txt[:80]}", flush=True)
            await self.maybe_answer(txt)
        await self.set_state("idle")

async def ws_session(reader, writer, peer, key):
    accept = base64.b64encode(hashlib.sha1((key + GUID).encode()).digest()).decode()
    writer.write(("HTTP/1.1 101 Switching Protocols\r\n"
                  "Upgrade: websocket\r\nConnection: Upgrade\r\n"
                  f"Sec-WebSocket-Accept: {accept}\r\n\r\n").encode())
    await writer.drain()
    sess = Session(writer)
    print(f"[ws] {peer} upgraded", flush=True)
    try:
        while True:
            try:
                hdr = await asyncio.wait_for(reader.readexactly(2), 30)
            except (asyncio.IncompleteReadError, asyncio.TimeoutError):
                break
            opcode = hdr[0] & 0x0F
            ln = hdr[1] & 0x7F
            if ln == 126: ln = int.from_bytes(await reader.readexactly(2), "big")
            elif ln == 127: ln = int.from_bytes(await reader.readexactly(8), "big")
            if ln > MAX_FRAME: break
            mask = await reader.readexactly(4) if (hdr[1] >> 7) else None
            payload = await reader.readexactly(ln)
            if mask: payload = bytes(b ^ mask[i % 4] for i, b in enumerate(payload))
            if opcode == 8:
                writer.write(ws_frame(8, payload[:125])); await writer.drain(); break
            elif opcode == 9:
                writer.write(ws_frame(10, payload)); await writer.drain()
            elif opcode == 1:
                try:
                    m = json.loads(payload.decode())
                    if m.get("type") == "ping":
                        await sess.send({"type": "pong", "t": m.get("t")})
                    elif m.get("type") == "hello":
                        sess.rate = int(m.get("rate", 48000))
                        sess.res = None
                        print(f"[ws] {peer} hello rate={sess.rate}", flush=True)
                    elif m.get("type") == "ask":
                        sess._fu_stop()               # manual tap takes over follow-up state
                        sess.ask_mode = bool(m.get("on", False))
                        if sess.ask_mode:
                            # Button tap while she is speaking = INTENTIONAL interrupt: cut her
                            # and clear the echo lockout: their next words are a real question,
                            # not her trailing playback (which drop_utt would otherwise swallow).
                            sess.cut_speech()
                            sess.tts_ended_at = 0.0
                        if not sess.ask_mode:
                            sess.manual_off_at = time.monotonic()
                        print(f"[ws] {peer} ask_mode={sess.ask_mode}", flush=True)
                        await sess.send({"type": "ask_state", "on": sess.ask_mode})
                    elif m.get("type") == "stop":
                        print(f"[ws] {peer} STOP", flush=True)
                        sess.stop_now()
                    elif m.get("type") == "reset":
                        print(f"[ws] {peer} RESET (new conversation)", flush=True)
                        sess.reset_conversation("button")
                except Exception: pass
            elif opcode == 2:
                ev = sess.feed_audio(payload)
                if ev == "start":
                    # Anti-self-trigger (AUTO_BARGE_IN=False): speech starting while she is
                    # talking or within the post-TTS lockout is HER OWN VOICE through the
                    # room (sentence-pause echo; some devices lack AEC). Never cut on it,
                    # never answer it: mark for drop. Real interrupts = STOP/tap.
                    if (sess.speaking
                            or time.monotonic() < sess.tts_ended_at + POST_TTS_LOCKOUT_S):
                        sess.drop_utt = True
                    else:
                        sess.drop_utt = False
                        if AUTO_BARGE_IN:
                            sess.cut_speech()      # barge-in: user talking over her
                        sess.last_partial = time.monotonic()
                        if sess.fin_task and not sess.fin_task.done():
                            sess.fin_task.cancel()
                            sess.fin_task = None
                    await sess.set_state("speech")
                if sess.speech and not sess.drop_utt:
                    await sess.set_state("speech")
                    await sess.maybe_partial()
                elif ev == "end":
                    await sess.set_state("processing")
                    if sess.fin_task and not sess.fin_task.done():
                        sess.fin_task.cancel()
                    sess.fin_task = asyncio.create_task(sess.finalize_soon())
    except Exception as e:
        print(f"[ws] {peer} err {e}", flush=True)
    print(f"[ws] {peer} closed", flush=True)

async def handle(reader, writer):
    peer = writer.get_extra_info("peername")
    print(f"[conn] {peer}", flush=True)
    is_tls = writer.get_extra_info("ssl_object") is not None   # True on the 8444 TLS mirror
    try:
        while True:
            try:
                head = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), 15)
            except Exception:
                break
            text = head.decode("latin1")
            if "upgrade: websocket" in text.lower():
                # ---- auth gate: no valid cookie -> no upgrade (runs BEFORE any work)
                cookie = _read_cookie(text)
                if not (cookie and kiosk_auth.verify_cookie(cookie)):
                    print(f"[auth] ws denied {peer}", flush=True)
                    writer.write(http_response_full(401, b"unauthorized"))
                    await writer.drain()
                    break
                m = re.search(r"Sec-WebSocket-Key:\s*(.+)", text, re.I)
                await ws_session(reader, writer, peer, m.group(1).strip() if m else "")
                break
            m = re.search(r"^(GET|POST)\s+(\S+)", text)
            method = m.group(1) if m else "GET"
            path = m.group(2).split("?")[0] if m else "/"
            cookie = _read_cookie(text)
            authed = bool(cookie and kiosk_auth.verify_cookie(cookie))
            ip = peer[0] if peer else "?"

            if method == "POST" and path == "/auth":
                mlen = re.search(r"Content-Length:\s*(\d+)", text, re.I)
                try:
                    body = await asyncio.wait_for(reader.readexactly(int(mlen.group(1))), 10) if mlen else b"{}"
                    pw = json.loads(body).get("password", "")
                except Exception:
                    pw = ""
                if not kiosk_auth.limiter.ok(ip):
                    writer.write(http_response_full(429, b'{"ok":false,"msg":"rate-limited"}', "application/json"))
                elif kiosk_auth.check_password(pw):
                    kiosk_auth.limiter.reset(ip)
                    c = kiosk_auth.issue_cookie()
                    secure = "; Secure" if is_tls else ""     # plain-http 8443 (tablet) rejects Secure cookies
                    writer.write(http_response_full(
                        200, b'{"ok":true}', "application/json",
                        extra_headers=(("Set-Cookie",
                            f"oracle_auth={c}; Path=/; HttpOnly; SameSite=Strict; Max-Age={kiosk_auth.MAX_AGE}{secure}"),)))
                    print(f"[auth] ok {peer}", flush=True)
                else:
                    kiosk_auth.limiter.fail(ip)
                    print(f"[auth] FAIL {peer}", flush=True)
                    writer.write(http_response_full(401, b'{"ok":false}', "application/json"))
                await writer.drain()
                break

            if path == "/login":
                try:
                    with open(os.path.join(WEB_DIR, "login.html"), "rb") as f:
                        body = f.read()
                    writer.write(http_response_full(200, body, "text/html; charset=utf-8"))
                except OSError:
                    writer.write(http_response_full(404, b"login.html missing"))
                await writer.drain()
                break

            if not authed:
                writer.write(http_response_full(302, b"", extra_headers=(("Location", "/login"),)))
                await writer.drain()
                break

            if path in ("/", "/index.html"):
                try:
                    with open(os.path.join(WEB_DIR, "index.html"), "rb") as f:
                        body = f.read()
                    writer.write(http_response_full(200, body, "text/html; charset=utf-8"))
                except OSError:
                    writer.write(http_response_full(404, b"index.html missing"))
            elif path == "/oracle":          # backup client: procedural avatar canvas scene
                try:
                    with open(os.path.join(WEB_DIR, "index-oracle.html"), "rb") as f:
                        body = f.read()
                    writer.write(http_response_full(200, body, "text/html; charset=utf-8"))
                except OSError:
                    writer.write(http_response_full(404, b"index-oracle.html missing"))
            elif path == "/favicon.ico":
                writer.write(http_response_full(404, b""))
            else:
                writer.write(http_response_full(404, b"not found"))
            await writer.drain()
            break
    except Exception as e:
        print(f"[conn] {peer} err {e}", flush=True)
    try:
        writer.close(); await writer.wait_closed()
    except Exception: pass

async def warm_tts():
    """Preload Kokoro at boot AND tune real inference shapes (oneDNN benchmarks
    per shape on first use: a tiny warmup text leaves the real sizes cold)."""
    try:
        t0 = time.time()
        pipe = await asyncio.to_thread(_tts_pipeline)
        warm_text = ("The weather in Oceanside today is overcast with drizzle. "
                     "It is seventy two degrees and it feels like seventy three. "
                     "The high today will be eighty six degrees, with light winds.")
        await asyncio.to_thread(_synth_text, pipe, warm_text)
        print(f"[tts] warmed in {time.time()-t0:.1f}s", flush=True)
    except Exception as e:
        print(f"[tts] warm FAILED: {e}", flush=True)
    if RVC_ON:
        await asyncio.to_thread(_rvc_boot)   # trained-voice sidecar (Kokoro fallback if down)
    if OMNI_ON:
        await asyncio.to_thread(_omni_boot)  # OmniVoice clone sidecar (primary voice)

async def main():
    srv = await asyncio.start_server(handle, HOST, PORT)
    print(f"[srv] ears listening on http://{HOST}:{PORT}  (ctrl-c)", flush=True)
    # TLS mirror for iOS/iPhone (getUserMedia requires a secure context). Tablet's
    # plain-http path on 8443 is untouched. Cert: certs/server.{crt,key} (self-signed, EC).
    tls_srv = None
    try:
        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        ctx.load_cert_chain(os.path.join(CERT_DIR, "server.crt"),
                            os.path.join(CERT_DIR, "server.key"))
        tls_srv = await asyncio.start_server(handle, HOST, TLS_PORT, ssl=ctx)
        print(f"[srv] TLS mirror on https://{HOST}:{TLS_PORT} (iPhone)", flush=True)
    except Exception as e:
        print(f"[srv] TLS mirror DISABLED: {e}", flush=True)
    asyncio.create_task(warm_tts())
    async with srv:
        if tls_srv is not None:
            async with tls_srv:
                await asyncio.gather(srv.serve_forever(), tls_srv.serve_forever())
        else:
            await srv.serve_forever()

if __name__ == "__main__":
    from http import HTTPStatus
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("[srv] stopped")