feat: add optional background transcription jobs

CHANGELOG.md

@@ -27,6 +27,8 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
   overlapping character-library refreshes cannot overwrite a newer table/card view.
 - **Constrained Browser Smoke Coverage.** Added a 4 GB `navigator.deviceMemory` regression that verifies shared
   batch work is capped at two concurrent operations.
+- **Optional Background Transcription.** `/api/transcribe` now accepts `background: true` and exposes bounded job
+  status/cancellation endpoints with timeout and failure states; existing synchronous callers remain compatible.
 - **API Boundary Hardening.** External API authentication is enabled by default, request and mutation limits are
   enforced, and settings responses redact provider credentials for non-browser callers.
 - **Server-side Provider Credentials.** Provider keys are now masked even for same-origin Settings responses, are

TODO.md

@@ -62,9 +62,9 @@ the first block before investing in architecture or polish.
 - [ ] **M — Add constrained-device validation.** Test with Chrome CPU/memory throttling,
   a 4 GB VM/container, and measure browser, Docker and temporary-disk usage. Automated browser emulation now proves
   the 4 GB shared-pool cap; physical VM/container and resource measurements remain.
-- [ ] **L — Add background jobs and bounded queues.** Give voice design, transcription,
+- [ ] **L — Add background jobs and bounded queues.** Image generation already has the full interactive job path,
-  casting and image generation IDs, progress, cancellation, timeout, retry and
+  and transcription now has an optional ID/status/cancel/timeout path while preserving synchronous callers. Voice
-  idempotency support.
+  design and casting still need migration to the same user-facing durable job model, plus retries and persistence.
 - [ ] **L — Run the Book 2 generalization regression.** Confirm speaker-attribution and
   confidence rules are not overfit to *Der Sturm*.
 - [x] **L — Add character forms inside one character record.** `sheet.forms[]` stores bounded alternate appearances,
@@ -124,7 +124,8 @@ the first block before investing in architecture or polish.
 
 ## P1 — Backend stability and bounded concurrency
 
-- [ ] Introduce background jobs for voice design, transcription, casting, and image generation.
+- [ ] Introduce background jobs for voice design, transcription, casting, and image generation. Image generation is
+  complete and transcription has an optional background endpoint; voice design and casting remain synchronous.
 - [ ] Give every job an ID, progress, cancellation, timeout, retry, and idempotency key. Consistent image jobs now
   provide all except retry; voice design and casting remain synchronous request paths.
 - [x] Prevent duplicate generation when a user clicks twice for image, voice-design, and casting submissions.

routes/stt.py

@@ -25,6 +25,9 @@ from core.voice import _AUDIO_EXTS, _UPLOAD_EXTS
 
 router = APIRouter()
 _STT_GENERATION_SLOT = asyncio.Semaphore(1)
+_STT_JOBS: dict[str, dict] = {}
+_STT_JOB_MAX = 128
+_STT_JOB_TIMEOUT = 15 * 60
 
 
 async def _bounded_stt_call(fn, *args):
@@ -38,6 +41,41 @@ async def _bounded_stt_call(fn, *args):
     finally:
         _STT_GENERATION_SLOT.release()
 
