"""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'

…it also mutters: {e(mutter)}

' if mutter else "" ) audio_block = ( f'' if audio_url else "" ) mutter_audio_block = ( f'' if mutter_audio else "" ) return f""" {e(title)}

p a r e i d o l i a

{e(name, quote=True)}

{e(name)}

“{e(grudge)}”
{audio_block} {mutter_block} {mutter_audio_block} step into the menagerie →
""" # --------------------------------------------------------------------------- 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())