"""Unified DriftCall Space — every artefact at a slash-path, served locally.
URL surface:
/ static project site (Vite-built React + Pretext)
/assets/* site bundle
/healthz OpenEnv health probe
/reset POST OpenEnv canonical reset
/step POST OpenEnv step
/state GET OpenEnv read-only state
/close POST OpenEnv close
/openenv.yaml OpenEnv v1.0 manifest
/docs FastAPI / Swagger UI for the env routes
/demo voice demo Gradio app, MOUNTED LOCALLY (no iframe)
/env landing page documenting env routes (curl recipes)
/lora landing page with LoRA metadata + download link
/source landing page with the project file tree + repo link
Nothing under this Space depends on a 302 redirect to another origin —
the demo Gradio Blocks are mounted via gr.mount_gradio_app, the LoRA
and source pages are rendered server-side from local data.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
from fastapi import FastAPI
from fastapi.responses import FileResponse, HTMLResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
# `app.py` refuses to start without DRIFTCALL_ENV_TOKEN. For the hackathon
# demo Space we publish a known public token so anyone can hit the OpenEnv
# API end-to-end. To restrict access, override via Space Settings → Secrets.
os.environ.setdefault("DRIFTCALL_ENV_TOKEN", "driftcall-demo")
# `app.py` eager-loads Kokoro TTS + faster-whisper ASR at lifespan startup.
# On this unified Space those engines aren't needed for the OpenEnv API
# surface — they only matter inside the /demo Gradio app, which constructs
# its own engines lazily on first audio I/O. We monkey-patch the eager-load
# step BEFORE importing app.py so the lifespan handler skips it. This also
# dodges the misaki/kokoro version-drift error (`module 'misaki.en' has no
# attribute 'MutableToken'`) that was crashing startup.
import importlib
_audio = importlib.import_module("cells.step_09_audio")
def _noop_engine() -> None:
return None
_audio.get_tts_engine = _noop_engine # type: ignore[assignment]
_audio.get_asr_engine = _noop_engine # type: ignore[assignment]
from app import app as openenv_app # noqa: E402 # type: ignore[import-not-found]
from online_trainer import get_online_trainer # noqa: E402 # type: ignore[import-not-found]
LORA_HUB_URL = "https://huggingface.co/DGXAI/gemma-3n-e2b-driftcall-lora"
SOURCE_URL = "https://github.com/saumilyagupta/openenv-DGXAI"
SITE_DIR = Path(__file__).parent / "site"
MANIFEST_PATH = Path(__file__).parent / "openenv.yaml"
BLOG_PATH = Path(__file__).parent / "BLOG.md"
# ---------------------------------------------------------------------------
# Shared chrome — same dark editorial brutalism as the React site.
# ---------------------------------------------------------------------------
_HEAD = """
__TITLE__DriftCall · __SLUG__
__BODY__
"""
_NAV_LINKS = [
("/", "site"),
("/demo", "demo"),
("/blog", "blog"),
("/env", "env"),
("/openenv.yaml", "manifest"),
("/docs", "docs"),
("/lora", "lora"),
("/source", "source"),
]
def _page(title: str, slug: str, body: str, active: str) -> str:
nav_items: list[str] = []
for href, label in _NAV_LINKS:
cls = ' class="is-active"' if href == active else ""
nav_items.append(f'{label}')
nav = "\n ".join(nav_items)
return (
_HEAD
.replace("__TITLE__", title)
.replace("__SLUG__", slug)
.replace("__NAV__", nav)
.replace("__BODY__", body)
)
# ---------------------------------------------------------------------------
# /blog — render BLOG.md with the site chrome.
# ---------------------------------------------------------------------------
_BLOG_EXTRA_CSS = """
"""
def _render_blog_html() -> str:
"""Render BLOG.md to HTML. Returns an empty-state notice if missing."""
if not BLOG_PATH.exists():
return (
f"{_BLOG_EXTRA_CSS}"
'
'
"BLOG.md not bundled in this build."
"
"
)
raw = BLOG_PATH.read_text(encoding="utf-8")
# Strip Hugging Face blog YAML frontmatter (between leading --- fences).
if raw.startswith("---"):
try:
_, _, rest = raw.split("---", 2)
raw = rest.lstrip("\n")
except ValueError:
pass
try:
import markdown as md # type: ignore[import-not-found]
body_html = md.markdown(
raw,
extensions=[
"extra",
"tables",
"fenced_code",
"sane_lists",
"attr_list",
"toc",
"pymdownx.tilde",
"pymdownx.superfences",
],
)
except Exception:
# Hard fallback: escape and dump in a
so the page never 500s.
from html import escape
body_html = f"
Every endpoint sits at the bare path the OpenEnv v1.0 spec expects —
no /api prefix, no rewriting. Auth is bearer
(DRIFTCALL_ENV_TOKEN) plus X-Session-Id on
every mutating call.
method
path
description
GET
/healthz
health probe (unauthenticated, returns "ok")
POST
/reset
create or recycle a session (seed / curriculum_stage / language_weights / audio_boundary_enabled)
POST
/step
advance one turn — body {"action": <DriftCallAction>}
GET
/state
read-only DriftCallState snapshot
POST
/close
evict the server-side session
GET
/openenv.yaml
OpenEnv v1.0 manifest
GET
/docs
auto-generated FastAPI / Swagger UI
Try it
This Space publishes a known public bearer token so anyone can exercise
the API end-to-end:
driftcall-demo.
"""
# ---------------------------------------------------------------------------
# /lora — local landing page (no 302) with model card details.
# ---------------------------------------------------------------------------
_LORA_BODY = f"""
Trained LoRA adapter
GRPO-tuned LoRA over Gemma-3n-E2B-it. Five reward components, three-stage
curriculum (no drift → single drift → compound drift), 240 steps total
on H100 80GB. Adapter-only — never the merged 16-bit weights, per
DESIGN.md §10.5.
"""
# ---------------------------------------------------------------------------
# /source — local landing page (no 302) with the project file tree.
# ---------------------------------------------------------------------------
_SOURCE_BODY = f"""
Project source
This Space bundles the entire DriftCall project — runtime modules,
training scripts, the Colab-runnable notebook, design docs, the test
suite, and the spec corpus. Browse the full file tree under the
Files tab
of this Space, or open the canonical mirror on GitHub.
"""
def _build_demo_blocks() -> Any:
"""Build the Gradio Blocks lazily so a missing dep at import time
doesn't kill the whole FastAPI app — instead /demo will return 503
with a clear message until the deps come back online."""
try:
from demo_app import build_ui # type: ignore[import-not-found]
return build_ui()
except Exception as exc: # noqa: BLE001
# We log via FastAPI's normal startup logs; nothing more to do.
import logging
logging.getLogger("unified").exception("demo blocks build failed: %s", exc)
return None
def build_unified_app() -> FastAPI:
app: FastAPI = openenv_app
@app.get("/openenv.yaml", include_in_schema=False)
async def serve_manifest() -> Any:
if MANIFEST_PATH.exists():
return FileResponse(MANIFEST_PATH, media_type="text/yaml")
return {"error": "openenv.yaml not found"}
@app.get("/env", include_in_schema=False)
async def serve_env_page() -> HTMLResponse:
return HTMLResponse(_page("DriftCall — OpenEnv API", "/env", _ENV_BODY, "/env"))
@app.get("/lora", include_in_schema=False)
async def serve_lora_page() -> HTMLResponse:
return HTMLResponse(_page("DriftCall — LoRA", "/lora", _LORA_BODY, "/lora"))
@app.get("/source", include_in_schema=False)
async def serve_source_page() -> HTMLResponse:
return HTMLResponse(_page("DriftCall — source", "/source", _SOURCE_BODY, "/source"))
@app.get("/blog", include_in_schema=False)
async def serve_blog_page() -> HTMLResponse:
return HTMLResponse(_page("DriftCall — blog", "/blog", _render_blog_html(), "/blog"))
# ── Live online RL — subprocess wrapper around scripts/train_driftcall_grpo.py.
@app.get("/training", include_in_schema=False)
async def training_status() -> Any:
return get_online_trainer().status()
@app.post("/training/start", include_in_schema=False)
async def training_start() -> Any:
return get_online_trainer().start()
@app.post("/training/stop", include_in_schema=False)
async def training_stop() -> Any:
return get_online_trainer().stop()
# Note: @app.on_event('startup') is a no-op here because app.py already
# registers a custom `lifespan` context manager — the two APIs are mutually
# exclusive. We instead kick off the trainer at module load below, after
# `app = build_unified_app()` returns. The trainer's start() spawns a
# subprocess and returns immediately, so it never blocks uvicorn boot.
# Mount the Gradio voice demo at /demo — runs locally, no iframe.
# Gradio mounts under /demo/ (with trailing slash). Register a
# bare /demo handler that redirects to /demo/ so users typing the
# short URL don't hit the SPA static catch-all and 404.
@app.get("/demo", include_in_schema=False)
async def demo_trailing_slash() -> RedirectResponse:
return RedirectResponse(url="/demo/", status_code=308)
blocks = _build_demo_blocks()
if blocks is not None:
try:
import gradio as gr # type: ignore[import-not-found]
gr.mount_gradio_app(app, blocks, path="/demo")
except Exception:
import logging
logging.getLogger("unified").exception("gr.mount_gradio_app failed")
# SPA static mount — must come LAST so OpenEnv routes and our
# explicit /env, /lora, /source, /demo handlers take precedence.
if SITE_DIR.exists():
app.mount(
"/",
StaticFiles(directory=SITE_DIR, html=True),
name="frontend",
)
return app
app = build_unified_app()
# Autostart the online GRPO trainer subprocess at module load.
# We can't use @app.on_event('startup') because app.py registers a custom
# `lifespan` context manager — the two APIs are mutually exclusive and the
# decorator silently no-ops. Module-level call runs once on Space boot,
# spawns a subprocess, and returns immediately so uvicorn isn't blocked.
if os.environ.get("DRIFTCALL_ONLINE_AUTOSTART", "1") == "1":
try:
get_online_trainer().start()
except Exception:
import logging
logging.getLogger("unified").exception("trainer autostart failed")