proxy.py

"""Server-side proxy to the Modal NIGHTWAVE engine.

Python 3.9 compatible. The browser NEVER calls Modal directly -- it calls the
Space's same-origin /api/* routes (see server.py), which call into these
functions, which in turn call Modal with the proxy-auth headers held in Space
Secrets.

Mock mode (NIGHTWAVE_MOCK=="1" or MODAL_URL unset) returns canned,
stage-appropriate data plus a short silent WAV, so the whole UI runs with no
Modal backend at all -- ideal for local dev and CI.
"""

import base64
import datetime
import io
import os
import random
import re
import struct
import threading
import wave
from typing import Any, Dict, List, Optional

import httpx

import arc
import content

# Reuse one client across calls (connection pooling). 60s timeout per the
# contract; Modal cold starts can be slow.
_TIMEOUT = 60.0
_client: Optional[httpx.Client] = None


def _get_client() -> httpx.Client:
    global _client
    if _client is None:
        _client = httpx.Client(timeout=_TIMEOUT)
    return _client


# ---------------------------------------------------------------------------
# Environment / auth
# ---------------------------------------------------------------------------
def _modal_url() -> str:
    return (os.environ.get("MODAL_URL") or "").rstrip("/")


def _modal_headers() -> Dict[str, str]:
    """Modal proxy-auth headers from Space Secrets."""
    return {
        "Modal-Key": os.environ.get("MODAL_KEY", ""),
        "Modal-Secret": os.environ.get("MODAL_SECRET", ""),
        "Content-Type": "application/json",
    }


def is_mock() -> bool:
    """True when we should serve canned data instead of hitting Modal."""
    if os.environ.get("NIGHTWAVE_MOCK") == "1":
        return True
    return not os.environ.get("MODAL_URL")


# ---------------------------------------------------------------------------
# Mock helpers
# ---------------------------------------------------------------------------
def _silent_wav_b64(seconds: float = 0.4, rate: int = 24000) -> str:
    """Build a short silent 24kHz mono PCM16 WAV in pure Python.

    Returns base64 of the WAV bytes with NO data: prefix (per the contract).
    """
    n_frames = int(seconds * rate)
    buf = io.BytesIO()
    wf = wave.open(buf, "wb")
    try:
        wf.setnchannels(1)
        wf.setsampwidth(2)  # 16-bit PCM
        wf.setframerate(rate)
        # All-zero (silent) samples.
        silence = struct.pack("<%dh" % n_frames, *([0] * n_frames))
        wf.writeframes(silence)
    finally:
        wf.close()
    return base64.b64encode(buf.getvalue()).decode("ascii")


_MOCK_BRAIN_LINE = (
    "Stay right where you are, friend -- the dial's warm and the night is long. "
    "This one's for everyone still awake out there."
)


def _mock_brain() -> Dict[str, Any]:
    return {"text": _MOCK_BRAIN_LINE, "mood": "warm", "arc_cue": "none"}


def _mock_asr() -> Dict[str, Any]:
    return {"text": "is anybody really out there tonight?"}


def _mock_speak() -> Dict[str, Any]:
    return {
        "audio_b64": _silent_wav_b64(),
        "words": [],
        "wtimes": [],
        "wdurations": [],
    }


# ---------------------------------------------------------------------------
# Real local weather/location resolution (server-side; never logs raw coords).
# Open-Meteo (current weather + sunrise + local time) and BigDataCloud (city).
# Both are free, key-less, CORS-irrelevant (server-side). All failures degrade
# to the fictional-town weather, so the show never breaks.
# ---------------------------------------------------------------------------
_WMO = {
    0: "clear skies", 1: "mostly clear", 2: "partly cloudy", 3: "overcast",
    45: "fog", 48: "freezing fog", 51: "light drizzle", 53: "drizzle",
    61: "light rain", 63: "steady rain", 65: "heavy rain", 71: "light snow",
    73: "snow", 80: "passing showers", 81: "showers", 95: "a thunderstorm",
}


def _wmo_phrase(code) -> str:
    try:
        return _WMO.get(int(code), "quiet skies")
    except (TypeError, ValueError):
        return "quiet skies"


