app.py

"""PAREIDOLIA — Space entrypoint. Everything secretly has a face.

Wiring (ARCHITECTURE.md is the contract; the godseed skeleton is the proven base):

  Menagerie wall (web/)     <--  GET /                      static, zero GPU
  the séance                <--  @app.api(name="awaken")    gradio endpoint via
                                 @gradio/client — the VISITOR's browser-authenticated
                                 request pays the ZeroGPU quota
                                 (POST /api/awaken is its mock/dev REST twin,
                                  not registered on the zerogpu backend)
  publish to the wall       <--  POST /api/menagerie        {record_token}; rate-limited,
                                 gate re-checked server-side
  the wall, paginated       <--  GET  /api/menagerie        newest first
  one soul, by id           <--  GET  /api/menagerie/{id}   the share page's data twin
  the share page            <--  GET  /o/{id}               HTML + OG meta; composite
                                 card at /media/{id}_og.jpg (made at publish time)
  live wall updates         <--  GET  /api/stream           SSE 'awakened' broadcasts
  persisted media           <--  GET  /media/{id}.(jpg|wav) id->path map only
                                 (+ {id}_mutter.wav, {id}_og.jpg siblings)
  liveness for the watchdog <--  GET  /api/health           CPU-only; always 200;
                                 degraded=true on 2 consecutive backend errors or
                                 a wedged ZeroGPU context (server/health.py)

Run:       python app.py                  (port from $PORT, default 7860)
Backends:  PAREIDOLIA_BACKEND = mock | zerogpu          (default mock)
Sync:      PAREIDOLIA_DATASET (+ HF_TOKEN) mirrors menagerie/ to an HF dataset.
Identity:  PAREIDOLIA_SECRET signs visitor cookies (publish <= 6/hour each).

The "her" pattern: the VLM owns taste (what counts as an eye, what the soul
sounds like); this file owns facts — bytes, tokens, limits, and what is allowed
onto the public wall. Errors and refusals are always poetry, never stack traces.
"""

from __future__ import annotations

import asyncio
import base64
import hashlib
import html
import io
import logging
import os
import re
import threading
import warnings

# ZeroGPU runtime: the `spaces` lib must be imported before torch anywhere in
# the process (it patches torch to virtualize the GPU). TORCHDYNAMO_DISABLE
# must also precede any torch import — VoxCPM's torch.compile warmup breaks
# ZeroGPU (§7). No-op everywhere else.
if os.environ.get("PAREIDOLIA_BACKEND", "").strip().lower() == "zerogpu":
    os.environ.setdefault("TORCHDYNAMO_DISABLE", "1")
    try:
        import spaces  # noqa: F401
    except ImportError:
        pass

from contextlib import asynccontextmanager
from pathlib import Path
from typing import Any, Optional

from fastapi import FastAPI, HTTPException, Query, Request, Response
from fastapi.responses import (
    FileResponse,
    HTMLResponse,
    JSONResponse,
    StreamingResponse,
)
from fastapi.staticfiles import StaticFiles

from server.health import HealthTracker, classify_exception
from server.og import compose_og
from server.persistence import PendingStore, PersistenceService
from server.ratelimit import (
    COOKIE_NAME,
    DEFAULT_IP_LIMIT,
    ClientIdentity,
    RateLimiter,
    client_ip,
)
from server.schemas import AwakenRequest, GateFlags, PublishRequest
from server.sse import SSEHub
from server.wiring import PipelineLike, make_pipeline

log = logging.getLogger("pareidolia")

ROOT = Path(__file__).resolve().parent
WEB_DIR = ROOT / "web"
MENAGERIE_DIR = ROOT / "menagerie"
SEEDS_RECORDS_DIR = ROOT / "seeds" / "records"

VALID_BACKENDS = ("mock", "zerogpu")

# Transport cap: 1.5MB of base64 text (the client downscales to <=350KB JPEG,
# so a compliant browser never gets near this; the cap stops abusive bodies).
MAX_IMAGE_B64_BYTES = 1_572_864

