feat: persist background job result artifacts

This commit is contained in:
mARTin-B78 2026-09-04 09:48:37 +02:00
parent b648bbdf0b
commit 690673625d
11 changed files with 171 additions and 12 deletions

View File

@ -10,7 +10,8 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
### Added ### Added
- **Background Job History Panel.** The Benchmark section now shows sanitized, restart-safe background-job status, - **Background Job History Panel.** The Benchmark section now shows sanitized, restart-safe background-job status,
progress, elapsed time, errors, and cancellation controls. Replay is deliberately limited to reopening the source progress, elapsed time, errors, and cancellation controls. Replay is deliberately limited to reopening the source
workflow because prompts, text, audio, and image inputs are never retained for implicit retries. workflow because prompts, text, audio, and image inputs are never retained for implicit retries. Completed voice-design
audio and consistent-image outputs are now stored as bounded atomic artifacts with authenticated download links.
- **Protected Backup Recovery.** Added an authenticated backup inventory and restore API for character and rehearsal - **Protected Backup Recovery.** Added an authenticated backup inventory and restore API for character and rehearsal
snapshots, with path-safe file selection and atomic pre-mutation backups. snapshots, with path-safe file selection and atomic pre-mutation backups.
- **Voice Trash Recovery.** Voice deletes and group deletes now expose a path-safe trash inventory and restore API, - **Voice Trash Recovery.** Voice deletes and group deletes now expose a path-safe trash inventory and restore API,

View File

