diff --git a/CHANGELOG.md b/CHANGELOG.md index 0efab46..1d67c1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. - **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. +- **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 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 8af9f8b..1bfe909 100644 --- a/TODO.md +++ b/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 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 plus voice design now have optional ID/status/cancel/timeout paths while preserving synchronous - callers. Casting still needs migration to the same user-facing durable job model, plus retries and persistence. + 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. - [ ] **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, diff --git a/VERSION b/VERSION index da9594f..013173a 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.22.5 +1.22.6 diff --git a/routes/conversation.py b/routes/conversation.py index b30440c..1375ed5 100644 --- a/routes/conversation.py +++ b/routes/conversation.py @@ -65,6 +65,10 @@ _consistent_image_idempotency: dict[str, str] = {} _consistent_image_slots = threading.BoundedSemaphore(2) _CAST_IDEMPOTENCY: dict[str, tuple[str, dict, float]] = {} _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_IDENTITY_FIELDS = ( "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: sheet = _character_form_sheet(sheet) 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() if len(idempotency_key) > 200: 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() if idempotency_key: now = time.time() @@ -742,6 +830,30 @@ async def match_characters_voices(request: Request): 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: """Pull the first {...} JSON object out of an LLM response.""" text = re.sub(r".*?", "", text, flags=re.DOTALL).strip() diff --git a/tests/test_security.py b/tests/test_security.py index dbdecb9..0b3f919 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -159,6 +159,24 @@ class SecurityValidationTests(unittest.TestCase): self.assertIn("same_asset_for_profile_and_reference", 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): from pathlib import Path from tempfile import TemporaryDirectory