"""SCENEGOD server — page + assets + scenes + render pipeline.

Single-file FastAPI app (MESHGOD convention). A plain folder tree is the asset
database (no index file to drift): assets/{characters,animations,props,
backdrops,audio}/<pack>/name.ext. Files sharing dir+stem = one entry, many
formats. See PLAN.md §4.3 for the HTTP contract.

    uvicorn scenegod.server:app --port 8020      # binds 127.0.0.1
"""
import array
import base64
import json
import math
import operator
import os
import re
import shutil
import subprocess
import threading
import time
import urllib.error
import urllib.request
import uuid
from pathlib import Path

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse

ROOT = Path(__file__).resolve().parent.parent
WEB = Path(__file__).resolve().parent / "web"   # scenegod/web (PLAN §2)
ASSETS = Path(os.environ.get("SCENEGOD_ASSETS", ROOT / "assets")).resolve()
SCENES = Path(os.environ.get("SCENEGOD_SCENES", ROOT / "scenes")).resolve()
RENDERS = Path(os.environ.get("SCENEGOD_RENDERS", ROOT / "renders")).resolve()
TEMPLATES = Path(os.environ.get("SCENEGOD_TEMPLATES", ROOT / "templates")).resolve()
SEQUENCES = Path(os.environ.get("SCENEGOD_SEQUENCES", ROOT / "sequences")).resolve()
FILMS = Path(os.environ.get("SCENEGOD_FILMS", ROOT / "films")).resolve()

MODEL_EXTS = {"glb", "gltf", "fbx", "obj", "bvh"}
IMG_EXTS = {"jpg", "jpeg", "png", "webp"}
AUDIO_EXTS = {"wav", "mp3"}
VIDEO_EXTS = {"mp4", "webm", "mov"}          # M6: Flow video plates
ASSET_EXTS = MODEL_EXTS | IMG_EXTS | AUDIO_EXTS | VIDEO_EXTS
# primary ext per category (others sharing a stem are thumbnails)
CATEGORIES = {"characters": MODEL_EXTS, "animations": MODEL_EXTS, "props": MODEL_EXTS,
              "backdrops": IMG_EXTS | VIDEO_EXTS, "audio": AUDIO_EXTS}
MEDIA_TYPES = {
    "glb": "model/gltf-binary", "gltf": "model/gltf+json", "fbx": "application/octet-stream",
    "obj": "text/plain", "bvh": "text/plain", "jpg": "image/jpeg", "jpeg": "image/jpeg",
    "png": "image/png", "webp": "image/webp", "wav": "audio/wav", "mp3": "audio/mpeg",
    "mp4": "video/mp4", "webm": "video/webm", "mov": "video/quicktime",
}

SCENES.mkdir(parents=True, exist_ok=True)
RENDERS.mkdir(parents=True, exist_ok=True)
SEQUENCES.mkdir(parents=True, exist_ok=True)
FILMS.mkdir(parents=True, exist_ok=True)
HAVE_FFMPEG = shutil.which("ffmpeg") is not None
HAVE_RHUBARB = shutil.which("rhubarb") is not None

# MODELBEAST proxy (M4) — off unless SCENEGOD_MB=1. Token read from disk at
# request time, never logged. Contract mirrors MESHGOD meshgod/ops.py.
MB_HOST = os.environ.get("MB_HOST", "http://100.89.131.57:8777")
MB_ENV = Path(os.environ.get("MB_ENV", "~/Documents/backnforth/.env")).expanduser()


def _mb_token() -> str:
    if os.environ.get("MB_TOKEN"):
        return os.environ["MB_TOKEN"]
    try:
        for line in MB_ENV.read_text().splitlines():
            if line.startswith("MB_TOKEN"):
                return line.split("=", 1)[1].strip().strip('"').strip("'")
    except OSError:
        pass
    raise HTTPException(503, f"no MB token (set MB_TOKEN or provide {MB_ENV})")


def reap_dirs(root: Path, max_age_s=48 * 3600):
    """Delete job dirs under root older than max_age_s. Called at boot (C5/M8)."""
    now = time.time()
    for d in root.iterdir() if root.is_dir() else []:
        try:
            if d.is_dir() and now - d.stat().st_mtime > max_age_s:
                shutil.rmtree(d, ignore_errors=True)
        except OSError:
            pass


reap_dirs(RENDERS)
reap_dirs(FILMS)
app = FastAPI(title="SCENEGOD")


# ---- path guard (ship-check rule: every path-taking endpoint uses this) ----
def safe_under(root: Path, relpath: str) -> Path:
    """Resolve relpath under root; reject traversal, absolute, symlink escape."""
    full = (root / relpath).resolve()
    if full != root and root not in full.parents:
        raise HTTPException(400, "bad path")
    return full


def _num(v, default=None):
    """Strict number coercion: None -> default, wrong type -> None (a violation)."""
    if v is None:
        return default
    if isinstance(v, bool) or not isinstance(v, (int, float)):
        return None
    return float(v)


def resolve_audio(a: dict) -> tuple[dict | None, str | None]:
    """One audio entry (scene `audio[]` / sequence bed) -> a muxable dict, or an
    error string. Shared by /render/end and plan_sequence so a clip means the
    same thing in a shot and in a film. `in`/`out` are SOURCE-relative seconds
    (Lane B's audio trim); `start` is where the trimmed piece lands on the
    timeline. Path-guarded like every other path-taking input."""
    if not isinstance(a, dict) or not isinstance(a.get("path"), str):
        return None, "needs a string path"
    try:
        f = safe_under(ASSETS, a["path"])
    except HTTPException:
        return None, f"bad path {a['path']!r}"
    if not f.is_file():
        return None, f"not found under assets: {a['path']!r}"
    start, gain = _num(a.get("start"), 0.0), _num(a.get("gain"), 1.0)
    t_in, t_out = _num(a.get("in"), 0.0), _num(a.get("out"))
    if start is None or gain is None or t_in is None or \
            (a.get("out") is not None and t_out is None):
        return None, "start/gain/in/out must be numbers"
    if start < 0 or t_in < 0:
        return None, f"start/in must be >= 0 (start={start}, in={t_in})"
    if t_out is not None and t_out <= t_in:
        return None, f"in ({t_in}) must be less than out ({t_out})"
    e = {"path": a["path"], "file": str(f), "start": start, "gain": gain}
    if t_in or t_out is not None:
        # a window entirely past the end of the file makes ffmpeg exit -22 mid-mux
        # ("Error muxing a packet"), which surfaces as a mystery failed render.
        src = _probe_dur(f)                     # only probed when a trim is asked for
        if src:
            if t_in >= src:
                return None, f"in ({t_in}) is past the end of {a['path']!r} ({src:.3f}s)"
            if t_out is not None and t_out > src:
                t_out = None                    # tail trim past EOF = no tail trim
    if t_in:
        e["in"] = t_in
    if t_out is not None:
        e["out"] = t_out
    return e, None


# ---- assets ----
_tree_cache = {"t": 0.0, "val": None}


def scan_assets() -> dict:
    """Walk each category dir, group files by (dir, stem) → one entry w/ formats.
    Live scan (hundreds of files); cached 5s in /assets/tree."""
    out = {c: [] for c in CATEGORIES}
    for cat, primary in CATEGORIES.items():
        base = ASSETS / cat
        if not base.is_dir():
            continue
        groups: dict[str, dict] = {}
        for dirpath, dirs, files in os.walk(base):
            dirs[:] = [d for d in dirs if not d.startswith(".")]
            for f in sorted(files):
                stem, dot, ext = f.rpartition(".")
                ext = ext.lower()
                if f.startswith(".") or not dot or ext not in ASSET_EXTS:
                    continue
                rel = os.path.relpath(os.path.join(dirpath, f), ASSETS).replace(os.sep, "/")
                key = os.path.relpath(os.path.join(dirpath, stem), ASSETS)  # dir+stem
                g = groups.setdefault(key, {"name": stem, "path": None, "formats": [], "thumb": None})
                if ext not in primary:
                    if ext in IMG_EXTS:
                        g["thumb"] = rel
                    continue
                g["formats"].append(ext)
                # prefer glb (models) / video (backdrops: mp4 beats its jpg poster), else first
                if g["path"] is None or ext == "glb" or ext in VIDEO_EXTS:
                    g["path"] = rel
                if ext in IMG_EXTS and g["thumb"] is None:  # primary img doubles as poster/thumb
                    g["thumb"] = rel
        entries = [g for g in groups.values() if g["formats"]]  # thumb-only stems aren't assets
        out[cat] = sorted(entries, key=lambda g: g["name"].lower())
    return out


