tests/test_proxy.py

"""Tests for space/proxy.py turn logic (offline; mock mode + monkeypatch)."""
import os
import sys

_HERE = os.path.dirname(os.path.abspath(__file__))
_SPACE_DIR = os.path.dirname(_HERE)
if _SPACE_DIR not in sys.path:
    sys.path.insert(0, _SPACE_DIR)

import arc        # noqa: E402
import content    # noqa: E402
import proxy      # noqa: E402


def test_templated_rejoin_names_the_dj():
    assert arc.DJ_NAME in proxy._templated_text("rejoin")


def test_templated_caller_intro_nonempty():
    assert proxy._templated_text("caller_intro")


def test_song_intro_template_includes_handle():
    t = proxy._templated_text(
        "song_intro",
        {"title": "Neon Rain", "artist": "The Tuesday Ghosts", "recommended_by": "@mara"},
    )
    assert "@mara" in t


def test_song_intro_template_omits_handle_when_absent():
    t = proxy._templated_text(
        "song_intro", {"title": "Neon Rain", "artist": "The Tuesday Ghosts"}
    )
    assert "@" not in t


def test_wmo_phrase_mapping():
    assert proxy._wmo_phrase(0) == "clear skies"
    assert proxy._wmo_phrase(61) == "light rain"
    assert proxy._wmo_phrase(999) == "quiet skies"   # unknown -> safe default


def test_fmt_clock():
    assert proxy._fmt_clock("2026-06-15T02:14") == "2:14 AM"
    assert proxy._fmt_clock("2026-06-15T13:05") == "1:05 PM"
    assert proxy._fmt_clock(None) is None


def test_resolve_locale_mock(monkeypatch):
    monkeypatch.setenv("NIGHTWAVE_MOCK", "1")
    f = proxy.resolve_locale(30.27, -97.74)
    assert f["resolved"] is True
    assert f["city"] and f["temp_f"] is not None and f["sky"]


def test_local_weather_falls_back_to_fictional_on_junk(monkeypatch):
    monkeypatch.setattr(proxy, "call_brain",
                        lambda system, messages: {"text": "<sentence 1>", "mood": "warm",
                                                  "arc_cue": "none"})
    monkeypatch.setattr(proxy, "call_speak",
                        lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [],
                                                       "wtimes": [], "wdurations": []})
    out = proxy.segment_turn("local_weather", {"city": "Austin", "temp_f": 58})
    assert out["text"] in content.WEATHER     # junk brain text -> fictional fallback
    assert out["kind"] == "local_weather"


# ---------------------------------------------------------------------------
# Regression coverage for the segment_turn junk-fallback path and the two
# segment helpers server.py calls (segment_fallback, get_stalls). `_clean_dj_text`
# is the junk-detector the Task-5 local_weather fallback test relies on; cover it
# directly so that load-bearing behaviour can't silently regress.
# ---------------------------------------------------------------------------
def test_clean_dj_text_rejects_degenerate_text_as_none():
    # Empty / placeholder / angle-bracket template / too-few-letters -> None,
    # so segment_turn falls back to a clean templated line instead of junk. This
    # is the exact mechanism the Task-5 local_weather fallback test depends on.
    assert proxy._clean_dj_text("") is None
    assert proxy._clean_dj_text(None) is None
    assert proxy._clean_dj_text("<sentence 1>") is None
    assert proxy._clean_dj_text("a b c") is None
    assert proxy._clean_dj_text("Sit tight, night owls.") == "Sit tight, night owls."


def test_get_stalls_returns_cached_clip_list(monkeypatch):
    # server.py /api/stalls calls this; reset the module cache for order-independence.
    monkeypatch.setattr(proxy, "_stall_cache", None)
    clips = proxy.get_stalls()
    assert isinstance(clips, list) and len(clips) == len(proxy._STALL_LINES)
    assert all(isinstance(c, str) and c for c in clips)   # each is a base64 WAV
    assert proxy.get_stalls() is clips                    # second call is cached


def test_segment_fallback_is_never_failing_and_well_shaped():
    # server.py /api/segment uses this on any failure so the show never breaks.
    fb = proxy.segment_fallback("thought")
    assert fb["kind"] == "thought"
    assert fb["text"] in content.THOUGHTS        # templated, stage-appropriate
    assert fb["audio_b64"]                        # a real (silent) WAV bed
    for key in ("mood", "arc_cue", "words", "wtimes", "wdurations"):
        assert key in fb


# ---------------------------------------------------------------------------
# Regression (bug 2): the all-caps station name must be normalized before TTS so
# Kokoro speaks "Nightwave" instead of spelling N-I-G-H-T-W-A-V-E. _speakable was
# dropped during a proxy rebuild and call_speak sent raw text.
# ---------------------------------------------------------------------------
def test_speakable_titlecases_nightwave():
    assert proxy._speakable("You're on NIGHTWAVE 98.6") == "You're on Nightwave 98.6"
    assert proxy._speakable("night wave") == "Nightwave"
    assert proxy._speakable(None) == ""