def _fmt_clock(iso: Optional[str]) -> Optional[str]:
    """'2026-06-15T02:14' -> '2:14 AM'. None/garbage -> None."""
    if not iso:
        return None
    try:
        t = datetime.datetime.fromisoformat(iso)
        h = t.hour % 12 or 12
        ap = "AM" if t.hour < 12 else "PM"
        return "%d:%02d %s" % (h, t.minute, ap)
    except (ValueError, TypeError):
        return None


def _reverse_city(lat: float, lon: float) -> Optional[str]:
    try:
        r = _get_client().get(
            "https://api.bigdatacloud.net/data/reverse-geocode-client",
            params={"latitude": lat, "longitude": lon, "localityLanguage": "en"},
        )
        r.raise_for_status()
        j = r.json()
        return j.get("city") or j.get("locality") or j.get("principalSubdivision") or None
    except Exception:
        return None


def resolve_locale(lat: float, lon: float) -> Dict[str, Any]:
    """Real weather/time/city for a lat/lon -> facts dict, or {'resolved': False}.

    NEVER logs raw coordinates (privacy). In mock mode returns canned facts so the
    whole flow runs offline.
    """
    if is_mock():
        return {"resolved": True, "city": "Old Ferris", "temp_f": 59,
                "sky": "clear skies", "local_time": "2:14 AM", "sunrise": "6:48 AM"}
    try:
        wx = _get_client().get(
            "https://api.open-meteo.com/v1/forecast",
            params={
                "latitude": lat, "longitude": lon,
                "current": "temperature_2m,weather_code",
                "daily": "sunrise,sunset", "timezone": "auto",
                "temperature_unit": "fahrenheit", "forecast_days": 1,
            },
        )
        wx.raise_for_status()
        w = wx.json()
        cur = w.get("current", {}) or {}
        daily = w.get("daily", {}) or {}
        sunrise_list = daily.get("sunrise") or [None]
        return {
            "resolved": True,
            "temp_f": cur.get("temperature_2m"),
            "sky": _wmo_phrase(cur.get("weather_code")),
            "local_time": _fmt_clock(cur.get("time")),
            "sunrise": _fmt_clock(sunrise_list[0]),
            "city": _reverse_city(lat, lon),
        }
    except Exception:
        return {"resolved": False}


def _local_weather_user(ctx: Optional[Dict[str, Any]] = None) -> str:
    """Build the USER-turn fact string for a local_weather segment."""
    ctx = ctx or {}
    parts: List[str] = []
    if ctx.get("city"):
        parts.append("out %s way" % ctx["city"])
    if ctx.get("temp_f") is not None:
        parts.append("it is about %d degrees" % int(ctx["temp_f"]))
    if ctx.get("sky"):
        parts.append("the sky is %s" % ctx["sky"])
    if ctx.get("local_time"):
        parts.append("the clock just read %s" % ctx["local_time"])
    if ctx.get("sunrise"):
        parts.append("the sun comes up around %s" % ctx["sunrise"])
    facts = ", and ".join(parts) if parts else "it is a still, clear night out their way"
    return (
        "Tell the listener about their own sky tonight, gently. Here is what is true "
        "right now: " + facts + ". Weave it into one or two warm spoken sentences. Do "
        'not read it as a forecast, do not list it, and never write anything as "label: value".'
    )


# ---------------------------------------------------------------------------
# Raw Modal calls
# ---------------------------------------------------------------------------
def call_brain(system: str, messages: List[Dict[str, str]]) -> Dict[str, Any]:
    """POST /brain -> {"text", "mood", "arc_cue"}."""
    if is_mock():
        return _mock_brain()
    resp = _get_client().post(
        _modal_url() + "/brain",
        headers=_modal_headers(),
        json={"system": system, "messages": messages},
    )
    resp.raise_for_status()
    return resp.json()


def call_asr(audio_b64: str) -> Dict[str, Any]:
    """POST /asr -> {"text"}."""
    if is_mock():
        return _mock_asr()
    resp = _get_client().post(
        _modal_url() + "/asr",
        headers=_modal_headers(),
        json={"audio_b64": audio_b64},
    )
    resp.raise_for_status()
    return resp.json()


