server.py

"""
PROJECT CANVAS-5 backend (Docker CPU Space).

Serves the full static app at / and exposes POST /convert, which decodes an
uploaded video frame-by-frame (OpenCV), renders each frame to ASCII (Pillow,
true color), and re-encodes the WHOLE clip with ffmpeg (imageio-ffmpeg). CPU
only - no browser, no GPU, no real-time race - so output is complete and never
black. Each result is also written into the mounted bucket at /data.
"""
import os
import time
import shutil
import pathlib
import tempfile
import json
import subprocess
import threading
import uuid
import queue
import socket
import ipaddress
from urllib.parse import urlparse
from collections import deque

import numpy as np
import cv2
import imageio
import yt_dlp
from PIL import Image, ImageDraw, ImageFont
from fastapi import FastAPI, UploadFile, File, Form, Request, Response
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles

GLYPHS = " .,:;irsXA253hMHGS#9B&@"
BG = (2, 6, 7)
FONT_CANDIDATES = [
    "/usr/share/fonts/truetype/dejavu/DejaVuSansMono-Bold.ttf",
    "/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf",
    "/system/fonts/DroidSansMono.ttf",
    "/System/Library/Fonts/Cache/Courier.ttc",
]

_DATA_OVERRIDE = os.environ.get("CANVAS5_DATA_DIR")
_DATA = pathlib.Path(_DATA_OVERRIDE or "/data")
_USING_DATA_ROOT = bool(_DATA_OVERRIDE) or (_DATA.is_dir() and os.access(_DATA, os.W_OK))
BUCKET = (_DATA / "renders") if _USING_DATA_ROOT else pathlib.Path("bucket")
BUCKET.mkdir(parents=True, exist_ok=True)
UPLOADS = (BUCKET.parent / "uploads") if _USING_DATA_ROOT else (BUCKET / "uploads")
JOBS_DIR = (BUCKET.parent / "jobs") if _USING_DATA_ROOT else (BUCKET / "jobs")
CONTROL_DIR = BUCKET / "control"
UPLOADS.mkdir(parents=True, exist_ok=True)
JOBS_DIR.mkdir(parents=True, exist_ok=True)
CONTROL_DIR.mkdir(parents=True, exist_ok=True)
CHUNK_SIZE = 8 * 1024 * 1024
JOBS = {}
JOBS_LOCK = threading.Lock()
# Environment Detection
IS_SPACE = bool(os.environ.get("SPACE_ID"))
IS_TERMUX = os.path.exists("/data/data/com.termux/files/home")
IS_IPHONE = not IS_TERMUX and os.path.exists("/private/var/mobile")

# Optimize slots: Cloud is shared (low), Phone is private (high)
CONVERT_SLOTS_DEFAULT = "4" if (IS_TERMUX or IS_IPHONE) else "1"
CONVERT_SLOTS = max(1, int(os.environ.get("CANVAS5_CONVERT_SLOTS", CONVERT_SLOTS_DEFAULT)))
CONVERT_GATE = threading.Semaphore(CONVERT_SLOTS)
CONTROL_TOKEN = os.environ.get("CANVAS5_CONTROL_TOKEN", "").strip()
CONTROL_CONTENT_LIMIT = max(4096, int(os.environ.get("CANVAS5_CONTROL_CONTENT_LIMIT", "262144")))
CONTROL_ALLOWED_EXTS = {".txt", ".md", ".json", ".html", ".htm", ".svg", ".css", ".ansi", ".log"}
CONTROL_LOCK = threading.Lock()
CONTROL_EVENTS = deque(maxlen=240)
CONTROL_OBSERVATIONS = deque(maxlen=80)
CONTROL_SUBSCRIBERS = []
CONTROL_STATE = {
    "schema": "project-canvas-5/control-state/v1",
    "settings": {
        "fx": True,
        "theme": {"color": "#00ff41", "density": 0.55, "glow": 1.0},
        "caption": "ASCII displacement field online.",
    },
    "last_command": None,
    "last_observation": None,
}

CONTROL_TOOLS = [
    {"name": "canvas5.telecast", "description": "Broadcast an overlay caption and pulse into connected Canvas-5 pages."},
    {"name": "canvas5.pulse", "description": "Send a short reactive ASCII pulse with optional x/y, color, energy, and life controls."},
    {"name": "canvas5.theme", "description": "Patch live theme controls such as color, hue, density, and glow."},
    {"name": "canvas5.fx", "description": "Turn the reactive background FX layer on or off."},
    {"name": "canvas5.patch", "description": "Apply a mixed live-state patch: caption, theme, FX, density, color, and telecast text."},
    {"name": "canvas5.clear", "description": "Clear transient live pulses/caption state in connected pages."},
    {"name": "canvas5.snapshot", "description": "Ask connected pages to report their current observable state."},
    {"name": "control.state", "description": "Return server-side control state, recent commands, observations, and artifacts."},
    {"name": "artifact.write", "description": "Create a bounded text/JSON/HTML/SVG artifact in the Space render bucket."},
    {"name": "artifact.list", "description": "List control artifacts written under the render bucket."},
    {"name": "artifact.read", "description": "Read a bounded text-like control artifact from the render bucket."},
    {"name": "canvas5_url_resolve", "description": "Resolve a public media URL and start ASCII conversion or preview. Returns a job_id."},
    {"name": "canvas5_job_status", "description": "Read progress/result of a conversion or resolve job by job_id."},
    {"name": "canvas5_index", "description": "Self-describing skillbook: the full tool/endpoint/range/constraint catalog with next-step hints."},
]

CANVAS5_LOOKS = [
    "natural", "vibrance", "vivid", "ektar", "portra", "verita", "kodachrome",
    "technicolor", "neonPop", "monoNeutral", "triX", "hp5", "delta100", "sepia",
    "cyanotype", "colorNegative", "monoNegative", "crossProcess", "bleachBypass", "solarize",
]


def _font(size):
    for p in FONT_CANDIDATES:
        try:
            return ImageFont.truetype(p, size)
        except Exception:
            continue
    return ImageFont.load_default()


def _even(n):
    n = int(round(n))
    return n if n % 2 == 0 else n + 1


def _safe_name(value, fallback="video.mp4"):
    name = pathlib.Path(str(value or fallback)).name
    cleaned = "".join(ch for ch in name if ch.isalnum() or ch in "._-")
    return cleaned[:160] or fallback


def _safe_control_name(value, fallback="canvas5-artifact.txt"):
    name = _safe_name(value, fallback)
    suffix = pathlib.Path(name).suffix.lower()
    if not suffix:
        name = f"{name}.txt"
        suffix = ".txt"
    if suffix not in CONTROL_ALLOWED_EXTS:
        raise ValueError(f"Control artifacts must use one of: {', '.join(sorted(CONTROL_ALLOWED_EXTS))}")
    return name


def _validate_public_media_url(value):
    url = str(value or "").strip()
    parsed = urlparse(url)
    if parsed.scheme not in {"http", "https"} or not parsed.hostname:
        raise ValueError("Use a public http:// or https:// video URL.")
    try:
        addresses = {
            item[4][0]
            for item in socket.getaddrinfo(parsed.hostname, parsed.port or (443 if parsed.scheme == "https" else 80))
        }
    except socket.gaierror as exc:
        raise ValueError("The media URL hostname could not be resolved.") from exc
    for address in addresses:
        ip = ipaddress.ip_address(address)
        if (
            ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast
            or ip.is_reserved or ip.is_unspecified
        ):
            raise ValueError("Private or local network media URLs are not allowed.")
    return url


def _write_json(path, payload):
    tmp = pathlib.Path(str(path) + ".tmp")
    tmp.write_text(json.dumps(payload, indent=2, default=str), encoding="utf-8")
    tmp.replace(path)


def _json_safe(value):
    try:
        json.dumps(value)
        return value
    except Exception:
        return str(value)


