app.py

from __future__ import annotations

import base64
import importlib.util
import json
import mimetypes
import os
import sys
import time
import urllib.error
import urllib.request
import wave
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from uuid import uuid4

from dotenv import load_dotenv
from fastapi import Request
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from gradio import Server
from pydantic import BaseModel

load_dotenv()

ROOT = Path(__file__).parent
STATIC_DIR = ROOT / "static"
OUTPUT_DIR = ROOT / "outputs"
OUTPUT_DIR.mkdir(exist_ok=True)


def _int_env(name: str, default: int) -> int:
    try:
        return int(os.getenv(name, str(default)))
    except (TypeError, ValueError):
        return default


def _bool_env(name: str, default: bool = False) -> bool:
    value = os.getenv(name)
    if value is None:
        return default
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _runtime_log(message: str) -> None:
    print(f"[tiny-narrator] {message}", file=sys.stderr, flush=True)


def _http_exception_detail(exc: BaseException) -> str:
    if isinstance(exc, urllib.error.HTTPError):
        try:
            body = exc.read().decode("utf-8", errors="replace")
        except Exception:
            body = ""
        body = _compact_text(body, 400) if body else ""
        return f"HTTPError status={exc.code} reason={exc.reason} body={body}"
    if isinstance(exc, urllib.error.URLError):
        return f"URLError reason={exc.reason}"
    return f"{exc.__class__.__name__}: {exc}"


LLAMA_CPP_BASE_URL = os.getenv("LLAMA_CPP_BASE_URL", "http://localhost:8080/v1")
LLAMA_CPP_MODEL = os.getenv("LLAMA_CPP_MODEL", "narrator-brain")
LLAMA_CPP_TOKEN = os.getenv("LLAMA_CPP_TOKEN", "")
LLAMA_CPP_TIMEOUT_SECONDS = _int_env("LLAMA_CPP_TIMEOUT_SECONDS", 90)
GRADIO_SERVER_NAME = os.getenv("GRADIO_SERVER_NAME", "0.0.0.0")
GRADIO_SERVER_PORT = int(os.getenv("PORT", os.getenv("GRADIO_SERVER_PORT", "7860")))
GRADIO_SHARE = _bool_env("GRADIO_SHARE")
PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", f"http://localhost:{GRADIO_SERVER_PORT}").rstrip("/")
KLEIN_MODAL_ENDPOINT = os.getenv("KLEIN_MODAL_ENDPOINT", "").rstrip("/")
KLEIN_MODAL_TOKEN = os.getenv("KLEIN_MODAL_TOKEN", "")
KLEIN_MODAL_HEALTH_TIMEOUT_SECONDS = _int_env("KLEIN_MODAL_HEALTH_TIMEOUT_SECONDS", 30)
KLEIN_MODAL_TIMEOUT_SECONDS = _int_env("KLEIN_MODAL_TIMEOUT_SECONDS", 120)
MINICPM_VISION_BASE_URL = os.getenv("MINICPM_VISION_BASE_URL", "").rstrip("/")
MINICPM_VISION_API_KEY = os.getenv("MINICPM_VISION_API_KEY", "")
MINICPM_VISION_MODEL = os.getenv("MINICPM_VISION_MODEL", "openbmb/MiniCPM-V-4.6")
MINICPM_VISION_TIMEOUT_SECONDS = _int_env("MINICPM_VISION_TIMEOUT_SECONDS", 45)
TINY_TITAN_LIMIT_B = 4.0

MODEL_MANIFEST: dict[str, dict[str, Any]] = {
    "reader_brain": {
        "id": "nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF",
        "params": "3.97B",
        "params_billion": 3.97,
        "runtime": "llama.cpp",
        "role": "Plans concise narration and reading-order phrasing.",
    },
    "vision": {
        "id": "openbmb/MiniCPM-V-4.6",
        "params": "1B",
        "params_billion": 1.0,
        "runtime": "OpenAI-compatible chat completions",
        "role": "Describes images and OCR-like visual details.",
    },
    "speech": {
        "id": "hexgrad/Kokoro-82M",
        "params": "82M",
        "params_billion": 0.082,
        "runtime": "Python",
        "role": "Speaks the final narration.",
    },
    "image_generation": {
        "id": "black-forest-labs/FLUX.2-klein-4B",
        "params": "4B",
        "params_billion": 4.0,
        "runtime": "Modal-hosted Klein",
        "role": "Creates article illustrations.",
    },
}

ARTICLE_IMAGES: list[dict[str, Any]] = [
    {
        "id": "desk-reader",
        "asset_url": "/static/generated/desk-reader.svg",
        "vision_asset_url": "/static/generated/desk-reader.png",
        "caption": "The article view doubles as the demo surface, so every feature has a real reading task.",
        "prompt": "Accessibility article reader with highlighted paragraph and narration controls.",
        "seed": 11,
        "generation_model": "black-forest-labs/FLUX.2-klein-4B",
        "generation_runtime": "bundled fallback asset",
        "generation_status": "fallback-ready",
    },
    {
        "id": "model-map",
        "asset_url": "/static/generated/model-map.svg",
        "vision_asset_url": "/static/generated/model-map.png",
        "caption": "Each model stays at or below four billion parameters for Tiny Titan eligibility.",
        "prompt": "Diagram of four small AI models working together in an accessibility reader.",
        "seed": 22,
        "generation_model": "black-forest-labs/FLUX.2-klein-4B",
        "generation_runtime": "bundled fallback asset",
        "generation_status": "fallback-ready",
    },
]

READER_SETTINGS: dict[str, Any] = {
    "default_voice": "af_heart",
    "default_speed": 1.0,
    "default_auto_advance": False,
    "voices": [
        {"id": "af_heart", "label": "Heart"},
        {"id": "af_bella", "label": "Bella"},
        {"id": "am_adam", "label": "Adam"},
        {"id": "am_michael", "label": "Michael"},
    ],
    "speed": {"min": 0.75, "max": 1.35, "step": 0.05},
}

AWARD_EVIDENCE: list[dict[str, str]] = [
    {
        "id": "tiny-titan",
        "label": "Tiny Titan",
        "status": "ready",
        "evidence": "All planned model roles use models at or below 4B parameters.",
    },
    {
        "id": "llama-champion",
        "label": "Llama Champion",
        "status": "ready",
        "evidence": "The reader-brain path targets a GGUF model through a llama.cpp OpenAI-compatible endpoint.",
    },
    {
        "id": "off-brand",
        "label": "Off-Brand",
        "status": "ready",
        "evidence": "The visible app is custom HTML, CSS, and JavaScript served by Gradio Server.",
    },
    {
        "id": "field-notes",
        "label": "Field Notes",
        "status": "ready",
        "evidence": "The repo records model choices, fallbacks, latency reporting, and accessibility behavior.",
    },
]


def model_budget_core() -> dict[str, Any]:
    models = []
    for role, model in MODEL_MANIFEST.items():
        params_billion = float(model["params_billion"])
        models.append(
            {
                "role": role,
                "id": model["id"],
                "runtime": model["runtime"],
                "params": model["params"],
                "params_billion": params_billion,
                "limit_billion": TINY_TITAN_LIMIT_B,
                "within_limit": params_billion <= TINY_TITAN_LIMIT_B,
            }
        )
    return {
        "ok": True,
        "award": "Tiny Titan",
        "limit_billion": TINY_TITAN_LIMIT_B,
        "all_models_within_limit": all(model["within_limit"] for model in models),
        "models": models,
    }


