Add authed STT proxy route POST /v1/stt/transcribe (#8854 step 1) (#9011)

## What

Adds `POST /v1/stt/transcribe` — an authenticated backend proxy in front
of the
parakeet GPU service's `/v2/transcribe`. This is step 1 of the parakeet
cleanup
plan laid out in #8854 ("Add new authed backend proxy route for mobile
app").

## Why

When a user selects the "Omi Parakeet" STT provider, the mobile app
currently
posts audio buffers directly to `parakeet.omiapi.com/v1/transcribe` with
no
authentication of any kind (`app/lib/models/stt_provider.dart:533`
builds the
request with no Authorization header, and the parakeet service has no
auth
guard of its own). That leaves a GPU transcription service effectively
open to
the public at that path.

This route puts the standard Omi stack in front of it:

- **Auth**: `get_current_user_uid` (Firebase ID token), same dependency
as every other authed endpoint.
- **Rate limiting**: new `stt:transcribe` policy (60/h per UID, mirrors
`voice:transcribe`).
- **Resource isolation**: dedicated `get_stt_proxy_client()` /
`get_stt_proxy_semaphore()`
(4 connections, keep-alive off) so bulk uploads can never starve the
listen
pipeline's internal STT pool (VAD, speaker embedding, speech profile).
Requests
fail fast with 503 if no upstream slot frees up within 30s, and the body
is only
buffered into RAM after a slot is held (max 4 resident bodies; waiters
stay in
  the multipart disk spool).
- **Abuse guards**: 200MB body cap (Content-Length + `file.size`
pre-checks, plus a
bounded read as defense-in-depth), and upstream filename sanitization —
parakeet
builds its temp path from the client filename, so path separators,
dot-prefixed
  and overlong names are never forwarded.
- **Error mapping**: parakeet's client-actionable 413 (audio too long)
and 503
(model loading / overloaded) are forwarded with their detail; any other
upstream
  failure maps to a generic 502 with nothing leaked.
- Route gets a 350s path timeout (`HTTP_STT_TRANSCRIBE_TIMEOUT`) to
cover the 300s
  upstream budget; the 120s POST default would kill long transcriptions.

The response mirrors parakeet `/v2/transcribe` verbatim
(`{"text", "segments", "detected_language"}`), so migrating the app is a
URL + auth-header swap.

## Follow-ups (out of scope, per the #8854 sequencing)

- Migrate the app's `omiParakeet` provider to this route — needs a
dynamic auth
header (Firebase tokens expire hourly; `SchemaBasedSttProvider` takes
static
  headers at socket creation), so it's a separate app-side change.
- Deprecate / remove the public `/v1/transcribe` exposure after the soak
period.

## Testing

- 28 unit tests (`backend/tests/unit/test_stt_router.py`, registered in
`test.sh`):
auth required, config guard, empty/oversized payloads (including a
chunked body
with no Content-Length and a direct-call test pinning the bounded read
when
`file.size` is unavailable), filename sanitization matrix, slot
exhaustion → 503,
response passthrough, `diarize` forwarding, upstream 413/503 forwarded
(dict,
non-dict JSON and HTML bodies), upstream 500 → 502 with no body leak,
network
  error → 502.
- End-to-end: ran the real router under uvicorn against a fake parakeet
upstream
and exercised it with `curl` + a real WAV — 401 without auth, 200 with
exact
  JSON passthrough, `diarize=false` forwarded, multipart filename
  `../../etc/evil.wav` arriving upstream as `evil.wav`.
- Full `backend/test.sh` run; `scan_async_blockers.py` and
  `scan_import_time_side_effects.py` clean.

Refs #8854


<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/BasedHardware/omi/pull/9011?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

Author: undivisible

backend/AGENTS.md

@@ -141,7 +141,7 @@ Serving STT provider/surface policy and canonical model order are owned exclusiv
 - **diarizer** (`diarizer/main.py`) — GPU. Speaker embeddings at `/v2/embedding`. Called by backend and pusher (`HOSTED_SPEAKER_EMBEDDING_API_URL`).
 - **vad** (`modal/main.py`) — GPU. `/v1/vad` and `/v1/speaker-identification`. Called by backend only.
 - **deepgram-self-hosted** — GPU STT deployment behind an explicit non-cloud endpoint. Hosted Deepgram serves the live streaming surface (`dg-nova-3` first, `DEEPGRAM_API_KEY`); it stays off batch. A deployment that sets `DEEPGRAM_SELF_HOSTED_ENABLED` must point at its own endpoint — never `api.deepgram.com`.
-- **parakeet** (`parakeet/`) — GPU STT service for streaming and pre-recorded transcription. Called by backend when `HOSTED_PARAKEET_API_URL` is set and Parakeet is selected.
+- **parakeet** (`parakeet/`) — GPU STT service for streaming and pre-recorded transcription. Called by backend when `HOSTED_PARAKEET_API_URL` is set and Parakeet is selected. Clients must not call it directly — they go through the authed backend proxy `POST /v1/stt/transcribe` (`routers/stt.py`).
 - **modulate** — Managed STT provider for configured languages. Called by backend when `MODULATE_API_KEY` is configured and Modulate is selected.
 - **nllb-translation** (`nllb_translation/`) — GPU translation service. Called by backend when `HOSTED_TRANSLATION_API_URL` is set and NLLB is selected.
 - **backend-sync** (`main.py`, same image as backend) — Cloud Run admission service for `/v2/sync-local-files`. The server classifies whole batches: recordings no more than six hours old enter `sync-jobs` (fresh), while older or untrusted batches enter `sync-backfill` and the scale-to-zero **backend-sync-backfill** worker. Fresh keeps its bounded inline fallback; backfill never falls into fresh/inline capacity. Backfill defaults to one in-flight job per UID, four processed speech hours per UID/day, 555 processed speech hours globally/day, a 30-day lookback, and four queue workers. Live fair-use reads only `realtime + sync_fresh`; `sync_backfill` is separately metered. A 45-day Firestore content ledger protects transcription and usage side effects across job expiry and re-upload. Audio playback merges (`/v1/sync/audio/*`) follow the same pattern via queue `audio-merge` building 30-day MP3 artifacts under `playback/` (`AUDIO_MERGE_DISPATCH_MODE`) — per-part files plus one dense per-conversation `conversation.mp3` whose spans manifest + audio_files fingerprint are stamped on the conversation doc (`conversation_audio`); a fingerprint mismatch after late chunks re-enqueues the build. In production, account deletion requires `ACCOUNT_DELETION_DISPATCH_MODE=cloud_tasks` and complete Cloud Tasks bindings to enqueue opaque job IDs to queue `account-deletion`, which posts `/v1/users/account-deletion-wipes/run`; startup rejects inline or incomplete configuration, reconciliation only re-dispatches tasks so the OIDC handler is the sole wipe executor, and the post-deploy queue-drain window accepts the former sync OIDC audience only for legacy UID payloads. API success is returned only after the deletion marker is persisted and the wipe task is durably enqueued.
@@ -249,7 +249,7 @@ black --line-length 120 --skip-string-normalization <files>
 Never block the event loop — it freezes health checks, HPA scaling, and all concurrent connections.
 
 - **Lane 1 — Async HTTP** (`utils/http_client.py`): Shared `httpx.AsyncClient` pools with semaphore-bounded concurrency. Never `requests.*` or sync `httpx.*` in async code.
-  - Clients: `get_webhook_client()`, `get_maps_client()`, `get_auth_client()`, `get_stt_client()`
+  - Clients: `get_webhook_client()`, `get_maps_client()`, `get_auth_client()`, `get_stt_client()`, `get_stt_proxy_client()`, `get_tts_client()`, `get_web_fetch_client()`
   - Semaphores: always wrap calls — `async with get_webhook_semaphore(): await client.post(...)`
   - Circuit breakers: `get_webhook_circuit_breaker(url)` for external targets — call `cb.record_success()`/`cb.record_failure()`
   - Lifecycle: lazy singletons, closed at shutdown via `close_all_clients()`

backend/main.py

@@ -80,6 +80,7 @@
     desktop_screen_crisp,
     desktop_tts_updates,
     scores,
+    stt,
     tts,
     memory_admin,
     memory_product,
@@ -210,6 +211,7 @@
 app.include_router(chat_sessions.router)
 app.include_router(chat_generation.router)
 app.include_router(scores.router)
+app.include_router(stt.router)
 app.include_router(tts.router)
 app.include_router(memory_admin.router)
 app.include_router(memory_product.router)
@@ -239,6 +241,10 @@
     "/v2/audio-merge-jobs/run": os.environ.get('HTTP_AUDIO_MERGE_RUN_TIMEOUT', 600),
     "/v1/users/account-deletion-wipes/run": os.environ.get('HTTP_ACCOUNT_DELETION_WIPE_RUN_TIMEOUT', 1500),
     "/v1/conversation-finalization-jobs/run": os.environ.get('HTTP_LISTEN_FINALIZATION_RUN_TIMEOUT', 1500),
+    # STT proxy: 30s slot wait + 300s parakeet client budget (get_stt_proxy_client)
+    # + headroom for auth and the multipart spool read; the default POST timeout
+    # would cut long files off mid-transcription.
+    "/v1/stt/transcribe": os.environ.get('HTTP_STT_TRANSCRIBE_TIMEOUT', 350),
 }
 
 app.add_middleware(TimeoutMiddleware, methods_timeout=methods_timeout, paths_timeout=paths_timeout)