def _control_authorized(request):
    if not CONTROL_TOKEN:
        return True
    auth = request.headers.get("authorization", "")
    if auth.lower().startswith("bearer ") and auth[7:].strip() == CONTROL_TOKEN:
        return True
    return request.query_params.get("token", "") == CONTROL_TOKEN


def _control_forbidden():
    return JSONResponse({"error": "Canvas-5 control token required."}, status_code=403)


def _sse_frame(event_name, payload):
    data = json.dumps(payload, separators=(",", ":"), ensure_ascii=False)
    lines = [
        f"id: {payload.get('id', uuid.uuid4().hex)}",
        f"event: {event_name}",
        f"data: {data}",
        "",
        "",
    ]
    return "\n".join(lines)


def _control_emit(kind, payload=None, event_name="canvas5"):
    envelope = {
        "schema": "project-canvas-5/control-event/v1",
        "id": uuid.uuid4().hex,
        "kind": kind,
        "payload": _json_safe(payload or {}),
        "created_at": time.time(),
    }
    with CONTROL_LOCK:
        CONTROL_EVENTS.append(envelope)
        CONTROL_STATE["last_command"] = envelope if event_name == "canvas5" else CONTROL_STATE.get("last_command")
        subscribers = list(CONTROL_SUBSCRIBERS)
    for subscriber in subscribers:
        try:
            subscriber.put_nowait((event_name, envelope))
        except queue.Full:
            try:
                subscriber.get_nowait()
                subscriber.put_nowait((event_name, envelope))
            except Exception:
                pass
    return envelope


def _control_observe(payload):
    observation = {
        "schema": "project-canvas-5/control-observation/v1",
        "id": uuid.uuid4().hex,
        "kind": str(payload.get("kind") or "canvas5.observation")[:80],
        "payload": _json_safe(payload.get("payload") or {}),
        "state": _json_safe(payload.get("state") or {}),
        "created_at": time.time(),
    }
    with CONTROL_LOCK:
        CONTROL_OBSERVATIONS.append(observation)
        CONTROL_STATE["last_observation"] = observation
        subscribers = list(CONTROL_SUBSCRIBERS)
    for subscriber in subscribers:
        try:
            subscriber.put_nowait(("canvas5.observation", observation))
        except queue.Full:
            pass
    return observation


def _artifact_record(path):
    return {
        "name": path.name,
        "bytes": path.stat().st_size,
        "modified": path.stat().st_mtime,
        "url": f"/api/control/artifacts/{path.name}",
    }


def _list_control_artifacts():
    items = []
    for path in CONTROL_DIR.iterdir():
        if path.is_file() and path.suffix.lower() in CONTROL_ALLOWED_EXTS:
            items.append(_artifact_record(path))
    return sorted(items, key=lambda item: item["modified"], reverse=True)


def _write_control_artifact(name, content):
    safe = _safe_control_name(name)
    if not isinstance(content, str):
        content = json.dumps(content, indent=2, ensure_ascii=False)
    raw = content.encode("utf-8")
    if len(raw) > CONTROL_CONTENT_LIMIT:
        raise ValueError(f"Control artifact exceeds {CONTROL_CONTENT_LIMIT} bytes.")
    path = CONTROL_DIR / safe
    path.write_bytes(raw)
    return _artifact_record(path)


def _read_control_artifact(name):
    safe = _safe_control_name(name)
    path = CONTROL_DIR / safe
    if not path.is_file():
        raise FileNotFoundError("Control artifact not found.")
    if path.stat().st_size > CONTROL_CONTENT_LIMIT:
        raise ValueError("Control artifact is too large to read through JSON.")
    record = _artifact_record(path)
    record["content"] = path.read_text(encoding="utf-8", errors="replace")
    return record


def _control_snapshot():
    with CONTROL_LOCK:
        events = list(CONTROL_EVENTS)[-24:]
        observations = list(CONTROL_OBSERVATIONS)[-12:]
        state = json.loads(json.dumps(CONTROL_STATE))
    return {
        "schema": "project-canvas-5/control-snapshot/v1",
        "ok": True,
        "sse_url": "/api/control/sse",
        "command_url": "/api/control/command",
        "tool_url": "/api/control/tool",
        "observe_url": "/api/control/observe",
        "artifacts_url": "/api/control/artifacts",
        "token_required": bool(CONTROL_TOKEN),
        "subscribers": len(CONTROL_SUBSCRIBERS),
        "state": state,
        "tools": CONTROL_TOOLS,
        "recent_events": events,
        "recent_observations": observations,
        "artifacts": _list_control_artifacts(),
    }


def _failure(error_type, message, next_step, detail=None):
    """The Yellow Brick Road: every failure carries an actionable next step."""
    envelope = {
        "ok": False,
        "schema": "canvas5/failure-mode/v1",
        "error_type": error_type,
        "message": str(message),
        "actionable_next_step": next_step,
    }
    if detail is not None:
        envelope["detail"] = _json_safe(detail)
    return envelope


