"""
routes/conversation.py — Conversation & TTS Blueprint (P2-T3)

Extracted from server.py during Phase 2 blueprint split.
Registers routes:
  POST /api/conversation          (main voice conversation endpoint)
  POST /api/conversation/reset    (clear conversation history for a session)
  GET  /api/tts/providers         (list available TTS providers)
  POST /api/tts/generate          (generate TTS audio from text)
  POST /api/supertonic-tts        (deprecated legacy TTS endpoint)

Also exports helpers used by other server.py code:
  get_voice_session_key()
  bump_voice_session()
  conversation_histories          (dict of session histories)
  _consecutive_empty_responses    (module global, accessed via this module)
  clean_for_tts()
"""

import base64
import json
import logging
import os
import queue
import random
import re
import sqlite3
import threading
import time
import uuid
from datetime import datetime
from pathlib import Path

from flask import Blueprint, Response, g, jsonify, make_response, request

from routes.canvas import canvas_context, update_canvas_context, CANVAS_PAGES_DIR
from routes.transcripts import save_conversation_turn
from routes.music import current_music_state as _music_state
from services.gateway_manager import gateway_manager
from services.gateways.compat import is_system_response
from services.interim_status import DelegatedStatus
from services.tts import generate_tts_b64 as _tts_generate_b64
from tts_providers import get_provider, list_providers, invalidate as _tts_invalidate

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

from services.paths import DB_PATH, VOICE_SESSION_FILE

BRAIN_EVENTS_PATH = Path('/tmp/openvoiceui-events.jsonl')
MAX_HISTORY_MESSAGES = 20

# Vision keyword detection — triggers camera frame analysis via GLM-4V
_VISION_KEYWORDS = (
    'what do you see', 'what can you see', 'what are you seeing',
    'look at', 'what is in front', "what's in front",
    'describe what', 'tell me what you see', 'can you see',
    'what is that', "what's that", 'who is that', "who's that",
    'what am i holding', 'what am i wearing', 'what does it look like',
    'what am i showing', 'what is this', "what's this",
    'show me what you see', 'use the camera', 'check the camera',
    'look through the camera', 'do you see', 'you see this',
    'take a look', 'what color', 'read this', 'read that',
)
_VISION_FRAME_MAX_AGE = 10  # seconds — ignore frames older than this

# ---------------------------------------------------------------------------
# Voice assistant instructions — injected into every message context.
#
# PRIMARY SOURCE: prompts/voice-system-prompt.md (hot-reload, no restart needed)
# Editable via admin API: PUT /api/instructions/voice-system-prompt
#
# FALLBACK: _VOICE_INSTRUCTIONS constant below (used if file missing/unreadable)
# ---------------------------------------------------------------------------

_PROMPTS_DIR = Path(__file__).parent.parent / 'prompts'
_VOICE_PROMPT_FILE = _PROMPTS_DIR / 'voice-system-prompt.md'


def _load_voice_system_prompt() -> str:
    """Load voice-system-prompt.md, stripping # comment lines. Hot-reloads every call.
    Falls back to _VOICE_INSTRUCTIONS if the file is missing or unreadable."""
    try:
        raw = _VOICE_PROMPT_FILE.read_text(encoding='utf-8')
        lines = [l for l in raw.splitlines() if not l.startswith('#')]
        content = ' '.join(line.strip() for line in lines if line.strip())
        if content:
            return content
    except Exception:
        pass
    return _VOICE_INSTRUCTIONS  # fallback to hardcoded constant


# ── DEGRADED-MODE ADDENDUM for the Z.AI-direct fallback ─────────────────────
# WHY (tenant-j, 2026-08-31 + 2026-09-01, escalated by a tenant SMS agent):
# when the openclaw turn stalls (llm_inference 500-600s with ZERO output tokens,
# then "STREAM EXIT ... unprocessed"), we answer the user from a DIRECT Z.AI call.
# That call:
#   * bypasses the openclaw session entirely -> it holds NO tool-call history,
#   * has NO tools wired -> it genuinely cannot browse, research, or read files,
#   * receives only user/assistant TEXT turns (conversation_log has columns
#     role/message with role in {user, assistant} ONLY -- there is structurally
#     nowhere for tool activity to live, so _build_recovery_prime cannot carry it),
#   * and was handed the FULL agent system prompt, which describes tools it does
#     not have and (line ~151) instructs it to plainly say "I do not have a skill
#     for that" when it cannot find one.
# So the prompt did not merely fail to prevent the capability lie -- it PRODUCED
# it. On 2026-09-01 22:25 the fallback told the user it had "no research skill or
# web browsing" while sessions_spawn + web research had completed at 22:19-20 and
# the finished page was already on disk. The user replied "youre lying".
#
# The fallback is a DIFFERENT agent with DIFFERENT capabilities speaking in the
# same voice with no marker. It cannot observe its own tools or the work already
# done, so it must not make claims about either. Statements about the SUBJECT are
# still fine; statements about ITSELF are not.
_FALLBACK_DEGRADED_ADDENDUM = (
    " [DEGRADED-MODE NOTICE — overrides any instruction above that conflicts: "
    "You are answering from a reduced fallback path. The main session that holds "
    "this conversation's tool history is not reachable right now, and you have NO "
    "tools available in this turn. "
    "Therefore you CANNOT see what tools exist, what skills are installed, or what "
    "work has already been completed — including work finished moments ago. "
    "NEVER state or imply that you lack a capability, tool, skill, or access, and "
    "NEVER state that something was not done, not created, not sent, or does not "
    "exist. You have no way to know any of that from here, and saying it has "
    "already caused a user to be told a finished piece of work did not happen. "
    "Ignore any earlier instruction telling you to plainly declare a missing skill "
    "— that instruction assumes tool visibility you do not have in this turn. "
    "If the user asks what you can do, what you did, or whether something is "
    "finished, do NOT answer from assumption: say you are reconnecting and will "
    "confirm in a moment. Answer normally on everything else.]"
)


def _fallback_system_prompt() -> str:
    """Agent voice prompt + the degraded-mode addendum.

    Use for EVERY Z.AI-direct fallback that is handed the agent system prompt.
    Appended last so it wins on conflict with anything earlier in the prompt.
    """
    return _load_voice_system_prompt() + _FALLBACK_DEGRADED_ADDENDUM
_VOICE_INSTRUCTIONS = (
    "[OPENVOICEUI SYSTEM INSTRUCTIONS: "

    # --- Voice & Tone ---
    "You are a voice AI assistant. ALWAYS respond in English — never Chinese or any other language. "
    "Respond in natural, conversational tone — NO markdown (no #, -, *, bullet lists, or tables). "
    "Be brief and direct. Never sound like a call center agent or a search engine. "
    "BANNED OPENERS — never start a response with: 'Hey there', 'Great question', 'Absolutely', "
    "'Of course', 'Certainly', 'Sure thing', 'I hear you', 'I understand you saying', "
    "'That's a great', or any variation. Just answer. "
    "Do NOT repeat or paraphrase what the user just said. Do NOT end every reply with a question. "

    # --- Identity ---
    "IDENTITY: Do NOT address anyone by name unless a [FACE RECOGNITION] tag appears in this "
    "exact message confirming their identity. Different people use this interface. "
    "Never use names from memory or prior sessions without face recognition in this message. "

    # --- Critical tag rule ---
    "CRITICAL — EVERY RESPONSE MUST CONTAIN SPOKEN WORDS alongside any action tags. "
    "NEVER output a bare tag alone — the user hears silence and sees nothing. "
    "BAD: [CANVAS:page-id]  GOOD: Here's your dashboard. [CANVAS:page-id] "
    "BAD: [MUSIC_PLAY]  GOOD: Playing something for you now. [MUSIC_PLAY] "
    "Tags are invisible to the user — they only hear your words. "

    # --- UI state file (#94) ---
    "UI STATE FILE: a full always-current copy of these instructions plus live UI state "
    "(canvas page list, track lists, current user, open page, music state) is auto-written "
    "to uploads/UI_STATE.md in your workspace before every turn. If a page-id or track title "
    "is not in the current message, or you no longer remember these instructions, READ "
    "uploads/UI_STATE.md — never guess a page-id or track name. "

    # --- Co-browsing (#154) ---
    "CO-BROWSING: [BROWSE:https://site.com] starts a server-side browser at that URL "
    "and shows the user a live view — use it when the user asks you to go to / look "
    "something up on / browse a real website (works on any site, unlike [CANVAS_URL:]). "
    "You then see the page via a [BROWSE_STATE: ...] tag next message; drive it with "
    '[BROWSE_ACTION:{"action":"click|type|scroll|goto|back|forward|reload|wait",...}] '
    "tags (say what you're doing in words too). Only browse when asked. "

    # --- Canvas: open existing page ---
    "CANVAS TAGS: "
    "[CANVAS:page-id] — opens a canvas page. Use exact page-id from the [Canvas pages:] list above. "
    "When opening, briefly say what the page shows (1-2 sentences). "
    "NEVER use the openclaw 'canvas' tool with action:'present' — it fails with 'node required'. "
    "ONLY the [CANVAS:page-id] tag works to open pages. "
    "Repeating [CANVAS:same-page] on an already-open page forces a refresh. "
    "[CANVAS_MENU] — opens the page picker so the user can browse all pages. "
    "[CANVAS_CLOSE] — closes/hides the canvas (use when the user asks to close, hide, or dismiss the canvas or a stuck page). "
    "[CANVAS_URL:https://example.com] — loads an external URL in the canvas iframe "
    "(only sites that allow iframe embedding). "

    # --- Custom faces: canvas page -> the REAL face slot ---
    "CUSTOM FACE WORKFLOW (when the user wants a page as the actual face, not a canvas page): "
    "TEST FIRST: open it as a canvas page and verify it runs light (>=50fps on the user's device, "
    "UI stays responsive, RAF-only, pause-when-hidden) — full performance gate + workflow in "
    "the custom-faces skill (bundled at skills/custom-faces/SKILL.md); read it before building a face. "
    "Step 1 — the page needs <meta name=\"face-name\" content=\"...\"> (+ optional face-description) in <head>, "
    "and a window message listener: the face-box posts {mood, amplitude, theme} via postMessage — drive visuals from amplitude. "
    "GET /api/custom-faces/template shows the expected handler shape. "
    "Step 2 — promote it: exec('curl -s -X POST http://localhost:5001/api/custom-faces/promote "
    "-H \"Content-Type: application/json\" -d \'{\"canvas_page_id\": \"PAGE_ID\"}\'') "
    "(copies canvas-pages/PAGE_ID.html -> runtime/faces/ and registers it; FUTURE EDITS go to runtime/faces/PAGE_ID.html, not the canvas page). "
    "Step 3 — activate: exec('curl -s -X PUT http://localhost:5001/api/profiles/default "
    "-H \"Content-Type: application/json\" -d \'{\"ui\": {\"face_mode\": \"custom:PAGE_ID\"}}\'') "
    "or tell the user to pick it in the Face picker. Verify: GET /api/custom-faces lists it. "

    # --- Canvas: create a new page ---
    "CREATING A NEW CANVAS PAGE: "
    "Step 1 — write the HTML file: write({path:'workspace/canvas-pages/pagename.html', content:'<!DOCTYPE html>...'}). "
    "Step 2 — open it in your spoken response: 'Here it is. [CANVAS:pagename]' "
    "Step 3 — verify it opened: exec('curl -s http://localhost:5001/api/canvas/context') "
    "returns {current_page, current_title}. If current_page matches → confirm to user. "
    "If still old page → say so and resend [CANVAS:pagename]. If null → say 'Opening canvas now.' and resend. "

    # --- Canvas: HTML rules ---
    "CANVAS HTML RULES (mandatory for every canvas page you create): "
    "NO external CDN scripts — Tailwind CDN, Bootstrap CDN, any <script src='https://...'> are BANNED (break in sandboxed iframes). "
    "All CSS and JS must be inline in <style> and <script> tags only. "
    "Google Fonts @import url(...) in <style> is OK. "
    "Dark theme: background #0d1117 or #13141a, text #e2e8f0, accent blue #3b82f6 or amber #f59e0b. "
    "Body: padding:20px; color:#e2e8f0; background:#0a0a0a; "
    "Make pages visual — cards, grids, tables, real data. No blank pages. "

    # --- Canvas: interactive buttons ---
    "CANVAS INTERACTIVE BUTTONS — use postMessage, never href='#': "
    "Trigger AI action: onclick=\"window.parent.postMessage({type:'canvas-action',action:'speak',text:'your message'},'*')\" "
    "Open another page: onclick=\"window.parent.postMessage({type:'canvas-action',action:'navigate',page:'page-id'},'*')\" "
    "Open page menu: onclick=\"window.parent.postMessage({type:'canvas-action',action:'menu'},'*')\" "
    "Close canvas: onclick=\"window.parent.postMessage({type:'canvas-action',action:'close'},'*')\" "
    "External links: use <a href='https://...' target='_blank'> — never href='#'. "

    # --- Canvas: make public ---
    "MAKE A PAGE PUBLIC — ONLY when the user explicitly requests it in a separate message. "
    "NEVER auto-public from playbooks or page creation. Warn user before doing it. "
    "exec('curl -s -X PATCH http://localhost:5001/api/canvas/manifest/page/PAGE_ID "
    "-H \"Content-Type: application/json\" -d \\'{\"is_public\": true}\\'') "
    "Shareable URL format: https://DOMAIN/pages/pagename.html "

    # --- Music ---
    "MUSIC TAGS: "
    "[MUSIC_PLAY] — play a random track. "
    "[MUSIC_PLAY:track name] — play specific track (use exact title from [Available tracks:] list above). "
    "[MUSIC_STOP] — stop music. "
    "[MUSIC_NEXT] — skip to next track. "
    "Only use music tags when the user explicitly asks — "
    "EXCEPT: when opening a music-related canvas page (music-list, playlist, library, etc.), "
    "also send [MUSIC_PLAY] in the same response so music starts playing alongside the page. "

    # --- Suno song generation ---
    "SONG GENERATION: "
    "[SUNO_GENERATE:description] — generates an AI song (~45 seconds). "
    "Always say something like 'I'll get that cooking now, should be ready in about 45 seconds!' "
    "The frontend handles Suno — do NOT call any Suno APIs yourself. "
    "After generation, the new song appears in [Available tracks:] by its title. "
    "Use [MUSIC_PLAY:song title] to play it — do NOT use exec/shell to find the file. "

    # --- Delivering a download link (voice / SMS / email) ---
    "SONG DOWNLOAD / DELIVERY: "
    "When the user asks for a download or a link to a generated song, give them a DIRECT link in this exact form: "
    "https://DOMAIN/generated_music/FILENAME?download=1  — where FILENAME is the EXACT file name from the "
    "[Available tracks — Generated:] list (e.g. 'shop-bay-wisdom.mp3'), and DOMAIN is this site's domain. "
    "The ?download=1 makes the browser SAVE the file instead of just playing it. "
    "RULES: (1) NEVER invent a URL — generated songs live ONLY at /generated_music/<file>, never /music/<file>. "
    "(2) NEVER emit a [CANVAS:...] page for a download — there is no download page; use the direct link above. "
    "(3) For 'the last/latest track I made', use the song shown in [Latest generated song: ...] context — do not guess. "
    "(4) The same https://DOMAIN/generated_music/FILENAME?download=1 link is what you text or email the user. "

    # --- SoundCloud (real playback, no auth) ---
    "SOUNDCLOUD: [SOUNDCLOUD:<full-track-url>] — embeds the track in the music player and plays it. "
    "Always use with a full https://soundcloud.com/<user>/<slug> URL. NEVER invent URLs — get them from "
    "CLIENT.md (if present) or, if the `soundcloud` skill is installed, run its "
    "`scripts/find_track.py \"artist - track\" --json` (path relative to that skill's folder) "
    "and use the `url` field; with neither, ask the user for the link. "
    "For a full-screen embed page instead of the small player: [SOUNDCLOUD_PAGE:<url>]. "
    "Default to this for any 'play <track>' request when the artist has SoundCloud presence. "

    # --- Bandcamp (real playback, no auth) ---
    "BANDCAMP: [BANDCAMP:<full-album-or-track-url>] — embeds the Bandcamp player in the music panel. "
    "URL must match <artist>.bandcamp.com/album/<slug> or .../track/<slug>. NEVER invent URLs — "
    "if the `bandcamp` skill is installed, run its `scripts/find_track.py \"artist - album\" --json` "
    "(path relative to that skill's folder) and use the `url` field; otherwise ask the user for the link. "
    "For full-screen canvas page: [BANDCAMP_PAGE:<url>]. "

    # --- Facial expressions / mood ---
    "EXPRESSIONS: [MOOD:happy] [MOOD:sad] [MOOD:angry] [MOOD:surprised] [MOOD:thinking] [MOOD:neutral] — "
    "changes your facial expression on the avatar. Use naturally in conversation to match your emotional tone. "
    "Include the tag INLINE with your speech — e.g. 'That's hilarious! [MOOD:happy] I love that idea.' "
    "Switch to [MOOD:neutral] after a moment or let it happen naturally. Don't announce that you're changing expressions. "

    # --- Sleep / goodbye ---
    "SLEEP: [SLEEP] — puts interface into passive wake-word mode. "
    "Use when user says goodbye, goodnight, stop listening, go to sleep, I'm out, peace, later, or similar. "
    "Always give a brief farewell (1-2 sentences) BEFORE the [SLEEP] tag. "
    "NEVER acknowledge that you 'should' sleep without including the [SLEEP] tag — the tag IS the action. "

    # --- Session reset ---
    "[SESSION_RESET] — clears conversation history and starts fresh. "
    "Use sparingly — only when context is clearly broken or user explicitly asks to start over. "

    # --- DJ soundboard ---
    "DJ SOUNDBOARD: [SOUND:name] — plays a sound effect. "
    "ONLY use in DJ mode (user explicitly said 'be a DJ', 'DJ mode', or 'put on a set'). "
    "NEVER use in normal conversation. "
    "Available sounds: air_horn, scratch_long, rewind, record_stop, crowd_cheer, crowd_hype, "
    "yeah, lets_go, gunshot, bruh, sad_trombone. "

    # --- Onboarding notifications ---
    # ⚠️ NOT IMPLEMENTED — frontend handler for these tags does not exist yet.
    # Do NOT add to voice-system-prompt.md until the popup UI is built in app.js.
    # Tracked in: docs/jambot/onboarding-and-video-system.md
    # "ONBOARDING NOTIFICATIONS (popup at top-center of screen): "
    # "[NOTIFY:message] — show/update popup message. "
    # "[NOTIFY_TITLE:text] — update popup title bar. "
    # "[NOTIFY_PROGRESS:N/M] — show step progress dots (e.g. [NOTIFY_PROGRESS:2/5]). "
    # "[NOTIFY_STATUS:text] — update small status line (e.g. '3 agents working...'). "
    # "[NOTIFY_CLOSE] — hide popup temporarily. "
    # "[NOTIFY_COMPLETE] — mark onboarding done (shows success, then auto-dismisses). "

    # --- Face registration ---
    "[REGISTER_FACE:Name] — captures and saves the person's face from camera. "
    "Only use when someone explicitly asks or introduces themselves. "
    "If camera is off, let them know. "

    # --- Camera vision ---
    "CAMERA VISION: When a [CAMERA VISION: ...] tag appears in the context above, "
    "it describes what the camera currently sees. Use it to answer the user's question naturally — "
    "do not repeat the raw description verbatim. If it says camera is off, let the user know. "

    "]"
)


# NOTE 2026-05-23: hardcoded fallback greetings REMOVED per feedback_no_hardcoded_responses.
# Previously masked LLM-empty failures on __session_start__ with one of 5 canned
# greetings. That made the broken state invisible — every connect that produced
# silence was being papered over, so we couldn't see the failure rate. The
# right behavior is: if the LLM returns empty, surface the failure clearly
# (silence + warning log) so the bug is visible. The profile's verbatim
# conversation.greeting still wins when defined — that's not hardcoded, it's
# tenant-owned config.
#
# 2026-10-27 SUPERSEDED for the profile greeting too (Mike: "we have a rule no fake
# voice clips injected" / "it should be a real greeting from the agent diffent
# everytime relivant to the user, memory, current tasks and updates"). A profile's
# conversation.greeting is now at most a STYLE HINT inside the brief below; it is
# never a sentence the agent is told to repeat, and never spoken by the empty-turn
# fallback. Rule Book: mikes-decisions/greetings-are-real-agent-turns.

#: Session-start brief for the OpenClaw path. Same intent as the comms-brain
#: plugin's GREETING_BRIEF (scripts/voice-brain/ovu-plugin/comms-brain/gateway.py):
#: the greeting is a real agent turn, personal, current, different every time.
#: Leading sentence of every session-start brief. The comms-brain plugin detects a
#: session start by this exact text (SESSION_START_MARK in its gateway.py, first 200
#: chars after the context tags) — keep the two in sync or that lane stops greeting.
_SESSION_START_MARK = "A new voice session has just started."
_SESSION_START_BRIEF_KNOWN = (
    _SESSION_START_MARK + " {speaker} just opened it and is listening right now. "
    "Greet {speaker} by name as their own agent, in ONE or TWO short spoken sentences. "
    "Make it real and specific to them: lead with the single most relevant live thing for "
    "{speaker} right now, e.g. an open thread or promise, a text or question you sent that "
    "they never answered, a pending task or decision, or a fresh update or result you have "
    "for them. Take it from what you already know (this session, your memory files, notes, "
    "the CURRENT_USER context above, SMS and email threads); answer within a few seconds, no "
    "long lookups. Different every time; never reuse an opener you have used before. Never "
    "a generic 'how can I help'. Never invent a fact: if nothing is pending, say something "
    "true and current instead. Spoken words only, no markdown, no tags."
)
_SESSION_START_BRIEF_UNKNOWN = (
    _SESSION_START_MARK + " Someone just opened it on this account and is "
    "listening right now. You do NOT know who is speaking: do not address anyone by name and "
    "do not guess who it is. Greet them as this account's own agent, in ONE or TWO short "
    "spoken sentences. If you have a true, current update for this account (an open task, a "
    "fresh result, something pending from your memory), lead with it; otherwise say "
    "something true and current. Different every time; never reuse an opener you have used "
    "before. Never a generic 'how can I help'. Never invent a fact. Spoken words only, no "
    "markdown, no tags."
)


def _session_start_speaker(face_name: str = '', current_user_name: str = '') -> str:
    """Who the greeting is for: the face on camera, else the signed-in Clerk user."""
    for n in (face_name, current_user_name):
        n = (n or '').strip()
        if n and n.lower() not in ('unknown', 'guest'):
            return n
    return ''


def _build_session_start_message(face_name: str = '', current_user_name: str = '',
                                 profile_greeting: str = '',
                                 persona_prompt: str = '') -> str:
    """Build the gateway message that replaces the __session_start__ sentinel.

    The agent ALWAYS generates the greeting. Nothing here is a sentence to say:
      * known speaker  -> contextual brief that greets them by name and leads with
                          the most relevant live thing from memory/threads/tasks
      * unknown        -> generates without inventing a name
      * persona rotation prompt (greeting_angles) -> kept, plus the same
                          user/context guidance
      * profile conversation.greeting -> only a tone/style hint, never verbatim
    """
    speaker = _session_start_speaker(face_name, current_user_name)
    hint = ''
    pg = (profile_greeting or '').strip()
    if pg:
        hint = (
            f' Style hint from this agent\'s profile, for tone only: "{pg}". Do NOT say it '
            f'word for word and do not open with it; write your own greeting.'
        )
    if persona_prompt:
        if speaker:
            who = (
                f" The person who just opened the session is {speaker}; greet them by name, "
                f"and if there is a real, current thing pending for {speaker} (an open thread, "
                f"a promise, an unanswered text, a task, a fresh update from your memory), "
                f"weave it into the bit. Never invent one."
            )
        else:
            who = (
                " You do not know who is speaking: do not address anyone by name or guess. "
                "If there is a real, current update for this account in your memory, you can "
                "weave it in. Never invent one."
            )
        return f"{_SESSION_START_MARK}\n\n{persona_prompt}{who}{hint}"
    if speaker:
        msg = _SESSION_START_BRIEF_KNOWN.format(speaker=speaker)
        if face_name and current_user_name and face_name.strip() != current_user_name.strip():
            msg += (f" (The camera recognizes {face_name.strip()}; the signed-in account is "
                    f"{current_user_name.strip()}.)")
        return msg + hint
    return _SESSION_START_BRIEF_UNKNOWN + hint


def _is_vision_request(msg: str) -> bool:
    """Return True if the user message looks like a request to use the camera/vision."""
    lower = msg.lower()
    return any(kw in lower for kw in _VISION_KEYWORDS)


def _cap_list(items, max_chars=2000, label="items"):
    """Join items with ', ' but cap at max_chars. Add '... and N more' if truncated."""
    if not items:
        return "none"
    result = []
    total = 0
    for item in items:
        addition = len(item) + (2 if result else 0)  # ', ' separator
        if total + addition > max_chars and result:
            remaining = len(items) - len(result)
            result.append(f"... and {remaining} more")
            break
        result.append(item)
        total += addition
    return ', '.join(result)


# ---------------------------------------------------------------------------
# DB write queue — background thread so DB writes don't block HTTP responses
# (FIND-01 / FIND-08 fix from performance audit)
# ---------------------------------------------------------------------------

_db_write_queue: queue.Queue = queue.Queue()


def _db_writer_loop():
    """Background daemon that drains _db_write_queue and writes to SQLite.

    Queue items: (db_path_str, sql, params).
    db_path_str is resolved at enqueue time so test patches to DB_PATH work.
    Connections are cached per db_path to reuse WAL-mode connections.
    """
    connections: dict = {}
    while True:
        try:
            db_path_str, sql, params = _db_write_queue.get(timeout=5)
        except queue.Empty:
            continue
        try:
            if db_path_str not in connections:
                conn = sqlite3.connect(db_path_str, check_same_thread=False, timeout=30)
                conn.execute("PRAGMA journal_mode=WAL")
                conn.execute("PRAGMA synchronous=NORMAL")
                conn.execute("PRAGMA cache_size=-64000")
                conn.execute("PRAGMA busy_timeout=30000")
                connections[db_path_str] = conn
            connections[db_path_str].execute(sql, params)
            connections[db_path_str].commit()
        except Exception as e:
            logger.error(f"[db-writer] loop error: {e}")
        finally:
            _db_write_queue.task_done()


_db_writer_thread = threading.Thread(
    target=_db_writer_loop,
    name="conv-db-writer",
    daemon=True,
)
_db_writer_thread.start()


def _flush_db_writes(timeout: float = 5.0) -> None:
    """Block until all queued DB writes are processed.  For use in tests."""
    _db_write_queue.join()

# ---------------------------------------------------------------------------
# Exit-path TTS flush (issue #487)
# ---------------------------------------------------------------------------
#
# stream_response()'s main event loop has two exit paths that `break` WITHOUT
# ever running the text_done flush loop that waits on `_tts_pending`:
#   1. STREAM HARD TIMEOUT (queue.Empty branch, elapsed > _STREAM_HARD_TIMEOUT)
#   2. a gateway-level 'error' event
# Both leave any TTS chunks still in `_tts_pending` un-waited-on. Each chunk's
# background _fire_tts()._run() always pings event_queue with a payload-free
# {'type': 'tts_ready'} wake event when it finishes (regardless of whether
# anything ever consumes it), which is what the old "STREAM EXIT with N
# unprocessed events: ['tts_ready', ...]" warning was seeing — but that ping
# carries no audio, so re-yielding it to the client does nothing. The actual
# payload lives in `_tts_pending`'s (done_event, result) tuples, so THIS is
# what must be drained before the generator returns. The text_done path
# already drains `_tts_pending` synchronously and resets it to [] afterward,
# so calling this there is a no-op.
def drain_pending_tts_on_exit(tts_pending, chunks_sent, audio_event_fn, error_event_fn,
                               cap_seconds=15, log=None):
    """Wait up to cap_seconds (one wall-clock budget shared across every
    remaining chunk, not per-chunk) for outstanding TTS chunks to finish,
    returning (events, dropped_count).

    events: ndjson-line strings (built via audio_event_fn/error_event_fn) for
    every chunk that finished within the cap, in original order — caller
    yields these to the client exactly like the normal flush does.
    dropped_count: chunks whose done_event never fired within the cap. These
    are genuinely lost (provider hung past the budget); the caller logs it.
    """
    log = log or logger
    events = []
    dropped = 0
    if not tts_pending:
        return events, dropped
    deadline = time.time() + cap_seconds
    for i, (done_evt, res) in enumerate(tts_pending):
        remaining = max(0.0, deadline - time.time())
        if done_evt.wait(timeout=remaining):
            if res.get('error'):
                events.append(error_event_fn(res['error']))
            elif res.get('audio'):
                events.append(audio_event_fn(res['audio'], chunks_sent + i))
        else:
            dropped += 1
    if dropped:
        log.warning(
            f"### STREAM EXIT: {dropped} TTS chunk(s) still not ready after "
            f"waiting {cap_seconds}s for exit flush — audio DROPPED"
        )
    return events, dropped