# Poetry for the unhappy paths — the spirits never return a stack trace.
OVERSIZE_REASON = (
    "That offering is too heavy for the séance table. "
    "Bring it under a megabyte and a half, and the spirits will look again."
)
NEEDS_IMAGE_REASON = "The spirits need a photograph to peer into. The table is bare."
BAD_IMAGE_REASON = "The spirits see no photograph here — only static."
VAST_IMAGE_REASON = (
    "The spirits cannot hold a vision that vast. "
    "Offer a smaller photograph and they will lean in close."
)
SEANCE_FAILED_REASON = (
    "The candle guttered mid-séance. Nothing was disturbed; try once more."
)
# The transient ZeroGPU quota/429 path: NOT a broken app — the visitor is out
# of free GPU seconds for the moment. Honest, actionable, retryable copy.
BUSY_REASON = (
    "So many have come to wake their things that the spirits are overwhelmed. "
    "Rest a moment, then try again — or sign in to Hugging Face to bring your "
    "own candle."
)
RATE_LIMIT_REASON = (
    "The menagerie admits six souls an hour from one hand. "
    "Sit with the ones you have woken; the wall will wait."
)
TOKEN_UNKNOWN_REASON = (
    "The spirits hold no séance by that name — or it has faded with the quarter hour."
)
GATE_RECHECK_REASON = "The wall declines this one. The spirits were clear."
NOT_FOUND_REASON = "The menagerie holds no such creature."
WALL_CROWDED_REASON = (
    "The wall is thick with watchers tonight. Linger a moment, then look again."
)

MEDIA_NAME_RE = re.compile(r"^([A-Za-z0-9_-]{4,64})\.(jpg|wav)$")
MEDIA_TYPES = {"jpg": "image/jpeg", "wav": "audio/wav"}

# Server-side re-encode bound: the client already downscales to <=1024px; this
# is the defensive ceiling for non-compliant callers (bounds VLM prefill too).
MAX_IMAGE_SIDE = 1024
JPEG_QUALITY = 88

# Decompression-bomb ceiling (security review #1): a tiny, highly-compressible
# PNG under the transport cap can DECLARE ~170M pixels and allocate 500MB+ on
# decode — before the thumbnail clamp, before any GPU quota. Cap total pixels:
# the 1024px working size with 4x headroom for the pre-thumbnail original.
MAX_PIXELS = MAX_IMAGE_SIDE * MAX_IMAGE_SIDE * 4  # ~4.2M pixels
try:
    from PIL import Image as _PILImage

    # PIL hard-stops at 2x this inside Image.open (DecompressionBombError -> our
    # poetic 400); our explicit dimension check below covers 1x-2x with a 413,
    # so the 1x warning is redundant noise — silence it, keep the error.
    _PILImage.MAX_IMAGE_PIXELS = MAX_PIXELS
    warnings.filterwarnings("ignore", category=_PILImage.DecompressionBombWarning)
except ImportError:  # pragma: no cover — PIL is a hard dependency in prod
    pass


def _backend_name() -> str:
    name = os.environ.get("PAREIDOLIA_BACKEND", "mock").strip().lower() or "mock"
    if name not in VALID_BACKENDS:
        log.warning("unknown PAREIDOLIA_BACKEND=%r; falling back to mock", name)
        name = "mock"
    return name


# --------------------------------------------------------------------------- helpers
def _set_identity_cookie(response: Response, request: Request, value: str) -> None:
    """Spaces serve over https inside an iframe (cross-site -> SameSite=None+Secure);
    local dev is plain http (Lax, not Secure)."""
    scheme = request.headers.get("x-forwarded-proto", request.url.scheme)
    secure = scheme == "https"
    response.set_cookie(
        COOKIE_NAME,
        value,
        max_age=365 * 24 * 3600,
        path="/",
        httponly=True,
        secure=secure,
        samesite="none" if secure else "lax",
    )


