From e234a6e9a3020900fe9f016bcd3f8f7000d08dd5 Mon Sep 17 00:00:00 2001 From: mARTin-B78 Date: Fri, 4 Sep 2026 09:23:17 +0200 Subject: [PATCH] feat: persist background job metadata --- CHANGELOG.md | 3 ++ TODO.md | 3 +- VERSION | 2 +- core/database.py | 81 ++++++++++++++++++++++++++++++++++++++ routes/conversation.py | 24 ++++++++--- routes/stt.py | 9 ++++- routes/tts.py | 7 ++++ tests/test_conversation.py | 4 ++ tests/test_security.py | 17 ++++++++ 9 files changed, 142 insertions(+), 8 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d7739de..de646a7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -37,6 +37,9 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi normalization across named speakers and narration without requiring a live provider. - **Visual Pair Review Workflow.** The character library now exposes the read-only visual-pair audit directly in its toolbar and displays actionable per-character findings without changing saved assets. +- **Durable Background Job Metadata.** Transcription, voice design, casting, and consistent-image jobs now persist + bounded status metadata in SQLite; unfinished jobs are reported as interrupted after a server restart, while prompts, + secrets, audio, and image payloads are excluded from persistence. - **STT Cancellation Safety.** Queued transcription cancels before consuming the single STT slot, and timed-out requests keep the device slot held until the underlying worker actually exits. - **Credential Boundary Review.** Provider credentials are now documented as server-only; only the explicitly diff --git a/TODO.md b/TODO.md index 24bab3a..7e1a68f 100644 --- a/TODO.md +++ b/TODO.md @@ -64,7 +64,8 @@ the first block before investing in architecture or polish. the 4 GB shared-pool cap; physical VM/container and resource measurements remain. - [ ] **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. Durable persistence and retry policy still need to be added. + synchronous callers. Job metadata now persists in SQLite and marks interrupted work after restart; durable binary + result storage and a unified retry policy still need to be added. - [ ] **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*. diff --git a/VERSION b/VERSION index 229a27c..d28b1eb 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.22.8 +1.22.9 diff --git a/core/database.py b/core/database.py index cd1c404..64eb0f2 100644 --- a/core/database.py +++ b/core/database.py @@ -12,6 +12,7 @@ import json import os import sqlite3 import tempfile +import time import uuid from datetime import datetime, timezone from pathlib import Path @@ -89,6 +90,21 @@ CREATE TABLE IF NOT EXISTS generation_history ( ); CREATE INDEX IF NOT EXISTS idx_generation_history_target ON generation_history(target_id, created DESC); +CREATE TABLE IF NOT EXISTS background_jobs ( + job_id TEXT PRIMARY KEY, + kind TEXT NOT NULL DEFAULT '', + stage TEXT NOT NULL DEFAULT 'queued', + detail TEXT NOT NULL DEFAULT '', + done INTEGER NOT NULL DEFAULT 0, + cancelled INTEGER NOT NULL DEFAULT 0, + error TEXT, + result TEXT, + fingerprint TEXT NOT NULL DEFAULT '', + created REAL NOT NULL, + updated REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_background_jobs_updated ON background_jobs(updated DESC); + CREATE TABLE IF NOT EXISTS voice_index ( voice_id TEXT PRIMARY KEY, path TEXT NOT NULL, @@ -143,6 +159,13 @@ def _init() -> None: "INSERT OR IGNORE INTO schema_migrations(version, applied_at) VALUES (?, ?)", (1, datetime.now(timezone.utc).isoformat()), ) + # A process restart cannot safely resume provider calls, but it must + # not leave clients polling a job that can never finish. + conn.execute( + "UPDATE background_jobs SET stage='error', detail='Interrupted by server restart', done=1, error=? " + "WHERE done=0", + ("Background job interrupted by server restart",), + ) conn.commit() @@ -206,6 +229,64 @@ def generation_history_list(target_id: str = "", limit: int = 100) -> list[dict] return [dict(row) for row in rows] +def _safe_background_result(value: Any) -> str | None: + if value is None: + return None + + def clean(item: Any) -> Any: + if isinstance(item, dict): + return {str(key): clean(val) for key, val in item.items() + if str(key).lower() not in {"prompt", "text", "content", "audio", "image", "profile", "reference"}} + if isinstance(item, (list, tuple)): + return [clean(val) for val in item[:100]] + if isinstance(item, (str, int, float, bool)) or item is None: + return item if not isinstance(item, str) else item[:500] + return str(item)[:200] + + return _j(clean(value))[:12000] + + +def background_job_create(job_id: str, kind: str, *, fingerprint: str = "", created: float | None = None) -> None: + now = float(created if created is not None else time.time()) + with _open() as conn: + conn.execute( + "INSERT OR REPLACE INTO background_jobs(job_id,kind,stage,detail,done,cancelled,error,result,fingerprint,created,updated) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?)", + (str(job_id), str(kind)[:80], "queued", "Queued", 0, 0, None, None, str(fingerprint)[:128], now, now), + ) + conn.commit() + + +def background_job_update(job_id: str, *, stage: str | None = None, detail: str | None = None, + done: bool | None = None, cancelled: bool | None = None, + error: str | None = None, result: Any = None, updated: float | None = None) -> None: + fields = {"updated": float(updated if updated is not None else time.time())} + if stage is not None: fields["stage"] = str(stage)[:40] + if detail is not None: fields["detail"] = str(detail)[:500] + if done is not None: fields["done"] = int(bool(done)) + if cancelled is not None: fields["cancelled"] = int(bool(cancelled)) + if error is not None: fields["error"] = str(error)[:1000] + if result is not None: fields["result"] = _safe_background_result(result) + assignments = ", ".join(f"{key}=?" for key in fields) + with _open() as conn: + conn.execute(f"UPDATE background_jobs SET {assignments} WHERE job_id=?", (*fields.values(), str(job_id))) + conn.commit() + + +def background_job_get(job_id: str) -> dict | None: + with _open() as conn: + row = conn.execute("SELECT * FROM background_jobs WHERE job_id=?", (str(job_id),)).fetchone() + if row is None: + return None + result = dict(row) + result["done"] = bool(result["done"]) + result["cancelled"] = bool(result["cancelled"]) + if result.get("result"): + try: result["result"] = json.loads(result["result"]) + except (TypeError, ValueError): pass + return result + + def _row_to_char(row: sqlite3.Row) -> dict: d = dict(row) d["sheet"] = json.loads(d.get("sheet") or "{}") diff --git a/routes/conversation.py b/routes/conversation.py index 1375ed5..eb73692 100644 --- a/routes/conversation.py +++ b/routes/conversation.py @@ -33,6 +33,7 @@ from core.voice import ( _voice_audio_files, _is_internal_voice_file, ) from core.voice_index import rebuild_voice_index +from core.database import background_job_create, background_job_get, background_job_update from core.tts_helpers import _preview_request_audio from routes.stt import _transcribe_audio, _clean_stt_backend @@ -92,6 +93,9 @@ def _cast_job_set(job_id: str, **fields) -> None: if job is not None: job.update(fields) job["updated"] = time.time() + background_job_update(job_id, stage=fields.get("stage"), detail=fields.get("detail"), + done=fields.get("done"), cancelled=fields.get("cancelled"), + error=fields.get("error"), result=fields.get("result")) def _cast_job_trim() -> None: @@ -171,6 +175,7 @@ def _cij_new(job_id: str, name: str) -> None: for key, mapped_id in list(_consistent_image_idempotency.items()): if mapped_id == stale_id: _consistent_image_idempotency.pop(key, None) + background_job_create(job_id, "consistent_images", created=now) def _cij_set(job_id: str, **fields) -> None: @@ -180,6 +185,10 @@ def _cij_set(job_id: str, **fields) -> None: return job.update(fields) job["updated"] = time.time() + if fields: + background_job_update(job_id, stage=fields.get("stage"), detail=fields.get("detail"), + done=fields.get("done"), cancelled=fields.get("cancelled"), + error=fields.get("error"), result=fields.get("result")) class _ConsistentImageCancelled(Exception): @@ -697,6 +706,7 @@ async def match_characters_voices(request: Request): "done": False, "cancel_requested": False, "cancelled": False, "error": None, "result": None, "request_fingerprint": job_fingerprint, } + background_job_create(job_id, "casting", fingerprint=job_fingerprint, created=now) if idempotency_key: with _CAST_IDEMPOTENCY_LOCK: _CAST_JOB_IDEMPOTENCY[idempotency_key] = (job_id, now) @@ -833,10 +843,12 @@ async def match_characters_voices(request: Request): @router.get("/api/match-characters-voices/status/{job_id}") async def match_characters_voices_status(job_id: str): job = _CAST_JOBS.get(job_id) + if job is None: + job = background_job_get(job_id) if job is None: raise HTTPException(404, "Unknown or expired casting job") payload = dict(job) - payload["elapsed"] = round(time.time() - payload.pop("started"), 1) + payload["elapsed"] = round(time.time() - payload.pop("started", payload.get("created", time.time())), 1) payload.pop("cancel_requested", None) payload.pop("request_fingerprint", None) return payload @@ -3416,11 +3428,13 @@ async def character_generate_consistent_images_status(job_id: str): """ with _consistent_image_jobs_lock: job = _consistent_image_jobs.get(job_id) - if job is None: - raise HTTPException(404, "Unknown or expired job id") - payload = dict(job) + payload = dict(job) if job is not None else None + if payload is None: + payload = background_job_get(job_id) + if payload is None: + raise HTTPException(404, "Unknown or expired job id") now = time.time() - payload["elapsed"] = round(now - payload.pop("started"), 1) + payload["elapsed"] = round(now - payload.pop("started", payload.get("created", now)), 1) payload["stale_seconds"] = round(now - payload.pop("updated"), 1) return payload diff --git a/routes/stt.py b/routes/stt.py index 30521a5..b1a663c 100644 --- a/routes/stt.py +++ b/routes/stt.py @@ -21,6 +21,7 @@ from core.constants import ( from core.registry import _registry_get, TEMP_DIR from core.validation import _validate_http_url, _copy_limited, _validated_upload_suffix from core.audio import _to_wav_16k +from core.database import background_job_create, background_job_get, background_job_update from core.voice import _AUDIO_EXTS, _UPLOAD_EXTS router = APIRouter() @@ -47,6 +48,9 @@ def _stt_job_set(job_id: str, **fields) -> None: if job is not None: job.update(fields) job["updated"] = time.time() + background_job_update(job_id, stage=fields.get("stage"), detail=fields.get("detail"), + done=fields.get("done"), cancelled=fields.get("cancelled"), + error=fields.get("error"), result=fields.get("result")) async def _run_stt_job(job_id: str, src: Path, settings: dict, backend: str) -> None: @@ -547,6 +551,7 @@ async def transcribe(request: Request): "stage": "queued", "detail": "Queued…", "started": now, "updated": now, "done": False, "cancel_requested": False, "cancelled": False, "error": None, "result": None, } + background_job_create(job_id, "transcription", created=now) _stt_jobs_trim() asyncio.create_task(_run_stt_job(job_id, src, settings, backend)) return {"job_id": job_id} @@ -560,10 +565,12 @@ async def transcribe(request: Request): @router.get("/api/transcribe/status/{job_id}") async def transcribe_status(job_id: str): job = _STT_JOBS.get(job_id) + if job is None: + job = background_job_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["elapsed"] = round(time.time() - payload.pop("started", payload.get("created", time.time())), 1) payload.pop("cancel_requested", None) return payload diff --git a/routes/tts.py b/routes/tts.py index e95a8c2..92e302f 100644 --- a/routes/tts.py +++ b/routes/tts.py @@ -49,6 +49,7 @@ from core.tts_helpers import ( _fishspeech_request_audio, ) from core.audio import _duration, _export_normalized_wav +from core.database import background_job_create, background_job_get, background_job_update from core.docker_client import _docker_available, _docker_post logger = logging.getLogger("uvicorn.error") @@ -80,6 +81,9 @@ def _voice_design_job_set(job_id: str, **fields) -> None: if job is not None: job.update(fields) job["updated"] = time.time() + background_job_update(job_id, stage=fields.get("stage"), detail=fields.get("detail"), + done=fields.get("done"), cancelled=fields.get("cancelled"), + error=fields.get("error"), result=fields.get("result")) def _voice_design_job_trim() -> None: @@ -890,6 +894,7 @@ async def voice_design(request: Request): "job_id": job_id, "stage": "queued", "detail": "Queued", "done": False, "created": time.time(), "updated": time.time(), "cancel_requested": False, } + background_job_create(job_id, "voice_design", created=_VOICE_DESIGN_JOBS[job_id]["created"]) _voice_design_job_trim() asyncio.create_task(_run_voice_design_job(job_id, job_data, idempotency_key, request_fingerprint)) return {"job_id": job_id, "stage": "queued"} @@ -899,6 +904,8 @@ async def voice_design(request: Request): @router.get("/api/voice-design/status/{job_id}") async def voice_design_status(job_id: str): job = _VOICE_DESIGN_JOBS.get(job_id) + if job is None: + job = background_job_get(job_id) if job is None: raise HTTPException(404, "Voice design job not found") return dict(job) diff --git a/tests/test_conversation.py b/tests/test_conversation.py index 3e07238..ba8bda7 100644 --- a/tests/test_conversation.py +++ b/tests/test_conversation.py @@ -82,6 +82,10 @@ class AttributionParserTests(unittest.TestCase): finally: _consistent_image_jobs.pop(job_id, None) _consistent_image_cancel_events.pop(job_id, None) + from core.database import _open + with _open() as conn: + conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,)) + conn.commit() if __name__ == "__main__": diff --git a/tests/test_security.py b/tests/test_security.py index 0b3f919..e382c2f 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -177,6 +177,23 @@ class SecurityValidationTests(unittest.TestCase): finally: conversation._CAST_JOBS.pop(job_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" + background_job_create(job_id, "test", fingerprint="fingerprint") + try: + background_job_update(job_id, stage="done", detail="Done", done=True, + result={"text": "must not persist", "id": "result-id", "image": "data:image/png;base64,huge"}) + job = background_job_get(job_id) + self.assertTrue(job["done"]) + self.assertEqual(job["result"], {"id": "result-id"}) + self.assertEqual(job["fingerprint"], "fingerprint") + finally: + from core.database import _open + with _open() as conn: + conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,)) + conn.commit() + def test_fish_cache_eviction_is_bounded(self): from pathlib import Path from tempfile import TemporaryDirectory