feat: add safe retries for background jobs

CHANGELOG.md

@@ -8,6 +8,8 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
 ## [Unreleased]
 
 ### Added
+- **Safe Background Retries.** Failed live transcription, voice-design, casting, and consistent-image jobs now expose
+  bounded retry actions from stripped in-memory inputs; retry payloads are never returned or persisted and expire on restart.
 - **Canonical Pair Asset Digests.** Generated profile/reference pairs now retain SHA-256 asset digests in character
   metadata, and visual-pair review reports either asset being replaced after generation.
 - **Live Image Result Preservation.** Durable image artifact persistence now leaves the active in-memory pair

TODO.md

@@ -64,12 +64,14 @@ the first block before investing in architecture or polish.
   the 4 GB shared-pool cap, and the current `1.23.3` image served successfully under a hard 4 GiB cgroup limit
   without OOM; desktop Chromium now has a 4× CPU-throttled navigation regression with DOM/audio/heap probes,
   while full workload resource measurements remain.
-- [ ] **L — Add background jobs and bounded queues.** Image generation already has the full interactive job path,
+- [x] **L — Add background jobs and bounded queues.** Image generation already has the full interactive job path,
   and transcription, voice design, and casting now have optional ID/status/cancel/timeout paths while preserving
   synchronous callers. Job metadata now persists in SQLite and marks interrupted work after restart; durable binary
   result storage remains; outbound TTS and LLM calls now share a bounded transient-failure retry policy, and `/api/jobs`
   exposes sanitized recent job history. Completed voice-design audio and consistent-image outputs now use bounded,
   atomic artifact files with authenticated download URLs; transcription text remains deliberately non-persistent.
+  Failed live transcription, voice-design, and casting jobs now offer bounded in-memory retries; after restart they
+  correctly require reopening the source workflow.
 - [ ] **L — Run the Book 2 generalization regression.** Added a neutral multi-speaker fixture and parser regression;
   a real Book 2 auto-cast/attribution output still needs to be captured and reviewed to confirm rules are not overfit
   to *Der Sturm*.
@@ -133,10 +135,11 @@ 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. 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] Introduce background jobs for voice design, transcription, casting, and image generation. Synchronous callers
+  remain supported while each long-running path has an optional bounded background job.
+- [x] Give every job an ID, progress, cancellation, timeout, retry, and idempotency key. Failed live jobs offer
+  bounded retries from stripped in-memory inputs; retry data is never returned or persisted, and restart correctly
+  requires reopening the source workflow.
 - [x] Prevent duplicate generation when a user clicks twice for image, voice-design, and casting submissions.
 - [ ] Centralize provider health checks and error handling.
 - [x] Limit concurrency to 1–2 image generations, 1–2 voice generations, and 1 STT request per user. Consistent

VERSION

@@ -1 +1 @@
-1.24.8
+1.24.9

routes/conversation.py

@@ -89,6 +89,14 @@ class _CastJobRequest:
         return self._data
 
 
+class _ConsistentImageRetryRequest:
+    def __init__(self, data: dict):
+        self._data = data
+        self.headers = {}
+    async def json(self):
+        return self._data
+
+
 def _cast_job_set(job_id: str, **fields) -> None:
     job = _CAST_JOBS.get(job_id)
     if job is not None:
@@ -109,6 +117,10 @@ def _cast_job_trim() -> None:
                 break
 
 
+def _cast_public_job(job: dict) -> dict:
+    return {key: value for key, value in dict(job).items() if key not in {"retry_data", "request_fingerprint"}}
+
+
 def _cast_job_run_sync(data: dict, headers: dict[str, str]):
     return asyncio.run(match_characters_voices(_CastJobRequest(data, headers)))
 
@@ -227,6 +239,15 @@ def _data_image_digest(value: object) -> str:
         return ""
 
 