@ -67,7 +67,8 @@ the first block before investing in architecture or polish.
and transcription, voice design, and casting now have optional ID/status/cancel/timeout paths while preserving 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 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` result storage remains; outbound TTS and LLM calls now share a bounded transient-failure retry policy, and `/api/jobs`
exposes sanitized recent job history. 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.
- [ ] **L — Run the Book 2 generalization regression.** Added a neutral multi-speaker fixture and parser regression; - [ ] **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 a real Book 2 auto-cast/attribution output still needs to be captured and reviewed to confirm rules are not overfit
to *Der Sturm*. to *Der Sturm*.

View File

@ -1 +1 @@
1.24.3 1.24.4

View File

@ -10,6 +10,8 @@ from __future__ import annotations
import json import json
import os import os
import re
import shutil
import sqlite3 import sqlite3
import tempfile import tempfile
import time import time
@ -22,6 +24,8 @@ from core.constants import CONFIG_DIR
from core.validation import _validate_character_record from core.validation import _validate_character_record
_DB_PATH = CONFIG_DIR / "tts_creator.db" _DB_PATH = CONFIG_DIR / "tts_creator.db"
_BACKGROUND_ARTIFACT_DIR = CONFIG_DIR / "background-artifacts"
_BACKGROUND_ARTIFACT_MAX_BYTES = 50 * 1024 * 1024
_DDL = """ _DDL = """
CREATE TABLE IF NOT EXISTS characters ( CREATE TABLE IF NOT EXISTS characters (
@ -246,6 +250,55 @@ def _safe_background_result(value: Any) -> str | None:
return _j(clean(value))[:12000] return _j(clean(value))[:12000]
def background_job_store_artifact(job_id: str, name: str, payload: bytes, media_type: str = "application/octet-stream") -> dict:
"""Persist one bounded binary result outside SQLite and return safe metadata."""
safe_job = str(job_id or "")
safe_name = str(name or "")
if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", safe_job) or not re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9]{1,8}", safe_name):
raise ValueError("invalid background artifact name")
data = bytes(payload or b"")
if not data:
raise ValueError("background artifact is empty")
if len(data) > _BACKGROUND_ARTIFACT_MAX_BYTES:
raise ValueError("background artifact exceeds 50 MB")
target_dir = _BACKGROUND_ARTIFACT_DIR / safe_job
target_dir.mkdir(parents=True, exist_ok=True)
target = target_dir / safe_name
fd, tmp_name = tempfile.mkstemp(prefix=f".{safe_name}.", suffix=".tmp", dir=target_dir)
try:
with os.fdopen(fd, "wb") as handle:
handle.write(data)
handle.flush()
os.fsync(handle.fileno())
os.replace(tmp_name, target)
except Exception:
try: os.unlink(tmp_name)
except OSError: pass
raise
return {"name": safe_name, "media_type": str(media_type or "application/octet-stream")[:100], "size": len(data)}
def background_job_artifact_path(job_id: str, name: str) -> Path | None:
"""Resolve an artifact only when it belongs to an existing job."""
if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", str(job_id or "")) or not re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9]{1,8}", str(name or "")):
return None
with _open() as conn:
exists = conn.execute("SELECT 1 FROM background_jobs WHERE job_id=?", (str(job_id),)).fetchone()
if exists is None:
return None
target = (_BACKGROUND_ARTIFACT_DIR / str(job_id) / str(name)).resolve()
try: target.relative_to(_BACKGROUND_ARTIFACT_DIR.resolve())
except ValueError: return None
return target if target.is_file() else None
def _remove_background_artifacts(job_id: str) -> None:
"""Remove artifacts when the corresponding bounded history row expires."""
safe_job = str(job_id or "")
if re.fullmatch(r"[A-Za-z0-9_-]{1,80}", safe_job):
shutil.rmtree(_BACKGROUND_ARTIFACT_DIR / safe_job, ignore_errors=True)
def background_job_create(job_id: str, kind: str, *, fingerprint: str = "", created: float | None = None) -> None: 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()) now = float(created if created is not None else time.time())
with _open() as conn: with _open() as conn:
@ -254,11 +307,12 @@ def background_job_create(job_id: str, kind: str, *, fingerprint: str = "", crea
"VALUES (?,?,?,?,?,?,?,?,?,?,?)", "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(str(job_id), str(kind)[:80], "queued", "Queued", 0, 0, None, None, str(fingerprint)[:128], now, now), (str(job_id), str(kind)[:80], "queued", "Queued", 0, 0, None, None, str(fingerprint)[:128], now, now),
) )
conn.execute( expired = conn.execute(
"DELETE FROM background_jobs WHERE job_id IN (" "SELECT job_id FROM background_jobs WHERE done=1 ORDER BY updated ASC LIMIT -1 OFFSET 255"
"SELECT job_id FROM background_jobs WHERE done=1 ORDER BY updated ASC " ).fetchall()
"LIMIT -1 OFFSET 255)" for row in expired:
) conn.execute("DELETE FROM background_jobs WHERE job_id=?", (row["job_id"],))
_remove_background_artifacts(row["job_id"])
conn.commit() conn.commit()

View File

@ -33,7 +33,7 @@ from core.voice import (
_voice_audio_files, _is_internal_voice_file, _voice_audio_files, _is_internal_voice_file,
) )
from core.voice_index import rebuild_voice_index from core.voice_index import rebuild_voice_index
from core.database import background_job_create, background_job_get, background_job_update from core.database import background_job_create, background_job_get, background_job_store_artifact, background_job_update
from core.retry import request_with_retries from core.retry import request_with_retries
from core.tts_helpers import _preview_request_audio from core.tts_helpers import _preview_request_audio
from routes.stt import _transcribe_audio, _clean_stt_backend from routes.stt import _transcribe_audio, _clean_stt_backend
@ -192,6 +192,29 @@ def _cij_set(job_id: str, **fields) -> None:
error=fields.get("error"), result=fields.get("result")) error=fields.get("error"), result=fields.get("result"))
def _persist_consistent_image_artifacts(job_id: str, result: dict) -> dict:
"""Move generated data-URI images to bounded durable files before status storage."""
persisted = dict(result or {})
artifacts = []
for key, filename in (("profile", "profile.png"), ("reference", "reference.png")):
value = persisted.get(key)
if not isinstance(value, str) or not value.startswith("data:") or ";base64," not in value:
continue
header, encoded = value.split(",", 1)
media_type = header[5:].split(";", 1)[0] or "image/png"
try:
payload = base64.b64decode(encoded, validate=True)
artifacts.append({"kind": key, **background_job_store_artifact(job_id, filename, payload, media_type)})
persisted.pop(key, None)
except Exception as exc:
persisted.pop(key, None)
persisted.setdefault("artifact_error", "One or more generated images could not be persisted")
print(f"[background-artifact] {key} persistence failed: {exc}")
if artifacts:
persisted["artifacts"] = artifacts
return persisted
class _ConsistentImageCancelled(Exception): class _ConsistentImageCancelled(Exception):
"""Internal cooperative cancellation signal for a consistent-image worker.""" """Internal cooperative cancellation signal for a consistent-image worker."""
@ -3403,6 +3426,7 @@ async def character_generate_consistent_images(request: Request):
result["prompt_version"] = _CHARACTER_VISUAL_PROMPT_VERSION result["prompt_version"] = _CHARACTER_VISUAL_PROMPT_VERSION
result["identity_consistent"] = True result["identity_consistent"] = True
result["identity_fingerprint"] = _character_identity_fingerprint(sheet) result["identity_fingerprint"] = _character_identity_fingerprint(sheet)
result = _persist_consistent_image_artifacts(job_id, result)
_cij_set(job_id, stage="done", detail="Done", done=True, result=result) _cij_set(job_id, stage="done", detail="Done", done=True, result=result)
except HTTPException as exc: except HTTPException as exc:
_cij_set(job_id, stage="error", detail="Failed", done=True, error=str(exc.detail)) _cij_set(job_id, stage="error", detail="Failed", done=True, error=str(exc.detail))

View File

@ -4,8 +4,9 @@ from __future__ import annotations
import time import time
from fastapi import APIRouter, HTTPException from fastapi import APIRouter, HTTPException
from fastapi.responses import FileResponse
from core.database import background_job_get, background_job_list from core.database import background_job_artifact_path, background_job_get, background_job_list
router = APIRouter() router = APIRouter()
@ -15,6 +16,18 @@ def _public_job(job: dict) -> dict:
payload.pop("fingerprint", None) payload.pop("fingerprint", None)
payload.pop("created", None) payload.pop("created", None)
payload.pop("updated", None) payload.pop("updated", None)
result = payload.get("result")
if isinstance(result, dict):
result = dict(result)
if isinstance(result.get("artifact"), dict) and result["artifact"].get("name"):
result["artifact"] = {**result["artifact"], "url": f"/api/jobs/{job.get('job_id')}/artifacts/{result['artifact']['name']}"}
if isinstance(result.get("artifacts"), list):
result["artifacts"] = [
{**item, "url": f"/api/jobs/{job.get('job_id')}/artifacts/{item['name']}"}
if isinstance(item, dict) and item.get("name") else item
for item in result["artifacts"]
]
payload["result"] = result
payload["elapsed"] = round(time.time() - float(job.get("created") or time.time()), 1) payload["elapsed"] = round(time.time() - float(job.get("created") or time.time()), 1)
return payload return payload
@ -30,3 +43,17 @@ async def jobs_get(job_id: str):
if job is None: if job is None:
raise HTTPException(404, "Unknown or expired job") raise HTTPException(404, "Unknown or expired job")
return _public_job(job) return _public_job(job)
@router.get("/api/jobs/{job_id}/artifacts/{name}")
async def jobs_artifact(job_id: str, name: str):
path = background_job_artifact_path(job_id, name)
if path is None:
raise HTTPException(404, "Unknown or expired job artifact")
media_type = "application/octet-stream"
suffix = path.suffix.lower()
if suffix == ".wav": media_type = "audio/wav"
elif suffix == ".mp3": media_type = "audio/mpeg"
elif suffix == ".png": media_type = "image/png"
elif suffix in {".jpg", ".jpeg"}: media_type = "image/jpeg"
return FileResponse(path, media_type=media_type, headers={"Cache-Control": "private, no-store"})

View File

@ -49,7 +49,7 @@ from core.tts_helpers import (
_fishspeech_request_audio, _fishspeech_request_audio,
) )
from core.audio import _duration, _export_normalized_wav from core.audio import _duration, _export_normalized_wav
from core.database import background_job_create, background_job_get, background_job_update from core.database import background_job_create, background_job_get, background_job_store_artifact, background_job_update
from core.docker_client import _docker_available, _docker_post from core.docker_client import _docker_available, _docker_post
logger = logging.getLogger("uvicorn.error") logger = logging.getLogger("uvicorn.error")
@ -151,6 +151,19 @@ async def _run_voice_design_job(job_id: str, data: dict, idempotency_key: str, r
_voice_design_job_set(job_id, stage="cancelled", detail="Cancelled", done=True, cancelled=True, _voice_design_job_set(job_id, stage="cancelled", detail="Cancelled", done=True, cancelled=True,
error="Voice design cancelled") error="Voice design cancelled")
else: else:
# The registry is process-local and its temp file is not restart-safe.
# Keep a bounded durable copy while leaving prompts and source text out
# of SQLite and out of the public job status payload.
try:
from core.registry import _registry_get
source = _registry_get(str((result or {}).get("id") or ""))
if source and source.is_file():
result = dict(result or {})
result["artifact"] = background_job_store_artifact(job_id, "voice.wav", source.read_bytes(), "audio/wav")
except Exception as exc:
result = dict(result or {})
result["artifact_error"] = "Voice audio could not be persisted"
logger.warning("background voice artifact persistence failed: %s", exc)
_voice_design_job_set(job_id, stage="done", detail="Done", done=True, result=result) _voice_design_job_set(job_id, stage="done", detail="Done", done=True, result=result)
except asyncio.TimeoutError: except asyncio.TimeoutError:
call.add_done_callback(lambda task: task.exception() if not task.cancelled() else None) call.add_done_callback(lambda task: task.exception() if not task.cancelled() else None)

File diff suppressed because one or more lines are too long

View File

@ -54,6 +54,19 @@
const meta = node(`${statusLabel(job)} · ${Number(job.elapsed || 0).toFixed(1)}s`); const meta = node(`${statusLabel(job)} · ${Number(job.elapsed || 0).toFixed(1)}s`);
meta.className = 'background-job-meta'; meta.className = 'background-job-meta';
main.appendChild(meta); main.appendChild(meta);
const storedResult = job.result && typeof job.result === 'object' ? job.result : {};
const artifactList = [];
if (storedResult.artifact) artifactList.push(storedResult.artifact);
if (Array.isArray(storedResult.artifacts)) artifactList.push(...storedResult.artifacts);
artifactList.filter(item => item && typeof item.url === 'string' && item.url.startsWith('/api/jobs/')).forEach(item => {
const link = document.createElement('a');
link.className = 'background-job-artifact';
link.href = item.url;
link.target = '_blank';
link.rel = 'noopener';
link.textContent = `Download ${item.kind || 'result'}`;
main.appendChild(link);
});
row.appendChild(main); row.appendChild(main);
const actions = document.createElement('div'); const actions = document.createElement('div');
actions.className = 'background-job-actions'; actions.className = 'background-job-actions';

View File

@ -3741,6 +3741,7 @@ code { background: var(--panel); border-radius: 4px; padding: 1px 5px; font-fami
.background-job-main strong { font-size: 13px; } .background-job-main strong { font-size: 13px; }
.background-job-main code { color: var(--subtext); font-size: 10px; } .background-job-main code { color: var(--subtext); font-size: 10px; }
.background-job-detail, .background-job-meta { grid-column: 1 / -1; color: var(--subtext); overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } .background-job-detail, .background-job-meta { grid-column: 1 / -1; color: var(--subtext); overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.background-job-artifact { grid-column: 1 / -1; color: var(--accent); font-size: 11px; }
.background-job-row.is-error .background-job-detail { color: var(--red); } .background-job-row.is-error .background-job-detail { color: var(--red); }
.background-job-row.is-done .background-job-meta { color: var(--green, #22c55e); } .background-job-row.is-done .background-job-meta { color: var(--green, #22c55e); }
.background-job-actions { display: flex; align-items: center; gap: 9px; flex-shrink: 0; } .background-job-actions { display: flex; align-items: center; gap: 9px; flex-shrink: 0; }

View File

@ -225,6 +225,31 @@ class SecurityValidationTests(unittest.TestCase):
conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,)) conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,))
conn.commit() conn.commit()
def test_background_job_artifact_is_atomic_bounded_and_job_scoped(self):
from core.database import background_job_artifact_path, background_job_create, background_job_store_artifact, background_job_update
job_id = "artifact-job-test"
background_job_create(job_id, "test")
try:
meta = background_job_store_artifact(job_id, "result.wav", b"RIFF-test", "audio/wav")
self.assertEqual(meta["name"], "result.wav")
path = background_job_artifact_path(job_id, "result.wav")
self.assertIsNotNone(path)
self.assertEqual(path.read_bytes(), b"RIFF-test")
background_job_update(job_id, stage="done", done=True, result={"artifact": meta})
from routes.jobs import jobs_artifact, jobs_get
self.assertEqual(asyncio.run(jobs_get(job_id))["result"]["artifact"]["url"], f"/api/jobs/{job_id}/artifacts/result.wav")
self.assertEqual(asyncio.run(jobs_artifact(job_id, "result.wav")).path, path)
self.assertIsNone(background_job_artifact_path(job_id, "../result.wav"))
with self.assertRaises(ValueError):
background_job_store_artifact(job_id, "bad.bin", b"x" * (50 * 1024 * 1024 + 1))
finally:
from core.database import _open, _BACKGROUND_ARTIFACT_DIR
with _open() as conn:
conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,))
conn.commit()
import shutil
shutil.rmtree(_BACKGROUND_ARTIFACT_DIR / job_id, ignore_errors=True)
def test_background_job_persistence_prunes_old_completed_rows(self): def test_background_job_persistence_prunes_old_completed_rows(self):
from core.database import _open, background_job_create, background_job_update from core.database import _open, background_job_create, background_job_update
prefix = "prune-job-" prefix = "prune-job-"