@app.get("/assets/tree")
def assets_tree():
    if not ASSETS.is_dir():
        raise HTTPException(500, f"asset root missing: {ASSETS} (set SCENEGOD_ASSETS)")
    now = time.monotonic()
    if _tree_cache["val"] is None or now - _tree_cache["t"] > 5.0:
        _tree_cache["val"] = scan_assets()
        _tree_cache["t"] = now
    return _tree_cache["val"]


@app.get("/assets/file")
def assets_file(path: str):
    full = safe_under(ASSETS, path)
    if not full.is_file():
        raise HTTPException(404, "not found")
    ext = full.suffix.lstrip(".").lower()
    return FileResponse(full, media_type=MEDIA_TYPES.get(ext, "application/octet-stream"))


@app.get("/assets/thumb")
def assets_thumb(path: str):
    """Sidecar thumbnail for an asset path (same dir+stem, any image ext), 404
    if none. Serves the file as-is — no resize (would need a Pillow dep).
    ponytail: pass-through; add downscaling if dock load is measurably slow."""
    full = safe_under(ASSETS, path)
    for ext in IMG_EXTS:
        thumb = full.with_suffix("." + ext)
        if thumb.is_file():
            return FileResponse(thumb, media_type=MEDIA_TYPES.get(ext),
                                headers={"Cache-Control": "max-age=3600"})
    raise HTTPException(404, "no thumbnail")


# ---- scenes ----
SLUG = re.compile(r"[^a-z0-9_-]")


def slugify(name: str) -> str:
    s = SLUG.sub("-", name.lower()).strip("-")
    if not s:
        raise HTTPException(422, "empty scene name after slugify")
    return s


def validate_scene(s: dict) -> list[str]:
    """Return list of violations (empty = valid). Stdlib only, no jsonschema."""
    errs = []
    if s.get("version") != 1:
        errs.append(f"version must be 1, got {s.get('version')!r}")
    ents = s.get("entities")
    if not isinstance(ents, list):
        return errs + ["entities must be a list"]
    ids, cam_ids = set(), set()
    for i, e in enumerate(ents):
        eid = e.get("id")
        if eid in ids:
            errs.append(f"entity[{i}]: duplicate id {eid!r}")
        ids.add(eid)
        if e.get("kind") == "camera":
            cam_ids.add(eid)
        tracks = e.get("tracks") or {}
        for tk in ("transform", "params"):
            keys = tracks.get(tk) or []
            ts = [k.get("t") for k in keys]
            if ts != sorted(ts):
                errs.append(f"entity {eid!r}: track {tk} not sorted by t")
        clips = sorted(tracks.get("clips") or [], key=lambda c: c.get("start", 0))
        for a, b in zip(clips, clips[1:]):
            end = a.get("start", 0) + (a.get("out", 0) - a.get("in", 0)) * a.get("loop", 1)
            if b.get("start", 0) < end - a.get("fade", 0) - 1e-6:
                errs.append(f"entity {eid!r}: clips overlap beyond fade "
                            f"(block ends {end:.3f}, next starts {b.get('start')})")
    for cut in s.get("cameraCuts") or []:
        if cut.get("camera") not in cam_ids:
            errs.append(f"cameraCut references non-camera entity {cut.get('camera')!r}")
    return errs


@app.get("/scenes")
def scenes_list():
    out = []
    for p in sorted(SCENES.glob("*.json")):
        try:
            dur = json.loads(p.read_text()).get("duration")
        except Exception:
            dur = None
        out.append({"name": p.stem, "mtime": int(p.stat().st_mtime), "duration": dur})
    return out


@app.get("/scenes/{name}")
def scene_get(name: str):
    p = safe_under(SCENES, slugify(name) + ".json")
    if not p.is_file():
        raise HTTPException(404, "not found")
    return json.loads(p.read_text())



# ship-check: body caps — this server now fronts digalot.fyi behind nginx
# basic-auth; never trust content-length alone, check the read bytes too.
async def capped_body(request: Request, limit: int) -> bytes:
    cl = request.headers.get("content-length")
    if cl and cl.isdigit() and int(cl) > limit:
        raise HTTPException(413, f"body over {limit} bytes")
    body = await request.body()
    if len(body) > limit:
        raise HTTPException(413, f"body over {limit} bytes")
    return body

@app.post("/scenes/{name}")
async def scene_save(name: str, request: Request):
    body = await capped_body(request, 5 * 1024 * 1024)
    try:
        scene = json.loads(body)
    except json.JSONDecodeError as e:
        raise HTTPException(400, f"invalid JSON: {e}")
    errs = validate_scene(scene)
    if errs:
        return JSONResponse({"errors": errs}, status_code=422)
    p = safe_under(SCENES, slugify(name) + ".json")
    tmp = p.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(scene, indent=2))
    tmp.replace(p)  # atomic
    return {"ok": True, "name": p.stem}


# ---- starter templates (M7-C) ----
# `templates/` ships WITH the repo (not gitignored like scenes/), so the app has
# something to open on instead of an empty grid. READ-ONLY on purpose: there is no
# POST/DELETE here — "new from template" is GET /templates/{name} followed by an
# ordinary POST /scenes/{name}. Presentation metadata rides in a `template` key
# (title/description/thumb/order) that validate_scene ignores and that never
# survives a Timeline round-trip, so saved scenes stay plain scene JSON.
def _template_meta(p: Path) -> dict | None:
    try:
        t = (json.loads(p.read_text()).get("template") or {})
    except (OSError, json.JSONDecodeError):
        return None                                  # a broken file is not a template
    m = {"name": p.stem, "title": t.get("title") or p.stem,
         "description": t.get("description") or "", "_order": t.get("order", 100)}
    if t.get("thumb"):
        m["thumb"] = t["thumb"]                      # asset-relative → GET /assets/file?path=
    return m


@app.get("/templates")
def templates_list():
    if not TEMPLATES.is_dir():
        return []
    metas = [m for m in (_template_meta(p) for p in TEMPLATES.glob("*.json")) if m]
    metas.sort(key=lambda m: (m["_order"], m["name"]))   # the wow one first, not alphabetical
    return [{k: v for k, v in m.items() if k != "_order"} for m in metas]


@app.get("/templates/{name}")
def template_get(name: str):
    p = safe_under(TEMPLATES, slugify(name) + ".json")   # same guard discipline as /scenes
    if not p.is_file():
        raise HTTPException(404, "not found")
    return json.loads(p.read_text())


# ---- render (C5) ----
# ponytail: in-memory state, single-process local tool. A server restart
# mid-encode loses live status; status() falls back to on-disk out.mp4.
_renders: dict[str, dict] = {}   # rid -> {state: queued|encoding|done|error, log: str}


def _render_dir(rid: str) -> Path:
    return safe_under(RENDERS, rid)   # rid is a hex uuid; guard anyway


