diff --git a/app/_launch_runtime.py b/app/_launch_runtime.py index 150ba869..32ee35b1 100644 --- a/app/_launch_runtime.py +++ b/app/_launch_runtime.py @@ -31722,6 +31722,25 @@ def cancelled() -> bool: ) +def _reserve_story_music_submission(body: dict, workspace: str): + """Persist command/task/candidate IDs before the MiniMax worker starts.""" + from services.music_submission import ( + MusicSubmissionConflict, + MusicSubmissionError, + submit_music_generation, + ) + + try: + return submit_music_generation( + workspace_dir=_workspace_dir(workspace), + request={**body, "output_folder": workspace, "workspace": workspace}, + ) + except MusicSubmissionConflict as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + except MusicSubmissionError as exc: + raise HTTPException(status_code=exc.status_code, detail=str(exc)) from exc + + @api.post("/api/v1/stories/music-candidates/jobs", status_code=202) def start_story_music_candidates_job(body: dict): """Start observable MiniMax Music generation and return immediately.""" @@ -31763,8 +31782,17 @@ def start_story_music_candidates_job(body: dict): detail="Upload a valid reference song before generating a cover", ) - job_id = f"minimax-music-{uuid.uuid4().hex[:12]}" - task_id = f"task-minimax-music-{job_id}" + reserved = _reserve_story_music_submission(body, workspace) or {} + if reserved.get("replay"): + existing = _load_minimax_music_job(str(reserved.get("job_id") or "")) + if existing: + public = _public_minimax_music_job(existing) + public["replay"] = True + return public + # Reservation survived but the MiniMax job did not: start the worker + # again with the reserved IDs instead of returning a dead 202. + job_id = str(reserved.get("job_id") or f"minimax-music-{uuid.uuid4().hex[:12]}") + task_id = str(reserved.get("task_id") or f"task-minimax-music-{job_id}") now = time.time() children = [] for index in range(count): @@ -31824,8 +31852,19 @@ def start_story_music_candidates_job(body: dict): "model": model, "reference_audio_path": reference_audio_path, }, + "generationId": reserved.get("generation_id"), + "commandId": reserved.get("command_id"), + "candidateId": reserved.get("candidate_id"), + "idempotencyKey": reserved.get("idempotency_key"), } with _minimax_music_jobs_lock: + existing_live = _minimax_music_jobs.get(job_id) + if existing_live is not None: + # Same reserved job_id is already in flight (concurrent replay + # missed the checkpoint). Do not start a second worker. + public = _public_minimax_music_job(existing_live) + public["replay"] = True + return public _minimax_music_jobs[job_id] = job _persist_minimax_music_job(job) _publish_minimax_music_job(job) diff --git a/app/services/music_submission.py b/app/services/music_submission.py new file mode 100644 index 00000000..5af01126 --- /dev/null +++ b/app/services/music_submission.py @@ -0,0 +1,334 @@ +"""Idempotent music generation submission (reservation before inference). + +This module does not import FastAPI, WanGP or launch. It reserves command, +generation, task and candidate IDs, verifies Story destinations by ID, and +deduplicates by idempotency key. Starting a GPU/provider worker is the +caller's optional ``after_persist`` hook. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import threading +import uuid +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Mapping + +from .minimax_music_service import ALLOWED_MODELS, COVER_MODELS +from .story_library import read_story_library +from .task_manager import TaskRegistry + + +SCHEMA_NAME = "hocuspocus.music-submission" +SCHEMA_VERSION = 1 +STORE_DIRNAME = "music-submissions" +LOCAL_MODELS = frozenset({"ace_step_v1_5_xl_sft_lm_4b", "minimax_music3"}) +REMOTE_MODELS = frozenset(ALLOWED_MODELS) +INTENTS = frozenset({"retransmit", "retry", "new_version"}) +_STORE_LOCK = threading.RLock() + + +class MusicSubmissionError(ValueError): + def __init__(self, message: str, status_code: int = 400): + super().__init__(message) + self.status_code = status_code + + +class MusicSubmissionConflict(MusicSubmissionError): + def __init__(self, message: str = "Idempotency key reused with a different spec"): + super().__init__(message, status_code=409) + + +def _clean(value: Any) -> str | None: + text = str(value or "").strip() + return text or None + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _portable_folder(value: Any) -> str | None: + text = _clean(value) + if not text: + return None + name = os.path.basename(text.replace("\\", "/")) + if not name or name in {".", ".."} or os.path.isabs(name) or "/" in name or "\\" in name: + raise MusicSubmissionError("output_folder must be a relative folder name, never a path") + return name + + +def classify_music_route(model: str | None) -> str: + token = _clean(model) or "music-3.0" + if token in LOCAL_MODELS or token.startswith("ace_step"): + return "local" + if token in REMOTE_MODELS: + return "remote_minimax" + raise MusicSubmissionError(f"Unsupported music model: {token}") + + +def _stable_dump(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + + +def spec_snapshot(request: Mapping[str, Any]) -> dict[str, Any]: + provenance = request.get("provenance") if isinstance(request.get("provenance"), Mapping) else {} + model = _clean(request.get("model")) or "music-3.0" + try: + count = max(1, min(3, int(request.get("count") or 2))) + except (TypeError, ValueError, OverflowError) as exc: + raise MusicSubmissionError( + "MiniMax Music candidate count must be an integer from 1 to 3", + ) from exc + return { + "model": model, + "prompt": str(request.get("prompt") or "").strip()[:300], + "lyrics": str(request.get("lyrics") or "").strip()[:3500], + "instrumental": bool(request.get("instrumental")), + "count": count, + "project_id": _clean(request.get("project_id") or provenance.get("project_id")), + "cue_id": _clean(request.get("cue_id") or provenance.get("cue_id")), + "candidate_id": _clean(request.get("candidate_id") or provenance.get("candidate_id")), + "output_folder": _portable_folder( + request.get("output_folder") or request.get("workspace"), + ), + "workspace_id": _clean(request.get("workspace_id")), + "library_revision": request.get("library_revision", request.get("expectedVersion")), + "reference_audio_filename": _clean(request.get("reference_audio_filename")), + "intent": _intent(request), + } + + +def spec_hash(spec: Mapping[str, Any]) -> str: + return hashlib.sha256(_stable_dump(dict(spec)).encode("utf-8")).hexdigest() + + +def _token_id(prefix: str) -> str: + return f"{prefix}-{uuid.uuid4().hex[:12]}" + + +def _story_row_by_id(items: Any, token: str) -> dict[str, Any] | None: + if not isinstance(items, list): + return None + for item in items: + if isinstance(item, dict) and _clean(item.get("id")) == token: + return item + return None + + +def verify_story_destination( + workspace_dir: str, + spec: Mapping[str, Any], +) -> None: + """Require Story rows by ID. Never resolve a project or cue by title.""" + project_id = spec.get("project_id") + if not project_id: + return + library = read_story_library(workspace_dir) + expected = spec.get("library_revision") + if expected not in (None, ""): + try: + wanted = int(expected) + except (TypeError, ValueError) as exc: + raise MusicSubmissionError("library_revision must be an integer") from exc + if wanted != int(library.get("revision") or 0): + raise MusicSubmissionConflict( + f"Story library revision conflict: expected {wanted}, current {library.get('revision')}", + ) + project = library.get("projects", {}).get(project_id) + if not isinstance(project, dict): + raise MusicSubmissionError(f"Story project {project_id!r} was not found", 404) + cue_id = spec.get("cue_id") + if not cue_id: + return + cue = _story_row_by_id((project.get("music") or {}).get("cues"), cue_id) + if cue is None: + raise MusicSubmissionError(f"Story cue {cue_id!r} was not found", 404) + candidate_id = spec.get("candidate_id") + if candidate_id and _story_row_by_id(cue.get("candidates"), candidate_id) is None: + raise MusicSubmissionError(f"Story song candidate {candidate_id!r} was not found", 404) + + +def _validate_spec(spec: dict[str, Any]) -> str: + if not spec.get("output_folder"): + raise MusicSubmissionError("output_folder is required") + route = classify_music_route(spec.get("model")) + if not spec.get("prompt"): + raise MusicSubmissionError("A music style prompt is required") + model = spec["model"] + if model not in COVER_MODELS and not spec["instrumental"] and not spec["lyrics"]: + raise MusicSubmissionError("Lyrics are required for a vocal song") + if model in COVER_MODELS and not spec.get("reference_audio_filename"): + raise MusicSubmissionError("Upload a valid reference song before generating a cover") + return route + + +class MusicSubmissionStore: + """One JSON document per idempotency key with atomic replace.""" + + def __init__(self, root: str | os.PathLike[str]): + self.root = Path(root) / STORE_DIRNAME + self._lock = threading.RLock() + + def _path(self, key: str) -> Path: + digest = hashlib.sha256(key.encode("utf-8")).hexdigest()[:40] + return self.root / f"{digest}.json" + + def load(self, key: str) -> dict[str, Any] | None: + path = self._path(key) + if not path.is_file(): + return None + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + return dict(value) if isinstance(value, dict) else None + + def persist(self, record: Mapping[str, Any]) -> dict[str, Any]: + key = str(record["idempotency_key"]) + path = self._path(key) + payload = json.loads(json.dumps(record, ensure_ascii=False)) + with _STORE_LOCK: + existing = self.load(key) + if existing: + if existing.get("spec_hash") != payload.get("spec_hash"): + raise MusicSubmissionConflict() + return existing + self.root.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + try: + with open(temporary, "w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + except Exception: + try: + temporary.unlink(missing_ok=True) + except OSError: + pass + raise + return payload + + +def _intent(request: Mapping[str, Any]) -> str: + raw = (_clean(request.get("intent")) or "retransmit").casefold() + if request.get("retry") is True: + raw = "retry" + if request.get("new_version") is True: + raw = "new_version" + if raw not in INTENTS: + raise MusicSubmissionError(f"Unsupported music submission intent: {raw}") + return raw + + +def submit_music_generation( + *, + workspace_dir: str, + request: Mapping[str, Any], + task_registry: TaskRegistry | None = None, + after_persist: Callable[[dict[str, Any]], None] | None = None, +) -> dict[str, Any]: + """Reserve durable IDs and return 202-shaped state without running inference.""" + spec = spec_snapshot(request) + route = _validate_spec(spec) + intent = spec["intent"] + digest = spec_hash(spec) + key = _clean(request.get("idempotency_key") or request.get("idempotencyKey")) + if not key: + # No key: a new attempt. Replay only happens when the caller repeats a key. + key = _token_id("idem") + store = MusicSubmissionStore(workspace_dir) + existing = store.load(key) + if existing: + if existing.get("spec_hash") != digest: + raise MusicSubmissionConflict() + existing = dict(existing) + existing["replay"] = True + return existing + verify_story_destination(workspace_dir, spec) + job_id = _token_id("minimax-music") if route == "remote_minimax" else _token_id("local-music") + task_id = _clean(request.get("task_id")) or f"task-{job_id}" + generation_id = _clean(request.get("generation_id")) or _token_id("gen") + candidate_id = spec.get("candidate_id") or _token_id("song") + command_id = _clean(request.get("command_id") or request.get("commandId")) or _token_id("cmd") + parent_generation_id = _clean(request.get("parent_generation_id")) if intent == "retry" else None + record = { + "schema": SCHEMA_NAME, + "schema_version": SCHEMA_VERSION, + "idempotency_key": key, + "spec_hash": digest, + "spec": spec, + "intent": intent, + "route": route, + "command_id": command_id, + "generation_id": generation_id, + "task_id": task_id, + "job_id": job_id, + "candidate_id": candidate_id, + "parent_generation_id": parent_generation_id, + "status": "queued", + "replay": False, + "created_at": _now_iso(), + "start_error": None, + } + stored = store.persist(record) + if stored.get("job_id") != record["job_id"]: + stored = dict(stored) + stored["replay"] = True + return stored + registry = task_registry or TaskRegistry(workspace_dir, interrupt_stale=False) + registry.create( + id=task_id, + kind="music", + workflow="generate_story_song", + status="queued", + title="Story song", + workspace=spec["output_folder"], + project_id=spec.get("project_id") or "", + backend_job_id=job_id, + metadata={ + "generation_id": generation_id, + "candidate_id": candidate_id, + "command_id": command_id, + "idempotency_key": key, + }, + ) + if after_persist is not None: + try: + after_persist(stored) + except Exception as exc: + stored = dict(stored) + stored["start_error"] = str(exc)[:500] + stored["status"] = "queued" + stored["replay"] = False + return stored + + +def public_music_job(record: Mapping[str, Any]) -> dict[str, Any]: + spec = record.get("spec") if isinstance(record.get("spec"), Mapping) else {} + return { + "jobId": record.get("job_id"), + "taskId": record.get("task_id"), + "rootTaskId": record.get("task_id"), + "workspace": spec.get("output_folder"), + "status": record.get("status") or "queued", + "phase": record.get("status") or "queued", + "message": "Music generation accepted", + "current": 0, + "total": spec.get("count") or 1, + "progress": 0, + "provider": "local" if record.get("route") == "local" else "minimax", + "model": spec.get("model"), + "candidates": [], + "error": record.get("start_error"), + "generationId": record.get("generation_id"), + "commandId": record.get("command_id"), + "candidateId": record.get("candidate_id"), + "idempotencyKey": record.get("idempotency_key"), + "replay": bool(record.get("replay")), + } diff --git a/docs/development/MUSIC_SUBMISSION.md b/docs/development/MUSIC_SUBMISSION.md new file mode 100644 index 00000000..91a37759 --- /dev/null +++ b/docs/development/MUSIC_SUBMISSION.md @@ -0,0 +1,42 @@ +# Music submission contract + +Status: accepted (phase 4). Authority is the Story library + TaskRegistry + +this reservation JSON. Inference is not part of acceptance. + +## Paths (kept) + +| Path | Endpoint | Models | +|---|---|---| +| Local ACE-Step / MiniMax-Music3 | existing `generateMusic` / start_generation | `ace_step_*`, `minimax_music3` | +| Remote MiniMax (durable) | `POST /api/v1/stories/music-candidates/jobs` → **202** | `music-3.0`, `music-2.6`, covers | +| Legacy sync | `POST /api/v1/stories/music-candidates` | same remote models; waits for bytes | + +Clients of the durable job route keep working. Extra fields on the 202 body +(`generationId`, `commandId`, `candidateId`, `idempotencyKey`, `replay`) are +additive. + +## Request + +`command_id` / `idempotency_key`, project/cue/candidate IDs (never titles), +physical `output_folder` (or `workspace` as the folder name), optional +Workspace `workspace_id`, optional Story `library_revision`, and an immutable +spec snapshot (`model`, prompt, lyrics, instrumental, count, reference). + +`intent`: `retransmit` (default) | `retry` | `new_version`. Retry and new +version mint a new attempt with lineage and a distinct key. + +## Dedup + +Same idempotency key + same spec hash → same `job_id` / `task_id` / +`generation_id` / `candidate_id` (HTTP replay). Same key + different spec → +**409**. IDs are reserved **before** `after_persist` (worker start). A worker +start failure does not delete the reservation. + +Story rows are looked up by ID in `.story-library-v1.json`. A title is never +a key. + +## Query + +`GET /api/v1/stories/music-candidates/jobs/{job_id}` remains the poll URL. +TaskRegistry owns the task row. This module does not download models or talk +to a GPU. diff --git a/fase4.md b/fase4.md index 7b651159..551821b6 100644 --- a/fase4.md +++ b/fase4.md @@ -10,14 +10,14 @@ Plan de ejecución basado en la auditoría del 5 de septiembre de 2026. Verifica ## Tareas de implementación -- [ ] F4.1 — Inventariar caminos local/ACE/Music3 y remoto; conservar endpoints y clientes existentes mediante compatibilidad explícita. -- [ ] F4.2 — Definir solicitud con command/idempotency key, proyecto/cue, carpeta, colección opcional, revisión y snapshot inmutable del spec. -- [ ] F4.3 — Reservar y persistir IDs de intento, tarea y candidato antes de aceptar ejecución. Verificar existencia y pertenencia del destino en servidor. -- [ ] F4.4 — Persistir deduplicación atómica: misma clave y payload devuelve el mismo resultado; clave reutilizada con payload distinto devuelve conflicto. -- [ ] F4.5 — Distinguir retransmisión de transporte de nueva versión creativa y retry explícito. Estos últimos generan nuevos intentos con lineage. -- [ ] F4.6 — Responder 202 con referencias y estado consultable sin esperar a inferencia; mantener una vía de compatibilidad documentada. -- [ ] F4.7 — Usar TaskRegistry y scheduler existentes. No introducir infraestructura distribuida, GPU ni descarga de modelos para probarlo. -- [ ] F4.8 — Probar fallo entre reserva y arranque, doble petición concurrente, respuesta perdida y referencia a proyecto inexistente. Nunca resolver por título si ya existe ID. +- [x] F4.1 — Inventario en `docs/development/MUSIC_SUBMISSION.md`: local ACE/Music3 (`generateMusic`), remoto 202 jobs, sync legado. Endpoints intactos. +- [x] F4.2 — `spec_snapshot` + `idempotency_key` / `command_id` / `output_folder` / `workspace_id` / `library_revision`. +- [x] F4.3 — `submit_music_generation` reserva `generation_id`, `task_id`, `job_id`, `candidate_id` y crea la fila TaskRegistry. Destino Story por ID. +- [x] F4.4 — Misma clave+spec → replay; misma clave+spec distinto → `MusicSubmissionConflict` 409. +- [x] F4.5 — `intent=retry|new_version` mint nuevo intento; retry guarda `parent_generation_id`. +- [x] F4.6 — Jobs POST sigue en 202. Sync legado documentado. Poll GET existente. +- [x] F4.7 — `TaskRegistry.create` (idempotente). Tests sin GPU/modelos. +- [x] F4.8 — `tests/test_music_submission.py`: after_persist falla, 8 hilos, título ≠ id, revisión stale. ## Pruebas y criterio de aceptación @@ -31,29 +31,28 @@ Parar para merge antes de fase 5. No abrir simultáneamente otra modificación d ## Protocolo obligatorio para cada fase -- [ ] Leer fase1.md y esta fase; comprobar dependencias mezcladas en main remoto. Si el trabajo ya existe, verificarlo y registrar evidencia en lugar de duplicarlo. -- [ ] Inspeccionar cambios locales y logs relevantes al diagnosticar. Trabajar en rama/worktree aislado desde el main actualizado; preservar WIP, stashes y archivos del usuario. -- [ ] Revisar PRs abiertos y sus archivos: máximo un PR por hotspot (_launch_runtime.py, useStore.ts, agentActions.ts, StoryLabPanel o runtime Director/Wizard). No usar ramas apiladas en esta ola. -- [ ] Registrar base SHA, archivos propios/prohibidos y pruebas antes de editar. Aplicar AGENTS.md; no tocar launchers ni código vendor/WanGP salvo paquete posterior explícito. -- [ ] Marcar [x] sólo tras cumplir la tarea y añadir evidencia breve: archivo, comando/resultado o URL/SHA. Un plan o test escrito sin ejecutar no acredita validación. -- [ ] Ejecutar tests focalizados y validación segura pertinente, lint/tipos/build si cambia UI, arquitectura si corresponde y ratchet contra base exacta. No refrescar baseline para ocultar regresiones. -- [ ] Revisar diff y archivos a añadir explícitamente. Nunca incorporar pesos, outputs, secretos, caches, entornos ni comunicaciones. No usar git add indiscriminado. -- [ ] Crear commit y PR hacia main, o actualizar el PR existente correspondiente. Descripción: problema, comportamiento final, alcance, pruebas, riesgos y limitaciones. -- [ ] Esperar CI del último head; resolver fallos atribuibles al cambio. Leer comentarios de Cursor, contrastarlos y corregir con tests. Repetir checks tras fixes; revisión de un commit anterior no acredita el actual. -- [ ] Entregar URL, head/base SHA y estado separado de implementación, CI, Cursor, merge y smoke. No hacer merge ni activar auto-merge. -- [ ] Continuar otra fase sólo si sus dependencias están mezcladas y no comparte hotspot/contrato en cambio. Si no queda trabajo independiente elegible, parar y pedir que se mezclen los PRs concretos. +- [x] Fases 2 y 3 mezcladas (`58b7a08a`, `8b010a68`). Ningún PR abierto. +- [x] Worktree `/tmp/hocus-fase4` desde `origin/main`. Stashes intactos. +- [x] Un solo PR sobre `_launch_runtime.py`. Sin useStore/agentActions/StoryLabPanel. +- [x] Base `8b010a68`. Propios: music_submission.py, tests, MUSIC_SUBMISSION.md, cableado mínimo launch. Prohibido: launchers, pesos, outputs. +- [x] F4.1–F4.8 con evidencia en tests/docs. +- [x] validate_local OK (E2E 7/7) + ratchet vs `8b010a68`. +- [x] Add explícito. Sin outputs. +- [ ] PR `feat/music-submission-contract`. +- [ ] CI y Cursor del head. Sin merge. +- [x] No abrir fase 5 ni otro PR de launch hasta mezclar este. ## Registro de entrega -- Base SHA: -- Rama / PR: -- Commit implementado: -- Tests ejecutados y resultado: -- CI del head: -- Revisión Cursor (SHA, hallazgos pendientes): -- Merge en main (lo completa quien lo verifique): -- Generación real: NO EJECUTADA salvo evidencia manual explícita. -- Bloqueos / siguiente fase elegible: +- Base SHA: `origin/main` `8b010a68` +- Rama / PR: `feat/music-submission-contract` (abrir) +- Commit implementado: (este commit) +- Tests ejecutados y resultado: pytest music_submission + minimax_music_jobs 14 passed +- CI del head: pendiente +- Revisión Cursor: pendiente (`cursor review` al abrir) +- Merge en main: no +- Generación real: NO EJECUTADA +- Bloqueos / siguiente fase elegible: fase 5 y 8 tras merge de este PR. Fase 6 tras este PR (fase 2 ya mezclada). Los checkboxes describen trabajo; los estados de entrega son independientes. No marcar una fase globalmente terminada sólo por abrir su PR. Tests reales requieren autorización manual separada. No son un requisito para abrir el PR y nunca se ejecutan en CI. diff --git a/tests/test_minimax_music_jobs.py b/tests/test_minimax_music_jobs.py index 0b0fad89..4f747b21 100644 --- a/tests/test_minimax_music_jobs.py +++ b/tests/test_minimax_music_jobs.py @@ -82,6 +82,8 @@ def _base_namespace(tmp_path: Path) -> dict: "_persist_minimax_music_job": lambda _job: None, "_publish_minimax_music_job": lambda _job: None, "_workspace_dir": lambda _workspace=None: str(tmp_path), + "_reserve_story_music_submission": lambda _body, _workspace: {}, + "_load_minimax_music_job": lambda _job_id: None, } @@ -172,6 +174,118 @@ def test_start_returns_a_durable_job_before_provider_work(tmp_path): ] +def test_replay_without_checkpoint_starts_the_worker(tmp_path): + namespace = _base_namespace(tmp_path) + started = [] + + class CaptureThread: + def __init__(self, *, target, args, **_kwargs): + self.args = args + + def start(self): + started.append(self.args) + + namespace.update({ + "HTTPException": _HTTPException, + "_get_active_workspace": lambda: "default", + "_safe_join": lambda root, name: os.path.join(root, name), + "_persist_minimax_music_job": lambda _job: None, + "_publish_minimax_music_job": lambda _job: None, + "_run_minimax_music_job": lambda _job_id: None, + "_public_minimax_music_job": lambda job: { + key: value for key, value in job.items() + if key not in {"request", "_cancel_requested"} + }, + "_load_minimax_music_job": lambda _job_id: None, + "_reserve_story_music_submission": lambda _body, _workspace: { + "replay": True, + "job_id": "minimax-music-abc123def456", + "task_id": "task-minimax-music-abc123def456", + "generation_id": "gen-1", + }, + "threading": SimpleNamespace(Thread=CaptureThread), + }) + start = _load("start_story_music_candidates_job", namespace=namespace)[ + "start_story_music_candidates_job" + ] + result = start({ + "prompt": "cinematic dream pop", + "lyrics": "[Verse]\nAcross the night", + "workspace": "default", + }) + assert started == [("minimax-music-abc123def456",)] + assert result["jobId"] == "minimax-music-abc123def456" + assert result["status"] == "queued" + + +def test_concurrent_replay_without_checkpoint_starts_one_worker(tmp_path): + namespace = _base_namespace(tmp_path) + started = [] + started_lock = threading.Lock() + barrier = threading.Barrier(8) + + class CaptureThread: + def __init__(self, *, target, args, **_kwargs): + self.args = args + + def start(self): + with started_lock: + started.append(self.args) + + namespace.update({ + "HTTPException": _HTTPException, + "_get_active_workspace": lambda: "default", + "_safe_join": lambda root, name: os.path.join(root, name), + "_persist_minimax_music_job": lambda _job: None, + "_publish_minimax_music_job": lambda _job: None, + "_run_minimax_music_job": lambda _job_id: None, + "_public_minimax_music_job": lambda job: { + key: value for key, value in job.items() + if key not in {"request", "_cancel_requested"} + }, + "_load_minimax_music_job": lambda _job_id: None, + "_reserve_story_music_submission": lambda _body, _workspace: { + "replay": True, + "job_id": "minimax-music-abc123def456", + "task_id": "task-minimax-music-abc123def456", + "generation_id": "gen-1", + }, + "threading": SimpleNamespace(Thread=CaptureThread), + }) + start = _load("start_story_music_candidates_job", namespace=namespace)[ + "start_story_music_candidates_job" + ] + results: list[dict | None] = [None] * 8 + errors: list[Exception] = [] + + def run(index: int) -> None: + try: + barrier.wait(timeout=2) + results[index] = start({ + "prompt": "cinematic dream pop", + "lyrics": "[Verse]\nAcross the night", + "workspace": "default", + }) + except Exception as exc: + errors.append(exc) + + threads = [ + threading.Thread(target=run, args=(index,)) + for index in range(8) + ] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert errors == [] + assert started == [("minimax-music-abc123def456",)] + assert all(result is not None for result in results) + assert {result["jobId"] for result in results} == {"minimax-music-abc123def456"} + assert sum(1 for result in results if result.get("replay") is True) == 7 + assert sum(1 for result in results if result.get("replay") is not True) == 1 + + def test_start_rejects_a_non_numeric_candidate_count_as_bad_input(tmp_path): namespace = _base_namespace(tmp_path) namespace.update({ diff --git a/tests/test_music_submission.py b/tests/test_music_submission.py new file mode 100644 index 00000000..786d5592 --- /dev/null +++ b/tests/test_music_submission.py @@ -0,0 +1,211 @@ +"""Model-free coverage for idempotent music submission before inference.""" +from __future__ import annotations + +import json +import threading +from pathlib import Path + +import pytest + +from app.services.music_submission import ( + LOCAL_MODELS, + REMOTE_MODELS, + MusicSubmissionConflict, + MusicSubmissionError, + classify_music_route, + public_music_job, + spec_hash, + spec_snapshot, + submit_music_generation, + verify_story_destination, +) +from app.services.story_library import write_story_library +from app.services.task_manager import TaskRegistry + + +def _request(**overrides): + payload = { + "prompt": "cinematic dream pop", + "lyrics": "[Verse]\nLa noche canta", + "model": "music-3.0", + "count": 1, + "output_folder": "night-shift", + "idempotency_key": "cmd-same-once", + } + payload.update(overrides) + return payload + + +def _write_library(tmp_path: Path, **project_fields): + project = { + "id": "story-1", + "title": "Night Choir", + "language": "Español", + "music": { + "cues": [{ + "id": "cue-1", + "title": "Opening", + "candidates": [{"id": "song-1", "status": "pending"}], + }], + }, + } + project.update(project_fields) + write_story_library( + str(tmp_path), + {"version": 2, "revision": 0, "activeId": "story-1", "projects": {"story-1": project}}, + base_revision=0, + ) + + +def test_inventory_keeps_local_and_remote_routes(): + assert classify_music_route("ace_step_v1_5_xl_sft_lm_4b") == "local" + assert classify_music_route("minimax_music3") == "local" + assert classify_music_route("music-3.0") == "remote_minimax" + assert "music-2.6" in REMOTE_MODELS + assert "minimax_music3" in LOCAL_MODELS + with pytest.raises(MusicSubmissionError, match="Unsupported"): + classify_music_route("unknown-model") + + +def test_same_key_and_spec_replays_without_a_second_task(tmp_path: Path): + first = submit_music_generation(workspace_dir=str(tmp_path), request=_request()) + second = submit_music_generation(workspace_dir=str(tmp_path), request=_request()) + assert first["replay"] is False + assert second["replay"] is True + assert second["job_id"] == first["job_id"] + assert second["task_id"] == first["task_id"] + assert second["generation_id"] == first["generation_id"] + assert second["candidate_id"] == first["candidate_id"] + registry = TaskRegistry(str(tmp_path), interrupt_stale=False) + assert registry.get(first["task_id"])["id"] == first["task_id"] + + +def test_same_key_different_spec_conflicts(tmp_path: Path): + submit_music_generation(workspace_dir=str(tmp_path), request=_request()) + with pytest.raises(MusicSubmissionConflict): + submit_music_generation( + workspace_dir=str(tmp_path), + request=_request(lyrics="[Verse]\nOtra letra"), + ) + + +def test_missing_idempotency_key_does_not_dedupe_repeats(tmp_path: Path): + first = submit_music_generation(workspace_dir=str(tmp_path), request=_request(idempotency_key=None)) + second = submit_music_generation(workspace_dir=str(tmp_path), request=_request(idempotency_key=None)) + assert first["job_id"] != second["job_id"] + assert first["replay"] is False + assert second["replay"] is False + + +def test_explicit_retry_mints_a_new_attempt_with_lineage(tmp_path: Path): + first = submit_music_generation(workspace_dir=str(tmp_path), request=_request()) + with pytest.raises(MusicSubmissionConflict): + submit_music_generation( + workspace_dir=str(tmp_path), + request=_request(retry=True, parent_generation_id=first["generation_id"]), + ) + retry = submit_music_generation( + workspace_dir=str(tmp_path), + request=_request( + idempotency_key="cmd-same-once:retry", + retry=True, + parent_generation_id=first["generation_id"], + ), + ) + assert retry["job_id"] != first["job_id"] + assert retry["generation_id"] != first["generation_id"] + assert retry["intent"] == "retry" + replay = submit_music_generation( + workspace_dir=str(tmp_path), + request=_request( + idempotency_key="cmd-same-once:retry", + retry=True, + parent_generation_id=first["generation_id"], + ), + ) + assert replay["replay"] is True + assert replay["job_id"] == retry["job_id"] + assert retry["parent_generation_id"] == first["generation_id"] + + +def test_missing_project_id_is_not_resolved_by_title(tmp_path: Path): + _write_library(tmp_path) + with pytest.raises(MusicSubmissionError, match="was not found"): + submit_music_generation( + workspace_dir=str(tmp_path), + request=_request( + project_id="Night Choir", + cue_id="cue-1", + candidate_id="song-1", + ), + ) + accepted = submit_music_generation( + workspace_dir=str(tmp_path), + request=_request( + idempotency_key="cmd-existing-ids", + project_id="story-1", + cue_id="cue-1", + candidate_id="song-1", + ), + ) + assert accepted["spec"]["project_id"] == "story-1" + + +def test_stale_library_revision_conflicts(tmp_path: Path): + _write_library(tmp_path) + with pytest.raises(MusicSubmissionConflict, match="revision"): + submit_music_generation( + workspace_dir=str(tmp_path), + request=_request( + idempotency_key="cmd-rev", + project_id="story-1", + cue_id="cue-1", + candidate_id="song-1", + library_revision=99, + ), + ) + + +def test_failure_after_reserve_still_returns_queryable_ids(tmp_path: Path): + def boom(_record): + raise RuntimeError("worker did not start") + + record = submit_music_generation( + workspace_dir=str(tmp_path), + request=_request(), + after_persist=boom, + ) + assert record["job_id"] + assert record["start_error"] + replay = submit_music_generation(workspace_dir=str(tmp_path), request=_request()) + assert replay["replay"] is True + assert replay["job_id"] == record["job_id"] + public = public_music_job(record) + assert public["status"] == "queued" + assert public["jobId"] == record["job_id"] + + +def test_concurrent_duplicate_posts_share_one_job(tmp_path: Path): + results: list[dict] = [] + + def worker(): + results.append(submit_music_generation(workspace_dir=str(tmp_path), request=_request())) + + threads = [threading.Thread(target=worker) for _ in range(8)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + job_ids = {item["job_id"] for item in results} + assert len(job_ids) == 1 + assert sum(1 for item in results if not item["replay"]) == 1 + + +def test_spec_hash_is_stable_and_ignores_transport_noise(): + first = spec_snapshot(_request(idempotency_key="a")) + second = spec_snapshot(_request(idempotency_key="b")) + assert spec_hash(first) == spec_hash(second) + assert first["output_folder"] == "night-shift" + assert first["workspace_id"] is None + omitted = spec_snapshot({k: v for k, v in _request().items() if k != "count"}) + assert omitted["count"] == 2