def test_call_speak_normalizes_nightwave_before_tts(monkeypatch):
    sent = {}

    class _Resp:
        def raise_for_status(self):
            pass

        def json(self):
            return {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []}

    class _Client:
        def post(self, url, headers=None, json=None):
            sent.update(json or {})
            return _Resp()

    monkeypatch.setattr(proxy, "is_mock", lambda: False)
    monkeypatch.setattr(proxy, "_modal_url", lambda: "http://modal.test")
    monkeypatch.setattr(proxy, "_modal_headers", lambda: {})
    monkeypatch.setattr(proxy, "_get_client", lambda: _Client())
    proxy.call_speak("This is NIGHTWAVE, the last station on the dial.")
    assert "NIGHTWAVE" not in sent["text"]
    assert "Nightwave" in sent["text"]


# ---------------------------------------------------------------------------
# Regression (bug 4): the caller turn must ANSWER the caller, not echo. The
# straight station uses build_host_prompt("caller") + an answer-directive user
# turn so a 1B model doesn't parrot a question that appears verbatim in context.
# call_turn had reverted to the arc builder + raw caller_text as the user message.
# ---------------------------------------------------------------------------
def test_call_turn_answers_without_echoing_the_question(monkeypatch):
    captured = {}
    monkeypatch.setattr(proxy, "call_asr", lambda a: {"text": "are you real?"})

    def _brain(system, messages):
        captured["system"] = system
        captured["messages"] = messages
        return {"text": "Real as the rain on the glass, friend.", "mood": "warm", "arc_cue": "none"}

    monkeypatch.setattr(proxy, "call_brain", _brain)
    monkeypatch.setattr(proxy, "call_speak",
                        lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [],
                                                       "wtimes": [], "wdurations": []})
    out = proxy.call_turn("oblivious", 0, "<b64>")
    # straight-station host prompt (names the DJ), NOT the arc builder
    assert arc.DJ_NAME in captured["system"]
    # the user turn carries an answer-directive, not the bare echoed question
    user = captured["messages"][-1]["content"]
    assert "do not repeat" in user.lower()
    assert user.strip() != "are you real?"
    assert out["caller_text"] == "are you real?"
    assert out["meter_delta"] >= 14   # an identity/reality question pushes the meter


# ---------------------------------------------------------------------------
# Prompt-audit fixes: structured field labels (title:/artist:/temp:/...) must
# never reach TTS; degenerate/echo output is rejected -> templated fallback.
# ---------------------------------------------------------------------------
def test_clean_dj_text_rejects_label_leakage():
    assert proxy._clean_dj_text("title: Neon Rain, artist: The Tuesday Ghosts") is None
    assert proxy._clean_dj_text('{"text": "hi", "mood": "warm"}') is None
    assert proxy._clean_dj_text("[song]") is None
    assert proxy._clean_dj_text("temp: 58, sky: clear skies") is None
    assert proxy._clean_dj_text("Coming up next, a quiet one for the night owls.") == \
        "Coming up next, a quiet one for the night owls."


def test_call_turn_degenerate_uses_caller_fallback_not_crash(monkeypatch):
    # The model leaks labels -> _clean_dj_text returns None -> must use a clean
    # caller fallback line (the old arc.sample_line was undefined and crashed here).
    monkeypatch.setattr(proxy, "call_asr", lambda a: {"text": "you there?"})
    monkeypatch.setattr(proxy, "call_brain",
                        lambda s, m: {"text": "title: X, artist: Y", "mood": "warm", "arc_cue": "none"})
    monkeypatch.setattr(proxy, "call_speak",
                        lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [],
                                                       "wtimes": [], "wdurations": []})
    out = proxy.call_turn("oblivious", 0, "<b64>")
    assert out["text"] in content.CALLER_FALLBACKS


def test_broadcast_turn_uses_host_path_and_sanitizes(monkeypatch):
    captured = {}

    def _brain(system, messages):
        captured["system"] = system
        return {"text": "artist: X", "mood": "warm", "arc_cue": "none"}  # degenerate

    monkeypatch.setattr(proxy, "call_brain", _brain)
    monkeypatch.setattr(proxy, "call_speak",
                        lambda text, voice=arc.VOICE: {"audio_b64": "", "words": [],
                                                       "wtimes": [], "wdurations": []})
    out = proxy.broadcast_turn("oblivious", 0, None)
    assert arc.DJ_NAME in captured["system"]          # straight-station host, not arc builder
    assert out["text"] in content.THOUGHTS            # degenerate -> templated thought
    assert out["arc_cue"] == "none"


def test_extract_name_place():
    assert proxy._extract_name("Hi, I'm Mira and I'm calling in") == "Mira"
    assert proxy._extract_name("just wanted to say hi") is None
    assert proxy._extract_place("I'm driving in Austin tonight") == "Austin"
    assert proxy._extract_place("no place here") is None


def test_caller_gist():
    assert proxy._caller_gist("  I'm driving home   after a double shift ") == \
        "I'm driving home after a double shift"
    assert proxy._caller_gist("hi") is None
    assert proxy._caller_gist("") is None