def audio_input(a: dict) -> list[str]:
    """ffmpeg input args for one audio entry, honouring the SOURCE-relative
    `in`/`out` window (Lane B's audio trim; absent = whole file).

    The trim rides on the INPUT (-ss/-t) rather than an `atrim` in the filter
    graph, which is not a style choice: `atrim,...,apad` yields a zero-sample
    audio stream on ffmpeg 8.x (apad emits frames at ~INT64_MAX PTS, the muxer
    drops the lot — an mp4 with an aac stream and no audio in it). Input seeking
    keeps the graph byte-identical to the chain that has been shipping."""
    pre = []
    if a.get("in"):
        pre += ["-ss", f"{a['in']:.6f}"]
    if a.get("out") is not None:
        pre += ["-t", f"{a['out'] - (a.get('in') or 0.0):.6f}"]
    return pre + ["-i", a["file"]]


def audio_filters(audio: list[dict], first_input: int) -> list[str]:
    """Filter-graph fragments for the audio inputs, ending in [aout]: delay each
    track to its timeline `start`, apply gain, mix down to one stereo stream."""
    if not audio:
        return []
    parts = [f"[{first_input + k}:a]adelay={int(a['start']*1000)}:all=1,volume={a['gain']}[a{k}]"
             for k, a in enumerate(audio)]
    # apad + -shortest: the VIDEO decides the length. Without the pad, audio
    # shorter than the shot truncates the picture (silent footage loss).
    parts.append("".join(f"[a{k}]" for k in range(len(audio)))
                 + f"amix=inputs={len(audio)}:normalize=0,apad[aout]")
    return parts


def _encode(rid: str, fps: int, audio: list[dict]):
    """audio = [{file: abs path, start: sec, gain: float, in?, out?}] (resolved)."""
    d = _render_dir(rid)
    _renders[rid] = {"state": "encoding", "log": ""}
    cmd = ["ffmpeg", "-y", "-framerate", str(fps), "-i", str(d / "frames" / "%06d.png")]
    for a in audio:
        cmd += audio_input(a)                  # in/out trim rides on the input (M9)
    cmd += ["-c:v", "libx264", "-pix_fmt", "yuv420p", "-crf", "18"]
    if audio:  # delay to start, apply gain, mix down to one stereo stream
        cmd += ["-filter_complex", ";".join(audio_filters(audio, 1)),
                "-map", "0:v", "-map", "[aout]", "-c:a", "aac", "-shortest"]
    cmd += [str(d / "out.mp4")]
    try:
        p = subprocess.run(cmd, capture_output=True, text=True)
        ok = p.returncode == 0 and (d / "out.mp4").is_file()
        _renders[rid] = {"state": "done" if ok else "error", "log": p.stderr[-2000:]}
    except Exception as e:  # noqa: BLE001 — surface any spawn failure as error state
        _renders[rid] = {"state": "error", "log": str(e)}


@app.post("/render/begin")
async def render_begin(request: Request):
    if not HAVE_FFMPEG:
        raise HTTPException(503, "ffmpeg not found on server")
    rid = uuid.uuid4().hex
    (RENDERS / rid / "frames").mkdir(parents=True)
    try:
        meta = json.loads(await capped_body(request, 64 * 1024) or b"{}")
    except json.JSONDecodeError:
        meta = {}
    if not (1 <= int(meta.get("fps", 30)) <= 60) or \
       not (16 <= int(meta.get("width", 1280)) <= 4096) or \
       not (16 <= int(meta.get("height", 720)) <= 4096):
        raise HTTPException(400, "fps/width/height out of range")
    (RENDERS / rid / "meta.json").write_text(json.dumps(meta))
    _renders[rid] = {"state": "queued", "log": ""}
    return {"renderId": rid}


@app.post("/render/{rid}/frame/{n}")
async def render_frame(rid: str, n: int, request: Request):
    frames = _render_dir(rid) / "frames"
    if not frames.is_dir():
        raise HTTPException(404, "unknown renderId")
    if not (0 <= n <= 20000):
        raise HTTPException(400, "frame index out of range")
    body = await capped_body(request, 25 * 1024 * 1024)
    if not body:
        raise HTTPException(400, "empty frame body")
    (frames / f"{n:06d}.png").write_bytes(body)
    return {"ok": True}


@app.post("/render/{rid}/end")
async def render_end(rid: str, request: Request):
    d = _render_dir(rid)
    if not (d / "frames").is_dir():
        raise HTTPException(404, "unknown renderId")
    if not any((d / "frames").glob("*.png")):
        raise HTTPException(400, "no frames posted")
    try:
        fps = int(json.loads((d / "meta.json").read_text()).get("fps", 30))
    except (OSError, json.JSONDecodeError):
        fps = 30
    try:
        body = json.loads(await capped_body(request, 1024 * 1024) or b"{}")
    except json.JSONDecodeError:
        body = {}
    audio = []
    for a in body.get("audio") or []:              # scene JSON audio[] (M4, +in/out M9)
        e, err = resolve_audio(a if isinstance(a, dict) else {})
        if err:
            raise HTTPException(400, f"audio: {err}")
        audio.append(e)
    threading.Thread(target=_encode, args=(rid, fps, audio), daemon=True).start()
    return {"ok": True}


@app.get("/render/{rid}/status")
def render_status(rid: str):
    st = _renders.get(rid)
    if st:
        return st
    d = _render_dir(rid)          # fell out of memory (restart) — infer from disk
    if (d / "out.mp4").is_file():
        return {"state": "done", "log": ""}
    if d.is_dir():
        return {"state": "queued", "log": ""}
    raise HTTPException(404, "unknown renderId")


@app.get("/render/{rid}/out.mp4")
def render_out(rid: str):
    out = _render_dir(rid) / "out.mp4"
    if not out.is_file():
        raise HTTPException(404, "not ready")
    return FileResponse(out, media_type="video/mp4")


# ---- sequences: multi-shot films (M8-C) ----
# A scene is ONE take. A sequence is the film-level object: an ordered list of
# shots, each a trimmed window [in,out) of a saved scene, joined by cuts or
# dissolves, over an optional film-level audio bed.
#
# The server CANNOT render a shot — three.js lives in the browser. So a film is a
# client-driven job the server orchestrates: POST /sequences/{n}/render hands back
# the resolved shot plan, the client renders each shot through the ORDINARY
# /render/* session endpoints and reports which renderId belongs to which shot
# index, and /films/{id}/end concatenates those mp4s with ffmpeg. See web/film.js.
SEQ_NAME = re.compile(r"[a-z0-9_-]{1,64}\Z")     # same alphabet slugify() emits
TRANSITIONS = {"cut", "dissolve"}
MAX_SHOTS = 200
MAX_TRANSITION_S = 10.0