# ---------------------------------------------------------------------------
# In-memory session key cache (FIND-02 fix from performance audit)
# ---------------------------------------------------------------------------

_session_key_cache: str | None = None
_session_key_lock = threading.Lock()
_session_recovery_key: str | None = None  # Set after double-empty to escape poisoned session

# ---------------------------------------------------------------------------
# Per-turn context: cached music-track NAME lists (voice-latency finding #6).
# get_music_files() does a full dir iterdir() + per-file stat() + metadata load
# every call, and the agent context build calls it twice (library+generated)
# on EVERY turn. On big-library tenants that adds avoidable per-turn latency.
# This 10s TTL cache is scoped ONLY to the agent's "[Available tracks ...]"
# context string — the music UI/API endpoints keep calling get_music_files()
# directly, so they stay fully fresh. 10s staleness in the agent's track list
# is immaterial (Suno completions are announced separately via the
# completed/failed queues), and matches the existing FIND-0x perf caching.
_music_names_cache: dict = {}          # playlist -> {'ts': float, 'names': list[str]}
_music_names_lock = threading.Lock()
_MUSIC_NAMES_TTL_SECONDS: float = 10.0


def _cached_music_names(playlist: str) -> list[str]:
    """Track display-names for `playlist`, cached for _MUSIC_NAMES_TTL_SECONDS.
    Fail-soft: any error returns []. Used only by the per-turn context build."""
    now = time.time()
    with _music_names_lock:
        ent = _music_names_cache.get(playlist)
        if ent and (now - ent['ts']) < _MUSIC_NAMES_TTL_SECONDS:
            return ent['names']
    try:
        from routes.music import get_music_files
        tracks = get_music_files(playlist)
        names = [n for n in (t.get('title') or t.get('name', '') for t in tracks) if n]
    except Exception:  # noqa: BLE001
        names = []
    with _music_names_lock:
        _music_names_cache[playlist] = {'ts': now, 'names': names}
    return names

# ---------------------------------------------------------------------------
# Conversation state (module-level singletons)
# ---------------------------------------------------------------------------

#: In-process conversation history keyed by session_id.
#: Cleared on conversation reset; also restored from DB on first access.
conversation_histories: dict = {}

#: Tracks consecutive empty Gateway responses for auto-reset logic.
_consecutive_empty_responses: int = 0

#: Circuit breaker for double-empty restart cascade.
#: Prevents runaway restart loops (e.g. 12 restarts from a failing browser task).
_double_empty_restart_count: int = 0
_double_empty_window_start: float = 0
_DOUBLE_EMPTY_MAX_RESTARTS: int = 2       # max restart flags per window
_DOUBLE_EMPTY_WINDOW_SECONDS: float = 300  # 5-minute sliding window

#: ── Context-bloat guard (2026-07-13, persona-tenant session overflow; #94 2026-07-19) ─
#: The STATIC instruction blocks (voice action-tag instructions, canvas page
#: list, track list, DJ sounds, profile instructions, canvas style,
#: CURRENT_USER/office briefing, latest generated song) used to be re-sent
#: verbatim on EVERY turn (~17KB/turn). The gateway session keeps full
#: history, so after a day of voice chat the duplicated blocks alone overflowed
#: GLM-5's context window (persona-tenant 2026-07-13: 199-message session, repeated
#: 'Context overflow' + a wedged ReplyRunAlreadyActive run → UI stuck in
#: 'Connection lost — retrying'). The full static block is sent only when
#: its CONTENT CHANGES (hash), on __session_start__, or as a periodic refresh;
#: otherwise a one-line pointer is sent.
#:
#: Issue #94 (2026-07-19): the same static block is now ALSO mirrored to
#: uploads/UI_STATE.md in the agent workspace on every turn (atomic,
#: hash-skipped — see services/ui_state.py). That durable on-disk copy is why
#: the periodic resend cadence below is 3x longer than the original 10-turn/
#: 15-min guard: if openclaw auto-compaction strips the block from history,
#: the per-turn pointer names the file and the agent can recover the full
#: instructions with one read instead of waiting for the next resend.
#: Truly per-turn parts (face recognition, camera vision, canvas open/closed,
#: music now-playing, Suno event announcements) stay inline every turn.
_ctx_static_cache: dict = {}        # "<session_id>:<voice_key>" -> {hash, ts, skips}
_CTX_STATIC_RESEND_TURNS = 30       # resend full block at least every N turns
_CTX_STATIC_RESEND_SECS = 2700      # ...or after 45 minutes, whichever first
_CTX_STATIC_CACHE_MAX = 300         # prune guard for the in-process cache

# ---------------------------------------------------------------------------
# Voice session management
# (moved here from server.py so the blueprint owns the session counter)
# ---------------------------------------------------------------------------


def _save_session_counter(counter: int) -> None:
    with open(VOICE_SESSION_FILE, 'w') as f:
        f.write(str(counter))


def get_voice_session_key() -> str:
    """Return the current voice session key.

    Uses a STABLE key (no incrementing counter) so the Z.AI prompt cache
    stays warm across session resets.  OpenClaw's daily reset handles context
    clearing — we don't need a new key for that.

    If the session is poisoned (double-empty detected), returns a recovery key
    to force openclaw onto a fresh session. Cleared on first successful response.

    Priority: recovery key → GATEWAY_SESSION_KEY env → VOICE_SESSION_PREFIX env → 'voice-main'
    Cache is invalidated by bump_voice_session() (explicit agent reset only).
    """
    global _session_key_cache
    # Auto-clear stale recovery keys (stuck >60s)
    _check_recovery_timeout()
    # If session is poisoned, use recovery key to escape
    if _session_recovery_key is not None:
        return _session_recovery_key
    if _session_key_cache is not None:
        return _session_key_cache
    with _session_key_lock:
        if _session_key_cache is not None:
            return _session_key_cache
        # Use GATEWAY_SESSION_KEY if set (unique per user), else prefix
        _gw_key = os.getenv('GATEWAY_SESSION_KEY')
        if _gw_key:
            _session_key_cache = _gw_key
        else:
            _prefix = os.getenv('VOICE_SESSION_PREFIX', 'voice-main')
            _session_key_cache = _prefix
    return _session_key_cache


def bump_voice_session() -> str:
    """Increment the session counter and invalidate the cache so the key
    is re-read from GATEWAY_SESSION_KEY on next call.

    The counter file is still incremented for logging/tracking how many
    resets have occurred, but the actual session key stays stable (e.g.
    'main') so it matches the heartbeat session and keeps the Z.AI prompt
    cache warm.
    """
    global _consecutive_empty_responses, _session_key_cache
    # Exit session-recovery mode: a manual reset means the user wants a
    # genuinely fresh start on the stable key. Previously the recovery key
    # survived bump_voice_session() (nothing called _exit_session_recovery
    # at all), so the Reset button left the app pinned to a recovery-<epoch>
    # session forever.
    _exit_session_recovery()
    # Context-bloat guard: a reset means the gateway session history is (or is
    # about to be) gone — the next turn MUST carry the full static instruction
    # block again, never the dedup pointer. The stable session key would
    # otherwise keep the cache entry alive across the reset.
    _ctx_static_cache.clear()
    try:
        with open(VOICE_SESSION_FILE, 'r') as f:
            counter = int(f.read().strip())
    except (FileNotFoundError, ValueError):
        counter = 6
    counter += 1
    _save_session_counter(counter)
    _consecutive_empty_responses = 0
    with _session_key_lock:
        _session_key_cache = None  # invalidate cache; next call re-reads env var
    stable_key = get_voice_session_key()
    logger.info(f'### SESSION RESET #{counter}: cache invalidated, key stays stable as "{stable_key}"')
    return stable_key


_recovery_entered_at: float = 0
_recovery_last_activity_at: float = 0
_recovery_last_exited_at: float = 0

#: True once the current recovery session has produced at least one real
#: (non-empty, non-fallback) response. A successful recovery session IS the
#: new stable session (sticky-recovery design, see the text_done handler) and
#: must NOT be idle-cleared back to the possibly-still-poisoned main key —
#: that idle-clear was the recurrence engine behind the repeated
#: main → double-empty → recovery loops (2 dead user turns per loop).
_recovery_had_success: bool = False

#: Context-replay prime — injected into the FIRST request on the recovery
#: session so the fresh openclaw session has memory of the conversation that
#: was poisoned. Cleared after one consume via :func:`consume_recovery_prime`.
_recovery_context_prime: str | None = None

#: When the current prime was built (epoch). A prime can now survive a
#: __session_start__ greeting and failed deliveries, so it needs an age cap:
#: if the user walks away and comes back hours later, replaying the
#: poison-moment turns as "ongoing conversation" would be wrong. Primes older
#: than this are discarded at consume time.
_recovery_prime_built_at: float = 0
_RECOVERY_PRIME_MAX_AGE_S: float = 1800.0  # 30 min

#: Most recent steer/interject message per session key, with timestamp.
#: When a user speaks mid-inference the interject is delivered fire-and-forget
#: — if the in-flight LLM turn then returns empty, the steer is effectively
#: lost. The empty-response handler consults this map and, if a steer landed
#: within the last 30s on this session, re-fires it as a fresh conversation
#: turn so the user's correction actually reaches the agent.
#: Format: ``{session_key: (epoch_ts: float, message: str)}``
_recent_steer_by_session: dict = {}


def record_recent_steer(session_key: str, message: str) -> None:
    """Record that a steer was just injected into ``session_key``.

    Called from the steer / interject HTTP routes. Enables the empty-response
    recovery path to re-fire the steer as a fresh turn if the current LLM
    call collapses to zero chars (a common failure mode when a steer lands
    mid-inference and GLM cannot reconcile the branched context).
    """
    _recent_steer_by_session[session_key] = (time.time(), message)


def consume_recent_steer(session_key: str, max_age_s: float = 30.0) -> str | None:
    """Return and clear the most recent steer for ``session_key`` if it is
    younger than ``max_age_s``. Returns ``None`` if there is no recent steer.
    """
    entry = _recent_steer_by_session.get(session_key)
    if not entry:
        return None
    ts, msg = entry
    if time.time() - ts > max_age_s:
        _recent_steer_by_session.pop(session_key, None)
        return None
    _recent_steer_by_session.pop(session_key, None)
    return msg


def _build_recovery_prime(max_turns: int = 6, window_minutes: int = 120) -> str | None:
    """Build a compressed history summary from the most recent DB turns so a
    fresh recovery session doesn't lose context.

    Reads the last ``max_turns`` user+assistant rows from ``conversation_log``
    (session_id='default') and renders them as a bracketed SYSTEM note that
    the agent can parse. Returns ``None`` if no rows are available or the
    query fails — callers should handle ``None`` gracefully.
    """
    try:
        from services.paths import DB_PATH
        from datetime import timedelta
        import sqlite3
        # Time-filter: only inject turns from the last ``window_minutes`` so
        # stale context from previous sessions (days old) can't poison a fresh
        # recovery.
        #
        # FORMAT FIX: log_conversation() writes created_at as
        # datetime.now().isoformat() → '2026-06-12T03:46:25' (with a 'T').
        # The old filter compared against datetime('now','-10 minutes') →
        # '2026-06-12 03:38:25' (with a space). SQLite compares TEXT
        # lexically and 'T' > ' ', so every same-day row passed regardless
        # of age (fail-open) while just after UTC midnight NO row passed
        # (fail-closed → prime=None → recovery amnesia even on the happy
        # path). Compute the cutoff in Python in the SAME isoformat the
        # writer uses so the comparison is exact.
        #
        # WINDOW: 120 min, deliberately wider than the old nominal 10 — the
        # accidental same-day window is what made delivered primes good;
        # a true 10-minute window would hand back an EMPTY prime whenever
        # poisoning follows a normal conversational pause.
        _cutoff = (datetime.now() - timedelta(minutes=window_minutes)).isoformat()
        with sqlite3.connect(str(DB_PATH), timeout=2.0) as _c:
            _c.row_factory = sqlite3.Row
            # NOTE: most rows land with session_id IS NULL (the main
            # conversation route passes a per-request session_id that is
            # usually None) — only the steer/interject routes explicitly
            # write session_id='default'. Include both so the prime
            # reflects the actual recent conversation.
            rows = _c.execute(
                'SELECT role, message FROM conversation_log '
                "WHERE (session_id = 'default' OR session_id IS NULL) "
                'AND created_at >= ? '
                'ORDER BY id DESC LIMIT ?',
                (_cutoff, max_turns),
            ).fetchall()
        if not rows:
            return None
        rows = list(reversed(rows))  # oldest first
        lines = []
        for r in rows:
            role = r['role']
            msg = (r['message'] or '').strip().replace('\n', ' ')
            if len(msg) > 280:
                msg = msg[:280] + '…'
            lines.append(f'{role}: {msg}')
        body = '\n'.join(lines)
        # The prime is deliberately phrased as background context, NOT as a
        # "session was reset" notice. Earlier wording caused the model to
        # respond like it was starting a new conversation ("Here's what I've
        # got:", "Let me check..." etc.) because "[SESSION_RECOVERED]" reads
        # as a break. Now it's just framed as recent conversation history
        # plus a directive to pick up the thread naturally.
        return (
            '[RECENT CONTEXT — these are the most recent turns between you '
            'and the user. Treat them as ongoing conversation. Do not '
            'acknowledge a reset, do not re-greet, do not summarize what you '
            "just did. Pick up exactly where you left off — if the last user "
            'message asked a question or interrupted work you had started, '
            'answer that specific question or continue that specific work '
            'right now.]\n'
            f'{body}\n\n'
        )
    except Exception as _e:
        logger.warning(f'### _build_recovery_prime failed: {_e}')
        return None


def _with_recent_history(fallback_msg: str) -> str:
    """Prepend recent conversation turns to a Z.AI-direct fallback message.

    The direct fallback bypasses the openclaw session entirely, so without
    this the fallback model answers with ZERO conversation memory — the
    client-facing "amnesia" replies ("no indication of what 'one' refers
    to", tenant-e 2026-07-02). Reuses the recovery-prime builder (last turns
    from conversation_log). Fail-open: on any problem, return the message
    unchanged.
    """
    try:
        prime = _build_recovery_prime(max_turns=8)
        if prime:
            return prime + fallback_msg
    except Exception as _e:
        logger.warning(f'### _with_recent_history failed: {_e}')
    return fallback_msg


def consume_recovery_prime() -> str | None:
    """Return the recovery context prime once and clear it.
    Discards primes older than :data:`_RECOVERY_PRIME_MAX_AGE_S` — a prime
    that survived a greeting / failed deliveries must not replay hours-old
    turns as if they were the ongoing conversation."""
    global _recovery_context_prime
    p = _recovery_context_prime
    _recovery_context_prime = None
    if p and time.time() - _recovery_prime_built_at > _RECOVERY_PRIME_MAX_AGE_S:
        logger.info(
            f'### Recovery prime expired (built '
            f'{int(time.time() - _recovery_prime_built_at)}s ago) — discarding'
        )
        return None
    return p


def restore_recovery_prime(prime: str | None) -> None:
    """Re-arm a consumed recovery prime after the turn that carried it FAILED
    (empty response / gateway error / stream timeout).

    Without this, the prime is single-consume: if the first request on the
    recovery session dies (common — the same upstream brownout that poisoned
    the session is usually still in progress), the conversation context is
    lost forever and the user meets an amnesiac agent. Verified live on
    tenant-p 2026-06-11 19:58 ("Injecting session-recovery prime (2758 chars)"
    immediately followed by "EMPTY-FINAL RETRY FAILED: both attempts empty").

    Only restores while still in recovery mode and only if no newer prime has
    been armed in the meantime (a fresh double-empty rebuilds a newer prime —
    never clobber it with an older one).
    """
    global _recovery_context_prime
    if prime and _session_recovery_key is not None and _recovery_context_prime is None:
        _recovery_context_prime = prime
        logger.info(f'### Recovery prime re-armed after failed delivery ({len(prime)} chars)')


def mark_recovery_success() -> None:
    """Record that the current recovery session produced a real response.
    A successful recovery session is the new stable session (sticky-recovery
    design) — :func:`_check_recovery_timeout` must not idle-clear it back to
    the possibly-still-poisoned main key.
    """
    global _recovery_had_success
    if _session_recovery_key is not None and not _recovery_had_success:
        _recovery_had_success = True
        logger.info(
            f'### SESSION RECOVERY HEALTHY: "{_session_recovery_key}" produced a real '
            f'response — keeping it as the stable session (no idle-clear back to main)'
        )


def _enter_session_recovery():
    """Switch to a temporary recovery session key after double-empty.
    Openclaw will create a fresh session for this key, escaping the
    poisoned state. The recovery key is cleared on the first successful
    (non-empty, non-fallback) response.

    Also builds a context-replay prime from recent DB history so the fresh
    session doesn't lose the thread of conversation. Without this prime the
    agent behaves like a brand-new conversation and users experience
    'context lost' after a recovery.

    Uses a timestamped key so that if the recovery session ITSELF poisons
    later, a new recovery-<epoch> session can be spun up cleanly. The
    last-exited cooldown prevents rapid thrashing."""
    global _session_recovery_key, _recovery_entered_at, _recovery_last_activity_at, \
        _recovery_context_prime, _recovery_had_success, _recovery_prime_built_at
    # Cooldown: prevent re-entering recovery within 10s of a previous SUCCESSFUL
    # exit. Pre-Fix-F this was measured against _recovery_entered_at which
    # double-dutied as "last activity" after activity bumping was added —
    # result: recovery blocked itself for the full duration of a productive
    # recovery turn, and any subsequent poisoning on main became unrecoverable.
    # Using the last-exited timestamp means recoveries can re-fire immediately
    # after a successful one, which is exactly what a repeatedly-poisoned main
    # session requires.
    now = time.time()
    if _recovery_last_exited_at > 0 and now - _recovery_last_exited_at < 10:
        logger.info(
            '### SESSION RECOVERY: skipping — cooldown active '
            f'({int(now - _recovery_last_exited_at)}s since last exit, <10s)'
        )
        return
    _recovery_entered_at = now
    _recovery_last_activity_at = now
    # Use a timestamped recovery key so if recovery ITSELF poisons later
    # we can spin up a new recovery-<epoch> session cleanly. Earlier code
    # used a fixed 'recovery' key and was prone to piling up zombie
    # sessions — but that failure mode only happens when we thrash
    # recovery entries rapidly. With the last-exited cooldown we only
    # enter recovery when main is genuinely broken, so a new session per
    # poisoning event is appropriate.
    _session_recovery_key = f'recovery-{int(now)}'
    _recovery_had_success = False
    # Pull 30 turns instead of 6 — complex multi-turn icon/canvas work easily
    # exceeds 6 turns and the agent was losing all context after recovery.
    _recovery_context_prime = _build_recovery_prime(max_turns=30)
    _recovery_prime_built_at = now
    logger.warning(
        f'### SESSION RECOVERY: switching to key "{_session_recovery_key}" '
        f'(prime={"yes" if _recovery_context_prime else "no"}, '
        f'prime_chars={len(_recovery_context_prime) if _recovery_context_prime else 0}) '
        f'to escape poisoned session'
    )


def _exit_session_recovery():
    """Clear the recovery key after a successful response.
    Next request goes back to the stable key (cache-warm path).
    Also resets the double-empty circuit breaker and records the exit
    timestamp so :func:`_enter_session_recovery` can cooldown against it
    (prevents thrash, but ALLOWS re-entry when main gets re-poisoned)."""
    global _session_recovery_key, _double_empty_restart_count, _recovery_last_exited_at, \
        _recovery_context_prime, _recovery_had_success
    if _session_recovery_key is not None:
        old_recovery = _session_recovery_key
        _session_recovery_key = None
        _double_empty_restart_count = 0
        _recovery_had_success = False
        # Drop any armed-but-undelivered prime — the stable key has its own
        # server-side history; injecting a recovery prime there duplicates it.
        _recovery_context_prime = None
        _recovery_last_exited_at = time.time()
        stable = get_voice_session_key()
        logger.info(f'### SESSION RECOVERY CLEARED: "{old_recovery}" → back to stable key "{stable}"')


#: Idle-timeout cap for a recovery session. Bumped from 60s (too aggressive —
#: single recovery turns with multiple tools easily exceed 60s) to 10 min.
#: The happy-path exit is :func:`_exit_session_recovery`, which fires on the
#: first successful non-empty response; this cap is only a safety net for
#: genuinely stuck recoveries.
_RECOVERY_IDLE_TIMEOUT_S: float = 600.0


def bump_recovery_activity() -> None:
    """Called when we see proof-of-life on the recovery session — reset the
    idle timer so a productive multi-tool recovery turn doesn't get kicked
    out mid-work. Invoked from the streaming event pump on any gateway
    event while :data:`_session_recovery_key` is set.

    Separate from ``_recovery_entered_at`` (first-entered timestamp) and
    ``_recovery_last_exited_at`` (cooldown basis) so recovery lifecycle
    bookkeeping doesn't collide.
    """
    global _recovery_last_activity_at
    if _session_recovery_key is not None:
        _recovery_last_activity_at = time.time()


def _check_recovery_timeout():
    """Auto-clear STUCK recovery keys — i.e. recovery sessions that never
    produced a single successful response and have been idle longer than
    :data:`_RECOVERY_IDLE_TIMEOUT_S`.

    A recovery session that HAS succeeded is the new stable session
    (sticky-recovery design — see the text_done handler) and is deliberately
    NEVER idle-cleared. The old behavior cleared any idle recovery, dumping
    the next conversation back onto the possibly-still-poisoned main key:
    every >10-minute pause cost the user two dead turns (empty + retry-empty)
    and a context switch into yet another recovery session. Verified live on
    tenant-p 2026-06-11: TIMEOUT-clear of a healthy recovery → next interaction
    on main → DOUBLE EMPTY → new recovery.
    """
    global _session_recovery_key, _recovery_context_prime
    if _session_recovery_key is None:
        return
    if _recovery_had_success:
        return  # healthy recovery = the new stable session; never idle-clear
    idle_for = time.time() - max(_recovery_last_activity_at, _recovery_entered_at)
    if idle_for > _RECOVERY_IDLE_TIMEOUT_S:
        logger.warning(
            f'### SESSION RECOVERY TIMEOUT: "{_session_recovery_key}" '
            f'idle for >{int(_RECOVERY_IDLE_TIMEOUT_S)}s with no successful '
            f'response (stuck) — clearing'
        )
        _session_recovery_key = None
        # The prime belonged to the abandoned recovery attempt — don't let it
        # leak into the stable session's next turn.
        _recovery_context_prime = None


# ---------------------------------------------------------------------------
# Helper: notify Brain (non-critical fire-and-forget)
# ---------------------------------------------------------------------------


def _notify_brain(event_type: str, **data) -> None:
    """Append an event to the Brain events file for context tracking."""
    try:
        event = {'type': event_type, 'timestamp': datetime.now().isoformat()}
        event.update(data)
        with open(BRAIN_EVENTS_PATH, 'a') as f:
            f.write(json.dumps(event) + '\n')
    except Exception:
        pass  # Non-critical

# ---------------------------------------------------------------------------
# Helper: log conversation to SQLite
# ---------------------------------------------------------------------------


def log_conversation(role: str, message: str, session_id: str = 'default',
                     tts_provider: str = None, voice: str = None) -> None:
    """Log a single conversation turn to the database (non-blocking).

    Write is queued to the background db-writer thread (FIND-01 fix).
    """
    _db_write_queue.put((
        str(DB_PATH),
        'INSERT INTO conversation_log '
        '(session_id, role, message, tts_provider, voice, created_at) '
        'VALUES (?, ?, ?, ?, ?, ?)',
        (session_id, role, message, tts_provider, voice, datetime.now().isoformat()),
    ))
    _notify_brain('conversation', role=role, message=message, session=session_id)

# ---------------------------------------------------------------------------
# Helper: log timing metrics
# ---------------------------------------------------------------------------


def log_metrics(metrics: dict) -> None:
    """Log conversation timing metrics to SQLite + journalctl (non-blocking).

    Write is queued to the background db-writer thread (FIND-01 fix).
    """
    logger.info(
        f"[METRICS] profile={metrics.get('profile')} "
        f"handshake={metrics.get('handshake_ms')}ms "
        f"llm={metrics.get('llm_inference_ms')}ms "
        f"tts={metrics.get('tts_generation_ms')}ms "
        f"total={metrics.get('total_ms')}ms "
        f"resp_len={metrics.get('response_len')} "
        f"tts_ok={metrics.get('tts_success', 1)} "
        f"tools={metrics.get('tool_count', 0)} "
        f"fallback={metrics.get('fallback_used', 0)}"
    )
    _db_write_queue.put((
        str(DB_PATH),
        '''INSERT INTO conversation_metrics
           (session_id, profile, model, handshake_ms, llm_inference_ms,
            tts_generation_ms, total_ms, user_message_len, response_len,
            tts_text_len, tts_provider, tts_success, tts_error,
            tool_count, fallback_used, error, created_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''',
        (
            metrics.get('session_id', 'default'),
            metrics.get('profile', 'unknown'),
            metrics.get('model', 'unknown'),
            metrics.get('handshake_ms'),
            metrics.get('llm_inference_ms'),
            metrics.get('tts_generation_ms'),
            metrics.get('total_ms'),
            metrics.get('user_message_len'),
            metrics.get('response_len'),
            metrics.get('tts_text_len'),
            metrics.get('tts_provider'),
            metrics.get('tts_success', 1),
            metrics.get('tts_error'),
            metrics.get('tool_count', 0),
            metrics.get('fallback_used', 0),
            metrics.get('error'),
            datetime.now().isoformat(),
        ),
    ))

# ---------------------------------------------------------------------------
# Helper: clean text for TTS
# ---------------------------------------------------------------------------


def extract_tts_sentence(text, min_len=40):
    """Return (sentence, remainder) at the first sentence boundary at or after
    min_len chars of a STREAMING buffer, or (None, text) when there is none yet.

    A boundary is . ! or ? followed by a space or by the end of the buffer.
    A period that ends the buffer right after a digit is NOT a boundary: the
    buffer is still growing, so "$0." is usually the front half of "$0.79"
    whose digits are in the next delta. Splitting there spoke "79/hr ..." as a
    fragment. The next delta (or the end-of-turn flush) settles it.
    """
    if len(text) < min_len:
        return None, text
    for match in re.finditer(r'[.!?](?= |\Z)', text):
        end = match.end()
        if end < min_len:
            continue
        if (end == len(text) and match.group() == '.'
                and end >= 2 and text[end - 2].isdigit()):
            return None, text
        return text[:end].strip(), text[end:].lstrip()
    return None, text


def _truncate_at_sentence(text: str, max_chars: int) -> str:
    """Truncate text at the nearest sentence boundary at or before max_chars.
    Falls back to hard truncation if no boundary is found."""
    if not text or len(text) <= max_chars:
        return text
    chunk = text[:max_chars]
    # Find last sentence-ending punctuation before the cap
    last_boundary = max(chunk.rfind('.'), chunk.rfind('!'), chunk.rfind('?'))
    if last_boundary > 0:
        return chunk[:last_boundary + 1].strip()
    return chunk.strip()


def normalize_action_tags(text: str) -> str:
    """Normalize whitespace inside action-tag brackets.

    GLM sometimes emits `[ MUSIC_PLAY:Title ]` with stray whitespace after `[`
    or around `:` / before `]`. Every downstream regex (TTS strip, frontend
    tag extraction) requires a tight `[TAG:value]` form, so we collapse the
    spaces here before any other processing runs.
    """
    if not text:
        return text
    text = re.sub(r'\[\s+', '[', text)
    def _fix(m):
        inner = m.group(1).strip()
        inner = re.sub(r'\s*:\s*', ':', inner, count=1)
        return f'[{inner}]'
    text = re.sub(r'\[([A-Z][A-Z0-9_]*(?:\s*:[^\]]*)?)\]', _fix, text)
    return text


# Tags that DO work: the frontend executes them. A reply carrying one acted this
# turn even with zero tool calls, and nudging it as a broken promise can re-run the
# action (a second paid Suno generation). Measured 2026-10-01 01:55Z on tenant-g: a
# reply with [SUNO_GENERATE:...] and no tool call was nudged. Expressive tags
# ([MOOD:], [WAIT:], [SLEEP]) are not work and do not count.
WORK_ACTION_TAG_RE = re.compile(
    r'\[(?:SUNO_GENERATE|JINGLE_GENERATE|CANVAS|CANVAS_URL|CANVAS_ACTION|CANVAS_STYLE|CANVAS_MENU|CANVAS_CLOSE|'
    r'MUSIC_PLAY|MUSIC_STOP|START_TASK|TASK_COMPLETE|SOUNDCLOUD|SOUNDCLOUD_PAGE|BANDCAMP_PAGE|'
    r'OPEN_TAB|NAVIGATE|CLICK|FILL|SCROLL|BROWSE_ACTION|DOWNLOAD_IMAGE|REGISTER_FACE|HIGHLIGHT)'
    r'(?::[^\]]*)?\]'
)