def _decode_image(image_b64: str):
    """base64 -> validated, bounded RGB PIL image. Raises poetic HTTPExceptions:
    413 oversize, 400 undecodable, 422 missing. PIL import stays local so the
    transport module imports clean even in odd environments."""
    if not image_b64 or not image_b64.strip():
        raise HTTPException(status_code=422, detail=NEEDS_IMAGE_REASON)
    if len(image_b64) > MAX_IMAGE_B64_BYTES:
        raise HTTPException(status_code=413, detail=OVERSIZE_REASON)
    payload = image_b64.strip()
    if payload.startswith("data:"):  # tolerate a full data URL from the client
        payload = payload.partition(",")[2]
    try:
        raw = base64.b64decode(payload, validate=True)
    except Exception as exc:
        raise HTTPException(status_code=400, detail=BAD_IMAGE_REASON) from exc
    try:
        from PIL import Image

        image = Image.open(io.BytesIO(raw))
        # Dimension gate BEFORE any pixel decode (security review #1): open()
        # only parses the header; load() is what allocates w*h*3 bytes. Refuse
        # on DECLARED size so the bomb never costs more than a header parse.
        width, height = image.size
        if width * height > MAX_PIXELS:
            raise HTTPException(status_code=413, detail=VAST_IMAGE_REASON)
        image.load()
        image = image.convert("RGB")
    except HTTPException:
        raise
    except Exception as exc:
        # Includes PIL's DecompressionBombError (the 2x MAX_PIXELS hard stop
        # raised inside Image.open) — same poetic 400 as any undecodable image.
        raise HTTPException(status_code=400, detail=BAD_IMAGE_REASON) from exc
    if max(image.size) > MAX_IMAGE_SIDE:
        image.thumbnail((MAX_IMAGE_SIDE, MAX_IMAGE_SIDE))
    return image


def _encode_jpeg(image) -> bytes:
    buf = io.BytesIO()
    image.save(buf, format="JPEG", quality=JPEG_QUALITY)
    return buf.getvalue()


# ----------------------------------------------------------------- the share page
def _public_base(request: Request) -> str:
    """Absolute origin for OG tags (crawlers refuse relative og:image URLs).
    On Spaces SPACE_HOST is the public hostname; locally fall back to the
    forwarded/request host."""
    space_host = os.environ.get("SPACE_HOST", "").strip()
    if space_host:
        return f"https://{space_host}"
    host = request.headers.get("x-forwarded-host") or request.headers.get("host")
    if not host:
        return ""
    scheme = request.headers.get("x-forwarded-proto", request.url.scheme or "https")
    return f"{scheme}://{host}"


def _share_display_name(record: dict[str, Any]) -> str:
    """Mirror web/main.js displayName: 'the rusted fire hydrant'."""
    name = " ".join(
        str(p) for p in (record.get("condition"), record.get("object")) if p
    )
    return f"the {name}".lower() if name else "an awakened thing"


def _share_page(record: dict[str, Any], base: str, og_image: str) -> str:
    """One published soul as a standalone, crawler-friendly HTML page.

    Server-persisted records only (the wall's no-client-HTML rule holds:
    every interpolated field is escaped). Inline styles on purpose — the page
    must unfurl and render even if the SPA assets change underneath it."""
    rid = str(record.get("id", ""))
    name = _share_display_name(record)
    lines = record.get("lines") or {}
    grudge = str(lines.get("grudge") or "").strip()
    mutter = str(lines.get("mutter") or "").strip()
    title = f"{name} — PAREIDOLIA"
    description = grudge or "It has a face. It has a grudge. It will tell you."
    url = f"{base}/o/{rid}"
    image_url = record.get("image_url", f"/media/{rid}.jpg")
    audio_url = record.get("audio_url", "")
    mutter_audio = record.get("mutter_audio", "")

    e = html.escape
    mutter_block = (
        f'<p class="mutter">…it also mutters: <em>{e(mutter)}</em></p>' if mutter else ""
    )
    audio_block = (
        f'<audio controls preload="none" src="{e(audio_url, quote=True)}"></audio>'
        if audio_url
        else ""
    )
    mutter_audio_block = (
        f'<audio controls preload="none" src="{e(mutter_audio, quote=True)}"></audio>'
        if mutter_audio
        else ""
    )
    return f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{e(title)}</title>