+def _strip_image_payloads(value):
+    if isinstance(value, dict):
+        return {key: _strip_image_payloads(item) for key, item in value.items()
+                if key not in {"image", "concept_art_image", "identity_image", "profile_asset_digest", "reference_asset_digest"}}
+    if isinstance(value, list):
+        return [_strip_image_payloads(item) for item in value[:20]]
+    return value
+
+
 class _ConsistentImageCancelled(Exception):
     """Internal cooperative cancellation signal for a consistent-image worker."""
 
@@ -741,6 +762,7 @@ async def match_characters_voices(request: Request):
             "job_id": job_id, "stage": "queued", "detail": "Queued", "started": now, "updated": now,
             "done": False, "cancel_requested": False, "cancelled": False, "error": None, "result": None,
             "request_fingerprint": job_fingerprint,
+            "retry_data": dict(job_data),
         }
         background_job_create(job_id, "casting", fingerprint=job_fingerprint, created=now)
         if idempotency_key:
@@ -887,6 +909,7 @@ async def match_characters_voices_status(job_id: str):
     payload["elapsed"] = round(time.time() - payload.pop("started", payload.get("created", time.time())), 1)
     payload.pop("cancel_requested", None)
     payload.pop("request_fingerprint", None)
+    payload.pop("retry_data", None)
     return payload
 
 
@@ -902,6 +925,20 @@ async def match_characters_voices_cancel(job_id: str):
     return {"ok": True, "cancelled": True}
 
 
+@router.post("/api/match-characters-voices/retry/{job_id}")
+async def match_characters_voices_retry(job_id: str):
+    job = _CAST_JOBS.get(job_id)
+    if job is None:
+        raise HTTPException(404, "Original casting input is no longer available")
+    if not job.get("done") or not job.get("error"):
+        raise HTTPException(409, "Only failed casting jobs can be retried")
+    data = dict(job.get("retry_data") or {})
+    if not data.get("names") or not data.get("voices"):
+        raise HTTPException(410, "Original casting input is no longer available")
+    data["background"] = True
+    return await match_characters_voices(_CastJobRequest(data, {}))
+
+
 def _extract_json_block(text: str) -> str:
     """Pull the first {...} JSON object out of an LLM response."""
     text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
@@ -3376,6 +3413,13 @@ async def character_generate_consistent_images(request: Request):
     # fetch that a page reload would then throw away mid-render.
     job_id = uuid.uuid4().hex
     _cij_new(job_id, name or "Character")
+    with _consistent_image_jobs_lock:
+        _consistent_image_jobs[job_id]["retry_data"] = {
+            "name": name, "book": book, "sheet": _strip_image_payloads(sheet),
+            "profile_prompt": saved_profile_prompt, "reference_prompt": saved_reference_prompt,
+            "visual_style": visual_style, "book_profile": book_profile,
+            "identity_image": "", "reuse_identity": False,
+        }
     if idempotency_key:
         with _consistent_image_jobs_lock:
             _consistent_image_idempotency[idempotency_key] = job_id
@@ -3475,6 +3519,7 @@ async def character_generate_consistent_images_status(job_id: str):
     now = time.time()
     payload["elapsed"] = round(now - payload.pop("started", payload.get("created", now)), 1)
     payload["stale_seconds"] = round(now - payload.pop("updated"), 1)
+    payload.pop("retry_data", None)
     return payload
 
 
@@ -3499,6 +3544,20 @@ async def character_generate_consistent_images_cancel(job_id: str):
     return {"ok": True, "cancelled": True}
 
 
+@router.post("/api/character-generate-consistent-images/retry/{job_id}")
+async def character_generate_consistent_images_retry(job_id: str):
+    with _consistent_image_jobs_lock:
+        job = _consistent_image_jobs.get(job_id)
+        retry_data = dict(job.get("retry_data") or {}) if job else {}
+    if job is None:
+        raise HTTPException(404, "Original image-generation input is no longer available")
+    if not job.get("done") or not job.get("error"):
+        raise HTTPException(409, "Only failed image jobs can be retried")
+    if not retry_data.get("profile_prompt") or not retry_data.get("reference_prompt"):
+        raise HTTPException(410, "Original image-generation input is no longer available")
+    return await character_generate_consistent_images(_ConsistentImageRetryRequest({**retry_data, "background": False}))
+
+
 def _character_generate_image_sync(provider: str, model: str, prompt: str, keys: dict, _settings: dict, image_kind: str = "portrait") -> dict:
     if provider == "openai":
         api_key = (keys.get("openai_image") or "").strip()