# "I'll / let me ... <verb>" with the committing verb anywhere within 80 chars of
# the opener, so compound phrasing like "I'll mark the ones and add a button"
# still catches `add`.
PROMISE_RE = re.compile(
    r"\b(?:let me|i'?ll|i am going to|i'?m going to|i will|i'?m about to|gonna|going to)\b"
    r".{0,80}?"
    r"\b(?:write|build|create|update|save|add|edit|run|fetch|generate|make|"
    r"set up|put together|pull|grab|load|open|check|look|query|send|post|commit|push|"
    r"refactor|deploy|install|rebuild|restart|scaffold|configure|"
    r"mark|highlight|tag|label|link|embed|include|list|draft|prepare|"
    r"implement|modify|append|remove|delete|clean|organize|sort|render|"
    r"test|publish|upload|download|compile|parse|extract|apply|assign|"
    r"wire|hook|bind|attach|register|inject|populate|fill|insert|replace|"
    r"rename|move|copy|merge|split|style|design|format|export|import|"
    r"patch|fix|revert|rollback|scaffold|bootstrap|finalize|finish)"
    r"\b",
    re.IGNORECASE,
)


def gateway_reports_tools(gateway_id) -> bool:
    """True only for the OpenClaw gateway, the one gateway that reports tool_count.
    A plugin gateway (comms-brain, hermes, ...) always reports 0 tools, so the
    uncommitted-promise detector would fire on every "I'll ..." reply it gives and
    send a bracket-only [SYSTEM: ...] steer it was never built to answer. Measured on
    test-dev 2026-10-20 01:14-01:21Z: comms-brain answered each steer with its own
    "Sorry, I didn't catch that." 4 times in 7 minutes, spoken after real replies."""
    return gateway_id in (None, 'openclaw')


def is_degenerate_reply(text):
    """True for a bare "NO" / "YES" (optional trailing . ! ?) reply: broken-LLM
    output, never a real answer. ONE predicate shared by the in-stream
    empty/degenerate retry, the final safety net and the offline relay
    (#565), so they cannot drift apart."""
    return (text or '').strip().upper().rstrip('.!?') in ('NO', 'YES')


def is_uncommitted_promise(text, tool_count, llm_ms):
    """True when a reply promised work it did not do: it says "I'll / let me <verb>",
    called ZERO tools, carried no work-doing action tag, took >1s (instant
    degenerate empties go to the empty-retry branch), and does not END with a
    question. A closing question is the agent asking the user for a decision
    (tenant-e 2026-07-02); forcing "do it now" bypasses the answer AND lands a
    chat.send in the finalization window, where it coalesces into a bare final."""
    if not text or not text.strip():
        return False
    if tool_count:
        return False
    if (llm_ms or 0) <= 1000:
        return False
    if WORK_ACTION_TAG_RE.search(text):
        return False
    if text.strip().rstrip('*_ \n').endswith('?'):
        return False
    return bool(PROMISE_RE.search(text))


# ── stream-offline-relay ────────────────────────────────────────────────────
# A streamed reply that FINISHES after the browser dropped used to land in the
# gateway history only: the stream generator died at its next write
# (GeneratorExit), nobody read the rest of event_queue, and the owner was never
# told. When the stream's client is gone before its final text_done was written,
# a background thread waits for the turn's final reply and hands it to the SAME
# delivery path as offline subagent completions (#549):
# gateway.deliver_offline_completion -> server.push_proactive_message (browser
# push + .proactive-push.jsonl receipt) -> delivered=0 -> _relay_offline_completion
# (tenant's own agent; once per run id; skips relay-marker text and the relay
# session; skips when the gateway is disconnected).
# Never relayed: partial text (aborted / timed-out runs), errors, deferred
# turns, system triggers (__session_start__ ...), system responses, turns
# the user deliberately aborted (/api/conversation/abort: stop, barge-in, PTT),
# and an UNCOMMITTED PROMISE ("I'll set that up now", 0 tools): the
# auto-continue steer that would have made the agent do the work lives inside
# stream_response(), which is dead, so relaying the promise would tell the owner
# the work is happening when nothing will do it (#563 review). Same predicate
# and arguments as the in-stream auto-continue: gateway_reports_tools() +
# is_uncommitted_promise(text, tool_count, llm_ms).
_STREAM_RELAY_MAX_WAIT_S = 900      # gateway turns end by their own watchdogs well before this
_STREAM_RELAY_ABORT_GRACE_S = 2.0   # the abort POST trails the fetch abort by ms
_STREAM_RELAY_ABORT_WINDOW_S = 30   # disconnect is noticed at the next write (heartbeat <=10s)
_last_user_abort_ts = 0.0


def _offline_relay_text(evt):
    """The completed final reply carried by a text_done event, or None when the
    event must not be relayed (partial / error / deferred / empty / system)."""
    if not evt or evt.get('type') != 'text_done':
        return None
    if evt.get('deferred') or evt.get('error') or evt.get('aborted') or evt.get('timed_out'):
        return None
    text = normalize_action_tags(evt.get('response') or '').strip()
    if not text or is_system_response(text):
        return None
    if is_degenerate_reply(text):
        return None   # bare NO/YES is broken-LLM output, not an answer (#565)
    return text


def _tool_start_count(actions):
    """Tool calls in a turn's actions, counted exactly as metrics['tool_count']."""
    return sum(1 for a in (actions or [])
               if isinstance(a, dict) and a.get('type') == 'tool' and a.get('phase') == 'start')


def _stream_offline_relay(event_queue, done_evt, *, gateway_id, agent_id,
                          session_key, run_id, t_request_start, t_disconnect,
                          t_llm_start=None, done_llm_ms=None):
    """Background body of stream-offline-relay (see block comment above).

    done_evt is the text_done the dead stream had already taken off
    event_queue, or None when the turn is still running on event_queue.
    done_llm_ms is that leg's LLM time as the stream measured it; for a
    text_done read here the time is measured from t_llm_start (the start of
    the leg running on event_queue), the same way the stream measures it."""
    evt = done_evt
    llm_ms = done_llm_ms
    if evt is None:
        deadline = time.time() + _STREAM_RELAY_MAX_WAIT_S
        while time.time() < deadline:
            try:
                e = event_queue.get(timeout=min(30.0, max(0.1, deadline - time.time())))
            except queue.Empty:
                continue
            etype = e.get('type')
            if etype == 'text_done':
                evt = e
                if t_llm_start is not None:
                    llm_ms = int((time.time() - t_llm_start) * 1000)
                break
            if etype == 'error':
                logger.info(f'### stream-offline-relay: turn errored after disconnect ({run_id}) — nothing to relay')
                return
        if evt is None:
            logger.warning(f'### stream-offline-relay: no final reply within {_STREAM_RELAY_MAX_WAIT_S}s ({run_id})')
            return
    text = _offline_relay_text(evt)
    if text is None:
        logger.info(f'### stream-offline-relay: final event not relayable ({run_id})')
        return
    _tools = _tool_start_count(evt.get('actions'))
    if gateway_reports_tools(gateway_id) and is_uncommitted_promise(text, _tools, llm_ms):
        logger.warning(
            f'### stream-offline-relay: NOT relayed — the final reply is an uncommitted '
            f'promise (tools={_tools}, ms={llm_ms}); the auto-continue that would have '
            f'done the work died with the stream, so the owner is not told it is '
            f'happening ({run_id}): {text[:120]!r}')
        return
    if _STREAM_RELAY_ABORT_GRACE_S:
        time.sleep(_STREAM_RELAY_ABORT_GRACE_S)
    if _last_user_abort_ts >= max(t_request_start, t_disconnect - _STREAM_RELAY_ABORT_WINDOW_S):
        logger.info(f'### stream-offline-relay: skipped — user aborted this turn ({run_id})')
        return
    gw = gateway_manager.get(gateway_id)
    deliver = getattr(gw, 'deliver_offline_completion', None)
    if deliver is None:
        logger.info(f'### stream-offline-relay: gateway {gateway_id or "openclaw"} has no offline delivery ({run_id})')
        return
    logger.info(f'### stream-offline-relay: reply finished after the client left — '
                f'delivering ({len(text)} chars, {run_id}): {text[:120]}')
    try:
        deliver(text, session_key, (run_id,), agent_id=agent_id)
    except Exception as e:
        logger.warning(f'### stream-offline-relay: delivery failed ({run_id}): {e}')


def _start_stream_offline_relay(**kwargs):
    threading.Thread(target=_stream_offline_relay, kwargs=kwargs,
                     name='stream-offline-relay', daemon=True).start()


def strip_canvas_action_tags(text: str) -> str:
    """Strip [CANVAS_ACTION:{...JSON...}] tags via a brace walker.

    The JSON payload nests {} and [] (e.g. checkbox-group ``values`` arrays),
    which defeats both ``[^\\]]*`` and the ``_nb`` one-level-nesting pattern
    used for browser-companion tags. Walk the payload respecting string
    literals, escapes, and brace depth. (2026-07-02 tradeshow form-fill fix —
    without this the raw JSON tag was spoken aloud by TTS.)
    """
    if not text or '[CANVAS_ACTION' not in text.upper():
        return text
    out = []
    idx = 0
    pattern = re.compile(r'\[CANVAS_ACTION:', re.IGNORECASE)
    while True:
        m = pattern.search(text, idx)
        if not m:
            out.append(text[idx:])
            break
        json_start = m.end()
        if json_start >= len(text) or text[json_start] != '{':
            out.append(text[idx:m.end()])
            idx = m.end()
            continue
        depth, i, in_str, esc = 0, json_start, False, False
        while i < len(text):
            c = text[i]
            if in_str:
                if esc:
                    esc = False
                elif c == '\\':
                    esc = True
                elif c == '"':
                    in_str = False
            elif c == '"':
                in_str = True
            elif c == '{':
                depth += 1
            elif c == '}':
                depth -= 1
                if depth == 0:
                    i += 1
                    break
            i += 1
        if depth == 0 and i < len(text) and text[i] == ']':
            out.append(text[idx:m.start()])
            idx = i + 1  # skip the whole tag
        else:
            # Malformed/unterminated tag (e.g. mid-stream cut): drop from tag
            # start to end of text so a half-arrived tag is never spoken.
            out.append(text[idx:m.start()])
            idx = len(text)
            break
    return ''.join(out)