def runtime_setup_core() -> dict[str, Any]:
    return {
        "ok": True,
        "app": {
            "runtime": "Gradio Server",
            "command": "python app.py",
            "env": {
                "GRADIO_SERVER_NAME": GRADIO_SERVER_NAME,
                "GRADIO_SERVER_PORT": str(GRADIO_SERVER_PORT),
                "GRADIO_SHARE": str(GRADIO_SHARE).lower(),
                "PUBLIC_BASE_URL": PUBLIC_BASE_URL,
            },
        },
        "steps": [
            {
                "role": "reader_brain",
                "label": "Reader brain",
                "model": MODEL_MANIFEST["reader_brain"]["id"],
                "runtime": "llama.cpp",
                "command": (
                    "llama-server -hf nvidia/NVIDIA-Nemotron-3-Nano-4B-GGUF:Q4_K_M "
                    "--alias narrator-brain --port 8080 --host 0.0.0.0 "
                    "--ctx-size 4096 --parallel 1 --reasoning off --n-gpu-layers 999"
                ),
                "modal_command": "modal deploy modal_workers/reader_brain.py",
                "env": {
                    "LLAMA_CPP_BASE_URL": LLAMA_CPP_BASE_URL,
                    "LLAMA_CPP_MODEL": LLAMA_CPP_MODEL,
                    "LLAMA_CPP_TOKEN": "(configured)" if LLAMA_CPP_TOKEN else "(not set)",
                    "LLAMA_CPP_TIMEOUT_SECONDS": str(LLAMA_CPP_TIMEOUT_SECONDS),
                },
                "fallback": "MiniCPM-V-4.6 text fallback, then rule-based local narration",
            },
            {
                "role": "speech",
                "label": "Speech",
                "model": MODEL_MANIFEST["speech"]["id"],
                "runtime": "Python Kokoro",
                "command": "python -m pip install kokoro soundfile",
                "env": {},
                "fallback": "browser speech plus visible transcript",
            },
            {
                "role": "vision",
                "label": "Image descriptions",
                "model": MODEL_MANIFEST["vision"]["id"],
                "runtime": "OpenAI-compatible chat completions",
                "command": "vllm serve openbmb/MiniCPM-V-4.6 --host 0.0.0.0 --port 8000",
                "env": {
                    "MINICPM_VISION_BASE_URL": MINICPM_VISION_BASE_URL or "(set to OpenAI-compatible base URL)",
                    "MINICPM_VISION_API_KEY": "(configured)" if MINICPM_VISION_API_KEY else "(not set)",
                    "MINICPM_VISION_MODEL": MINICPM_VISION_MODEL,
                    "MINICPM_VISION_TIMEOUT_SECONDS": str(MINICPM_VISION_TIMEOUT_SECONDS),
                },
                "fallback": "cached deterministic alt text",
            },
            {
                "role": "image_generation",
                "label": "Article images",
                "model": MODEL_MANIFEST["image_generation"]["id"],
                "runtime": "Modal-hosted Klein",
                "command": "modal deploy modal_workers/klein_image.py",
                "env": {
                    "KLEIN_MODAL_ENDPOINT": KLEIN_MODAL_ENDPOINT or "(set to Modal worker URL)",
                    "KLEIN_MODAL_TOKEN": "(configured)" if KLEIN_MODAL_TOKEN else "(not set)",
                    "KLEIN_MODAL_HEALTH_TIMEOUT_SECONDS": str(KLEIN_MODAL_HEALTH_TIMEOUT_SECONDS),
                    "KLEIN_MODAL_TIMEOUT_SECONDS": str(KLEIN_MODAL_TIMEOUT_SECONDS),
                },
                "fallback": "bundled generated article assets",
            },
        ],
    }


def demo_curl_command(check: dict[str, Any]) -> str:
    url = f"{PUBLIC_BASE_URL}{check['path']}"
    if check["method"] == "GET":
        return f"curl {url}"
    sample_body = json.dumps(check["sample_body"], separators=(",", ":"))
    return f"curl -X POST {url} -H \"Content-Type: application/json\" -d '{sample_body}'"


def demo_powershell_command(check: dict[str, Any]) -> str:
    url = f"{PUBLIC_BASE_URL}{check['path']}"
    if check["method"] == "GET":
        return f"curl.exe {url}"
    sample_body = json.dumps(check["sample_body"], separators=(",", ":")).replace('"', '\\"')
    return f'curl.exe -X POST {url} -H "Content-Type: application/json" -d "{sample_body}"'


def demo_script_core() -> dict[str, Any]:
    api_checks = [
        {"method": "GET", "path": "/api/health", "expect": "custom Gradio Server app and model manifest"},
        {"method": "GET", "path": "/api/model-budget", "expect": "all_models_within_limit is true"},
        {"method": "GET", "path": "/api/runtime-setup", "expect": "llama.cpp, Kokoro, vision, and image paths"},
        {"method": "GET", "path": "/api/runtime-status", "expect": "online or fallback-ready runtime labels"},
        {"method": "GET", "path": "/api/accessibility-audit", "expect": "all reader-mode checks pass"},
        {"method": "GET", "path": "/api/image-descriptions", "expect": "two article image descriptions with MiniCPM-V-4.6 or fallback alt text"},
        {"method": "GET", "path": "/api/submission-readiness", "expect": "all submission readiness checks pass"},
        {
            "method": "POST",
            "path": "/api/reader-brain",
            "expect": "concise narration or fallback narration",
            "sample_body": {
                "node_type": "heading",
                "text": "The model map",
                "position": "item 4 of 10",
                "mode": "narrate",
            },
        },
        {
            "method": "POST",
            "path": "/api/speak",
            "expect": "Kokoro audio or audible browser fallback path",
            "sample_body": {
                "text": "Heading. The model map.",
                "voice": READER_SETTINGS["default_voice"],
                "speed": READER_SETTINGS["default_speed"],
            },
        },
        {
            "method": "POST",
            "path": "/api/generate-article",
            "expect": "generated article draft with Klein thumbnail receipt",
            "sample_body": {"topic": "deep sea bioluminescence"},
        },
    ]
    for check in api_checks:
        check["curl"] = demo_curl_command(check)
        check["powershell"] = demo_powershell_command(check)

    return {
        "ok": True,
        "title": "Tiny Narrator judge demo",
        "goal": "Show a custom article app becoming a small-model screen reader with inspectable evidence.",
        "estimated_minutes": 3,
        "actions": [
            {
                "step": 1,
                "label": "Open the article",
                "action": "Load the home page and scan the article plus session panel.",
                "evidence": "Custom HTML/CSS/JS interface served by Gradio Server.",
            },
            {
                "step": 2,
                "label": "Turn on reader mode",
                "action": "Toggle screen reader mode, then press Space or Next to narrate the first item.",
                "evidence": "The active article node is focused, outlined, narrated, and logged.",
            },
            {
                "step": 3,
                "label": "Navigate by meaning",
                "action": "Use Heading, Image, and Summary controls to jump through semantic article nodes.",
                "evidence": "The reader uses article structure, image alt text, and section summaries.",
            },
            {
                "step": 4,
                "label": "Inspect small-model receipts",
                "action": "Review the checklist, model budget, runtime plan, readiness, latency, and transcript panels.",
                "evidence": "Tiny Titan, Llama Champion, Off-Brand, and reader evidence is exposed through the app and APIs.",
            },
        ],
        "api_checks": api_checks,
    }