<meta name="description" content="{e(description, quote=True)}">
<meta property="og:type" content="website">
<meta property="og:site_name" content="PAREIDOLIA">
<meta property="og:title" content="{e(title, quote=True)}">
<meta property="og:description" content="{e(description, quote=True)}">
<meta property="og:url" content="{e(url, quote=True)}">
<meta property="og:image" content="{e(base + og_image, quote=True)}">
<meta property="og:image:width" content="1200">
<meta property="og:image:height" content="630">
<meta name="twitter:card" content="summary_large_image">
<meta name="twitter:title" content="{e(title, quote=True)}">
<meta name="twitter:description" content="{e(description, quote=True)}">
<meta name="twitter:image" content="{e(base + og_image, quote=True)}">
<style>
  @font-face {{ font-family: "Cormorant Garamond"; font-style: normal; font-weight: 300 700;
    src: url("/vendor/fonts/cormorant-garamond-latin.woff2") format("woff2"); }}
  @font-face {{ font-family: "Cormorant Garamond"; font-style: italic; font-weight: 300 700;
    src: url("/vendor/fonts/cormorant-garamond-latin-italic.woff2") format("woff2"); }}
  body {{ margin: 0; background: #0b0a09; color: #efe6d4;
    font-family: "Cormorant Garamond", "Iowan Old Style", Georgia, serif;
    display: grid; place-items: center; min-height: 100vh; }}
  main {{ max-width: 640px; padding: 48px 24px 64px; text-align: center; }}
  .label {{ color: #e8a849; letter-spacing: 0.14em; text-transform: lowercase;
    font-size: 14px; }}
  img.photo {{ width: 100%; max-width: 520px; border: 1px solid rgba(232,168,73,.22);
    border-radius: 4px; margin: 24px 0 18px; }}
  h1 {{ font-size: 42px; font-weight: 500; margin: 0 0 10px; }}
  blockquote {{ font-style: italic; font-size: 26px; margin: 8px 0 16px; }}
  .mutter {{ color: rgba(239,230,212,.58); font-size: 18px; margin: 8px 0; }}
  audio {{ width: 100%; max-width: 420px; margin: 8px 0; filter: sepia(.4) saturate(.6); }}
  a.cta {{ display: inline-block; margin-top: 28px; color: #e8a849;
    text-decoration: none; border-bottom: 1px solid rgba(232,168,73,.4);
    font-size: 20px; }}
  footer {{ margin-top: 36px; color: rgba(239,230,212,.32); font-style: italic;
    font-size: 16px; }}
</style>
</head>
<body>
<main>
  <p class="label">p a r e i d o l i a</p>
  <img class="photo" src="{e(image_url, quote=True)}" alt="{e(name, quote=True)}">
  <h1>{e(name)}</h1>
  <blockquote>“{e(grudge)}”</blockquote>
  {audio_block}
  {mutter_block}
  {mutter_audio_block}
  <a class="cta" href="/">step into the menagerie →</a>
  <footer>everything secretly has a face.</footer>
