le-mystere-professeur-zacus/tools/zacus-gateway/main.py

"""Zacus Hub gateway — thin aggregator for the SwiftUI hub app.

Sprint 1+ scope: real backend probes, WS live state, scenario write/validate,
multipart audio transcribe proxy. See docs/specs/2026-05-24-zacus-hub-app.md.
"""
from __future__ import annotations

import asyncio
import io
import json
import logging
import os
import re
import secrets
import struct
import subprocess
import tempfile
import time
import types
import wave
from contextlib import asynccontextmanager
from pathlib import Path
from typing import AsyncIterator, Literal

import httpx
import yaml
from fastapi import Depends, FastAPI, Form, HTTPException, Request, UploadFile, File, WebSocket, WebSocketDisconnect, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, PlainTextResponse, Response
from pydantic import BaseModel
from pydantic_settings import BaseSettings, SettingsConfigDict

from blocks_to_runtime3 import compile_blocks, CompileError


def _load_boards_registry() -> dict[str, dict]:
    # ZACUS_HUB_BOARDS_FILE lets a local/E2E run point at a working copy with the
    # real LAN IPs without touching the versioned boards.yaml (placeholder IPs).
    override = os.environ.get("ZACUS_HUB_BOARDS_FILE")
    path = Path(override) if override else Path(__file__).parent / "boards.yaml"
    if not path.is_file():
        return {}
    raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    return raw.get("boards") or {}


BOARDS = _load_boards_registry()


def _master_board_name() -> str | None:
    """Name of the board that owns an ESP-NOW relay (the master). First wins."""
    for name, cfg in BOARDS.items():
        if cfg.get("espnow_relay_peers"):
            return name
    return None


class BoardRoute(BaseModel):
    """How to reach a board for hot HTTP control.

    - kind="direct": the board has its own IP; POST straight to http://<ip>.
    - kind="relay":  no IP; reachable only via the master's ESP-NOW relay.
                     `base_url` then points at the master's HTTP API and
                     `relay_via` names the master board.
    """
    board: str
    kind: Literal["direct", "relay"]
    ip: str | None = None           # the board's own IP (direct) or the master's (relay)
    base_url: str | None = None     # http://<ip> to POST against
    relay_via: str | None = None    # master board name, when kind == "relay"


def resolve_board_route(board: str) -> BoardRoute:
    """Resolve a board name to a concrete HTTP target.

    master (has an ip)              -> direct  http://<ip>
    annex (no ip, peer of master)   -> relay   via master's http://<master_ip>
    unknown / unreachable           -> HTTPException(404/422) with a clear reason
    """
    if board not in BOARDS:
        raise HTTPException(404, f"unknown board '{board}' — declare it in boards.yaml")
    cfg = BOARDS[board]
    ip = cfg.get("ip")
    if ip:
        return BoardRoute(board=board, kind="direct", ip=ip, base_url=f"http://{ip}")
    # No IP of its own: look for a master that relays to this board.
    master = _master_board_name()
    if master and board in (BOARDS[master].get("espnow_relay_peers") or []):
        master_ip = BOARDS[master].get("ip")
        if not master_ip:
            raise HTTPException(
                422,
                f"board '{board}' is relayed via '{master}' but '{master}' has no ip in boards.yaml",
            )
        return BoardRoute(
            board=board, kind="relay", ip=master_ip,
            base_url=f"http://{master_ip}", relay_via=master,
        )
    raise HTTPException(
        422,
        f"board '{board}' has no ip and no ESP-NOW relay master — cannot route to it",
    )


REPO_ROOT = Path(__file__).resolve().parents[2]


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="ZACUS_HUB_", env_file=".env", extra="ignore")

    token: str = "change-me-run-gen_token.py"
    voice_bridge_url: str = "http://studio:8200"
    hints_url: str = "http://macm1:8311"
    hints_admin_key: str | None = None  # X-Admin-Key for /hints/sessions, if the engine enforces it
    esp32_url: str | None = None
    scenarios_dir: str = ""  # empty → REPO_ROOT/game/scenarios
    share_dir: str = ""  # empty → REPO_ROOT/share-store (Atelier "Partager" link store)
    ailiance_url: str = "https://gateway.ailiance.fr"  # LLM gateway for text asset generation
    ailiance_tts_url: str = "https://gateway.ailiance.fr"  # TTS endpoint base (same gateway, distinct setting)
    ailiance_tts_model: str = "tts-1"
    ailiance_tts_voice: str = "nova"
    ailiance_chat_model: str = "gpt-4o-mini"
    # Local LLM (vllm-mlx on this Mac M1, OpenAI-compatible) — primary chat
    # backend for the NPC voice loop: native FR, fast, no flaky remote gateway.
    local_chat_url: str = "http://127.0.0.1:8520"
    local_chat_model: str = "granite4-tiny"
    kokoro_tts_url: str = "http://macm1:8520"  # vllm-mlx native /v1/audio/speech (stock voices)
    kyutai_stt_url: str = "http://127.0.0.1:8300"  # local Kyutai stt-1b-en_fr MLX server (/transcribe)
    kyutai_tts_url: str = "http://127.0.0.1:8302"  # local Kyutai tts-1.6b-en_fr MLX server (/tts, native FR)
    request_timeout: float = 10.0
    probe_interval: float = 2.0
    # Browser clients (atelier DeployPanel). Comma-separated origins; "*" for
    # the LAN-tool default — auth still goes through the bearer token.
    cors_origins: str = "*"


settings = Settings()
SCENARIOS_DIR = Path(settings.scenarios_dir) if settings.scenarios_dir else REPO_ROOT / "game" / "scenarios"
SHARE_DIR = Path(settings.share_dir) if settings.share_dir else REPO_ROOT / "share-store"
# Generated audio previews land here before an author promotes them to the
# hotline_tts pool. Served read-only via GET /v1/_staged/{key}.
STAGED_DIR = SHARE_DIR / "_staged"


def load_phone_directory() -> dict[str, dict]:
    """Load game/scenarios/phone_directory.yaml and return the numbers mapping."""
    path = SCENARIOS_DIR / "phone_directory.yaml"
    if not path.is_file():
        return {}
    data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
    return data.get("numbers", {})


PHONE_DIRECTORY: dict[str, dict] = load_phone_directory()


# NOTE: in-process dict — single uvicorn worker only; multi-worker would need shared store (Redis).
class VoiceSessions:
    """In-memory per-session chat history with TTL and turn cap."""

    _SWEEP_EVERY = 50  # purge expired entries every N appends

    def __init__(self, ttl_s: float = 600.0, max_turns: int = 20) -> None:
        self._ttl = ttl_s
        self._max = max_turns
        self._store: dict[str, dict] = {}  # sid -> {msgs: list, ts: float}
        self._n = 0  # append counter for lazy sweep

    def _sweep(self) -> None:
        now = time.time()
        for k in [k for k, v in self._store.items() if now - v["ts"] > self._ttl]:
            del self._store[k]

    def append(self, sid: str, role: str, content: str) -> None:
        self._n += 1
        if self._n % self._SWEEP_EVERY == 0:
            self._sweep()
        if sid not in self._store:
            self._store[sid] = {"msgs": [], "ts": time.time()}
        entry = self._store[sid]
        entry["msgs"].append({"role": role, "content": content})
        entry["ts"] = time.time()
        # cap to max_turns * 2 messages (user + assistant pairs)
        cap = self._max * 2
        if len(entry["msgs"]) > cap:
            entry["msgs"] = entry["msgs"][-cap:]

    def history(self, sid: str) -> list[dict]:
        entry = self._store.get(sid)
        if entry is None:
            return []
        if time.time() - entry["ts"] > self._ttl:
            self._store.pop(sid, None)
            return []
        return list(entry["msgs"])

    def end(self, sid: str) -> None:
        self._store.pop(sid, None)


VOICE_SESSIONS = VoiceSessions()


# Spoken-telephone style: the reply is sent straight to TTS and played in an
# earpiece, so it must be short, plain spoken French — no markdown, no stage
# directions, no lists, no headings.
_PHONE_STYLE = (
    "\n\nCONTRAINTE DE FORME : tu réponds au téléphone. Réponds en UNE seule "
    "phrase courte (15 mots maximum), en français parlé naturel. JAMAIS de "
    "markdown (pas de **, *, #, ---), JAMAIS de didascalies entre crochets, "
    "JAMAIS de listes ni de titres. Uniquement les mots que le personnage "
    "dirait à voix haute, et fais court."
)


# Spoken fallback when the LLM gateway is unreachable/slow: the conversational
# loop must never leave the caller hanging on silence, so the NPC says a natural
# "bad line, repeat please" instead of returning a 502.
_CHAT_FALLBACK = "Excusez-moi, la ligne est très mauvaise, pouvez-vous répéter ?"


async def _chat_one(http: httpx.AsyncClient, url: str, model: str,
                    messages: list[dict], timeout: float) -> str:
    """One OpenAI-compatible chat call. Raises on transport/HTTP/shape error."""
    resp = await http.post(
        f"{url.rstrip('/')}/v1/chat/completions",
        json={
            "model": model,
            "messages": messages,
            "max_tokens": 48,
            # granite4-tiny loops ("parlez par-dessus, parlez par-dessous…")
            # without a repetition penalty; a little temperature + penalty keep
            # the one-sentence phone reply varied and on-track.
            "temperature": 0.7,
            "repetition_penalty": 1.3,
        },
        timeout=timeout,
    )
    if resp.status_code != 200:
        raise RuntimeError(f"chat HTTP {resp.status_code}: {resp.text[:160]}")
    return resp.json()["choices"][0]["message"]["content"].strip()


async def _chat_reply(http: httpx.AsyncClient, persona: str, history: list[dict]) -> str:
    """Produce the NPC reply text from the persona + conversation history.

    LOCAL-ONLY (user requirement): the only LLM is the on-box vllm-mlx granite.
    There is NO remote fallback — if granite fails, we return a LOCAL spoken
    line so the voice loop never hangs, but we never leave this Mac.
    """
    messages = [{"role": "system", "content": persona + _PHONE_STYLE}] + history
    backends = [
        # Local granite only. Generous timeout: granite answers in ~0.4 s idle
        # but is slowed by Metal contention with the on-box Kyutai STT during a
        # turn, so allow it to finish rather than giving up early. On a true hang
        # we fall back to a LOCAL spoken line (never a remote backend).
        ("local", settings.local_chat_url, settings.local_chat_model, 15.0),
    ]
    last_exc: Exception | None = None
    for name, url, model, timeout in backends:
        try:
            return await _chat_one(http, url, model, messages, timeout)
        except Exception as exc:  # noqa: BLE001 — try the next backend on any failure
            last_exc = exc
            logging.warning("LLM backend %s failed (%s) — local spoken fallback", name, type(exc).__name__)
    logging.error("local LLM unreachable (%s) — local spoken fallback", last_exc)
    return _CHAT_FALLBACK