backend/route_policy_manifest.yaml

@@ -77,6 +77,29 @@ routes:
     method: POST
     path: /v1/tts/synthesize
     policy: *desktop_migration_policy
+  - route_type: http
+    method: POST
+    path: /v1/stt/transcribe
+    policy:
+      review_status: reviewed
+      auth:
+        mechanisms:
+          - firebase_id_token
+        placement: dependency
+        scopes: []
+      byok: not_applicable
+      rate_limit:
+        policy_name: stt:transcribe
+        key_subject: uid
+        enforcement: fail_closed
+        placement: wrapper
+      timeout_class: unknown
+      surface: first_party_app
+      visibility: first_party
+      data_domain: conversations
+      deprecation:
+        state: active
+      owner: backend
   - route_type: http
     method: POST
     path: /v1/webhooks/sentry

backend/routers/stt.py

@@ -0,0 +1,139 @@
+"""STT proxy route — authed proxy in front of the parakeet transcription service.
+
+Step 1 of the parakeet cleanup plan (issue #8854): clients must not call the
+parakeet service directly (it has no auth of its own — only the internal LB
+shields it). This route puts the standard Omi auth guard plus per-UID rate
+limiting in front of parakeet's `/v2/transcribe`, so the mobile app's
+"Omi Parakeet" provider can migrate off the direct `parakeet.omiapi.com` URL.
+
+The response mirrors parakeet `/v2/transcribe` verbatim:
+    {"text": str, "segments": [{"start", "end", "text", "speaker"}], "detected_language": str}
+"""
+
+import asyncio
+import logging
+import os
+import re
+
+import httpx
+from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile
+
+from utils.http_client import get_stt_proxy_client, get_stt_proxy_semaphore
+from utils.log_sanitizer import sanitize
+from utils.other import endpoints as auth
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# Mirrors _MAX_PCM_BODY_BYTES in routers/chat.py. Parakeet enforces its own
+# duration cap (PARAKEET_MAX_FILE_DURATION); this only bounds backend memory.
+_MAX_UPLOAD_BYTES = 200_000_000
+
+# How long a request may wait for an upstream slot before failing fast with
+# 503. Waiters hold no RAM (the body stays in starlette's disk spool until a
+# slot is acquired), so this only bounds client-perceived latency.
+_UPSTREAM_WAIT_SECS = 30.0
+
+_SAFE_FILENAME_RE = re.compile(r'[^A-Za-z0-9._-]')
+_MAX_FILENAME_LEN = 64
+_MAX_EXTENSION_LEN = 8
+
+
+def _safe_upstream_filename(filename) -> str:
+    """Parakeet builds its temp file path from the client-supplied filename —
+    never forward path separators, dot-prefixed names, or overlong names.
+    Truncation keeps a short extension so upstream decoders that sniff by
+    suffix still see it.
+    """
+    base = os.path.basename(filename or '')
+    base = _SAFE_FILENAME_RE.sub('_', base).lstrip('.')
+    if len(base) > _MAX_FILENAME_LEN:
+        stem, dot, ext = base.rpartition('.')
+        if dot and 0 < len(ext) <= _MAX_EXTENSION_LEN:
+            base = stem[: _MAX_FILENAME_LEN - len(ext) - 1] + '.' + ext
+        else:
+            base = base[:_MAX_FILENAME_LEN]
+    return base or 'audio.wav'
+
+
+@router.post('/v1/stt/transcribe', tags=['stt'])
+async def stt_transcribe(
+    request: Request,
+    file: UploadFile = File(...),
+    diarize: bool = Form(True),
+    uid: str = Depends(auth.with_rate_limit(auth.get_current_user_uid, "stt:transcribe")),
+):
+    """Transcribe an uploaded audio file via parakeet. Per-user rate limited."""
+    api_url = os.getenv('HOSTED_PARAKEET_API_URL')
+    if not api_url:
+        logger.error('stt_transcribe: HOSTED_PARAKEET_API_URL not configured')
+        raise HTTPException(status_code=503, detail='Transcription service not configured')
+
+    # Starlette has already spooled the multipart body to disk by this point;
+    # these checks bound backend RAM, not the upload itself. file.size is the
+    # primary size enforcement — starlette populates it for every multipart
+    # part, chunked uploads included. The Content-Length check just rejects a
+    # step earlier when the header is present.
+    content_length = request.headers.get('content-length')
+    if content_length and content_length.isdigit() and int(content_length) > _MAX_UPLOAD_BYTES:
+        raise HTTPException(status_code=413, detail=f'Body too large (max {_MAX_UPLOAD_BYTES} bytes)')
+    if file.size is not None and file.size > _MAX_UPLOAD_BYTES:
+        raise HTTPException(status_code=413, detail=f'Body too large (max {_MAX_UPLOAD_BYTES} bytes)')
+
+    upstream_url = api_url.rstrip('/') + '/v2/transcribe'
+    client = get_stt_proxy_client()
+    semaphore = get_stt_proxy_semaphore()
+
+    try:
+        await asyncio.wait_for(semaphore.acquire(), timeout=_UPSTREAM_WAIT_SECS)
+    except asyncio.TimeoutError:
+        logger.warning(f'stt_transcribe: no upstream slot within {_UPSTREAM_WAIT_SECS}s uid={uid}')
+        raise HTTPException(status_code=503, detail='Transcription service busy — try again shortly')
+
+    try:
+        # Buffer only after holding a slot, so at most 4 request bodies are
+        # resident in RAM at once — waiters keep theirs in the disk spool.
+        # The bounded read is defense-in-depth should file.size be absent.
+        audio_bytes = await file.read(_MAX_UPLOAD_BYTES + 1)
+        if not audio_bytes:
+            raise HTTPException(status_code=400, detail='No audio data provided')
+        if len(audio_bytes) > _MAX_UPLOAD_BYTES:
+            del audio_bytes
+            raise HTTPException(status_code=413, detail=f'Body too large (max {_MAX_UPLOAD_BYTES} bytes)')
+
+        files = {'file': (_safe_upstream_filename(file.filename), audio_bytes, file.content_type or 'audio/wav')}
+        data = {'diarize': 'true' if diarize else 'false'}
+        try:
+            resp = await client.post(upstream_url, files=files, data=data)
+        except httpx.HTTPError as e:
+            logger.error(f'stt_transcribe: parakeet request failed uid={uid}: {sanitize(str(e))}')
+            raise HTTPException(status_code=502, detail='Transcription upstream unavailable')
+        finally:
+            del files
+            del audio_bytes
+    finally:
+        semaphore.release()
+
+    if resp.status_code >= 400:
+        # 413 (audio too long) and 503 (model loading / overloaded) carry
+        # client-actionable details — forward them as-is. Anything else is an
+        # internal upstream failure the client can't act on.
+        if resp.status_code in (413, 503):
+            detail = 'Transcription failed upstream'
+            try:
+                payload = resp.json()
+            except ValueError:
+                payload = None
+            if isinstance(payload, dict) and payload.get('detail'):
+                detail = str(payload['detail'])
+            logger.warning(f'stt_transcribe: parakeet returned {resp.status_code} uid={uid}: {sanitize(detail)}')
+            raise HTTPException(status_code=resp.status_code, detail=detail)
+        logger.error(f'stt_transcribe: parakeet returned {resp.status_code} uid={uid}: {sanitize(resp.text[:200])}')
+        raise HTTPException(status_code=502, detail='Transcription failed upstream')
+
+    try:
+        return resp.json()
+    except ValueError:
+        logger.error(f'stt_transcribe: parakeet returned non-JSON success body uid={uid}')
+        raise HTTPException(status_code=502, detail='Transcription upstream returned an invalid response')