</main>
</body>
</html>"""


# --------------------------------------------------------------------------- the app
def create_app(
    *,
    pipeline: Optional[PipelineLike] = None,
    persistence: Optional[PersistenceService] = None,
    pending: Optional[PendingStore] = None,
    hub: Optional[SSEHub] = None,
    rate_limiter: Optional[RateLimiter] = None,
    ip_rate_limiter: Optional[RateLimiter] = None,
    identity: Optional[ClientIdentity] = None,
    health: Optional[HealthTracker] = None,
    menagerie_dir: Optional[Path | str] = None,
    backend: Optional[str] = None,
    mount_web: bool = True,
) -> FastAPI:
    """Build the ASGI app. With no arguments this wires the real (env-selected)
    backend; tests inject fakes for everything mind/cv-shaped."""
    backend_name = backend or _backend_name()
    pipeline = pipeline if pipeline is not None else make_pipeline(backend_name)
    persistence = (
        persistence
        if persistence is not None
        else PersistenceService(
            menagerie_dir if menagerie_dir is not None else MENAGERIE_DIR,
            seeds_dir=SEEDS_RECORDS_DIR,
        )
    )
    pending = pending if pending is not None else PendingStore()
    hub = hub if hub is not None else SSEHub()
    rate_limiter = rate_limiter or RateLimiter()
    ip_rate_limiter = ip_rate_limiter or RateLimiter(limit=DEFAULT_IP_LIMIT)
    identity = identity or ClientIdentity()
    health = health or HealthTracker()

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        yield
        await persistence.drain()

    # Gradio Server mode (the "her" idiom, proven on godseed): gr.Server IS a
    # FastAPI app with gradio's API engine attached — the sdk:gradio runtime
    # serves it natively, and @app.api endpoints are gradio endpoints the
    # browser calls via @gradio/client, which forwards the HF auth headers
    # ZeroGPU quota needs.
    try:
        import gradio as gr

        app: FastAPI = gr.Server(title="PAREIDOLIA")
        app.router.lifespan_context = lifespan
    except Exception:  # pragma: no cover — gradio always present in prod
        app = FastAPI(
            title="PAREIDOLIA", docs_url=None, redoc_url=None, lifespan=lifespan
        )
    app.state.persistence = persistence
    app.state.pending = pending
    app.state.hub = hub
    app.state.health = health

    async def ensure_ready() -> None:
        """Boot persistence lazily on the serving loop (godseed's ensure idiom:
        launch() does not honor an injected lifespan, so the first request —
        or the warm-boot thread in _serve — does the waking). Idempotent and
        thread-safe; the dataset restore runs off the event loop."""
        if not persistence.booted:
            await asyncio.to_thread(persistence.boot)

    # ------------------------------------------------------------------ the séance
    async def _awaken(image_b64: str) -> dict[str, Any]:
        """§2 pipeline: decode -> medium -> gate -> snap -> TTS -> pending token.
        Refusals are HTTP 200 + {refused: true}; only transport sins get codes."""
        await ensure_ready()
        image = _decode_image(image_b64)
        try:
            outcome = await asyncio.to_thread(pipeline.run, image)
        except Exception as exc:
            # Outcome tracking (reliability): label the failure ONCE so health
            # and the user-facing copy agree. Three shapes, decided here:
            #   * busy — a transient ZeroGPU quota/429 rejection. The backend is
            #     fine; the visitor is out of free GPU seconds. Healthy outcome,
            #     honest+retryable copy, HTTP 200 (NOT a 500): a 500 would read
            #     as "broken app" AND flip /api/health toward a needless restart.
            #   * gpu_context_error — the wedge (NVML/allocator); terminal.
            #   * backend_error — anything else; a flake.
            status = classify_exception(exc)
            if status == "busy":
                health.record("busy")  # single classification, single record
                log.warning("séance backpressure (busy/quota): %s", exc)
                return {
                    "refused": True,
                    "reason": BUSY_REASON,
                    "quota": True,
                    "retry": True,
                }
            health.record(status)  # same label classify_exception just produced
            log.exception("séance pipeline failed (%s)", status)
            raise HTTPException(
                status_code=500, detail=SEANCE_FAILED_REASON
            ) from exc
        if outcome.refused:
            health.record("refused")  # a refusal is a HEALTHY backend outcome
            return {"refused": True, "reason": outcome.reason or SEANCE_FAILED_REASON}
        if outcome.result is None or outcome.grudge_wav is None:
            # Not refused yet missing its substance — the user hears poetry,
            # health hears the truth: the backend produced a malformed séance.
            health.record("backend_error")
            return {"refused": True, "reason": outcome.reason or SEANCE_FAILED_REASON}
        health.record("ok")
        result = outcome.result
        jpeg = _encode_jpeg(image)
        record: dict[str, Any] = {
            "object": result.object,
            "material": result.material,
            "condition": result.condition,
            "setting": result.setting,
            "persona": result.persona.model_dump(),
            "features": [f.model_dump() for f in outcome.features],
            "lines": result.lines.model_dump(),
            "critique": result.critique,
            "image_sha256": hashlib.sha256(jpeg).hexdigest(),
            "backend": backend_name,
        }
        if getattr(outcome, "punchup", None) is not None:
            # Punch-up provenance (live medium only): {"applied", "original"}.
            # Absent on the mock path so its records stay byte-identical.
            record["punchup"] = outcome.punchup
        if getattr(outcome, "geometry", None) is not None:
            # Geometry-gate provenance (live medium only): {"violations_initial",
            # "retried", "violations_final"}. Absent on the mock path so its
            # records stay byte-identical.
            record["geometry"] = outcome.geometry
        token = pending.put(
            record=record,
            image_jpeg=jpeg,
            grudge_wav=outcome.grudge_wav,
            gate=result.gate.model_dump(),
            # Best-effort sibling (contract): None when the GPU window margin
            # skipped the mutter TTS — publish persists it only when present.
            mutter_wav=getattr(outcome, "mutter_wav", None),
        )
        return {
            "refused": False,
            "record_token": token,
            "record": {
                **record,
                "grudge_audio_b64": base64.b64encode(outcome.grudge_wav).decode(
                    "ascii"
                ),
            },
        }

    # Gradio endpoint: the visitor's browser calls this via @gradio/client so
    # the visitor's own ZeroGPU quota pays for the awakening (§0).
    if hasattr(app, "api"):

        @app.api(name="awaken")
        async def awaken(image_b64: str = "") -> dict:
            try:
                return await _awaken(image_b64)
            except HTTPException as exc:
                # gradio endpoints cannot speak HTTP status codes; same poetry,
                # flat shape, status carried in-band for the client.
                return {
                    "refused": True,
                    "reason": str(exc.detail),
                    "status": exc.status_code,
                }
            except Exception:
                log.exception("awaken failed")
                return {"refused": True, "reason": SEANCE_FAILED_REASON, "status": 500}

    # REST twin for the mock/dev backend (tests, curl, local frontend work).
    # Never registered on zerogpu: GPU work must stay inside the visitor's
    # browser-authenticated gradio request (§0), so prod matches §6 exactly.
    if backend_name != "zerogpu":

        @app.post("/api/awaken")
        async def awaken_rest(payload: AwakenRequest) -> JSONResponse:
            return JSONResponse(await _awaken(payload.image_b64))

    # ------------------------------------------------------------------ GET /api/health
    @app.get("/api/health")
    async def health_check() -> JSONResponse:
        """CPU-only liveness for tools/watchdog.py (laptop-side): per-process
        awaken counters + the degraded flag that detects the wedged-ZeroGPU
        failure mode. No auth, no GPU, never boots persistence — this must
        answer instantly even when the backend is on fire. Always HTTP 200:
        the ``degraded`` flag carries the signal, not the status code."""
        return JSONResponse(health.snapshot(backend_name))

    # -------------------------------------------------------------- POST /api/menagerie
    @app.post("/api/menagerie")
    async def publish(payload: PublishRequest, request: Request) -> JSONResponse:
        await ensure_ready()
        cid, new_cookie = identity.resolve(request)

        # Order is peek-validate -> rate-limit -> claim (security review #2c):
        # the token must prove itself BEFORE the limiters record a hit, or an
        # unauthenticated junk-token flood grows limiter state for nothing.
        # Peek (not claim) first so a rate-limited publish never spends the
        # one-time token; peek->claim cannot race — requests share one event
        # loop and nothing awaits between the two calls.
        if pending.peek(payload.record_token) is None:
            raise HTTPException(status_code=404, detail=TOKEN_UNKNOWN_REASON)

        ip_ok, ip_retry = ip_rate_limiter.hit(f"ip:{client_ip(request)}")
        allowed, retry_after = rate_limiter.hit(cid) if ip_ok else (False, ip_retry)
        if not allowed:
            raise HTTPException(
                status_code=429,
                detail=RATE_LIMIT_REASON,
                headers={"Retry-After": str(int(retry_after) + 1)},
            )

        held = pending.claim(payload.record_token)
        if held is None:  # pragma: no cover — peek above guarantees presence
            raise HTTPException(status_code=404, detail=TOKEN_UNKNOWN_REASON)

        # Gate re-check, server-side (§0: layered and non-negotiable). awaken
        # never parks a gated record, so this is pure defense-in-depth.
        if not GateFlags.model_validate(held.gate).passes():
            raise HTTPException(status_code=403, detail=GATE_RECHECK_REASON)

        # The share card is composed HERE, once, at publish time (contract):
        # strictly best-effort — a Pillow hiccup must never cost the publish;
        # the /o/ page falls back to the plain photo when no composite exists.
        og_jpeg: Optional[bytes] = None
        try:
            og_jpeg = await asyncio.to_thread(compose_og, held.image_jpeg, held.record)
        except Exception:  # noqa: BLE001 — polish never costs the wall
            log.exception("og composite failed; publishing without it")

        record = await persistence.publish(
            held.record,
            held.image_jpeg,
            held.grudge_wav,
            mutter_wav=getattr(held, "mutter_wav", None),
            og_jpeg=og_jpeg,
        )
        hub.publish({"type": "awakened", "record": record})
        # The cookie is set on the returned Response directly: FastAPI drops
        # `response`-param mutations when a handler returns its own Response
        # (and on raise paths — cookieless retries fall to the IP limiter).
        out = JSONResponse(record)
        if new_cookie:
            _set_identity_cookie(out, request, new_cookie)
        return out

    # --------------------------------------------------------------- GET /api/menagerie
    @app.get("/api/menagerie")
    async def menagerie_index(
        offset: int = Query(default=0, ge=0),
        limit: int = Query(default=60, ge=1, le=200),
    ) -> JSONResponse:
        await ensure_ready()
        records, total = persistence.page(offset, limit)
        return JSONResponse(
            {
                "records": records,
                "total": total,
                "offset": offset,
                "limit": limit,
                "awake_count": total,
            }
        )

    # ----------------------------------------------------- GET /api/menagerie/{id}
    @app.get("/api/menagerie/{record_id}")
    async def menagerie_record(record_id: str) -> JSONResponse:
        """One published record by id (the share page's data twin). The id is
        only ever a dict key — no client string becomes a path."""
        await ensure_ready()
        record = persistence.find(record_id)
        if record is None:
            raise HTTPException(status_code=404, detail=NOT_FOUND_REASON)
        return JSONResponse(record)

    # --------------------------------------------------------------- GET /o/{id}
    @app.get("/o/{record_id}", include_in_schema=False)
    async def share_page(record_id: str, request: Request) -> HTMLResponse:
        """The share page: one soul, standalone, with OG meta so the link
        unfurls (composite card when {id}_og.jpg exists, the photo otherwise).
        Published records only — pending souls have no public page."""
        await ensure_ready()
        record = persistence.find(record_id)
        if record is None:
            raise HTTPException(status_code=404, detail=NOT_FOUND_REASON)
        og = persistence.media_path(f"{record_id}_og", "jpg")
        og_image = (
            f"/media/{record_id}_og.jpg"
            if og is not None and og.is_file()
            else record.get("image_url", f"/media/{record_id}.jpg")
        )
        page = _share_page(record, _public_base(request), og_image)
        return HTMLResponse(page, headers={"Cache-Control": "public, max-age=300"})

    # ------------------------------------------------------------------ /api/stream
    @app.get("/api/stream")
    async def stream(
        request: Request,
        limit: Optional[int] = Query(
            default=None,
            ge=1,
            le=10_000,
            description="debug/test aid: close the stream after N events",
        ),
    ) -> StreamingResponse:
        await ensure_ready()
        # Capacity gate (security review #3): every subscriber holds a bounded
        # queue + a generator. The slot is reserved HERE — before the response
        # body starts — so over-capacity is an honest 503, and the generator's
        # finally-unsubscribe releases the slot on disconnect as before.
        queue = hub.subscribe(key=f"ip:{client_ip(request)}")
        if queue is None:
            raise HTTPException(
                status_code=503,
                detail=WALL_CROWDED_REASON,
                headers={"Retry-After": "15"},
            )
        initial = [{"type": "hello", "awake_count": persistence.count()}]
        return StreamingResponse(
            hub.event_stream(initial_events=initial, limit=limit, queue=queue),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",  # required on Spaces or the proxy buffers
                "Connection": "keep-alive",
            },
        )

    # --------------------------------------------------------------- GET /media/{name}
    @app.get("/media/{name}", include_in_schema=False)
    async def media(name: str) -> FileResponse:
        """Persisted menagerie media. Traversal-safe by construction: the name
        must match the mint alphabet, and the id resolves through the
        server-built id->path map only — no client string ever becomes a path."""
        await ensure_ready()
        match = MEDIA_NAME_RE.fullmatch(name)
        if match is None:
            raise HTTPException(status_code=404, detail=NOT_FOUND_REASON)
        path = persistence.media_path(match.group(1), match.group(2))
        if path is None or not path.is_file():
            raise HTTPException(status_code=404, detail=NOT_FOUND_REASON)
        return FileResponse(
            path,
            media_type=MEDIA_TYPES[match.group(2)],
            headers={"Cache-Control": "public, max-age=31536000, immutable"},
        )

    # ------------------------------------------------------------------------ static
    def _mount_static() -> None:
        """Root static mount. A "/" catch-all registered BEFORE gr.Server.launch()
        shadows the /gradio_api routes gradio adds at launch time (godseed,
        verified June 12: the launch self-check 404s and the app dies). So the
        mount is deferred: _serve() calls this AFTER launch; the plain-FastAPI
        path mounts immediately below."""
        if WEB_DIR.is_dir():
            # gradio's launch registers its own SPA index at "/" (GET+HEAD) —
            # evict exactly those so the wall owns the root; every other gradio
            # route (/config, /gradio_api/*, assets) must survive for
            # @gradio/client connectivity.
            app.router.routes[:] = [
                r for r in app.router.routes if getattr(r, "path", None) != "/"
            ]
            app.mount("/", StaticFiles(directory=str(WEB_DIR), html=True), name="web")
        else:
            log.warning("web/ not found at %s; serving API only", WEB_DIR)

            @app.get("/", include_in_schema=False)
            async def root_placeholder() -> JSONResponse:
                return JSONResponse({"ok": True, "hint": "frontend not built"})

    app.state.mount_static = _mount_static
    if mount_web and not hasattr(app, "launch"):
        _mount_static()

    return app


# --------------------------------------------------------------------------- entrypoint
def _serve(application: FastAPI) -> None:
    port = int(os.environ.get("PORT", os.environ.get("GRADIO_SERVER_PORT", 7860)))
    log.info("PAREIDOLIA listening on http://0.0.0.0:%d", port)
    if hasattr(application, "launch"):
        # Gradio Server mode — launch non-blocking so the web root can mount
        # AFTER gradio registers its /gradio_api routes (order = precedence).
        application.launch(
            server_name="0.0.0.0",
            server_port=port,
            show_error=False,
            prevent_thread_lock=True,
        )
        mount_static = getattr(application.state, "mount_static", None)
        if mount_static is not None:
            mount_static()
        # Warm-boot the menagerie off the serving thread so visitor #1 sees a
        # full wall without paying the dataset-restore wait themselves.
        persistence = getattr(application.state, "persistence", None)
        if persistence is not None and not persistence.booted:
            threading.Thread(target=persistence.boot, daemon=True).start()
        import time as _time

        while True:  # keep the process alive; the server runs in gradio's thread
            _time.sleep(3600)
    else:  # pragma: no cover — gradio-less dev fallback
        import uvicorn

        uvicorn.run(application, host="0.0.0.0", port=port)


# HF Spaces (sdk:gradio) executes this file; locally it's `python app.py`.
# Test imports leave both conditions false and get no side effects.
if __name__ == "__main__" or os.environ.get("SPACE_ID"):
    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s"
    )
    _serve(create_app())