routes/jobs.py

@@ -33,6 +33,23 @@ def _public_job(job: dict) -> dict:
     payload.pop("fingerprint", None)
     payload.pop("created", None)
     payload.pop("updated", None)
+    retry_available = False
+    if payload.get("done") and payload.get("error"):
+        maps = {
+            "transcription": ("routes.stt", "_STT_JOBS"),
+            "voice_design": ("routes.tts", "_VOICE_DESIGN_JOBS"),
+            "casting": ("routes.conversation", "_CAST_JOBS"),
+            "consistent_images": ("routes.conversation", "_consistent_image_jobs"),
+        }
+        module_name, map_name = maps.get(str(payload.get("kind")), (None, None))
+        if module_name:
+            try:
+                import importlib
+                live = getattr(importlib.import_module(module_name), map_name).get(str(payload.get("job_id")), {})
+                retry_available = bool(live.get("retry_data") or live.get("retry_source_id"))
+            except Exception:
+                retry_available = False
+    payload["retry_available"] = retry_available
     result = payload.get("result")
     if isinstance(result, dict):
         result = dict(result)

routes/stt.py

@@ -31,6 +31,13 @@ _STT_JOB_MAX = 128
 _STT_JOB_TIMEOUT = 15 * 60
 
 
+class _SttRetryRequest:
+    def __init__(self, data: dict):
+        self._data = data
+    async def json(self):
+        return self._data
+
+
 async def _bounded_stt_call(fn, *args):
     """Serialize heavyweight STT work so one process cannot overcommit a GPU."""
     try:
@@ -550,6 +557,7 @@ async def transcribe(request: Request):
         _STT_JOBS[job_id] = {
             "stage": "queued", "detail": "Queued…", "started": now, "updated": now,
             "done": False, "cancel_requested": False, "cancelled": False, "error": None, "result": None,
+            "retry_source_id": fid, "retry_backend": backend,
         }
         background_job_create(job_id, "transcription", created=now)
         _stt_jobs_trim()
@@ -572,6 +580,8 @@ async def transcribe_status(job_id: str):
     payload = dict(job)
     payload["elapsed"] = round(time.time() - payload.pop("started", payload.get("created", time.time())), 1)
     payload.pop("cancel_requested", None)
+    payload.pop("retry_source_id", None)
+    payload.pop("retry_backend", None)
     return payload
 
 
@@ -587,6 +597,20 @@ async def transcribe_cancel(job_id: str):
     return {"ok": True, "cancelled": True}
 
 