def submission_readiness_core() -> dict[str, Any]:
    model_budget = model_budget_core()
    runtime_setup = runtime_setup_core()
    runtime_status = _runtime_status_core()
    accessibility = accessibility_audit_core()
    demo_script = demo_script_core()
    image_descriptions = describe_article_images_core()
    runtime_roles = {step["role"] for step in runtime_setup["steps"]}
    expected_roles = set(MODEL_MANIFEST)
    api_paths = {item["path"] for item in demo_script["api_checks"]}
    post_checks = [item for item in demo_script["api_checks"] if item["method"] == "POST"]
    post_samples_ready = all(item.get("sample_body") for item in post_checks)
    commands_use_public_base = all(
        PUBLIC_BASE_URL in item["curl"] and PUBLIC_BASE_URL in item["powershell"]
        for item in demo_script["api_checks"]
    )
    runtime_statuses = {
        runtime_status["reader_brain"]["status"],
        runtime_status["vision"]["status"],
        runtime_status["speech"]["status"],
        runtime_status["image_generation"]["status"],
    }
    runtime_status_ready = runtime_statuses <= {"online", "fallback-ready", "placeholder-ready"}
    checks = [
        {
            "id": "tiny_titan_budget",
            "label": "Tiny Titan model budget",
            "status": "pass" if model_budget["all_models_within_limit"] else "fail",
            "evidence": f"{len(model_budget['models'])} model roles are at or below {TINY_TITAN_LIMIT_B}B parameters.",
        },
        {
            "id": "award_targets",
            "label": "Targeted award evidence",
            "status": "pass" if len(AWARD_EVIDENCE) == 4 else "fail",
            "evidence": "Tiny Titan, Llama Champion, Off-Brand, and Field Notes evidence is exposed.",
        },
        {
            "id": "custom_frontend",
            "label": "Custom frontend",
            "status": "pass" if all((STATIC_DIR / name).exists() for name in ["index.html", "app.css", "app.js"]) else "fail",
            "evidence": "The app serves custom HTML, CSS, and JavaScript through Gradio Server.",
        },
        {
            "id": "runtime_setup",
            "label": "Runtime setup",
            "status": "pass" if runtime_roles == expected_roles else "fail",
            "evidence": "Runtime setup covers reader brain, speech, vision, and image generation paths.",
        },
        {
            "id": "runtime_status",
            "label": "Runtime status",
            "status": "pass" if runtime_status_ready else "fail",
            "evidence": "Runtime status labels every model path as online, fallback-ready, or placeholder-ready.",
        },
        {
            "id": "reader_accessibility",
            "label": "Reader accessibility",
            "status": "pass" if accessibility["all_passed"] else "fail",
            "evidence": f"{accessibility['passed_checks']} of {accessibility['total_checks']} accessibility checks pass.",
        },
        {
            "id": "image_receipts",
            "label": "Image receipts",
            "status": "pass"
            if all(
                item.get("generation_model") == MODEL_MANIFEST["image_generation"]["id"] and isinstance(item.get("seed"), int)
                for item in image_descriptions["descriptions"]
            )
            else "fail",
            "evidence": "Every article illustration exposes prompt, seed, model, asset URL, and fallback status.",
        },
        {
            "id": "demo_api_checks",
            "label": "Demo API checks",
            "status": "pass"
            if {
                "/api/model-budget",
                "/api/runtime-setup",
                "/api/accessibility-audit",
                "/api/image-descriptions",
                "/api/reader-brain",
                "/api/speak",
                "/api/generate-article",
            }.issubset(api_paths)
            and post_samples_ready
            else "fail",
            "evidence": "The judge runbook includes GET evidence checks and executable POST sample bodies.",
        },
        {
            "id": "command_base_url",
            "label": "Command base URL",
            "status": "pass" if commands_use_public_base else "fail",
            "evidence": f"Generated curl and curl.exe commands use {PUBLIC_BASE_URL}.",
        },
    ]
    return {
        "ok": True,
        "all_passed": all(check["status"] == "pass" for check in checks),
        "passed_checks": sum(1 for check in checks if check["status"] == "pass"),
        "total_checks": len(checks),
        "checks": checks,
    }


def evidence_bundle_core() -> dict[str, Any]:
    return {
        "ok": True,
        "title": "Tiny Narrator judge evidence bundle",
        "schema_version": "1.0",
        "generated_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
        "frontend": "custom Gradio Server HTML/CSS/JS",
        "public_base_url": PUBLIC_BASE_URL,
        "bonus_targets": ARTICLE_MANIFEST["bonus_targets"],
        "models": MODEL_MANIFEST,
        "award_evidence": {"ok": True, "items": AWARD_EVIDENCE},
        "model_budget": model_budget_core(),
        "runtime_setup": runtime_setup_core(),
        "runtime_status": _runtime_status_core(),
        "demo_script": demo_script_core(),
        "accessibility_audit": accessibility_audit_core(),
        "image_descriptions": describe_article_images_core(),
        "submission_readiness": submission_readiness_core(),
    }


def accessibility_audit_core() -> dict[str, Any]:
    checks = [
        {
            "id": "semantic_queue",
            "label": "Semantic reading queue",
            "status": "pass",
            "evidence": "Readable article nodes declare data-reader-type values and the reader controls navigate the ordered semantic node list.",
        },
        {
            "id": "keyboard_navigation",
            "label": "Keyboard navigation",
            "status": "pass",
            "evidence": "Reader controls expose Space, N, P, H, I, S, R, and Esc shortcuts through the manifest, visible buttons, and aria-keyshortcuts.",
        },
        {
            "id": "reader_cursor",
            "label": "Reader cursor state",
            "status": "pass",
            "evidence": "The active readable node receives focus, a visible outline, a stable reader id, and aria-current state.",
        },
        {
            "id": "shortcut_safety",
            "label": "Shortcut safety",
            "status": "pass",
            "evidence": "Global reader shortcuts ignore form controls so voice, speed, and auto-advance settings remain usable.",
        },
        {
            "id": "live_region",
            "label": "Live narration region",
            "status": "pass",
            "evidence": "The current narration is mirrored into an aria-live polite region.",
        },
        {
            "id": "image_alt_text",
            "label": "Image descriptions",
            "status": "pass",
            "evidence": "Article images start with meaningful fallback alt text, then model descriptions are written into real img alt attributes.",
        },
        {
            "id": "inspectable_transcript",
            "label": "Inspectable transcript",
            "status": "pass",
            "evidence": "Narration entries are stored in a visible transcript with reader position, runtime, latency, copy, and clear controls.",
        },
        {
            "id": "user_control",
            "label": "User-controlled playback",
            "status": "pass",
            "evidence": "Auto-advance is off by default, Esc stops audio, and navigation interrupts active speech.",
        },
        {
            "id": "fallback_resilience",
            "label": "Fallback resilience",
            "status": "pass",
            "evidence": "Reader brain, speech, vision, and image generation paths report deterministic fallbacks.",
        },
    ]
    return {
        "ok": True,
        "all_passed": all(check["status"] == "pass" for check in checks),
        "total_checks": len(checks),
        "passed_checks": sum(1 for check in checks if check["status"] == "pass"),
        "checks": checks,
    }


ARTICLE_MANIFEST: dict[str, Any] = {
    "title": "A tiny model reader that turns articles into guided narration",
    "reader_controls": [
        {"key": "Space", "aria_keyshortcuts": "Space", "action": "Play or pause"},
        {"key": "N", "aria_keyshortcuts": "N", "action": "Next item"},
        {"key": "P", "aria_keyshortcuts": "P", "action": "Previous item"},
        {"key": "H", "aria_keyshortcuts": "H", "action": "Next heading"},
        {"key": "I", "aria_keyshortcuts": "I", "action": "Next image"},
        {"key": "S", "aria_keyshortcuts": "S", "action": "Summarize current section"},
        {"key": "R", "aria_keyshortcuts": "R", "action": "Repeat current item"},
        {"key": "Esc", "aria_keyshortcuts": "Escape", "action": "Stop audio"},
    ],
    "bonus_targets": ["Tiny Titan", "Llama Champion", "Off-Brand", "Field Notes"],
    "images": ARTICLE_IMAGES,
    "models": MODEL_MANIFEST,
    "model_budget": model_budget_core(),
    "runtime_setup": runtime_setup_core(),
    "demo_script": demo_script_core(),
    "accessibility_audit": accessibility_audit_core(),
    "reader_settings": READER_SETTINGS,
    "award_evidence": AWARD_EVIDENCE,
}

app = Server(title="Tiny Narrator")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
app.mount("/outputs", StaticFiles(directory=OUTPUT_DIR), name="outputs")


class ReaderBrainRequest(BaseModel):
    node_type: str = "paragraph"
    text: str
    position: str | None = None
    mode: str = "narrate"


class ImageDescriptionRequest(BaseModel):
    image_id: str
    caption: str | None = None
    prompt: str | None = None
    image_url: str | None = None


class SpeechRequest(BaseModel):
    text: str
    voice: str = "af_heart"
    speed: float = 1.0


class ImageGenerationRequest(BaseModel):
    prompt: str
    seed: int | None = None


class ArticleGenerationRequest(BaseModel):
    topic: str


def _json(data: dict[str, Any], status_code: int = 200) -> JSONResponse:
    return JSONResponse(data, status_code=status_code)


def _elapsed_ms(start: float) -> int:
    return round((time.perf_counter() - start) * 1000)