def plan_sequence(seq: dict) -> tuple[list[str], dict]:
    """Validate a sequence AND resolve it to a concrete shot plan in one pass, so
    save-time validation and render-time planning can never disagree about what a
    sequence means. Returns (violations, plan); plan is meaningful only when the
    violation list is empty."""
    errs: list[str] = []
    warnings: list[str] = []
    plan = {"fps": 30, "shots": [], "audio": [], "duration": 0.0, "warnings": warnings}
    if seq.get("version") != 1:
        errs.append(f"version must be 1, got {seq.get('version')!r}")
    shots = seq.get("shots")
    if not isinstance(shots, list) or not shots:
        errs.append("shots must be a non-empty list")
        shots = []
    if len(shots) > MAX_SHOTS:
        errs.append(f"too many shots ({len(shots)} > {MAX_SHOTS})")
        shots = shots[:MAX_SHOTS]

    fps_seen: dict[int, str] = {}
    has_bed = bool(seq.get("audio"))
    for i, sh in enumerate(shots):
        if not isinstance(sh, dict):
            errs.append(f"shot[{i}]: must be an object")
            continue
        nm = sh.get("scene")
        if not isinstance(nm, str) or not SEQ_NAME.match(nm):
            errs.append(f"shot[{i}]: bad scene name {nm!r}")
            continue
        p = SCENES / (nm + ".json")            # name is [a-z0-9_-] only: no traversal possible
        if not p.is_file():
            errs.append(f"shot[{i}]: unknown scene {nm!r}")
            continue
        try:
            sc = json.loads(p.read_text())
        except (OSError, json.JSONDecodeError) as e:
            errs.append(f"shot[{i}]: scene {nm!r} unreadable ({e.__class__.__name__})")
            continue
        sdur = _num(sc.get("duration"), 0.0) or 0.0
        fps_seen.setdefault(int(_num(sc.get("fps"), 30) or 30), nm)
        if (sc.get("audio") or []) and not has_bed:
            warnings.append(f"scene {nm!r} has audio[] — shot audio is NOT carried into the "
                            f"film; put the soundtrack in the sequence's own audio[]")

        t_in, t_out = _num(sh.get("in"), 0.0), _num(sh.get("out"), sdur)
        if t_in is None or t_out is None:
            errs.append(f"shot[{i}]: in/out must be numbers")
            continue
        if t_in < 0:
            errs.append(f"shot[{i}]: in must be >= 0, got {t_in}")
        if t_out <= t_in:
            errs.append(f"shot[{i}]: in ({t_in}) must be less than out ({t_out})")
        if t_out > sdur + 1e-6:
            errs.append(f"shot[{i}]: out ({t_out}) is past the end of scene {nm!r} ({sdur})")

        tr, td = sh.get("transition", "cut"), _num(sh.get("transitionDur"), 0.5)
        if tr not in TRANSITIONS:
            errs.append(f"shot[{i}]: unknown transition {tr!r} (one of {sorted(TRANSITIONS)})")
        elif td is None or td < 0 or td > MAX_TRANSITION_S:
            errs.append(f"shot[{i}]: transitionDur must be 0..{MAX_TRANSITION_S}s, "
                        f"got {sh.get('transitionDur')!r}")
        elif tr == "dissolve" and td <= 0:
            errs.append(f"shot[{i}]: a dissolve needs transitionDur > 0")
        plan["shots"].append({"index": i, "scene": nm, "in": t_in, "out": t_out,
                              "len": max(0.0, t_out - t_in),
                              "transition": tr if tr in TRANSITIONS else "cut",
                              "transitionDur": (td or 0.0) if tr == "dissolve" else 0.0})

    # a dissolve eats the tail of this shot and the head of the next: ffmpeg xfade
    # needs it shorter than BOTH, or the fold produces garbage instead of a film.
    ps = plan["shots"]
    for k, s in enumerate(ps):
        if s["transition"] != "dissolve" or k + 1 >= len(ps):
            continue
        nxt = ps[k + 1]
        if s["transitionDur"] >= min(s["len"], nxt["len"]) - 1e-6:
            errs.append(f"shot[{s['index']}]: dissolve {s['transitionDur']}s must be shorter than "
                        f"both shots it joins ({s['len']:.3f}s, {nxt['len']:.3f}s)")
    if len(fps_seen) > 1:
        errs.append("shots mix scene fps ("
                    + ", ".join(f"{f} in {n!r}" for f, n in sorted(fps_seen.items()))
                    + ") — one film, one fps")
    if fps_seen:
        plan["fps"] = min(fps_seen)

    aud = seq.get("audio") or []
    if not isinstance(aud, list):
        errs.append("audio must be a list")
        aud = []
    for k, a in enumerate(aud):
        e, err = resolve_audio(a)
        if err:
            errs.append(f"audio[{k}]: {err}")
        else:
            plan["audio"].append(e)

    plan["duration"] = round(sum(s["len"] for s in ps) - sum(s["transitionDur"] for s in ps[:-1]), 6)
    plan["warnings"] = list(dict.fromkeys(warnings))              # dedupe, keep order
    return errs, plan


def validate_sequence(seq: dict) -> list[str]:
    return plan_sequence(seq)[0]


def _seq_path(name: str) -> Path:
    return safe_under(SEQUENCES, slugify(name) + ".json")


@app.get("/sequences")
def sequences_list():
    out = []
    for p in sorted(SEQUENCES.glob("*.json")):
        try:
            seq = json.loads(p.read_text())
        except (OSError, json.JSONDecodeError):
            continue
        errs, plan = plan_sequence(seq)
        out.append({"name": p.stem, "title": seq.get("title") or p.stem,
                    "mtime": int(p.stat().st_mtime), "shots": len(seq.get("shots") or []),
                    "duration": None if errs else plan["duration"], "ok": not errs})
    return out


@app.get("/sequences/{name}")
def sequence_get(name: str):
    p = _seq_path(name)
    if not p.is_file():
        raise HTTPException(404, "not found")
    return json.loads(p.read_text())


@app.post("/sequences/{name}")
async def sequence_save(name: str, request: Request):
    body = await capped_body(request, 1024 * 1024)
    try:
        seq = json.loads(body)
    except json.JSONDecodeError as e:
        raise HTTPException(400, f"invalid JSON: {e}")
    errs, plan = plan_sequence(seq)
    if errs:
        return JSONResponse({"errors": errs}, status_code=422)
    p = _seq_path(name)
    tmp = p.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(seq, indent=2))
    tmp.replace(p)  # atomic
    return {"ok": True, "name": p.stem, "duration": plan["duration"],
            "shots": len(plan["shots"]), "warnings": plan["warnings"]}


@app.delete("/sequences/{name}")
def sequence_delete(name: str):
    p = _seq_path(name)
    if not p.is_file():
        raise HTTPException(404, "not found")
    p.unlink()
    return {"ok": True, "name": p.stem}


# ---- film render jobs (M8-C) ----
_films: dict[str, dict] = {}      # fid -> {state: queued|encoding|done|error, log} (encode only)


def _film_dir(fid: str) -> Path:
    return safe_under(FILMS, fid)


def _film_meta(fid: str) -> dict:
    """Plan + shot registrations live on disk: a film spans minutes of client
    rendering, and /end must still work if the process was restarted underneath."""
    p = _film_dir(fid) / "meta.json"
    if not p.is_file():
        raise HTTPException(404, "unknown filmId")
    return json.loads(p.read_text())


def _film_write(fid: str, meta: dict):
    p = _film_dir(fid) / "meta.json"
    tmp = p.with_suffix(".json.tmp")
    tmp.write_text(json.dumps(meta))
    tmp.replace(p)


def _probe_dur(f: Path) -> float | None:
    """Actual mp4 duration — xfade offsets are absolute, so a half-frame of drift
    between nominal and encoded length shows up as a stutter at the join."""
    try:
        out = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration",
                              "-of", "default=nw=1:nokey=1", str(f)],
                             capture_output=True, text=True, timeout=30)
        return float(out.stdout.strip())
    except (OSError, ValueError, subprocess.SubprocessError):
        return None