def _canvas5_skillbook(args=None):
    """Self-describing index so an SSE agent discovers the whole surface cold.

    Multi-view for different learning models: `view` selects one of
    all|tools|cookbook|recipes|quickref|narrative (default all).
    """
    args = args or {}
    book = {
        "ok": True,
        "schema": "project-canvas-5/skillbook/v1",
        "summary": (
            "Canvas-5 ASCII media lab. Resolve/convert public video URLs to ASCII video, "
            "drive live visual state, and read artifacts. Poll job_status as the source of "
            "truth; transient SSE events are hints and may be missed on reconnect."
        ),
        "transports": {
            "sse": "/api/control/sse",
            "tool_call": "/api/control/tool",
            "mcp_call_alias": "/api/mcp/call",
            "tools_list": "/api/control/tools",
        },
        "call_shape": {"name": "<tool name>", "arguments": {"...": "..."}},
                "tools": [
            {
                "name": "canvas5_url_resolve",
                "aliases": ["url_resolve", "resolve"],
                "description": "Resolve a public media URL and start ASCII conversion (or preview).",
                "parameters": {
                    "url": {"type": "string", "format": "uri", "required": True},
                    "operation": {"type": "string", "enum": ["convert", "preview"], "default": "convert"},
                    "look": {"type": "string", "enum": CANVAS5_LOOKS},
                    "cols": {"type": "integer", "min": 24, "max": 220, "default": 96},
                    "fps": {"type": "integer", "min": 0, "max": 60, "default": 0},
                }
            },
            {
                "name": "canvas5_job_status",
                "aliases": ["job_status", "status"],
                "description": "Read the status/progress/result of a conversion or resolve job.",
                "parameters": {"job_id": {"type": "string", "required": True}}
            },
            {
                "name": "canvas5_telecast",
                "aliases": ["telecast"],
                "description": "Broadcast an overlay caption and pulse into connected pages.",
                "parameters": {
                    "text": {"type": "string", "required": True},
                    "color": {"type": "string", "description": "Hex color for the pulse"},
                    "energy": {"type": "number", "default": 1.0}
                }
            },
            {
                "name": "canvas5_pulse",
                "aliases": ["pulse"],
                "description": "Send a reactive ASCII pulse with coordinate controls.",
                "parameters": {
                    "text": {"type": "string", "default": "PULSE"},
                    "x": {"type": "integer", "min": 0, "max": 100},
                    "y": {"type": "integer", "min": 0, "max": 100},
                    "color": {"type": "string"}
                }
            },
            {
                "name": "canvas5_theme_patch",
                "aliases": ["theme"],
                "description": "Patch live theme controls: color, hue, density, and glow.",
                "parameters": {
                    "color": {"type": "string"},
                    "density": {"type": "number", "min": 0, "max": 1},
                    "glow": {"type": "number"}
                }
            },
            {
                "name": "canvas5_fx_toggle",
                "aliases": ["fx"],
                "description": "Turn the reactive background FX layer on or off.",
                "parameters": {"enabled": {"type": "boolean", "default": True}}
            },
            {
                "name": "canvas5_surface_patch",
                "aliases": ["patch"],
                "description": "Combined patch for caption, theme, and FX state.",
                "parameters": {
                    "caption": {"type": "string"},
                    "fx": {"type": "boolean"},
                    "theme": {"type": "object"}
                }
            },
            {
                "name": "canvas5_clear",
                "aliases": ["clear"],
                "description": "Clear transient live pulses and caption state.",
                "parameters": {}
            },
            {
                "name": "canvas5_snapshot",
                "aliases": ["snapshot"],
                "description": "Ask connected pages to report their current observable state.",
                "parameters": {}
            },
            {
                "name": "control_state",
                "aliases": ["state"],
                "description": "Get server-side control state, recent commands, and observations.",
                "parameters": {}
            },
            {
                "name": "artifact_write",
                "aliases": ["write"],
                "description": "Create a bounded text/JSON/HTML/SVG artifact in the bucket.",
                "parameters": {
                    "name": {"type": "string", "required": True},
                    "content": {"type": "string", "required": True}
                }
            },
            {
                "name": "artifact_list",
                "aliases": ["list"],
                "description": "List control artifacts in the bucket.",
                "parameters": {}
            },
            {
                "name": "artifact_read",
                "aliases": ["read"],
                "description": "Read a bounded text-like artifact from the bucket.",
                "parameters": {"name": {"type": "string", "required": True}}
            },
            {
                "name": "canvas5_index",
                "aliases": ["playbook"],
                "description": "This skillbook: the full discoverable tool catalog.",
                "parameters": {"view": {"type": "string", "enum": ["all", "tools", "cookbook", "recipes", "quickref", "narrative"], "default": "all"}}
            }
        ],
        "governance": {
            "ranges": {
                "cols": {"min": 24, "max": 220, "default": 96},
                "fps": {"min": 0, "max": 60, "note": "0 keeps source fps, clamped to 1-60"},
            },
            "concurrency": {"convert_slots_env": "CANVAS5_CONVERT_SLOTS", "default": 1},
            "constraints": [
                "Facebook share URLs fail in yt-dlp (Cannot parse data). Fallback: use the in-browser CAPTURE_TAB route, or supply a direct media file URL.",
                "Only public http/https URLs resolve; private/loopback/link-local hosts are rejected.",
                "Linked downloads are capped by CANVAS5_URL_MAX_BYTES (default 8 GiB).",
            ],
        },
        "error_envelope_schema": "canvas5/failure-mode/v1",
        "truth_source": "Poll canvas5_job_status (or GET /api/jobs/{id}); SSE events may be missed on reconnect.",
    }
    book["recipes"] = [
        {
            "goal": "Convert a public video URL to ASCII video",
            "steps": [
                "Call canvas5_url_resolve {url, look?, cols?, fps?} -> get job_id.",
                "Poll canvas5_job_status {job_id} every ~1s until status='complete'.",
                "GET the returned artifact_url (under /api/artifacts/) to fetch the MP4.",
            ],
            "on_failure": "If status='error' on a Facebook/unsupported source, switch to the in-browser CAPTURE_TAB route.",
        },
        {
            "goal": "Get a fast browser-playable preview of a link",
            "steps": [
                "Call canvas5_url_resolve {url, operation:'preview'} -> job_id.",
                "Poll canvas5_job_status until complete; artifact_url is a H.264/AAC preview.",
            ],
        },
        {
            "goal": "Drive the live visuals of a connected page",
            "steps": [
                "Subscribe GET /api/control/sse (the target page must be open and subscribed).",
                "Call canvas5_surface_patch {caption?, color?, density?, glow?, fx?} to mutate state.",
                "Watch the canvas5.command events arrive on the SSE stream.",
            ],
        },
        {
            "goal": "Make and read a text artifact (banner/ansi/json)",
            "steps": [
                "Call artifact.write {name, content} (.txt/.md/.json/.html/.svg/.css, bounded).",
                "Discover names with artifact.list; read with canvas5_artifact_read {name}.",
            ],
        },
        {
            "goal": "Recover after a disconnect",
            "steps": [
                "Re-open the page; it auto-reattaches running/finished jobs saved in localStorage.",
                "Or call canvas5_job_status {job_id} directly if you kept the id.",
            ],
        },
    ]
    book["quickref"] = (
        "url_resolve{url,look,cols(24-220),fps(0-60)}->job_id | "
        "job_status{job_id}->progress/artifact_url | "
        "surface_patch{caption,color,density(0-1),fx} | "
        "artifact_read{name} | index{view:all|tools|cookbook|quickref|narrative}. "
        "Truth=poll job_status. Facebook=>CAPTURE_TAB."
    )
    book["narrative"] = (
        "You are talking to Canvas-5, an ASCII media lab. To turn a video link into ASCII video, "
        "call canvas5_url_resolve with the URL; you receive a job_id. Poll canvas5_job_status with "
        "that id until it reports complete, then download artifact_url. Sizes are governed: columns "
        "24-220 (default 96), fps 0-60. To paint the live page, subscribe to /api/control/sse and "
        "call canvas5_surface_patch. Every error returns a quinesmith/failure-mode/v1 envelope with "
        "an actionable_next_step - follow it. Facebook share links cannot be resolved server-side; "
        "use CAPTURE_TAB instead. Trust polled job status over transient SSE events."
    )
    view = str(args.get("view") or args.get("format") or "all").lower()
    view_fields = {
        "tools": ["summary", "transports", "call_shape", "tools", "error_envelope_schema"],
        "cookbook": ["summary", "recipes", "governance", "error_envelope_schema"],
        "recipes": ["summary", "recipes"],
        "quickref": ["summary", "quickref", "governance"],
        "narrative": ["summary", "narrative", "truth_source"],
    }
    if view in view_fields:
        keep = set(view_fields[view]) | {"ok", "schema", "view"}
        slim = {k: v for k, v in book.items() if k in keep}
        slim["view"] = view
        return slim
    book["view"] = "all"
    book["available_views"] = ["all", "tools", "cookbook", "recipes", "quickref", "narrative"]
    return book


def _tool_url_resolve(args):
    raw = args.get("url") or args.get("source") or ""
    try:
        url = _validate_public_media_url(raw)
    except ValueError as exc:
        host = (urlparse(str(raw)).hostname or "").lower()
        if "facebook." in host or "fb.watch" in host or host == "fb.com":
            return _failure(
                "UnsupportedSource", str(exc),
                "Facebook share URLs cannot be parsed by yt-dlp. Use the in-browser CAPTURE_TAB route, or supply a direct media file URL.",
            )
        return _failure(
            "InvalidUrl", str(exc),
            "Supply a public http(s) URL to a direct media file or a yt-dlp-supported page.",
        )
    settings = dict(args.get("settings")) if isinstance(args.get("settings"), dict) else {}
    for key in ("look", "cols", "fps", "color"):
        if key in args and key not in settings:
            settings[key] = args[key]
    operation = "preview" if str(args.get("operation") or settings.get("operation") or "").lower() == "preview" else "convert"
    settings["operation"] = operation
    job_id = uuid.uuid4().hex
    _job_update(
        job_id, status="queued", stage="Queued", progress=0.0,
        operation=operation, source_url=url, created_at=time.time(),
    )
    threading.Thread(
        target=_run_url_job, args=(job_id, url, settings),
        daemon=True, name=f"canvas5-url-{operation}-{job_id[:8]}",
    ).start()
    return {
        "ok": True, "job_id": job_id, "operation": operation,
        "next_step": "Poll canvas5_job_status with this job_id until status=complete, then read artifact_url.",
    }