async def _warm_persona(http: httpx.AsyncClient, persona: str) -> None:
    """Prime granite's KV cache with this persona's system-prompt prefix.

    The first granite call for a persona pays ~15-17 s of prompt processing
    (511-char system prompt, slow on the M1 while the Kyutai STT shares Metal);
    later calls reuse the cached prefix in ~0.5 s. Fired in the background at
    greeting time so the player's FIRST reply already hits a warm cache instead
    of timing out. Best-effort: errors are swallowed.
    """
    try:
        messages = [{"role": "system", "content": persona + _PHONE_STYLE},
                    {"role": "user", "content": "Allô ?"}]
        await _chat_one(http, settings.local_chat_url, settings.local_chat_model, messages, 30.0)
        logging.info("LLM persona prewarm done")
    except Exception as exc:  # noqa: BLE001
        logging.warning("LLM persona prewarm failed (%s)", type(exc).__name__)


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Surface our own INFO logs (HEARD/SAID transcriptions, prewarm) — uvicorn
    # only configures its own loggers, leaving the root logger at WARNING.
    root = logging.getLogger()
    if root.level > logging.INFO or not root.handlers:
        logging.basicConfig(level=logging.INFO)
        root.setLevel(logging.INFO)
    app.state.http = httpx.AsyncClient(timeout=settings.request_timeout)
    app.state.health_cache = HealthCache(app.state.http)
    app.state.audio_jobs = {}  # job_id -> AudioJob (in-process async TTS queue)
    STAGED_DIR.mkdir(parents=True, exist_ok=True)

    async def _prewarm_greetings() -> None:
        """Synthesise every directory greeting into the TTS cache in the
        background so the first phone call to each number plays instantly."""
        for number, entry in PHONE_DIRECTORY.items():
            greeting = entry.get("greeting")
            if not greeting:
                continue
            try:
                await _voice_tts_16k(app.state.http, greeting,
                                     entry.get("voice", settings.ailiance_tts_voice))
                logging.info("TTS prewarm: cached greeting for %s", number)
            except Exception as exc:  # noqa: BLE001 — prewarm is best-effort
                logging.warning("TTS prewarm failed for %s: %s", number, str(exc)[:100])

    async def _keep_llm_warm() -> None:
        """Ping the local LLM periodically so it never auto-unloads/cools between
        calls — a cold (re)load takes ~47 s, long enough to blow the per-call
        timeout and fall back. A tiny request keeps the model resident. Ping
        every 4 min: well under the server's 1800 s idle-unload, with margin for
        faster cooling under memory contention. On the very first ping (gateway
        just started) the model loads cold — that's expected and only delays the
        first call, which the 'un instant' filler covers."""
        while True:
            try:
                await _chat_one(
                    app.state.http, settings.local_chat_url, settings.local_chat_model,
                    [{"role": "user", "content": "ok"}], timeout=90.0,
                )
                logging.info("LLM keep-warm: %s resident", settings.local_chat_model)
            except Exception as exc:  # noqa: BLE001 — best-effort
                logging.warning("LLM keep-warm failed: %s", type(exc).__name__)
            await asyncio.sleep(240)  # 4 min ≪ 1800 s idle-unload, margin for contention

    prewarm_task = asyncio.create_task(_prewarm_greetings())
    warm_task = asyncio.create_task(_keep_llm_warm())
    try:
        yield
    finally:
        prewarm_task.cancel()
        warm_task.cancel()
        await app.state.http.aclose()


app = FastAPI(title="Zacus Hub Gateway", version="0.2.0", lifespan=lifespan)

# Browser clients (atelier DeployPanel) hit the gateway cross-origin; native
# clients (SwiftUI hub) are unaffected. Authentication stays on require_token.
app.add_middleware(
    CORSMiddleware,
    allow_origins=[o.strip() for o in settings.cors_origins.split(",") if o.strip()],
    allow_methods=["*"],
    allow_headers=["*"],
)


def require_token(request: Request) -> None:
    auth = request.headers.get("authorization", "")
    if not auth.lower().startswith("bearer "):
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "missing bearer token")
    if auth.split(" ", 1)[1].strip() != settings.token:
        raise HTTPException(status.HTTP_403_FORBIDDEN, "invalid token")


# ---------- Backend health ----------

BackendName = Literal["voice_bridge", "hints", "esp32"]


class BackendHealth(BaseModel):
    name: BackendName
    url: str | None
    online: bool
    latency_ms: float | None = None
    detail: str | None = None


class HealthCache:
    """Probes upstreams at most every `probe_interval` seconds."""

    def __init__(self, client: httpx.AsyncClient) -> None:
        self.client = client
        self._cache: dict[BackendName, BackendHealth] = {}
        self._stamps: dict[BackendName, float] = {}

    def targets(self) -> dict[BackendName, str | None]:
        return {
            "voice_bridge": settings.voice_bridge_url,
            "hints": settings.hints_url,
            "esp32": settings.esp32_url,
        }

    async def probe(self, name: BackendName, url: str | None) -> BackendHealth:
        if not url:
            return BackendHealth(name=name, url=None, online=False, detail="not configured")
        start = time.perf_counter()
        try:
            resp = await self.client.get(f"{url.rstrip('/')}/healthz", timeout=3.0)
            elapsed = (time.perf_counter() - start) * 1000
            return BackendHealth(
                name=name,
                url=url,
                online=200 <= resp.status_code < 500,
                latency_ms=round(elapsed, 1),
                detail=f"HTTP {resp.status_code}",
            )
        except Exception as exc:
            elapsed = (time.perf_counter() - start) * 1000
            return BackendHealth(name=name, url=url, online=False, latency_ms=round(elapsed, 1), detail=str(exc)[:120])

    async def get(self, force: bool = False) -> list[BackendHealth]:
        now = time.monotonic()
        results: list[BackendHealth] = []
        for name, url in self.targets().items():
            if not force and (now - self._stamps.get(name, 0)) < settings.probe_interval and name in self._cache:
                results.append(self._cache[name])
                continue
            health = await self.probe(name, url)
            self._cache[name] = health
            self._stamps[name] = now
            results.append(health)
        return results


# ---------- Routes: meta ----------

@app.get("/healthz", response_class=PlainTextResponse)
async def healthz() -> str:
    return "ok"


@app.get("/v1/auth/ping")
async def auth_ping(_: None = Depends(require_token)) -> dict[str, str]:
    return {"status": "authenticated"}


@app.get("/v1/backends/health", response_model=list[BackendHealth])
async def backends_health(request: Request, _: None = Depends(require_token)) -> list[BackendHealth]:
    return await request.app.state.health_cache.get(force=True)


# ---------- Game master state ----------

class SessionSnapshot(BaseModel):
    """Per-group live state derived from the hints engine session tracker."""
    session_id: str
    total_hints: int = 0
    total_penalty: int = 0
    active_puzzle: str | None = None  # most recently touched puzzle for the group
    last_activity_ms: int = 0


class GameState(BaseModel):
    scene_id: str | None = None
    scene_index: int = 0
    last_hint: str | None = None
    voice_session: str | None = None
    backends: list[BackendHealth] = []
    # Live game read-model aggregated from the hints engine (sprint 1).
    sessions: list[SessionSnapshot] = []
    hints_total: int = 0
    sessions_active: int = 0
    state_detail: str | None = None  # human note when live state is partial/unavailable


def _hints_admin_headers() -> dict[str, str]:
    return {"X-Admin-Key": settings.hints_admin_key} if settings.hints_admin_key else {}


def _summarise_sessions(payload: dict) -> tuple[list[SessionSnapshot], int]:
    """Reduce the hints engine /hints/sessions dump to the hub read-model."""
    snapshots: list[SessionSnapshot] = []
    total_hints = 0
    for raw in payload.get("sessions", []) or []:
        puzzles = raw.get("puzzles", []) or []
        # The most recently touched puzzle stands in for the group's current step.
        latest = max(puzzles, key=lambda p: p.get("last_at_ms", 0), default=None)
        session_hints = int(raw.get("total_hints", 0))
        total_hints += session_hints
        snapshots.append(SessionSnapshot(
            session_id=str(raw.get("session_id", "?")),
            total_hints=session_hints,
            total_penalty=int(raw.get("total_penalty", 0)),
            active_puzzle=(latest or {}).get("puzzle_id"),
            last_activity_ms=int((latest or {}).get("last_at_ms", 0)),
        ))
    snapshots.sort(key=lambda s: s.last_activity_ms, reverse=True)
    return snapshots, total_hints


async def build_state(request: Request) -> GameState:
    health = await request.app.state.health_cache.get()
    state = GameState(backends=health)

    hints_online = any(b.name == "hints" and b.online for b in health)
    if not hints_online:
        state.state_detail = "hints engine offline — live session state unavailable"
        return state

    # Pull the live session read-model from the hints engine. This is the only
    # backend that currently exposes per-group game progress; scene_id/voice_session
    # stay null until the ESP32 master / voice-bridge expose equivalent state.
    try:
        resp = await request.app.state.http.get(
            f"{settings.hints_url.rstrip('/')}/hints/sessions",
            headers=_hints_admin_headers(),
            timeout=3.0,
        )
        if resp.status_code == 401:
            state.state_detail = "hints /hints/sessions requires X-Admin-Key — set ZACUS_HUB_HINTS_ADMIN_KEY"
            return state
        if resp.status_code >= 400:
            state.state_detail = f"hints /hints/sessions → HTTP {resp.status_code}"
            return state
        sessions, total_hints = _summarise_sessions(resp.json())
    except Exception as exc:
        state.state_detail = f"hints state probe failed: {str(exc)[:120]}"
        return state

    state.sessions = sessions
    state.hints_total = total_hints
    state.sessions_active = len(sessions)
    # Surface the most-recently-active group's puzzle as the headline scene so the
    # GM panel reflects real progress even before scene-level state exists upstream.
    if sessions:
        head = sessions[0]
        state.scene_id = head.active_puzzle
        state.last_hint = f"{head.session_id}: {head.total_hints} indice(s), pénalité {head.total_penalty}"
    return state


@app.get("/v1/state", response_model=GameState)
async def get_state(request: Request, _: None = Depends(require_token)) -> GameState:
    return await build_state(request)


@app.websocket("/v1/state/ws")
async def state_ws(ws: WebSocket) -> None:
    token = ws.query_params.get("token", "")
    if token != settings.token:
        await ws.close(code=1008)
        return
    await ws.accept()
    try:
        while True:
            state = await build_state(ws)  # type: ignore[arg-type]
            await ws.send_json(state.model_dump())
            await asyncio.sleep(settings.probe_interval)
    except WebSocketDisconnect:
        return


# ---------- Game master actions (sprint 2 stubs) ----------

class HintTrigger(BaseModel):
    scene: str
    level: int = 1


@app.post("/v1/gm/hint")
async def gm_hint(request: Request, body: HintTrigger, _: None = Depends(require_token)) -> JSONResponse:
    """Forward a hint request to the hints engine on behalf of the GM."""
    upstream = await request.app.state.http.post(
        f"{settings.hints_url}/hints/ask",
        json={"scene": body.scene, "level": body.level, "source": "gm"},
    )
    return JSONResponse(content=upstream.json(), status_code=upstream.status_code)


# ---------- Companion: voice proxies ----------