+@router.post("/api/transcribe/retry/{job_id}")
+async def transcribe_retry(job_id: str):
+    job = _STT_JOBS.get(job_id)
+    if job is None:
+        raise HTTPException(404, "Original transcription input is no longer available")
+    if not job.get("done") or not job.get("error"):
+        raise HTTPException(409, "Only failed transcription jobs can be retried")
+    source_id = str(job.get("retry_source_id") or "")
+    source = _registry_get(source_id)
+    if not source or not source.exists():
+        raise HTTPException(410, "Original transcription audio is no longer available")
+    return await transcribe(_SttRetryRequest({"id": source_id, "backend": job.get("retry_backend"), "background": True}))
+
+
 @router.post("/api/transcribe-bytes")
 async def transcribe_bytes(
     file: UploadFile = File(...),

routes/tts.py

@@ -64,6 +64,14 @@ _VOICE_DESIGN_JOB_MAX = 128
 _VOICE_DESIGN_JOB_TIMEOUT = 15 * 60
 
 
+class _VoiceDesignRetryRequest:
+    def __init__(self, data: dict):
+        self._data = data
+        self.headers = {}
+    async def json(self):
+        return self._data
+
+
 async def _bounded_tts_call(fn, *args):
     """Run one blocking synthesis through the process-local voice queue."""
     try:
@@ -96,6 +104,10 @@ def _voice_design_job_trim() -> None:
                 break
 
 
+def _voice_design_public_job(job: dict) -> dict:
+    return {key: value for key, value in dict(job).items() if key != "retry_data"}
+
+
 async def _voice_design_execute(data: dict, idempotency_key: str, request_fingerprint: str) -> dict:
     instruct: str = data.get("instruct", "").strip()
     sample_text: str = data.get("sample_text", "Hello! This is a voice design sample.").strip()
@@ -906,6 +918,7 @@ async def voice_design(request: Request):
         _VOICE_DESIGN_JOBS[job_id] = {
             "job_id": job_id, "stage": "queued", "detail": "Queued", "done": False,
             "created": time.time(), "updated": time.time(), "cancel_requested": False,
+            "retry_data": dict(job_data),
         }
         background_job_create(job_id, "voice_design", created=_VOICE_DESIGN_JOBS[job_id]["created"])
         _voice_design_job_trim()
@@ -921,7 +934,7 @@ async def voice_design_status(job_id: str):
         job = background_job_get(job_id)
     if job is None:
         raise HTTPException(404, "Voice design job not found")
-    return dict(job)
+    return _voice_design_public_job(job)
 
 
 @router.post("/api/voice-design/cancel/{job_id}")
@@ -930,10 +943,24 @@ async def voice_design_cancel(job_id: str):
     if job is None:
         raise HTTPException(404, "Voice design job not found")
     if job.get("done"):
-        return dict(job)
+        return _voice_design_public_job(job)
     job["cancel_requested"] = True
     _voice_design_job_set(job_id, detail="Cancellation requested")
-    return dict(_VOICE_DESIGN_JOBS[job_id])
+    return _voice_design_public_job(_VOICE_DESIGN_JOBS[job_id])
+
+
+@router.post("/api/voice-design/retry/{job_id}")
+async def voice_design_retry(job_id: str):
+    job = _VOICE_DESIGN_JOBS.get(job_id)
+    if job is None:
+        raise HTTPException(404, "Original voice design input is no longer available")
+    if not job.get("done") or not job.get("error"):
+        raise HTTPException(409, "Only failed voice design jobs can be retried")
+    data = dict(job.get("retry_data") or {})
+    if not data.get("instruct"):
+        raise HTTPException(410, "Original voice design input is no longer available")
+    data["background"] = True
+    return await voice_design(_VoiceDesignRetryRequest(data))
 
 
 @router.post("/api/tts-route-test")

static/js/jobs.js

@@ -32,6 +32,10 @@
     const paths = { transcription: '/api/transcribe/cancel/', voice_design: '/api/voice-design/cancel/', casting: '/api/match-characters-voices/cancel/', consistent_images: '/api/character-generate-consistent-images/cancel/' };
     return paths[kind] ? paths[kind] + encodeURIComponent(id) : '';
   }