def test_build_memory_patch():
    p = proxy._build_memory_patch("I'm Mira, driving home after a double shift", "tender")
    assert p["caller_name"] == "Mira" and p["topic"] and p["mood"] == "tender"
    assert proxy._build_memory_patch("hi", "warm") is None        # too short -> None
    assert proxy._build_memory_patch("I'm driving all night", "bogus")["mood"] == "warm"  # bad mood -> warm


def test_call_turn_returns_memory_patch(monkeypatch):
    monkeypatch.setattr(proxy, "call_asr",
                        lambda a: {"text": "I'm Mira, driving home after a double shift"})
    monkeypatch.setattr(proxy, "call_brain",
                        lambda s, m: {"text": "Drive safe, friend.", "mood": "tender", "arc_cue": "none"})
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    out = proxy.call_turn("oblivious", 0, "<b64>")
    assert out["queue_dedication"] is True
    assert out["memory_patch"]["caller_name"] == "Mira"
    assert out["memory_patch"]["mood"] == "tender"


def test_call_turn_no_memory_on_short_caller(monkeypatch):
    monkeypatch.setattr(proxy, "call_asr", lambda a: {"text": "hi"})
    monkeypatch.setattr(proxy, "call_brain",
                        lambda s, m: {"text": "Hey there, friend.", "mood": "warm", "arc_cue": "none"})
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    out = proxy.call_turn("oblivious", 0, "<b64>")
    assert out["queue_dedication"] is False and out["memory_patch"] is None


def test_segment_dedication_memory_aware(monkeypatch):
    captured = {}

    def _brain(system, messages):
        captured["user"] = messages[-1]["content"]
        return {"text": "This one's for the night-shift drivers, headlights home.",
                "mood": "tender", "arc_cue": "none"}

    monkeypatch.setattr(proxy, "call_brain", _brain)
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    out = proxy.segment_turn("dedication",
                             {"topic": "driving home after a double shift", "caller_name": "Mira", "mood": "tender"})
    assert "Mira" in captured["user"] and "double shift" in captured["user"]
    assert out["text"].startswith("This one's for")


def test_segment_dedication_no_memory_is_templated(monkeypatch):
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    out = proxy.segment_turn("dedication", None)          # no model call on this path
    assert out["text"].startswith("This next one's for")  # from content.DEDICATIONS


def test_segment_dedication_junk_falls_back_to_name(monkeypatch):
    monkeypatch.setattr(proxy, "call_brain", lambda s, m: {"text": "title: X", "mood": "warm", "arc_cue": "none"})
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    out = proxy.segment_turn("dedication", {"topic": "waiting up", "caller_name": "Sam"})
    assert "Sam" in out["text"]                            # _dedication_fallback weaves the name


def test_song_intro_weaves_segue(monkeypatch):
    captured = {}

    def _brain(system, messages):
        captured["user"] = messages[-1]["content"]
        return {"text": "After that caller, a soft one -- 'Neon Rain'.", "mood": "tender", "arc_cue": "none"}

    monkeypatch.setattr(proxy, "call_brain", _brain)
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    proxy.segment_turn("song_intro", {"title": "Neon Rain", "artist": "The Tuesday Ghosts",
                                      "vibe": "melancholy", "segue": "right after that caller"})
    assert "right after that caller" in captured["user"]


def test_song_intro_no_segue_when_absent(monkeypatch):
    captured = {}

    def _brain(system, messages):
        captured["user"] = messages[-1]["content"]
        return {"text": "Coming up, 'Neon Rain'.", "mood": "warm", "arc_cue": "none"}

    monkeypatch.setattr(proxy, "call_brain", _brain)
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    proxy.segment_turn("song_intro", {"title": "Neon Rain", "artist": "The Tuesday Ghosts", "vibe": "melancholy"})
    assert "through-line" not in captured["user"]


def test_fragment_user_mentions_another_station():
    u = proxy._fragment_user(None)
    assert "another station" in u and len(u) > 20


def test_segment_fragment_is_persona_free(monkeypatch):
    captured = {}

    def _brain(system, messages):
        captured["system"] = system
        return {"text": "calling all night nurses near mile marker nine", "mood": "warm", "arc_cue": "none"}

    monkeypatch.setattr(proxy, "call_brain", _brain)
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    out = proxy.segment_turn("fragment", None)
    assert arc.HOST_PERSONA not in captured["system"]     # not the Sam Dusk host persona
    assert out["text"].startswith("calling all night nurses")


def test_segment_fragment_junk_falls_back(monkeypatch):
    monkeypatch.setattr(proxy, "call_brain", lambda s, m: {"text": "title: X", "mood": "warm", "arc_cue": "none"})
    monkeypatch.setattr(proxy, "call_speak",
                        lambda t, voice=arc.VOICE: {"audio_b64": "", "words": [], "wtimes": [], "wdurations": []})
    out = proxy.segment_turn("fragment", None)
    assert out["text"] in content.FRAGMENTS               # junk -> templated bank