@app.post("/v1/companion/voice/intent")
async def voice_intent(request: Request, _: None = Depends(require_token)) -> JSONResponse:
    body = await request.body()
    upstream = await request.app.state.http.post(
        f"{settings.voice_bridge_url}/voice/intent",
        content=body,
        headers={"content-type": request.headers.get("content-type", "application/json")},
    )
    return JSONResponse(content=upstream.json(), status_code=upstream.status_code)


@app.post("/v1/companion/voice/transcribe")
async def voice_transcribe(
    request: Request,
    audio: UploadFile = File(...),
    _: None = Depends(require_token),
) -> JSONResponse:
    """Multipart audio → voice-bridge /voice/transcribe (which fans out to whisper.cpp)."""
    data = await audio.read()
    # whisper.cpp's /inference (which the bridge proxies to verbatim) expects
    # the multipart file field named "file"; "audio" was silently ignored ->
    # empty transcription through the gateway.
    files = {"file": (audio.filename or "clip.wav", data, audio.content_type or "audio/wav")}
    upstream = await request.app.state.http.post(
        f"{settings.voice_bridge_url}/voice/transcribe",
        files=files,
        timeout=30.0,
    )
    try:
        payload = upstream.json()
    except Exception:
        payload = {"raw": upstream.text[:500]}
    return JSONResponse(content=payload, status_code=upstream.status_code)


@app.post("/v1/companion/voice/tts")
async def voice_tts(request: Request, _: None = Depends(require_token)) -> Response:
    """Proxy → voice-bridge /tts. Returns raw audio bytes."""
    body = await request.body()
    upstream = await request.app.state.http.post(
        f"{settings.voice_bridge_url}/tts",
        content=body,
        headers={"content-type": request.headers.get("content-type", "application/json")},
        timeout=60.0,
    )
    return Response(
        content=upstream.content,
        status_code=upstream.status_code,
        media_type=upstream.headers.get("content-type", "audio/wav"),
    )


@app.post("/v1/companion/hint")
async def hint_ask(request: Request, _: None = Depends(require_token)) -> JSONResponse:
    body = await request.json()
    upstream = await request.app.state.http.post(f"{settings.hints_url}/hints/ask", json=body)
    return JSONResponse(content=upstream.json(), status_code=upstream.status_code)


# ---------- Studio ----------

class ScenarioMeta(BaseModel):
    name: str
    path: str
    size: int
    modified: float


class ScenarioWrite(BaseModel):
    yaml: str


class ValidationResult(BaseModel):
    ok: bool
    errors: list[str] = []
    warnings: list[str] = []
    top_level_keys: list[str] = []


def _resolve_scenario(name: str) -> Path:
    target = (SCENARIOS_DIR / name).resolve()
    base = SCENARIOS_DIR.resolve()
    if base != target.parent and base not in target.parents:
        raise HTTPException(404, "scenario not found")
    return target


@app.get("/v1/studio/scenarios", response_model=list[ScenarioMeta])
async def list_scenarios(_: None = Depends(require_token)) -> list[ScenarioMeta]:
    if not SCENARIOS_DIR.exists():
        return []
    return [
        ScenarioMeta(
            name=p.name,
            path=str(p.relative_to(SCENARIOS_DIR.parent.parent)) if SCENARIOS_DIR.parent.parent in p.parents else p.name,
            size=p.stat().st_size,
            modified=p.stat().st_mtime,
        )
        for p in sorted(SCENARIOS_DIR.glob("*.yaml"))
    ]


@app.get("/v1/studio/scenario/{name}")
async def get_scenario(name: str, _: None = Depends(require_token)) -> dict:
    target = _resolve_scenario(name)
    if not target.is_file():
        raise HTTPException(404, "scenario not found")
    text = target.read_text(encoding="utf-8")
    return {"name": name, "yaml": text, "modified": target.stat().st_mtime}


@app.put("/v1/studio/scenario/{name}", response_model=ValidationResult)
async def write_scenario(name: str, body: ScenarioWrite, _: None = Depends(require_token)) -> ValidationResult:
    target = _resolve_scenario(name)
    validation = _validate_yaml(body.yaml)
    if not validation.ok:
        raise HTTPException(422, detail=validation.model_dump())
    # backup first
    if target.exists():
        backup = target.with_suffix(target.suffix + f".bak.{int(time.time())}")
        backup.write_bytes(target.read_bytes())
    target.write_text(body.yaml, encoding="utf-8")
    return validation


class RenameRequest(BaseModel):
    new_name: str


def _scenario_meta(p: Path) -> ScenarioMeta:
    st = p.stat()
    return ScenarioMeta(
        name=p.name,
        path=str(p.relative_to(SCENARIOS_DIR.parent.parent)) if SCENARIOS_DIR.parent.parent in p.parents else p.name,
        size=st.st_size,
        modified=st.st_mtime,
    )


@app.delete("/v1/studio/scenario/{name}")
async def delete_scenario(name: str, _: None = Depends(require_token)) -> dict:
    """Delete a scenario YAML. _resolve_scenario guards against path traversal."""
    target = _resolve_scenario(name)
    if not target.is_file():
        raise HTTPException(404, "scenario not found")
    target.unlink()
    return {"deleted": name}


@app.post("/v1/studio/scenario/{name}/duplicate", response_model=ScenarioMeta)
async def duplicate_scenario(name: str, body: RenameRequest, _: None = Depends(require_token)) -> ScenarioMeta:
    """Copy scenario `name` → `new_name`. Both names pass through _resolve_scenario."""
    src = _resolve_scenario(name)
    if not src.is_file():
        raise HTTPException(404, "scenario not found")
    dst = _resolve_scenario(body.new_name)
    if dst.exists():
        raise HTTPException(409, f"destination '{body.new_name}' already exists")
    dst.write_bytes(src.read_bytes())
    return _scenario_meta(dst)


@app.post("/v1/studio/scenario/{name}/rename", response_model=ScenarioMeta)
async def rename_scenario(name: str, body: RenameRequest, _: None = Depends(require_token)) -> ScenarioMeta:
    """Rename scenario `name` → `new_name`. Both names pass through _resolve_scenario."""
    src = _resolve_scenario(name)
    if not src.is_file():
        raise HTTPException(404, "scenario not found")
    dst = _resolve_scenario(body.new_name)
    if dst.exists():
        raise HTTPException(409, f"destination '{body.new_name}' already exists")
    src.rename(dst)
    return _scenario_meta(dst)


# ---------- Share (Atelier "Partager" — public short links) ----------
#
# Deliberately token-free: a player opening a shared link has no bearer token.
# Scope is intentionally narrow — store a blocks_studio v2 YAML, hand back a
# short id, serve it back by id. Size-capped and YAML-validated to keep the
# store from becoming an open dumping ground.

SHARE_ID_RE = re.compile(r"[A-Za-z0-9_-]{4,32}")
SHARE_MAX_BYTES = 512 * 1024


class ShareCreate(BaseModel):
    yaml: str


class ShareResult(BaseModel):
    id: str


def _share_path(share_id: str) -> Path:
    if not SHARE_ID_RE.fullmatch(share_id):
        raise HTTPException(404, "share not found")
    return SHARE_DIR / f"{share_id}.yaml"


@app.post("/v1/share", response_model=ShareResult)
async def create_share(body: ShareCreate) -> ShareResult:
    if len(body.yaml.encode("utf-8")) > SHARE_MAX_BYTES:
        raise HTTPException(413, "scenario too large to share")
    validation = _validate_yaml(body.yaml)
    if not validation.ok:
        raise HTTPException(422, detail=validation.model_dump())
    SHARE_DIR.mkdir(parents=True, exist_ok=True)
    for _ in range(5):
        share_id = secrets.token_urlsafe(6)[:8]
        path = SHARE_DIR / f"{share_id}.yaml"
        if not path.exists():
            path.write_text(body.yaml, encoding="utf-8")
            return ShareResult(id=share_id)
    raise HTTPException(500, "could not allocate a share id, retry")


@app.get("/v1/share/{share_id}")
async def get_share(share_id: str) -> dict:
    path = _share_path(share_id)
    if not path.is_file():
        raise HTTPException(404, "share not found")
    return {"id": share_id, "yaml": path.read_text(encoding="utf-8")}


# ---------- Asset generation (P1: text via gateway.ailiance.fr LLM) ----------
#
# Token-free like /v1/share — the Atelier calls it same-origin from the browser.
# Proxies the ailiance LLM gateway (OpenAI-compatible) with a per-kind system
# prompt so authors generate in-voice NPC lines, hints and screen text.

_TEXTGEN_SYSTEM = {
    "npc_reply": (
        "Tu es le Professeur Zacus, savant excentrique d'une escape room. "
        "Réponds par UNE phrase courte, mystérieuse, en français, sans guillemets."
    ),
    "hint": (
        "Tu es le maître du jeu d'une escape room. Donne un indice progressif, "
        "clair et bref, en français, sans jamais révéler directement la solution."
    ),
    "scene_text": (
        "Tu écris un court texte d'écran (titre ou sous-titre) pour une borne "
        "d'escape room. Français, percutant, 60 caractères maximum, sans guillemets."
    ),
    "freeform": "Tu assistes l'auteur d'une escape room. Réponds en français, concis.",
}


class TextGenRequest(BaseModel):
    kind: str = "freeform"   # npc_reply | hint | scene_text | freeform
    prompt: str              # author instruction / context
    model: str = "ailiance"
    max_tokens: int = 160
    temperature: float = 0.8


class TextGenResult(BaseModel):
    kind: str
    text: str
    model: str | None = None


@app.post("/v1/assets/text/generate", response_model=TextGenResult)
async def generate_text(body: TextGenRequest, request: Request) -> TextGenResult:
    if not body.prompt or not body.prompt.strip():
        raise HTTPException(400, "prompt is required")
    system = _TEXTGEN_SYSTEM.get(body.kind, _TEXTGEN_SYSTEM["freeform"])
    payload = {
        "model": body.model,
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": body.prompt},
        ],
        "max_tokens": max(16, min(body.max_tokens, 512)),
        "temperature": max(0.0, min(body.temperature, 1.5)),
    }
    try:
        resp = await request.app.state.http.post(
            f"{settings.ailiance_url.rstrip('/')}/v1/chat/completions",
            json=payload, timeout=30.0,
        )
    except Exception as exc:
        raise HTTPException(502, f"ailiance unreachable: {str(exc)[:120]}") from exc
    if resp.status_code != 200:
        raise HTTPException(502, f"ailiance HTTP {resp.status_code}: {resp.text[:120]}")
    data = resp.json()
    try:
        text = data["choices"][0]["message"]["content"].strip()
    except (KeyError, IndexError, TypeError) as exc:
        raise HTTPException(502, "ailiance: unexpected response shape") from exc
    return TextGenResult(kind=body.kind, text=text, model=data.get("model"))


# ---------- P2: Audio asset generation (async TTS queue) ----------
# Token-free / same-origin like /v1/assets/text/generate. Generation can run
# well past the gateway request timeout (F5 cold synth, long lines), so it is
# dispatched to an in-process asyncio job and polled via /status. The WAV is
# staged under share-store/_staged and previewed via /v1/_staged/{key}.