def _tool_job_status(args):
    job_id = str(args.get("job_id") or args.get("id") or "").strip()
    if not job_id:
        return _failure("MissingArgument", "job_id is required.", "Call canvas5_url_resolve first to obtain a job_id.")
    with JOBS_LOCK:
        job = JOBS.get(job_id)
    if job is None:
        path = JOBS_DIR / f"{_safe_name(job_id, '')}.json"
        if path.exists():
            job = json.loads(path.read_text(encoding="utf-8"))
    if job is None:
        return _failure("JobNotFound", f"No job {job_id}.", "The job id may be wrong or its record was cleaned. Start a new canvas5_url_resolve.")
    result = {"ok": True, **job}
    if job.get("status") == "error":
        result["failure"] = _failure(
            "JobFailed", job.get("error") or "Conversion failed.",
            "Inspect the error; for unsupported sources use CAPTURE_TAB or a direct media file URL.",
        )
    return result


_CANVAS5_MEDIA_TOOLS = {
    "canvas5_index": "index", "canvas5.index": "index", "canvas5_playbook": "index", "canvas5.playbook": "index", "playbook": "index",
    "canvas5_url_resolve": "url_resolve", "canvas5.url_resolve": "url_resolve", "url_resolve": "url_resolve", "resolve": "url_resolve",
    "canvas5_job_status": "job_status", "canvas5.job_status": "job_status", "job_status": "job_status", "status": "job_status",
    "canvas5.telecast": "telecast", "canvas5_telecast": "telecast", "telecast": "telecast",
    "canvas5.pulse": "pulse", "canvas5_pulse": "pulse", "pulse": "pulse",
    "canvas5.theme": "theme", "canvas5_theme": "theme", "theme": "theme",
    "canvas5.fx": "fx", "canvas5_fx": "fx", "fx": "fx",
    "canvas5.patch": "patch", "canvas5_patch": "patch", "patch": "patch",
    "canvas5.clear": "clear", "canvas5_clear": "clear", "clear": "clear",
    "canvas5.snapshot": "snapshot", "canvas5_snapshot": "snapshot", "snapshot": "snapshot",
    "control.state": "control_state", "control_state": "control_state",
    "artifact.write": "artifact_write", "artifact_write": "artifact_write",
    "artifact.list": "artifact_list", "artifact_list": "artifact_list",
    "artifact.read": "artifact_read", "artifact_read": "artifact_read",
}


def _control_tool_result(name, args):
    args = args or {}
    op = _CANVAS5_MEDIA_TOOLS.get(name)
    if not op:
        # Fallback to prefix stripping if not in map
        if name.startswith("canvas5."):
            op = name.replace("canvas5.", "", 1)
        elif name.startswith("artifact."):
            op = name.replace(".", "_", 1)
        else:
            op = name

    if op == "index":
        return _canvas5_skillbook(args)
    if op == "url_resolve":
        return _tool_url_resolve(args)
    if op == "job_status":
        return _tool_job_status(args)
    if op == "control_state" or op == "state":
        return _control_snapshot()
    if op == "artifact_write":
        record = _write_control_artifact(args.get("name") or args.get("path"), args.get("content") or "")
        event = _control_emit("canvas5.artifact", {"action": "telecast", "text": "Artifact written: {0}".format(record['name']), "artifact": record})
        return {"ok": True, "artifact": record, "event": event}
    if op == "artifact_list":
        return {"ok": True, "artifacts": _list_control_artifacts()}
    if op == "artifact_read":
        return {"ok": True, "artifact": _read_control_artifact(args.get("name") or args.get("path"))}
    
    # Visual Patch family
    if op in {"telecast", "pulse", "theme", "fx", "patch", "clear", "snapshot"}:
        payload = {"action": op}
        if op == "telecast":
            payload["text"] = args.get("text") or args.get("message") or args.get("value") or ""
            payload.update({k: v for k, v in args.items() if k not in payload})
        elif op == "pulse":
            payload["text"] = args.get("text") or args.get("message") or args.get("value") or "PULSE"
            payload.update({k: v for k, v in args.items() if k not in payload})
        elif op == "theme":
            CONTROL_STATE["settings"]["theme"].update({k: v for k, v in args.items() if k in {"color", "hue", "density", "glow"}})
            payload.update(args)
        elif op == "fx":
            enabled = args.get("enabled", args.get("value", True))
            CONTROL_STATE["settings"]["fx"] = bool(enabled)
            payload["enabled"] = bool(enabled)
        elif op == "patch":
            if "caption" in args: CONTROL_STATE["settings"]["caption"] = str(args["caption"])
            if "fx" in args: CONTROL_STATE["settings"]["fx"] = bool(args["fx"])
            theme = args.get("theme") if isinstance(args.get("theme"), dict) else args
            CONTROL_STATE["settings"]["theme"].update({k: v for k, v in theme.items() if k in {"color", "hue", "density", "glow"}})
            payload.update(args)
        
        event = _control_emit("canvas5.command", payload)
        return {"ok": True, "event": event, "state": CONTROL_STATE}

    raise ValueError("Unknown Canvas-5 control tool: {0} (mapped to {1})".format(name, op))