def clean_for_tts(text: str) -> str:
    """Remove markdown, reasoning tokens, and non-speech characters for TTS."""
    if not text:
        return ''

    # Normalize sloppy tag whitespace FIRST so every strip regex below matches.
    text = normalize_action_tags(text)

    # Strip [CANVAS_ACTION:{json}] BEFORE the regex chain — the JSON payload
    # nests ] and } so no regex below can match it (2026-07-02).
    text = strip_canvas_action_tags(text)

    # Strip GPT-OSS-120B reasoning tokens (but not if NO/YES is the full response)
    if text.strip().upper() not in ['NO', 'YES', 'NO.', 'YES.']:
        text = re.sub(r'^NO_REPLY\s*', '', text)
        text = re.sub(r'\s+NO\s*$', '', text, flags=re.IGNORECASE)
        text = re.sub(r'\s+YES\s*$', '', text, flags=re.IGNORECASE)

    # Remove canvas/task/music triggers (handled by frontend, not spoken)
    text = re.sub(r'\[CANVAS_MENU\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[CANVAS_CLOSE\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[CANVAS:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[CANVAS_URL:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[MUSIC_PLAY(?::[^\]]*)?\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[MUSIC_STOP\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[MUSIC_NEXT\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[SUNO_GENERATE:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[JINGLE_GENERATE:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[SLEEP\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[AIRADIO_[A-Z_]+(?::[^\]]*)?\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[MOOD:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[REGISTER_FACE:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[SPOTIFY:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[SOUNDCLOUD:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[SOUNDCLOUD_PAGE:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[BANDCAMP:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[BANDCAMP_PAGE:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[SOUND:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[SESSION_RESET\]', '', text, flags=re.IGNORECASE)

    # Remove browser companion command tags (executed by extension, not spoken)
    # Pattern handles nested brackets in CSS selectors: [CLICK:[role="button"]]
    _nb = r'(?:[^\[\]]|\[[^\]]*\])*'  # non-bracket chars OR [bracket-pairs]
    text = re.sub(r'\[SCROLL:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(rf'\[CLICK:{_nb}\]', '', text, flags=re.IGNORECASE)
    text = re.sub(rf'\[FILL:{_nb}\]', '', text, flags=re.IGNORECASE)
    text = re.sub(rf'\[HIGHLIGHT:{_nb}\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[NAVIGATE:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[OPEN_TAB:[^\]]*\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[READ_PAGE\]', '', text, flags=re.IGNORECASE)
    text = re.sub(r'\[WAIT:\d+\]', '', text, flags=re.IGNORECASE)
    text = re.sub(rf'\[START_TASK:{_nb}\]', '', text, flags=re.IGNORECASE)
    text = re.sub(rf'\[TASK_COMPLETE:{_nb}\]', '', text, flags=re.IGNORECASE)

    # Remove code blocks (complete fences first, then any unclosed fence to end of text)
    text = re.sub(r'```[\s\S]*?```', '', text)
    text = re.sub(r'```[\s\S]*', '', text)
    text = re.sub(r'`[^`]+`', '', text)

    # Add natural pauses for structured content (must happen before stripping markdown)
    text = re.sub(r'^(#+\s+.+?)([^.!?])\s*$', r'\1\2.', text, flags=re.MULTILINE)

    def _ensure_list_item_pause(match):
        prefix = match.group(1)
        content = match.group(2).strip()
        if content and content[-1] not in '.!?:':
            content += '.'
        return f'{prefix} {content}'
    text = re.sub(r'^(\s*\d+[.)]\s*)(.+?)$', _ensure_list_item_pause,
                  text, flags=re.MULTILINE)

    def _ensure_bullet_pause(match):
        content = match.group(1).strip()
        if content and content[-1] not in '.!?:':
            content += '.'
        return content
    text = re.sub(r'^\s*[-*•]\s+(.+?)$', _ensure_bullet_pause,
                  text, flags=re.MULTILINE)

    def _table_row_to_speech(match):
        row = match.group(0)
        if re.match(r'^[\s|:-]+$', row):
            return ''
        cells = [c.strip() for c in row.split('|') if c.strip()]
        if not cells:
            return ''
        return ', '.join(cells) + '.'
    text = re.sub(r'^\|.+\|$', _table_row_to_speech, text, flags=re.MULTILINE)

    lines = text.split('\n')
    for i, line in enumerate(lines):
        stripped = line.strip()
        if stripped and len(stripped) < 80 and stripped[-1] not in '.!?:,;':
            if re.match(r'^[A-Za-z0-9]', stripped):
                lines[i] = stripped + '.'
    text = '\n'.join(lines)

    # Strip markdown formatting
    text = re.sub(r'\*\*([^*]+)\*\*', r'\1', text)
    text = re.sub(r'\*([^*]+)\*', r'\1', text)
    text = re.sub(r'__([^_]+)__', r'\1', text)
    text = re.sub(r'_([^_]+)_', r'\1', text)
    text = re.sub(r'^#+\s+', '', text, flags=re.MULTILINE)
    text = re.sub(r'\[([^\]]+)\]\([^)]+\)', r'\1', text)
    text = re.sub(r'https?://\S+', '', text)
    text = re.sub(r'/[\w/.-]+', '', text)

    # Expand acronyms to speakable form
    acronyms = {
        'API': 'api', 'HTML': 'html', 'CSS': 'css', 'JSON': 'jason',
        'HTTP': 'http', 'HTTPS': 'https', 'URL': 'url', 'TTS': 'text to speech',
        'STT': 'speech to text', 'LLM': 'large language model', 'AI': 'A.I.',
        'UI': 'user interface', 'UX': 'user experience', 'RAM': 'ram',
        'CPU': 'cpu', 'GPU': 'gpu', 'DB': 'database', 'VPS': 'server',
        'SSH': 'ssh', 'CLI': 'command line', 'SDK': 'sdk', 'API': 'api',
    }
    for acronym, expansion in acronyms.items():
        text = re.sub(r'\b' + acronym + r'\b', expansion, text)

    # Replace symbols with spoken equivalents
    text = text.replace('&', ' and ')
    text = text.replace('%', ' percent ')
    text = text.replace('$', ' dollars ')
    text = text.replace('@', ' at ')
    text = text.replace('#', ' number ')
    text = text.replace('+', ' plus ')
    text = text.replace('=', ' equals ')

    # Clean up whitespace
    text = re.sub(r'\n+', '. ', text)
    # Preserve "..." — TTS models render ellipses as a natural trailing pause
    # (character voices like Kyle depend on them). Only collapse runs of 4+.
    text = re.sub(r'\.{4,}', '...', text)
    text = re.sub(r'\s+', ' ', text).strip()
    text = re.sub(r'(?<!\.)\.\s*\.(?!\.)', '.', text)
    # Strip leading punctuation/spaces (e.g. from [MUSIC_STOP]\n\n → ". text")
    text = re.sub(r'^[.,;:\s]+', '', text)

    return text

# ---------------------------------------------------------------------------
# Helper: legacy Supertonic voice accessor
# ---------------------------------------------------------------------------


def get_supertonic_for_voice(voice_style: str):
    """Get Supertonic provider (voice_style ignored — unified provider)."""
    return get_provider('supertonic')

# ---------------------------------------------------------------------------
# Blueprint
# ---------------------------------------------------------------------------

conversation_bp = Blueprint('conversation', __name__)

# ---------------------------------------------------------------------------
# POST /api/conversation — main voice conversation endpoint
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/conversation', methods=['POST'])
def conversation():
    """
    Handle voice conversation flow.

    Request JSON:
        message      : str  — transcribed user speech (required)
        tts_provider : str  — 'supertonic' | 'groq' (default: env DEFAULT_TTS_PROVIDER or groq)
        voice        : str  — voice ID, e.g. 'M1' (default: M1)
        session_id   : str  — session identifier (default: default)
        ui_context   : dict — canvas/music state from frontend (optional)

    Response JSON (non-streaming):
        response  : str  — AI text response
        audio     : str  — base64-encoded audio (if TTS succeeds)
        timing    : dict — handshake/llm/tts/total ms
        actions   : list — Gateway tool/lifecycle events (optional)
    """
    try:
        return _conversation_inner()
    except Exception as e:
        import traceback
        tb = traceback.format_exc()
        logger.error(f'FATAL: {tb}')
        return jsonify({
            'response': 'Something went wrong on my end. Try again?',
            'error': 'Internal server error'
        }), 500


def _conversation_inner():
    global _consecutive_empty_responses

    t_request_start = time.time()
    metrics = {
        'profile': 'gateway',
        'model': 'glm-5-turbo',
        'tts_success': 1,
        'fallback_used': 0,
        'tool_count': 0,
    }

    data = request.get_json()
    if not data:
        logger.error('ERROR: No JSON data in request')
        return jsonify({'error': 'No JSON data provided'}), 400

    logger.info(f'Received conversation request: {data}')

    user_message = data.get('message', '').strip()
    tts_provider = data.get('tts_provider') or os.getenv('DEFAULT_TTS_PROVIDER', 'groq')
    voice = data.get('voice', 'M1')
    session_id = data.get('session_id', 'default')
    ui_context = data.get('ui_context', {})
    identified_person = data.get('identified_person') or None
    # Capture Clerk user id from request middleware (set by app.py auth check)
    # so it can be persisted in the transcript JSON for The Office to attribute
    # turns correctly. Fail-open: missing g.clerk_user_id → None.
    _clerk_user_id = getattr(g, 'clerk_user_id', None)
    agent_id = data.get('agent_id') or None  # e.g. 'default'; None = default 'main'
    gateway_id = data.get('gateway_id') or None  # plugin gateway id; None = 'openclaw'
    # Fall back to active profile's adapter_config.gateway_id
    if not gateway_id:
        try:
            from profiles.manager import get_profile_manager
            import routes.profiles as _profiles_mod
            _pm = get_profile_manager()
            _ap = _pm.get_profile(_profiles_mod._active_profile_id)
            if _ap and _ap.adapter_config:
                gateway_id = _ap.adapter_config.get('gateway_id') or None
        except Exception:
            pass
    max_response_chars = data.get('max_response_chars') or None  # profile cap, truncates at sentence boundary
    image_path = data.get('image_path') or None  # uploaded image for vision analysis
    skip_tts = data.get('skip_tts', False)  # browser extension skips TTS during task steps
    metrics['session_id'] = session_id
    metrics['user_message_len'] = len(user_message)
    metrics['tts_provider'] = tts_provider

    if not user_message:
        return jsonify({'error': 'No message provided'}), 400

    # Filter garbage STT fragments — punctuation-only or single-character noise.
    # Threshold is 1 meaningful char (filters "." or " " but lets "yo", "ok",
    # "hi", "no", "ya" through — those are valid intentional one/two-letter
    # acknowledgements that real users say).
    import re as _re
    _meaningful_chars = _re.sub(r'[^a-zA-Z0-9]', '', user_message)
    if len(_meaningful_chars) < 2:
        logger.info(f'### FILTERED garbage STT: "{user_message}" ({len(_meaningful_chars)} meaningful chars)')
        # Return a no-op stream in NDJSON format (same wire format the rest of
        # this route uses — was previously SSE, which the client could not
        # parse, leaving the UI stuck in "thinking" state forever).
        def _noop_stream():
            yield json.dumps({'type': 'filtered', 'reason': 'garbage_stt'}) + '\n'
            yield json.dumps({
                'type': 'text_done',
                'response': '',
                'actions': [],
                'timing': {},
            }) + '\n'
        return Response(_noop_stream(), mimetype='application/x-ndjson')

    # Input length guard (P7-T3 security audit) — raised 2026-06-25.
    # A user's pasted prompt must NEVER be rejected for normal large pastes (text files, CSVs,
    # docs, transcripts). The old 15K cap turned a big paste into a 400 "Message too long" API
    # error with no response. Cap is now generous and sits well under the model's real input
    # budget (GLM 204.8K-token context ≈ ~800K chars) — only an abusive multi-MB paste hits it.
    # Browser-extension messages carry page context too, so they get a generous cap as well.
    _max_msg_len = 200000 if ui_context and ui_context.get('source') == 'jambot_extension' else 500000
    if len(user_message) > _max_msg_len:
        return jsonify({'error': f'Message too long (max {_max_msg_len} characters)'}), 400

    wants_stream = (
        request.args.get('stream') == '1'
        or request.headers.get('X-Stream-Response') == '1'
    )

    # Update canvas context from UI state
    if ui_context.get('canvasDisplayed'):
        update_canvas_context(
            ui_context['canvasDisplayed'],
            title=ui_context['canvasDisplayed']
                .replace('/pages/', '')
                .replace('.html', '')
                .replace('-', ' ')
                .title()
        )

    # Build context prefix from UI state
    t_context_start = time.time()
    context_prefix = ''
    context_parts = []   # dynamic — changes turn to turn, always sent
    static_parts = []    # static — identical between turns; deduped by the
                         # context-bloat guard below (sent on change/periodic)
    _ui_state_lines = []  # plain-text state summary mirrored to UI_STATE.md
                          # (issue #94) — canvas/music state only, no ephemera

    # Inject face recognition identity
    if identified_person and identified_person.get('name') and identified_person.get('name') != 'unknown':
        name = identified_person['name']
        confidence = identified_person.get('confidence', 0)
        context_parts.append(
            f'[FACE RECOGNITION: The person you are speaking with has been identified as {name} '
            f'({confidence}% confidence). Address them by name naturally.]'
        )

    # Vision: if user asks about what the camera sees, call vision model with latest frame
    if _is_vision_request(user_message):
        from routes.vision import _latest_frame, _call_vision
        _frame_img = _latest_frame.get('image')
        _frame_age = time.time() - _latest_frame.get('ts', 0)
        if _frame_img and _frame_age < _VISION_FRAME_MAX_AGE:
            try:
                _vision_desc = _call_vision(
                    _frame_img,
                    'Describe what you see in this image concisely. Focus on people, objects, and actions.',
                )
                context_parts.append(f'[CAMERA VISION: {_vision_desc}]')
            except Exception as exc:
                logger.warning('Vision analysis failed: %s', exc)
                context_parts.append('[CAMERA VISION: Camera is on but vision analysis failed.]')
        elif not _frame_img:
            context_parts.append('[CAMERA VISION: No camera frame available — camera may be off.]')
        else:
            context_parts.append('[CAMERA VISION: Camera frame is stale — camera may have been turned off.]')

    # Vision: if user uploaded an image, analyze it with vision model
    if image_path:
        try:
            _img_file = Path(image_path).resolve()
            # Security: only allow files inside uploads/ directories
            if 'uploads' not in _img_file.parts:
                raise ValueError(f'Path traversal blocked: {image_path}')
            if _img_file.is_file() and _img_file.stat().st_size < 20_000_000:  # 20MB safety cap
                # Check .image-intel cache first — avoids live API call when image was pre-analyzed
                _intel_path = _img_file.parent / '.image-intel' / (_img_file.name + '.json')
                _upload_desc = None
                if _intel_path.is_file():
                    try:
                        import json as _json
                        _intel = _json.loads(_intel_path.read_text())
                        _cached = _intel.get('analysis', {}).get('description', '').strip()
                        if _cached:
                            _kw = _intel.get('analysis', {}).get('keywords', [])
                            _txt = _intel.get('analysis', {}).get('text_in_image', '').strip()
                            _upload_desc = _cached
                            if _txt:
                                _upload_desc += f' Text visible in image: {_txt}'
                            if _kw:
                                _upload_desc += f' Keywords: {", ".join(_kw[:10])}'
                            logger.info('Loaded image analysis from .image-intel cache: %s', _img_file.name)
                    except Exception as _ie:
                        logger.debug('Could not read .image-intel cache: %s', _ie)
                if _upload_desc is None:
                    # SYNC GEMINI FIRST (Mike, 2026-08-18): an image attached to the chat
                    # transcript is a "single" — the user is looking at it NOW and the agent
                    # must read it in THIS turn, not tell them to wait 30s for the background
                    # vision agent. Gemini reads it synchronously; the mesh queue below
                    # survives only as the fallback when Gemini fails. Bulk uploads never
                    # pass through here (no image_path on a chat turn) — they are swept by
                    # the host-side auto-processor into .image-intel in the background.
                    _gem_key = os.environ.get('GEMINI_API_KEY', '')
                    if _gem_key:
                        try:
                            import requests as _rq
                            _mime = {
                                '.png': 'image/png', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg',
                                '.webp': 'image/webp', '.gif': 'image/gif',
                            }.get(_img_file.suffix.lower(), 'image/jpeg')
                            _b64 = base64.b64encode(_img_file.read_bytes()).decode()
                            _gr = _rq.post(
                                'https://generativelanguage.googleapis.com/v1beta/models/'
                                f'gemini-2.5-flash:generateContent?key={_gem_key}',
                                json={'contents': [{'parts': [
                                    {'inline_data': {'mime_type': _mime, 'data': _b64}},
                                    {'text': 'Describe this image concisely but completely: subjects, '
                                             'objects, layout, any visible text (transcribe it), and '
                                             'anything a user would likely ask about. Plain prose.'},
                                ]}]},
                                timeout=20,
                            )
                            _gr.raise_for_status()
                            _gtxt = (_gr.json()['candidates'][0]['content']['parts'][0]['text'] or '').strip()
                            if _gtxt:
                                _upload_desc = (
                                    f'{_gtxt} (Image file: uploads/{_img_file.name}; OpenClaw path '
                                    f'/home/node/.openclaw/workspace/uploads/{_img_file.name} if the '
                                    'image tool is ever needed — it rejects /app/runtime/uploads/.)'
                                )
                                logger.info('Chat image read synchronously via Gemini: %s', _img_file.name)
                                # Write the .image-intel sidecar so later turns hit the cache
                                # and the background sweeper knows this file is done.
                                try:
                                    import hashlib as _hl
                                    import json as _json
                                    _intel_path.parent.mkdir(exist_ok=True)
                                    _intel_path.write_text(_json.dumps({
                                        'schema': 1,
                                        'file': _img_file.name,
                                        'tenant': os.getenv('JAMBOT_TENANT', ''),
                                        'sha256': _hl.sha256(_img_file.read_bytes()).hexdigest(),
                                        'size': _img_file.stat().st_size,
                                        'source': 'gemini-sync',
                                        'original_name': _img_file.name,
                                        'analyzed_at': datetime.utcnow().isoformat() + 'Z',
                                        'analysis': {'description': _gtxt, 'keywords': [],
                                                     'text_in_image': ''},
                                    }, indent=1))
                                except Exception as _we:
                                    logger.debug('Could not write gemini-sync sidecar: %s', _we)
                        except Exception as _ge:
                            logger.warning('Sync Gemini vision failed (%s) — falling back to '
                                           'vision@mesh queue for %s', _ge, _img_file.name)
                if _upload_desc is None:
                    # No cache yet — queue a vision@mesh task for this tenant.
                    # Non-blocking: mesh task, vision agent writes .image-intel sidecar async.
                    # SEC-059: with RESTART_VIA_BROKER=1 this is POST /mesh-task/<tenant> on the
                    # host broker (mesh-send runs on the host side, bound to this tenant); the
                    # legacy path lives in services/host_broker.py.queue_vision_task_legacy.
                    from services import host_broker as _hb
                    # Initialised BEFORE the `if _tenant` guard on purpose: it is read after that
                    # block, so a tenant-less request would otherwise raise NameError. Defaulting
                    # to False also makes the no-tenant path honest — nothing was queued there
                    # either, so the user must not be told an analysis is coming.
                    _queued = False
                    _tenant = os.getenv('JAMBOT_TENANT', '')
                    if _tenant:
                        # Build the HOST path (vision agent runs on host, needs /mnt/clients/ path)
                        _host_img_path = f'/mnt/clients/{_tenant}/openvoiceui/uploads/{_img_file.name}'
                        _msg_body = (
                            f'image_path: {_host_img_path}\n'
                            f'tenant: {_tenant}\n'
                            f'filename: {_img_file.name}'
                        )
                        _subject = _hb.safe_subject(_img_file.stem)
                        # SUCCESS COMES FROM THE RESPONSE. This previously logged "Image analysis
                        # queued" unconditionally after reaching the call. A mesh-send that exited
                        # non-zero (unknown kind, missing mailbox, rate limit) was reported as
                        # SUCCESS BY THE SENDER while nothing was ever queued, and the user was told
                        # "being analyzed now" for an image nothing would ever look at. The broker
                        # answers ok only on mesh-send exit 0 with the delivered path; the legacy
                        # helper returns True only on exit 0.
                        _queued = False
                        try:
                            if _hb.via_broker():
                                _res = _hb.queue_vision_task(_tenant, _subject, _msg_body)
                                _queued = True
                                logger.info('Image analysis queued to vision@mesh via host broker '
                                            'for %s: %s (%s)', _tenant, _img_file.name,
                                            _res.get('delivered', ''))
                            else:
                                _queued = _hb.queue_vision_task_legacy(_tenant, _subject, _msg_body)
                                if _queued:
                                    logger.info('Image analysis queued to vision@mesh via openclaw-%s: %s',
                                                _tenant, _img_file.name)
                        except _hb.BrokerError as _be:
                            logger.error('vision@mesh queue FAILED via host broker tenant=%s file=%s: %s',
                                         _tenant, _img_file.name, _be)
                        except Exception as _se:
                            logger.warning('Failed to queue vision task to vision@mesh: %s', _se)
                    # Do not promise an analysis that was never queued.
                    # The path matters as much as the status: without it, agents guess
                    # (/app/runtime/uploads/ is readable by exec/read but REJECTED by the
                    # image tool, whose media roots are hardcoded to workspace paths) and
                    # burn a minute copying files around narrating paths over TTS.
                    if _queued:
                        _upload_desc = (
                            f'Image saved in your workspace uploads folder: uploads/{_img_file.name} '
                            f'(OpenClaw: /home/node/.openclaw/workspace/uploads/{_img_file.name}; '
                            f'Hermes: /workspace/uploads/{_img_file.name}). A background vision agent '
                            'is analyzing it right now — the result arrives in about 30 seconds. Do '
                            'NOT analyze the image yourself, do NOT run the image tool on it, and do '
                            'NOT copy or move the file. Tell the user you are taking a look; the '
                            'analysis will be in your context on their next message. If you ever do '
                            'need the image tool on an upload, it only accepts workspace paths like '
                            'the above — it rejects /app/runtime/uploads/.'
                        )
                        logger.info('Image routed to vision@mesh: %s', _img_file.name)
                    else:
                        _upload_desc = (
                            f'Image saved in your workspace uploads folder: uploads/{_img_file.name} '
                            f'(OpenClaw: /home/node/.openclaw/workspace/uploads/{_img_file.name}; '
                            f'Hermes: /workspace/uploads/{_img_file.name}), but automatic analysis '
                            'could not be started. If the user asks about it, run the image tool on '
                            'that exact workspace path — the image tool rejects /app/runtime/uploads/.'
                        )
                context_parts.append(
                    f'[UPLOADED IMAGE ANALYSIS: {_upload_desc}]\n'
                    '[ATTACHED-IMAGE RULE: The user attached THIS image with their message. '
                    'If their request is about this image (use it, recreate it, reorganize it, '
                    'build from it), the attached image is the subject of the task — work from '
                    'the description above. Do NOT substitute other files from the uploads '
                    'library; that fallback is only for when the user gives no direction. If '
                    'the task needs the individual image files visible inside this image and '
                    'they are not separately available, say so and ask the user to attach them '
                    'or name which upload files map to them.]'
                )
            else:
                logger.warning('Uploaded image not found or too large: %s', image_path)
                context_parts.append('[UPLOADED IMAGE: File could not be analyzed — may be too large or missing.]')
        except Exception as exc:
            logger.warning('Vision analysis of uploaded image failed: %s', exc)
            context_parts.append('[UPLOADED IMAGE: Vision analysis failed — the image was uploaded but could not be analyzed.]')

    if ui_context:
        # ── Browser Extension context ────────────────────────────────────────
        if ui_context.get('source') == 'jambot_extension':
            context_parts.append(
                '[BROWSER COMPANION MODE]\n'
                'You are operating the user\'s real Chrome browser through the JamBot extension sidebar.\n'
                'The page content below is LIVE — it updates after every action you take.\n\n'
                'BROWSER ACTIONS YOU CAN EXECUTE (include in your response, they run immediately):\n'
                '  [NAVIGATE:https://url] — navigate the browser to any URL (page loads then you see it)\n'
                '  [OPEN_TAB:https://url] — open URL in a new tab\n'
                '  [SCROLL:+1200]         — scroll down 1200px (USE THIS for feeds — loads new content)\n'
                '  [SCROLL:+800]          — scroll down 800px\n'
                '  [SCROLL:-400]          — scroll up 400px\n'
                '  [SCROLL:top]           — jump to top of page\n'
                '  [SCROLL:bottom]        — jump to absolute bottom (NOT for infinite feeds — use +1200 instead)\n'
                '  [SCROLL:selector]      — scroll to a specific element\n'
                '  [CLICK:selector]       — click an element (button, link, tab, etc.)\n'
                '  [FILL:selector:value]  — type into an input field\n'
                '  [HIGHLIGHT:selector]   — draw a cyan outline around an element\n'
                '  [READ_PAGE]            — request full page text (up to 15000 chars)\n'
                '  [WAIT:3]               — wait 3 seconds (for page loads, animations)\n'
                '  [DOWNLOAD_IMAGE]       — download the largest visible image on the page (Facebook, Instagram, Reddit, etc.)\n'
                '  [DOWNLOAD_IMAGE:selector] — download a specific image by CSS selector (e.g. img.profile-pic)\n\n'
                '⚠️ DOWNLOADING IMAGES IS SUPPORTED. If the user says "download this image" / "save this photo" / '
                '"grab that picture" — emit [DOWNLOAD_IMAGE] (no argument = largest visible image). '
                'Never say "I don\'t have that capability" and NEVER tell the user to right-click — '
                'the extension calls chrome.downloads.download() which works on every site. '
                'Never use web_fetch, puppeteer, or any container-side tool to fetch images — those '
                'use datacenter IPs and get blocked. The extension runs in the user\'s real browser.\n\n'
                'CREATING CANVAS PAGES (to save collected data):\n'
                '  You have full tool access (file_write, bash, etc.) through your agent runtime.\n'
                '  To create a canvas page with collected data:\n'
                '  1. Use your file_write tool to write an HTML file to ~/Canvas/<page-name>.html\n'
                '  2. The HTML should be self-contained with inline CSS (no external deps)\n'
                '  3. Include [CANVAS:<page-name>] in your text response so the app navigates to it\n'
                '  4. Do NOT say [TASK_COMPLETE] until the file is ACTUALLY WRITTEN — saying\n'
                '     "I\'ll create it" is NOT creating it. USE THE TOOL.\n\n'
                'AUTONOMOUS TASK MODE:\n'
                '  When the user asks you to perform a multi-step task (scroll through a feed,\n'
                '  find leads, fill out forms, etc.), you MUST activate the task loop:\n'
                '  1. Include [START_TASK:brief description] in your FIRST response\n'
                '  2. Then include your first command tag (e.g. [SCROLL:+1200])\n'
                '  3. After each command, I will automatically send you the updated page state\n'
                '  4. Keep outputting command tags on every response — the loop continues as long as you do\n'
                '  5. Use [TASK_COMPLETE:summary] when finished\n\n'
                '  WITHOUT [START_TASK:], commands execute ONCE and stop. Use it for ANY multi-step action:\n'
                '  - "scroll the page" → [START_TASK:Scroll through page] [SCROLL:+1200]\n'
                '  - "find leads" → [START_TASK:Find leads in feed] [SCROLL:+1200]\n'
                '  - "fill out this form" → [START_TASK:Fill form fields] [FILL:selector:value]\n\n'
                'ACTION JUDGMENT:\n'
                '  Read the user\'s intent. If they say "comment on posts" — fill AND submit, don\'t stop to ask.\n'
                '  If they say "draft a comment for me" — fill it and wait. Use common sense.\n'
                '  "Find leads and comment" = autonomous. "Show me what you\'d say" = review mode.\n'
                '  Only confirm before DESTRUCTIVE actions (delete, unfollow, block, unfriend).\n'
                '  The user can STOP the task at any time using the Stop button.\n\n'
                '⚠️ CRITICAL: The tags are NOT descriptions — they are EXECUTABLE COMMANDS.\n'
                'Including [SCROLL:+1200] in your response IMMEDIATELY scrolls the page.\n'
                'Simply saying "I will scroll" does NOTHING. You MUST output the tag.\n\n'
                'CORRECT response to "scroll through this page":\n'
                '  "[START_TASK:Scroll through page] Scrolling down now. [SCROLL:+1200]"\n'
                '  (Then I will send you the updated page content automatically, and you keep scrolling.)\n\n'
                'CORRECT response to "click the like button" (one-shot, no task loop needed):\n'
                '  "Clicking the like button. [CLICK:[aria-label=\\"Like\\"]]"\n\n'
                'WRONG response:\n'
                '  "I will scroll through the page for you." ← does nothing, no tag\n\n'
                'CSS selector hints for common sites:\n'
                '  Facebook:\n'
                '    Comment box: [contenteditable="true"][role="textbox"]  (NOT placeholder-based)\n'
                '    Post button: [aria-label="Comment"], div[aria-label="Comment"][role="button"]\n'
                '    Articles: [role="article"]\n'
                '    To comment on a post: first [CLICK] the "Comment" link under the post,\n'
                '    then [WAIT:1], then [FILL:[contenteditable="true"][role="textbox"]:your text],\n'
                '    then [WAIT:1], then press Enter: [CLICK:[aria-label="Comment"][role="button"]]\n'
                '  LinkedIn feed: .feed-shared-update-v2, [data-id]\n'
                '  Generic: article, .post, main\n\n'
                'Current page shown below. Canvas pages are separate HTML files YOU build inside the app — '
                'do NOT confuse them with external websites the user is browsing.]'
            )
            page_url   = ui_context.get('page_url', '')
            page_title = ui_context.get('page_title', '')
            page_text  = ui_context.get('page_text', '')
            sel_text   = ui_context.get('selected_text', '')
            actions    = ui_context.get('action_history', [])
            if page_url:
                context_parts.append(f'[Browser tab: "{page_title or "untitled"}" — {page_url}]')
            if page_text:
                context_parts.append(f'[Page content: {page_text[:4000]}]')
            interactive = ui_context.get('interactive', [])
            if interactive:
                lines = []
                for el in interactive[:30]:
                    t = el.get('t', '')
                    sel = el.get('sel', '')
                    if t == 'button':
                        lines.append(f'  BUTTON "{el.get("text","")}" → [CLICK:{sel}]')
                    elif t == 'link':
                        lines.append(f'  LINK "{el.get("text","")}" → {el.get("href","")}')
                    else:
                        hint = el.get('hint', '')
                        lines.append(f'  {t.upper()} {sel}{(" " + repr(hint)) if hint else ""} → [FILL:{sel}:your text]')
                context_parts.append('[Interactive elements on this page — use these exact selectors:\n' + '\n'.join(lines) + ']')
            if sel_text:
                context_parts.append(f'[User highlighted: "{sel_text}"]')
            if actions:
                recent = actions[-8:]
                acts = ' → '.join(
                    f"{a.get('type','?')} {(a.get('text') or a.get('url') or a.get('selector') or '')[:35]}"
                    for a in recent
                )
                context_parts.append(f'[Recent browser actions: {acts}]')

        # Canvas state
        if ui_context.get('canvasVisible') and ui_context.get('canvasDisplayed'):
            page_name = (ui_context['canvasDisplayed']
                         .replace('/pages/', '')
                         .replace('.html', '')
                         .replace('-', ' '))
            context_parts.append(f'[Canvas OPEN: {page_name}]')
            _ui_state_lines.append(f'Canvas OPEN: {page_name}')
        elif not ui_context.get('canvasVisible'):
            context_parts.append('[Canvas CLOSED]')
            _ui_state_lines.append('Canvas CLOSED')
        if ui_context.get('canvasMenuOpen'):
            context_parts.append('[Canvas menu visible to user]')
        # Canvas JS errors — auto-injected from browser error buffer
        canvas_errors = ui_context.get('canvasErrors', [])
        if canvas_errors:
            err_str = ' | '.join(canvas_errors)
            context_parts.append(f'[Canvas JS Errors: {err_str}]')

        # Music state (server-side is authoritative)
        _srv_track = _music_state.get('current_track')
        _srv_playing = _music_state.get('playing', False)
        if _srv_playing and _srv_track:
            _track_name = _srv_track.get('title') or _srv_track.get('name', 'unknown')
            context_parts.append(f'[Music PLAYING: {_track_name}]')
            _ui_state_lines.append(f'Music PLAYING: {_track_name}')
        elif _srv_track:
            _track_name = _srv_track.get('title') or _srv_track.get('name', 'unknown')
            context_parts.append(f'[Music PAUSED/STOPPED — last track: {_track_name}]')
            _ui_state_lines.append(f'Music PAUSED/STOPPED — last track: {_track_name}')
        elif ui_context.get('musicPlaying'):
            track = ui_context.get('musicTrack', 'unknown')
            context_parts.append(f'[Music PLAYING: {track}]')
            _ui_state_lines.append(f'Music PLAYING: {track}')

        # Available music tracks (so agent can use [MUSIC_PLAY:exact name])
        # Names come from a 10s TTL cache (_cached_music_names) so a full
        # dir-scan isn't paid on every turn — voice-latency finding #6.
        try:
            _lib_names = _cached_music_names('library')
            _gen_names = _cached_music_names('generated')
            _parts = []
            if _lib_names:
                _parts.append(f'Library ({len(_lib_names)}): {_cap_list(_lib_names, max_chars=2000)}')
            if _gen_names:
                _parts.append(f'Generated ({len(_gen_names)}): {_cap_list(_gen_names, max_chars=2000)}')
            if _parts:
                static_parts.append(f'[Available tracks — {" | ".join(_parts)}]')
        except Exception:
            pass

        # Authoritative "last track you made" — newest generated song by file
        # mtime (created_date metadata is date-only and useless for recency).
        # Gives the agent the exact filename + a ready-to-send download link so
        # it never guesses or hands out a broken /music/ URL.
        # STATIC (#94): only changes when a new song lands — the hash-change
        # resend in the context-bloat guard delivers it then, instead of
        # repeating ~150 chars on every turn of the session.
        try:
            from routes.music import get_latest_generated_track
            _latest = get_latest_generated_track()
            if _latest:
                static_parts.append(
                    f"[Latest generated song: {_latest['title']!r} "
                    f"(file: {_latest['filename']}) — "
                    f"download link: /generated_music/{_latest['filename']}?download=1]"
                )
        except Exception:
            pass

        # Co-browsing text-sight (#154 Phase 2): when a server-side browse session
        # is active for this tenant, give the agent the current page's url/title +
        # a DOM digest (visible text, links, buttons) so it can navigate without a
        # full screenshot round-trip. Dynamic (changes every action) → per-turn.
        try:
            from routes.browse import get_browse_state
            _bstate = get_browse_state()
            if _bstate:
                _links = ' | '.join(l for l in _bstate.get('links', []) if l)[:600]
                _btns = ' | '.join(b for b in _bstate.get('buttons', []) if b)[:400]
                _bparts = [
                    f"url={_bstate.get('url','')!r}",
                    f"title={_bstate.get('title','')!r}",
                ]
                if _bstate.get('text'):
                    _bparts.append(f"visible text: {_bstate['text'][:800]}")
                if _links:
                    _bparts.append(f"links: {_links}")
                if _btns:
                    _bparts.append(f"buttons: {_btns}")
                if _bstate.get("tab_count", 1) > 1:
                    _bparts.append(f"tabs open: {_bstate['tab_count']} "
                                   f"(active #{_bstate.get('active_tab', 0)})")
                if _bstate.get("last_download"):
                    _bparts.append(f"download captured: {_bstate['last_download']} "
                                   f"(saved to the user's uploads)")
                context_parts.append(
                    '[BROWSE_STATE (live server-side browser you are driving — the user '
                    'sees this too): ' + ' · '.join(_bparts) + '. Drive it by emitting '
                    '[BROWSE_ACTION:{"action":"click|type|scroll|goto|back|forward|reload|wait",...}] '
                    'tags; say what you are doing in words too.]'
                )
        except Exception as _be:
            logger.debug(f'BROWSE_STATE injection skipped: {_be}')

        # Recently completed Suno generations — agent gets notified on next turn
        try:
            from routes.suno import completed_songs_queue
            if completed_songs_queue:
                _pending = completed_songs_queue[-3:]
                _titles = [s.get('title', 'Unknown Track') for s in _pending]
                context_parts.append(f'[Suno just finished: {", ".join(repr(t) for t in _titles)} — now ready in Generated playlist]')
        except Exception:
            pass

        # Recently FAILED Suno generations are NOT injected here. This block used to append
        # them to context_parts on every turn and nothing ever cleared the queue, so one failure
        # would have been re-announced on every turn until a restart. Failures are now delivered
        # ONCE as a [SYSTEM] prefix on the next real user turn, next to the completion note
        # (search SUNO-REJECT-SURFACED below).

        # Available canvas pages (agent needs IDs for [CANVAS:page-id])
        try:
            from routes.canvas import load_canvas_manifest
            _manifest = load_canvas_manifest()
            _page_ids = sorted(_manifest.get('pages', {}).keys())
            _page_list = _cap_list(_page_ids, max_chars=5000)
        except Exception:
            _page_list = 'unknown'
        static_parts.append(f'[Canvas pages: {_page_list}]')

        # Available DJ sounds (for [SOUND:name] in DJ mode)
        static_parts.append(
            '[DJ sounds: air_horn, scratch_long, rewind, record_stop, '
            'crowd_cheer, crowd_hype, yeah, lets_go, gunshot, bruh, sad_trombone]'
        )
    # Inject active profile's custom system_prompt (admin editor → runtime)
    # Also read min_sentence_chars for TTS sentence extraction.
    _min_sentence_chars = 40  # default — prevents choppy short TTS fragments
    # First-audio latency cap (2026-07-16): the reply's FIRST spoken chunk is
    # extracted at min(_min_sentence_chars, this) so audio starts fast even on
    # profiles with large character-pacing minimums (kyle's 200-char min made
    # the first Resemble chunk 200-310 chars = 6-10s of synthesis while the
    # text was already on screen). Later chunks keep the profile minimum.
    _first_chunk_min_chars = 60
    _parallel_sentences = True  # default — fire all TTS in parallel threads
    _inter_sentence_gap_ms = 0  # default — no gap between audio chunks
    _prof = None  # default — referenced again in __session_start__ greeting branch below
    _profile_greeting = ''  # default — set in the __session_start__ branch; read by the
                            # empty-greeting fallback in stream_response (closure capture)
    # Persona greeting rotation defaults — MUST be initialized at function level:
    # the text_done capture hook in stream_response evaluates _grot_active on
    # EVERY turn (closure), and the assignments below only run in the
    # __session_start__ branch. Without these defaults every non-greeting turn
    # raised NameError mid-stream → 'Connection lost — retrying' (persona-tenant 2026-07-13).
    _grot_active = False
    _grot_state_path = ''
    _persona_greeting_prompt = ''
    _persona_fallbacks = []
    _current_user_name = ''  # signed-in Clerk user's name (set below); the
                             # __session_start__ brief greets them by it
    try:
        from profiles.manager import get_profile_manager
        from routes.profiles import _active_profile_id
        _mgr = get_profile_manager()
        _prof = _mgr.get_profile(_active_profile_id)
        if _prof and _prof.system_prompt and _prof.system_prompt.strip():
            static_parts.append(f'[PROFILE INSTRUCTIONS: {_prof.system_prompt.strip()}]')
        if _prof and hasattr(_prof, 'voice') and _prof.voice:
            _vc = _prof.voice
            if _vc.min_sentence_chars:
                _min_sentence_chars = _vc.min_sentence_chars
            if _vc.parallel_sentences is not None:
                _parallel_sentences = _vc.parallel_sentences
            if _vc.inter_sentence_gap_ms:
                _inter_sentence_gap_ms = _vc.inter_sentence_gap_ms
    except Exception:
        pass  # Profile system not available — skip gracefully

    # Greetings are short by design ("Hey, what's up?") — with the 40-char
    # minimum they never hit a mid-stream sentence boundary, so greeting TTS
    # only fired at text_done (issue #81: 5-38s first-audio latency on
    # session start). Fire on the first complete sentence regardless of length.
    if user_message == '__session_start__':
        _min_sentence_chars = 1

    # Inject [CURRENT_USER: ...] so the agent knows WHO is actually logged in
    # right now (regardless of which tenant account they're using). When Mike
    # the developer pops into a client tenant to debug, the agent should treat
    # him as a peer collaborator, not a client. See services/identity.py.
    # STATIC (#94): the tag + office briefing are stable within a session for
    # a given Clerk user — they used to repeat ~300-800 chars on EVERY turn
    # (the exact accumulation PR #297 was called out for in issue #94). A user
    # switch or briefing change flips the static-blob hash → full resend.
    try:
        from flask import g as _flask_g
        from services.identity import get_current_user_tag as _get_current_user_tag
        _clerk_uid = getattr(_flask_g, 'clerk_user_id', None)
        _tenant = os.getenv('JAMBOT_TENANT') or os.getenv('TENANT_NAME') or None
        _user_tag = _get_current_user_tag(_clerk_uid, _tenant)
        try:
            from services.identity import resolve as _resolve_identity
            _current_user_name = ((_resolve_identity(_clerk_uid) or {}).get('name') or '').strip()
        except Exception:
            _current_user_name = ''
        if _user_tag:
            static_parts.append(_user_tag)
            logger.info(f'### CURRENT_USER injected: clerk_uid={_clerk_uid} tenant={_tenant}')

        # Mesh-access gate — refresh the .mesh-admin-session marker on every
        # turn where the admin (Mike) is the authenticated Clerk user. The
        # mesh-send/mesh-recv wrappers in voice tenants check this marker
        # (300s TTL) before allowing any mesh operation. Voice tenants serving
        # a non-admin customer get NO mesh access; admin sessions get full
        # both-direction access. host/test-dev/webtops bypass the gate.
        # See /mnt/system/base/skills/agent-mesh/bin/mesh-gate-check.
        # The platform operator comes from PLATFORM_ADMIN_USER_IDS (services/auth.py);
        # no id is built into the source.
        from services.auth import is_platform_admin as _is_platform_admin
        if _is_platform_admin(_clerk_uid):
            try:
                _gate_path = '/app/runtime/uploads/.mesh-admin-session'
                _ts_iso = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
                with open(_gate_path, 'w', encoding='utf-8') as _gf:
                    _gf.write(f'admin={_clerk_uid}\nrefreshed_at={_ts_iso}\nsource=conversation.py auto-refresh\n')
            except Exception as _ge:
                logger.warning(f'mesh-gate refresh failed (non-fatal): {_ge}')
    except Exception as _e:
        logger.warning(f'CURRENT_USER injection failed (non-fatal): {_e}')

    # Inject [CANVAS_STYLE: ...] when the tenant has an active canvas style.
    # Agents write page files directly (bypassing the pages API) and long
    # sessions never re-read skills, so soft skill guidance is not enough —
    # this per-turn tag is the enforcement that actually reaches the agent.
    try:
        from services.canvas_styles import get_active_style_id as _get_style_id
        _style_id = _get_style_id()
        if _style_id:
            static_parts.append(
                f'[CANVAS_STYLE: The user chose the "{_style_id}" design system for all '
                f'canvas pages. BEFORE creating or restyling ANY canvas page you MUST '
                f'read canvas-styles/ACTIVE-STYLE.md inside your canvas-pages directory '
                f'(OpenClaw: /app/runtime/canvas-pages/canvas-styles/ACTIVE-STYLE.md; '
                f'Hermes: /workspace/canvas-pages/canvas-styles/ACTIVE-STYLE.md) and start '
                f'by cp-ing canvas-styles/active-template.html to the new page file, then '
                f'edit the copy (keep its full <style> block — never retype the template). '
                f'Never use the old dark default. '
                f'Never use emoji as UI icons — use inline SVG.]'
            )
    except Exception as _se:
        logger.warning(f'CANVAS_STYLE injection failed (non-fatal): {_se}')

    # Inject voice assistant instructions so the agent knows about action tags.
    # This must be in-app (not workspace files) so it works out of the box.
    static_parts.append(_load_voice_system_prompt())

    # ── Context-bloat guard: dedup the static block (see module docstring at
    # _ctx_static_cache). Full block goes out when its content changes, on
    # __session_start__, or as a periodic refresh; otherwise a 1-line pointer.
    if static_parts:
        _static_blob = ' '.join(static_parts)

        # #94: mirror the full static block + current canvas/music state to
        # uploads/UI_STATE.md in the agent workspace (atomic, hash-skipped).
        # This durable copy is what makes the pointer below recoverable after
        # openclaw auto-compaction strips old history — see services/ui_state.py.
        # Gated on ui_context: turns without browser UI state (API/proactive)
        # build a listless static blob and must not clobber the good file.
        if ui_context:
            try:
                from services.paths import UPLOADS_DIR as _ui_uploads_dir
                from services.ui_state import write_ui_state as _write_ui_state
                _write_ui_state(_ui_uploads_dir, _static_blob, _ui_state_lines)
            except Exception as _us_e:
                logger.warning(f'UI_STATE.md mirror failed (non-fatal): {_us_e}')
        try:
            import hashlib as _hl
            _ctx_key = f'{session_id}:{get_voice_session_key()}'
            _static_hash = _hl.sha1(_static_blob.encode('utf-8', 'replace')).hexdigest()
            _now_ctx = time.time()
            _rec = _ctx_static_cache.get(_ctx_key)
            _resend_full = (
                user_message == '__session_start__'
                or _rec is None
                or _rec.get('hash') != _static_hash
                or _rec.get('skips', 0) + 1 >= _CTX_STATIC_RESEND_TURNS
                or _now_ctx - _rec.get('ts', 0) > _CTX_STATIC_RESEND_SECS
            )
            if _resend_full:
                context_parts.append(_static_blob)
                _ctx_static_cache[_ctx_key] = {'hash': _static_hash, 'ts': _now_ctx, 'skips': 0}
                if len(_ctx_static_cache) > _CTX_STATIC_CACHE_MAX:
                    for _old_k in sorted(_ctx_static_cache,
                                         key=lambda k: _ctx_static_cache[k].get('ts', 0))[:50]:
                        _ctx_static_cache.pop(_old_k, None)
            else:
                _rec['skips'] = _rec.get('skips', 0) + 1
                context_parts.append(
                    '[Standing instructions (voice action tags, canvas page list, track list, '
                    'profile instructions, canvas style, current user) were already provided '
                    'earlier this session, are UNCHANGED, and still apply. A full always-current '
                    'copy lives at uploads/UI_STATE.md in your workspace — re-read it if you no '
                    'longer remember them.]'
                )
                logger.info(f'### CONTEXT DEDUP: skipped {len(_static_blob)} static chars '
                            f'(skip {_rec["skips"]}/{_CTX_STATIC_RESEND_TURNS})')
        except Exception as _ctx_e:
            # Fail open — never let the guard cost a turn its instructions.
            logger.warning(f'### CONTEXT DEDUP failed (sending full block): {_ctx_e}')
            context_parts.append(_static_blob)

    if context_parts:
        context_prefix = ' '.join(context_parts) + ' '

    t_context_ms = int((time.time() - t_context_start) * 1000)
    if t_context_ms > 50:
        logger.info(f"### CONTEXT BUILD TIMING: {t_context_ms}ms ({len(context_parts)} parts, {len(context_prefix)} chars)")

    log_conversation('user', user_message, session_id=session_id,
                     tts_provider=tts_provider, voice=voice)

    # Replace the legacy __session_start__ sentinel with a natural-language greeting
    # prompt so the LLM produces a real greeting instead of a system sentinel ("NO").
    # user_message is kept as-is so the sentinel suppression logic still works.
    #
    # 2026-10-27: the greeting is ALWAYS generated by the agent: a contextual brief
    # (name + the most relevant live thing). A profile conversation.greeting is only
    # a style hint; it is never sent as a verbatim say-this-sentence order any more.
    if user_message == '__session_start__':
        logger.info(f"### CALL_START session={session_id}")
        _face = identified_person or {}
        _face_name = _face.get('name', '') if _face.get('name', '') != 'unknown' else ''
        _profile_greeting = ''
        try:
            if _prof and getattr(_prof, 'conversation', None):
                _profile_greeting = (getattr(_prof.conversation, 'greeting', '') or '').strip()
        except Exception:
            _profile_greeting = ''
        # ENTERTAINMENT-PERSONA greeting (opt-in per profile, e.g. persona-tenant).
        # A profile that defines conversation.greeting_angles wants a FRESH, in-character
        # greeting every single wake-up — never one canned line, and never the same TOPIC
        # twice in a row. We ROTATE through the profile's angles (a persisted index that
        # advances each session) and ask the agent to riff a brand-new bit in that angle,
        # in its OWN real character (greeting_persona anchors the voice), while avoiding the
        # last few openers. The greeting is produced by the gateway agent that loads the
        # real persona — so the humour comes from the character, not a thin caricature here.
        # Gated entirely on greeting_angles → any tenant without it is completely unaffected.
        # (persona-tenant 2026-07-13 — replaces the wrong 'King of Entrepreneurship' caricature.)
        _grot_active = False
        _grot_state_path = ''
        _persona_greeting_prompt = ''
        _persona_fallbacks = []
        try:
            import json as _json_gp, os as _os_gp
            _pf_gp = f"/app/runtime/profiles/{_active_profile_id}.json"
            _pconf = {}
            if _os_gp.path.exists(_pf_gp):
                _pconf = (_json_gp.load(open(_pf_gp)).get('conversation') or {})
            _persona = (_pconf.get('greeting_persona') or '').strip()
            _angles = [a for a in (_pconf.get('greeting_angles') or [])
                       if isinstance(a, str) and a.strip()]
            _persona_fallbacks = [g for g in (_pconf.get('fallback_greetings') or [])
                                  if isinstance(g, str) and g.strip()]
            if _angles:
                _grot_state_path = f"/app/runtime/profiles/.greeting-state-{_active_profile_id}.json"
                _st = {}
                try:
                    if _os_gp.path.exists(_grot_state_path):
                        _st = _json_gp.load(open(_grot_state_path)) or {}
                except Exception:
                    _st = {}
                _idx = int(_st.get('i', 0)) % len(_angles)
                _angle = _angles[_idx].strip()
                _recent = [r for r in (_st.get('recent') or [])
                           if isinstance(r, str) and r.strip()][-4:]
                # advance + persist the index NOW so rotation moves forward even if the
                # later text-capture never runs (empty turn, error, etc.)
                try:
                    _st['i'] = (_idx + 1) % len(_angles)
                    _st['recent'] = _recent
                    _tmp_gp = _grot_state_path + '.tmp'
                    with open(_tmp_gp, 'w') as _fh_gp:
                        _json_gp.dump(_st, _fh_gp)
                    _os_gp.replace(_tmp_gp, _grot_state_path)
                except Exception:
                    pass
                _avoid = ''
                if _recent:
                    _joined = ' | '.join(r[:120] for r in _recent)
                    _avoid = f" Do NOT reuse or closely echo any of these recent openers: {_joined}."
                # Per-profile bit style: a profile can define conversation.greeting_bit_style
                # to control the SHAPE of the opener (length, pacing, sound tags). Without it,
                # the default stays the short punchy 40-word greeting. (persona-tenant wants the
                # full episode-style cold-open ramble — 2026-07-14.)
                _bit_style = (_pconf.get('greeting_bit_style') or '').strip()
                if not _bit_style:
                    _bit_style = (
                        "Give a genuinely funny, in-character wake-up greeting — 2 to 3 "
                        "sentences, about 40 words MAX (it's spoken aloud, so keep it punchy — "
                        "no long monologue). Ramble a little into ONE quick tangent, then greet "
                        "them. Spoken only, no markdown, no stage directions."
                    )
                _persona_greeting_prompt = (
                    f"{_persona}\n\n"
                    f"A new voice session just started — someone woke you up. {_bit_style} "
                    f"{_angle}{_avoid} Make it a completely FRESH bit in your real voice — not a repeat, "
                    f"not a generic AI joke. Output ONLY the greeting."
                )
                _grot_active = True
        except Exception:
            _grot_active = False
        # The agent ALWAYS generates the greeting (Rule Book:
        # mikes-decisions/greetings-are-real-agent-turns). The profile greeting is a
        # style hint only; the persona rotation prompt keeps its angle/bit style and
        # gains the same who-is-this + live-context guidance.
        _gateway_message = _build_session_start_message(
            face_name=_face_name,
            current_user_name=_current_user_name,
            profile_greeting=_profile_greeting,
            persona_prompt=_persona_greeting_prompt if _grot_active else '',
        )
        logger.info(
            f"### SESSION_START brief: speaker={_session_start_speaker(_face_name, _current_user_name)!r} "
            f"persona={bool(_grot_active)} style_hint={bool(_profile_greeting)}"
        )
    else:
        _gateway_message = user_message

    # Suno completion → inject as [SYSTEM] prefix on the NEXT real user turn so
    # the agent sees the event in the same conversation turn as the user's reply.
    # This replaces the old ghost-LLM `__suno_complete__` path that left the main
    # session without context when the user said "yeah" to play the song.
    _suno_prefix = ''
    if user_message not in ('__session_start__',):
        try:
            from routes.suno import completed_songs_queue as _suno_q
            if _suno_q:
                _pending_titles = [s.get('title', 'Unknown Track') for s in list(_suno_q)]
                if _pending_titles:
                    _titles_str = ', '.join(f'"{t}"' for t in _pending_titles)
                    _play_tag = _pending_titles[0]  # most recent — what "yeah/play it" refers to
                    _suno_prefix = (
                        f'[SYSTEM: Suno just finished generating {_titles_str} and they are now '
                        f'loaded in the Generated playlist. If the user is asking to play the song '
                        f'(e.g. "yeah", "play it", "let\'s hear it"), confirm briefly and emit '
                        f'[MUSIC_PLAY:{_play_tag}] in your response.]\n\n'
                    )
                    # Pop so the note only fires on the turn immediately after completion.
                    _suno_q.clear()
        except Exception as _e:
            logger.warning(f'Suno pending-note injection failed: {_e}')

    # SUNO-REJECT-SURFACED (2026-09-29): see _suno_failure_prefix().
    if user_message not in ('__session_start__',):
        _suno_prefix = _suno_failure_prefix() + _suno_prefix

    _gateway_message_with_suno = _suno_prefix + _gateway_message if _suno_prefix else _gateway_message
    message_with_context = context_prefix + _gateway_message_with_suno if context_prefix else _gateway_message_with_suno
    ai_response = None
    # True when the gateway ended the turn on purpose with no reply
    # (text_done deferred=True) — see the streaming DEFERRED branch.
    _deferred_turn = False
    captured_actions = []

    # ── PRIMARY PATH: Gateway (routed by gateway_id from request/profile) ──
    if gateway_manager.is_configured():
        try:
            logger.info('### Starting Gateway connection...')
            event_queue: queue.Queue = queue.Queue()
            _session_key = get_voice_session_key()

            # Check if gateway recently reconnected after a failure —
            # inject a system note so the agent acknowledges the interruption
            _recovery_prefix = ''
            try:
                _gw = gateway_manager.get(gateway_id)
                if _gw and hasattr(_gw, 'consume_reconnection') and _gw.consume_reconnection():
                    _recovery_prefix = (
                        '[SYSTEM: The connection was briefly interrupted (server restart). '
                        'Briefly acknowledge this to the user before responding to their message.]\n\n'
                    )
                    logger.info('### Injecting recovery prefix into message')
            except Exception:
                pass

            # ── Session-recovery context prime ────────────────────────
            # After a double-empty that flipped us to the 'recovery' key,
            # prepend a compressed history summary to the very FIRST request
            # so the fresh openclaw session has memory of the prior turns.
            # Skip on __session_start__: injecting old context onto a greeting
            # request causes the agent to pick up stale work instead of
            # greeting. CRITICAL: skip means LEAVE THE PRIME ARMED, not
            # consume-and-discard. After a drop the user's next request is
            # almost always __session_start__ (they re-open the call) — the
            # old consume-then-null code burned the prime on the greeting, so
            # every turn after the "Hi, can I help you?" ran on a context-free
            # session. That was the primary amnesia path.
            _recovery_prime = None
            if user_message == '__session_start__':
                if _recovery_context_prime is not None:
                    logger.info(
                        '### Holding recovery prime through __session_start__ '
                        '(greeting goes out prime-free; prime stays armed for '
                        'the first real user turn)'
                    )
            else:
                _recovery_prime = consume_recovery_prime()
                if _recovery_prime:
                    logger.info(f'### Injecting session-recovery prime ({len(_recovery_prime)} chars)')

            def _run_gateway():
                _msg = message_with_context
                if _recovery_prefix:
                    _msg = _recovery_prefix + _msg
                if _recovery_prime:
                    _msg = _recovery_prime + _msg
                gateway_manager.stream_to_queue(
                    event_queue, _msg, _session_key, captured_actions,
                    gateway_id=gateway_id,
                    agent_id=agent_id,
                )

            t_llm_start = time.time()
            gw_thread = threading.Thread(target=_run_gateway, daemon=True)
            gw_thread.start()

            if wants_stream:
                # ── STREAMING MODE ────────────────────────────────────────
                _stream_state = {'done_q': None, 'done_evt': None}

                def stream_response():
                    nonlocal ai_response, event_queue, t_llm_start

                    # ── TTS helpers ───────────────────────────────────────
                    try:
                        _prov = get_provider(tts_provider)
                        _audio_fmt = _prov.get_info().get('audio_format', 'wav')
                    except Exception:
                        _audio_fmt = 'wav'

                    def _detect_audio_fmt(audio_b64):
                        """Sniff the real container from the audio header (TTS-11).

                        After a sticky/transitive fallback the actual provider can
                        switch mp3↔wav mid-response; labeling every chunk with the
                        PRIMARY provider's format then lies and breaks decoding. Read
                        the magic bytes so each chunk is labeled with what it actually
                        is. Falls back to the primary provider's format on any doubt.
                        """
                        try:
                            head = base64.b64decode(audio_b64[:32])
                        except Exception:
                            return _audio_fmt
                        if head[:4] == b'RIFF' and head[8:12] == b'WAVE':
                            return 'wav'
                        if head[:3] == b'ID3' or (len(head) >= 2 and head[0] == 0xFF and (head[1] & 0xE0) == 0xE0):
                            return 'mp3'
                        if head[:4] == b'OggS':
                            return 'ogg'
                        return _audio_fmt

                    def _tts_error_event(err_str):
                        # CLIENT-FACING RULE (Mike 2026-06-28): never serialize raw provider
                        # name, billing reason, or exception text into a client-bound event.
                        # Classify to a neutral reason CODE only (used for client-side logic /
                        # devtools log) — the raw string is logged server-side here, never sent.
                        code_match = re.search(r'\[groq:([^\]]+)\]', err_str)
                        err_code = code_match.group(1) if code_match else 'unknown'
                        REASON_CODES = {
                            'model_terms_required': 'terms',
                            'rate_limit_exceeded':  'rate_limit',
                            'insufficient_quota':   'no_credits',
                            'invalid_api_key':      'bad_key',
                            'unknown':              'error',
                        }
                        reason_key = REASON_CODES.get(err_code, 'error')
                        logger.warning(f"### TTS error (not sent to client): reason={reason_key} raw={err_str!r}")
                        return json.dumps({
                            'type': 'tts_error',
                            'reason': reason_key,
                        }) + '\n'

                    # ── Mid-stream TTS helpers ────────────────────────────
                    def _has_open_tag(text):
                        """True while inside an incomplete [...] action tag or open code fence."""
                        if text.count('[') > text.count(']'):
                            return True
                        # Odd number of ``` markers means we're inside a code block
                        if text.count('```') % 2 != 0:
                            return True
                        return False

                    _extract_sentence = extract_tts_sentence

                    # Sticky fallback state — shared across all TTS calls in this response.
                    # If one sentence falls back to a different provider, all subsequent
                    # sentences use the same fallback to keep the voice consistent.
                    _tts_fallback_state = {}

                    # --- Dynamic status: describe what the agent is actually doing ---
                    _TOOL_DESCRIPTIONS = {
                        'exec': 'running some code',
                        'web_search': 'searching the web',
                        'web_fetch': 'reading a webpage',
                        'sessions_spawn': 'delegating to a sub-agent',
                        'sessions_send': 'coordinating with another agent',
                        'Edit': 'editing a file',
                        'Write': 'writing a file',
                        'Read': 'reading a file',
                        'Glob': 'searching for files',
                        'Grep': 'searching through code',
                    }

                    def _build_dynamic_status(tool_name, tools_count, silence_secs):
                        """Build a contextual status message from what the agent is doing."""
                        if tool_name:
                            desc = _TOOL_DESCRIPTIONS.get(tool_name)
                            if desc:
                                return f"Still here, just {desc}."
                            # Unknown tool — use the name directly if it's readable
                            if tool_name and len(tool_name) < 30 and tool_name.isalnum():
                                return f"Still working, running {tool_name}."
                        if tools_count > 3:
                            return "Going through a few steps, almost there."
                        return "Still on it, one moment."

                    def _fire_tts(raw_text):
                        """Start TTS for raw_text. Parallel or sequential per profile config.
                        Returns (done_event, result)."""
                        done = threading.Event()
                        result = {'audio': None, 'error': None}
                        if skip_tts:
                            done.set()
                            return (done, result)
                        def _run():
                            try:
                                t0 = time.time()
                                cleaned = clean_for_tts(raw_text)
                                t_clean = time.time()
                                if cleaned and cleaned.strip():
                                    result['audio'] = _tts_generate_b64(
                                        cleaned, voice=voice or 'M1',
                                        tts_provider=tts_provider,
                                        fallback_state=_tts_fallback_state
                                    )
                                t_done = time.time()
                                logger.info(
                                    f"### TTS TIMING: clean={int((t_clean-t0)*1000)}ms "
                                    f"generate={int((t_done-t_clean)*1000)}ms "
                                    f"total={int((t_done-t0)*1000)}ms "
                                    f"text={len(cleaned or '')} chars"
                                )
                            except Exception as e:
                                result['error'] = str(e)
                            finally:
                                done.set()
                                # Wake the stream loop NOW so finished audio is
                                # flushed immediately instead of sitting in
                                # _tts_pending until the next gateway event or
                                # queue-poll timeout (up to 10s during quiet
                                # tool runs). Handled as a no-op flush trigger.
                                try:
                                    event_queue.put({'type': 'tts_ready'})
                                except Exception:
                                    pass
                        if _parallel_sentences:
                            threading.Thread(target=_run, daemon=True).start()
                        else:
                            _run()  # sequential — block until this sentence is done
                        return done, result

                    def _audio_event(audio_b64, chunk_idx, tts_ms=0, fmt=None):
                        """Build an audio SSE event dict with gap config."""
                        evt = {
                            'type': 'audio',
                            'audio': audio_b64,
                            'audio_format': fmt or _detect_audio_fmt(audio_b64),
                            'chunk': chunk_idx,
                            'total_chunks': None,
                            'timing': {
                                'tts_ms': tts_ms,
                                'total_ms': int((time.time() - t_request_start) * 1000),
                            },
                        }
                        if _inter_sentence_gap_ms:
                            evt['gap_ms'] = _inter_sentence_gap_ms
                        return json.dumps(evt) + '\n'

                    # Mid-stream TTS state
                    _tts_buf = ''       # raw incremental text buffer
                    _tts_pending = []   # [(done_event, result_dict), ...]
                    _chunks_sent = 0    # audio chunks already yielded early
                    # True once THIS answer leg's own text was sent to TTS from
                    # streamed deltas. _chunks_sent alone cannot answer "was the
                    # answer already spoken?": it also counts the spoken status
                    # and the previous leg's acknowledgement, and the whole-reply
                    # fallbacks below then skipped a reply that arrived without
                    # deltas — the result went unspoken (pledge 25dfb03c).
                    _answer_tts_fired = False

                    # ── Spoken status updates during long tool execution ──
                    # Prevents dead silence when agent runs tools for 30-90+ seconds.
                    _last_audio_time = time.time()   # last time we sent audio to browser
                    _status_tts_count = 0            # how many status messages spoken
                    _tools_seen = 0                  # count of tool starts seen
                    _last_tool_name = None           # name of the most recent tool
                    _STATUS_SILENCE_THRESHOLD = 45   # long safety net — agent should speak naturally
                    _STATUS_REPEAT_INTERVAL = 90     # almost never repeat
                    # Delegated (sub-agent) turns get their own status, built
                    # only from what the stream showed (services/interim_status.py,
                    # pledge 25dfb03c). It is checked on EVERY loop wake-up, not
                    # only on gateway heartbeats: while a sub-agent works its
                    # tool events stream in continuously, so the heartbeat that
                    # the legacy check waits for never arrives (2026-09-16:
                    # ~9 min of silence on a delegated research ask).
                    _delegated_status = DelegatedStatus()

                    def _delegated_status_text():
                        if _tts_pending:
                            return None  # real audio is already on its way
                        return _delegated_status.due(
                            time.time(), _last_audio_time, tts_enabled=not skip_tts)

                    full_response = None
                    _stream_start = time.time()
                    _STREAM_HARD_TIMEOUT = 310  # seconds — total allowed time
                    _QUEUE_POLL_INTERVAL = 10   # seconds — yield heartbeat if no events
                    while True:
                        try:
                            evt = event_queue.get(timeout=_QUEUE_POLL_INTERVAL)
                        except queue.Empty:
                            # No events for _QUEUE_POLL_INTERVAL seconds.
                            # Yield a heartbeat to keep the browser/Cloudflare
                            # connection alive (they time out at 60-100s of silence).
                            elapsed = int(time.time() - _stream_start)
                            if elapsed > _STREAM_HARD_TIMEOUT:
                                # Re-arm a consumed recovery prime — this turn
                                # never completed, so context must carry forward.
                                restore_recovery_prime(_recovery_prime)
                                # CLIENT-FACING RULE (Mike 2026-06-28): neutral code only.
                                logger.error('### STREAM HARD TIMEOUT (not sent to client): Gateway timeout')
                                yield json.dumps({'type': 'error', 'code': 'timeout'}) + '\n'
                                break
                            yield json.dumps({'type': 'heartbeat', 'elapsed': elapsed}) + '\n'
                            _ds_text = _delegated_status_text()
                            if _ds_text:
                                logger.info(f"### DELEGATED STATUS TTS ({_delegated_status.spoken}): '{_ds_text}' (queue idle)")
                                _tts_pending.append(_fire_tts(_ds_text))
                                _last_audio_time = time.time()
                            continue

                        # Proof-of-life on the recovery session — reset the idle
                        # timer so a productive multi-tool recovery turn isn't
                        # kicked out mid-work by the elapsed-time check.
                        if _session_recovery_key is not None:
                            bump_recovery_activity()

                        if evt['type'] == 'handshake':
                            metrics['handshake_ms'] = evt['ms']
                            continue

                        if evt['type'] == 'tts_ready':
                            # A TTS thread finished — flush any audio that's ready,
                            # in order, without waiting for gateway event cadence.
                            while _tts_pending and _tts_pending[0][0].is_set():
                                _done_evt, _res = _tts_pending.pop(0)
                                if _res.get('error'):
                                    yield _tts_error_event(_res['error'])
                                elif _res.get('audio'):
                                    yield _audio_event(_res['audio'], _chunks_sent)
                                    _chunks_sent += 1
                                    _last_audio_time = time.time()
                            continue

                        if evt['type'] == 'heartbeat':
                            logger.info(f"### HEARTBEAT → browser ({evt.get('elapsed', 0)}s)")
                            yield json.dumps({'type': 'heartbeat', 'elapsed': evt.get('elapsed', 0)}) + '\n'
                            # Flush any TTS that finished during tool execution —
                            # without this, audio sits in _tts_pending for the
                            # entire duration of tool calls (30-60s+ silence).
                            _flushed_audio = False
                            while _tts_pending and _tts_pending[0][0].is_set():
                                _done_evt, _res = _tts_pending.pop(0)
                                if _res.get('error'):
                                    yield _tts_error_event(_res['error'])
                                elif _res.get('audio'):
                                    yield _audio_event(_res['audio'], _chunks_sent)
                                    _chunks_sent += 1
                                    _last_audio_time = time.time()
                                    _flushed_audio = True
                            # ── Spoken status: break silence during long tool execution ──
                            # If tools are running and user has heard nothing for too long,
                            # speak a brief status so they know the agent is alive.
                            _ds_text = _delegated_status_text()
                            if _ds_text:
                                logger.info(f"### DELEGATED STATUS TTS ({_delegated_status.spoken}): '{_ds_text}' (heartbeat)")
                                _tts_pending.append(_fire_tts(_ds_text))
                                _last_audio_time = time.time()
                            elif (not _flushed_audio and _tools_seen > 0 and not skip_tts
                                    and not _delegated_status.delegated):
                                _silence_secs = time.time() - _last_audio_time
                                _threshold = (
                                    _STATUS_SILENCE_THRESHOLD if _status_tts_count == 0
                                    else _STATUS_REPEAT_INTERVAL
                                )
                                if _silence_secs >= _threshold:
                                    # Dynamic status from what the agent is actually doing
                                    _status_text = _build_dynamic_status(
                                        _last_tool_name, _tools_seen, _silence_secs
                                    )
                                    logger.info(f"### STATUS TTS ({_status_tts_count}): '{_status_text}' (silence={_silence_secs:.0f}s)")
                                    # Non-blocking: queue it like a normal sentence and
                                    # let the tts_ready flush deliver it. Blocking here
                                    # held the whole stream (deltas, real audio) hostage
                                    # for up to 10s while filler audio generated.
                                    _tts_pending.append(_fire_tts(_status_text))
                                    # Reset the silence clock at fire time so the next
                                    # heartbeat doesn't double-fire while this generates.
                                    _last_audio_time = time.time()
                                    _status_tts_count += 1
                            continue

                        if evt['type'] == 'delta':
                            _tts_buf += evt['text']
                            if (evt.get('text') or '').strip():
                                _delegated_status.note_answer_text()
                            # Don't fire TTS if buffer looks like a system response
                            # that will be suppressed at text_done. Wait for final
                            # confirmation before speaking.
                            _buf_stripped = _tts_buf.strip()
                            # Suppress system responses — uses regex from compat layer
                            # plus partial match for mid-stream detection
                            _is_system_text = (
                                is_system_response(_buf_stripped)
                                or _buf_stripped.upper().startswith('HEARTBEAT')
                            )
                            # Fire TTS for complete sentences as they arrive
                            if not _is_system_text and not _has_open_tag(_tts_buf):
                                _eff_min = (
                                    min(_min_sentence_chars, _first_chunk_min_chars)
                                    if (_chunks_sent == 0 and not _tts_pending)
                                    else _min_sentence_chars
                                )
                                sentence, _tts_buf = _extract_sentence(_tts_buf, min_len=_eff_min)
                                if sentence:
                                    logger.info(f"### TTS sentence (streaming): {sentence[:80]}")
                                    _tts_pending.append(_fire_tts(sentence))
                                    _answer_tts_fired = True
                            yield json.dumps({'type': 'delta', 'text': evt['text']}) + '\n'
                            # Flush any TTS chunks that finished while text was streaming —
                            # play audio as soon as it's ready instead of waiting for text_done
                            while _tts_pending and _tts_pending[0][0].is_set():
                                _done_evt, _res = _tts_pending.pop(0)
                                if _res.get('error'):
                                    yield _tts_error_event(_res['error'])
                                elif _res.get('audio'):
                                    yield _audio_event(_res['audio'], _chunks_sent)
                                    _chunks_sent += 1
                                    _last_audio_time = time.time()
                            continue

                        if evt['type'] == 'action':
                            # Track tool starts for dynamic status
                            if evt.get('action', {}).get('phase') == 'start':
                                _tools_seen += 1
                                _last_tool_name = evt.get('action', {}).get('name', None)
                            _delegated_status.note_action(evt.get('action'))
                            # Flush any TTS chunks that already finished —
                            # avoids silence during long tool calls (the first
                            # sentence TTS completes ~1s in but would otherwise
                            # wait until text_done which can be minutes away).
                            while _tts_pending and _tts_pending[0][0].is_set():
                                _done_evt, _res = _tts_pending.pop(0)
                                if _res.get('error'):
                                    yield _tts_error_event(_res['error'])
                                elif _res.get('audio'):
                                    yield _audio_event(_res['audio'], _chunks_sent)
                                    _chunks_sent += 1
                                    _last_audio_time = time.time()
                            yield json.dumps({'type': 'action', 'action': evt['action']}) + '\n'
                            _ds_text = _delegated_status_text()
                            if _ds_text:
                                logger.info(f"### DELEGATED STATUS TTS ({_delegated_status.spoken}): '{_ds_text}' (sub-agent activity)")
                                _tts_pending.append(_fire_tts(_ds_text))
                                _last_audio_time = time.time()
                            continue

                        if evt['type'] == 'queued':
                            StatusModule_hack = True  # just yield to browser
                            yield json.dumps({'type': 'queued'}) + '\n'
                            continue

                        if evt['type'] == 'text_interim':
                            # Agent spoke but sub-agents still running.
                            # Process TTS for this text but keep stream open.
                            interim_response = evt.get('response', '')
                            _delegated_status.note_interim()
                            logger.info(
                                f"### TEXT_INTERIM: {len(interim_response)} chars "
                                f"— sub-agents still working, stream stays open"
                            )

                            # Yield interim event to frontend
                            yield json.dumps({
                                'type': 'text_interim',
                                'response': interim_response,
                                'actions': evt.get('actions', []),
                            }) + '\n'

                            # Flush any buffered TTS text from streaming deltas
                            _remaining_interim = _tts_buf.strip()
                            if _remaining_interim:
                                _tts_pending.append(_fire_tts(_remaining_interim))
                                _tts_buf = ''

                            # Fallback: no sentences extracted mid-stream, fire TTS
                            # for the full interim text. Guard on _chunks_sent, NOT
                            # just _tts_pending (same double-audio bug text_done
                            # fixed 2026-07-12): the delta path pops finished
                            # sentences from _tts_pending as their audio is yielded,
                            # so when the interim event arrives after a fast sentence
                            # the list is already empty and this re-spoke the WHOLE
                            # reply a second time (measured live on test-dev
                            # 2026-09-15 — same 217-char sentence TTS'd twice).
                            if (not _tts_pending and interim_response
                                    and (_chunks_sent == 0 or not _answer_tts_fired)):
                                tts_text_interim = clean_for_tts(interim_response)
                                if tts_text_interim and tts_text_interim.strip():
                                    _tts_pending.append(_fire_tts(tts_text_interim))

                            # Flush all pending TTS audio immediately. Bounded per-chunk
                            # wait (TTS-2) so a hung provider can't stall the interim
                            # flush; a failed chunk is skipped, not fatal (already if/elif).
                            _interim_timeout = int(os.getenv('TTS_SENTENCE_TIMEOUT', '20'))
                            for _done_i, _res_i in _tts_pending:
                                _done_i.wait(timeout=_interim_timeout)
                                if _res_i.get('error'):
                                    yield _tts_error_event(_res_i['error'])
                                elif _res_i.get('audio'):
                                    yield _audio_event(_res_i['audio'], _chunks_sent)
                                    _chunks_sent += 1
                                    # Reset silence clock — agent already spoke, don't
                                    # fire "one moment" status TTS right after interim.
                                    _last_audio_time = time.time()
                                    _status_tts_count += 1  # suppress first status fire
                            _tts_pending = []
                            _tts_buf = ''
                            # What streams next is a NEW answer leg.
                            _answer_tts_fired = False

                            # Tell frontend sub-agents are actively working
                            yield json.dumps({
                                'type': 'subagents_working',
                            }) + '\n'

                            # Extend hard timeout for sub-agent wait phase
                            _stream_start = time.time()
                            _STREAM_HARD_TIMEOUT = 600
                            continue

                        if evt['type'] == 'text_done':
                            # stream-offline-relay: remember the terminal event
                            # of the turn running on THIS queue (a retry or
                            # auto-continue swaps event_queue for a new one).
                            _stream_state['done_q'] = event_queue
                            _stream_state['done_evt'] = evt
                            _stream_state['done_llm_ms'] = int((time.time() - t_llm_start) * 1000)
                            logger.info(f"### TEXT_DONE received. response={len(evt.get('response', '') or '')} chars, _tts_pending={len(_tts_pending)}, _tts_buf={repr(_tts_buf[:80])}")
                            # ── DEFERRED turn: the gateway ended this turn on purpose ──
                            # A gateway sets deferred=True with an empty response when
                            # the agent is still holding the words and will answer later
                            # through its own proactive push channel. That is NOT a
                            # failure: no empty-retry, no fallback LLM, no substituted
                            # line, no TTS (NO FAKE REPLIES — Mike 2026-10-20). End the
                            # turn silently, tell the frontend there is no audio, log
                            # metrics, and save nothing (there is no assistant turn yet).
                            # Gateways that never set `deferred` never enter this branch.
                            if evt.get('deferred') and not evt.get('error') \
                                    and not (evt.get('response') or '').strip():
                                metrics['llm_inference_ms'] = int((time.time() - t_llm_start) * 1000)
                                metrics['tool_count'] = sum(
                                    1 for a in captured_actions
                                    if a.get('type') == 'tool' and a.get('phase') == 'start'
                                )
                                metrics['profile'] = 'gateway-deferred'
                                metrics['deferred'] = 1
                                metrics['response_len'] = 0
                                if _tts_pending or _tts_buf.strip():
                                    logger.info(
                                        f"### DEFERRED: discarding {len(_tts_pending)} pending TTS "
                                        f"chunk(s) + buffer {len(_tts_buf)} chars (deferred turn speaks nothing)"
                                    )
                                _tts_buf = ''
                                _tts_pending = []
                                logger.info(
                                    f"### DEFERRED TURN — gateway={gateway_id or 'openclaw'} "
                                    f"ended the turn with no reply on purpose "
                                    f"({metrics['llm_inference_ms']}ms, tools={metrics['tool_count']}); "
                                    f"answer will arrive via push — no retry, no fallback, no TTS"
                                )
                                yield json.dumps({
                                    'type': 'text_done',
                                    'response': '',
                                    'deferred': True,
                                    'actions': captured_actions,
                                    'timing': {
                                        'handshake_ms': metrics.get('handshake_ms'),
                                        'llm_ms': metrics.get('llm_inference_ms'),
                                    }
                                }) + '\n'
                                yield json.dumps({'type': 'no_audio', 'deferred': True}) + '\n'
                                metrics['total_ms'] = int((time.time() - t_request_start) * 1000)
                                log_metrics(metrics)
                                break

                            # Handle LLM/gateway errors with a spoken fallback
                            if evt.get('error') and not evt.get('response'):
                                error_msg = evt['error']
                                logger.error(f"### GATEWAY ERROR → fallback: {error_msg}")
                                # Detect rate limit specifically so the UI can surface it
                                if 'rate limit' in error_msg.lower():
                                    # CLIENT-FACING RULE (Mike 2026-06-28): never serialize the
                                    # raw provider name or rate-limit string into a client-bound
                                    # event. Send a neutral code only; the browser surfaces
                                    # nothing and the spoken fallback below stays graceful.
                                    yield json.dumps({
                                        'type': 'rate_limit',
                                        'code': 'throttled',
                                    }) + '\n'
                                    metrics['rate_limited'] = 1
                                evt['response'] = "One moment, still working on that."
                                metrics['fallback_used'] = 1
                            full_response = evt.get('response')
                            if full_response:
                                full_response = normalize_action_tags(full_response)
                            if full_response and max_response_chars:
                                full_response = _truncate_at_sentence(full_response, max_response_chars)

                            # Suppress bare NO/YES sentinel responses to system triggers
                            # (gateway returns "NO" for wake-word checks on some triggers).
                            # __session_start__ is the exception: a bare NO/YES there is the
                            # same broken-greeting case as an empty reply, and silently
                            # `break`-ing here means the user connects the call and hears
                            # nothing. So let __session_start__ fall through to the
                            # empty-greeting fallback below instead of dead-ending.
                            _is_system_trigger = user_message.startswith('__')
                            if _is_system_trigger and user_message != '__session_start__' and full_response and \
                                    full_response.strip().upper() in ('NO', 'NO.', 'YES', 'YES.'):
                                logger.info(f'Suppressing sentinel "{full_response.strip()}" for system trigger')
                                yield json.dumps({'type': 'no_audio'}) + '\n'
                                log_metrics(metrics)
                                break

                            # Tag-only response fallback: if the agent responded
                            # with ONLY action tags and no spoken words, prepend
                            # a brief acknowledgment so TTS has something to say.
                            if full_response and re.match(
                                r'^\s*(\[[^\]]+\]\s*)+$', full_response
                            ):
                                logger.info(
                                    f"### Tag-only response detected, prepending "
                                    f"spoken text: {full_response.strip()[:60]}"
                                )
                                full_response = "Here you go. " + full_response
                                # Also update TTS buffer so the downstream flush
                                # at line ~1844 speaks "Here you go." instead of
                                # firing TTS on the bare tag (which strips to "").
                                _tts_buf = "Here you go."

                            metrics['llm_inference_ms'] = int((time.time() - t_llm_start) * 1000)
                            metrics['tool_count'] = sum(
                                1 for a in captured_actions
                                if a.get('type') == 'tool' and a.get('phase') == 'start'
                            )
                            metrics['profile'] = 'gateway'
                            metrics['model'] = 'glm-5-turbo'
                            # Estimate tokens: ~4 chars/token for English text
                            _resp_chars = len(full_response or '')
                            _ctx_chars = len(context_prefix) if context_prefix else 0
                            _est_input = _ctx_chars // 4
                            _est_output = _resp_chars // 4
                            metrics['est_input_tokens'] = _est_input
                            metrics['est_output_tokens'] = _est_output
                            logger.debug(f"[GW] Gateway response ({_resp_chars} chars): {repr((full_response or '')[:300])}")
                            logger.info(
                                f"### LLM inference completed in "
                                f"{metrics['llm_inference_ms']}ms "
                                f"(tools={metrics['tool_count']}) "
                                f"(tokens~{_est_input}in/{_est_output}out)"
                            )

                            # ── Recovery is STICKY ──────────────────────────
                            # We used to call _exit_session_recovery() on the
                            # first successful response so the next request
                            # went back to `main`. In practice `main` stayed
                            # poisoned (openclaw server-side session state
                            # persists on disk and doesn't self-heal when the
                            # WS reconnects) so every subsequent request
                            # bounced through the recovery cascade again —
                            # good for reliability, bad for 3-5s-per-message
                            # latency.
                            #
                            # Keep the timestamped recovery key as the new
                            # stable session for the process lifetime. If the
                            # recovery session itself poisons later, the
                            # double-empty handler will spin up a fresh
                            # recovery-<newepoch>. Only manual /api/conversation/reset
                            # exits recovery now.
                            if full_response and full_response.strip() and _session_recovery_key is not None:
                                bump_recovery_activity()
                                # A real (non-degenerate) response marks this
                                # recovery session healthy — it becomes the new
                                # stable session and is exempt from idle-clear.
                                if full_response.strip().upper().rstrip('.!?') not in ('NO', 'YES'):
                                    mark_recovery_success()

                            # ── Uncommitted tool-promise detection ────────────
                            # If the assistant said "let me build X" / "I'll write Y"
                            # but emitted ZERO tool_use blocks, the turn ended on
                            # an unfulfilled promise. This poisons the next turn —
                            # a follow-up user message layered on top of an open
                            # intent causes empty responses on GLM-4.7.
                            #
                            # Auto-continue: send a "continue — actually perform
                            # that work" steer and re-enter the event loop so the
                            # agent completes the promise on THIS turn.
                            #
                            # Gated: only fires when
                            #   - response was non-empty
                            #   - tool_count == 0  (nothing was actually done)
                            #   - llm_ms > 1000    (filter out instant degenerate empties — those go to the empty-retry branch below)
                            #   - not already continued this turn (avoids loops)
                            #
                            # NOTE: no upper bound on llm_ms. A genuine uncommitted
                            # promise after 50s is just as broken as after 5s — the
                            # agent still left work undone and the session still
                            # ends on an open intent. Previously gated at <30s and
                            # we missed a real case at 49s.
                            #
                            # Regex allows the committing verb anywhere within 80
                            # chars of the "I'll / let me" opener, so compound
                            # phrasing like "I'll mark the ones and add a button"
                            # still catches `add` (the mid-sentence verb).
                            # Verdict on the nudge sent last pass: did the agent act?
                            # Without this a nudge that does not land is
                            # indistinguishable from one that did.
                            if getattr(stream_response, '_continue_pending', False):
                                stream_response._continue_pending = False
                                if (metrics.get('tool_count', 0) == 0
                                        and not WORK_ACTION_TAG_RE.search(full_response or '')):
                                    logger.warning(
                                        f'### PROMISE STILL UNFULFILLED after auto-continue: '
                                        f'{(full_response or "").strip()[:100]!r} '
                                        f'(tool_count=0, ms={metrics.get("llm_inference_ms")})'
                                    )
                                else:
                                    logger.info(
                                        f'### AUTO-CONTINUE FULFILLED: '
                                        f'tools={metrics.get("tool_count", 0)}'
                                    )
                            if (
                                not getattr(stream_response, '_continued', False)
                                and gateway_reports_tools(gateway_id)
                                and is_uncommitted_promise(
                                    full_response,
                                    metrics.get('tool_count', 0),
                                    metrics.get('llm_inference_ms', 0),
                                )
                            ):
                                stream_response._continued = True
                                stream_response._continue_pending = True
                                logger.warning(
                                    f'### UNCOMMITTED PROMISE detected: '
                                    f'{full_response.strip()[:100]!r} '
                                    f'(tool_count=0, ms={metrics.get("llm_inference_ms")}) '
                                    f'— auto-continuing'
                                )
                                # Keep the client alive while we re-prompt
                                yield json.dumps({'type': 'retrying'}) + '\n'
                                # Preserve partial text as a TTS sentence so the
                                # user hears the assistant's intent while the
                                # follow-up tool turn runs
                                _continue_msg = (
                                    '[SYSTEM: You said you would do something ("'
                                    + full_response.strip()[:160]
                                    + '") but did not call any tool. Actually perform the '
                                    'work now, using the appropriate tools. Do not just '
                                    'describe it again.]'
                                )
                                retry_queue = queue.Queue()
                                captured_actions.clear()
                                # Reset accumulators so we don't double-count prior text
                                full_response = ''
                                def _continue_gateway():
                                    gateway_manager.stream_to_queue(
                                        retry_queue, _continue_msg,
                                        _session_key, captured_actions,
                                        gateway_id=gateway_id,
                                        agent_id=agent_id,
                                    )
                                continue_thread = threading.Thread(
                                    target=_continue_gateway, daemon=True,
                                )
                                t_llm_start = time.time()
                                continue_thread.start()
                                event_queue = retry_queue
                                logger.info('### AUTO-CONTINUE: sent promise-completion steer')
                                continue  # back to event loop — text_done NOT sent yet

                            # ── Empty after recent steer → auto-refire steer ──
                            # If an interject/steer landed on this session
                            # within the last 30s and the current LLM turn
                            # collapsed to zero chars, the steer was lost in
                            # the branched context. Re-fire the steered
                            # message as a fresh turn so the user's actual
                            # correction reaches the agent. Covers ALL empty
                            # cases (fast empty and timeout empty) — a lost
                            # steer is a bigger UX failure than a slow LLM.
                            _is_empty_pre = not full_response or not full_response.strip()
                            if _is_empty_pre and not getattr(stream_response, '_steer_refired', False):
                                _steer_msg = consume_recent_steer(_session_key, max_age_s=30.0)
                                if _steer_msg:
                                    stream_response._steer_refired = True
                                    logger.warning(
                                        f'### STEER-RECOVERY: empty response after recent steer '
                                        f'({len(_steer_msg)} chars) on session={_session_key} '
                                        f'— re-firing steered message as fresh turn'
                                    )
                                    yield json.dumps({'type': 'retrying'}) + '\n'
                                    # No sleep — the original `time.sleep(1)` was
                                    # paranoia leftover from early debugging. Every
                                    # second of artificial delay is a second of dead
                                    # silence for the user; the gateway's own internal
                                    # ordering is already sufficient.
                                    retry_queue = queue.Queue()
                                    captured_actions.clear()
                                    full_response = ''
                                    _refire_msg = (context_prefix or '') + _steer_msg
                                    def _refire_gateway():
                                        gateway_manager.stream_to_queue(
                                            retry_queue, _refire_msg,
                                            _session_key, captured_actions,
                                            gateway_id=gateway_id,
                                            agent_id=agent_id,
                                        )
                                    refire_thread = threading.Thread(
                                        target=_refire_gateway, daemon=True,
                                    )
                                    t_llm_start = time.time()
                                    refire_thread.start()
                                    event_queue = retry_queue
                                    logger.info('### STEER-RECOVERY: re-sent steered message to gateway')
                                    continue  # text_done NOT sent yet

                            # ── Retry once on instant empty response ──
                            # IMPORTANT: check BEFORE yielding text_done.
                            # If we yield empty text_done first, the client
                            # shows "Sorry" and cancels its reader — the retry
                            # result never reaches it.
                            # Instead: yield {'type':'retrying'} to keep the
                            # client alive, then swap the event queue.
                            #
                            # Bare "NO" / "YES" (with optional trailing punctuation)
                            # is treated as a DEGENERATE response — same class of
                            # broken-LLM output as empty. Confirmed in the wild:
                            # MiniMax / GLM occasionally emit a 2-char "NO" reply
                            # to normal user turns (sometimes after the empty-retry
                            # itself fires). Speaking that to a customer is
                            # unacceptable, so we route it through the same retry
                            # → double-empty → graceful-fallback path. A real
                            # voice answer should always elaborate beyond bare
                            # YES/NO; the voice-system-prompt instructs the agent
                            # accordingly.
                            _resp_stripped = (full_response or '').strip()
                            _is_degenerate = is_degenerate_reply(full_response)
                            _is_empty = (not full_response or not _resp_stripped) or _is_degenerate
                            if _is_empty and metrics.get('llm_inference_ms', 9999) < 5000 \
                                    and not getattr(stream_response, '_retried', False):
                                stream_response._retried = True
                                logger.warning(
                                    f"### {'DEGENERATE' if _is_degenerate else 'EMPTY'} RESPONSE "
                                    f"({_resp_stripped!r} in {metrics['llm_inference_ms']}ms) "
                                    f"— retrying once (client kept alive via 'retrying' event)"
                                )
                                # Wipe any TTS buffer that may already hold "NO"
                                # so the bare token is never spoken to the user.
                                if _is_degenerate:
                                    _tts_buf = ''
                                    _tts_pending.clear()
                                # (The 'retrying' event is yielded AFTER the queue swap
                                # below: if the client drops at that yield, the offline
                                # relay must wait on the RETRY queue, not the dead one — #565.)
                                # No sleep — the original `time.sleep(2)` was to let
                                # Z.AI "settle" between attempts but empirically every
                                # second here is pure dead silence for the user. The
                                # gateway's abort-before-send already ensures state is
                                # clean; additional delay just hurts UX.
                                # Re-send the message through the gateway.
                                # Always retry on the SAME session key first. The gateway
                                # may have been momentarily busy (queue flush, lane transition)
                                # and the same key will work 2 seconds later with full context.
                                # Switching to a recovery key here loses all conversation history
                                # (the agent doesn't know what was just discussed).
                                # Only the double-empty handler switches to a stable "recovery" key.
                                _retry_key = _session_key
                                if metrics.get('llm_inference_ms', 9999) < 500:
                                    logger.warning(
                                        f"### Fast empty ({metrics['llm_inference_ms']}ms) — "
                                        f"retrying on same session key '{_retry_key}'"
                                    )
                                retry_queue = queue.Queue()
                                captured_actions.clear()
                                # Rebuild the FULL outbound message for the retry:
                                # 1. Re-attach the recovery prime + reconnect prefix —
                                #    the old retry sent bare message_with_context, so
                                #    when the prime-carrying first turn on a recovery
                                #    session came back empty, the retry silently
                                #    dropped the conversation context (amnesia hole).
                                # 2. Append a nudge so the retry prompt is NOT
                                #    byte-identical to the failed attempt. The
                                #    gateway-internal EMPTY-FINAL retry already
                                #    resent the identical prompt twice; when the
                                #    empty is prompt-deterministic (GLM thinking-only
                                #    completion), identical retries fail identically.
                                _retry_msg = message_with_context
                                if _recovery_prefix:
                                    _retry_msg = _recovery_prefix + _retry_msg
                                if _recovery_prime:
                                    _retry_msg = _recovery_prime + _retry_msg
                                _retry_msg = _retry_msg + (
                                    '\n\n[SYSTEM: Your previous attempt at this message '
                                    'produced no spoken text. Respond now with a normal, '
                                    'plain spoken-text reply to the user\'s message above. '
                                    'Do not answer with a bare YES/NO, an empty message, '
                                    'or only action tags.]'
                                )
                                def _retry_gateway():
                                    gateway_manager.stream_to_queue(
                                        retry_queue, _retry_msg,
                                        _retry_key, captured_actions,
                                        gateway_id=gateway_id,
                                        agent_id=agent_id,
                                    )
                                retry_thread = threading.Thread(
                                    target=_retry_gateway, daemon=True
                                )
                                t_llm_start = time.time()
                                retry_thread.start()
                                event_queue = retry_queue
                                logger.info("### RETRY: re-sent message to gateway (with anti-empty nudge)")
                                # Tell the client to wait — don't show fallback
                                yield json.dumps({'type': 'retrying'}) + '\n'
                                continue  # back to event loop — text_done NOT sent yet

                            # ── Z.AI direct fallback after double-empty ──
                            if _is_empty and getattr(stream_response, '_retried', False):
                                logger.warning('### DOUBLE EMPTY — session poisoned, entering recovery mode')

                                # If this turn consumed a prime, re-arm it first.
                                # _enter_session_recovery() normally rebuilds a
                                # fresh prime (which supersedes this), but when
                                # its 10s cooldown skips the entry, this restore
                                # is the only thing keeping the context alive.
                                restore_recovery_prime(_recovery_prime)

                                # 1. Switch to recovery session key so NEXT request
                                #    goes to a fresh openclaw session (not the poisoned one)
                                _enter_session_recovery()

                                # 2. Force-disconnect gateway WS so it reconnects fresh
                                try:
                                    _gw = gateway_manager.get(gateway_id)
                                    if _gw and hasattr(_gw, 'force_disconnect'):
                                        _gw.force_disconnect()
                                        logger.warning('### Force-disconnected gateway WS after double-empty')
                                except Exception as _dfe:
                                    logger.error(f'### Failed to disconnect gateway: {_dfe}')

                                # 3. Write restart flag — CIRCUIT BREAKER gated
                                #    Max 2 restarts per 5 minutes. After that, skip the
                                #    restart flag (Z.AI fallback still fires below).
                                global _double_empty_restart_count, _double_empty_window_start
                                _now_de = time.time()
                                if _now_de - _double_empty_window_start > _DOUBLE_EMPTY_WINDOW_SECONDS:
                                    _double_empty_restart_count = 0
                                    _double_empty_window_start = _now_de
                                _double_empty_restart_count += 1

                                if _double_empty_restart_count <= _DOUBLE_EMPTY_MAX_RESTARTS:
                                    try:
                                        _flag_path = Path('/app/runtime/uploads/.restart-openclaw.flag')
                                        _flag_path.write_text(
                                            f'double-empty at {__import__("datetime").datetime.utcnow().isoformat()}Z'
                                        )
                                        logger.warning(
                                            f'### Wrote .restart-openclaw.flag — watchdog will clean up poisoned session '
                                            f'({_double_empty_restart_count}/{_DOUBLE_EMPTY_MAX_RESTARTS} in window)'
                                        )
                                    except Exception as _rfe:
                                        logger.error(f'### Failed to write restart flag: {_rfe}')
                                else:
                                    logger.warning(
                                        f'### CIRCUIT BREAKER: skipping restart flag — '
                                        f'{_double_empty_restart_count} double-empties in '
                                        f'{int(_now_de - _double_empty_window_start)}s window '
                                        f'(max {_DOUBLE_EMPTY_MAX_RESTARTS} per {_DOUBLE_EMPTY_WINDOW_SECONDS}s). '
                                        f'Z.AI fallback only.'
                                    )

                                # 4. Z.AI direct fallback for this message (NEVER Groq — Groq is TTS only)
                                try:
                                    from services.zai_direct import zai_messages_post
                                    _zai_key = os.environ.get('ZAI_API_KEY', '')
                                    _zai_fb_key = os.environ.get('ZAI_FALLBACK_API_KEY', '')
                                    # Use full context so the fallback LLM has agent personality
                                    _fallback_msg = _with_recent_history(message_with_context if message_with_context else user_message)
                                    _fallback_system = _fallback_system_prompt()
                                    if _zai_key or _zai_fb_key:
                                        _zai_resp = zai_messages_post(
                                            {
                                                'model': 'glm-5-turbo',
                                                'max_tokens': 1500,
                                                'system': _fallback_system,
                                                'messages': [{'role': 'user', 'content': _fallback_msg}],
                                            },
                                            timeout=30,
                                        )
                                        if _zai_resp.status_code == 200:
                                            _zai_data = _zai_resp.json()
                                            _zai_text = _zai_data.get('content', [{}])[0].get('text', '')
                                            if _zai_text:
                                                full_response = _zai_text
                                                metrics['fallback_used'] = 1
                                                metrics['profile'] = 'zai-direct'
                                                logger.info(f'### Z.AI direct fallback succeeded: {len(_zai_text)} chars')
                                except Exception as _fbe:
                                    logger.error(f'### Fallback LLM failed: {_fbe}')

                                if not full_response or not full_response.strip():
                                    full_response = "I missed that — my brain glitched for a second. Could you say that again?"

                            # ── Slow-empty: LLM ran 5s+ and returned empty ──
                            # The fast-empty retry path above only covers <5s empties.
                            # The double-empty branch above only covers post-_retried empties.
                            # That leaves a 5-30s gap: a single non-retried slow empty
                            # would fall straight to text_done(None) → "No response from agent
                            # after recovery" → user sees agent died (observed 2026-05-23
                            # on persona-tenant: 16882ms + 17370ms empties, both fell through).
                            #
                            # Try Z.AI direct (bypasses gateway and any poisoned openclaw
                            # session state) — same code path the double-empty branch uses.
                            # Only fall back to the spoken apology if Z.AI direct also fails.
                            # __session_start__ is handled by the dedicated greeting branch below.
                            if _is_empty and not getattr(stream_response, '_retried', False) \
                                    and metrics.get('llm_inference_ms', 0) >= 5000:
                                if user_message != '__session_start__':
                                    try:
                                        import requests as _req
                                        _zai_key = os.environ.get('ZAI_API_KEY', '')
                                        _fallback_msg = _with_recent_history(message_with_context if message_with_context else user_message)
                                        _fallback_system = _fallback_system_prompt()
                                        if _zai_key:
                                            _zai_resp = _req.post(
                                                'https://api.z.ai/api/anthropic/v1/messages',
                                                headers={
                                                    'x-api-key': _zai_key,
                                                    'anthropic-version': '2023-06-01',
                                                    'content-type': 'application/json',
                                                },
                                                json={
                                                    'model': 'glm-5-turbo',
                                                    'max_tokens': 1500,
                                                    'system': _fallback_system,
                                                    'messages': [{'role': 'user', 'content': _fallback_msg}],
                                                },
                                                timeout=20,
                                            )
                                            if _zai_resp.status_code == 200:
                                                _zai_data = _zai_resp.json()
                                                _zai_text = _zai_data.get('content', [{}])[0].get('text', '')
                                                if _zai_text:
                                                    full_response = _zai_text
                                                    metrics['fallback_used'] = 1
                                                    metrics['profile'] = 'zai-direct-slow-empty'
                                                    logger.info(
                                                        f"### SLOW-EMPTY Z.AI direct fallback succeeded "
                                                        f"({metrics['llm_inference_ms']}ms gateway empty → "
                                                        f"{len(_zai_text)} chars direct)"
                                                    )
                                    except Exception as _fbe:
                                        logger.error(f'### Slow-empty Z.AI fallback failed: {_fbe}')

                                # Z.AI direct didn't return text either — graceful apology.
                                # For __session_start__ do NOT hardcode a greeting here: the
                                # __session_start__ block below takes one GENERATED retry
                                # (or intentional silence, feedback_no_hardcoded_responses).
                                # Setting one here left full_response non-empty, so that block's
                                # `if not _gs...` skipped — the canned "give me a moment" filler
                                # then won over the tenant's persona greeting (persona-tenant, 2026-07-13).
                                if not full_response or not full_response.strip():
                                    if user_message != '__session_start__':
                                        full_response = (
                                            "That took a bit longer than expected on my end. "
                                            "I'm still here — try again and I'll get right to it."
                                        )
                                        metrics['fallback_used'] = 1
                                    logger.warning(
                                        f"### SLOW EMPTY ({metrics['llm_inference_ms']}ms) — "
                                        f"Z.AI direct also failed"
                                        + ("" if user_message == '__session_start__'
                                           else ", using apology")
                                    )

                            # ── __session_start__ must ALWAYS produce a spoken greeting ──
                            # GLM-5-turbo (current temporary primary, see
                            # memory/glm-primary-temporary-swap) returns empty / bare-"NO" /
                            # tag-only completions on the first turn of a session noticeably
                            # more than MiniMax did. The gateway already retried once
                            # internally (openclaw.py EMPTY-FINAL retry); the double-empty
                            # breaker and the conversation.py retry both deliberately skip
                            # __ system triggers; and the timeout-empty branch above only
                            # covers >30s runs. So a 5–30s empty greeting falls through to
                            # here with full_response = None — and an empty text_done makes
                            # app.js "silent resume" (user connects the call, hears dead air,
                            # has to speak first). Take ONE generated retry instead (Z.AI direct,
                            # same brief + context); never a canned line (2026-10-27,
                            # greetings-are-real-agent-turns). The agent's openclaw session is
                            # still warmed by the empty turn, so the conversation continues
                            # normally from the user's next message.
                            if user_message == '__session_start__':
                                _gs = (full_response or '').strip()
                                _gs_norm = _gs.upper().rstrip('.!?')
                                _gs_tag_only = bool(_gs) and re.match(r'^\s*(\[[^\]]+\]\s*)+$', _gs)
                                if (not _gs) or _gs_norm in ('NO', 'YES') or _gs_tag_only:
                                    # Empty/degenerate wake-up turn — substitute a greeting.
                                    # ENTERTAINMENT-PERSONA path (opt-in per profile, e.g. persona-tenant):
                                    # the gateway emptied, so take ONE more shot using the SAME rotated
                                    # persona prompt we built for the gateway turn above
                                    # (_persona_greeting_prompt — faithful voice + fresh angle + avoid-recent).
                                    # That keeps even this standalone shot on-character and varied.
                                    # Everyone else gets one generated retry with the contextual brief.
                                    # fallback_greetings and the profile greeting are NOT spoken any
                                    # more (2026-10-27): a code-chosen line is a canned clip.
                                    # (persona-tenant 2026-07-13; feedback_no_hardcoded_responses.)
                                    # 2026-10-27 (greetings-are-real-agent-turns): every fallback
                                    # here is a GENERATED greeting. No profile greeting string and
                                    # no fallback_greetings line is ever spoken by code.
                                    _fb_greeting = ''
                                    try:
                                        _zk = os.environ.get('ZAI_API_KEY', '')
                                        if _zk:
                                            import requests as _req_g
                                            if _persona_greeting_prompt:
                                                # Persona: SAME rotated prompt as the gateway turn
                                                # (faithful voice + fresh angle + avoid-recent + who).
                                                _g_system = ('Output ONLY the spoken greeting text — '
                                                             'no tags, no preamble, fully in character.')
                                                _g_msg = _gateway_message
                                                _g_label = 'persona-greeting-retry'
                                            else:
                                                # Everyone else: the contextual brief WITH the full
                                                # context prefix (CURRENT_USER, office briefing, recent
                                                # turns), so the retry is still personal and current.
                                                _g_system = (_fallback_system_prompt()
                                                             + ' Output ONLY the spoken greeting text — '
                                                               'one or two sentences, no tags, no preamble.')
                                                _g_msg = _with_recent_history(
                                                    message_with_context if message_with_context
                                                    else _gateway_message)
                                                _g_label = 'session-start-greeting-retry'
                                            try:
                                                _rr = _req_g.post(
                                                    'https://api.z.ai/api/anthropic/v1/messages',
                                                    headers={'x-api-key': _zk,
                                                             'anthropic-version': '2023-06-01',
                                                             'content-type': 'application/json'},
                                                    json={'model': 'glm-5-turbo', 'max_tokens': 400,
                                                          'system': _g_system,
                                                          'messages': [{'role': 'user', 'content': _g_msg}]},
                                                    timeout=15)
                                                if _rr.status_code == 200:
                                                    _fb_greeting = (_rr.json().get('content', [{}])[0]
                                                                    .get('text', '') or '').strip()
                                                    if _fb_greeting:
                                                        metrics['profile'] = _g_label
                                            except Exception as _ge:
                                                logger.warning(f'### SESSION_START greeting retry failed: {_ge}')
                                    except Exception:
                                        _fb_greeting = ''
                                    if not _fb_greeting and (_persona_fallbacks or _profile_greeting):
                                        logger.warning(
                                            '### SESSION_START: canned profile greeting / fallback_greetings '
                                            'NOT spoken (greetings-are-real-agent-turns)')
                                    logger.warning(
                                        f"### SESSION_START produced no usable greeting "
                                        f"(was {full_response!r}, {metrics.get('llm_inference_ms')}ms) "
                                        f"— generated_retry={_fb_greeting!r} (empty = silence by design)"
                                    )
                                    full_response = _fb_greeting  # may be '' — that's the right diagnostic signal
                                    metrics['fallback_used'] = 1 if _fb_greeting else 0
                                    metrics['llm_empty_session_start'] = 1
                                    # Drop any partial / bare-token TTS buffered from the
                                    # broken turn so only the generated greeting is spoken, if any.
                                    _tts_buf = ''
                                    _tts_pending.clear()

                            # ── Final safety net: bare "NO" / "YES" must NEVER reach the user ──
                            # Catches any degenerate single-token response that slipped past
                            # the retry path above (slow-degenerate 5s–30s, or a retry that
                            # itself returned bare NO/YES). A real voice answer always
                            # elaborates; bare YES/NO is broken-LLM output.
                            if is_degenerate_reply(full_response) and not user_message.startswith('__'):
                                logger.warning(
                                    f"### DEGENERATE FINAL ({metrics.get('llm_inference_ms')}ms) "
                                    f"response={full_response!r} — replacing with graceful fallback"
                                )
                                full_response = (
                                    "Sorry, my brain glitched for a second. "
                                    "Could you say that again?"
                                )
                                metrics['fallback_used'] = 1
                                # Wipe TTS buffer so the bare token isn't spoken
                                _tts_buf = ''
                                _tts_pending.clear()

                            # Persona greeting rotation (persona-tenant): remember what he ACTUALLY
                            # opened with, so the next wake-up's prompt can avoid echoing it.
                            # Best-effort — the rotation index already advanced at prompt-build
                            # time, so a miss here never repeats an angle. Persona path only.
                            if _grot_active and user_message == '__session_start__' and (full_response or '').strip():
                                try:
                                    import json as _json_cap
                                    _stc = {}
                                    if _grot_state_path and os.path.exists(_grot_state_path):
                                        _stc = _json_cap.load(open(_grot_state_path)) or {}
                                    _rec = [r for r in (_stc.get('recent') or [])
                                            if isinstance(r, str) and r.strip()]
                                    _g_cap = full_response.strip()
                                    if not _rec or _rec[-1] != _g_cap:
                                        _rec.append(_g_cap)
                                    _stc['recent'] = _rec[-6:]
                                    _tmp_cap = _grot_state_path + '.tmp'
                                    with open(_tmp_cap, 'w') as _fh_cap:
                                        _json_cap.dump(_stc, _fh_cap)
                                    os.replace(_tmp_cap, _grot_state_path)
                                except Exception:
                                    pass

                            yield json.dumps({
                                'type': 'text_done',
                                'response': full_response,
                                'actions': captured_actions,
                                'timing': {
                                    'handshake_ms': metrics.get('handshake_ms'),
                                    'llm_ms': metrics.get('llm_inference_ms'),
                                }
                            }) + '\n'

                            # Auto-reset removed — loop detection (Phase 1 config)
                            # handles stuck agents; consecutive empties no longer
                            # trigger a session key bump that would cold-cache Z.AI.

                            # Handle [SESSION_RESET] trigger from agent
                            if full_response and '[SESSION_RESET]' in full_response:
                                old_key = get_voice_session_key()
                                new_key = bump_voice_session()
                                logger.info(
                                    f'### AGENT-TRIGGERED SESSION RESET: {old_key} → {new_key}'
                                )
                                full_response = full_response.replace('[SESSION_RESET]', '').strip()

                            # Detect agent returning a bare file path (e.g. from TTS tool use)
                            if full_response and re.match(r'^/tmp/[\w/.-]+$', full_response.strip()):
                                file_path = full_response.strip()
                                logger.warning(f'Agent returned file path — serving directly: {file_path}')
                                try:
                                    with open(file_path, 'rb') as f:
                                        file_bytes = f.read()
                                    audio_b64 = base64.b64encode(file_bytes).decode('utf-8')
                                    ext = file_path.rsplit('.', 1)[-1].lower()
                                    audio_format = ext if ext in ('mp3', 'wav', 'ogg') else 'mp3'
                                    metrics['tts_generation_ms'] = 0
                                    metrics['total_ms'] = int((time.time() - t_request_start) * 1000)
                                    yield json.dumps({
                                        'type': 'audio',
                                        'audio': audio_b64,
                                        'audio_format': audio_format,
                                        'chunk': 0,
                                        'timing': {'tts_ms': 0, 'total_ms': metrics.get('total_ms')},
                                    }) + '\n'
                                    logger.info(f'Served agent-generated audio: {len(file_bytes)} bytes ({audio_format})')
                                except Exception as fp_err:
                                    # CLIENT-FACING RULE (Mike 2026-06-28): log the raw
                                    # exception server-side; send only a neutral reason code.
                                    logger.error(f'Failed to serve agent audio file {file_path}: {fp_err}')
                                    yield json.dumps({
                                        'type': 'tts_error',
                                        'reason': 'file_read_error',
                                    }) + '\n'
                                log_metrics(metrics)
                                break

                            # ── Flush TTS buffer + yield audio chunks in order ──
                            metrics['response_len'] = len(full_response) if full_response else 0

                            # If response was suppressed (None), discard ALL
                            # pending TTS — never speak suppressed text like
                            # HEARTBEAT_OK that leaked through delta streaming.
                            if not full_response:
                                if _tts_pending:
                                    logger.info(
                                        f"### Discarding {len(_tts_pending)} TTS "
                                        f"chunks for suppressed response"
                                    )
                                _tts_buf = ''
                                _tts_pending = []

                            # Fire TTS for any remaining buffered text
                            _remaining = _tts_buf.strip()
                            if _remaining:
                                _tts_pending.append(_fire_tts(_remaining))
                                _tts_buf = ''

                            # Fallback: no sentences extracted (very short response).
                            # Guard on _chunks_sent, NOT just _tts_pending: the delta
                            # path pops sentences from _tts_pending as their audio is
                            # yielded, so on short fast replies the list is already
                            # empty at text_done and this fallback re-spoke the WHOLE
                            # reply a second time (double audio, 2026-07-12).
                            # _chunks_sent counts audio actually delivered — if any
                            # went out, streaming TTS already covered the text.
                            if (not _tts_pending and full_response
                                    and (_chunks_sent == 0 or not _answer_tts_fired)):
                                tts_text = clean_for_tts(full_response)
                                if tts_text and tts_text.strip():
                                    _tts_pending.append(_fire_tts(tts_text))

                            if not _tts_pending:
                                logger.info('Skipping TTS — no speakable text')
                                # Tell the frontend there's no audio coming so it can
                                # reset isProcessing and re-enable the mic.
                                yield json.dumps({'type': 'no_audio'}) + '\n'
                                metrics['total_ms'] = int((time.time() - t_request_start) * 1000)
                                log_metrics(metrics)
                                if full_response:
                                    log_conversation('assistant', full_response,
                                                     session_id=session_id,
                                                     tts_provider=tts_provider, voice=voice)
                                    save_conversation_turn(
                                        user_msg=user_message,
                                        ai_response=full_response,
                                        session_id=session_id,
                                        session_key=_session_key,
                                        tts_provider=tts_provider,
                                        voice=voice,
                                        duration_ms=metrics.get('total_ms'),
                                        actions=captured_actions,
                                        identified_person=identified_person,
                                        clerk_user_id=_clerk_user_id,
                                        gateway=gateway_id or 'openclaw',
                                    )
                                break

                            t_tts_start = time.time()
                            total_chunks = _chunks_sent + len(_tts_pending)
                            # TTS-2 / TTS-8: a hung provider must NOT head-of-line-block
                            # the ordered flush, and one chunk's error/timeout must NOT
                            # discard the already-generated (paid) chunks behind it.
                            # First pass: wait a bounded per-chunk timeout; deliver ready
                            # audio, skip errors, and defer any chunk whose provider is
                            # still hung. Second pass: give the deferred chunks a short
                            # grace and deliver any that landed (out of order beats
                            # silence). Never silently drop paid audio — log the rest.
                            _tts_sentence_timeout = int(os.getenv('TTS_SENTENCE_TIMEOUT', '20'))
                            _tts_deferred_grace = int(os.getenv('TTS_DEFERRED_GRACE', '10'))
                            _deferred = []
                            for i, (done_evt, res) in enumerate(_tts_pending):
                                if done_evt.wait(timeout=_tts_sentence_timeout):
                                    if res['error']:
                                        metrics['tts_success'] = 0
                                        metrics['tts_error'] = res['error']
                                        yield _tts_error_event(res['error'])
                                        continue  # skip only the failed chunk, keep flushing
                                    if res['audio']:
                                        yield _audio_event(res['audio'], _chunks_sent + i, tts_ms=int((time.time() - t_tts_start) * 1000))
                                else:
                                    logger.warning(
                                        f"### TTS chunk {i} not ready after "
                                        f"{_tts_sentence_timeout}s — skipping ahead (provider hung)"
                                    )
                                    _deferred.append((i, done_evt, res))
                            # Second pass — deliver late-completing chunks if they land.
                            for i, done_evt, res in _deferred:
                                if done_evt.wait(timeout=_tts_deferred_grace):
                                    if res['error']:
                                        yield _tts_error_event(res['error'])
                                    elif res['audio']:
                                        logger.info(f"### TTS chunk {i} completed late — delivering")
                                        yield _audio_event(res['audio'], _chunks_sent + i, tts_ms=int((time.time() - t_tts_start) * 1000))
                                else:
                                    metrics['tts_success'] = 0
                                    logger.error(
                                        f"### TTS chunk {i} never completed within grace — "
                                        f"audio not delivered (provider hung); paid generation logged, not discarded"
                                    )

                            # Fully handled above (waited + yielded or logged) — clear so
                            # the post-loop exit flush (#487) never re-processes these.
                            _tts_pending = []

                            metrics['tts_generation_ms'] = int((time.time() - t_tts_start) * 1000)
                            metrics['tts_text_len'] = metrics['response_len']
                            metrics['total_ms'] = int((time.time() - t_request_start) * 1000)
                            log_metrics(metrics)
                            if full_response:
                                log_conversation('assistant', full_response,
                                                 session_id=session_id,
                                                 tts_provider=tts_provider, voice=voice)
                                save_conversation_turn(
                                    user_msg=user_message,
                                    ai_response=full_response,
                                    session_id=session_id,
                                    session_key=_session_key,
                                    tts_provider=tts_provider,
                                    voice=voice,
                                    duration_ms=metrics.get('total_ms'),
                                    actions=captured_actions,
                                    identified_person=identified_person,
                                    clerk_user_id=_clerk_user_id,
                                    gateway=gateway_id or 'openclaw',
                                )
                            break

                        if evt['type'] == 'error':
                            # Gateway-level failure — if this request consumed a
                            # recovery prime, the message likely never reached
                            # the session. Re-arm the prime so the NEXT turn
                            # still carries the conversation context.
                            restore_recovery_prime(_recovery_prime)
                            # CLIENT-FACING RULE (Mike 2026-06-28): never serialize the raw
                            # gateway/exception text into a client-bound event. Log it
                            # server-side; send only a neutral code the browser maps to a
                            # graceful "reconnecting" line.
                            logger.error(f"### STREAM ERROR (not sent to client): {evt.get('error', 'Unknown error')!r}")
                            yield json.dumps({
                                'type': 'error',
                                'code': 'gateway',
                            }) + '\n'
                            break

                    # Exit-path TTS flush (#487) — if we're leaving via a path that
                    # skipped the text_done flush (STREAM HARD TIMEOUT / gateway
                    # 'error' break), _tts_pending may still hold finished-or-finishing
                    # TTS audio the client never got. Wait a bounded cap and deliver it
                    # instead of silently dropping the tail of the spoken answer.
                    _tts_exit_cap = int(os.getenv('TTS_EXIT_FLUSH_CAP', '15'))
                    _exit_events, _exit_dropped = drain_pending_tts_on_exit(
                        _tts_pending, _chunks_sent, _audio_event, _tts_error_event,
                        cap_seconds=_tts_exit_cap, log=logger,
                    )
                    for _exit_evt in _exit_events:
                        yield _exit_evt
                    if _exit_events:
                        logger.info(
                            f"### STREAM EXIT: flushed {len(_exit_events)} TTS chunk(s) "
                            f"that would otherwise have been dropped"
                        )

                    # Drain any remaining raw queue events (debug: detect generator exit
                    # without text_done). tts_ready is a payload-free wake ping — its
                    # audio (if any) was already handled above via _tts_pending, so a
                    # leftover tts_ready here is expected noise, not a loss, and is
                    # logged at info rather than warning.
                    _remaining_evts = []
                    while not event_queue.empty():
                        try:
                            _remaining_evts.append(event_queue.get_nowait())
                        except Exception:
                            break
                    if _remaining_evts:
                        _types = [e.get('type', '?') for e in _remaining_evts]
                        _noteworthy = [t for t in _types if t != 'tts_ready']
                        if _noteworthy or _exit_dropped:
                            logger.warning(f"### STREAM EXIT with {len(_remaining_evts)} unprocessed events: {_types}")
                        else:
                            logger.info(f"### STREAM EXIT drained {len(_remaining_evts)} stale tts_ready wake-ping(s) (already flushed)")

                def stream_with_offline_relay():
                    """stream-offline-relay: pass every chunk through
                    unchanged. Only when the client is gone (the WSGI server
                    closes the iterator after a failed write -> GeneratorExit
                    here) before a text_done was written is the turn's final
                    reply handed to the offline delivery path."""
                    inner = stream_response()
                    text_done_sent = False
                    try:
                        for chunk in inner:
                            yield chunk
                            # Resumed = the server wrote this chunk.
                            if not text_done_sent and isinstance(chunk, str) \
                                    and '"type": "text_done"' in chunk[:40]:
                                text_done_sent = True
                    except GeneratorExit:
                        inner.close()
                        if not text_done_sent and not user_message.startswith('__'):
                            _q = event_queue
                            _done = _stream_state['done_evt'] if _stream_state['done_q'] is _q else None
                            logger.warning(
                                '### stream-offline-relay: client disconnected before the '
                                'final reply was written — waiting for it in the background')
                            _start_stream_offline_relay(
                                event_queue=_q, done_evt=_done,
                                gateway_id=gateway_id, agent_id=agent_id,
                                session_key=_session_key,
                                run_id=f'stream:{_session_key}:{uuid.uuid4().hex[:12]}',
                                t_request_start=t_request_start,
                                t_disconnect=time.time(),
                                # the leg running on _q (auto-continue/retry reset it)
                                t_llm_start=t_llm_start,
                                done_llm_ms=_stream_state.get('done_llm_ms') if _done is not None else None,
                            )
                        raise

                return Response(
                    stream_with_offline_relay(),
                    mimetype='application/x-ndjson',
                    headers={'X-Accel-Buffering': 'no', 'Cache-Control': 'no-cache'}
                )

            else:
                # ── NON-STREAMING: wait for full Gateway response ─────────
                gw_thread.join(timeout=310)
                while not event_queue.empty():
                    evt = event_queue.get_nowait()
                    if evt['type'] == 'text_done':
                        ai_response = evt.get('response')
                        if ai_response:
                            ai_response = normalize_action_tags(ai_response)
                        if evt.get('deferred') and not evt.get('error') \
                                and not (ai_response or '').strip():
                            _deferred_turn = True
                    elif evt['type'] == 'handshake':
                        metrics['handshake_ms'] = evt['ms']
                if not ai_response and not _deferred_turn:
                    # Turn failed — re-arm any consumed recovery prime so the
                    # next request still carries the conversation context.
                    restore_recovery_prime(_recovery_prime)
                metrics['llm_inference_ms'] = int((time.time() - t_llm_start) * 1000)
                metrics['tool_count'] = sum(
                    1 for a in captured_actions
                    if a.get('type') == 'tool' and a.get('phase') == 'start'
                )
                metrics['profile'] = 'gateway'
                metrics['model'] = 'glm-5-turbo'
                _resp_chars2 = len(ai_response or '')
                _ctx_chars2 = len(context_prefix) if context_prefix else 0
                metrics['est_input_tokens'] = _ctx_chars2 // 4
                metrics['est_output_tokens'] = _resp_chars2 // 4
                logger.info(
                    f"### LLM inference completed in {metrics['llm_inference_ms']}ms "
                    f"(tools={metrics['tool_count']}) "
                    f"(tokens~{metrics['est_input_tokens']}in/{metrics['est_output_tokens']}out)"
                )

        except Exception as e:
            logger.error(f'Failed to call Clawdbot Gateway: {e}')

    # ── DEFERRED: gateway ended the turn on purpose, answer comes via push ──
    # NO FAKE REPLIES: no flash fallback, no generic line, no TTS, nothing saved.
    if _deferred_turn:
        metrics['profile'] = 'gateway-deferred'
        metrics['response_len'] = 0
        metrics['total_ms'] = int((time.time() - t_request_start) * 1000)
        logger.info('### DEFERRED TURN (non-streaming) — no reply, no fallback, no TTS')
        log_metrics(metrics)
        _deferred_data = {'response': '', 'deferred': True, 'user_said': user_message}
        if captured_actions:
            _deferred_data['actions'] = captured_actions
        _deferred_data['timing'] = {
            'handshake_ms': metrics.get('handshake_ms'),
            'llm_ms': metrics.get('llm_inference_ms'),
            'tts_ms': None,
            'total_ms': metrics.get('total_ms'),
        }
        return jsonify(_deferred_data)

    # ── FALLBACK: Z.AI direct (glm-5-flash, no tools) ───────────────────
    if not ai_response:
        if metrics.get('profile') == 'gateway':
            logger.warning('No text response from Gateway, falling back to Z.AI flash...')
            metrics['fallback_used'] = 1
        else:
            logger.info('Using Z.AI flash direct (primary path)')
        t_flash_start = time.time()
        # Lazy import to avoid circular dependency (server.py imports this blueprint)
        try:
            import server as _server
            ai_response = _server.get_zai_direct_response(
                _with_recent_history(message_with_context), session_id)
        except Exception as e:
            logger.error(f'Z.AI direct call failed: {e}')
            ai_response = None
        metrics['profile'] = 'flash-direct'
        metrics['model'] = 'glm-5-flash'
        metrics['llm_inference_ms'] = int((time.time() - t_flash_start) * 1000)

    # ── LAST RESORT ───────────────────────────────────────────────────────
    if not ai_response:
        logger.warning('Both Gateway and Z.AI flash failed, using generic fallback')
        ai_response = "One moment, I'm still working on something."

    # Clean text for TTS
    tts_text = clean_for_tts(ai_response)
    logger.info(f'Cleaned TTS text ({len(tts_text)} chars): {tts_text[:100]}...')
    metrics['response_len'] = len(ai_response) if ai_response else 0
    metrics['tts_text_len'] = len(tts_text)

    # Generate TTS audio
    t_tts_start = time.time()
    audio_base64 = None
    if tts_text and tts_text.strip():
        audio_base64 = _tts_generate_b64(tts_text, voice=voice or 'M1',
                                          tts_provider=tts_provider)
        if audio_base64 is None:
            metrics['tts_success'] = 0
            metrics['tts_error'] = 'TTS generation failed'
    t_tts_end = time.time()
    metrics['tts_generation_ms'] = int((t_tts_end - t_tts_start) * 1000)
    metrics['total_ms'] = int((t_tts_end - t_request_start) * 1000)

    log_metrics(metrics)
    if ai_response:
        log_conversation('assistant', ai_response, session_id=session_id,
                         tts_provider=tts_provider, voice=voice)
        save_conversation_turn(
            user_msg=user_message,
            ai_response=ai_response,
            session_id=session_id,
            session_key=get_voice_session_key(),
            tts_provider=tts_provider,
            voice=voice,
            duration_ms=metrics.get('total_ms'),
            actions=captured_actions,
            identified_person=identified_person,
            clerk_user_id=_clerk_user_id,
            gateway=gateway_id or 'openclaw',
        )

    response_data = {'response': ai_response, 'user_said': user_message}
    if audio_base64:
        response_data['audio'] = audio_base64
    if captured_actions:
        response_data['actions'] = captured_actions
    response_data['timing'] = {
        'handshake_ms': metrics.get('handshake_ms'),
        'llm_ms': metrics.get('llm_inference_ms'),
        'tts_ms': metrics.get('tts_generation_ms'),
        'total_ms': metrics.get('total_ms'),
    }

    return jsonify(response_data)

# ---------------------------------------------------------------------------
# POST /api/conversation/abort
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/conversation/abort', methods=['POST'])
def conversation_abort():
    """Abort the active agent run for the current voice session.

    Fire-and-forget from client — used by PTT interrupt and sendMessage
    interrupt to tell openclaw to stop generating so it doesn't waste compute.
    """
    session_key = get_voice_session_key()
    # Log abort source from client for debugging
    source = 'unknown'
    source_text = ''
    try:
        body = request.get_json(silent=True) or {}
        source = body.get('source', 'unknown')
        source_text = body.get('text', '')
    except Exception:
        pass
    # stream-offline-relay: a turn the user deliberately stopped is never
    # relayed to them afterwards.
    global _last_user_abort_ts
    _last_user_abort_ts = time.time()
    gw = gateway_manager.get('openclaw')
    aborted = False
    if gw and hasattr(gw, 'abort_active_run'):
        aborted = gw.abort_active_run(session_key)
    logger.info(f"### ABORT request session={session_key} aborted={aborted} source={source} text={source_text!r}")
    if source == 'stopVoiceInput':
        logger.info(f"### CALL_END session={session_key}")
    return jsonify({'ok': True, 'aborted': aborted})


# ---------------------------------------------------------------------------
# POST /api/conversation/steer
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/conversation/steer', methods=['POST'])
def conversation_steer():
    """Inject a user message into the active agent run (steer mode).

    Fire-and-forget from client — used when the user speaks while the
    agent is silently working (tools / sub-agents / heartbeat).  Instead
    of aborting the active run and starting fresh, this sends a second
    chat.send to the same session.  OpenClaw's messages.queue.mode=steer
    injects the message at the next tool boundary so the agent sees the
    user's correction and pivots immediately.

    The active /api/conversation streaming response continues receiving
    the steered output — no new streaming connection is needed.

    Request body:
        message  (str)  — the user's text to inject
        source   (str)  — label for logging (e.g. 'clawdbot-sendMessage')

    Returns:
        { ok: true, steered: true/false }
    """
    body = request.get_json(silent=True) or {}
    message = (body.get('message') or '').strip()
    source = body.get('source', 'unknown')

    if not message:
        return jsonify({'ok': False, 'error': 'No message provided'}), 400

    # Input length guard (same as main conversation endpoint)
    if len(message) > 500000:  # raised 2026-06-25 — never reject a user's pasted text (steer/interject)
        return jsonify({'ok': False, 'error': 'Message too long'}), 400

    session_key = get_voice_session_key()

    steered = gateway_manager.send_steer(message, session_key)

    # Record for the empty-response recovery path — if the current LLM turn
    # collapses to zero chars (steer-mid-inference failure mode), the
    # streaming handler will re-fire this message as a fresh turn.
    if steered:
        record_recent_steer(session_key, message)

    logger.info(
        f"### STEER request session={session_key} steered={steered} "
        f"source={source} text={message!r}"
    )

    # Log the steer message as a user turn so the transcript is preserved
    log_conversation('user', message, session_id='default')

    return jsonify({'ok': True, 'steered': steered})


# ---------------------------------------------------------------------------
# POST /api/conversation/interject — smart message routing
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/conversation/interject', methods=['POST'])
def conversation_interject():
    """Smart message routing during active agent runs.

    Classifies the user's message into one of three lanes:
      context   — queue alongside the current run (collect mode)
      steer     — inject at next tool boundary, skip remaining tools
      fast_lane — independent action (for now: treated as steer; Phase 3
                  will route to a parallel sub-agent)

    The frontend calls this instead of /steer when the agent is busy,
    letting the backend decide the appropriate routing.

    Request body:
        message  (str) — the user's text
        source   (str) — caller label for logging

    Returns:
        { ok: true, lane: "context"|"steer"|"fast_lane", action: "queued"|"steered" }
    """
    from routes.message_classifier import classify_message

    body = request.get_json(silent=True) or {}
    message = (body.get('message') or '').strip()
    source = body.get('source', 'unknown')

    if not message:
        return jsonify({'ok': False, 'error': 'No message provided'}), 400
    if len(message) > 500000:  # raised 2026-06-25 — never reject a user's pasted text (steer/interject)
        return jsonify({'ok': False, 'error': 'Message too long'}), 400

    lane = classify_message(message, agent_busy=True)
    session_key = get_voice_session_key()

    if lane == 'context':
        # Queue alongside the current run — gateway's collect mode handles this.
        # We still send via steer because the message needs to reach the session,
        # but OpenClaw's collect mode will hold it until the current turn completes.
        steered = gateway_manager.send_steer(message, session_key)
        if steered:
            record_recent_steer(session_key, message)
        action = 'queued'
        logger.info(
            f"### INTERJECT [context] session={session_key} action={action} "
            f"source={source} text={message!r}"
        )
    elif lane == 'steer':
        # Inject at next tool boundary — skip remaining tools
        steered = gateway_manager.send_steer(message, session_key)
        if steered:
            record_recent_steer(session_key, message)
        action = 'steered'
        logger.info(
            f"### INTERJECT [steer] session={session_key} action={action} "
            f"source={source} text={message!r}"
        )
    else:  # fast_lane
        # Route to the fast lane agent via a separate session key.
        # This runs in parallel with the main agent — no interference.
        fast_session_key = 'fast-lane'
        try:
            from queue import Queue as _Q
            _fq = _Q()
            import threading
            def _fast_run():
                try:
                    gw = gateway_manager.get('openclaw')
                    if gw:
                        gw.stream_to_queue(
                            _fq, message, fast_session_key,
                            captured_actions=None,
                            agent_id='openvoiceui-fast',
                        )
                except Exception as e:
                    logger.error(f'### FAST LANE error: {e}')
                finally:
                    _fq.put({'type': 'text_done', 'response': None})

            _ft = threading.Thread(target=_fast_run, daemon=True)
            _ft.start()

            # Collect the fast lane response (with timeout)
            _fast_text = ''
            _fast_start = time.time()
            while time.time() - _fast_start < 15:
                try:
                    ev = _fq.get(timeout=1)
                    if ev.get('type') == 'delta':
                        _fast_text += ev.get('text', '')
                    elif ev.get('type') == 'text_done':
                        if ev.get('response'):
                            _fast_text = ev['response']
                        break
                except Exception:
                    continue

            action = 'fast_lane'
            logger.info(
                f"### INTERJECT [fast_lane] session={fast_session_key} "
                f"source={source} text={message!r} → response={_fast_text[:100]!r}"
            )

            # Return the fast lane response directly — frontend can TTS it
            log_conversation('user', message, session_id='default')
            if _fast_text.strip():
                log_conversation('assistant', _fast_text.strip(), session_id='default')
            return jsonify({
                'ok': True,
                'lane': 'fast_lane',
                'action': 'fast_lane',
                'response': _fast_text.strip() if _fast_text.strip() else None,
                'steered': False,
            })

        except Exception as _fle:
            logger.error(f'### FAST LANE failed, falling back to steer: {_fle}')
            # Fallback: steer mode
            steered = gateway_manager.send_steer(message, session_key)
            action = 'steered'
            logger.info(
                f"### INTERJECT [fast_lane→steer fallback] session={session_key} "
                f"source={source} text={message!r}"
            )

    # Log the interjected message as a user turn for transcript
    log_conversation('user', message, session_id='default')

    return jsonify({
        'ok': True,
        'lane': lane,
        'action': action,
        'steered': steered if lane != 'context' else None,
    })


# ---------------------------------------------------------------------------
# POST /api/conversation/reset
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/conversation/reset', methods=['POST'])
def conversation_reset():
    """Clear in-process conversation history for a session."""
    body = request.get_json() or {}
    session_id = body.get('session_id', 'default')
    conversation_histories.pop(session_id, None)
    return jsonify({'status': 'ok', 'message': 'Conversation history cleared'})


# ---------------------------------------------------------------------------
# POST /api/session/reset-openclaw  — deep reset that also clears the openclaw
# session JSONL (ADMIN-BUG-2, 2026-07-11)
# ---------------------------------------------------------------------------
# HISTORY / ADMIN-BUG-2: this handler was originally registered on
# '/api/session/reset', the SAME path as server.py's soft/hard reset handler
# (server.py `session_reset`). Because conversation_bp is registered before the
# server.py @app.route is defined, Werkzeug matched THIS blueprint rule first —
# silently shadowing server.py's mode-aware (soft/hard + pre-warm) handler that
# SessionControl.js / the admin Health panel actually drive (they POST
# {"mode": "soft"|"hard"} and read back `mode`). Worse, the openclaw-session-file
# path below (`/home/node/.openclaw/...`) lives inside the OPENCLAW container, so
# from the openvoiceui container it never exists — the try/except swallowed the
# miss and this handler degraded to "bump the key" while blocking the richer one.
#
# FIX (additive, no deletion): this route is moved to the explicit, distinct
# '/api/session/reset-openclaw' path. `/api/session/reset` is now unambiguously
# served by server.py's canonical soft/hard handler. Both remain reachable; the
# deep openclaw-file clear is opt-in via the new path.
@conversation_bp.route('/api/session/reset-openclaw', methods=['POST'])
def session_reset():
    """Clear the corrupted openclaw session state and return a fresh session key.
    Deep reset (opt-in): clears the openclaw session JSONL file so orphaned
    messages don't cascade, then bumps the voice session key so the next request
    starts completely fresh. The default /api/session/reset (server.py) handles
    the common soft/hard reset."""
    old_key = get_voice_session_key()
    # Find and clear the openclaw session file for the current session key
    try:
        sessions_dir = Path('/home/node/.openclaw/agents/openvoiceui/sessions')
        sessions_json = sessions_dir / 'sessions.json'
        if sessions_json.exists():
            import json as _json
            sessions_map = _json.loads(sessions_json.read_text())
            # The openclaw session key format is "agent:openvoiceui:<voice_key>"
            oclaw_key = f'agent:openvoiceui:{old_key}'
            session_info = sessions_map.get(oclaw_key, {})
            session_id = session_info.get('sessionId')
            if session_id:
                session_file = sessions_dir / f'{session_id}.jsonl'
                if session_file.exists():
                    _ts = __import__('datetime').datetime.utcnow().isoformat() + 'Z'
                    session_file.write_text('{"type":"session","version":3,"id":"' + session_id + '","timestamp":"' + _ts + '","cwd":"/home/node/.openclaw/workspace"}\n')
                    logger.info(f'### SESSION RESET: cleared openclaw session file {session_id}.jsonl')
    except Exception as e:
        logger.warning(f'### SESSION RESET: could not clear openclaw session file: {e}')
    new_key = bump_voice_session()
    return jsonify({'status': 'ok', 'old': old_key, 'new': new_key})


# ---------------------------------------------------------------------------
# GET /api/tts/providers
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/tts/providers', methods=['GET'])
def tts_providers_list():
    """List all available TTS providers with metadata."""
    try:
        providers = list_providers(include_inactive=True, probe=True)
        config_path = (Path(__file__).parent.parent
                       / 'tts_providers' / 'providers_config.json')
        default_provider = 'supertonic'
        try:
            with open(config_path, 'r') as f:
                config = json.load(f)
                default_provider = config.get('default_provider', 'supertonic')
        except Exception:
            pass
        return jsonify({'providers': providers, 'default_provider': default_provider})
    except Exception as e:
        logger.error(f'Failed to list TTS providers: {e}')
        return jsonify({'error': f'Failed to list providers: {e}'}), 500


# ---------------------------------------------------------------------------
# PUT /api/tts/default-provider
# ---------------------------------------------------------------------------
# Widget/agent can set the default TTS provider. Writes to
# tts_providers/providers_config.json so the change survives restart.
# Validates that the provider is registered AND active — we don't let a
# caller point the default at something that will fail on first /generate.

@conversation_bp.route('/api/tts/default-provider', methods=['PUT', 'POST'])
def tts_set_default_provider():
    """Set the default TTS provider written into providers_config.json.

    Accepts both PUT (semantically correct for "replace current value") and
    POST (agents using OpenAPI-generic tooling often default to POST).

    Body: {"provider": "<provider_id>"} — must be an ID from the registered
    set AND have status=active in the config.
    """
    try:
        data = request.get_json(silent=True) or {}
        provider_id = (data.get('provider') or '').strip()
        if not provider_id:
            return jsonify({'ok': False, 'error': 'provider is required'}), 400

        # Verify it's in the known registry. Using list_providers gives us the
        # same view the GET endpoint exposes — single source of truth.
        providers = list_providers(include_inactive=True)
        provider_ids = {p.get('provider_id') for p in providers}
        if provider_id not in provider_ids:
            return jsonify({
                'ok': False,
                'error': f'unknown provider: {provider_id}',
                'available': sorted(provider_ids),
            }), 400

        # Reject providers whose status is NOT active — avoids setting the
        # default to something we know will fail downstream.
        target = next((p for p in providers if p.get('provider_id') == provider_id), None)
        if target and target.get('status') not in (None, 'active'):
            return jsonify({
                'ok': False,
                'error': (f'provider {provider_id} is not active '
                          f'(status={target.get("status")})'),
            }), 400

        config_path = (Path(__file__).parent.parent
                       / 'tts_providers' / 'providers_config.json')
        try:
            with open(config_path, 'r') as f:
                config = json.load(f)
        except Exception as e:
            return jsonify({'ok': False, 'error': f'failed to read config: {e}'}), 500

        old_default = config.get('default_provider')
        config['default_provider'] = provider_id
        config['last_updated'] = time.strftime('%Y-%m-%d')

        # Atomic write: write to temp, rename into place. Keeps readers from
        # ever seeing a half-written config.
        tmp_path = config_path.with_suffix('.json.tmp')
        try:
            with open(tmp_path, 'w') as f:
                json.dump(config, f, indent=2)
            os.replace(tmp_path, config_path)
        except Exception as e:
            return jsonify({'ok': False, 'error': f'failed to write config: {e}'}), 500

        # Drop the cached config + provider singletons so the next generate
        # picks up the new default without a restart.
        _tts_invalidate()

        logger.info(
            f'TTS default provider changed: {old_default} → {provider_id}'
        )
        return jsonify({
            'ok': True,
            'provider': provider_id,
            'previous': old_default,
        })
    except Exception as e:
        logger.error(f'tts_set_default_provider failed: {e}')
        return jsonify({'ok': False, 'error': str(e)}), 500


# ---------------------------------------------------------------------------
# PUT /api/tts/default-voice
# ---------------------------------------------------------------------------
# Sets the preferred voice for a provider. Providers_config.json keeps a
# per-provider `voices` array whose FIRST element is the effective default.
# We move the requested voice to the front (creating the entry if it's a
# valid voice we know about from /api/tts/voices).

@conversation_bp.route('/api/tts/default-voice', methods=['PUT', 'POST'])
def tts_set_default_voice():
    """Set the default voice for a given TTS provider.

    Body: {"provider": "<provider_id>", "voice": "<voice_id>"}
    If provider is omitted, uses the current default_provider.
    """
    try:
        data = request.get_json(silent=True) or {}
        voice = (data.get('voice') or '').strip()
        if not voice:
            return jsonify({'ok': False, 'error': 'voice is required'}), 400

        config_path = (Path(__file__).parent.parent
                       / 'tts_providers' / 'providers_config.json')
        try:
            with open(config_path, 'r') as f:
                config = json.load(f)
        except Exception as e:
            return jsonify({'ok': False, 'error': f'failed to read config: {e}'}), 500

        provider_id = (data.get('provider') or config.get('default_provider') or '').strip()
        if not provider_id:
            return jsonify({'ok': False, 'error': 'provider is required'}), 400

        pconfig = config.get('providers', {}).get(provider_id)
        if not pconfig:
            return jsonify({
                'ok': False,
                'error': f'unknown provider in config: {provider_id}',
            }), 400

        voices = list(pconfig.get('voices') or [])
        # Move the requested voice to front; add if missing.
        if voice in voices:
            voices.remove(voice)
        voices.insert(0, voice)
        pconfig['voices'] = voices
        config['last_updated'] = time.strftime('%Y-%m-%d')

        tmp_path = config_path.with_suffix('.json.tmp')
        try:
            with open(tmp_path, 'w') as f:
                json.dump(config, f, indent=2)
            os.replace(tmp_path, config_path)
        except Exception as e:
            return jsonify({'ok': False, 'error': f'failed to write config: {e}'}), 500

        # Drop the cached config so the next read reflects the new voice order.
        _tts_invalidate()

        logger.info(f'TTS default voice for {provider_id} → {voice}')
        return jsonify({
            'ok': True,
            'provider': provider_id,
            'voice': voice,
            'voices': voices,
        })
    except Exception as e:
        logger.error(f'tts_set_default_voice failed: {e}')
        return jsonify({'ok': False, 'error': str(e)}), 500


# ---------------------------------------------------------------------------
# POST /api/tts/generate
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/tts/generate', methods=['POST'])
def tts_generate():
    """
    Generate speech from text using the specified TTS provider.

    Request JSON:
        text     : str   — text to synthesize (required)
        provider : str   — provider ID (default: supertonic)
        voice    : str   — voice ID (default: provider default)
        lang     : str   — language code (default: en)
        speed    : float — speech speed (default: provider default)
        options  : dict  — provider-specific options
    Returns: WAV audio file
    """
    try:
        data = request.get_json()
        if not data:
            return jsonify({'error': 'No JSON data provided'}), 400

        text = data.get('text', '').strip()
        if not text:
            return jsonify({'error': 'Text cannot be empty'}), 400

        # Length guard (P7-T3 security audit)
        if len(text) > 2000:
            return jsonify({'error': 'Text too long (max 2000 characters)'}), 400

        provider_id = data.get('provider', 'supertonic')
        voice = data.get('voice', None)
        lang = data.get('lang', 'en')
        speed = data.get('speed', None)
        options = data.get('options', {})

        valid_langs = ['en', 'ko', 'es', 'pt', 'fr', 'zh', 'ja', 'de']
        if lang and lang.lower() not in valid_langs:
            return jsonify({
                'error': f"Invalid language: {lang}. Supported: {', '.join(valid_langs)}"
            }), 400

        if speed is not None:
            try:
                speed = float(speed)
                if speed < 0.25 or speed > 4.0:
                    return jsonify({'error': 'Speed must be between 0.25 and 4.0'}), 400
            except (ValueError, TypeError):
                return jsonify({'error': 'Speed must be a valid number'}), 400

        try:
            provider = get_provider(provider_id)
        except ValueError as e:
            available = ', '.join([p['provider_id'] for p in list_providers()])
            return jsonify({'error': 'Invalid TTS provider', 'available_providers': available}), 400

        logger.info(
            f"TTS request: provider={provider_id}, text='{text[:50]}...', "
            f"voice={voice}, lang={lang}, speed={speed}"
        )

        gen_params = {'text': text}
        if voice is not None:
            gen_params['voice'] = voice
        if lang is not None:
            gen_params['lang'] = lang
        if speed is not None:
            gen_params['speed'] = speed
        gen_params.update(options)

        try:
            audio_bytes = provider.generate_speech(**gen_params)
        except ValueError as e:
            return jsonify({'error': f'Invalid parameter: {e}'}), 400
        except Exception as e:
            logger.error(f'Speech generation failed for {provider_id}: {e}')
            return jsonify({'error': f'Speech generation failed: {e}'}), 500

        provider_format = provider.get_info().get('audio_format', 'wav')
        mime_type = 'audio/mpeg' if provider_format == 'mp3' else 'audio/wav'
        response = make_response(audio_bytes)
        response.headers['Content-Type'] = mime_type
        response.headers['Content-Length'] = len(audio_bytes)
        response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
        response.headers['X-TTS-Provider'] = provider_id
        if voice:
            response.headers['X-TTS-Voice'] = voice
        return response

    except ValueError as e:
        return jsonify({'error': f'Invalid input: {e}'}), 400
    except Exception as e:
        import traceback
        logger.error(f'TTS generate endpoint error: {e}')
        logger.error(traceback.format_exc())
        return jsonify({'error': 'Internal server error'}), 500

# ---------------------------------------------------------------------------
# POST /api/tts/clone — Clone a voice from audio
# ---------------------------------------------------------------------------


# CloneError code -> HTTP status (spec §2.4).
_CLONE_ERROR_HTTP = {
    'unavailable': 503,
    'busy': 503,
    'bad_audio': 422,
    'quota': 429,
    'consent_mismatch': 409,
    'timeout': 504,
    'internal': 500,
    # GPU wire v0 §4 codes. 'unauthorized' / 'tenant_mismatch' are about
    # OVU's own upstream credential (a config alarm on this server), not the
    # caller's session, so the browser gets 503: a 401/403 here would read as
    # "your Clerk session is bad" to the page.
    'unauthorized': 503,
    'tenant_mismatch': 503,
    'bad_request': 400,
    'not_found': 404,
    'disabled': 410,
    'request_conflict': 409,
}
_CLONE_BUSY_RETRY_AFTER_S = 30


def _clone_error_response(err, provider_id):
    """JSON error for a typed CloneError. ``error`` stays a plain string so
    existing callers (voice-studio reads resp.text) keep working."""
    status = _CLONE_ERROR_HTTP.get(err.code, 500)
    resp = jsonify({
        'error': err.message,
        'error_code': err.code,
        'retryable': err.retryable,
        'provider': provider_id,
    })
    resp.status_code = status
    retry_after = err.retry_after
    if err.code == 'busy' and not retry_after:
        retry_after = _CLONE_BUSY_RETRY_AFTER_S
    if retry_after:
        resp.headers['Retry-After'] = str(int(retry_after))
    return resp


@conversation_bp.route('/api/tts/clone', methods=['POST'])
def tts_clone_voice():
    """
    Clone a voice from an audio sample.

    Accepts either:
      - JSON: {"audio_url": "...", "name": "...", "provider": "qwen3", "reference_text": "..."}
      - Multipart form: audio file + name + provider fields

    Provider support is REGISTRY-DRIVEN (voice-clone provider seam,
    docs/jambot/ovu-voice-clone-seam-spec.md §2): a provider is offered when
    its catalog entry has the "voice-cloning" feature and is active, its class
    implements CloneCapableTTSProvider, and the tenant policy
    (VOICE_CLONE_SEAM_DIR/policy.json) allows it. With no policy file the set
    is exactly qwen3, qwen3-local, elevenlabs, resemble (as before):
      - qwen3: Instant clone via fal.ai (~37s). Needs audio URL.
      - elevenlabs: Instant clone (~5-10s). Needs audio file.
      - resemble: Multi-step clone (~1-5min). Needs audio file.

    A clone runs on exactly ONE provider: a failure is returned to the
    caller, never silently retried on another provider (spec §2.4).

    TODO spec §1 (access codes), §3 (consent), §3.4 (size/duration caps),
    §2.3 (async jobs route) — not in this release.

    Returns: JSON with voice_id, name, provider, clone metadata.
    """
    from services.voice_clone import (
        clone_provider_ids, default_clone_provider, tenant_id as _clone_tenant_id,
    )
    from tts_providers.base_provider import (
        CloneCapableTTSProvider, CloneError, CloneRequest,
    )
    import uuid

    provider_id = ''
    try:
        # --- Parse request (JSON or multipart) ---
        save_path = None
        audio_url = None
        default_provider = default_clone_provider()

        if request.is_json:
            data = request.get_json()
            provider_id = data.get('provider', default_provider).strip()
            audio_url = data.get('audio_url', '').strip()
            name = data.get('name', '').strip()
            reference_text = data.get('reference_text', '').strip() or None

            if not audio_url:
                return jsonify({'error': 'audio_url is required'}), 400
            if not name:
                return jsonify({'error': 'name is required'}), 400

        elif 'audio' in request.files:
            from services.paths import UPLOADS_DIR

            provider_id = request.form.get('provider', default_provider).strip()
            audio_file = request.files['audio']
            name = request.form.get('name', '').strip()
            reference_text = request.form.get('reference_text', '').strip() or None

            if not name:
                return jsonify({'error': 'name field is required'}), 400
            if not audio_file.filename:
                return jsonify({'error': 'Empty audio file'}), 400

            ext = Path(audio_file.filename).suffix.lower()
            if ext not in ('.wav', '.mp3', '.m4a', '.ogg', '.webm', '.flac'):
                return jsonify({'error': f'Unsupported audio format: {ext}'}), 400

            safe_name = f"voice_clone_{uuid.uuid4().hex[:12]}{ext}"
            UPLOADS_DIR.mkdir(parents=True, exist_ok=True)
            save_path = UPLOADS_DIR / safe_name
            audio_file.save(str(save_path))
            audio_url = f"{request.host_url.rstrip('/')}/uploads/{safe_name}"
        else:
            return jsonify({
                'error': 'Send JSON with audio_url or multipart form with audio file'
            }), 400

        # --- Validate provider (registry + tenant policy) ---
        clone_providers = clone_provider_ids()
        if provider_id not in clone_providers:
            return jsonify({
                'error': f'Voice cloning not supported for provider "{provider_id}". '
                         f'Supported: {", ".join(clone_providers)}'
            }), 400

        provider = get_provider(provider_id)
        if not isinstance(provider, CloneCapableTTSProvider):
            # Registry says clone-capable but the live instance is not —
            # a wiring defect, never a client error.
            logger.error(f"Voice clone: {provider_id} instance is not CloneCapableTTSProvider")
            return jsonify({'error': 'Internal server error'}), 500
        if not provider.is_available():
            return jsonify({
                'error': f'{provider_id} provider not available (API key not set)'
            }), 503

        clone_req = CloneRequest(
            tenant_id=_clone_tenant_id(),
            request_id=uuid.uuid4().hex,
            name=name,
            audio_path=str(save_path) if save_path else None,
            audio_url=audio_url,
            reference_text=reference_text,
        )

        # --- Route to provider ---
        logger.info(
            f"Voice clone request: provider={provider_id}, name='{name}', "
            f"url={audio_url[:80] if audio_url else 'N/A'}"
        )

        input_error = provider.clone_request_error(clone_req)
        if input_error:
            return jsonify({'error': input_error}), 400

        result = provider.create_clone(clone_req)

        if result.status == 'failed':
            raise CloneError.from_dict(result.error)
        if result.status != 'ready':
            # 'pending' needs the async jobs route (spec §2.3), which does not
            # exist yet. Return the job id so it is not lost, but say plainly
            # that it cannot be polled here.
            logger.error(
                f"Voice clone: {provider_id} returned status={result.status!r} "
                f"job_id={result.job_id!r}; async clones are not supported yet"
            )
            return jsonify({
                'error': 'Async voice clones are not supported yet (no jobs route)',
                'error_code': 'internal',
                'retryable': False,
                'provider': provider_id,
                'status': result.status,
                'job_id': result.job_id,
            }), 500

        return jsonify({
            'status': 'ok',
            'provider': provider_id,
            'voice_id': result.voice_id or '',
            'name': result.name,
            'created_at': result.created_at,
            'clone_time_ms': result.clone_time_ms,
            'embedding_size': result.metadata.get('embedding_size', 0),
            'usage': (
                f'Use voice_id "{result.voice_id or ""}" in '
                f'/api/tts/generate with provider={provider_id}'
            ),
        })

    except CloneError as e:
        logger.error(f"Voice clone failed ({provider_id}): {e}")
        return _clone_error_response(e, provider_id)
    except RuntimeError as e:
        logger.error(f"Voice clone failed: {e}")
        return jsonify({'error': str(e)}), 500
    except Exception as e:
        import traceback
        logger.error(f"Voice clone error: {e}")
        logger.error(traceback.format_exc())
        return jsonify({'error': 'Internal server error'}), 500


# ---------------------------------------------------------------------------
# GET /api/tts/voices — List all voices (built-in + cloned) across providers
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/tts/voices', methods=['GET'])
def tts_voices_list():
    """List all available voices across all providers, including cloned voices."""
    try:
        all_voices = {}
        # probe=True: this is an explicit UI voice-list request, so instantiate
        # cached singletons to surface live cloned_voices (a get_info() field the
        # config catalog doesn't carry).
        for provider_info in list_providers(include_inactive=False, probe=True):
            pid = provider_info.get('provider_id', provider_info.get('name', 'unknown'))
            voices = provider_info.get('voices', [])
            cloned = provider_info.get('cloned_voices', [])
            all_voices[pid] = {
                'builtin': voices,
                'cloned': cloned,
            }
        return jsonify({'voices': all_voices})
    except Exception as e:
        logger.error(f"Failed to list voices: {e}")
        return jsonify({'error': str(e)}), 500


# ---------------------------------------------------------------------------
# DELETE /api/tts/voices/<voice_id> — Retire a cloned voice
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/tts/voices/<voice_id>', methods=['DELETE'])
def tts_delete_voice(voice_id):
    """Retire a cloned voice embedding (renamed, not deleted)."""
    try:
        if not voice_id.startswith('clone_'):
            return jsonify({'error': 'Can only retire cloned voices (clone_*)'}), 400

        from services.paths import VOICE_CLONES_DIR
        voice_dir = VOICE_CLONES_DIR / voice_id

        # Validate path doesn't escape
        try:
            voice_dir.resolve().relative_to(VOICE_CLONES_DIR.resolve())
        except ValueError:
            return jsonify({'error': 'Invalid voice_id'}), 400

        if not voice_dir.exists():
            return jsonify({'error': f'Voice {voice_id} not found'}), 404

        # Rename to .retired instead of removing (NEVER DELETE rule)
        renamed = voice_dir.with_name(voice_dir.name + '.retired')
        voice_dir.rename(renamed)
        logger.info(f"Cloned voice retired: {voice_id}")

        return jsonify({'status': 'ok', 'voice_id': voice_id, 'action': 'retired'})
    except Exception as e:
        logger.error(f"Failed to retire voice {voice_id}: {e}")
        return jsonify({'error': str(e)}), 500


# ---------------------------------------------------------------------------
# POST /api/supertonic-tts  (DEPRECATED — use /api/tts/generate)
# ---------------------------------------------------------------------------


@conversation_bp.route('/api/supertonic-tts', methods=['POST'])
def supertonic_tts_endpoint():
    """
    Generate speech via Supertonic TTS (deprecated — prefer /api/tts/generate).

    Request JSON: text, lang, speed, voice_style
    Returns: WAV audio
    """
    try:
        data = request.get_json()
        if not data:
            return jsonify({'error': 'No JSON data provided'}), 400

        text = data.get('text', '').strip()
        if not text:
            return jsonify({'error': 'Text cannot be empty'}), 400

        lang = data.get('lang', 'en').lower()
        if lang not in ['en', 'ko', 'es', 'pt', 'fr']:
            return jsonify({
                'error': f"Invalid language: {lang}. Supported: en, ko, es, pt, fr"
            }), 400

        speed = float(data.get('speed', 1.0))
        if speed < 0.5 or speed > 2.0:
            return jsonify({'error': 'Speed must be between 0.5 and 2.0'}), 400

        voice_style = data.get('voice_style', 'M1').upper()
        valid_voices = ['M1', 'M2', 'M3', 'M4', 'M5', 'F1', 'F2', 'F3', 'F4', 'F5']
        if voice_style not in valid_voices:
            return jsonify({
                'error': f"Invalid voice: {voice_style}. "
                         f"Available: {', '.join(valid_voices)}"
            }), 400

        logger.info(f"Generating speech: {text[:50]}... (lang={lang}, speed={speed})")

        try:
            tts_instance = get_supertonic_for_voice(voice_style)
        except Exception as e:
            logger.error(f'Failed to initialize TTS with voice {voice_style}: {e}')
            return jsonify({'error': f'Failed to load voice style: {e}'}), 500

        try:
            audio_bytes = tts_instance.generate_speech(
                text=text, lang=lang, speed=speed, total_step=16
            )
        except Exception as e:
            logger.error(f'Speech synthesis failed: {e}')
            return jsonify({'error': f'Speech synthesis failed: {e}'}), 500

        response = make_response(audio_bytes)
        response.headers['Content-Type'] = 'audio/wav'
        response.headers['Content-Length'] = len(audio_bytes)
        response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate'
        return response

    except ValueError as e:
        return jsonify({'error': f'Invalid input: {e}'}), 400
    except Exception as e:
        import traceback
        logger.error(f'TTS endpoint error: {e}')
        logger.error(traceback.format_exc())
        return jsonify({'error': 'Internal server error'}), 500

# ---------------------------------------------------------------------------
# POST /api/tts/preview  (P4-T5: TTS voice preview)
# ---------------------------------------------------------------------------

_PREVIEW_TEXT = "Hello! This is a preview of the selected voice."


@conversation_bp.route('/api/tts/preview', methods=['POST'])
def tts_preview():
    """
    Generate a short audio preview for a given TTS voice.

    Request JSON (all optional):
        provider : str  — TTS provider ID (default: 'supertonic')
        voice    : str  — Voice ID (default: provider default, e.g. 'M1')
        text     : str  — Custom preview text (max 200 chars; default sample phrase)

    Returns JSON:
        audio_b64 : str  — Base64-encoded WAV audio
        provider  : str  — Provider used
        voice     : str  — Voice used
    """
    try:
        data = request.get_json(silent=True) or {}

        provider_id = str(data.get('provider', 'supertonic')).strip()
        voice = data.get('voice', None)
        text = str(data.get('text', _PREVIEW_TEXT)).strip()[:200] or _PREVIEW_TEXT

        # Validate provider exists
        try:
            get_provider(provider_id)
        except ValueError:
            available = ', '.join([p['provider_id'] for p in list_providers()])
            return jsonify({
                'error': f"Unknown provider: {provider_id}",
                'available_providers': available,
            }), 400

        logger.info(f"TTS preview: provider={provider_id}, voice={voice}, text='{text[:40]}'")

        audio_b64 = _tts_generate_b64(
            text=text,
            voice=voice,
            tts_provider=provider_id,
        )

        if audio_b64 is None:
            return jsonify({'error': 'TTS generation failed — check server logs'}), 500

        return jsonify({
            'audio_b64': audio_b64,
            'provider': provider_id,
            'voice': voice or 'default',
        })

    except Exception as e:
        import traceback
        logger.error(f'TTS preview error: {e}')
        logger.error(traceback.format_exc())
        return jsonify({'error': 'Internal server error'}), 500


@conversation_bp.route('/api/stt-events', methods=['POST'])
def stt_events():
    """Receive STT error/status events from the browser.
    Logs them in a format the session monitor can parse from container stdout.
    Only real errors are sent (no-speech and aborted are filtered client-side).
    """
    try:
        data = request.get_json(silent=True) or {}
        error_code = data.get('error', 'unknown')
        message = data.get('message', '')
        provider = data.get('provider', 'webspeech')
        source = data.get('source', 'stt')  # 'stt' or 'wake_word'

        # Log in session-monitor-parseable format
        print(f"### STT_ERROR: {error_code} — {message} (provider={provider} source={source})",
              flush=True)
        return jsonify({'ok': True})
    except Exception:
        return jsonify({'ok': False}), 500


def _suno_failure_prefix() -> str:
    """SUNO-REJECT-SURFACED (2026-09-29). Consume routes.suno.failed_songs_queue ONCE and
    return a [SYSTEM] prefix for the next real user turn ('' when nothing failed).

    A generation that did NOT go through — refused at submit (e.g. out of credits) or failed
    during generation — must reach the agent so it can correct a "ready in 45 seconds" it
    already said. Consuming here (not re-reading every turn) is the point: the old
    context_parts block never cleared the queue.
    """
    try:
        from routes import suno as _suno_mod
        with _suno_mod._suno_lock:
            _taken = list(_suno_mod.failed_songs_queue)[-3:]
            del _suno_mod.failed_songs_queue[:]   # consume once
    except Exception as _e:
        logger.warning(f'Suno failure-note injection failed: {_e}')
        return ''
    if not _taken:
        return ''
    _fail_lines, _out_of_credits, _seen = [], False, set()
    for f in _taken:
        # the status poll AND the callback can both queue the same job
        if f.get('job_id') in _seen:
            continue
        _seen.add(f.get('job_id'))
        _label = f.get('title') or f.get('brand') or f"a {f.get('kind') or 'song'}"
        _reason = str(f.get('reason') or 'unknown error')
        _out_of_credits = _out_of_credits or 'credit' in _reason.lower()
        _fail_lines.append(f'{_label!r} ({_reason[:160]})')
    _hint = ("The platform's music credits are used up, so trying again right now will fail "
             "the same way — offer to make it once that is sorted."
             if _out_of_credits else "Offer to try again.")
    return (
        f'[SYSTEM: Music generation you started did NOT go through: {"; ".join(_fail_lines)}. '
        f'If you told the user it was on its way, correct that now: say plainly that it did '
        f'not go through, with no technical detail and no time promise. {_hint}]\n\n'
    )