# All-caps "NIGHTWAVE" makes the TTS spell it out letter-by-letter; normalize to
# title case for SPEECH only (captions/UI keep the branded all-caps original).
_SPEAK_FIX = re.compile(r"\bNIGHT\s*WAVE\b", re.IGNORECASE)


def _speakable(text: Optional[str]) -> str:
    return _SPEAK_FIX.sub("Nightwave", text or "")


def call_speak(text: str, voice: str = arc.VOICE) -> Dict[str, Any]:
    """POST /speak -> {"audio_b64", "words", "wtimes", "wdurations"}."""
    if is_mock():
        return _mock_speak()
    resp = _get_client().post(
        _modal_url() + "/speak",
        headers=_modal_headers(),
        json={"text": _speakable(text), "voice": voice},
    )
    resp.raise_for_status()
    return resp.json()


# ---------------------------------------------------------------------------
# DJ-text sanitizer: a small 1B model sometimes leaks the JSON labels into the
# spoken text ("...mood: warm") or emits placeholders ("<sentence 1>"). We never
# want the DJ to SAY those, so we scrub the text before TTS and, if what's left
# is degenerate, fall back to a clean stage-appropriate canned line.
# ---------------------------------------------------------------------------
_LABEL_RE = re.compile(
    r"[\(\[]?\s*\b(?:mood|arc[_ ]?cue|title|artist|song|track|vibe|genre|by|city|town|"
    r"temp(?:erature)?|sky|weather|conditions?|forecast|time|local[_ ]?time|sunrise|sunset|"
    r"handle|requested[_ ]?by|recommended[_ ]?by|dedication|name)\b\s*[:=]\s*[^\n,;]*",
    re.IGNORECASE,
)
_PLACEHOLDER_RE = re.compile(
    r"<[^>]{0,40}>|\[[^\]]{0,40}\]|\{[^}]{0,40}\}|\bsentence\s*\d+\b", re.IGNORECASE
)


def _clean_dj_text(text: Optional[str]) -> Optional[str]:
    """Strip label leakage / markdown; return cleaned spoken text or None if junk."""
    t = text or ""
    t = _LABEL_RE.sub("", t)          # drop "mood: warm" / "title: X" / "artist: Y" runs
    t = re.sub(r"[*#`]+", "", t)       # drop markdown emphasis/headings
    t = t.strip()
    if len(t) >= 2 and t[0] in "\"'" and t[-1] in "\"'":
        t = t[1:-1].strip()            # unwrap a fully-quoted line
    t = re.sub(r"\s+", " ", t).strip()  # it's spoken: collapse whitespace
    # Degenerate? empty / placeholder / bracketed / structured key:value / echo.
    if not t or _PLACEHOLDER_RE.search(t):
        return None
    if any(c in t for c in "<>[]{}|"):
        return None
    if re.search(r"\b(?:title|artist|mood|arc_cue|temp|sky|sunrise|vibe)\b\s*[:=]", t, re.IGNORECASE):
        return None
    if t.count(":") >= 2:
        return None
    if re.search(r"\b(one or two sentences|do not repeat|in your own words|spoken sentences|stage direction)\b", t, re.IGNORECASE):
        return None
    if len(re.sub(r"[^a-zA-Z]", "", t)) < 6:
        return None
    return t


# ---------------------------------------------------------------------------
# Caller session memory (SP1): deterministic extraction from the ASR text.
# No model call, no Modal engine change -- mood is reused from the answer brain
# call; topic is the caller's gist; name/place are best-effort (often None).
# ---------------------------------------------------------------------------
_NAME_RE = re.compile(
    r"\b(?i:i'?m|i am|this is|it'?s|my name is|name'?s)\s+([A-Z][a-z]{1,18})\b"
)
_PLACE_RE = re.compile(r"\b(?i:in|from|out in|over in)\s+([A-Z][a-z]+(?:\s[A-Z][a-z]+)?)\b")


def _caller_gist(caller_text: Optional[str]) -> Optional[str]:
    t = re.sub(r"\s+", " ", (caller_text or "")).strip()
    if len(t.split()) < 3:
        return None
    return t[:140].strip()