def _has_audio(path):
    probe = subprocess.run(
        [
            "ffprobe", "-v", "error", "-select_streams", "a:0",
            "-show_entries", "stream=codec_name", "-of", "default=nw=1:nk=1",
            str(path),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    return probe.returncode == 0 and bool(probe.stdout.strip())


def _probe_duration(path):
    probe = subprocess.run(
        [
            "ffprobe", "-v", "error", "-show_entries", "format=duration",
            "-of", "default=nw=1:nk=1", str(path),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    try:
        duration = float(probe.stdout.strip())
    except (TypeError, ValueError):
        duration = 0.0
    return max(0.0, duration)


def transcode_browser_preview(in_path, progress=None):
    stamp = time.strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:8]
    out_path = BUCKET / f"canvas5-preview-{stamp}.mp4"
    duration = _probe_duration(in_path)
    command = [
        "ffmpeg", "-y", "-loglevel", "error",
        "-i", str(in_path),
        "-map", "0:v:0", "-map", "0:a?",
        "-vf", "scale=w='min(1280,iw)':h=-2",
        "-c:v", "libx264", "-preset", "ultrafast", "-crf", "25",
        "-pix_fmt", "yuv420p",
        "-c:a", "aac", "-b:a", "128k",
        "-movflags", "+faststart", "-shortest",
        "-progress", "pipe:1", "-nostats",
        str(out_path),
    ]
    process = subprocess.Popen(
        command,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    diagnostic = deque(maxlen=30)
    if process.stdout is not None:
        for raw_line in process.stdout:
            line = raw_line.strip()
            if not line:
                continue
            diagnostic.append(line)
            if progress and duration and line.startswith("out_time_ms="):
                try:
                    seconds = int(line.split("=", 1)[1]) / 1_000_000
                    progress(min(98.0, seconds / duration * 98.0), seconds, duration)
                except (TypeError, ValueError):
                    pass
    return_code = process.wait()
    if return_code or not out_path.is_file() or out_path.stat().st_size == 0:
        out_path.unlink(missing_ok=True)
        detail = " | ".join(diagnostic) or "ffmpeg could not create a browser preview"
        raise RuntimeError("Browser preview transcode failed: " + detail)
    if progress:
        progress(100.0, duration, duration)
    return out_path, duration, _has_audio(out_path)



def _concat_segments(job_dir, output_path):
    segments = sorted(job_dir.glob("seg-*.mp4"))
    if not segments:
        raise RuntimeError("No segments found for concatenation.")
    
    concat_file = job_dir / "concat.txt"
    with concat_file.open("w", encoding="utf-8") as f:
        for seg in segments:
            # Escape single quotes for ffmpeg
            safe_path = str(seg.absolute()).replace("'", "'\\''")
            f.write("file '{0}'\n".format(safe_path))
            
    command = [
        "ffmpeg", "-y", "-loglevel", "error",
        "-f", "concat", "-safe", "0",
        "-i", str(concat_file),
        "-c", "copy",
        str(output_path),
    ]
    result = subprocess.run(command, capture_output=True, text=True, check=False)
    if result.returncode:
        raise RuntimeError("Segment concatenation failed: " + (result.stderr.strip() or "ffmpeg error"))





def _mux_source_audio(rendered_path, source_path, output_path):
    command = [
        "ffmpeg", "-y", "-loglevel", "error",
        "-i", str(rendered_path), "-i", str(source_path),
        "-map", "0:v:0", "-map", "1:a?",
        "-c:v", "copy", "-c:a", "aac", "-b:a", "192k",
        "-strict", "-2", "-shortest", "-movflags", "+faststart", str(output_path),
    ]
    result = subprocess.run(command, capture_output=True, text=True, check=False)
    if result.returncode:
        raise RuntimeError("Audio transfer failed: " + (result.stderr.strip() or "ffmpeg error"))


def convert_video(in_path, cols, fps_cap, color, progress=None, job_dir=None, resume_state=None):
    cap = cv2.VideoCapture(in_path)
    if not cap.isOpened():
        raise RuntimeError("Could not open the uploaded video.")
    src_fps = cap.get(cv2.CAP_PROP_FPS) or 24.0
    if not (src_fps and src_fps == src_fps and src_fps > 0):
        src_fps = 24.0
    out_fps = float(min(src_fps, fps_cap)) if fps_cap else float(src_fps)
    out_fps = max(1.0, min(60.0, out_fps))
    w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH) or 16)
    h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT) or 9)
    source_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
    source_duration = source_frames / src_fps if source_frames else 0.0

    cols = int(max(24, min(220, cols)))
    font_size = 12
    font = _font(font_size)
    box = font.getbbox("M")
    cell_w = max(6, box[2] - box[0])
    cell_h = max(8, int(round(font_size * 1.25)))
    rows = int(max(8, round(cols * (h / max(1, w)) * (cell_w / cell_h))))
    out_w = _even(cols * cell_w)
    out_h = _even(rows * cell_h)

    stamp = time.strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:8]
    out_path = BUCKET / "canvas5-ascii-{0}.mp4".format(stamp)
    rendered_path = BUCKET / ".canvas5-video-{0}.mp4".format(stamp)
    
    n = 0
    source_index = 0
    next_output_time = 0.0
    segments_done = 0
    
    if resume_state:
        segments_done = resume_state.get("segments_done", 0)
        n = resume_state.get("frames_done", 0)
        source_index = resume_state.get("source_index", 0)
        next_output_time = resume_state.get("next_output_time", 0.0)
        cap.set(cv2.CAP_PROP_POS_FRAMES, source_index)
        if int(cap.get(cv2.CAP_PROP_POS_FRAMES)) != source_index:
            cap.set(cv2.CAP_PROP_POS_FRAMES, 0)
            for _ in range(source_index):
                cap.read()
    
    seg_frames = 240
    current_seg_frames = 0
    writer = None
    
    def open_writer(seg_idx=None):
        if job_dir:
            job_dir.mkdir(parents=True, exist_ok=True)
            p = job_dir / "seg-{0:04d}.mp4".format(seg_idx)
        else:
            p = rendered_path
        return imageio.get_writer(
            str(p), fps=out_fps, codec="libx264", quality=8,
            pixelformat="yuv420p", macro_block_size=None,
        )

    if not job_dir:
        writer = open_writer()
    else:
        writer = open_writer(segments_done)

    try:
        while True:
            ok, frame = cap.read()
            if not ok:
                break
            frame_time = source_index / src_fps
            source_index += 1
            if frame_time + (0.5 / src_fps) < next_output_time:
                if progress and source_frames:
                    progress(min(90.0, source_index / source_frames * 90.0), source_index, source_frames)
                continue
            
            if job_dir and current_seg_frames >= seg_frames:
                writer.close()
                segments_done += 1
                job_id = job_dir.name
                _job_update(
                    job_id,
                    segments_done=segments_done,
                    frames_done=n,
                    source_index=source_index - 1,
                    next_output_time=next_output_time,
                    out_fps=out_fps,
                    cols=cols,
                    rows=rows,
                    resumable=True,
                    source_path=in_path
                )
                writer = open_writer(segments_done)
                current_seg_frames = 0

            small = cv2.resize(frame, (cols, rows), interpolation=cv2.INTER_AREA)
            small = cv2.cvtColor(small, cv2.COLOR_BGR2RGB)
            luma = 0.2126 * small[:, :, 0] + 0.7152 * small[:, :, 1] + 0.0722 * small[:, :, 2]
            idx = np.clip((luma / 255.0 * (len(GLYPHS) - 1)).astype(int), 0, len(GLYPHS) - 1)
            img = Image.new("RGB", (out_w, out_h), BG)
            draw = ImageDraw.Draw(img)
            if color:
                for y in range(rows):
                    yy = y * cell_h
                    for x in range(cols):
                        ch = GLYPHS[idx[y, x]]
                        if ch == " ":
                            continue
                        r, g, b = small[y, x]
                        draw.text((x * cell_w, yy), ch, font=font, fill=(int(r), int(g), int(b)))
            else:
                for y in range(rows):
                    line = "".join(GLYPHS[idx[y, x]] for x in range(cols))
                    draw.text((0, y * cell_h), line, font=font, fill=(0, 255, 65))
            writer.append_data(np.asarray(img))
            n += 1
            current_seg_frames += 1
            next_output_time += 1.0 / out_fps
            if progress and source_frames:
                progress(min(90.0, source_index / source_frames * 90.0), source_index, source_frames)
    finally:
        cap.release()
        if writer:
            writer.close()
            if job_dir and current_seg_frames > 0:
                segments_done += 1
                job_id = job_dir.name
                _job_update(
                    job_id,
                    segments_done=segments_done,
                    frames_done=n,
                    source_index=source_index,
                    next_output_time=next_output_time,
                    out_fps=out_fps,
                    cols=cols,
                    rows=rows,
                    resumable=True,
                    source_path=in_path
                )
    if n == 0:
        raise RuntimeError("No frames decoded from the uploaded video.")
    try:
        if progress:
            progress(94.0, source_index, source_frames)
        if job_dir:
            _concat_segments(job_dir, rendered_path)
        if _has_audio(in_path):
            _mux_source_audio(rendered_path, in_path, out_path)
        else:
            rendered_path.replace(out_path)
        if job_dir:
            shutil.rmtree(job_dir, ignore_errors=True)
    finally:
        rendered_path.unlink(missing_ok=True)
    if progress:
        progress(100.0, source_index, source_frames)
    return out_path, n, out_fps, out_w, out_h, source_duration, _has_audio(out_path)


def _job_update(job_id, **values):
    with JOBS_LOCK:
        current = dict(JOBS.get(job_id, {}))
        current.update(values)
        current["updated_at"] = time.time()
        JOBS[job_id] = current
        _write_json(JOBS_DIR / f"{job_id}.json", current)
        return current


def _run_conversion_job(job_id, source_path, settings):
    job = _job_update(job_id, status="queued", stage="Waiting for converter", progress=0.0)
    last_report = [0.0]

    def report(percent, frame, total):
        now = time.monotonic()
        if percent < 100.0 and now - last_report[0] < 0.75:
            return
        last_report[0] = now
        _job_update(
            job_id,
            status="converting",
            stage="Rendering ASCII frames" if percent < 94 else "Transferring source audio",
            progress=round(float(percent), 2),
            source_frame=frame,
            source_frames=total,
        )

    try:
        with CONVERT_GATE:
            resume_state = job if job.get("resumable") else None
            _job_update(job_id, status="converting", stage="Rendering ASCII frames", progress=job.get("progress", 1.0))
            result = convert_video(
                source_path,
                int(settings.get("cols") or job.get("cols") or 96),
                int(settings.get("fps") or job.get("fps") or 0),
                str(settings.get("color") or job.get("color") or "color") != "mono",
                report,
                job_dir=JOBS_DIR / job_id,
                resume_state=resume_state
            )
        out_path, frames, out_fps, width, height, duration, audio = result
        _job_update(
            job_id,
            status="complete",
            stage="Download ready",
            progress=100.0,
            artifact=out_path.name,
            artifact_url="/api/artifacts/{0}".format(out_path.name),
            bytes=out_path.stat().st_size,
            frames=frames,
            fps=out_fps,
            width=width,
            height=height,
            duration=duration,
            audio=audio,
        )
    except Exception as exc:
        _job_update(job_id, status="error", stage="Conversion failed", error=str(exc))
    finally:
        current_job = JOBS.get(job_id, {})
        if current_job.get("status") == "complete" or not current_job.get("resumable"):
            pathlib.Path(source_path).unlink(missing_ok=True)