+
+def _stt_job_set(job_id: str, **fields) -> None:
+    job = _STT_JOBS.get(job_id)
+    if job is not None:
+        job.update(fields)
+        job["updated"] = time.time()
+
+
+async def _run_stt_job(job_id: str, src: Path, settings: dict, backend: str) -> None:
+    _stt_job_set(job_id, stage="running", detail="Transcribing…")
+    try:
+        text, used_backend = await asyncio.wait_for(
+            _bounded_stt_call(_transcribe_audio, src, settings, backend),
+            timeout=_STT_JOB_TIMEOUT,
+        )
+        if _STT_JOBS.get(job_id, {}).get("cancel_requested"):
+            _stt_job_set(job_id, stage="cancelled", detail="Cancelled", done=True, cancelled=True, error="Transcription cancelled")
+        else:
+            _stt_job_set(job_id, stage="done", detail="Done", done=True, result={"text": text, "backend": used_backend})
+    except asyncio.TimeoutError:
+        _stt_job_set(job_id, stage="error", detail="Timed out", done=True, error="STT transcription timed out")
+    except Exception as exc:
+        message = str(exc.detail) if isinstance(exc, HTTPException) else str(exc)
+        _stt_job_set(job_id, stage="error", detail="Failed", done=True, error=message)
+
+
+def _stt_job_trim() -> None:
+    if len(_STT_JOBS) <= _STT_JOB_MAX:
+        return
+    for job_id, job in sorted(_STT_JOBS.items(), key=lambda item: item[1].get("updated", 0)):
+        if job.get("done"):
+            _STT_JOBS.pop(job_id, None)
+            if len(_STT_JOBS) <= _STT_JOB_MAX:
+                break
+
 # ── STT backend definitions ───────────────────────────────────────────────────
 
 _STT_BACKEND_ALIASES = {
@@ -493,6 +531,17 @@ async def transcribe(request: Request):
     # back anywhere. An explicit request still wins over the preference.
     raw_backend = data.get("backend") or data.get("stt_backend") or settings.get("stt_preferred_backend") or "configured"
     backend = _clean_stt_backend(str(raw_backend))
+    if data.get("background") is True:
+        job_id = uuid.uuid4().hex
+        now = time.time()
+        _stt_jobs_trim = _stt_job_trim
+        _STT_JOBS[job_id] = {
+            "stage": "queued", "detail": "Queued…", "started": now, "updated": now,
+            "done": False, "cancel_requested": False, "cancelled": False, "error": None, "result": None,
+        }
+        _stt_jobs_trim()
+        asyncio.create_task(_run_stt_job(job_id, src, settings, backend))
+        return {"job_id": job_id}
     try:
         text, used_backend = await _bounded_stt_call(_transcribe_audio, src, settings, backend)
         return {"text": text, "backend": used_backend}
@@ -500,6 +549,29 @@ async def transcribe(request: Request):
         raise HTTPException(502, f"STT error ({backend}): {e}")
 
 
+@router.get("/api/transcribe/status/{job_id}")
+async def transcribe_status(job_id: str):
+    job = _STT_JOBS.get(job_id)
+    if job is None:
+        raise HTTPException(404, "Unknown or expired transcription job")
+    payload = dict(job)
+    payload["elapsed"] = round(time.time() - payload.pop("started"), 1)
+    payload.pop("cancel_requested", None)
+    return payload
+
+
+@router.post("/api/transcribe/cancel/{job_id}")
+async def transcribe_cancel(job_id: str):
+    job = _STT_JOBS.get(job_id)
+    if job is None:
+        raise HTTPException(404, "Unknown or expired transcription job")
+    if job.get("done"):
+        return {"ok": True, "already_done": True, "cancelled": bool(job.get("cancelled"))}
+    job["cancel_requested"] = True
+    _stt_job_set(job_id, stage="cancelling", detail="Cancelling…")
+    return {"ok": True, "cancelled": True}
+
+
 @router.post("/api/transcribe-bytes")
 async def transcribe_bytes(
     file: UploadFile = File(...),

tests/test_security.py

@@ -103,6 +103,28 @@ class SecurityValidationTests(unittest.TestCase):
         self.assertTrue(response.json().get("external_api_key"))
         self.assertIn("no-store", response.headers.get("cache-control", ""))
 
+    def test_transcription_background_job_reports_completion(self):
+        from pathlib import Path
+        from tempfile import TemporaryDirectory
+        from unittest.mock import patch
+        import routes.stt as stt
+        with TemporaryDirectory() as directory:
+            source = Path(directory) / "sample.wav"
+            source.write_bytes(b"RIFF")
+            job_id = "stt-test-job"
+            now = 0.0
+            stt._STT_JOBS[job_id] = {
+                "stage": "queued", "detail": "Queued", "started": now, "updated": now,
+                "done": False, "cancel_requested": False, "cancelled": False, "error": None, "result": None,
+            }
+            try:
+                with patch.object(stt, "_transcribe_audio", return_value=("hello", "test")):
+                    asyncio.run(stt._run_stt_job(job_id, source, {}, "configured"))
+                self.assertTrue(stt._STT_JOBS[job_id]["done"])
+                self.assertEqual(stt._STT_JOBS[job_id]["result"]["text"], "hello")
+            finally:
+                stt._STT_JOBS.pop(job_id, None)
+
     def test_fish_cache_eviction_is_bounded(self):
         from pathlib import Path
         from tempfile import TemporaryDirectory