def _concat_film(fid: str):
    """Join the shot mp4s in order: cut = concat filter, dissolve = xfade. The
    concat FILTER (not the demuxer) because a film may need xfade anyway, and it
    re-encodes once instead of twice."""
    meta = json.loads((_film_dir(fid) / "meta.json").read_text())
    _films[fid] = {"state": "encoding", "log": ""}
    d, shots = _film_dir(fid), meta["shots"]
    w, h, fps = meta["width"], meta["height"], meta["fps"]
    srcs = [RENDERS / s["renderId"] / "out.mp4" for s in shots]
    missing = [str(f) for f in srcs if not f.is_file()]
    if missing:
        _films[fid] = {"state": "error", "log": "shot mp4 missing (reaped?): " + ", ".join(missing)}
        return
    durs = [_probe_dur(f) or s["len"] for f, s in zip(srcs, shots)]

    cmd = ["ffmpeg", "-y"] + [x for f in srcs for x in ("-i", str(f))]
    cmd += [x for a in meta["audio"] for x in audio_input(a)]     # in/out trim (M9)
    # normalise every shot to the film's grid — mismatches are rejected at
    # registration, this is the belt to that braces (SAR/fps oddities included).
    parts = [f"[{i}:v]scale={w}:{h},setsar=1,fps={fps},format=yuv420p[v{i}]" for i in range(len(srcs))]
    cur, acc = "v0", durs[0]
    for i in range(1, len(srcs)):
        prev = shots[i - 1]
        td = prev["transitionDur"] if prev["transition"] == "dissolve" else 0.0
        if td > 0:
            parts.append(f"[{cur}][v{i}]xfade=transition=fade:duration={td}"
                         f":offset={max(0.0, acc - td):.6f}[x{i}]")
            acc += durs[i] - td
        else:
            parts.append(f"[{cur}][v{i}]concat=n=2:v=1:a=0[x{i}]")
            acc += durs[i]
        cur = f"x{i}"
    parts += audio_filters(meta["audio"], len(srcs))   # delay/gain/mix the film-level bed
    cmd += ["-filter_complex", ";".join(parts), "-map", f"[{cur}]"]
    if meta["audio"]:
        cmd += ["-map", "[aout]", "-c:a", "aac", "-shortest"]   # apad+shortest: video sets the length
    cmd += ["-c:v", "libx264", "-pix_fmt", "yuv420p", "-crf", "18", str(d / "out.mp4")]
    try:
        p = subprocess.run(cmd, capture_output=True, text=True)
        ok = p.returncode == 0 and (d / "out.mp4").is_file()
        _films[fid] = {"state": "done" if ok else "error", "log": p.stderr[-2000:]}
    except Exception as e:  # noqa: BLE001 — surface any spawn failure as error state
        _films[fid] = {"state": "error", "log": str(e)}
    meta["state"] = _films[fid]["state"]
    _film_write(fid, meta)


@app.post("/sequences/{name}/render")
async def film_begin(name: str, request: Request):
    """Start a film job. Returns the ordered shot list the CLIENT must render
    (web/film.js) — one /render/* session per shot, reported back per index."""
    if not HAVE_FFMPEG:
        raise HTTPException(503, "ffmpeg not found on server")
    p = _seq_path(name)
    if not p.is_file():
        raise HTTPException(404, "not found")
    errs, plan = plan_sequence(json.loads(p.read_text()))
    if errs:
        return JSONResponse({"errors": errs}, status_code=422)   # a scene may have vanished since save
    try:
        body = json.loads(await capped_body(request, 64 * 1024) or b"{}")
    except json.JSONDecodeError:
        body = {}
    try:
        w, h = int(body.get("width", 1920)), int(body.get("height", 1080))
    except (TypeError, ValueError):
        raise HTTPException(400, "width/height must be integers")
    if not (16 <= w <= 4096 and 16 <= h <= 4096):
        raise HTTPException(400, "width/height out of range")
    fid = uuid.uuid4().hex
    (FILMS / fid).mkdir(parents=True)
    meta = {"sequence": p.stem, "fps": plan["fps"], "width": w, "height": h,
            "audio": plan["audio"], "state": "collecting", "created": time.time(),
            "shots": [dict(s, renderId=None) for s in plan["shots"]]}
    _film_write(fid, meta)
    return {"filmId": fid, "fps": plan["fps"], "width": w, "height": h,
            "duration": plan["duration"], "warnings": plan["warnings"],
            "shots": [{k: s[k] for k in ("index", "scene", "in", "out", "len",
                                         "transition", "transitionDur")} for s in plan["shots"]]}


@app.post("/films/{fid}/shot/{i}")
async def film_shot(fid: str, i: int, request: Request):
    """The client says 'renderId R is shot i'. The render must be finished and cut
    to the film's grid — one resolution and one fps per film, checked here so the
    failure is a clear 400 and not a cryptic ffmpeg filter error 10 minutes later."""
    meta = _film_meta(fid)
    if not (0 <= i < len(meta["shots"])):
        raise HTTPException(400, f"shot index {i} out of range (0..{len(meta['shots'])-1})")
    try:
        rid = str(json.loads(await capped_body(request, 8 * 1024)).get("renderId") or "")
    except json.JSONDecodeError:
        raise HTTPException(400, "invalid JSON")
    if not re.fullmatch(r"[0-9a-f]{8,64}", rid):
        raise HTTPException(400, "bad renderId")
    rd = safe_under(RENDERS, rid)
    if not (rd / "out.mp4").is_file():
        raise HTTPException(400, f"render {rid} has no out.mp4 yet (encode it before registering)")
    try:
        rm = json.loads((rd / "meta.json").read_text())
        rw, rh, rf = int(rm.get("width", 0)), int(rm.get("height", 0)), int(rm.get("fps", 0))
    except (OSError, json.JSONDecodeError, TypeError, ValueError):
        rw = rh = rf = 0                       # no meta (restart/manual) — trust the pixels
    if rw and rh and (rw, rh) != (meta["width"], meta["height"]):
        raise HTTPException(400, f"shot {i} rendered {rw}x{rh}, film is "
                                 f"{meta['width']}x{meta['height']} — one resolution per film")
    if rf and rf != meta["fps"]:
        raise HTTPException(400, f"shot {i} rendered at {rf}fps, film is {meta['fps']}fps")
    meta["shots"][i]["renderId"] = rid
    _film_write(fid, meta)
    done = sum(1 for s in meta["shots"] if s.get("renderId"))
    return {"ok": True, "shot": i, "of": len(meta["shots"]), "registered": done}


@app.post("/films/{fid}/end")
async def film_end(fid: str, request: Request):
    await capped_body(request, 8 * 1024)          # no payload today; cap it anyway (ship-check)
    meta = _film_meta(fid)
    missing = [s["index"] for s in meta["shots"] if not s.get("renderId")]
    if missing:
        raise HTTPException(400, f"shots not registered yet: {missing}")
    meta["state"] = "queued"
    _film_write(fid, meta)
    _films[fid] = {"state": "queued", "log": ""}
    threading.Thread(target=_concat_film, args=(fid,), daemon=True).start()
    return {"ok": True, "shots": len(meta["shots"])}


@app.get("/films/{fid}/status")
def film_status(fid: str):
    meta = _film_meta(fid)
    n = len(meta["shots"])
    done = sum(1 for s in meta["shots"] if s.get("renderId"))
    st = _films.get(fid)
    if not st:      # fell out of memory (restart) — infer from disk
        st = {"state": "done" if (_film_dir(fid) / "out.mp4").is_file()
                       else meta.get("state", "collecting"), "log": ""}
    return {"state": st["state"], "shot": done, "of": n, "log": st["log"]}


@app.get("/films/{fid}/out.mp4")
def film_out(fid: str):
    out = _film_dir(fid) / "out.mp4"
    if not out.is_file():
        raise HTTPException(404, "not ready")
    return FileResponse(out, media_type="video/mp4")


# ---- viseme lip-sync (M4) — feeds Lane A's viseme lane ----
@app.get("/rhubarb")
def rhubarb(path: str):
    """Rhubarb Lip Sync on an asset audio file -> {metadata, mouthCues:[...]}.
    503 if rhubarb absent (same pattern as ffmpeg). ponytail: synchronous —
    VO clips are short; make it a render-style session if that ever bites."""
    if not HAVE_RHUBARB:
        raise HTTPException(503, "rhubarb not installed (see github.com/DanielSWolf/rhubarb-lip-sync)")
    f = safe_under(ASSETS, path)
    if not f.is_file():
        raise HTTPException(404, "not found")
    p = subprocess.run(["rhubarb", "-f", "json", str(f)], capture_output=True, text=True)
    if p.returncode != 0:
        raise HTTPException(500, f"rhubarb failed: {p.stderr[-500:]}")
    return json.loads(p.stdout)


