feat: add background casting jobs
This commit is contained in:
parent
cc8fcdbd69
commit
4300d22cd9
@ -31,6 +31,8 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
|
|||||||
status/cancellation endpoints with timeout and failure states; existing synchronous callers remain compatible.
|
status/cancellation endpoints with timeout and failure states; existing synchronous callers remain compatible.
|
||||||
- **Optional Background Voice Design.** `/api/voice-design` now accepts `background: true` and exposes bounded job
|
- **Optional Background Voice Design.** `/api/voice-design` now accepts `background: true` and exposes bounded job
|
||||||
status/cancellation endpoints with queue state, timeout and failure handling; existing synchronous callers remain compatible.
|
status/cancellation endpoints with queue state, timeout and failure handling; existing synchronous callers remain compatible.
|
||||||
|
- **Optional Background Casting.** `/api/match-characters-voices` now accepts `background: true` and exposes bounded
|
||||||
|
job status/cancellation endpoints while reusing the existing casting validation, retries, and idempotency behavior.
|
||||||
- **STT Cancellation Safety.** Queued transcription cancels before consuming the single STT slot, and timed-out
|
- **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.
|
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
|
- **Credential Boundary Review.** Provider credentials are now documented as server-only; only the explicitly
|
||||||
|
|||||||
4
TODO.md
4
TODO.md
@ -63,8 +63,8 @@ the first block before investing in architecture or polish.
|
|||||||
a 4 GB VM/container, and measure browser, Docker and temporary-disk usage. Automated browser emulation now proves
|
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.
|
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,
|
- [ ] **L — Add background jobs and bounded queues.** Image generation already has the full interactive job path,
|
||||||
and transcription plus voice design now have optional ID/status/cancel/timeout paths while preserving synchronous
|
and transcription, voice design, and casting now have optional ID/status/cancel/timeout paths while preserving
|
||||||
callers. Casting still needs migration to the same user-facing durable job model, plus retries and persistence.
|
synchronous callers. Durable persistence and retry policy still need to be added.
|
||||||
- [ ] **L — Run the Book 2 generalization regression.** Confirm speaker-attribution and
|
- [ ] **L — Run the Book 2 generalization regression.** Confirm speaker-attribution and
|
||||||
confidence rules are not overfit to *Der Sturm*.
|
confidence rules are not overfit to *Der Sturm*.
|
||||||
- [x] **L — Add character forms inside one character record.** `sheet.forms[]` stores bounded alternate appearances,
|
- [x] **L — Add character forms inside one character record.** `sheet.forms[]` stores bounded alternate appearances,
|
||||||
|
|||||||
@ -65,6 +65,10 @@ _consistent_image_idempotency: dict[str, str] = {}
|
|||||||
_consistent_image_slots = threading.BoundedSemaphore(2)
|
_consistent_image_slots = threading.BoundedSemaphore(2)
|
||||||
_CAST_IDEMPOTENCY: dict[str, tuple[str, dict, float]] = {}
|
_CAST_IDEMPOTENCY: dict[str, tuple[str, dict, float]] = {}
|
||||||
_CAST_IDEMPOTENCY_LOCK = threading.Lock()
|
_CAST_IDEMPOTENCY_LOCK = threading.Lock()
|
||||||
|
_CAST_JOBS: dict[str, dict] = {}
|
||||||
|
_CAST_JOB_IDEMPOTENCY: dict[str, tuple[str, float]] = {}
|
||||||
|
_CAST_JOB_MAX = 128
|
||||||
|
_CAST_JOB_TIMEOUT = 15 * 60
|
||||||
_CHARACTER_VISUAL_PROMPT_VERSION = "v3"
|
_CHARACTER_VISUAL_PROMPT_VERSION = "v3"
|
||||||
_CHARACTER_IDENTITY_FIELDS = (
|
_CHARACTER_IDENTITY_FIELDS = (
|
||||||
"age", "age_estimate", "hair", "eyes", "physical", "build", "skin_tone", "race_species",
|
"age", "age_estimate", "hair", "eyes", "physical", "build", "skin_tone", "race_species",
|
||||||
@ -72,6 +76,60 @@ _CHARACTER_IDENTITY_FIELDS = (
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class _CastJobRequest:
|
||||||
|
"""Small internal request adapter used to reuse the validated cast route."""
|
||||||
|
|
||||||
|
def __init__(self, data: dict, headers: dict[str, str]):
|
||||||
|
self._data = data
|
||||||
|
self.headers = 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:
|
||||||
|
job.update(fields)
|
||||||
|
job["updated"] = time.time()
|
||||||
|
|
||||||
|
|
||||||
|
def _cast_job_trim() -> None:
|
||||||
|
if len(_CAST_JOBS) <= _CAST_JOB_MAX:
|
||||||
|
return
|
||||||
|
for job_id, job in sorted(_CAST_JOBS.items(), key=lambda item: item[1].get("updated", 0)):
|
||||||
|
if job.get("done"):
|
||||||
|
_CAST_JOBS.pop(job_id, None)
|
||||||
|
if len(_CAST_JOBS) <= _CAST_JOB_MAX:
|
||||||
|
break
|
||||||
|
|
||||||
|
|
||||||
|
def _cast_job_run_sync(data: dict, headers: dict[str, str]):
|
||||||
|
return asyncio.run(match_characters_voices(_CastJobRequest(data, headers)))
|
||||||
|
|
||||||
|
|
||||||
|
async def _run_cast_job(job_id: str, data: dict, headers: dict[str, str]) -> None:
|
||||||
|
if _CAST_JOBS.get(job_id, {}).get("cancel_requested"):
|
||||||
|
_cast_job_set(job_id, stage="cancelled", detail="Cancelled", done=True, cancelled=True,
|
||||||
|
error="Casting cancelled")
|
||||||
|
return
|
||||||
|
_cast_job_set(job_id, stage="running", detail="Matching characters to voices…")
|
||||||
|
call = asyncio.create_task(asyncio.to_thread(_cast_job_run_sync, data, headers))
|
||||||
|
try:
|
||||||
|
result = await asyncio.wait_for(asyncio.shield(call), timeout=_CAST_JOB_TIMEOUT)
|
||||||
|
if _CAST_JOBS.get(job_id, {}).get("cancel_requested"):
|
||||||
|
_cast_job_set(job_id, stage="cancelled", detail="Cancelled", done=True, cancelled=True,
|
||||||
|
error="Casting cancelled")
|
||||||
|
else:
|
||||||
|
_cast_job_set(job_id, stage="done", detail="Done", done=True, result=result)
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
call.add_done_callback(lambda task: task.exception() if not task.cancelled() else None)
|
||||||
|
_cast_job_set(job_id, stage="error", detail="Timed out", done=True, error="Voice casting timed out")
|
||||||
|
except Exception as exc:
|
||||||
|
message = str(exc.detail) if isinstance(exc, HTTPException) else str(exc)
|
||||||
|
_cast_job_set(job_id, stage="error", detail="Failed", done=True, error=message)
|
||||||
|
|
||||||
|
|
||||||
def _character_identity_fingerprint(sheet: dict) -> str:
|
def _character_identity_fingerprint(sheet: dict) -> str:
|
||||||
sheet = _character_form_sheet(sheet)
|
sheet = _character_form_sheet(sheet)
|
||||||
values = {key: sheet.get(key, "") for key in _CHARACTER_IDENTITY_FIELDS}
|
values = {key: sheet.get(key, "") for key in _CHARACTER_IDENTITY_FIELDS}
|
||||||
@ -618,6 +676,36 @@ async def match_characters_voices(request: Request):
|
|||||||
idempotency_key = str(request.headers.get("idempotency-key") or "").strip()
|
idempotency_key = str(request.headers.get("idempotency-key") or "").strip()
|
||||||
if len(idempotency_key) > 200:
|
if len(idempotency_key) > 200:
|
||||||
raise HTTPException(400, "Idempotency-Key is too long")
|
raise HTTPException(400, "Idempotency-Key is too long")
|
||||||
|
if data.get("background") is True:
|
||||||
|
job_fingerprint = hashlib.sha256(json.dumps({k: v for k, v in data.items() if k != "background"}, sort_keys=True, default=str).encode()).hexdigest()
|
||||||
|
if idempotency_key:
|
||||||
|
with _CAST_IDEMPOTENCY_LOCK:
|
||||||
|
previous_job_id, created = _CAST_JOB_IDEMPOTENCY.get(idempotency_key, ("", 0))
|
||||||
|
if previous_job_id and time.time() - created <= 1800:
|
||||||
|
previous_job = _CAST_JOBS.get(previous_job_id)
|
||||||
|
if previous_job is not None:
|
||||||
|
if previous_job.get("request_fingerprint") != job_fingerprint:
|
||||||
|
raise HTTPException(409, "Idempotency-Key was already used for a different request")
|
||||||
|
return {"job_id": previous_job_id, "deduplicated": True}
|
||||||
|
_CAST_JOB_IDEMPOTENCY.pop(idempotency_key, None)
|
||||||
|
job_id = uuid.uuid4().hex
|
||||||
|
job_data = dict(data)
|
||||||
|
job_data.pop("background", None)
|
||||||
|
now = time.time()
|
||||||
|
_CAST_JOBS[job_id] = {
|
||||||
|
"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,
|
||||||
|
}
|
||||||
|
if idempotency_key:
|
||||||
|
with _CAST_IDEMPOTENCY_LOCK:
|
||||||
|
_CAST_JOB_IDEMPOTENCY[idempotency_key] = (job_id, now)
|
||||||
|
if len(_CAST_JOB_IDEMPOTENCY) > 256:
|
||||||
|
oldest = min(_CAST_JOB_IDEMPOTENCY, key=lambda key: _CAST_JOB_IDEMPOTENCY[key][1])
|
||||||
|
_CAST_JOB_IDEMPOTENCY.pop(oldest, None)
|
||||||
|
_cast_job_trim()
|
||||||
|
asyncio.create_task(_run_cast_job(job_id, job_data, {"idempotency-key": idempotency_key}))
|
||||||
|
return {"job_id": job_id, "stage": "queued"}
|
||||||
request_fingerprint = hashlib.sha256(json.dumps(data, sort_keys=True, default=str).encode()).hexdigest()
|
request_fingerprint = hashlib.sha256(json.dumps(data, sort_keys=True, default=str).encode()).hexdigest()
|
||||||
if idempotency_key:
|
if idempotency_key:
|
||||||
now = time.time()
|
now = time.time()
|
||||||
@ -742,6 +830,30 @@ async def match_characters_voices(request: Request):
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
@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:
|
||||||
|
raise HTTPException(404, "Unknown or expired casting job")
|
||||||
|
payload = dict(job)
|
||||||
|
payload["elapsed"] = round(time.time() - payload.pop("started"), 1)
|
||||||
|
payload.pop("cancel_requested", None)
|
||||||
|
payload.pop("request_fingerprint", None)
|
||||||
|
return payload
|
||||||
|
|
||||||
|
|
||||||
|
@router.post("/api/match-characters-voices/cancel/{job_id}")
|
||||||
|
async def match_characters_voices_cancel(job_id: str):
|
||||||
|
job = _CAST_JOBS.get(job_id)
|
||||||
|
if job is None:
|
||||||
|
raise HTTPException(404, "Unknown or expired casting job")
|
||||||
|
if job.get("done"):
|
||||||
|
return {"ok": True, "already_done": True, "cancelled": bool(job.get("cancelled"))}
|
||||||
|
job["cancel_requested"] = True
|
||||||
|
_cast_job_set(job_id, stage="cancelling", detail="Cancelling…")
|
||||||
|
return {"ok": True, "cancelled": True}
|
||||||
|
|
||||||
|
|
||||||
def _extract_json_block(text: str) -> str:
|
def _extract_json_block(text: str) -> str:
|
||||||
"""Pull the first {...} JSON object out of an LLM response."""
|
"""Pull the first {...} JSON object out of an LLM response."""
|
||||||
text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
|
text = re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
|
||||||
|
|||||||
@ -159,6 +159,24 @@ class SecurityValidationTests(unittest.TestCase):
|
|||||||
self.assertIn("same_asset_for_profile_and_reference", result["issues"][0]["issues"])
|
self.assertIn("same_asset_for_profile_and_reference", result["issues"][0]["issues"])
|
||||||
self.assertIn("asset_reused_across_characters", result["issues"][0]["issues"])
|
self.assertIn("asset_reused_across_characters", result["issues"][0]["issues"])
|
||||||
|
|
||||||
|
def test_cast_background_job_honours_queued_cancellation(self):
|
||||||
|
import routes.conversation as conversation
|
||||||
|
from unittest.mock import patch
|
||||||
|
job_id = "cast-test-job"
|
||||||
|
now = 0.0
|
||||||
|
conversation._CAST_JOBS[job_id] = {
|
||||||
|
"job_id": job_id, "stage": "queued", "detail": "Queued", "started": now, "updated": now,
|
||||||
|
"done": False, "cancel_requested": True, "cancelled": False,
|
||||||
|
}
|
||||||
|
try:
|
||||||
|
with patch.object(conversation, "match_characters_voices") as match:
|
||||||
|
asyncio.run(conversation._run_cast_job(job_id, {}, {}))
|
||||||
|
self.assertTrue(conversation._CAST_JOBS[job_id]["done"])
|
||||||
|
self.assertEqual(conversation._CAST_JOBS[job_id]["stage"], "cancelled")
|
||||||
|
match.assert_not_called()
|
||||||
|
finally:
|
||||||
|
conversation._CAST_JOBS.pop(job_id, None)
|
||||||
|
|
||||||
def test_fish_cache_eviction_is_bounded(self):
|
def test_fish_cache_eviction_is_bounded(self):
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from tempfile import TemporaryDirectory
|
from tempfile import TemporaryDirectory
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user