def _extract_name(caller_text: Optional[str]) -> Optional[str]:
    m = _NAME_RE.search(caller_text or "")
    return m.group(1).strip()[:40] if m else None


def _extract_place(caller_text: Optional[str]) -> Optional[str]:
    m = _PLACE_RE.search(caller_text or "")
    return m.group(1).strip()[:40] if m else None


def _build_memory_patch(caller_text: Optional[str], mood: str) -> Optional[Dict[str, Any]]:
    topic = _caller_gist(caller_text)
    if not topic:
        return None
    return {
        "caller_name": _extract_name(caller_text),
        "place": _extract_place(caller_text),
        "topic": topic,
        "mood": mood if mood in arc.MOODS else "warm",
    }


# ---------------------------------------------------------------------------
# Pre-cached "stall" lines: instant filler audio played the moment a caller stops
# talking, while the real reply generates underneath -- so the call never has
# dead air. Synthesized once (per container) via Modal /speak, then cached.
# ---------------------------------------------------------------------------
_STALL_LINES = [
    "Mm. Good question, friend. Let me sit with that a second.",
    "That's a thing to ask at this hour. Hold on now, let me find the words.",
    "Hm. Give me a breath, caller. The night makes me slow.",
    "Let me think on that one. Stay right there with me.",
]
_stall_cache: Optional[List[str]] = None
_stall_lock = threading.Lock()


def get_stalls() -> List[str]:
    """Return a list of base64 WAV stall clips (cached after first synthesis)."""
    global _stall_cache
    if _stall_cache is not None:
        return _stall_cache
    with _stall_lock:
        if _stall_cache is not None:
            return _stall_cache
        clips: List[str] = []
        for line in _STALL_LINES:
            try:
                sp = call_speak(line)
                if sp.get("audio_b64"):
                    clips.append(sp["audio_b64"])
            except Exception:
                pass
        _stall_cache = clips
        return clips


# ---------------------------------------------------------------------------
# High-level turns (what server.py routes call)
# ---------------------------------------------------------------------------
def broadcast_turn(
    stage: str, meter: int, topic: Optional[str] = None
) -> Dict[str, Any]:
    """Produce one solo on-air broadcast turn.

    -> /api/broadcast response shape:
       {text, mood, arc_cue, audio_b64, words, wtimes, wdurations}
    """
    system = arc.build_host_prompt("on_air", {"topic": topic})
    user = (topic.strip() if (topic and topic.strip())
            else "Go on the air now with a short stretch of warm late-night patter.")
    try:
        brain = call_brain(system, [{"role": "user", "content": user}])
        text = _clean_dj_text(brain.get("text", "")) or _templated_text("thought")
        mood = brain.get("mood", "warm")
        arc_cue = brain.get("arc_cue", "none")
    except Exception:
        text, mood, arc_cue = _templated_text("thought"), "warm", "none"
    speak = call_speak(text)
    return {
        "text": text,
        "mood": mood,
        "arc_cue": arc_cue,
        "audio_b64": speak.get("audio_b64", ""),
        "words": speak.get("words", []),
        "wtimes": speak.get("wtimes", []),
        "wdurations": speak.get("wdurations", []),
    }


def call_turn(stage: str, meter: int, audio_b64: str) -> Dict[str, Any]:
    """Handle one live call-in turn.

    -> /api/call response shape:
       {caller_text, text, mood, arc_cue, audio_b64, words, wtimes,
        wdurations, meter_delta}
    """
    asr = call_asr(audio_b64)
    caller_text = asr.get("text", "")
    meter_delta = arc.detect_triggers(caller_text)

    # Straight station: the host always answers the caller in his own voice. The
    # caller's words are delivered as an ANSWER-DIRECTIVE in the user turn (NOT
    # embedded verbatim) because a 1B model under the JSON grammar tends to echo a
    # question that appears verbatim in its context.
    system = arc.build_host_prompt("caller", {"caller_text": caller_text})
    if caller_text and caller_text.strip():
        user_msg = (
            "The caller is asking about this: " + caller_text.strip()
            + "\nRespond by speaking ONLY your warm reply in your own words -- a fresh "
            "sentence that answers them. Begin your reply with a word other than the "
            'caller\'s. Do not quote or restate their words, do not write "X: Y", and do '
            "not repeat this instruction."
        )
    else:
        user_msg = "(the caller's line is too crackly to make out)"
    brain = call_brain(system, [{"role": "user", "content": user_msg}])
    text = _clean_dj_text(brain.get("text", "")) or _pick(content.CALLER_FALLBACKS)
    speak = call_speak(text)
    memory_patch = _build_memory_patch(caller_text, brain.get("mood", "warm"))
    return {
        "caller_text": caller_text,
        "text": text,
        "mood": brain.get("mood", arc.stage_default_mood(stage)),
        "arc_cue": brain.get("arc_cue", "none"),
        "audio_b64": speak.get("audio_b64", ""),
        "words": speak.get("words", []),
        "wtimes": speak.get("wtimes", []),
        "wdurations": speak.get("wdurations", []),
        "meter_delta": meter_delta,
        "memory_patch": memory_patch,
        "queue_dedication": bool(memory_patch),
    }