def _run_preview_job(job_id, source_path):
    _job_update(job_id, status="queued", stage="Waiting for preview transcoder", progress=0.0)
    last_report = [0.0]

    def report(percent, current, duration):
        now = time.monotonic()
        if percent < 100.0 and now - last_report[0] < 0.75:
            return
        last_report[0] = now
        _job_update(
            job_id,
            status="converting",
            stage="Preparing browser-playable preview",
            progress=round(float(percent), 2),
            preview_time=round(float(current), 3),
            duration=round(float(duration), 3),
        )

    try:
        with CONVERT_GATE:
            _job_update(
                job_id,
                status="converting",
                stage="Preparing browser-playable preview",
                progress=1.0,
            )
            out_path, duration, audio = transcode_browser_preview(source_path, report)
        _job_update(
            job_id,
            status="complete",
            stage="ASCII preview ready",
            progress=100.0,
            operation="preview",
            artifact=out_path.name,
            artifact_url=f"/api/artifacts/{out_path.name}",
            bytes=out_path.stat().st_size,
            duration=duration,
            audio=audio,
        )
    except Exception as exc:
        _job_update(job_id, status="error", stage="Preview preparation failed", error=str(exc))
    finally:
        pathlib.Path(source_path).unlink(missing_ok=True)


def _download_url_source(job_id, url):
    output_template = str(UPLOADS / f"{job_id}-source.%(ext)s")

    def progress_hook(status):
        if status.get("status") != "downloading":
            return
        downloaded = int(status.get("downloaded_bytes") or 0)
        total = int(status.get("total_bytes") or status.get("total_bytes_estimate") or 0)
        percent = min(44.0, downloaded / total * 44.0) if total else 4.0
        _job_update(
            job_id,
            status="downloading",
            stage="Downloading linked video",
            progress=round(percent, 2),
            downloaded_bytes=downloaded,
            total_bytes=total,
            speed=status.get("speed"),
            eta=status.get("eta"),
        )

    options = {
        "outtmpl": output_template,
        "format": "bv*+ba/b",
        "merge_output_format": "mp4",
        "noplaylist": True,
        "quiet": True,
        "no_warnings": True,
        "restrictfilenames": True,
        "socket_timeout": 30,
        "progress_hooks": [progress_hook],
        "max_filesize": max(64 * 1024 * 1024, int(os.environ.get("CANVAS5_URL_MAX_BYTES", str(8 * 1024 * 1024 * 1024)))),
    }
    with yt_dlp.YoutubeDL(options) as downloader:
        info = downloader.extract_info(url, download=True)
    candidates = [
        path for path in UPLOADS.glob(f"{job_id}-source.*")
        if path.is_file() and path.suffix != ".part"
    ]
    if not candidates:
        raise RuntimeError("The linked video downloaded, but no media file was produced.")
    source_path = max(candidates, key=lambda path: path.stat().st_size)
    title = str((info or {}).get("title") or source_path.stem)
    return source_path, title


def _run_url_job(job_id, url, settings):
    operation = "preview" if str(settings.get("operation") or "").lower() == "preview" else "convert"
    source_path = None
    try:
        _job_update(job_id, status="resolving", stage="Resolving linked video", progress=1.0)
        safe_url = _validate_public_media_url(url)
        source_path, title = _download_url_source(job_id, safe_url)
        _job_update(
            job_id,
            status="queued",
            stage="Waiting for preview transcoder" if operation == "preview" else "Waiting for ASCII converter",
            progress=45.0,
            title=title,
            source_bytes=source_path.stat().st_size,
        )
        with CONVERT_GATE:
            if operation == "preview":
                def preview_report(percent, current, duration):
                    _job_update(
                        job_id,
                        status="converting",
                        stage="Preparing browser-playable preview",
                        progress=round(45.0 + float(percent) * 0.55, 2),
                        preview_time=round(float(current), 3),
                        duration=round(float(duration), 3),
                    )

                out_path, duration, audio = transcode_browser_preview(source_path, preview_report)
                result = {
                    "operation": "preview",
                    "artifact": out_path.name,
                    "artifact_url": f"/api/artifacts/{out_path.name}",
                    "bytes": out_path.stat().st_size,
                    "duration": duration,
                    "audio": audio,
                }
            else:
                def convert_report(percent, frame, total):
                    _job_update(
                        job_id,
                        status="converting",
                        stage="Rendering linked video as ASCII" if percent < 94 else "Transferring source audio",
                        progress=round(45.0 + float(percent) * 0.55, 2),
                        source_frame=frame,
                        source_frames=total,
                    )

                resume_state = JOBS.get(job_id) if JOBS.get(job_id, {}).get("resumable") else None
                out_path, frames, out_fps, width, height, duration, audio = convert_video(
                    source_path,
                    int(settings.get("cols") or 96),
                    int(settings.get("fps") or 0),
                    str(settings.get("color") or "color") != "mono",
                    convert_report,
                    job_dir=JOBS_DIR / job_id,
                    resume_state=resume_state
                )
                result = {
                    "operation": "convert",
                    "artifact": out_path.name,
                    "artifact_url": f"/api/artifacts/{out_path.name}",
                    "bytes": out_path.stat().st_size,
                    "frames": frames,
                    "fps": out_fps,
                    "width": width,
                    "height": height,
                    "duration": duration,
                    "audio": audio,
                }
        _job_update(
            job_id,
            status="complete",
            stage="ASCII preview ready" if operation == "preview" else "Download ready",
            progress=100.0,
            **result,
        )
    except Exception as exc:
        _job_update(job_id, status="error", stage="Linked video failed", error=str(exc))
    finally:
        if source_path is not None:
            current_job = JOBS.get(job_id, {})
            if current_job.get("status") == "complete" or not current_job.get("resumable"):
                pathlib.Path(source_path).unlink(missing_ok=True)



def _resume_incomplete_jobs():
    for job_path in JOBS_DIR.glob("*.json"):
        try:
            job = json.loads(job_path.read_text(encoding="utf-8"))
            job_id = job.get("job_id") or job_path.stem
            status = job.get("status")
            source_path = job.get("source_path")
            
            if status in {"queued", "converting", "downloading"} and source_path:
                p = pathlib.Path(source_path)
                if p.is_file():
                    operation = job.get("operation", "convert")
                    if operation == "convert":
                        target = _run_conversion_job
                        args = (job_id, source_path, job.get("settings", {}))
                    elif operation == "preview":
                        target = _run_preview_job
                        args = (job_id, source_path)
                    else:
                        continue
                    
                    # Update local JOBS cache
                    with JOBS_LOCK:
                        JOBS[job_id] = job
                        
                    threading.Thread(
                        target=target,
                        args=args,
                        daemon=True,
                        name="canvas5-resume-{0}-{1}".format(operation, job_id[:8]),
                    ).start()
        except Exception as e:
            print("Failed to resume job {0}: {1}".format(job_path, e))