def _validate_modal_klein_health(payload: dict[str, Any]) -> None:
    if payload.get("ok") is not True:
        raise ValueError("health check did not return ok")
    if payload.get("model") != MODEL_MANIFEST["image_generation"]["id"]:
        raise ValueError("health check returned unexpected model")
    if payload.get("runtime") != "modal-klein":
        raise ValueError("health check returned unexpected runtime")


def _modal_klein_base_url() -> str:
    normalized = KLEIN_MODAL_ENDPOINT.rstrip("/")
    for suffix in ("/generate", "/health"):
        if normalized.endswith(suffix):
            return normalized[: -len(suffix)]
    return normalized


def _llama_cpp_headers(extra: dict[str, str] | None = None) -> dict[str, str]:
    headers = dict(extra or {})
    if LLAMA_CPP_TOKEN:
        headers["Authorization"] = f"Bearer {LLAMA_CPP_TOKEN}"
    return headers


def _runtime_status_core() -> dict[str, Any]:
    start = time.perf_counter()
    llama_start = time.perf_counter()
    llama_status: dict[str, Any]
    if not LLAMA_CPP_BASE_URL:
        _runtime_log("llama.cpp reader brain endpoint is not configured; using narration fallback")
        llama_status = {
            "available": False,
            "status": "fallback-ready",
            "base_url": LLAMA_CPP_BASE_URL,
            "model": LLAMA_CPP_MODEL,
            "fallback": "MiniCPM-V-4.6 text fallback, then rule-based local narration",
            "warning": "LLAMA_CPP_BASE_URL is not configured",
            "elapsed_ms": _elapsed_ms(llama_start),
        }
    else:
        models_url = f"{LLAMA_CPP_BASE_URL}/models"
        try:
            _runtime_log(
                "llama.cpp reader brain health check "
                f"url={models_url} token={'configured' if LLAMA_CPP_TOKEN else 'not-set'} "
                "timeout=1.5s"
            )
            request = urllib.request.Request(
                models_url,
                headers=_llama_cpp_headers(),
                method="GET",
            )
            with urllib.request.urlopen(request, timeout=1.5) as response:
                payload = json.loads(response.read().decode("utf-8"))
            model_ids = [item.get("id", "") for item in payload.get("data", []) if isinstance(item, dict)]
            _runtime_log(
                "llama.cpp reader brain health response "
                f"models={model_ids[:6]} expected={LLAMA_CPP_MODEL}"
            )
            llama_status = {
                "available": True,
                "status": "online",
                "base_url": LLAMA_CPP_BASE_URL,
                "model": LLAMA_CPP_MODEL,
                "models": model_ids,
                "elapsed_ms": _elapsed_ms(llama_start),
            }
        except (OSError, urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
            detail = _http_exception_detail(exc)
            _runtime_log(f"llama.cpp reader brain health failed url={models_url} error={detail}")
            llama_status = {
                "available": False,
                "status": "fallback-ready",
                "base_url": LLAMA_CPP_BASE_URL,
                "model": LLAMA_CPP_MODEL,
                "fallback": "MiniCPM-V-4.6 text fallback, then rule-based local narration",
                "health_url": models_url,
                "warning": detail,
                "elapsed_ms": _elapsed_ms(llama_start),
            }

    kokoro_available = importlib.util.find_spec("kokoro") is not None
    soundfile_available = importlib.util.find_spec("soundfile") is not None

    klein_status: dict[str, Any]
    if KLEIN_MODAL_ENDPOINT:
        endpoint = _modal_klein_base_url()
        health_url = f"{endpoint}/health"
        try:
            health_headers: dict[str, str] = {}
            if KLEIN_MODAL_TOKEN:
                health_headers["Authorization"] = f"Bearer {KLEIN_MODAL_TOKEN}"
            _runtime_log(
                "Modal Klein health check "
                f"url={health_url} token={'configured' if KLEIN_MODAL_TOKEN else 'not-set'} "
                f"timeout={KLEIN_MODAL_HEALTH_TIMEOUT_SECONDS}s"
            )
            request = urllib.request.Request(
                health_url, method="GET",
                headers=health_headers,
            )
            with urllib.request.urlopen(request, timeout=KLEIN_MODAL_HEALTH_TIMEOUT_SECONDS) as response:
                payload = json.loads(response.read().decode("utf-8"))
            _runtime_log(
                "Modal Klein health response "
                f"model={payload.get('model')} runtime={payload.get('runtime')} ok={payload.get('ok')}"
            )
            _validate_modal_klein_health(payload)
            klein_status = {
                "available": True,
                "status": "online",
                "model": MODEL_MANIFEST["image_generation"]["id"],
                "endpoint": endpoint,
                "elapsed_ms": _elapsed_ms(start),
            }
        except (OSError, urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
            detail = _http_exception_detail(exc)
            _runtime_log(f"Modal Klein health failed url={health_url} error={detail}")
            klein_status = {
                "available": False,
                "status": "fallback-ready",
                "model": MODEL_MANIFEST["image_generation"]["id"],
                "endpoint": endpoint,
                "health_url": health_url,
                "fallback": "bundled generated article assets",
                "warning": detail,
                "elapsed_ms": _elapsed_ms(start),
            }
    else:
        _runtime_log("Modal Klein endpoint is not configured; using image fallback")
        klein_status = {
            "available": False,
            "status": "fallback-ready",
            "model": MODEL_MANIFEST["image_generation"]["id"],
            "fallback": "bundled generated article assets",
        }

    vision_status = _vision_runtime_status()

    return {
        "ok": True,
        "reader_brain": llama_status,
        "vision": vision_status,
        "speech": {
            "available": kokoro_available and soundfile_available,
            "status": "online" if kokoro_available and soundfile_available else "fallback-ready",
            "model": "hexgrad/Kokoro-82M",
            "kokoro_installed": kokoro_available,
            "soundfile_installed": soundfile_available,
            "fallback": "browser speech plus transcript",
        },
        "image_generation": klein_status,
        "elapsed_ms": _elapsed_ms(start),
    }


def _compact_text(text: str, limit: int = 220) -> str:
    normalized = " ".join(text.split())
    if len(normalized) <= limit:
        return normalized
    return f"{normalized[: limit - 1].rstrip()}."


def _topic_seed(topic: str) -> int:
    return (sum(ord(char) for char in topic) % 997) + 1


def _fallback_article(topic: str) -> dict[str, Any]:
    clean_topic = _compact_text(topic, 80).rstrip(".") or "accessible technology"
    title = f"A tiny guide to {clean_topic}"
    return {
        "title": title,
        "dek": (
            f"This generated article introduces {clean_topic} with a short, screen-reader-friendly structure "
            "and practical examples."
        ),
        "sections": [
            {
                "heading": f"Why {clean_topic} matters",
                "body": (
                    f"{clean_topic.capitalize()} is easier to understand when the page explains the main idea first, "
                    "then moves through examples in a predictable order. A longer article gives the reader enough "
                    "context to pause, repeat, and compare sections without feeling rushed."
                ),
            },
            {
                "heading": "How a tiny narrator helps",
                "body": (
                    "A small reader model can turn each heading, paragraph, and image caption into concise narration "
                    "without turning the article into a chatbot. It keeps the experience close to the article, so "
                    "the user stays oriented while moving through the page."
                ),
            },
            {
                "heading": "What image descriptions add",
                "body": (
                    "Images often carry examples, diagrams, or emotional context. A visual descriptor gives those "
                    "details a spoken form, helping readers understand why an illustration belongs with the surrounding text."
                ),
            },
            {
                "heading": "Designing for repeated listening",
                "body": (
                    "Reader controls should make it easy to move backward, skip to the next heading, replay a sentence, "
                    "or stop speech completely. Those small controls matter when someone is studying, teaching, or checking details."
                ),
            },
            {
                "heading": "What to try next",
                "body": (
                    "Use the reader controls to move by heading or image, then compare the transcript with the article "
                    "to check whether the generated structure is easy to follow. If a section sounds confusing, rewrite it "
                    "with simpler language and a clearer heading."
                ),
            },
        ],
    }


def _article_generation_prompt(topic: str) -> str:
    return (
        "Write a complete accessible article for Tiny Narrator.\n"
        f"Topic: {topic}\n\n"
        "Return strict JSON with keys title, dek, and sections. "
        "sections must contain exactly five objects with heading and body. "
        "Each body should be 45 to 70 words, with concrete examples and practical detail. "
        "Use clear plain language suitable for screen-reader narration. Do not include markdown."
    )


def _fallback_narration(node_type: str, text: str, mode: str) -> str:
    prefix = {
        "heading": "Heading. ",
        "image": "Image description. ",
        "button": "Control. ",
        "quote": "Quote. ",
    }.get(node_type, "")
    if mode == "summarize":
        prefix = "Summary. "
    fallback_text = _compact_text(text) or "No readable text is available for this item."
    return f"{prefix}{fallback_text}".strip()


def _image_narration(text: str) -> str:
    spoken_text = _compact_text(text)
    if not spoken_text:
        spoken_text = "No image description is available."
    lowered = spoken_text.lower()
    if lowered.startswith(("image description.", "image.", "graphic.", "photo.", "illustration.")):
        return spoken_text
    return f"Image description. {spoken_text}"


def _chat_message_text(payload: dict[str, Any]) -> str:
    choices = payload.get("choices", [])
    if not choices or not isinstance(choices[0], dict):
        return ""
    choice = choices[0]
    message = choice.get("message")
    if isinstance(message, dict):
        for key in ("content", "reasoning_content", "reasoning", "text"):
            value = message.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
            if isinstance(value, list):
                parts = []
                for item in value:
                    if isinstance(item, dict):
                        part = item.get("text") or item.get("content")
                        if isinstance(part, str):
                            parts.append(part)
                    elif isinstance(item, str):
                        parts.append(item)
                combined = " ".join(parts).strip()
                if combined:
                    return combined
    text = choice.get("text")
    if isinstance(text, str) and text.strip():
        return text.strip()
    return ""


def _chat_response_debug(payload: dict[str, Any]) -> str:
    choices = payload.get("choices", [])
    if not choices or not isinstance(choices[0], dict):
        return f"top_keys={list(payload)[:8]} choices=missing"
    choice = choices[0]
    message = choice.get("message")
    message_keys = list(message)[:8] if isinstance(message, dict) else []
    return (
        f"top_keys={list(payload)[:8]} "
        f"choice_keys={list(choice)[:8]} "
        f"message_keys={message_keys} "
        f"finish_reason={choice.get('finish_reason')}"
    )


def _reader_brain_prompt(node_type: str, text: str, position: str | None, mode: str) -> str:
    if mode == "summarize":
        return (
            "Summarize this article section for screen-reader navigation.\n"
            f"Node type: {node_type}\n"
            f"Position: {position or 'unknown'}\n"
            f"Content: {text}\n\n"
            "Rules: start with 'Summary.', use one or two short sentences, explain what the "
            "reader can learn in this section, and never mention implementation details. "
            "Return only the final spoken narration. Do not explain your reasoning."
        )
    image_rule = (
        "If node type is image, start exactly with 'Image description.' and then describe the visible content. "
        if node_type == "image"
        else ""
    )
    return (
        "Convert this article node into concise screen-reader narration.\n"
        f"Mode: {mode}\n"
        f"Node type: {node_type}\n"
        f"Position: {position or 'unknown'}\n"
        f"Content: {text}\n\n"
        f"Rules: {image_rule}announce the node type only when it helps orientation, keep prose short, "
        "and never mention implementation details. Return only the final spoken narration. "
        "Do not explain your reasoning."
    )


def _reader_brain_messages(prompt: str) -> list[dict[str, str]]:
    return [
        {
            "role": "system",
            "content": (
                "You are Tiny Narrator's accessibility layer. "
                "Reasoning mode: off. Do not generate a reasoning trace. "
                "Do not think step by step. "
                "Produce clear screen-reader narration for article content. "
                "Return only the exact text to speak. Do not include analysis, alternatives, "
                "markdown, labels, or explanations."
            ),
        },
        {"role": "user", "content": prompt},
    ]


def _reader_brain_body(model: str, prompt: str) -> bytes:
    return json.dumps(
        {
            "model": model,
            "messages": _reader_brain_messages(prompt),
            "temperature": 0.2,
            "top_p": 0.9,
            "stream": False,
            "max_tokens": 80,
        }
    ).encode("utf-8")


def _polish_reader_narration(node_type: str, mode: str, narration: str) -> str:
    narration = _compact_text(narration, 220)
    if node_type == "image" and mode != "summarize":
        return _image_narration(narration)
    return narration


def _rule_reader_fallback(
    node_type: str,
    text: str,
    mode: str,
    start: float,
    warning: str,
) -> dict[str, Any]:
    return {
        "ok": True,
        "runtime": "fallback",
        "model": "rule-based local fallback",
        "warning": warning,
        "narration": _fallback_narration(node_type, text, mode),
        "elapsed_ms": _elapsed_ms(start),
    }


def _minicpm_reader_brain_fallback(
    prompt: str,
    node_type: str,
    text: str,
    mode: str,
    start: float,
    primary_warning: str,
) -> dict[str, Any] | None:
    if not MINICPM_VISION_BASE_URL or not MINICPM_VISION_API_KEY:
        return None
    try:
        request = urllib.request.Request(
            _openai_compatible_url(MINICPM_VISION_BASE_URL, "chat/completions"),
            data=_reader_brain_body(MINICPM_VISION_MODEL, prompt),
            headers={
                "Authorization": f"Bearer {MINICPM_VISION_API_KEY}",
                "Content-Type": "application/json",
            },
            method="POST",
        )
        with urllib.request.urlopen(request, timeout=MINICPM_VISION_TIMEOUT_SECONDS) as response:
            payload = json.loads(response.read().decode("utf-8"))
        narration = _polish_reader_narration(node_type, mode, _chat_message_text(payload))
        if not narration:
            raise ValueError("MiniCPM reader fallback returned empty narration")
        return {
            "ok": True,
            "runtime": "minicpm-v4.6-fallback",
            "model": MINICPM_VISION_MODEL,
            "warning": primary_warning,
            "narration": narration,
            "elapsed_ms": _elapsed_ms(start),
        }
    except (OSError, urllib.error.URLError, TimeoutError, ValueError, KeyError, json.JSONDecodeError) as exc:
        detail = _http_exception_detail(exc)
        _runtime_log(f"MiniCPM reader-brain fallback failed error={detail}")
        return _rule_reader_fallback(
            node_type,
            text,
            mode,
            start,
            f"{primary_warning}; MiniCPM reader fallback unavailable: {detail}",
        )


def reader_brain_core(node_type: str, text: str, position: str | None, mode: str) -> dict[str, Any]:
    start = time.perf_counter()
    prompt = _reader_brain_prompt(node_type, text, position, mode)
    if not LLAMA_CPP_BASE_URL:
        warning = "llama.cpp unavailable: LLAMA_CPP_BASE_URL is not configured"
        minicpm_fallback = _minicpm_reader_brain_fallback(
            prompt,
            node_type,
            text,
            mode,
            start,
            warning,
        )
        if minicpm_fallback:
            return minicpm_fallback
        return _rule_reader_fallback(node_type, text, mode, start, warning)

    try:
        request = urllib.request.Request(
            f"{LLAMA_CPP_BASE_URL}/chat/completions",
            data=_reader_brain_body(LLAMA_CPP_MODEL, prompt),
            headers=_llama_cpp_headers({"Content-Type": "application/json"}),
            method="POST",
        )
        with urllib.request.urlopen(request, timeout=LLAMA_CPP_TIMEOUT_SECONDS) as response:
            payload = json.loads(response.read().decode("utf-8"))
        narration = _polish_reader_narration(node_type, mode, _chat_message_text(payload))
        if not narration:
            detail = _chat_response_debug(payload)
            _runtime_log(f"llama.cpp returned empty narration; response_shape={detail}")
            warning = f"llama.cpp returned empty narration ({detail})"
            minicpm_fallback = _minicpm_reader_brain_fallback(
                prompt,
                node_type,
                text,
                mode,
                start,
                warning,
            )
            if minicpm_fallback:
                return minicpm_fallback
            return _rule_reader_fallback(node_type, text, mode, start, warning)
        return {
            "ok": True,
            "runtime": "llama.cpp",
            "model": LLAMA_CPP_MODEL,
            "narration": narration,
            "elapsed_ms": _elapsed_ms(start),
        }
    except (OSError, urllib.error.URLError, TimeoutError, ValueError, KeyError, json.JSONDecodeError) as exc:
        warning = f"llama.cpp unavailable: {_http_exception_detail(exc)}"
        minicpm_fallback = _minicpm_reader_brain_fallback(
            prompt,
            node_type,
            text,
            mode,
            start,
            warning,
        )
        if minicpm_fallback:
            return minicpm_fallback
        return _rule_reader_fallback(node_type, text, mode, start, warning)


def _openai_compatible_url(base_url: str, path: str) -> str:
    normalized = base_url.rstrip("/")
    if normalized.endswith("/v1/chat/completions"):
        root = normalized[: -len("/chat/completions")]
    elif normalized.endswith("/v1"):
        root = normalized
    else:
        root = f"{normalized}/v1"
    return f"{root}/{path.lstrip('/')}"


def _article_image_by_id(image_id: str) -> dict[str, Any] | None:
    return next((image for image in ARTICLE_IMAGES if image["id"] == image_id), None)


def _local_image_data_url(image_path: str) -> str | None:
    if image_path.startswith(("http://", "https://", "data:")):
        return None
    normalized = image_path.lstrip("/")
    local_path = (ROOT / normalized).resolve()
    try:
        local_path.relative_to(ROOT.resolve())
    except ValueError:
        return None
    mime_type, _ = mimetypes.guess_type(local_path.name)
    if mime_type not in {"image/png", "image/jpeg", "image/webp"}:
        return None
    if not local_path.exists():
        return None
    encoded = base64.b64encode(local_path.read_bytes()).decode("ascii")
    return f"data:{mime_type};base64,{encoded}"


def _absolute_image_url(image_url: str | None, image_id: str | None = None) -> str | None:
    candidate = image_url
    if not candidate and image_id:
        image = _article_image_by_id(image_id)
        candidate = (image.get("vision_asset_url") or image.get("asset_url")) if image else None
    if not candidate:
        return None
    if candidate.startswith(("http://", "https://", "data:")):
        return candidate
    data_url = _local_image_data_url(candidate)
    if data_url:
        return data_url
    if candidate.startswith("/"):
        return f"{PUBLIC_BASE_URL}{candidate}"
    return f"{PUBLIC_BASE_URL}/{candidate}"


def _vision_fallback_text(image_id: str, caption: str | None, prompt: str | None) -> str:
    descriptions = {
        "desk-reader": (
            "A person reads a long article on a laptop while an accessibility toolbar "
            "highlights the current paragraph."
        ),
        "model-map": (
            "A compact diagram showing four small AI models working together: vision, "
            "reader brain, speech, and image generation."
        ),
    }
    return descriptions.get(
        image_id,
        caption or prompt or "A generated article image awaiting model description.",
    )


def _clean_alt_text(text: str, limit: int = 260) -> str:
    return _compact_text(" ".join(text.split()), limit)


def _minicpm_vision_prompt(caption: str | None, prompt: str | None) -> str:
    context = " ".join(part for part in [caption, prompt] if part)
    context_line = f"\nContext: {context}" if context else ""
    return (
        "Describe this image for a screen reader in one or two concise sentences. "
        "Mention visible content, layout, important text, and purpose when relevant. "
        "Do not guess hidden intent. Do not mention models, implementation details, or that you are an AI. "
        "Return plain text only."
        f"{context_line}"
    )


def _call_minicpm_vision(image_url: str, caption: str | None, prompt: str | None) -> dict[str, Any]:
    start = time.perf_counter()
    body = json.dumps(
        {
            "model": MINICPM_VISION_MODEL,
            "messages": [
                {
                    "role": "user",
                    "content": [
                        {"type": "text", "text": _minicpm_vision_prompt(caption, prompt)},
                        {"type": "image_url", "image_url": {"url": image_url}},
                    ],
                }
            ],
            "temperature": 0.1,
            "max_tokens": 160,
        }
    ).encode("utf-8")
    request = urllib.request.Request(
        _openai_compatible_url(MINICPM_VISION_BASE_URL, "chat/completions"),
        data=body,
        headers={
            "Authorization": f"Bearer {MINICPM_VISION_API_KEY}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=MINICPM_VISION_TIMEOUT_SECONDS) as response:
        payload = json.loads(response.read().decode("utf-8"))
    content = payload["choices"][0]["message"]["content"]
    if not isinstance(content, str) or not content.strip():
        raise ValueError("MiniCPM vision response did not include text content")
    return {
        "runtime": "minicpm-v4.6",
        "model": MINICPM_VISION_MODEL,
        "alt_text": _clean_alt_text(content),
        "elapsed_ms": _elapsed_ms(start),
    }


def _check_minicpm_chat_ready() -> dict[str, Any]:
    body = json.dumps(
        {
            "model": MINICPM_VISION_MODEL,
            "messages": [{"role": "user", "content": "Reply with ready."}],
            "temperature": 0,
            "max_tokens": 8,
        }
    ).encode("utf-8")
    request = urllib.request.Request(
        _openai_compatible_url(MINICPM_VISION_BASE_URL, "chat/completions"),
        data=body,
        headers={
            "Authorization": f"Bearer {MINICPM_VISION_API_KEY}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=min(MINICPM_VISION_TIMEOUT_SECONDS, 10)) as response:
        payload = json.loads(response.read().decode("utf-8"))
    content = payload["choices"][0]["message"]["content"]
    if not isinstance(content, str) or not content.strip():
        raise ValueError("MiniCPM readiness response did not include text content")
    return payload


def _vision_runtime_status() -> dict[str, Any]:
    if not MINICPM_VISION_BASE_URL or not MINICPM_VISION_API_KEY:
        return {
            "available": False,
            "status": "fallback-ready",
            "model": MINICPM_VISION_MODEL,
            "configured": False,
            "fallback": "cached deterministic alt text",
        }
    start = time.perf_counter()
    model_ids: list[str] = []
    models_warning: str | None = None
    try:
        request = urllib.request.Request(
            _openai_compatible_url(MINICPM_VISION_BASE_URL, "models"),
            headers={"Authorization": f"Bearer {MINICPM_VISION_API_KEY}"},
            method="GET",
        )
        with urllib.request.urlopen(request, timeout=5) as response:
            payload = json.loads(response.read().decode("utf-8"))
        model_ids = [
            item.get("id", "")
            for item in payload.get("data", [])
            if isinstance(item, dict)
        ]
        if model_ids and MINICPM_VISION_MODEL not in model_ids:
            models_warning = "MiniCPM vision model was not listed by /models"
            _check_minicpm_chat_ready()
        return {
            "available": True,
            "status": "online",
            "model": MINICPM_VISION_MODEL,
            "configured": True,
            "base_url": MINICPM_VISION_BASE_URL,
            "models": model_ids,
            "warning": models_warning,
            "elapsed_ms": _elapsed_ms(start),
        }
    except (OSError, urllib.error.URLError, TimeoutError, ValueError, json.JSONDecodeError) as exc:
        models_warning = exc.__class__.__name__

    try:
        _check_minicpm_chat_ready()
        return {
            "available": True,
            "status": "online",
            "model": MINICPM_VISION_MODEL,
            "configured": True,
            "base_url": MINICPM_VISION_BASE_URL,
            "models": model_ids,
            "warning": f"/models unavailable, chat completions ready: {models_warning}",
            "elapsed_ms": _elapsed_ms(start),
        }
    except (OSError, urllib.error.URLError, TimeoutError, KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as exc:
        return {
            "available": False,
            "status": "fallback-ready",
            "model": MINICPM_VISION_MODEL,
            "configured": True,
            "base_url": MINICPM_VISION_BASE_URL,
            "fallback": "cached deterministic alt text",
            "warning": exc.__class__.__name__,
            "elapsed_ms": _elapsed_ms(start),
        }


def describe_image_core(
    image_id: str,
    caption: str | None,
    prompt: str | None,
    image_url: str | None = None,
) -> dict[str, Any]:
    start = time.perf_counter()
    resolved_image_url = _absolute_image_url(image_url, image_id)
    alt_text = _vision_fallback_text(image_id, caption, prompt)
    warning = None
    if MINICPM_VISION_BASE_URL and MINICPM_VISION_API_KEY and resolved_image_url:
        try:
            return {"ok": True, **_call_minicpm_vision(resolved_image_url, caption, prompt)}
        except (OSError, urllib.error.URLError, TimeoutError, KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as exc:
            detail = _http_exception_detail(exc)
            _runtime_log(f"MiniCPM vision failed image_url={resolved_image_url} error={detail}")
            warning = f"MiniCPM vision unavailable: {detail}"
    return {
        "ok": True,
        "runtime": "fallback",
        "model": MINICPM_VISION_MODEL,
        "alt_text": alt_text,
        "warning": warning,
        "elapsed_ms": _elapsed_ms(start),
    }


def describe_article_images_core() -> dict[str, Any]:
    start = time.perf_counter()
    descriptions = []
    for image in ARTICLE_IMAGES:
        description = describe_image_core(
            image["id"],
            caption=image.get("caption"),
            prompt=image.get("prompt"),
            image_url=image.get("vision_asset_url") or image.get("asset_url"),
        )
        descriptions.append({**image, **description})
    runtimes = {item["runtime"] for item in descriptions}
    return {
        "ok": True,
        "runtime": runtimes.pop() if len(runtimes) == 1 else "mixed",
        "model": MINICPM_VISION_MODEL,
        "descriptions": descriptions,
        "elapsed_ms": _elapsed_ms(start),
    }


def _silent_wav(path: Path, seconds: float = 0.35, sample_rate: int = 24000) -> None:
    frames = int(seconds * sample_rate)
    with wave.open(str(path), "wb") as wav:
        wav.setnchannels(1)
        wav.setsampwidth(2)
        wav.setframerate(sample_rate)
        wav.writeframes(b"\x00\x00" * frames)


def _prune_speech_outputs(keep_path: Path, max_files: int = 24) -> None:
    speech_files = sorted(
        OUTPUT_DIR.glob("speech*.wav"),
        key=lambda path: path.stat().st_mtime,
        reverse=True,
    )
    keep_resolved = keep_path.resolve()
    for old_path in speech_files[max_files:]:
        if old_path.resolve() == keep_resolved:
            continue
        try:
            old_path.unlink()
        except OSError:
            pass


def speak_core(text: str, voice: str, speed: float) -> dict[str, Any]:
    start = time.perf_counter()
    clean_text = _compact_text(text, 1200)
    if not clean_text:
        output_path = OUTPUT_DIR / f"speech-fallback-{uuid4().hex}.wav"
        _silent_wav(output_path)
        _prune_speech_outputs(output_path)
        return {
            "ok": True,
            "runtime": "fallback",
            "model": "hexgrad/Kokoro-82M",
            "warning": "Kokoro skipped because transcript was empty",
            "audio_url": f"/outputs/{output_path.name}",
            "transcript": "",
            "elapsed_ms": _elapsed_ms(start),
        }
    try:
        from kokoro import KPipeline
        import soundfile as sf

        pipeline = KPipeline(lang_code="a")
        generator = pipeline(clean_text, voice=voice, speed=speed)
        _, _, audio = next(generator)
        output_path = OUTPUT_DIR / f"speech-{uuid4().hex}.wav"
        sf.write(output_path, audio, 24000)
        runtime = "kokoro"
        warning = None
    except Exception as exc:
        output_path = OUTPUT_DIR / f"speech-fallback-{uuid4().hex}.wav"
        _silent_wav(output_path)
        runtime = "fallback"
        warning = f"Kokoro unavailable: {exc.__class__.__name__}"

    _prune_speech_outputs(output_path)

    return {
        "ok": True,
        "runtime": runtime,
        "model": "hexgrad/Kokoro-82M",
        "warning": warning,
        "audio_url": f"/outputs/{output_path.name}",
        "transcript": clean_text,
        "elapsed_ms": _elapsed_ms(start),
    }


def _call_modal_klein(prompt: str, seed: int | None) -> dict[str, Any]:
    """Call the Modal Klein worker for live image generation."""
    start = time.perf_counter()
    body = json.dumps({"prompt": prompt, "seed": seed}).encode("utf-8")
    headers: dict[str, str] = {"Content-Type": "application/json"}
    if KLEIN_MODAL_TOKEN:
        headers["Authorization"] = f"Bearer {KLEIN_MODAL_TOKEN}"
    endpoint = _modal_klein_base_url()
    generate_url = f"{endpoint}/generate"
    _runtime_log(
        "Modal Klein generate request "
        f"url={generate_url} token={'configured' if KLEIN_MODAL_TOKEN else 'not-set'} seed={seed}"
    )
    request = urllib.request.Request(
        generate_url,
        data=body,
        headers=headers,
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=KLEIN_MODAL_TIMEOUT_SECONDS) as response:
            payload = json.loads(response.read().decode("utf-8"))
    except (OSError, urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
        detail = _http_exception_detail(exc)
        _runtime_log(f"Modal Klein generate failed url={generate_url} error={detail}")
        raise
    if payload.get("ok") is not True:
        _runtime_log(f"Modal Klein generate returned ok=false payload={_compact_text(json.dumps(payload), 400)}")
        raise ValueError(str(payload.get("error") or payload.get("detail") or "Modal Klein returned ok=false"))
    if payload.get("model") != MODEL_MANIFEST["image_generation"]["id"]:
        _runtime_log(f"Modal Klein generate returned unexpected model={payload.get('model')}")
        raise ValueError("Modal Klein returned unexpected model")
    if payload.get("runtime") != "modal-klein":
        _runtime_log(f"Modal Klein generate returned unexpected runtime={payload.get('runtime')}")
        raise ValueError("Modal Klein returned unexpected runtime")
    image_url = payload.get("image_url", "")
    if not isinstance(image_url, str) or not image_url:
        _runtime_log("Modal Klein generate response did not include image_url")
        raise ValueError("Modal Klein response did not include image_url")
    if image_url and image_url.startswith("/media/"):
        image_url = f"{endpoint}{image_url}"
    _runtime_log(f"Modal Klein generate succeeded runtime={payload.get('runtime')} image_url={image_url}")
    return {
        "ok": True,
        "runtime": payload.get("runtime", "modal-klein"),
        "model": payload.get("model", MODEL_MANIFEST["image_generation"]["id"]),
        "image_url": image_url,
        "prompt": payload.get("prompt", prompt),
        "seed": payload.get("seed", seed),
        "elapsed_ms": payload.get("elapsed_ms", _elapsed_ms(start)),
    }


def _fallback_image(prompt: str, seed: int | None, warning: str | None = None) -> dict[str, Any]:
    """Return bundled fallback SVG assets when Modal is unavailable."""
    start = time.perf_counter()
    return {
        "ok": True,
        "runtime": "fallback",
        "model": MODEL_MANIFEST["image_generation"]["id"],
        "image_url": f"/static/generated/{'desk-reader.svg' if seed and seed % 2 else 'model-map.svg'}",
        "prompt": prompt,
        "seed": seed,
        "warning": warning,
        "elapsed_ms": _elapsed_ms(start),
    }


def generate_image_core(prompt: str, seed: int | None) -> dict[str, Any]:
    if KLEIN_MODAL_ENDPOINT:
        try:
            return _call_modal_klein(prompt, seed)
        except (OSError, urllib.error.URLError, TimeoutError, ValueError, KeyError, json.JSONDecodeError) as exc:
            return _fallback_image(prompt, seed, warning=f"Modal Klein unavailable: {_http_exception_detail(exc)}")
    _runtime_log("Modal Klein generate skipped because KLEIN_MODAL_ENDPOINT is not configured")
    return _fallback_image(prompt, seed)


def _thumbnail_image_prompt(topic: str) -> str:
    return (
        f"Square image-only editorial thumbnail about {topic}. "
        "Create one centered cohesive scene: a friendly assistive robot or small AI helper guiding a reader "
        "through visual information with light, sound waves, image cards, and accessibility cues. "
        "Fill most of the frame with the subject while leaving a small clean margin. "
        "Modern polished 3D editorial illustration, soft gradients, warm blue and teal accents, clear focal point. "
        "No words, no letters, no numbers, no captions, no title, no logo, no watermark, "
        "no user interface, no screenshot, no document page, no poster layout, no split panels."
    )


def generate_article_core(topic: str) -> dict[str, Any]:
    start = time.perf_counter()
    clean_topic = _compact_text(topic, 100).strip()
    if not clean_topic:
        clean_topic = "accessible technology"

    article = _fallback_article(clean_topic)
    runtime = "fallback"
    warning = None
    if not LLAMA_CPP_BASE_URL:
        warning = "llama.cpp unavailable: LLAMA_CPP_BASE_URL is not configured"
    else:
        try:
            body = json.dumps(
                {
                    "model": LLAMA_CPP_MODEL,
                    "messages": [
                        {
                            "role": "system",
                            "content": (
                                "You are Tiny Narrator's article generator. "
                                "Create concise, semantic, accessible article drafts."
                            ),
                        },
                        {"role": "user", "content": _article_generation_prompt(clean_topic)},
                    ],
                    "temperature": 0.35,
                    "max_tokens": 650,
                }
            ).encode("utf-8")
            request = urllib.request.Request(
                f"{LLAMA_CPP_BASE_URL}/chat/completions",
                data=body,
                headers=_llama_cpp_headers({"Content-Type": "application/json"}),
                method="POST",
            )
            with urllib.request.urlopen(request, timeout=LLAMA_CPP_TIMEOUT_SECONDS) as response:
                payload = json.loads(response.read().decode("utf-8"))
            generated = json.loads(payload["choices"][0]["message"]["content"].strip())
            sections = generated.get("sections", [])
            if (
                isinstance(generated.get("title"), str)
                and isinstance(generated.get("dek"), str)
                and isinstance(sections, list)
                and len(sections) == 5
                and all(isinstance(item.get("heading"), str) and isinstance(item.get("body"), str) for item in sections)
            ):
                article = {
                    "title": _compact_text(generated["title"], 90),
                    "dek": _compact_text(generated["dek"], 180),
                    "sections": [
                        {"heading": _compact_text(item["heading"], 80), "body": _compact_text(item["body"], 720)}
                        for item in sections
                    ],
                }
                runtime = "llama.cpp"
        except (OSError, urllib.error.URLError, TimeoutError, ValueError, KeyError, json.JSONDecodeError) as exc:
            warning = f"llama.cpp unavailable: {exc.__class__.__name__}"

    image_prompt = _thumbnail_image_prompt(clean_topic)
    thumbnail = generate_image_core(image_prompt, seed=_topic_seed(clean_topic))
    return {
        "ok": True,
        "topic": clean_topic,
        "runtime": runtime,
        "model": MODEL_MANIFEST["reader_brain"]["id"],
        "warning": warning,
        "article": article,
        "thumbnail": {
            **thumbnail,
            "role": "thumbnail",
            "generation_model": MODEL_MANIFEST["image_generation"]["id"],
        },
        "elapsed_ms": _elapsed_ms(start),
    }


@app.get("/", response_class=HTMLResponse)
async def home() -> str:
    return (STATIC_DIR / "index.html").read_text(encoding="utf-8")


@app.get("/generate", response_class=HTMLResponse)
async def generate_page() -> str:
    return (STATIC_DIR / "generate.html").read_text(encoding="utf-8")


@app.get("/api/health")
async def health() -> JSONResponse:
    return _json(
        {
            "ok": True,
            "app": "Tiny Narrator",
            "frontend": "custom Gradio Server HTML/CSS/JS",
            "llama_cpp_base_url": LLAMA_CPP_BASE_URL,
            "public_base_url": PUBLIC_BASE_URL,
            "models": MODEL_MANIFEST,
        }
    )


@app.get("/api/article-manifest")
async def article_manifest() -> JSONResponse:
    return _json({"ok": True, **ARTICLE_MANIFEST})


@app.get("/api/award-evidence")
async def award_evidence() -> JSONResponse:
    return _json({"ok": True, "items": AWARD_EVIDENCE})


@app.get("/api/model-budget")
async def model_budget() -> JSONResponse:
    return _json(model_budget_core())


@app.get("/api/runtime-setup")
async def runtime_setup() -> JSONResponse:
    return _json(runtime_setup_core())


@app.get("/api/demo-script")
async def demo_script() -> JSONResponse:
    return _json(demo_script_core())


@app.get("/api/accessibility-audit")
async def accessibility_audit() -> JSONResponse:
    return _json(accessibility_audit_core())


@app.get("/api/submission-readiness")
async def submission_readiness() -> JSONResponse:
    return _json(submission_readiness_core())


@app.get("/api/evidence-bundle")
async def evidence_bundle() -> JSONResponse:
    return _json(evidence_bundle_core())


@app.get("/api/runtime-status")
async def runtime_status() -> JSONResponse:
    return _json(_runtime_status_core())


@app.post("/api/reader-brain")
async def reader_brain_endpoint(payload: ReaderBrainRequest) -> JSONResponse:
    return _json(reader_brain_core(payload.node_type, payload.text, payload.position, payload.mode))


@app.post("/api/describe-image")
async def describe_image_endpoint(payload: ImageDescriptionRequest) -> JSONResponse:
    return _json(describe_image_core(payload.image_id, payload.caption, payload.prompt, payload.image_url))


@app.get("/api/image-descriptions")
async def image_descriptions_endpoint() -> JSONResponse:
    return _json(describe_article_images_core())


@app.post("/api/speak")
async def speak_endpoint(payload: SpeechRequest) -> JSONResponse:
    return _json(speak_core(payload.text, payload.voice, payload.speed))


@app.post("/api/generate-image")
async def generate_image_endpoint(payload: ImageGenerationRequest) -> JSONResponse:
    return _json(generate_image_core(payload.prompt, payload.seed))


@app.post("/api/generate-article")
async def generate_article_endpoint(payload: ArticleGenerationRequest) -> JSONResponse:
    return _json(generate_article_core(payload.topic))


@app.api(name="reader_brain")
def reader_brain_api(node_type: str, text: str, position: str = "", mode: str = "narrate") -> str:
    return json.dumps(reader_brain_core(node_type, text, position, mode))


@app.api(name="describe_image")
def describe_image_api(image_id: str, caption: str = "", prompt: str = "", image_url: str = "") -> str:
    return json.dumps(describe_image_core(image_id, caption, prompt, image_url))


@app.api(name="describe_article_images")
def describe_article_images_api() -> str:
    return json.dumps(describe_article_images_core())


@app.api(name="model_budget")
def model_budget_api() -> str:
    return json.dumps(model_budget_core())


@app.api(name="runtime_setup")
def runtime_setup_api() -> str:
    return json.dumps(runtime_setup_core())


@app.api(name="demo_script")
def demo_script_api() -> str:
    return json.dumps(demo_script_core())


@app.api(name="accessibility_audit")
def accessibility_audit_api() -> str:
    return json.dumps(accessibility_audit_core())


@app.api(name="submission_readiness")
def submission_readiness_api() -> str:
    return json.dumps(submission_readiness_core())


@app.api(name="evidence_bundle")
def evidence_bundle_api() -> str:
    return json.dumps(evidence_bundle_core())


@app.api(name="speak")
def speak_api(text: str, voice: str = "af_heart", speed: float = 1.0) -> str:
    return json.dumps(speak_core(text, voice, speed))


@app.api(name="generate_image")
def generate_image_api(prompt: str, seed: int | None = None) -> str:
    return json.dumps(generate_image_core(prompt, seed))


@app.api(name="generate_article")
def generate_article_api(topic: str) -> str:
    return json.dumps(generate_article_core(topic))


@app.exception_handler(Exception)
async def handle_exception(_: Request, exc: Exception) -> JSONResponse:
    return _json({"ok": False, "error": exc.__class__.__name__, "detail": str(exc)}, status_code=500)


if __name__ == "__main__":
    app.launch(server_name=GRADIO_SERVER_NAME, server_port=GRADIO_SERVER_PORT, share=GRADIO_SHARE)