_AUDIO_KEY_RE = re.compile(r"[^a-zA-Z0-9_-]+")


def _audio_staged_key(text: str, backend: str, voice: str, key: str | None) -> str:
    """Stable, filesystem-safe key for a generated clip."""
    if key:
        return _AUDIO_KEY_RE.sub("_", key).strip("_")[:80] or "clip"
    digest = secrets.token_hex(4)
    slug = _AUDIO_KEY_RE.sub("_", text.lower()).strip("_")[:32] or "clip"
    return f"{backend}_{slug}_{digest}"


class AudioGenRequest(BaseModel):
    text: str
    backend: Literal["f5", "kokoro"] = "f5"   # f5=cloned Zacus voice, kokoro=stock
    voice: str = "ff_siwis"                    # kokoro voice id (ignored by f5)
    key: str | None = None                     # manifest/phrase id; default = auto slug


class AudioJob(BaseModel):
    id: str
    status: Literal["pending", "running", "done", "error"]
    backend: str
    staged_key: str | None = None
    error: str | None = None
    created: float
    updated: float


async def _synthesise_wav(http: httpx.AsyncClient, body: AudioGenRequest) -> bytes:
    """Call the selected TTS backend, return WAV bytes."""
    if body.backend == "kokoro":
        resp = await http.post(
            f"{settings.kokoro_tts_url.rstrip('/')}/v1/audio/speech",
            json={"model": "kokoro", "voice": body.voice,
                  "input": body.text, "response_format": "wav"},
            timeout=120.0,
        )
        if resp.status_code != 200:
            raise RuntimeError(f"kokoro HTTP {resp.status_code}: {resp.text[:160]}")
        return resp.content
    if body.backend == "ailiance":
        voice = body.voice if body.voice else settings.ailiance_tts_voice
        resp = await http.post(
            f"{settings.ailiance_tts_url.rstrip('/')}/v1/audio/speech",
            json={"model": settings.ailiance_tts_model, "input": body.text,
                  "voice": voice, "response_format": "wav"},
            timeout=120.0,
        )
        if resp.status_code != 200:
            raise RuntimeError(f"ailiance TTS HTTP {resp.status_code}: {resp.text[:160]}")
        return resp.content
    # default: F5 cloned Zacus voice via the voice-bridge
    resp = await http.post(
        f"{settings.voice_bridge_url.rstrip('/')}/tts",
        json={"text": body.text}, timeout=120.0,
    )
    if resp.status_code != 200:
        raise RuntimeError(f"voice-bridge HTTP {resp.status_code}: {resp.text[:160]}")
    return resp.content


try:  # high-quality anti-aliased resampling (avoids the harsh aliasing of naive linear interp)
    import numpy as _np
    import soxr as _soxr
except Exception:  # pragma: no cover
    _np = None
    _soxr = None


def _softknee_compressor(x, sr: int, threshold: float, ratio: float = 3.0,
                         knee_db: float = 6.0, attack_ms: float = 5.0,
                         release_ms: float = 120.0):
    """Feed-forward soft-knee compressor (float samples in/out).

    Evens out the voice dynamics (quiet parts up, loud parts down) so the
    perceived level is uniform before the limiter. `threshold` is a linear
    amplitude in the same units as x. Smooth (soft) knee around the threshold.
    """
    n = len(x)
    if n == 0:
        return x
    eps = 1e-9
    thr_db = 20.0 * _np.log10(threshold + eps)
    half_knee = knee_db / 2.0
    aa = _np.exp(-1.0 / (sr * attack_ms / 1000.0))
    ar = _np.exp(-1.0 / (sr * release_ms / 1000.0))
    absx = _np.abs(x)
    # Peak envelope follower (one-pole, separate attack/release) — Python loop.
    env = _np.empty(n, dtype=_np.float64)
    e = 0.0
    for i in range(n):
        a = float(absx[i])
        coef = aa if a > e else ar
        e = a + (e - a) * coef
        env[i] = e
    lvl_db = 20.0 * _np.log10(env + eps)
    over = lvl_db - thr_db
    gain_db = _np.zeros(n, dtype=_np.float64)
    above = over > half_knee
    in_knee = _np.abs(over) <= half_knee
    gain_db[above] = (1.0 / ratio - 1.0) * over[above]
    k = over[in_knee] + half_knee
    gain_db[in_knee] = (1.0 / ratio - 1.0) * (k * k) / (2.0 * knee_db)
    return x * (10.0 ** (gain_db / 20.0))


def _lookahead_limiter(x, ceiling: float, sr: int,
                       lookahead_ms: float = 5.0, release_ms: float = 60.0):
    """Brick-wall look-ahead limiter (float in/out). Guarantees |out| <= ceiling
    without hard clipping: gain ducks *before* a peak (look-ahead = instant
    attack) and recovers smoothly (release one-pole)."""
    n = len(x)
    if n == 0:
        return x
    la = max(1, int(sr * lookahead_ms / 1000.0))
    absx = _np.abs(x)
    desired = _np.ones(n, dtype=_np.float64)
    over = absx > ceiling
    desired[over] = ceiling / absx[over]
    env = desired.copy()
    for s in range(1, la + 1):
        env[: n - s] = _np.minimum(env[: n - s], desired[s:])
    rel = _np.exp(-1.0 / (sr * release_ms / 1000.0))
    g = _np.empty(n, dtype=_np.float64)
    prev = 1.0
    for i in range(n):
        t = float(env[i])
        prev = t if t < prev else t + (prev - t) * rel
        g[i] = prev
    return x * g