async def _handle_mcp_message(request: Request):
    payload = await request.json()
    msg_id = payload.get("id")
    method = payload.get("method")
    params = payload.get("params", {})
    
    if method == "initialize":
        return {
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {
                    "tools": {"listChanged": False},
                    "resources": {"subscribe": False, "listChanged": False},
                    "prompts": {"listChanged": False}
                },
                "serverInfo": {
                    "name": "project-canvas-5",
                    "version": "1.0.0"
                }
            }
        }
    
    if method == "tools/list":
        tools = []
        book = _canvas5_skillbook({"view": "tools"})
        for t in book.get("tools", []):
            mcp_tool = {
                "name": t["name"],
                "description": t["description"],
                "inputSchema": {
                    "type": "object",
                    "properties": {},
                    "required": []
                }
            }
            # Map parameters to JSON Schema
            for p_name, p_def in t.get("parameters", {}).items():
                p_type = p_def.get("type", "string")
                prop = {"type": p_type}
                if "description" in p_def: prop["description"] = p_def["description"]
                if "default" in p_def: prop["default"] = p_def["default"]
                if "enum" in p_def: prop["enum"] = p_def["enum"]
                if "min" in p_def: prop["minimum"] = p_def["min"]
                if "max" in p_def: prop["maximum"] = p_def["max"]
                
                mcp_tool["inputSchema"]["properties"][p_name] = prop
                if p_def.get("required"):
                    mcp_tool["inputSchema"]["required"].append(p_name)
            
            tools.append(mcp_tool)
            
        return {
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {"tools": tools}
        }
    
    if method == "tools/call":
        tool_name = params.get("name")
        tool_args = params.get("arguments", {})
        try:
            result = _control_tool_result(tool_name, tool_args)
            return {
                "jsonrpc": "2.0",
                "id": msg_id,
                "result": {
                    "content": [{"type": "text", "text": json.dumps(result, indent=2)}]
                }
            }
        except Exception as e:
            return {
                "jsonrpc": "2.0",
                "id": msg_id,
                "error": {"code": -32603, "message": str(e)}
            }

    if method == "notifications/initialized":
        return None # No response needed

    return {
        "jsonrpc": "2.0",
        "id": msg_id,
        "error": {"code": -32601, "message": f"Method {method} not found"}
    }



app = FastAPI()


@app.post("/api/monkey/push")
async def monkey_push_retired(request: Request):
    return JSONResponse(
        {
            "ok": False,
            "error": "Monkey MJPEG stream is retired. Use OBS Window Capture on the Canvas-5 Monkey deck.",
        },
        status_code=410,
    )


@app.get("/api/monkey/stream")
async def monkey_stream_retired(request: Request):
    return JSONResponse(
        {
            "ok": False,
            "error": "Monkey MJPEG stream is retired. Use OBS Window Capture on the Canvas-5 Monkey deck.",
        },
        status_code=410,
    )

@app.on_event("startup")
def on_startup():
    _resume_incomplete_jobs()









@app.get("/healthz")
def healthz():
    return {
        "ok": True,
        "bucket": str(BUCKET),
        "chunk_size": CHUNK_SIZE,
        "large_file_jobs": True,
        "audio_transfer": True,
        "browser_preview_transcode": True,
        "linked_video_jobs": True,
        "control_sse": True,
        "control_tools": True,
        "control_artifacts": True,
        "control_token_required": bool(CONTROL_TOKEN),
    }