# ---------------------------------------------------------------------------
# Autonomous show segments (the straight station).
# LLM kinds (thought, song_intro) -> host persona + brain + sanitizer.
# Templated kinds (station_id, weather, dedication) -> content banks. All spoken.
# ---------------------------------------------------------------------------
def _pick(seq):
    return random.choice(seq) if seq else ""


def _templated_text(kind: str, ctx: Optional[Dict[str, Any]] = None) -> str:
    ctx = ctx or {}
    if kind in ("weather", "local_weather"):
        # Fictional-town weather is also the graceful fallback for local_weather.
        return _pick(content.WEATHER)
    if kind == "dedication":
        d = _pick(content.DEDICATIONS) or {"name": "someone out there",
                                           "message": "you're not as alone as it feels"}
        return "This next one's for %s -- %s." % (d["name"], d["message"])
    if kind == "song_intro":
        base = "Coming up next, '%s' by %s." % (
            ctx.get("title", "this next one"), ctx.get("artist", "a friend of the show"))
        rb = ctx.get("recommended_by")
        if rb:
            base = base[:-1] + " -- sent in by %s." % rb
        return base
    if kind == "rejoin":
        return _pick(content.REJOINS)
    if kind == "caller_intro":
        return _pick(content.CALLER_INTROS)
    if kind == "thought":
        return _pick(content.THOUGHTS)
    if kind == "fragment":
        return _pick(content.FRAGMENTS)
    return _pick(content.STATION_IDS)  # station_id (default)


# ---------------------------------------------------------------------------
# Generative song cards (SP3): model writes title+artist; server picks the vibe
# and assigns ALWAYS-VALID musical params so a card can never break the engine.
# ---------------------------------------------------------------------------
_VIBE_MUSIC = {
    "melancholy": ("minor", "rhodes", 70, ("A", "E", "D")),
    "lonely": ("minor", "music_box", 64, ("A", "E", "Bb")),
    "tender": ("major", "music_box", 68, ("C", "F", "Bb")),
    "warm": ("major", "rhodes", 78, ("C", "D", "G")),
    "cozy": ("major", "rhodes", 80, ("F", "C", "Bb")),
    "hopeful": ("lydian", "soft_saw", 80, ("D", "G", "C")),
    "nostalgic": ("major", "triangle_pluck", 74, ("C", "Bb", "G")),
    "wistful": ("dorian", "sine_pad", 72, ("C", "D", "F")),
    "bittersweet": ("minor", "rhodes", 70, ("Bb", "A", "E")),
    "jazzy": ("dorian", "rhodes", 84, ("D", "F", "G")),
    "dreamy": ("lydian", "sine_pad", 64, ("F", "D", "Eb")),
    "pensive": ("dorian", "sine_pad", 76, ("C", "E", "G")),
    "hypnotic": ("pentatonic_minor", "soft_saw", 70, ("G", "A", "D")),
    "mysterious": ("lydian", "sine_pad", 68, ("Eb", "A", "F")),
    "eerie": ("pentatonic_minor", "sine_pad", 60, ("A", "Eb", "G")),
    "ambient": ("lydian", "sine_pad", 64, ("F", "D", "C")),
    "romantic": ("minor", "rhodes", 66, ("Ab", "D", "A")),
    "gentle": ("major", "triangle_pluck", 75, ("C", "G", "D")),
    "breezy": ("major", "triangle_pluck", 88, ("G", "D", "C")),
}
_DEFAULT_MUSIC = ("major", "rhodes", 74, ("C", "D", "G"))
_MOOD_TO_VIBE = {
    "warm": "warm", "nostalgic": "nostalgic", "tender": "tender",
    "searching": "pensive", "hollow": "lonely", "uneasy": "eerie",
}


