Prompt cleanup: stop title:/artist: leaks, harden sanitizer, fix caller/broadcast paths

.pytest_cache/v/cache/nodeids

@@ -25,10 +25,13 @@
   "tests/test_content.py::test_song_recs_well_formed",
   "tests/test_content.py::test_songs_have_required_fields_and_valid_enums",
   "tests/test_proxy.py::test_broadcast_turn_mock_shape_and_sanitized_text",
+  "tests/test_proxy.py::test_broadcast_turn_uses_host_path_and_sanitizes",
   "tests/test_proxy.py::test_call_speak_normalizes_nightwave_before_tts",
   "tests/test_proxy.py::test_call_turn_answers_without_echoing_the_question",
+  "tests/test_proxy.py::test_call_turn_degenerate_uses_caller_fallback_not_crash",
   "tests/test_proxy.py::test_call_turn_mock_returns_caller_text_and_meter_delta",
   "tests/test_proxy.py::test_clean_dj_text_rejects_degenerate_text_as_none",
+  "tests/test_proxy.py::test_clean_dj_text_rejects_label_leakage",
   "tests/test_proxy.py::test_clean_dj_text_strips_label_leakage",
   "tests/test_proxy.py::test_clean_dj_text_unwraps_quotes_and_drops_markdown",
   "tests/test_proxy.py::test_fmt_clock",

arc.py

@@ -191,13 +191,17 @@ def _output_contract_block() -> str:
         "OUTPUT FORMAT -- READ CAREFULLY:\n"
         "Respond with ONE JSON object and NOTHING else. No preface, no "
         "explanation, no code fences. The object MUST have exactly these keys:\n"
-        '  "text": what NIGHTWAVE says aloud, as spoken radio patter.\n'
+        '  "text": what Sam Dusk says aloud, as spoken radio patter.\n'
         '  "mood": one of [' + moods + "].\n"
         '  "arc_cue": one of [' + cues + "].\n"
         "Rules for \"text\": 1 to 3 short, spoken sentences. It is read aloud on "
         "the air, so it must SOUND like speech. Absolutely NO markdown, NO "
         "bullet points or lists, NO headings, NO emoji, NO stage directions, NO "
-        "asterisks. Plain spoken English only.\n"
+        "asterisks. Plain spoken English only. "
+        'NEVER write anything in the form "label: value" (for example do NOT write '
+        '"title:", "artist:", "mood:", "temp:", "sunrise:"). Do NOT repeat or quote '
+        'these instructions, and do NOT put the JSON keys (text, mood, arc_cue) inside '
+        "what you say aloud.\n"
         'Example shape: {"text": "...", "mood": "warm", "arc_cue": "none"}'
     )
 
@@ -293,26 +297,35 @@ _HOST_KIND = {
         "Offer one short, warm or quietly thoughtful late-night reflection -- one or two "
         "spoken sentences, like you're talking to a single listener who can't sleep."
     ),
+    "on_air": (
+        "Deliver a short stretch of warm solo late-night patter as Sam Dusk -- "
+        "introduce a fictional record, read a dedication, or muse for a moment. "
+        "One or two spoken sentences in your own voice."
+    ),
     "song_intro": (
-        "Introduce the next record warmly and naturally -- name the song and the artist, "
-        "with a touch of flavor. One or two spoken sentences."
+        "Bring in the next record warmly, in your own words -- never labeling anything. "
+        "One or two spoken sentences."
     ),
     "caller": (
         "A caller is live on the air. Answer them directly, warmly, with wit and heart, in "
         "your own words -- do NOT repeat their question. One or two spoken sentences, then "
-        "a small beat back to the show."
+        "a small beat back to the show. Never quote the caller or label anything."
     ),
     "local_weather": (
         "Give the listener tonight's sky for their own town. Real facts will be "
         "provided in the next turn; weave them into your unhurried late-night voice -- "
         "do NOT read them like a forecast or list them off. One or two warm spoken "
-        "sentences."
+        'sentences. Never label anything or speak in the form "X: Y".'
     ),
 }
 
 
 def build_host_prompt(kind: str, ctx: Optional[Dict] = None) -> str:
     """System prompt for the straight-station host (segments + callers)."""