@app.get("/api/control/status")
def control_status(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    return _control_snapshot()


@app.get("/api/control/tools")
def control_tools(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    return {
        "schema": "project-canvas-5/control-tools/v1",
        "ok": True,
        "tools": CONTROL_TOOLS,
        "tool_url": "/api/control/tool",
        "mcp_aliases": ["/api/mcp/tools", "/api/mcp/call"],
    }


@app.get("/api/mcp/tools")
def mcp_tools_alias(request: Request):
    return control_tools(request)


@app.get("/api/control/sse")
def control_sse(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()

    def stream():
        subscriber = queue.Queue(maxsize=64)
        with CONTROL_LOCK:
            CONTROL_SUBSCRIBERS.append(subscriber)
        hello = {
            "schema": "project-canvas-5/control-event/v1",
            "id": uuid.uuid4().hex,
            "kind": "canvas5.control.hello",
            "payload": {
                "message": "Canvas-5 server control bus connected.",
                "tools": [tool["name"] for tool in CONTROL_TOOLS],
                "observe_url": "/api/control/observe",
            },
            "created_at": time.time(),
        }
        try:
            yield _sse_frame("canvas5.control", hello)
            while True:
                try:
                    event_name, payload = subscriber.get(timeout=15)
                    yield _sse_frame(event_name, payload)
                except queue.Empty:
                    yield ": canvas5 keepalive\n\n"
        finally:
            with CONTROL_LOCK:
                if subscriber in CONTROL_SUBSCRIBERS:
                    CONTROL_SUBSCRIBERS.remove(subscriber)

    return StreamingResponse(
        stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.post("/api/control/observe")
async def control_observe(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    payload = await request.json()
    return {"ok": True, "observation": _control_observe(payload)}


@app.post("/api/control/command")
async def control_command(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    payload = await request.json()
    if "payload" in payload and "kind" in payload:
        command_payload = dict(payload.get("payload") or {})
        command_payload.setdefault("action", str(payload.get("kind") or "canvas5.telecast").replace("canvas5.", ""))
    else:
        command_payload = dict(payload)
    event = _control_emit("canvas5.command", command_payload)
    return {"ok": True, "event": event}


@app.post("/api/control/tool")
async def control_tool(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    payload = await request.json()
    name = str(payload.get("name") or payload.get("tool") or "").strip()
    args = payload.get("arguments")
    if args is None:
        args = payload.get("args")
    if args is None:
        args = payload.get("payload")
    try:
        if not name:
            raise ValueError("Missing control tool name.")
        return _control_tool_result(name, args or {})
    except Exception as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=400)


@app.post("/api/mcp/call")
async def mcp_call_alias(request: Request):
    return await control_tool(request)


@app.get("/api/control/artifacts")
def control_artifacts(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    return {"ok": True, "artifacts": _list_control_artifacts()}


@app.post("/api/control/artifacts")
async def write_control_artifact(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    payload = await request.json()
    try:
        record = _write_control_artifact(payload.get("name") or payload.get("path"), payload.get("content") or "")
        _control_emit("canvas5.artifact", {"action": "telecast", "text": f"Artifact written: {record['name']}", "artifact": record})
        return {"ok": True, "artifact": record}
    except Exception as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=400)


@app.get("/api/control/artifacts/{name}")
def get_control_artifact(request: Request, name: str):
    if not _control_authorized(request):
        return _control_forbidden()
    try:
        safe = _safe_control_name(name, "")
    except ValueError as exc:
        return JSONResponse({"error": str(exc)}, status_code=400)
    path = CONTROL_DIR / safe
    if not safe or not path.is_file():
        return JSONResponse({"error": "Control artifact not found."}, status_code=404)
    media = {
        ".css": "text/css",
        ".html": "text/html",
        ".htm": "text/html",
        ".json": "application/json",
        ".svg": "image/svg+xml",
    }.get(path.suffix.lower(), "text/plain")
    return FileResponse(str(path), media_type=media, filename=path.name)


@app.post("/convert")
def convert(video: UploadFile = File(...), cols: int = Form(96),
            fps: int = Form(0), color: str = Form("color")):
    suffix = os.path.splitext(video.filename or "input.mp4")[1] or ".mp4"
    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
    try:
        with tmp as fh:
            shutil.copyfileobj(video.file, fh)
        with CONVERT_GATE:
            out_path, n, out_fps, ow, oh, duration, audio = convert_video(
                tmp.name, cols, fps, color != "mono"
            )
    except Exception as e:
        return JSONResponse({"error": str(e)}, status_code=500)
    finally:
        try:
            os.unlink(tmp.name)
        except Exception:
            pass
    return FileResponse(
        str(out_path), media_type="video/mp4", filename=out_path.name,
        headers={"X-Canvas5-Frames": str(n), "X-Canvas5-Fps": str(int(out_fps)),
                 "X-Canvas5-Size": f"{ow}x{oh}",
                 "X-Canvas5-Duration": f"{duration:.3f}",
                 "X-Canvas5-Audio": "yes" if audio else "no"},
    )


@app.post("/api/preview")
def create_browser_preview(video: UploadFile = File(...)):
    suffix = os.path.splitext(video.filename or "input.mp4")[1] or ".mp4"
    tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
    try:
        with tmp as handle:
            shutil.copyfileobj(video.file, handle)
        with CONVERT_GATE:
            out_path, duration, audio = transcode_browser_preview(tmp.name)
    except Exception as exc:
        return JSONResponse({"error": str(exc)}, status_code=500)
    finally:
        pathlib.Path(tmp.name).unlink(missing_ok=True)
    return {
        "ok": True,
        "operation": "preview",
        "artifact": out_path.name,
        "artifact_url": f"/api/artifacts/{out_path.name}",
        "bytes": out_path.stat().st_size,
        "duration": duration,
        "audio": audio,
    }


@app.post("/api/url-jobs")
async def create_url_job(request: Request):
    payload = await request.json()
    try:
        url = _validate_public_media_url(payload.get("url"))
    except ValueError as exc:
        return JSONResponse({"error": str(exc)}, status_code=400)
    settings = payload.get("settings") if isinstance(payload.get("settings"), dict) else {}
    operation = "preview" if str(payload.get("operation") or settings.get("operation") or "").lower() == "preview" else "convert"
    settings = {**settings, "operation": operation}
    job_id = uuid.uuid4().hex
    _job_update(
        job_id,
        status="queued",
        stage="Queued",
        progress=0.0,
        operation=operation,
        source_url=url,
        created_at=time.time(),
    )
    threading.Thread(
        target=_run_url_job,
        args=(job_id, url, settings),
        daemon=True,
        name=f"canvas5-url-{operation}-{job_id[:8]}",
    ).start()
    return {"job_id": job_id, "status": "queued", "operation": operation}


@app.post("/api/uploads/init")
async def init_upload(request: Request):
    payload = await request.json()
    upload_id = uuid.uuid4().hex
    filename = _safe_name(payload.get("filename"))
    size = max(0, int(payload.get("size") or 0))
    upload_path = UPLOADS / f"{upload_id}-{filename}.part"
    upload_path.touch()
    meta = {
        "upload_id": upload_id,
        "filename": filename,
        "size": size,
        "uploaded": 0,
        "path": str(upload_path),
        "created_at": time.time(),
    }
    _write_json(UPLOADS / f"{upload_id}.json", meta)
    return {"upload_id": upload_id, "chunk_size": CHUNK_SIZE, "uploaded": 0}


@app.put("/api/uploads/{upload_id}/chunk")
async def upload_chunk(upload_id: str, request: Request, offset: int = 0):
    meta_path = UPLOADS / f"{upload_id}.json"
    if not meta_path.exists():
        return JSONResponse({"error": "Upload session not found."}, status_code=404)
    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    expected = int(meta.get("uploaded") or 0)
    if offset != expected:
        return JSONResponse(
            {"error": f"Chunk offset mismatch. Expected {expected}, received {offset}.", "uploaded": expected},
            status_code=409,
        )
    data = await request.body()
    if not data:
        return JSONResponse({"error": "Empty upload chunk."}, status_code=400)
    path = pathlib.Path(meta["path"])
    with path.open("ab") as handle:
        handle.write(data)
    meta["uploaded"] = expected + len(data)
    _write_json(meta_path, meta)
    return {"uploaded": meta["uploaded"], "size": meta["size"]}


@app.post("/api/uploads/{upload_id}/complete")
async def complete_upload(upload_id: str, request: Request):
    meta_path = UPLOADS / f"{upload_id}.json"
    if not meta_path.exists():
        return JSONResponse({"error": "Upload session not found."}, status_code=404)
    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    if meta["size"] and meta["uploaded"] != meta["size"]:
        return JSONResponse(
            {"error": f"Upload incomplete: {meta['uploaded']} of {meta['size']} bytes."},
            status_code=409,
        )
    settings = await request.json()
    source_path = pathlib.Path(meta["path"])
    suffix = pathlib.Path(meta["filename"]).suffix or ".mp4"
    ready_path = UPLOADS / f"{upload_id}{suffix}"
    source_path.replace(ready_path)
    meta_path.unlink(missing_ok=True)
    job_id = uuid.uuid4().hex
    operation = "preview" if str(settings.get("operation") or "").lower() == "preview" else "convert"
    job = {
        "job_id": job_id,
        "status": "queued",
        "stage": "Queued",
        "progress": 0.0,
        "filename": meta["filename"],
        "operation": operation,
        "created_at": time.time(),
    }
    _job_update(job_id, **{key: value for key, value in job.items() if key != "job_id"})
    target = _run_preview_job if operation == "preview" else _run_conversion_job
    args = (job_id, str(ready_path)) if operation == "preview" else (job_id, str(ready_path), settings)
    threading.Thread(
        target=target,
        args=args,
        daemon=True,
        name=f"canvas5-{operation}-{job_id[:8]}",
    ).start()
    return job


@app.get("/api/uploads/{upload_id}")
def upload_status(upload_id: str):
    """Report chunked-upload progress so a disconnected client can resume.

    The client stores its upload_id locally; on reconnect it reads this and
    continues PUT /chunk from `uploaded` instead of restarting from byte 0.
    """
    meta_path = UPLOADS / f"{_safe_name(upload_id, '')}.json"
    if not upload_id or not meta_path.exists():
        return JSONResponse(
            {"error": "Upload session not found.", "recoverable": False},
            status_code=404,
        )
    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    size = int(meta.get("size") or 0)
    uploaded = int(meta.get("uploaded") or 0)
    return {
        "upload_id": upload_id,
        "filename": meta.get("filename"),
        "size": size,
        "uploaded": uploaded,
        "chunk_size": CHUNK_SIZE,
        "complete": bool(size and uploaded >= size),
        "remaining": max(0, size - uploaded),
    }


@app.get("/api/jobs/{job_id}")
def job_status(job_id: str):
    with JOBS_LOCK:
        job = JOBS.get(job_id)
    if job is None:
        path = JOBS_DIR / f"{job_id}.json"
        if path.exists():
            job = json.loads(path.read_text(encoding="utf-8"))
    if job is None:
        return JSONResponse({"error": "Conversion job not found."}, status_code=404)
    return job


@app.get("/api/artifacts/{name}")
def get_artifact(name: str):
    safe = _safe_name(name, "")
    path = BUCKET / safe
    if not safe or not path.is_file():
        return JSONResponse({"error": "Artifact not found."}, status_code=404)
    return FileResponse(str(path), media_type="video/mp4", filename=path.name)


# Serve the full static app at / (mounted last so /convert and /healthz win).

@app.get("/mcp/sse")
async def mcp_sse(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()

    def stream():
        subscriber = queue.Queue(maxsize=64)
        with CONTROL_LOCK:
            CONTROL_SUBSCRIBERS.append(subscriber)
        
        # The crucial 'endpoint' event for standard MCP clients
        base_url = str(request.base_url).rstrip('/')
        yield f"event: endpoint\ndata: {base_url}/api/mcp/messages\n\n"
        
        try:
            while True:
                try:
                    event_name, payload = subscriber.get(timeout=15)
                    yield _sse_frame(event_name, payload)
                except queue.Empty:
                    yield ": ping\n\n"
        finally:
            with CONTROL_LOCK:
                if subscriber in CONTROL_SUBSCRIBERS:
                    CONTROL_SUBSCRIBERS.remove(subscriber)

    return StreamingResponse(
        stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

@app.post("/api/mcp/messages")
async def mcp_messages(request: Request):
    if not _control_authorized(request):
        return _control_forbidden()
    res = await _handle_mcp_message(request)
    if res is None:
        return Response(status_code=204)
    return JSONResponse(res)


app.mount("/", StaticFiles(directory=".", html=True), name="app")