def _wav_to_16k_mono(wav_bytes: bytes, max_seconds: float | None = None) -> tuple[bytes, float, bool]:
    """Pure-Python WAV resampler: any-rate mono/stereo → 16000 Hz mono 16-bit PCM.

    Does NOT use audioop (removed in Python 3.13) or ffmpeg.
    Returns (wav16_bytes, duration_s, truncated).
    """
    with wave.open(io.BytesIO(wav_bytes), "rb") as src:
        src_rate = src.getframerate()
        src_channels = src.getnchannels()
        src_width = src.getsampwidth()
        src_nframes = src.getnframes()
        raw = src.readframes(src_nframes)

    # Decode to signed 16-bit integers (handles 8, 16, 24, 32-bit input)
    if src_width == 1:
        # 8-bit WAV is unsigned; convert to signed 16-bit
        samples_8 = struct.unpack(f"{len(raw)}B", raw)
        interleaved = [(s - 128) << 8 for s in samples_8]
    elif src_width == 2:
        n = len(raw) // 2
        interleaved = list(struct.unpack(f"<{n}h", raw))
    elif src_width == 3:
        # 24-bit: 3 bytes little-endian signed → shift to 16-bit
        n = len(raw) // 3
        interleaved = []
        for i in range(n):
            b0, b1, b2 = raw[i * 3], raw[i * 3 + 1], raw[i * 3 + 2]
            val = (b2 << 16) | (b1 << 8) | b0
            if val >= 0x800000:
                val -= 0x1000000
            interleaved.append(val >> 8)
    elif src_width == 4:
        n = len(raw) // 4
        interleaved = [s >> 16 for s in struct.unpack(f"<{n}i", raw)]
    else:
        raise ValueError(f"unsupported WAV sample width: {src_width}")

    # Downmix stereo → mono
    if src_channels == 1:
        mono = interleaved
    elif src_channels == 2:
        mono = [(interleaved[i * 2] + interleaved[i * 2 + 1]) // 2
                for i in range(len(interleaved) // 2)]
    else:
        # Multi-channel: average all channels
        mono = [
            sum(interleaved[i * src_channels:(i + 1) * src_channels]) // src_channels
            for i in range(len(interleaved) // src_channels)
        ]

    # Resample to 16000 Hz. Prefer soxr (anti-aliased, VHQ) — naive linear interp
    # downsampling folds >8 kHz content back into the band → harsh "saturated"
    # artefacts. Fall back to linear interp only if soxr/numpy are unavailable.
    dst_rate = 16000
    if src_rate == dst_rate:
        out_samples = mono
    elif _soxr is not None and _np is not None:
        arr = _np.asarray(mono, dtype=_np.float32)
        res = _soxr.resample(arr, src_rate, dst_rate, quality="VHQ")
        out_samples = res.astype(_np.int32).tolist()
    else:
        ratio = src_rate / dst_rate
        src_len = len(mono)
        dst_len = int(src_len / ratio)
        out_samples = []
        for i in range(dst_len):
            pos = i * ratio
            idx = int(pos)
            frac = pos - idx
            a = mono[idx]
            b = mono[idx + 1] if idx + 1 < src_len else a
            out_samples.append(int(a + frac * (b - a)))

    # Apply max_seconds cap
    truncated = False
    if max_seconds is not None:
        max_frames = int(max_seconds * dst_rate)
        if len(out_samples) > max_frames:
            out_samples = out_samples[:max_frames]
            truncated = True

    duration_s = len(out_samples) / dst_rate

    # ── Dynamics chain: normalize → soft-knee compressor → look-ahead limiter ──
    # Evens out the voice (uniform perceived level) and guarantees no clipping.
    if _np is not None and out_samples:
        FS = 32767.0
        x = _np.asarray(out_samples, dtype=_np.float64)
        peak = float(_np.max(_np.abs(x)))
        if peak > 0:
            x *= min(0.95 * FS / peak, 12.0)              # normalize to a known scale
        # Soft-knee compressor: tame peaks above ~0.30 FS, 3:1, then make up gain.
        x = _softknee_compressor(x, dst_rate, threshold=0.30 * FS, ratio=3.0, knee_db=6.0)
        peak2 = float(_np.max(_np.abs(x)))
        if peak2 > 0:
            x *= min(0.95 * FS / peak2, 12.0)             # makeup → back to 0.95 FS
        # Brick-wall look-ahead limiter at 0.95 FS (no hard clip, catches overshoot).
        x = _lookahead_limiter(x, ceiling=0.95 * FS, sr=dst_rate)
        out_samples = _np.clip(_np.round(x), -32768, 32767).astype(_np.int32).tolist()
    else:
        # Fallback (no numpy): simple peak normalize + hard clamp.
        peak = max((abs(s) for s in out_samples), default=0)
        if peak > 0:
            gain = min(0.70 * 32767 / peak, 12.0)
            out_samples = [int(s * gain) for s in out_samples]
        out_samples = [max(-32768, min(32767, s)) for s in out_samples]

    # Pack back to WAV
    pcm = struct.pack(f"<{len(out_samples)}h", *out_samples)
    buf = io.BytesIO()
    with wave.open(buf, "wb") as dst:
        dst.setnchannels(1)
        dst.setsampwidth(2)
        dst.setframerate(dst_rate)
        dst.writeframes(pcm)
    return buf.getvalue(), duration_s, truncated


async def _run_audio_job(app: FastAPI, job_id: str, body: AudioGenRequest, staged_key: str) -> None:
    jobs = app.state.audio_jobs
    job = jobs[job_id]
    job.status = "running"
    job.updated = time.time()
    try:
        wav = await _synthesise_wav(app.state.http, body)
        (STAGED_DIR / f"{staged_key}.wav").write_bytes(wav)
        job.staged_key = staged_key
        job.status = "done"
    except Exception as exc:  # noqa: BLE001 — surfaced verbatim to the author
        job.error = str(exc)[:200]
        job.status = "error"
    finally:
        job.updated = time.time()


@app.post("/v1/assets/audio/generate", response_model=AudioJob)
async def generate_audio(body: AudioGenRequest, request: Request) -> AudioJob:
    if not body.text or not body.text.strip():
        raise HTTPException(400, "text is required")
    staged_key = _audio_staged_key(body.text, body.backend, body.voice, body.key)
    now = time.time()
    job = AudioJob(id=secrets.token_urlsafe(8), status="pending",
                   backend=body.backend, created=now, updated=now)
    request.app.state.audio_jobs[job.id] = job
    asyncio.create_task(_run_audio_job(request.app, job.id, body, staged_key))
    return job


@app.get("/v1/assets/audio/{job_id}/status", response_model=AudioJob)
async def audio_status(job_id: str, request: Request) -> AudioJob:
    job = request.app.state.audio_jobs.get(job_id)
    if job is None:
        raise HTTPException(404, "job not found")
    return job


class StagedAsset(BaseModel):
    key: str
    size: int
    modified: float


def _resolve_staged(key: str) -> Path:
    """Resolve a client key to a *.wav under STAGED_DIR, guarding traversal.

    Mirrors the cleanup/guard used by GET /v1/_staged/{key} so the staged store
    never sees a raw client path.
    """
    safe = _AUDIO_KEY_RE.sub("_", key).strip("_")
    target = (STAGED_DIR / f"{safe}.wav").resolve()
    if STAGED_DIR.resolve() not in target.parents:
        raise HTTPException(404, "staged asset not found")
    return target


@app.get("/v1/_staged", response_model=list[StagedAsset])
async def list_staged(_: None = Depends(require_token)) -> list[StagedAsset]:
    """List staged audio assets (the *.wav under STAGED_DIR), newest first."""
    if not STAGED_DIR.exists():
        return []
    assets = [
        StagedAsset(key=p.stem, size=p.stat().st_size, modified=p.stat().st_mtime)
        for p in STAGED_DIR.glob("*.wav")
        if p.is_file()
    ]
    assets.sort(key=lambda a: a.modified, reverse=True)
    return assets


@app.get("/v1/_staged/{key}")
async def staged_preview(key: str) -> Response:
    target = _resolve_staged(key)
    if not target.is_file():
        raise HTTPException(404, "staged asset not found")
    return Response(content=target.read_bytes(), media_type="audio/wav")


@app.delete("/v1/_staged/{key}")
async def delete_staged(key: str, _: None = Depends(require_token)) -> dict:
    """Delete a staged audio asset. Same key cleanup + STAGED_DIR guard as GET."""
    target = _resolve_staged(key)
    if not target.is_file():
        raise HTTPException(404, "staged asset not found")
    target.unlink()
    return {"deleted": target.stem}


# ---------- P4: stage assets onto a board's SD card ----------
# Pushes the staged audio (and any other staged file) to a board's microSD via
# the firmware's existing POST /game/file?path=sd/... route, which already
# mkdir -p's intermediate dirs and writes under /sdcard. The on-card layout is
# /sdcard/zacus/{board}/<file> (the plan's /sd/zacus/{board_id}/). No firmware
# change is needed; NFS and a PULL-side /assets/sync remain hardware-gated.

class AssetStageRequest(BaseModel):
    board: str                 # board name from boards.yaml
    keys: list[str] = []       # staged asset keys; empty = every *.wav staged
    strategy: str = "auto"     # auto | hot (push over HTTP) | cold (host bundle)


class AssetStageStep(BaseModel):
    asset: str
    status: str                # ok | warn | skip | error
    detail: str = ""


class AssetStageResult(BaseModel):
    board: str
    strategy_used: str
    ok: bool
    base_path: str             # on-card dir or host bundle dir
    steps: list[AssetStageStep]
    manifest: list[str] = []   # filenames staged


def _staged_assets(keys: list[str]) -> list[Path]:
    files = sorted(p for p in STAGED_DIR.glob("*.wav") if p.is_file())
    if keys:
        wanted = {_AUDIO_KEY_RE.sub("_", k).strip("_") for k in keys}
        files = [p for p in files if p.stem in wanted]
    return files


@app.post("/v1/assets/stage", response_model=AssetStageResult)
async def stage_assets(body: AssetStageRequest, request: Request, _: None = Depends(require_token)) -> AssetStageResult:
    if body.board not in BOARDS:
        raise HTTPException(404, f"unknown board '{body.board}' — declare it in boards.yaml")
    board = BOARDS[body.board]
    assets = _staged_assets(body.keys)
    base = f"zacus/{body.board}"
    steps: list[AssetStageStep] = []
    manifest: list[str] = []
    ok = True

    if not assets:
        return AssetStageResult(board=body.board, strategy_used="none", ok=False,
                                base_path=f"/sdcard/{base}",
                                steps=[AssetStageStep(asset="-", status="skip",
                                       detail="no staged assets (generate some in the Audio panel first)")])

    strategy = body.strategy
    if strategy == "auto":
        strategy = "hot" if board.get("ip") else "cold"

    if strategy == "hot":
        ip = board.get("ip")
        if not ip:
            return AssetStageResult(board=body.board, strategy_used="hot", ok=False,
                                    base_path=f"/sdcard/{base}",
                                    steps=[AssetStageStep(asset="-", status="error",
                                           detail=f"no IP for '{body.board}' — set it in boards.yaml")])
        for p in assets:
            url = f"http://{ip}/game/file?path=sd/{base}/{p.name}"
            try:
                resp = await request.app.state.http.post(
                    url, content=p.read_bytes(),
                    headers={"content-type": "audio/wav"}, timeout=30.0)
                if 200 <= resp.status_code < 300:
                    steps.append(AssetStageStep(asset=p.name, status="ok", detail=f"{resp.status_code}"))
                    manifest.append(p.name)
                else:
                    detail = "sd_not_mounted" in resp.text and "SD card not mounted on board" or resp.text[:120]
                    steps.append(AssetStageStep(asset=p.name, status="error", detail=f"{resp.status_code} {detail}"))
                    ok = False
            except Exception as exc:  # noqa: BLE001
                steps.append(AssetStageStep(asset=p.name, status="error", detail=str(exc)[:160]))
                ok = False
        base_path = f"/sdcard/{base}"
    elif strategy == "cold":
        bundle = SHARE_DIR / "_bundles" / body.board
        bundle.mkdir(parents=True, exist_ok=True)
        for p in assets:
            (bundle / p.name).write_bytes(p.read_bytes())
            steps.append(AssetStageStep(asset=p.name, status="ok", detail="bundled"))
            manifest.append(p.name)
        (bundle / "manifest.json").write_text(
            __import__("json").dumps({"board": body.board, "base": f"/sdcard/{base}", "files": manifest}, indent=2),
            encoding="utf-8")
        base_path = str(bundle.relative_to(REPO_ROOT)) if REPO_ROOT in bundle.parents else str(bundle)
        steps.append(AssetStageStep(asset="manifest.json", status="ok",
                     detail=f"copy this dir to the SD card under /{base}/"))
    else:
        raise HTTPException(400, f"unknown strategy '{strategy}' (auto|hot|cold)")

    return AssetStageResult(board=body.board, strategy_used=strategy, ok=ok,
                            base_path=base_path, steps=steps, manifest=manifest)


class BoardInfo(BaseModel):
    name: str
    label: str
    type: str
    ip: str | None
    mdns: str | None
    hot_endpoint: str | None
    cold_data_dir: str | None
    espnow_relay_peers: list[str] = []


@app.get("/v1/flash/boards", response_model=list[BoardInfo])
async def list_boards(_: None = Depends(require_token)) -> list[BoardInfo]:
    out = []
    for name, cfg in BOARDS.items():
        out.append(BoardInfo(
            name=name,
            label=cfg.get("label", name),
            type=cfg.get("type", "unknown"),
            ip=cfg.get("ip"),
            mdns=cfg.get("mdns"),
            hot_endpoint=cfg.get("hot_endpoint"),
            cold_data_dir=cfg.get("cold_data_dir"),
            espnow_relay_peers=cfg.get("espnow_relay_peers") or [],
        ))
    return out


class FlashRequest(BaseModel):
    scenario: str            # YAML filename (the blocks_studio v2 source)
    strategy: str = "auto"   # "auto" | "hot" | "cold"


class FlashStep(BaseModel):
    label: str
    status: str              # "ok" | "warn" | "skip" | "error"
    detail: str = ""


class FlashResult(BaseModel):
    board: str
    strategy_used: str
    ok: bool
    steps: list[FlashStep]
    ir_path: str | None = None
    cold_command: str | None = None
    relayed_to: list[str] = []


@app.post("/v1/flash/{board_name}", response_model=FlashResult)
async def flash_board(board_name: str, body: FlashRequest, request: Request, _: None = Depends(require_token)) -> FlashResult:
    if board_name not in BOARDS:
        raise HTTPException(404, f"unknown board '{board_name}' — declare it in boards.yaml")
    board = BOARDS[board_name]

    # 1. compile current YAML to IR
    yaml_path = _resolve_scenario(body.scenario)
    if not yaml_path.is_file():
        raise HTTPException(404, f"scenario '{body.scenario}' not found")
    try:
        ir = compile_blocks(yaml_path.read_text(encoding="utf-8"),
                            scenario_id=body.scenario.replace(".yaml", "").upper())
    except CompileError as exc:
        raise HTTPException(422, f"compile: {exc}") from exc
    import json
    ir_text = json.dumps(ir, indent=2, ensure_ascii=False)
    ir_path = yaml_path.with_suffix(".ir.json")
    ir_path.write_text(ir_text, encoding="utf-8")

    steps: list[FlashStep] = [FlashStep(label="compile IR", status="ok",
                                         detail=f"{len(ir['steps'])} steps → {ir_path.name}")]

    strategy = body.strategy
    if strategy == "auto":
        strategy = "hot" if board.get("ip") else "cold"

    cold_command: str | None = None
    relayed_to: list[str] = []
    ok = True

    if strategy == "hot":
        # Try POST to firmware /game/scenario
        ip = board.get("ip")
        hot_endpoint = board.get("hot_endpoint") or "/game/scenario"
        if not ip:
            steps.append(FlashStep(label="hot POST", status="error",
                                   detail=f"no IP for board '{board_name}' — set ip in boards.yaml or use ESP-NOW relay via master"))
            ok = False
        else:
            url = f"http://{ip}{hot_endpoint}"
            try:
                resp = await request.app.state.http.post(url, content=ir_text,
                                                          headers={"content-type": "application/json"},
                                                          timeout=20.0)
                if 200 <= resp.status_code < 300:
                    steps.append(FlashStep(label="hot POST", status="ok",
                                           detail=f"{resp.status_code} {resp.text[:120]}"))
                elif resp.status_code == 404:
                    steps.append(FlashStep(label="hot POST", status="error",
                                           detail=f"404 — firmware doesn't expose {hot_endpoint} yet. Apply firmware Phase 2 spec, or use strategy=cold."))
                    ok = False
                else:
                    steps.append(FlashStep(label="hot POST", status="error",
                                           detail=f"{resp.status_code} {resp.text[:200]}"))
                    ok = False
            except Exception as exc:
                steps.append(FlashStep(label="hot POST", status="error", detail=str(exc)[:200]))
                ok = False

        # If master and ok, relay via ESP-NOW to peers
        if ok and board.get("espnow_relay_peers"):
            relay_url = f"http://{ip}/game/scenario/relay"
            try:
                resp = await request.app.state.http.post(
                    relay_url,
                    json={"peers": board["espnow_relay_peers"], "ir": ir},
                    timeout=20.0,
                )
                if 200 <= resp.status_code < 300:
                    relayed_to = list(board["espnow_relay_peers"])
                    steps.append(FlashStep(label="ESP-NOW relay", status="ok",
                                           detail=f"forwarded to {','.join(relayed_to)}"))
                else:
                    steps.append(FlashStep(label="ESP-NOW relay", status="warn",
                                           detail=f"{resp.status_code} — relay endpoint not ready (Phase 2)"))
            except Exception as exc:
                steps.append(FlashStep(label="ESP-NOW relay", status="warn", detail=f"skipped: {exc}"[:200]))

    elif strategy == "cold":
        cold_dir = board.get("cold_data_dir")
        if not cold_dir:
            steps.append(FlashStep(label="cold stage", status="error",
                                   detail=f"no cold_data_dir for '{board_name}'"))
            ok = False
        else:
            # Stage the IR into the firmware data dir on the gateway host
            # (assumes the repo is cloned alongside the gateway; otherwise just
            # report the file path back to the developer).
            stage_dir = REPO_ROOT / cold_dir
            if stage_dir.is_dir():
                target = stage_dir / "scenario.json"
                target.write_text(ir_text, encoding="utf-8")
                steps.append(FlashStep(label="cold stage", status="ok", detail=f"wrote {target}"))
                cold_command = _cold_command_for(board.get("type", ""), cold_dir)
            else:
                steps.append(FlashStep(label="cold stage", status="warn",
                                       detail=f"directory {stage_dir} not present on gateway — copy IR manually from {ir_path}"))
                cold_command = _cold_command_for(board.get("type", ""), cold_dir)
    else:
        raise HTTPException(400, f"unknown strategy '{strategy}' (auto|hot|cold)")

    return FlashResult(
        board=board_name,
        strategy_used=strategy,
        ok=ok,
        steps=steps,
        ir_path=str(ir_path.relative_to(REPO_ROOT)) if REPO_ROOT in ir_path.parents else str(ir_path),
        cold_command=cold_command,
        relayed_to=relayed_to,
    )


# ---------- Board-aware runtime control (media cue + step arm) ----------
#
# Both routes resolve a board → HTTP target via resolve_board_route(). The
# master is hit directly on http://<ip>; annexes (no IP) are relayed via the
# master's ESP-NOW. For relayed boards the firmware-side consumer of the relayed
# CMD does not exist yet (D5 contract) — we issue a best-effort
# POST /game/espnow/cmd and flag it as a warning so the caller is not misled.

def _resolve_media_path(path: str) -> str:
    """Resolve a client media path to an ABSOLUTE firmware path.

    The firmware's POST /game/media/play needs an absolute path: a relative
    one is mis-resolved (firmware prepends `/sdcard/music/`). Conventions:
      - already absolute (`/sdcard/...`, `/littlefs/...`) → pass through;
      - `apps/...`  → `/littlefs/apps/...` (the /game/file apps/ write target);
      - anything else (e.g. the staged `zacus/<board>/<file>`) → `/sdcard/...`
        (mirrors POST /v1/assets/stage which writes under /sdcard/zacus/<board>/).
    """
    p = path.strip()
    if p.startswith("/sdcard/") or p.startswith("/littlefs/"):
        return p
    if p.startswith("apps/"):
        return f"/littlefs/{p}"
    return f"/sdcard/{p.lstrip('/')}"


class MediaPlayRequest(BaseModel):
    board: str
    path: str            # client path, e.g. "zacus/master/cue.wav" or "apps/foo/cue.wav"


class MediaPlayResult(BaseModel):
    board: str
    route: Literal["direct", "relay"]
    ok: bool
    status: int | None = None    # firmware HTTP status (direct or relay POST)
    detail: str = ""


def _is_master_board(board: str) -> bool:
    """Return True if this board is the master (has espnow_relay_peers)."""
    return bool(BOARDS.get(board, {}).get("espnow_relay_peers"))


def _board_has_ip(board: str) -> bool:
    return bool(BOARDS.get(board, {}).get("ip"))


@app.post("/v1/media/play", response_model=MediaPlayResult)
async def media_play(body: MediaPlayRequest, request: Request, _: None = Depends(require_token)) -> MediaPlayResult:
    if not body.path or not body.path.strip():
        raise HTTPException(400, "path is required")
    route = resolve_board_route(body.board)
    fw_path = _resolve_media_path(body.path)

    # Master board (has espnow_relay_peers) → use its native /game/media/play endpoint.
    if route.kind == "direct" and _is_master_board(body.board):
        url = f"{route.base_url}/game/media/play"
        try:
            resp = await request.app.state.http.post(
                url, json={"path": fw_path},
                headers={"content-type": "application/json"}, timeout=20.0)
        except Exception as exc:  # noqa: BLE001
            return MediaPlayResult(board=body.board, route="direct", ok=False,
                                   detail=f"POST {url} failed: {str(exc)[:160]}")
        ok = 200 <= resp.status_code < 300
        return MediaPlayResult(board=body.board, route="direct", ok=ok,
                               status=resp.status_code, detail=resp.text[:200])

    # Annex board with an IP → route directly to its /game/cmd endpoint (WiFi-direct,
    # no ESP-NOW channel mismatch). This path is preferred over ESP-NOW relay.
    if route.kind == "direct" and _board_has_ip(body.board):
        url = f"{route.base_url}/game/cmd"
        cmd_payload = {"op": "play", "a": {"p": fw_path, "l": 0}}
        try:
            resp = await request.app.state.http.post(
                url, json=cmd_payload,
                headers={"content-type": "application/json"}, timeout=20.0)
        except Exception as exc:  # noqa: BLE001
            return MediaPlayResult(board=body.board, route="direct", ok=False,
                                   detail=f"POST {url} failed: {str(exc)[:160]}")
        ok = 200 <= resp.status_code < 300
        return MediaPlayResult(board=body.board, route="direct", ok=ok,
                               status=resp.status_code,
                               detail=f"direct /game/cmd: {resp.text[:200]}")

    # Fallback: no direct IP — relay via master's ESP-NOW (fire-and-forget).
    # 2xx from master means frame was emitted, not that the annex ran it.
    url = f"{route.base_url}/game/espnow/cmd"
    cmd = json.dumps({"op": "play", "a": {"p": fw_path, "l": 0}}, separators=(",", ":"))
    try:
        resp = await request.app.state.http.post(
            url, json={"peer": body.board, "command": cmd},
            headers={"content-type": "application/json"}, timeout=20.0)
    except Exception as exc:  # noqa: BLE001
        return MediaPlayResult(board=body.board, route="relay", ok=False,
                               detail=f"relay POST {url} (via {route.relay_via}) failed: {str(exc)[:160]}")
    ok = 200 <= resp.status_code < 300
    return MediaPlayResult(
        board=body.board, route="relay", ok=ok, status=resp.status_code,
        detail=(f"structured CMD {cmd} relayed to '{body.board}' via master "
                f"'{route.relay_via}' ESP-NOW (HTTP {resp.status_code}): {resp.text[:90]}"))


class StepRequest(BaseModel):
    step_id: str


class StepResult(BaseModel):
    board: str
    route: Literal["direct", "relay"]
    ok: bool
    status: int | None = None
    armed: str | None = None     # qr | sound | none (from firmware) on success
    detail: str = ""


@app.post("/v1/step/{board}", response_model=StepResult)
async def step_board(board: str, body: StepRequest, request: Request, _: None = Depends(require_token)) -> StepResult:
    if not body.step_id or not body.step_id.strip():
        raise HTTPException(400, "step_id is required")
    route = resolve_board_route(board)

    # Master board → use its native /game/step endpoint.
    if route.kind == "direct" and _is_master_board(board):
        url = f"{route.base_url}/game/step"
        try:
            resp = await request.app.state.http.post(
                url, json={"step_id": body.step_id},
                headers={"content-type": "application/json"}, timeout=20.0)
        except Exception as exc:  # noqa: BLE001
            return StepResult(board=board, route="direct", ok=False,
                              detail=f"POST {url} failed: {str(exc)[:160]}")
        ok = 200 <= resp.status_code < 300
        armed: str | None = None
        if ok:
            try:
                armed = resp.json().get("armed")
            except Exception:  # noqa: BLE001
                armed = None
        return StepResult(board=board, route="direct", ok=ok,
                          status=resp.status_code, armed=armed, detail=resp.text[:200])

    # Annex board with an IP → route directly to its /game/cmd endpoint (WiFi-direct).
    if route.kind == "direct" and _board_has_ip(board):
        url = f"{route.base_url}/game/cmd"
        cmd_payload = {"op": "step", "a": {"s": body.step_id}}
        try:
            resp = await request.app.state.http.post(
                url, json=cmd_payload,
                headers={"content-type": "application/json"}, timeout=20.0)
        except Exception as exc:  # noqa: BLE001
            return StepResult(board=board, route="direct", ok=False,
                              detail=f"POST {url} failed: {str(exc)[:160]}")
        ok = 200 <= resp.status_code < 300
        return StepResult(board=board, route="direct", ok=ok,
                          status=resp.status_code,
                          detail=f"direct /game/cmd: {resp.text[:200]}")

    # Fallback: no direct IP — relay via master's ESP-NOW. Fire-and-forget.
    url = f"{route.base_url}/game/espnow/cmd"
    cmd = json.dumps({"op": "step", "a": {"s": body.step_id}}, separators=(",", ":"))
    try:
        resp = await request.app.state.http.post(
            url, json={"peer": board, "command": cmd},
            headers={"content-type": "application/json"}, timeout=20.0)
    except Exception as exc:  # noqa: BLE001
        return StepResult(board=board, route="relay", ok=False,
                          detail=f"relay POST {url} failed: {str(exc)[:160]}")
    ok = 200 <= resp.status_code < 300
    return StepResult(
        board=board, route="relay", ok=ok, status=resp.status_code,
        detail=(f"structured CMD {cmd} relayed to '{board}' via master "
                f"'{route.relay_via}' ESP-NOW (HTTP {resp.status_code}): {resp.text[:90]}"))


# ---------- P5: TTS orchestration — synthesise, resample, push + play on board ----------

_MAX_BODY_BYTES = 240 * 1024  # firmware /game/file hard cap (PLIP MAX_BODY=256K, margin) — streaming upload/playback since PR #18


class VoiceSayRequest(BaseModel):
    board: str
    text: str
    voice: str | None = None
    backend: str = "ailiance"
    max_seconds: float = 7.5
    filename: str = "zacus_npc.wav"


class VoiceSayResult(BaseModel):
    board: str
    ok: bool
    tts_backend: str
    bytes: int
    seconds: float
    truncated: bool
    push_status: int | None = None
    play_status: int | None = None
    detail: str = ""


@app.post("/v1/voice/say", response_model=VoiceSayResult)
async def voice_say(body: VoiceSayRequest, request: Request, _: None = Depends(require_token)) -> VoiceSayResult:
    """Synthesise speech, resample to 16 kHz mono, push to board via /game/file, play via /game/cmd."""
    if not body.text or not body.text.strip():
        raise HTTPException(400, "text is required")
    if body.board not in BOARDS:
        raise HTTPException(404, f"unknown board '{body.board}' — declare it in boards.yaml")
    if not _board_has_ip(body.board):
        raise HTTPException(400, f"board '{body.board}' has no IP — cannot push audio")

    ip = BOARDS[body.board]["ip"]

    # 1. Synthesise WAV via selected backend
    try:
        synth_req = types.SimpleNamespace(text=body.text, backend=body.backend, voice=body.voice or "")
        raw_wav = await _synthesise_wav(request.app.state.http, synth_req)  # type: ignore[arg-type]
    except Exception as exc:
        raise HTTPException(502, f"TTS failed ({body.backend}): {str(exc)[:200]}") from exc

    # 2. Resample to 16 kHz mono with cap; loop shrinking cap until < 64 KB
    max_s = body.max_seconds
    truncated = False
    wav16 = b""
    duration_s = 0.0
    for _ in range(10):  # safety limit on iterations
        try:
            wav16, duration_s, truncated = _wav_to_16k_mono(raw_wav, max_seconds=max_s)
        except Exception as exc:
            raise HTTPException(500, f"resample failed: {str(exc)[:200]}") from exc
        if len(wav16) <= _MAX_BODY_BYTES:
            break
        # Recalculate a tighter cap so we stay safely under 64 KB
        # 16 kHz mono 16-bit = 32000 bytes/s; subtract WAV header margin (~44 bytes)
        max_s = (_MAX_BODY_BYTES - 128) / 32000
        truncated = True

    if len(wav16) > _MAX_BODY_BYTES:
        raise HTTPException(413, f"WAV still exceeds 64 KB ({len(wav16)} bytes) after cap — firmware cannot accept it")

    # 3. Push to board via /game/file?path=<filename>
    # The firmware handler prepends /spiffs internally, so we pass only the filename.
    push_url = f"http://{ip}/game/file?path={body.filename}"
    try:
        push_resp = await request.app.state.http.post(
            push_url, content=wav16,
            headers={"content-type": "audio/wav"}, timeout=20.0,
        )
        push_status = push_resp.status_code
    except Exception as exc:
        return VoiceSayResult(
            board=body.board, ok=False, tts_backend=body.backend,
            bytes=len(wav16), seconds=round(duration_s, 3), truncated=truncated,
            detail=f"push to {push_url} failed: {str(exc)[:200]}",
        )

    if push_status != 200:
        return VoiceSayResult(
            board=body.board, ok=False, tts_backend=body.backend,
            bytes=len(wav16), seconds=round(duration_s, 3), truncated=truncated,
            push_status=push_status,
            detail=f"push HTTP {push_status}: {push_resp.text[:120]}",
        )

    # 4. Play via /game/cmd
    play_url = f"http://{ip}/game/cmd"
    play_payload = {"op": "play", "a": {"p": f"/spiffs/{body.filename}", "l": 0}}
    try:
        play_resp = await request.app.state.http.post(
            play_url, json=play_payload,
            headers={"content-type": "application/json"}, timeout=10.0,
        )
        play_status = play_resp.status_code
        ok = 200 <= play_status < 300
        detail = f"push OK, play {play_status}: {play_resp.text[:120]}"
    except Exception as exc:
        return VoiceSayResult(
            board=body.board, ok=False, tts_backend=body.backend,
            bytes=len(wav16), seconds=round(duration_s, 3), truncated=truncated,
            push_status=push_status,
            detail=f"push OK, play POST failed: {str(exc)[:200]}",
        )

    return VoiceSayResult(
        board=body.board, ok=ok, tts_backend=body.backend,
        bytes=len(wav16), seconds=round(duration_s, 3), truncated=truncated,
        push_status=push_status, play_status=play_status, detail=detail,
    )


def _cold_command_for(board_type: str, cold_dir: str) -> str:
    if board_type in ("idf_zacus", "box3_voice", "puzzle"):
        proj_dir = cold_dir.rsplit("/data", 1)[0] if cold_dir.endswith("/data") else cold_dir
        return f"cd {proj_dir} && idf.py -p $(ls /dev/cu.usbserial* /dev/cu.SLAB* 2>/dev/null | head -1) flash monitor"
    if board_type == "plip_firmware":
        return "cd PLIP_FIRMWARE && pio run --target upload"
    return f"# manual flash needed — IR staged in {cold_dir}/scenario.json"


class CompileSummary(BaseModel):
    ok: bool
    steps_count: int
    entry_step_id: str | None
    errors: list[str] = []
    warnings: list[str] = []


@app.post("/v1/studio/scenario/{name}/compile", response_model=CompileSummary)
async def compile_scenario(name: str, _: None = Depends(require_token)) -> CompileSummary:
    """Read the scenario YAML, run blocks→Runtime 3 IR, write the IR next to it."""
    target = _resolve_scenario(name)
    if not target.is_file():
        raise HTTPException(404, "scenario not found")
    text = target.read_text(encoding="utf-8")
    try:
        ir = compile_blocks(text, scenario_id=name.replace(".yaml", "").upper())
    except CompileError as exc:
        raise HTTPException(422, detail=str(exc)) from exc
    ir_path = target.with_suffix(".ir.json")
    import json
    ir_path.write_text(json.dumps(ir, indent=2, ensure_ascii=False), encoding="utf-8")
    return CompileSummary(
        ok=True,
        steps_count=len(ir["steps"]),
        entry_step_id=ir["scenario"].get("entry_step_id") or None,
        errors=ir["metadata"]["errors"],
        warnings=ir["metadata"]["warnings"],
    )


@app.post("/v1/studio/scenario/{name}/validate", response_model=ValidationResult)
async def validate_scenario(name: str, body: ScenarioWrite | None = None, _: None = Depends(require_token)) -> ValidationResult:
    if body is not None:
        return _validate_yaml(body.yaml)
    target = _resolve_scenario(name)
    if not target.is_file():
        raise HTTPException(404, "scenario not found")
    return _validate_yaml(target.read_text(encoding="utf-8"))


def _validate_yaml(text: str) -> ValidationResult:
    errors: list[str] = []
    warnings: list[str] = []
    keys: list[str] = []
    try:
        parsed = yaml.safe_load(text)
    except yaml.YAMLError as exc:
        return ValidationResult(ok=False, errors=[f"YAML parse: {exc}"])
    if not isinstance(parsed, dict):
        return ValidationResult(ok=False, errors=["top-level must be a mapping"])
    keys = sorted(parsed.keys())
    # Soft schema hints aligned with zacus_v3.yaml — we don't enforce strictly here.
    for required in ("id", "version"):
        if required not in parsed:
            warnings.append(f"missing recommended top-level key '{required}'")
    if "acts" not in parsed and "steps_narrative" not in parsed and "scenes" not in parsed:
        warnings.append("no 'acts' / 'steps_narrative' / 'scenes' found — scenario looks empty")
    return ValidationResult(ok=True, errors=errors, warnings=warnings, top_level_keys=keys)


# ---------- P5: PLIP telephone — voice turn ----------

async def _tts_kyutai(http: httpx.AsyncClient, text: str, voice: str | None = None) -> bytes:
    """Synthesise text via the local Kyutai tts-1.6b-en_fr MLX server (native FR).

    Returns raw WAV bytes (24 kHz mono). `voice` is an optional Kyutai voice path
    (e.g. 'cml-tts/fr/...'); omit to use the server's default French voice.
    The OpenAI-style voice names in the directory (nova/alloy) are NOT Kyutai
    voices, so they are ignored here — the server picks its default FR voice.
    """
    payload: dict = {"text": text}
    if voice and "/" in voice:  # only forward genuine Kyutai voice paths
        payload["voice"] = voice
    resp = await http.post(
        f"{settings.kyutai_tts_url.rstrip('/')}/tts",
        json=payload,
        timeout=60.0,
    )
    resp.raise_for_status()
    return resp.content


_TTS_CACHE: dict[str, tuple[bytes, float, bool]] = {}
_TTS_CACHE_MAX = 64

# macOS `say` is the TTS engine: it emits a 16 kHz mono WAV directly, INSTANTLY
# and reliably (no MLX, no slow synthesis, no warmup). The previous Kyutai MLX
# TTS ran at ~0.3x realtime (~30-50 s/reply); `say` is ~real-time. French voice.
SAY_VOICE_FR = "Jacques"  # built-in fr_FR; `say -v '?'` lists others (Audrey, Eddy, Flo…)
# Optional per-NPC voice variety: map directory voice tags → `say` voices.
_SAY_VOICE_MAP = {"nova": "Audrey", "alloy": "Jacques"}


def _tts_say_sync(text: str, voice: str) -> bytes:
    """macOS `say` → 16 kHz mono 16-bit WAV bytes. Blocking (~real-time); call
    via asyncio.to_thread so the event loop isn't stalled."""
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
        path = tmp.name
    try:
        subprocess.run(
            ["say", "-v", voice, "--file-format=WAVE",
             "--data-format=LEI16@16000", "-o", path, text],
            check=True, capture_output=True, timeout=30,
        )
        with open(path, "rb") as f:
            return f.read()
    finally:
        try:
            os.unlink(path)
        except OSError:
            pass


async def _voice_tts_16k(http: httpx.AsyncClient, text: str, voice: str) -> tuple[bytes, float, bool]:
    """Synthesise text to a 16 kHz mono WAV for the PLIP via macOS `say`, then
    run the dynamics chain (normalise/compress/limit) for a consistent level.
    Cached by (voice, text) — greetings are fixed so repeats are free."""
    key = f"{voice}|{text}"
    cached = _TTS_CACHE.get(key)
    if cached is not None:
        return cached

    say_voice = _SAY_VOICE_MAP.get(voice, SAY_VOICE_FR)
    raw = await asyncio.to_thread(_tts_say_sync, text, say_voice)
    # say already emits 16 kHz mono → _wav_to_16k_mono skips resampling and just
    # applies the compressor + look-ahead limiter for a clean, even level.
    result = _wav_to_16k_mono(raw, max_seconds=7.5)

    if len(_TTS_CACHE) >= _TTS_CACHE_MAX:
        _TTS_CACHE.clear()
    _TTS_CACHE[key] = result
    return result


def _wav_pcm(wav: bytes) -> tuple[bytes, int, int, int] | None:
    """Parse a WAV by hand → (pcm, framerate, nchannels, sampwidth). Robust to a
    placeholder/zero data-chunk length (the PLIP streams /voice/capture with
    data_size = 0xFFFFFFFF), which the stdlib `wave` module chokes on."""
    if len(wav) < 44 or wav[:4] != b"RIFF" or wav[8:12] != b"WAVE":
        return None
    i, fmt = 12, None
    while i + 8 <= len(wav):
        cid = wav[i:i + 4]
        sz = int.from_bytes(wav[i + 4:i + 8], "little")
        i += 8
        if cid == b"fmt ":
            ch = int.from_bytes(wav[i + 2:i + 4], "little")
            sr = int.from_bytes(wav[i + 4:i + 8], "little")
            sw = int.from_bytes(wav[i + 14:i + 16], "little") // 8
            fmt = (sr, ch, sw)
            i += sz + (sz & 1)
        elif cid == b"data":
            avail = len(wav) - i
            dsz = avail if (sz == 0 or sz > avail or sz == 0xFFFFFFFF) else sz
            return (wav[i:i + dsz], *fmt) if fmt else None
        else:
            i += sz + (sz & 1)
    return None


def _build_wav(pcm: bytes, sr: int, ch: int, sw: int) -> bytes:
    out = io.BytesIO()
    with wave.open(out, "wb") as w:
        w.setnchannels(ch)
        w.setsampwidth(sw)
        w.setframerate(sr)
        w.writeframes(pcm)
    return out.getvalue()


def _normalize_wav_for_stt(wav_bytes: bytes, target_rms: float = 0.20,
                           max_gain: float = 50.0, hp_cutoff_hz: float = 110.0) -> bytes:
    """Condition a PLIP handset capture for the STT: high-pass then RMS-amplify.

    Two problems with the SLIC handset path: (1) low-frequency rumble/DC drift
    that swamps the speech — a high-pass (~110 Hz, box moving-average subtraction)
    removes it; (2) the voice body comes in VERY quiet (~0.5-1.5 % FS RMS). We
    amplify by RMS (NOT peak — a transient peak defeats peak-normalisation and
    leaves the body inaudible) to ~20 % FS, then hard-clip the rare transient.
    Verified: weak captures whisper returned '' for transcribe correctly once
    RMS-boosted (e.g. → "C'est parti !"). Silence gets blown up to noise but the
    whisper hallucination filter drops the result."""
    if _np is None:
        return wav_bytes
    parsed = _wav_pcm(wav_bytes)
    if parsed is None:
        return wav_bytes
    pcm, sr, ch, sw = parsed
    if sw != 2 or not pcm:
        return wav_bytes
    arr = _np.frombuffer(pcm, dtype=_np.int16).astype(_np.float32)
    # High-pass = signal minus its low-frequency moving average. Window length
    # sets the cutoff (~sr/k Hz); k≈145 @16 kHz → ~110 Hz.
    k = max(2, int(sr / max(hp_cutoff_hz, 1.0)))
    if arr.size > k:
        lp = _np.convolve(arr, _np.ones(k, dtype=_np.float32) / k, mode="same")
        arr = arr - lp
    rms = float(_np.sqrt(_np.mean(arr * arr))) or 1.0
    gain = min((target_rms * 32767.0) / rms, max_gain)
    arr = _np.clip(arr * gain, -32768, 32767).astype(_np.int16)
    logging.info("STT conditioning: high-pass %.0f Hz, RMS %.2f%% FS → gain x%.1f",
                 hp_cutoff_hz, 100 * rms / 32768, gain)
    return _build_wav(arr.tobytes(), sr, ch, sw)


def _pad_wav_silence(wav_bytes: bytes, ms: int = 800) -> bytes:
    """Append `ms` of trailing silence to a WAV. Kyutai STT is streaming with a
    ~0.5 s delay + semantic VAD; without an end-of-turn (trailing silence) the
    LAST words aren't flushed and the transcription is truncated. Padding acts as
    that end marker so the full utterance is transcribed."""
    parsed = _wav_pcm(wav_bytes)
    if parsed is None:
        return wav_bytes
    pcm, sr, ch, sw = parsed
    return _build_wav(pcm + b"\x00" * (int(sr * ms / 1000) * ch * sw), sr, ch, sw)


async def _transcribe_kyutai(http: httpx.AsyncClient, audio_wav: bytes) -> str:
    """Transcribe a WAV via the local Kyutai stt-1b-en_fr MLX server (/transcribe).

    Pads trailing silence so Kyutai flushes the full utterance. The server
    resamples internally (Mimi @ 24 kHz), so any input rate is fine.
    Returns the recognised text (possibly empty on silence/inaudible input).
    """
    resp = await http.post(
        f"{settings.kyutai_stt_url.rstrip('/')}/transcribe",
        content=_pad_wav_silence(_normalize_wav_for_stt(audio_wav), 800),
        headers={"Content-Type": "audio/wav"},
        timeout=30.0,
    )
    resp.raise_for_status()
    return (resp.json().get("text") or "").strip()


# whisper.cpp (large-v3-turbo, CoreML) — far more robust to the weak/noisy SLIC
# handset captures than Kyutai's streaming STT (decodes voice Kyutai returns ''
# for). Used as the primary STT for the voice loop.
_WHISPER_BIN = "/Users/electron/whisperx-server/whisper.cpp/build/bin/whisper-cli"
_WHISPER_MODEL = "/Users/electron/whisperx-server/models/ggml-large-v3-turbo-q5_0.bin"
_WHISPER_NOISE = re.compile(r"\[(?:_?BLANK_AUDIO_?|BLANK_AUDIO|silence|musique|music)\]|\(\s*\)", re.I)
# whisper hallucinates these training-data artefacts on silence/noise — treat as empty.
_WHISPER_HALLUC = re.compile(
    r"sous-titr|société radio-canada|radio-canada|merci d'avoir regard|"
    r"amara\.org|abonnez-vous|tipeur|souscripteur|merci à tous|"
    r"n'oubliez pas de vous abonner|^\s*\.+\s*$",
    re.I)


def _strip_whisper_artefacts(txt: str) -> str:
    """Drop whisper's silence/noise hallucinations so the NPC doesn't 'reply' to them."""
    cleaned = _WHISPER_NOISE.sub("", txt or "").strip()
    if not cleaned or _WHISPER_HALLUC.search(cleaned):
        return ""
    return cleaned


def _transcribe_whisper_sync(audio_wav: bytes) -> str:
    conditioned = _normalize_wav_for_stt(audio_wav)
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        f.write(conditioned)
        path = f.name
    try:
        out = subprocess.run(
            [_WHISPER_BIN, "-m", _WHISPER_MODEL, "-f", path,
             "-l", "fr", "-nt", "-np"],
            capture_output=True, text=True, timeout=30,
        )
        return _strip_whisper_artefacts(out.stdout or "")
    finally:
        try:
            os.unlink(path)
        except OSError:
            pass


async def _transcribe_whisper(audio_wav: bytes) -> str:
    """Transcribe a WAV via whisper.cpp turbo (high-pass conditioned), off-thread."""
    return await asyncio.to_thread(_transcribe_whisper_sync, audio_wav)


def _ascii_header(s: str) -> str:
    """Make a string safe as an HTTP header value: ASCII-only AND no control
    characters (newlines/CR/tab break the HTTP framing)."""
    ascii_s = s.encode("ascii", errors="replace").decode("ascii")
    return "".join(ch if 0x20 <= ord(ch) < 0x7F else " " for ch in ascii_s).strip()


class VoiceTurnRequest(BaseModel):
    session_id: str
    number: str
    kind: str = "reply"


class VoiceEndRequest(BaseModel):
    session_id: str


@app.post("/v1/voice/turn")
async def voice_turn(body: VoiceTurnRequest, request: Request, _: None = Depends(require_token)) -> Response:
    entry = PHONE_DIRECTORY.get(body.number)
    if entry is None:
        raise HTTPException(404, f"unknown phone number '{body.number}'")

    http: httpx.AsyncClient = request.app.state.http
    sid = body.session_id
    persona = entry.get("persona", "")
    heard = ""

    try:
        if body.kind == "greeting":
            said = entry.get("greeting") or await _chat_reply(http, persona, [])
            # Prime granite's KV cache with this persona NOW (background), so the
            # player's first spoken reply hits a warm cache (~0.5 s) instead of
            # the ~17 s cold prompt-processing that was timing out.
            asyncio.create_task(_warm_persona(http, persona))
        else:
            # Text-only reply (no audio input — heard is always empty here).
            # The transcription-driven path lives in POST /v1/voice/reply.
            VOICE_SESSIONS.append(sid, "user", heard or "(inaudible)")
            history = VOICE_SESSIONS.history(sid)
            said = await _chat_reply(http, persona, history)

        VOICE_SESSIONS.append(sid, "assistant", said)
        wav16, _, _ = await _voice_tts_16k(http, said, entry.get("voice", settings.ailiance_tts_voice))
    except HTTPException:
        raise
    except Exception as exc:
        raise HTTPException(502, f"voice backend unreachable: {str(exc)[:120]}") from exc

    return Response(
        content=wav16,
        media_type="audio/wav",
        headers={
            "X-Zacus-Heard": _ascii_header(heard[:200]),
            "X-Zacus-Said": _ascii_header(said[:200]),
        },
    )


@app.post("/v1/voice/reply")
async def voice_reply(
    request: Request,
    session_id: str = Form(...),
    number: str = Form(...),
    audio: UploadFile = File(...),
    _: None = Depends(require_token),
) -> Response:
    """Stage 3 conversational turn: the player's recorded speech drives the reply.

    Multipart form: session_id, number, audio (WAV).  The audio is transcribed
    by the local Kyutai STT, appended to the session as the user turn, the NPC
    persona LLM produces a reply, and the reply is synthesised to a 16 kHz WAV.
    X-Zacus-Heard carries the transcription so the firmware/dashboard can log it.
    """
    entry = PHONE_DIRECTORY.get(number)
    if entry is None:
        raise HTTPException(404, f"unknown phone number '{number}'")

    http: httpx.AsyncClient = request.app.state.http
    persona = entry.get("persona", "")

    try:
        audio_bytes = await audio.read()
        heard = await _transcribe_whisper(audio_bytes)
        logging.info("voice/reply[%s] HEARD: %r", number, heard)
        VOICE_SESSIONS.append(session_id, "user", heard or "(inaudible)")
        history = VOICE_SESSIONS.history(session_id)
        said = await _chat_reply(http, persona, history)
        logging.info("voice/reply[%s] SAID:  %r", number, said)
        VOICE_SESSIONS.append(session_id, "assistant", said)
        wav16, _, _ = await _voice_tts_16k(http, said, entry.get("voice", settings.ailiance_tts_voice))
    except HTTPException:
        raise
    except Exception as exc:
        raise HTTPException(502, f"voice backend unreachable: {str(exc)[:120]}") from exc

    return Response(
        content=wav16,
        media_type="audio/wav",
        headers={
            "X-Zacus-Heard": _ascii_header(heard[:200]),
            "X-Zacus-Said": _ascii_header(said[:200]),
        },
    )


@app.post("/v1/voice/end")
async def voice_end(body: VoiceEndRequest, _: None = Depends(require_token)) -> dict:
    VOICE_SESSIONS.end(body.session_id)
    return {"ok": True}