def _parse_title_artist(text):
    t = re.sub(r"\s+", " ", text or "").strip().strip("\"'")
    t = re.sub(r"^(?:title|song)\s*[:=]\s*", "", t, flags=re.IGNORECASE)
    parts = re.split(r"\s+by\s+", t, maxsplit=1, flags=re.IGNORECASE)
    if len(parts) != 2:
        return None, None
    title = parts[0].strip().strip("\"'")[:42]
    # Drop trailing model chatter ("... by The Velvet Sundays. Hope you enjoy!")
    # at the first real sentence break -- a period after >=4 letters then space --
    # so abbreviations like "St. Vincent" survive.
    artist = re.split(r"(?<=[A-Za-z]{4})\.\s", parts[1].strip().strip("\"'"), maxsplit=1)[0].strip()[:42]
    if 2 <= len(title) <= 42 and 2 <= len(artist) <= 42 and not any(c in t for c in "<>{}[]") and ":" not in title:
        return title, artist
    return None, None


def _procedural_title():
    return "%s %s" % (random.choice(content.CARD_TITLE_A), random.choice(content.CARD_TITLE_B))


def _procedural_artist():
    return "%s %s" % (random.choice(content.CARD_ARTIST_A), random.choice(content.CARD_ARTIST_B))


def _generate_title_artist(ctx, vibe):
    try:
        flavor = ""
        if ctx.get("topic"):
            flavor = " inspired by someone who said: %s" % str(ctx["topic"])[:80]
        elif ctx.get("city"):
            flavor = " for a night in %s" % ctx["city"]
        user = ("Invent a fictional late-night record with a %s feeling%s. Reply with ONLY "
                "the song title and the artist in the form: <title> by <artist>." % (vibe, flavor))
        brain = call_brain(arc.build_song_card_prompt(), [{"role": "user", "content": user}])
        title, artist = _parse_title_artist(brain.get("text", ""))
        if title and artist:
            return title, artist
    except Exception:
        pass
    return _procedural_title(), _procedural_artist()


def make_song_card(ctx=None):
    """Invent ONE fictional record. Musical params are always valid engine enums."""
    ctx = ctx or {}
    vibe = _MOOD_TO_VIBE.get(ctx.get("mood")) or random.choice(list(_VIBE_MUSIC.keys()))
    scale, timbre, tempo, keys = _VIBE_MUSIC.get(vibe, _DEFAULT_MUSIC)
    title, artist = _generate_title_artist(ctx, vibe)
    return {
        "title": title, "artist": artist, "vibe": vibe,
        "key": random.choice(keys), "scale": scale, "tempo": tempo, "timbre": timbre,
        "recommended_by": None, "generated": True,
    }


_FRAGMENT_FLAVORS = ["dedication", "numbers_weather", "station_id", "dream", "song_title"]
_FRAGMENT_HINT = {
    "dedication": "a half-heard dedication to someone you will never name",
    "numbers_weather": "a numbers-station weather reading for a place that may not exist",
    "station_id": "a fictional station identifier from a frequency that should be empty",
    "dream": "a single stray line, like a dream someone else is having out loud",
    "song_title": "a garbled song title and artist that dissolve into static",
}


def _fragment_user(ctx: Optional[Dict[str, Any]] = None) -> str:
    ctx = ctx or {}
    flavor = random.choice(_FRAGMENT_FLAVORS)
    hint = _FRAGMENT_HINT[flavor]
    extra = ""
    if flavor == "numbers_weather" and ctx.get("city"):
        extra = " somewhere near %s" % ctx["city"]
    if flavor == "dedication" and ctx.get("topic"):
        extra = " about %s" % ctx["topic"]
    return ("Speak ONE short fragment bleeding in from another station on the dial: " + hint
            + extra + ". At most a dozen words, eerie and distant, half-lost in static. "
            "Plain spoken words only, no labels.")