# ---- BEAT DETECTION (M9-C) — the grid a music video cuts itself to ----
# stdlib + ffmpeg only (repo rule: no librosa/aubio/numpy). The chain:
#   ffmpeg      -> mono s16 PCM @ 44100 (anything ffmpeg reads: wav/mp3/mp4/...)
#   framing     -> per-frame log energy, hop 512 = 86.13 envelope fps
#   onset       -> half-wave-rectified first difference   [ENERGY flux]
#   whitening   -> subtract a local mean so a dense mix has a floor near zero
#   tempo       -> normalised autocorrelation 60..180 BPM, harmonics added, then
#                  a log-gaussian prior centred on 125 BPM (John's material is
#                  90s house), then an explicit half/double check
#   phase       -> comb-filter search over one period
#   grid        -> least-squares refit of period+phase to the nearest onset peaks
# ENERGY flux, not spectral flux: a pure-Python radix-2 FFT would be ~20k
# transforms for a 4-minute track (tens of seconds of CPU), and on
# four-to-the-floor material the kick dominates the broadband energy anyway.
# Measured on ultra: 5-minute mp3 = 0.85s wall. 4/4 is ASSUMED — `downbeats` is
# every 4th beat, phase chosen as the strongest of the four cosets. Beats are a
# CONSTANT-tempo grid (no rubato tracking).
#
# DECODE AT 44100, NOT 22050 (M9-C fix — do not "optimise" this back down):
# a percussive hit is a high-frequency transient over a slower low-frequency
# body. Resampling to 22050 throws away everything above 11kHz, and in
# assets/audio/test/beat120.wav the >11k share of the attack grows 20x across
# the file (1402 -> 27711 peak sample) — so for the middle beats the surviving
# <11k energy peaks 30-50ms AFTER the attack while the first and last beats
# still peak at +0ms. A lateness that VARIES per beat is a tempo bias, not an
# offset: the detector read 119.62 on a file whose kicks are on 0.5000s to the
# microsecond, i.e. 0.32% slow = ~0.8s of slip across a 4-minute track. At 44100
# the transient survives and the same code reads 120.03. (172 fps was tried and
# rejected: it breaks octave choice on real records — a 90s house window went to
# 63.7 BPM — because the prior/harmonic weights below are tuned for 86 fps.)
BEAT_SR, BEAT_HOP = 44100, 512
BEAT_FPS = BEAT_SR / BEAT_HOP                 # 86.13 onset frames/s
BEAT_MIN_BPM, BEAT_MAX_BPM = 60.0, 180.0
BEAT_PRIOR_BPM, BEAT_PRIOR_OCT = 125.0, 0.9   # tempo prior: centre, width in octaves
BEAT_PREF = (90.0, 180.0)                     # preferred band for the octave check
BEAT_WHITEN_W = 32                            # local-mean window, +-0.37s
BEAT_MAX_S = 900                              # analyse at most 15 min (memory); grid extends
_beats_cache: dict[tuple, dict] = {}