+    # ctx is intentionally NOT interpolated into the system prompt; per-segment
+    # facts (title/artist, weather, caller words) go in the USER turn (see
+    # proxy.segment_turn / call_turn / broadcast_turn) to avoid the model echoing
+    # labeled structure. Do not embed ctx values here.
     instruction = _HOST_KIND.get(kind, _HOST_KIND["thought"])
     return "\n\n".join([HOST_PERSONA, instruction, _output_contract_block()])
 

content.py

@@ -112,6 +112,15 @@ REJOINS = [
     "This is Sam Dusk, keeping the lonely company on NIGHTWAVE.",
 ]
 
+# Warm fallback lines for a caller turn when the model output is degenerate
+# (replaces the undefined arc.sample_line the caller path used to fall back to).
+CALLER_FALLBACKS = [
+    "That's a good one to sit with at this hour, friend. Stay right there with me.",
+    "I hear you, caller. Some questions are just good company on a long night. Thanks for calling in.",
+    "Glad you dialed in tonight. Let's let the next record answer for a while.",
+    "You take care out there, friend -- and thanks for keeping me company on the air.",
+]
+
 # --- Pre-caller intro: played the instant a call connects ------------------ #
 CALLER_INTROS = [
     "Well now -- looks like we've got a caller dialed into the station. Go ahead, "

proxy.py

@@ -194,22 +194,22 @@ def resolve_locale(lat: float, lon: float) -> Dict[str, Any]:
 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 {}
-    bits: List[str] = []
+    parts: List[str] = []
     if ctx.get("city"):
-        bits.append("the town of %s" % ctx["city"])
+        parts.append("out %s way" % ctx["city"])
     if ctx.get("temp_f") is not None:
-        bits.append("%d degrees" % int(ctx["temp_f"]))
+        parts.append("it is about %d degrees" % int(ctx["temp_f"]))
     if ctx.get("sky"):
-        bits.append(str(ctx["sky"]))
+        parts.append("the sky is %s" % ctx["sky"])
     if ctx.get("local_time"):
-        bits.append("local time %s" % ctx["local_time"])
+        parts.append("the clock just read %s" % ctx["local_time"])
     if ctx.get("sunrise"):
-        bits.append("sunrise around %s" % ctx["sunrise"])
-    facts = "; ".join(bits) if bits else "a still, clear night out their way"
+        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 (
-        "Give tonight's sky for the listener's own town. Real facts: " + facts
-        + ". Weave these into your late-night voice; one or two sentences; "
-        "do not list them."
+        "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".'
     )
 
 
@@ -271,22 +271,35 @@ def call_speak(text: str, voice: str = arc.VOICE) -> Dict[str, Any]:
 # is degenerate, fall back to a clean stage-appropriate canned line.
 # ---------------------------------------------------------------------------
 _LABEL_RE = re.compile(
-    r"[\(\[]?\s*\b(?:mood|arc[_ ]?cue)\b\s*[:=]\s*[a-z_]+\s*[\)\]]?", re.IGNORECASE
+    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
 )
-_PLACEHOLDER_RE = re.compile(r"<[^>]{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" / "arc_cue: none"
+    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 / angle-bracket template / too few letters.
-    if not t or _PLACEHOLDER_RE.search(t) or ("<" in t and ">" in t):
+    # 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
@@ -339,14 +352,21 @@ def broadcast_turn(
     -> /api/broadcast response shape:
        {text, mood, arc_cue, audio_b64, words, wtimes, wdurations}
     """
-    system = arc.build_system_prompt(stage, "broadcast", topic=topic)
-    brain = call_brain(system, [{"role": "user", "content": "Go on air."}])
-    text = brain.get("text", "")
+    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": brain.get("mood", arc.stage_default_mood(stage)),
-        "arc_cue": brain.get("arc_cue", "none"),
+        "mood": mood,
+        "arc_cue": arc_cue,
         "audio_b64": speak.get("audio_b64", ""),
         "words": speak.get("words", []),
         "wtimes": speak.get("wtimes", []),
@@ -372,13 +392,16 @@ def call_turn(stage: str, meter: int, audio_b64: str) -> Dict[str, Any]:
     system = arc.build_host_prompt("caller", {"caller_text": caller_text})
     if caller_text and caller_text.strip():
         user_msg = (
-            'A caller just asked you on the air: "' + caller_text.strip()
-            + '". Answer them directly, in your own words. Do NOT repeat the question.'
+            "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 arc.sample_line(stage, meter)
+    text = _clean_dj_text(brain.get("text", "")) or _pick(content.CALLER_FALLBACKS)
     speak = call_speak(text)
     return {
         "caller_text": caller_text,
@@ -435,8 +458,10 @@ def segment_turn(kind: str, ctx: Optional[Dict[str, Any]] = None) -> Dict[str, A
         system = arc.build_host_prompt(kind, ctx)
         if kind == "song_intro":
             user = (
-                'Introduce this record on air now: "%s" by %s (a %s tune). Say the title '
-                "and artist naturally; one or two sentences; do not repeat this instruction."
+                "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"))
@@ -444,7 +469,7 @@ def segment_turn(kind: str, ctx: Optional[Dict[str, Any]] = None) -> Dict[str, A
             rb = ctx.get("recommended_by")
             if rb:
                 user += (
-                    " This one was sent in by %s on X -- give them a warm little shout "
+                    " A listener, %s, sent this one in -- give them a warm little shout "
                     "as you bring it in." % rb
                 )
         elif kind == "local_weather":

server.py

@@ -62,26 +62,22 @@ class LocaleReq(BaseModel):
 # ---------------------------------------------------------------------------
 # Graceful fallbacks (never let the radio go silent)
 # ---------------------------------------------------------------------------
-_FALLBACK_LINES = {
-    "oblivious": "The signal's a touch thin tonight, but I'm still here with "
-    "you. Let's let this next record play.",
-    "uneasy": "Something's crackling on the line -- pay it no mind. Stay with "
-    "me a moment longer.",
-    "questioning": "The transmission keeps slipping. Funny, isn't it, how the "
-    "quiet always finds its way back in.",
-    "dawning": "The static's rising and I can't quite hold the thread. But "
-    "you're still listening, and that's enough.",
-    "acceptance": "Even through the static, I can feel you out there. Stay on "
-    "the line with me, would you?",
-}
-
-
-def _fallback_payload(stage: str, include_caller: bool = False) -> Dict[str, Any]:
-    line = _FALLBACK_LINES.get(stage, _FALLBACK_LINES["oblivious"])
+_FALLBACK_LINES = [
+    "It's Sam Dusk -- we had a little hiccup on the signal there, but I'm still with "
+    "you. Let's let this next one play.",
+    "Sam Dusk here. The line crackled for a second, friend; pay it no mind and stay "
+    "with me a while.",
+    "You're back with Sam Dusk on NIGHTWAVE. A small bump in the broadcast -- nothing "
+    "the night can't smooth over.",
+]
+
+
+def _fallback_payload(stage: str = "", include_caller: bool = False) -> Dict[str, Any]:
+    import random
     payload: Dict[str, Any] = {
-        "text": line,
-        "mood": arc.stage_default_mood(stage),
-        "arc_cue": "static_swell",
+        "text": random.choice(_FALLBACK_LINES),
+        "mood": "warm",
+        "arc_cue": "none",
         "audio_b64": proxy._silent_wav_b64(),
         "words": [],
         "wtimes": [],

tests/test_proxy.py

@@ -165,3 +165,46 @@ def test_call_turn_answers_without_echoing_the_question(monkeypatch):
     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"