def _dedication_user(ctx: Optional[Dict[str, Any]] = None) -> str:
    ctx = ctx or {}
    msg = ("Read a short, warm dedication on the air for a caller. Here's what they "
           "shared: %s." % ctx.get("topic", "they're up late tonight"))
    if ctx.get("caller_name"):
        msg += " Their name is %s." % ctx["caller_name"]
    if ctx.get("place"):
        msg += " They're in %s." % ctx["place"]
    msg += (" Send the next record out to them in one or two warm spoken sentences. "
            "Do not quote them, do not label anything, and do not repeat this instruction.")
    return msg


def _dedication_fallback(ctx: Optional[Dict[str, Any]] = None) -> str:
    ctx = ctx or {}
    name = ctx.get("caller_name")
    if name:
        return "This next one's for %s -- you're not as alone as it feels tonight." % name
    return _templated_text("dedication")


def segment_turn(kind: str, ctx: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """Produce one show segment with audio."""
    ctx = ctx or {}
    mood, arc_cue = "warm", "none"
    memory_dedication = (kind == "dedication" and bool(ctx.get("topic")))
    if kind in ("thought", "song_intro", "local_weather", "fragment") or memory_dedication:
        system = arc.build_fragment_prompt() if kind == "fragment" else arc.build_host_prompt(kind, ctx)
        if kind == "fragment":
            user = _fragment_user(ctx)
        elif kind == "dedication":
            user = _dedication_user(ctx)
            mood = ctx.get("mood", "warm")
        elif kind == "song_intro":
            user = (
                "Bring in the next song on the air now. It is called %s and it is by %s "
                "-- it has a %s feel. Welcome it in with one or two warm spoken sentences "
                'in your own voice. Do not use the words "title" or "artist", do not label '
                'anything, do not write "X: Y", and do not repeat this instruction.'
                % (ctx.get("title", "this next one"),
                   ctx.get("artist", "a friend of the show"),
                   ctx.get("vibe", "late-night"))
            )
            if ctx.get("segue"):
                user = ("Open with a brief, warm half-sentence through-line -- %s -- and then "
                        % ctx["segue"]) + user[0].lower() + user[1:]
            rb = ctx.get("recommended_by")
            if rb:
                user += (
                    " A listener, %s, sent this one in -- give them a warm little shout "
                    "as you bring it in." % rb
                )
            if ctx.get("fresh"):
                user += (" Mention warmly that this record is brand new -- pressed just "
                         "tonight, never heard before.")
        elif kind == "local_weather":
            user = _local_weather_user(ctx)
        else:
            user = "Share one short late-night thought for the listeners right now."
        try:
            brain = call_brain(system, [{"role": "user", "content": user}])
            fb = _dedication_fallback(ctx) if kind == "dedication" else _templated_text(kind, ctx)
            text = _clean_dj_text(brain.get("text", "")) or fb
            mood = brain.get("mood", mood)
            arc_cue = brain.get("arc_cue", "none")
        except Exception:
            text = _dedication_fallback(ctx) if kind == "dedication" else _templated_text(kind, ctx)
    else:
        text = _templated_text(kind, ctx)
        if kind == "weather":
            mood = "nostalgic"

    speak = call_speak(text)
    return {
        "kind": kind,
        "text": text,
        "mood": mood,
        "arc_cue": arc_cue,
        "audio_b64": speak.get("audio_b64", ""),
        "words": speak.get("words", []),
        "wtimes": speak.get("wtimes", []),
        "wdurations": speak.get("wdurations", []),
    }


def segment_fallback(kind: str, ctx: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
    """Never-fail segment: templated text + a short silent WAV (bed covers it)."""
    return {
        "kind": kind,
        "text": _templated_text(kind, ctx),
        "mood": "warm",
        "arc_cue": "none",
        "audio_b64": _silent_wav_b64(),
        "words": [],
        "wtimes": [],
        "wdurations": [],
    }