def _frame_energy(f: Path) -> tuple[list[float], int]:
    """Decode and reduce to per-frame log energy in ONE streaming pass. 15
    minutes at 44.1k is 79MB of PCM; buffering that twice (subprocess buffer +
    array) is not a thing to do inside the prod container, and none of it is
    needed after the framing. stderr stays a pipe for the error message —
    safe only because `-v error` keeps it far under the pipe buffer."""
    p = subprocess.Popen(["ffmpeg", "-v", "error", "-nostdin", "-i", str(f), "-map", "0:a:0",
                          "-t", str(BEAT_MAX_S), "-f", "s16le", "-ac", "1",
                          "-ar", str(BEAT_SR), "-"],
                         stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    e, buf, stride, nbytes = [], b"", BEAT_HOP * 2, 0
    ap = e.append
    while True:
        chunk = p.stdout.read(stride * 4096)          # ~4MB reads
        if not chunk:
            break
        nbytes += len(chunk)
        buf = buf + chunk if buf else chunk
        nfr = len(buf) // stride
        if not nfr:
            continue
        a = array.array("h")
        a.frombytes(buf[:nfr * stride])
        buf = buf[nfr * stride:]
        for i in range(0, len(a), BEAT_HOP):
            fr = a[i:i + BEAT_HOP]
            ap(math.log1p(sum(map(operator.mul, fr, fr)) / BEAT_HOP))
    err = p.stderr.read()
    if p.wait() != 0:
        raise HTTPException(400, "cannot decode audio: "
                            + err[-200:].decode("utf-8", "replace").strip())
    return e, nbytes // 2          # sample count, so `duration` isn't a frame short


def _onset_envelope(e: list[float]) -> list[float]:
    """Per-frame log energy -> rectified first difference -> 3-tap smooth ->
    local-mean subtraction. The whitening is what makes a busy record's peaks
    stand out from its floor (and the confidence number mean anything)."""
    o = [0.0] + [max(0.0, e[i] - e[i - 1]) for i in range(1, len(e))]
    n = len(o)
    o = [(o[max(0, i - 1)] + o[i] + o[min(n - 1, i + 1)]) / 3 for i in range(n)]
    w = BEAT_WHITEN_W
    if n < 2 * w + 2:
        return o
    run, out = sum(o[:w + 1]), []
    for i in range(n):
        out.append(max(0.0, o[i] - run / (min(n - 1, i + w) - max(0, i - w) + 1)))
        if i + 1 + w < n:
            run += o[i + 1 + w]
        if i - w >= 0:
            run -= o[i - w]
    return out


def _beat_tempo(o: list[float]) -> tuple[float, dict]:
    """(period in frames, autocorrelation table). Peak is parabola-interpolated:
    the integer lag is only good to ~0.5 frame, which is seconds of drift across
    a 4-minute track before the least-squares refit cleans it up."""
    n = len(o)
    m = sum(o) / n
    oc = [x - m for x in o]
    e0 = sum(map(operator.mul, oc, oc)) / n or 1e-9
    lo = int(60 * BEAT_FPS / BEAT_MAX_BPM)
    hi = int(math.ceil(60 * BEAT_FPS / BEAT_MIN_BPM))
    r = {}
    for lag in range(lo, hi + 1):
        k = n - lag
        r[lag] = (sum(map(operator.mul, oc[:k], oc[lag:])) / k / e0) if k > 8 else 0.0
    p0 = 60 * BEAT_FPS / BEAT_PRIOR_BPM

    def score(g):
        # harmonics that fall OUTSIDE the lag table must not count as zero — that
        # silently penalises every lag whose 2x is past the table edge (it cost a
        # 118 BPM track 3 BPM: lag 43 kept its 2x at 86, lag 44 lost its 88).
        terms = [(1.0, r[g])] + [(w, r[h]) for w, h in
                                 ((.5, 2 * g), (.5, g // 2), (.3, 3 * g // 2)) if h in r]
        return (sum(w * v for w, v in terms) / sum(w for w, _ in terms)
                * math.exp(-.5 * (math.log2(g / p0) / BEAT_PRIOR_OCT) ** 2))

    best = max(r, key=score)
    # the harmonics and the prior choose WHICH peak; the raw autocorrelation
    # locates it. Walk uphill first: a parabola fitted to a point on a SLOPE
    # extrapolates instead of interpolating (that is how lag 43 became 44.95).
    while best + 1 in r and r[best + 1] > r[best]:
        best += 1
    while best - 1 in r and r[best - 1] > r[best]:
        best -= 1
    a, b, c = r.get(best - 1, 0.), r[best], r.get(best + 1, 0.)
    d = a - 2 * b + c
    return best + max(-0.5, min(0.5, 0.5 * (a - c) / d if d else 0.0)), r


def _comb(o: list[float], period: float, phase: float) -> float:
    """Mean onset strength on a beat grid — the score a phase/period is judged by.
    MEAN (not sum) is deliberate: it makes a double-tempo grid pay for the empty
    positions it adds, which is half the octave problem solved for free."""
    s, cnt, i, n = 0.0, 0, phase, len(o)
    while i < n - 1:
        s += o[int(i + 0.5)]
        cnt += 1
        i += period
    return s / cnt if cnt else 0.0


def _best_phase(o: list[float], period: float) -> tuple[float, int]:
    return max((_comb(o, period, ph), ph) for ph in range(max(1, int(period))))


def _fit_grid(o: list[float], period: float, phase: float, iters: int = 3):
    """Snap each predicted beat to the nearest onset peak (sub-frame, parabolic)
    and refit t_k = phase + k*period by onset-weighted least squares."""
    n = len(o)
    lo_p, hi_p = 60 * BEAT_FPS / BEAT_MAX_BPM, 60 * BEAT_FPS / BEAT_MIN_BPM
    for _ in range(iters):
        win = max(1, int(period * 0.12))
        xs, ys, ws = [], [], []
        k, t = 0, phase
        while t < n - 1:
            c = int(round(t))
            a, b = max(1, c - win), min(n - 2, c + win)
            if b > a:
                j = max(range(a, b + 1), key=o.__getitem__)
                p1, p2, p3 = o[j - 1], o[j], o[j + 1]
                d = p1 - 2 * p2 + p3
                frac = (0.5 * (p1 - p3) / d) if d else 0.0
                if abs(frac) > 1:
                    frac = 0.0
                if p2 > 0:
                    xs.append(k)
                    ys.append(j + frac)
                    ws.append(p2)
            k += 1
            t = phase + k * period
        if len(xs) < 4:
            break
        sw = sum(ws)
        mx = sum(w * x for w, x in zip(ws, xs)) / sw
        my = sum(w * y for w, y in zip(ws, ys)) / sw
        den = sum(w * (x - mx) ** 2 for w, x in zip(ws, xs))
        if den <= 0:
            break
        period = min(max(sum(w * (x - mx) * (y - my)
                             for w, x, y in zip(ws, xs, ys)) / den, lo_p), hi_p)
        phase = my - period * mx
    return period, phase


def analyse_beats(f: Path) -> dict:
    frames, nsamples = _frame_energy(f)
    analyzed = nsamples / BEAT_SR
    truncated = analyzed >= BEAT_MAX_S - 0.5
    full = (_probe_dur(f) or analyzed) if truncated else analyzed
    out = {"bpm": 0.0, "confidence": 0.0, "beats": [], "downbeats": [], "meter": 4,
           "duration": round(full, 3), "analyzed": round(analyzed, 3),
           "period": 0.0, "acPeak": 0.0, "onBeatContrast": 0.0}
    o = _onset_envelope(frames)
    n = len(o)
    lo_p, hi_p = 60 * BEAT_FPS / BEAT_MAX_BPM, 60 * BEAT_FPS / BEAT_MIN_BPM
    if n < 2 * hi_p or max(o, default=0.0) <= 1e-9:
        out["note"] = "no beat grid: silent, or shorter than two slow bars"
        return out
    lag, r = _beat_tempo(o)
    cands = []
    for p in (lag / 2, lag, lag * 2):        # explicit octave check: half, self, double
        if lo_p <= p <= hi_p:
            s, ph = _best_phase(o, p)
            cands.append((s, p, ph))
    top = max(c[0] for c in cands)
    keep = [c for c in cands if c[0] >= 0.85 * top] or cands
    # when half/double score comparably, take the one in the sane band — a 128 BPM
    # record whose onset envelope autocorrelates hardest at 64 is the normal case.
    _, period, phase = max(keep, key=lambda c: (BEAT_PREF[0] <= 60 * BEAT_FPS / c[1]
                                                <= BEAT_PREF[1], c[0]))
    period, phase = _fit_grid(o, period, phase)

    # frame i is the window [i, i+1) hops, so its centre is the honest onset time
    ph = phase + 0.5
    ph -= period * math.floor((ph + 0.05 * BEAT_FPS) / period)   # into [-50ms, period-50ms):
    beats_f = []                                                 # a track starting ON the
    k = 0                                                        # beat gets a beat at ~0
    while ph + k * period <= full * BEAT_FPS:
        beats_f.append(ph + k * period)
        k += 1
    inwin = [(i, int(round(t))) for i, t in enumerate(beats_f) if 0 <= round(t) < n]
    onb = [o[j] for _, j in inwin]
    onm = sum(onb) / len(onb) if onb else 0.0
    allm = sum(o) / n
    # confidence = geometric mean of two independent, honest checks:
    #   acPeak         how periodic the onset envelope is at this period (0..1)
    #   onBeatContrast how much onset energy lands ON the emitted grid vs the
    #                  whole track, (on-all)/(on+all), 0 = no better than chance
    # Measured: synthetic click tracks 0.83-0.93, real 90s house 0.32-0.51,
    # pink noise 0.09, silence 0.0. Treat >0.5 as a solid grid, <0.2 as mush.
    ac = max(0.0, min(1.0, r.get(int(round(period)), 0.0)))
    contrast = max(0.0, (onm - allm) / (onm + allm)) if onm + allm else 0.0
    dbc = max(range(4), key=lambda c: sum(o[j] for i, j in inwin if i % 4 == c)
              / max(1, sum(1 for i, _ in inwin if i % 4 == c)))
    out.update(bpm=round(60 * BEAT_FPS / period, 2),
               period=round(period / BEAT_FPS, 6),
               confidence=round(math.sqrt(ac * contrast), 3),
               acPeak=round(ac, 3), onBeatContrast=round(contrast, 3),
               beats=[round(max(0.0, t / BEAT_FPS), 4) for t in beats_f],
               downbeats=[round(max(0.0, t / BEAT_FPS), 4) for t in beats_f[dbc::4]])
    if truncated:
        out["truncated"] = True     # grid extrapolated past the analysed window
    return out


@app.get("/beats")
def beats(path: str):
    """Beat grid for an audio asset -> {bpm, confidence, beats[], downbeats[],
    duration, ...}. Times are absolute seconds from the start of the file; the
    grid is constant-tempo and 4/4 is assumed (see the section header for the
    algorithm and what `confidence` means). Cached by (path, mtime, size)."""
    if not HAVE_FFMPEG:
        raise HTTPException(503, "ffmpeg not found on server")
    f = safe_under(ASSETS, path)
    if not f.is_file():
        raise HTTPException(404, "not found")
    st = f.stat()
    key = (str(f), st.st_mtime_ns, st.st_size)
    if key not in _beats_cache:
        if len(_beats_cache) > 32:
            _beats_cache.clear()     # ponytail: tiny bounded cache; clearing beats an LRU here
        _beats_cache[key] = analyse_beats(f)
    return {"path": path, **_beats_cache[key]}


# ---- DIRECTOR MODE (M5) — NL text -> validated preset ops via tailnet LLM ----
# Vocab mirrors PLAN §5 M5. NOTE: Lane A's web/grammar.js hasn't landed yet —
# when it does, re-point these lists at that file (read at boot) so the server
# prompt and the client presets never drift. Single source of truth until then.
SHOTS = ["ECU", "CU", "MCU", "MS", "MWS", "WS", "EWS"]
ANGLES = ["eye", "low", "high", "dutch", "OTS-L", "OTS-R"]
FOCALS = [18, 24, 35, 50, 85]
LIGHTS = ["day", "golden-hour", "night", "noir", "horror", "neon", "overcast"]
MARKS = ["center", "two-shot-L", "two-shot-R", "walk-to-mark", "face-camera", "face-other"]

DIRECTOR_SYS = f"""You are a film director's assistant. Translate the direction into JSON: {{"ops": [...]}} and NOTHING else. Allowed ops (scenegod:direct contract):
- {{"op": "shot", "subject": "<entity id>", "shot": <shot>, "angle": <angle>, "focal": <mm>}} — frame the subject. shot sizes: {SHOTS}. angles: {ANGLES} (default "eye"). focal lengths mm: {FOCALS} (default 35).
- {{"op": "light", "preset": <preset>}} — lighting presets: {LIGHTS}.
- {{"op": "mark", "subject": "<entity id>", "mark": <mark>}} — blocking marks: {MARKS}. "face-other" and two-shots may add "target": "<entity id>".
Use ONLY entity ids from the scene roster. Skip anything you cannot express with these ops. At most 6 ops."""


def _validate_ops(ops, ids: set) -> tuple[list, list]:
    """Enforce the scenegod:direct contract server-side: unknown op names,
    vocab values, or entity ids not in the scene -> `rejected` (UI toasts it).
    Never forwards raw model output."""
    good, rejected = [], []

    def rej(o, why):
        rejected.append({"op": o, "reason": why})

    if not isinstance(ops, list):
        return [], [{"op": ops, "reason": "model output is not an op list"}]
    for o in ops:
        if not isinstance(o, dict):
            rej(o, "op is not an object")
            continue
        kind = o.get("op")
        if kind == "shot":
            o = {"op": "shot", "subject": o.get("subject"), "shot": o.get("shot"),
                 "angle": o.get("angle", "eye"), "focal": o.get("focal", 35)}
            try:
                o["focal"] = int(o["focal"])
            except (TypeError, ValueError):
                pass
            for field, allowed in (("shot", SHOTS), ("angle", ANGLES), ("focal", FOCALS)):
                if o[field] not in allowed:
                    rej(o, f"unknown {field} {o[field]!r}")
                    break
            else:
                if o["subject"] not in ids:
                    rej(o, f"unknown subject {o['subject']!r}")
                else:
                    good.append(o)
        elif kind == "light":
            o = {"op": "light", "preset": o.get("preset")}
            if o["preset"] not in LIGHTS:
                rej(o, f"unknown preset {o['preset']!r}")
            else:
                good.append(o)
        elif kind == "mark":
            keep = {"op": "mark", "subject": o.get("subject"), "mark": o.get("mark")}
            if "target" in o:
                keep["target"] = o["target"]
            if keep["mark"] not in MARKS:
                rej(keep, f"unknown mark {keep['mark']!r}")
            elif keep["subject"] not in ids:
                rej(keep, f"unknown subject {keep['subject']!r}")
            elif "target" in keep and keep["target"] not in ids:
                rej(keep, f"unknown target {keep['target']!r}")
            else:
                good.append(keep)
        else:
            rej(o, f"unknown op {kind!r}")
    return good, rejected


@app.post("/director")
async def director(request: Request):
    """{text, scene:{entities:[{id,kind,label}], time}} -> {ops, rejected}.
    OpenAI-compat chat call (tailnet Ollama); ops validated before returning."""
    url = os.environ.get("SCENEGOD_LLM_URL", "").rstrip("/")
    if not url:
        raise HTTPException(503, "no LLM configured (set SCENEGOD_LLM_URL, e.g. "
                                 "http://100.69.21.128:11434, and SCENEGOD_LLM_MODEL)")
    try:
        body = json.loads(await capped_body(request, 256 * 1024))
    except json.JSONDecodeError:
        raise HTTPException(400, "invalid JSON")
    text = (body.get("text") or "").strip()
    if not text:
        raise HTTPException(400, "text required")
    scene = body.get("scene") or {}
    ents = scene.get("entities") or []
    ids = {e.get("id") for e in ents}
    roster = "\n".join(f"- id={e.get('id')!r} kind={e.get('kind')} label={e.get('label')!r}"
                       for e in ents) or "(empty scene)"
    payload = {
        "model": os.environ.get("SCENEGOD_LLM_MODEL", "qwen2.5:7b"),
        "messages": [
            {"role": "system", "content": DIRECTOR_SYS},
            {"role": "user", "content": f"Scene roster:\n{roster}\n"
                                        f"playhead: {scene.get('time', 0)}s\n\nDirection: {text}"},
        ],
        "temperature": 0.1,
        "response_format": {"type": "json_object"},   # ignored by servers without it
    }
    req = urllib.request.Request(url + "/v1/chat/completions", method="POST",
        data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:   # 7b on m4pro can be slow
            data = json.load(resp)
    except urllib.error.HTTPError as e:
        raise HTTPException(502, f"LLM: {e.code} {e.read()[:200].decode('utf-8', 'replace')}")
    except OSError as e:
        raise HTTPException(502, f"LLM unreachable: {e.__class__.__name__}")
    try:
        parsed = json.loads(data["choices"][0]["message"]["content"])
    except (KeyError, IndexError, TypeError, json.JSONDecodeError):
        raise HTTPException(502, "LLM returned unparseable output")  # never forward raw text
    ops = parsed.get("ops") if isinstance(parsed, dict) else parsed  # accept bare array too
    good, rejected = _validate_ops(ops, ids)
    return {"ops": good, "rejected": rejected}


# ---- MODELBEAST proxy (M4, gated) ----
def _mb_post(path: str, data: bytes, content_type: str) -> dict:
    req = urllib.request.Request(MB_HOST + path, data=data, method="POST",
        headers={"Authorization": "Bearer " + _mb_token(), "Content-Type": content_type})
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return json.load(resp)
    except urllib.error.HTTPError as e:
        raise HTTPException(502, f"MB {path}: {e.code} {e.read()[:200].decode('utf-8', 'replace')}")
    except OSError as e:  # never surface the token in the error
        raise HTTPException(502, f"MB {path} unreachable: {e.__class__.__name__}")


@app.post("/mb/submit")
async def mb_submit(request: Request):
    """Submit a MODELBEAST job (tts/voice-clone/lipsync). Optional input file →
    /api/assets first, then /api/jobs — the meshgod/ops.py contract."""
    if os.environ.get("SCENEGOD_MB") != "1":
        raise HTTPException(503, "MODELBEAST proxy disabled (set SCENEGOD_MB=1)")
    try:
        body = json.loads(await capped_body(request, 256 * 1024))
    except json.JSONDecodeError:
        raise HTTPException(400, "invalid JSON")
    op = (body.get("operator") or "").strip()
    if not op:
        raise HTTPException(400, "operator required")
    asset_id = None
    if body.get("file_b64"):
        raw = base64.b64decode(body["file_b64"].split(",")[-1])
        name = os.path.basename(body.get("file_name") or "input.bin") or "input.bin"
        bnd = uuid.uuid4().hex
        mp = ((f'--{bnd}\r\nContent-Disposition: form-data; name="file"; filename="{name}"\r\n'
               f"Content-Type: application/octet-stream\r\n\r\n").encode()
              + raw + f"\r\n--{bnd}--\r\n".encode())
        a = _mb_post("/api/assets", mp, f"multipart/form-data; boundary={bnd}")
        asset_id = a.get("id") or (a.get("items") or [a])[0].get("id")
    payload = {"operator": op, "params": body.get("params") or {}}
    if asset_id:
        payload["asset_id"] = asset_id
    j = _mb_post("/api/jobs", json.dumps(payload).encode(), "application/json")
    return {"job_id": j.get("id"), "asset_id": asset_id}


# ---- statics (last: catch-all /web is defined after API routes) ----
@app.get("/", response_class=HTMLResponse)
def index():
    f = WEB / "index.html"
    html = f.read_text() if f.is_file() else "<h1>SCENEGOD</h1><p>Lane A page not landed yet.</p>"
    return HTMLResponse(html, headers={"Cache-Control": "no-store"})  # MESHGOD stale-cache lesson


@app.get("/web/{path:path}")
def web_static(path: str):
    full = safe_under(WEB, path)
    if not full.is_file():
        raise HTTPException(404, "not found")
    return FileResponse(full, headers={"Cache-Control": "max-age=60"})


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=int(os.environ.get("SCENEGOD_PORT", 8020)))