+  function retryEndpoint(kind, id) {
+    const paths = { transcription: '/api/transcribe/retry/', voice_design: '/api/voice-design/retry/', casting: '/api/match-characters-voices/retry/', consistent_images: '/api/character-generate-consistent-images/retry/' };
+    return paths[kind] && id ? paths[kind] + encodeURIComponent(id) : '';
+  }
   function render(jobs) {
     list.replaceChildren();
     if (!jobs.length) {
@@ -73,6 +77,22 @@
       row.appendChild(main);
       const actions = document.createElement('div');
       actions.className = 'background-job-actions';
+      if (job.done && job.error && job.retry_available) {
+        const button = document.createElement('button');
+        button.type = 'button'; button.className = 'btn-secondary btn-sm'; button.textContent = 'Retry';
+        button.addEventListener('click', async () => {
+          button.disabled = true;
+          try {
+            const response = await fetch(retryEndpoint(job.kind, job.job_id), { method: 'POST' });
+            if (!response.ok) { const detail = await response.json().catch(() => ({})); throw new Error(detail.detail || `HTTP ${response.status}`); }
+            await load();
+          } catch (error) {
+            button.disabled = false;
+            if (typeof toast === 'function') toast('Could not retry job: ' + error.message, 'error');
+          }
+        });
+        actions.appendChild(button);
+      }
       if (!job.done && !job.cancelled && job.job_id) {
         const endpoint = cancelEndpoint(job.kind, job.job_id);
         if (endpoint) {
@@ -97,7 +117,7 @@
       const note = node('Reopen source to retry');
       note.className = 'background-job-retry-note';
       note.title = 'Inputs are not retained for replay, by design.';
-      actions.appendChild(note);
+      if (!job.done || !job.error || !job.retry_available) actions.appendChild(note);
       row.appendChild(actions);
       list.appendChild(row);
     });

tests/test_security.py

@@ -226,6 +226,42 @@ class SecurityValidationTests(unittest.TestCase):
         finally:
             conversation._CAST_JOBS.pop(job_id, None)
 
+    def test_live_retry_metadata_is_available_without_exposing_input(self):
+        import routes.tts as tts
+        import routes.conversation as conversation
+        from routes.jobs import _public_job
+        voice_id = "retry-voice-test"
+        cast_id = "retry-cast-test"
+        tts._VOICE_DESIGN_JOBS[voice_id] = {"job_id": voice_id, "kind": "voice_design", "done": True, "error": "temporary", "retry_data": {"instruct": "private prompt"}}
+        conversation._CAST_JOBS[cast_id] = {"job_id": cast_id, "kind": "casting", "done": True, "error": "temporary", "retry_data": {"script": "private source"}}
+        try:
+            self.assertNotIn("retry_data", tts._voice_design_public_job(tts._VOICE_DESIGN_JOBS[voice_id]))
+            public_voice = _public_job({"job_id": voice_id, "kind": "voice_design", "done": True, "error": "temporary", "created": 0})
+            self.assertTrue(public_voice["retry_available"])
+            public_cast = _public_job({"job_id": cast_id, "kind": "casting", "done": True, "error": "temporary", "created": 0})
+            self.assertTrue(public_cast["retry_available"])
+        finally:
+            tts._VOICE_DESIGN_JOBS.pop(voice_id, None)
+            conversation._CAST_JOBS.pop(cast_id, None)
+
+    def test_failed_live_jobs_dispatch_safe_retries(self):
+        from unittest.mock import AsyncMock, patch
+        import routes.tts as tts
+        import routes.conversation as conversation
+        voice_id, cast_id = "dispatch-voice-retry", "dispatch-cast-retry"
+        tts._VOICE_DESIGN_JOBS[voice_id] = {"job_id": voice_id, "done": True, "error": "temporary", "retry_data": {"instruct": "voice"}}
+        conversation._CAST_JOBS[cast_id] = {"job_id": cast_id, "done": True, "error": "temporary", "retry_data": {"names": ["A"], "voices": [{"id": "v1"}]}}
+        try:
+            with patch.object(tts, "voice_design", new=AsyncMock(return_value={"job_id": "new-voice"})) as voice_route:
+                self.assertEqual(asyncio.run(tts.voice_design_retry(voice_id)), {"job_id": "new-voice"})
+                self.assertTrue(voice_route.await_args.args[0]._data["background"])
+            with patch.object(conversation, "match_characters_voices", new=AsyncMock(return_value={"job_id": "new-cast"})) as cast_route:
+                self.assertEqual(asyncio.run(conversation.match_characters_voices_retry(cast_id)), {"job_id": "new-cast"})
+                self.assertTrue(cast_route.await_args.args[0]._data["background"])
+        finally:
+            tts._VOICE_DESIGN_JOBS.pop(voice_id, None)
+            conversation._CAST_JOBS.pop(cast_id, None)
+
     def test_background_job_metadata_round_trips_without_large_payloads(self):
         from core.database import background_job_create, background_job_get, background_job_update
         job_id = "persistent-job-test"