diff --git a/.gitignore b/.gitignore index 766c04bd..9675c485 100644 --- a/.gitignore +++ b/.gitignore @@ -21,4 +21,9 @@ GEMINI.md ds-bundle/ .design-sync/.cache/ .design-sync/learnings/ -.design-sync/node_modules \ No newline at end of file +.design-sync/node_modules + +# Local dev artifacts (never commit) +.DS_Store +venv/ +extension/_metadata/ \ No newline at end of file diff --git a/README.md b/README.md index e4dd67ea..ae09b502 100644 --- a/README.md +++ b/README.md @@ -711,19 +711,35 @@ Optional narrator voice for scenes. Uses [OmniVoice](https://github.com/tuannguy ### Setup -See `skills/fk-gen-tts-template.md` for full install guide. Quick version: +See `skills/fk-gen-tts-template.md` for the full install guide. The launcher uses the FlowKit venv +at `$HOME/.venvs/flowkit`; install OmniVoice there: ```bash -pip install torch==2.8.0 torchaudio==2.8.0 # or +cu128 for NVIDIA -pip install omnivoice -python3 -c "from omnivoice import OmniVoice; print('OK')" +$HOME/.venvs/flowkit/bin/python -m pip install torch==2.8.0 torchaudio==2.8.0 +$HOME/.venvs/flowkit/bin/python -m pip install omnivoice +$HOME/.venvs/flowkit/bin/python -c 'from omnivoice import OmniVoice; print("OmniVoice OK")' ``` -If OmniVoice is in a separate venv, point to it: +The local subprocess defaults to the interpreter that launched FlowKit, so `TTS_PYTHON_BIN` is +optional. Set it only when OmniVoice intentionally lives in a separate venv. + +### Optional remote OmniVoice API + +`agent/omnivoice_api.py` is a standalone warm FastAPI service. It loads the model once and exposes +`POST /v1/tts` as multipart form data (`text`, `speed`, optional `instruct`, `ref_text`, and WAV +file `ref_audio`), returning raw `audio/wav` bytes. Run it from the FlowKit checkout: + ```bash -export TTS_PYTHON_BIN=/path/to/omnivoice-venv/bin/python3 +$HOME/.venvs/flowkit/bin/python -m pip install python-multipart +./flowkit/run-omnivoice-api.sh ``` +The main FlowKit API can call it by setting `TTS_BACKEND=remote` and `TTS_REMOTE_URL=http://127.0.0.1:8200`. +The remote backend reads each local `ref_audio` WAV, uploads its bytes, validates the returned WAV, +and writes it atomically to the existing local output path. For a non-loopback deployment, set +`OMNIVOICE_API_TOKEN` on the inference service and the matching `TTS_REMOTE_TOKEN` on FlowKit, +use HTTPS, and do not expose the service publicly without authentication. + ### Workflow 1. **Create voice template** — `/fk-gen-tts-template` — generates an anchor voice WAV @@ -731,7 +747,8 @@ export TTS_PYTHON_BIN=/path/to/omnivoice-venv/bin/python3 3. **Generate narration** — `/fk-gen-narrator` — voice-clones the template for each scene 4. **Concat with narration** — `/fk-concat-fit-narrator` — trims scene videos to match TTS duration -CPU-only recommended (MPS produces artifacts). ~15-30s per scene. +CPU-only is recommended for the local backend (MPS produces artifacts). Remote mode is serialized +per scene and keeps the model warm in the standalone inference service. ## YouTube Upload Pipeline diff --git a/agent/api/projects.py b/agent/api/projects.py index 600ebe8a..1d51fc59 100644 --- a/agent/api/projects.py +++ b/agent/api/projects.py @@ -7,7 +7,7 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel -from agent.config import BASE_DIR, USE_BATCH_RPC +from agent.config import BASE_DIR, USE_BATCH_RPC, FLOW_PROJECT_ID from agent.models.project import Project, ProjectCreate, ProjectUpdate from agent.models.character import Character from agent.sdk.persistence.sqlite_repository import SQLiteRepository @@ -172,8 +172,9 @@ async def create(body: ProjectCreate): detected_tier = await _detect_user_tier(client) - # On the batch path Flow no longer creates projects for us — the uuid comes - # from the request or from FLOW_PROJECT_ID. The legacy path still mints one. + # On the batch path a pinned FLOW_PROJECT_ID (or a uuid passed in the request) + # is reused; with neither, create_project now mints one via jHPbke. The legacy + # path always mints one. flow_project_id = client.flow_project_id(body.flow_project_id) if USE_BATCH_RPC else None if flow_project_id: logger.info("Flow project reused: %s", flow_project_id) @@ -186,6 +187,15 @@ async def create(body: ProjectCreate): repo = _get_repo() + # Reuse is idempotent: a pinned or request-supplied project id may already + # have a local row. Re-inserting that id violates the primary key + # (sqlite3.IntegrityError -> 500), so hand back the row we already have. A + # freshly minted id never matches, so this is a no-op on the create path. + existing = await repo.get_project(flow_project_id) + if existing: + logger.info("Flow project already tracked, reusing: %s", flow_project_id) + return existing + # Step 2: Create local project with the Flow-assigned ID and detected tier create_data = body.model_dump(exclude_none=True) create_data.pop("tool_name", None) @@ -259,9 +269,25 @@ async def update(pid: str, body: ProjectUpdate): @router.delete("/{pid}") async def delete(pid: str): + """Delete a project locally, and on Flow too when it is safe. + + The project must exist locally before anything is deleted, so a stray uuid + cannot reach through to Flow — only projects Flow Kit already tracks are + eligible. Flow is then deleted first (except the pinned FLOW_PROJECT_ID, the + shared/operator project, which is only ever removed locally) so a remote + failure leaves the local row in place rather than letting the two drift. + """ repo = _get_repo() - if not await repo.delete_project(pid): + if not await repo.get_project(pid): raise HTTPException(404, "Project not found") + if USE_BATCH_RPC and pid != (FLOW_PROJECT_ID or None): + client = get_flow_client() + if not client.connected: + raise HTTPException(503, "Extension not connected — cannot delete this project on Google Flow") + result = await client.delete_project(pid) + if result.get("error"): + raise HTTPException(502, f"Flow delete failed: {result['error']}") + await repo.delete_project(pid) return {"ok": True} diff --git a/agent/api/tts.py b/agent/api/tts.py index e2d3d068..342acfea 100644 --- a/agent/api/tts.py +++ b/agent/api/tts.py @@ -146,6 +146,10 @@ async def narrate_video(vid: str, body: NarrateVideoRequest): logger.info("Auto-resolved ref_text from template '%s'", tmpl["name"]) break + # The project/template fallback is operator data too; revalidate after resolution so remote TTS + # can never read and upload a path that bypassed the request-body check above. + if ref_audio: + _validate_ref_audio(ref_audio) project_name = project.get("name") or "unnamed_project" project_slug = slugify(project_name) out_dir = OUTPUT_DIR / project_slug / "tts" diff --git a/agent/config.py b/agent/config.py index 5d6edf2f..790ff380 100644 --- a/agent/config.py +++ b/agent/config.py @@ -86,9 +86,25 @@ MUSIC_OUTPUT_DIR = SHARED_OUTPUT_DIR / "music" # ─── TTS (OmniVoice) ───────────────────────────────────────── +TTS_BACKEND = os.environ.get("TTS_BACKEND", "local").strip().lower() TTS_MODEL = os.environ.get("TTS_MODEL", "k2-fsa/OmniVoice") TTS_DEVICE = os.environ.get("TTS_DEVICE", "cpu") # MPS produces gibberish; CPU+fp32 works TTS_SAMPLE_RATE = int(os.environ.get("TTS_SAMPLE_RATE", "24000")) +TTS_REMOTE_URL = os.environ.get("TTS_REMOTE_URL", "").strip().rstrip("/") +TTS_REMOTE_TOKEN = os.environ.get("TTS_REMOTE_TOKEN", "") +TTS_REMOTE_TIMEOUT_S = float(os.environ.get("TTS_REMOTE_TIMEOUT_S", "180")) +TTS_REMOTE_MAX_BYTES = int(os.environ.get("TTS_REMOTE_MAX_BYTES", str(50 * 1024 * 1024))) + +# ─── Standalone OmniVoice API ──────────────────────────────── +OMNIVOICE_API_HOST = os.environ.get("OMNIVOICE_API_HOST", "127.0.0.1") +OMNIVOICE_API_PORT = int(os.environ.get("OMNIVOICE_API_PORT", "8200")) +OMNIVOICE_API_TOKEN = os.environ.get("OMNIVOICE_API_TOKEN", "") +OMNIVOICE_MAX_REF_AUDIO_BYTES = int( + os.environ.get("OMNIVOICE_MAX_REF_AUDIO_BYTES", str(10 * 1024 * 1024)) +) +OMNIVOICE_MAX_REQUEST_BYTES = int( + os.environ.get("OMNIVOICE_MAX_REQUEST_BYTES", str(12 * 1024 * 1024)) +) # ─── Review / Claude Vision ────────────────────────────────── ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "") diff --git a/agent/omnivoice_api.py b/agent/omnivoice_api.py new file mode 100644 index 00000000..50c1d9d9 --- /dev/null +++ b/agent/omnivoice_api.py @@ -0,0 +1,224 @@ +"""Standalone warm OmniVoice HTTP service for remote FlowKit TTS.""" +from __future__ import annotations + +import asyncio +import hmac +import io +import math +import tempfile +import wave +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Optional + +from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile +from fastapi.responses import JSONResponse, Response + +from agent.config import ( + OMNIVOICE_API_HOST, + OMNIVOICE_API_PORT, + OMNIVOICE_API_TOKEN, + OMNIVOICE_MAX_REF_AUDIO_BYTES, + OMNIVOICE_MAX_REQUEST_BYTES, + TTS_DEVICE, + TTS_MODEL, + TTS_SAMPLE_RATE, +) + +_model = None +_inference_lock = asyncio.Lock() + + +def _validate_wav_bytes(audio_bytes: bytes) -> None: + if len(audio_bytes) < 12 or audio_bytes[:4] != b"RIFF" or audio_bytes[8:12] != b"WAVE": + raise HTTPException(status_code=400, detail="ref_audio must be a WAV file") + try: + with wave.open(io.BytesIO(audio_bytes), "rb") as wav: + frame_count = wav.getnframes() + frame_width = wav.getnchannels() * wav.getsampwidth() + if frame_count <= 0 or wav.getframerate() <= 0 or frame_width <= 0: + raise HTTPException(status_code=400, detail="ref_audio must contain audio") + if len(wav.readframes(frame_count)) != frame_count * frame_width: + raise HTTPException(status_code=400, detail="ref_audio is truncated") + except (wave.Error, EOFError) as error: + raise HTTPException(status_code=400, detail="ref_audio is an invalid WAV file") from error + + +def _load_model(): + global _model + import torch + from omnivoice import OmniVoice + + dtype = torch.float16 if TTS_DEVICE.startswith("cuda") else torch.float32 + _model = OmniVoice.from_pretrained(TTS_MODEL, device_map=TTS_DEVICE, dtype=dtype) + + +def _generate_wav_bytes( + text: str, + instruct: Optional[str], + ref_text: Optional[str], + ref_audio_bytes: Optional[bytes], + speed: float, +) -> bytes: + import torchaudio + + if _model is None: + raise RuntimeError("OmniVoice model is not loaded") + + with tempfile.TemporaryDirectory(prefix="omnivoice-") as temp_dir: + ref_audio_path = None + if ref_audio_bytes is not None: + ref_audio_path = Path(temp_dir) / "reference.wav" + ref_audio_path.write_bytes(ref_audio_bytes) + + output_path = Path(temp_dir) / "output.wav" + kwargs = {"text": text} + if ref_audio_path is not None and ref_text: + kwargs["ref_audio"] = str(ref_audio_path) + kwargs["ref_text"] = ref_text + elif instruct: + kwargs["instruct"] = instruct + if speed != 1.0: + kwargs["speed"] = speed + + audio = _model.generate(**kwargs) + torchaudio.save(str(output_path), audio[0], TTS_SAMPLE_RATE) + return output_path.read_bytes() + + +def _authorization_header_valid(actual: str) -> bool: + if not OMNIVOICE_API_TOKEN: + return True + return hmac.compare_digest(actual, f"Bearer {OMNIVOICE_API_TOKEN}") + + +def _authorization_valid(request: Request) -> bool: + return _authorization_header_valid(request.headers.get("authorization", "")) + + +async def _require_authorized(request: Request) -> None: + if not _authorization_valid(request): + raise HTTPException(status_code=401, detail="Invalid bearer token") + + +@asynccontextmanager +async def lifespan(app: FastAPI): + if OMNIVOICE_API_HOST not in {"127.0.0.1", "localhost", "::1"} and not OMNIVOICE_API_TOKEN: + raise RuntimeError("OMNIVOICE_API_TOKEN is required when the API is not loopback-only") + await asyncio.to_thread(_load_model) + yield + + +app = FastAPI(title="OmniVoice TTS", version="1.0.0", lifespan=lifespan) + + +class _RequestTooLarge(Exception): + pass + + +async def _send_json_error(send, status_code: int, detail: str) -> None: + body = f'{{"detail":"{detail}"}}'.encode("utf-8") + await send({"type": "http.response.start", "status": status_code, "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode("ascii")), + ]}) + await send({"type": "http.response.body", "body": body}) + + +class _TtsRequestLimitMiddleware: + def __init__(self, app): + self.app = app + + async def __call__(self, scope, receive, send): + if scope.get("type") != "http" or scope.get("path") != "/v1/tts": + await self.app(scope, receive, send) + return + + request_headers = {key.lower(): value for key, value in scope.get("headers", [])} + authorization = request_headers.get(b"authorization", b"").decode("latin-1") + if not _authorization_header_valid(authorization): + await _send_json_error(send, 401, "Invalid bearer token") + return + + content_length = request_headers.get(b"content-length") + if content_length is not None: + try: + if int(content_length) > OMNIVOICE_MAX_REQUEST_BYTES: + await _send_json_error(send, 413, "request is too large") + return + except ValueError: + await _send_json_error(send, 400, "invalid content-length") + return + + received = 0 + + async def limited_receive(): + nonlocal received + message = await receive() + if message["type"] == "http.request": + received += len(message.get("body", b"")) + if received > OMNIVOICE_MAX_REQUEST_BYTES: + raise _RequestTooLarge + return message + + try: + await self.app(scope, limited_receive, send) + except _RequestTooLarge: + await _send_json_error(send, 413, "request is too large") + + +app.add_middleware(_TtsRequestLimitMiddleware) + +@app.get("/health") +async def health(): + return {"ok": _model is not None, "model": TTS_MODEL, "device": TTS_DEVICE} + + +@app.post("/v1/tts", response_class=Response) +async def synthesize( + request: Request, + text: str = Form(...), + speed: float = Form(1.0), + instruct: Optional[str] = Form(None), + ref_text: Optional[str] = Form(None), + ref_audio: Optional[UploadFile] = File(None), +): + await _require_authorized(request) + if not text or len(text) > 5000: + raise HTTPException(status_code=422, detail="text must contain 1-5000 characters") + if not math.isfinite(speed) or speed < 0.5 or speed > 3.0: + raise HTTPException(status_code=422, detail="speed must be between 0.5 and 3.0") + if instruct is not None and len(instruct) > 200: + raise HTTPException(status_code=422, detail="instruct must contain at most 200 characters") + if ref_text is not None and len(ref_text) > 5000: + raise HTTPException(status_code=422, detail="ref_text is too long") + + ref_audio_bytes = None + if ref_audio is not None: + ref_audio_bytes = await ref_audio.read(OMNIVOICE_MAX_REF_AUDIO_BYTES + 1) + if len(ref_audio_bytes) > OMNIVOICE_MAX_REF_AUDIO_BYTES: + raise HTTPException(status_code=413, detail="ref_audio is too large") + _validate_wav_bytes(ref_audio_bytes) + + async with _inference_lock: + try: + wav_bytes = await asyncio.to_thread( + _generate_wav_bytes, + text, + instruct, + ref_text, + ref_audio_bytes, + speed, + ) + except HTTPException: + raise + except Exception: + raise HTTPException(status_code=500, detail="OmniVoice synthesis failed") + + return Response(content=wav_bytes, media_type="audio/wav") + + +if __name__ == "__main__": + import uvicorn + + uvicorn.run(app, host=OMNIVOICE_API_HOST, port=OMNIVOICE_API_PORT) diff --git a/agent/services/flow_batch.py b/agent/services/flow_batch.py index f0111632..b50bf510 100644 --- a/agent/services/flow_batch.py +++ b/agent/services/flow_batch.py @@ -40,6 +40,8 @@ RPC_PROJECT_MEDIA = "Zzl0ze" RPC_MEDIA = "as29s" RPC_UPLOAD_IMAGE = "maseQ" +RPC_CREATE_PROJECT = "jHPbke" +RPC_DELETE_PROJECT = "QI2zvc" CAPTCHA_IMAGE = "IMAGE_GENERATION" CAPTCHA_VIDEO = "VIDEO_GENERATION" @@ -382,6 +384,23 @@ def media_request(media_id: str) -> str: return build_envelope(RPC_MEDIA, [media_id]) +def create_project_request(title: str) -> str: + """Make a new Flow project titled ``title``. + + Captured off the UI's "New project" button. Inner payload is + ``["projects/*", [None, [title]], [None, SURFACE_ID]]`` — the trailing ``22`` + is the same surface id every other call stamps, not a tool type. This path + carries no tool slot, so it can only ask for the surface's default; a caller + that needs a specific tool cannot get it here (see FlowClient.create_project). + """ + return build_envelope(RPC_CREATE_PROJECT, ["projects/*", [None, [title]], [None, SURFACE_ID]]) + + +def delete_project_request(project_id: str) -> str: + """Delete a Flow project. Inner payload is ``["projects/"]``.""" + return build_envelope(RPC_DELETE_PROJECT, [f"projects/{project_id}"]) + + # ── response readers ───────────────────────────────────────────────────────── def _walk_strings(node: Any): @@ -429,6 +448,25 @@ def read_uploaded_media_id(payload: Any) -> str: return media_id +#: A Flow project id is a plain uuid; the create response leads with it. +_PROJECT_ID_RE = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.I) + + +def read_created_project_id(payload: Any) -> str: + """The uuid of a freshly created project. + + Captured shape: ``["", [""]]`` — the id leads the payload and + equals the ``/project/<uuid>`` the UI navigates to. Read that one slot and + nothing else: a 200 with the id somewhere unexpected is accepted and then + silently useless, so a non-uuid here is a failure rather than a cue to scan + the envelope for the first uuid-looking string. + """ + first = payload[0] if isinstance(payload, list) and payload else None + if isinstance(first, str) and _PROJECT_ID_RE.fullmatch(first): + return first + raise FlowBatchError(f"create response carried no project id: {payload!r}") + + def read_operation(payload: Any) -> Operation: """`[null, 50, [[opId, projectId, sceneId, status, …]]]`. diff --git a/agent/services/flow_client.py b/agent/services/flow_client.py index 667e2d51..7d7c092d 100644 --- a/agent/services/flow_client.py +++ b/agent/services/flow_client.py @@ -560,14 +560,50 @@ def flow_project_id(self, requested: str | None = None) -> str | None: return FLOW_PROJECT_ID or None async def create_project(self, project_title: str, tool_name: str = "PINHOLE") -> dict: + """Create a Flow project, or reuse the pinned one. + + Pin precedence: a pinned FLOW_PROJECT_ID (or a uuid the caller passes) is + reused rather than minting a new project every call — the route only + falls through to here when nothing is pinned. Unpinned, the batch path + now creates one for real via the ``jHPbke`` RPC captured off the UI's + "New project" button and hands back its uuid. ``tool_name`` is honoured + only on the legacy path; the batch create carries no tool slot, so a + non-default value is rejected rather than silently ignored. + """ if not USE_BATCH_RPC: return await self._legacy_create_project(project_title, tool_name) - pid = self.flow_project_id() - if not pid: - return {"error": _UNSUPPORTED_CREATE_PROJECT} - logger.info("Reusing pinned Flow project %s for '%s'", pid[:12], project_title) + pinned = self.flow_project_id() + if pinned: + logger.info("Reusing pinned Flow project %s for '%s'", pinned[:12], project_title) + return {"status": 200, "data": {"projectId": pinned}} + if tool_name and tool_name != "PINHOLE": + return {"error": "UNSUPPORTED_ON_BATCH_API: batch create carries no tool slot, " + f"cannot honour tool_name={tool_name!r}"} + try: + payload = await self._batch_payload(fb.RPC_CREATE_PROJECT, + fb.create_project_request(project_title)) + pid = fb.read_created_project_id(payload) + except (fb.FlowBatchError, fb.RpcError) as exc: + return {"error": f"CREATE_PROJECT_FAILED: {exc}"} + logger.info("Flow project created %s for '%s'", pid[:12], project_title) return {"status": 200, "data": {"projectId": pid}} + async def delete_project(self, project_id: str) -> dict: + """Delete a Flow project via the ``QI2zvc`` RPC (batch path only). + + A successful delete answers with an empty payload — there is nothing to + read back, so a clean return is success and a transport error surfaces as + ``{"error": …}`` like the rest of the client. + """ + if not USE_BATCH_RPC: + return {"error": "UNSUPPORTED_ON_BATCH_API: delete_project needs the batchexecute transport"} + try: + await self._batch_payload(fb.RPC_DELETE_PROJECT, fb.delete_project_request(project_id)) + except (fb.FlowBatchError, fb.RpcError) as exc: + return {"error": f"DELETE_PROJECT_FAILED: {exc}"} + logger.info("Flow project deleted %s", project_id[:12]) + return {"status": 200, "data": {"projectId": project_id}} + async def generate_images(self, prompt: str, project_id: str, aspect_ratio: str = "IMAGE_ASPECT_RATIO_PORTRAIT", user_paygate_tier: str = "PAYGATE_TIER_TWO", @@ -1207,12 +1243,6 @@ async def _legacy_upload_image(self, image_base64: str, mime_type: str = "image/ _CAPTURE_HINT = "see docs/CAPTURE.md to record its payload off the new UI" -_UNSUPPORTED_CREATE_PROJECT = ( - "NO_FLOW_PROJECT: Flow's project.createProject endpoint went with the September 2026 " - "migration, so Flow Kit cannot create one. Make a project in the Flow UI, then either " - "pass its uuid as flow_project_id or pin it as FLOW_PROJECT_ID." -) - def _unsupported(feature: str, why: str) -> str: return f"UNSUPPORTED_ON_BATCH_API: {feature} — {why}; {_CAPTURE_HINT}." diff --git a/agent/services/tts.py b/agent/services/tts.py index 14a72985..4283ebbf 100644 --- a/agent/services/tts.py +++ b/agent/services/tts.py @@ -1,20 +1,37 @@ -"""OmniVoice TTS service — subprocess-based for compatibility.""" +"""OmniVoice TTS service with local and remote backends.""" import asyncio +import io import json import logging import os import subprocess +import sys +import tempfile +import urllib.parse +import wave from pathlib import Path from typing import Optional -from agent.config import TTS_MODEL, TTS_SAMPLE_RATE +import httpx + +from agent.config import ( + TTS_MODEL, + TTS_REMOTE_MAX_BYTES, + TTS_REMOTE_TOKEN, + TTS_REMOTE_TIMEOUT_S, + TTS_REMOTE_URL, + TTS_SAMPLE_RATE, + TTS_BACKEND, +) logger = logging.getLogger(__name__) -# Default to python3.10 (has torch/torchaudio/omnivoice); override with TTS_PYTHON_BIN if needed -PYTHON_BIN = os.environ.get("TTS_PYTHON_BIN", "python3.10") +# The launcher starts FlowKit with its venv interpreter. Keep the override for deployments that +# intentionally use a separate OmniVoice environment, but never require an executable named +# python3.10 to be present on PATH. +PYTHON_BIN = os.environ.get("TTS_PYTHON_BIN", sys.executable) -# Inline script template for TTS generation via subprocess +# Inline script template for local subprocess generation. _TTS_SCRIPT = """ import sys, json, torch, torchaudio @@ -37,7 +54,7 @@ print(json.dumps({"ok": True, "path": args["output"]})) """ -# Batch script — loads model once, generates for multiple texts +# Batch script — loads model once for local generation of multiple texts. _TTS_BATCH_SCRIPT = """ import sys, json, torch, torchaudio from pathlib import Path @@ -73,6 +90,131 @@ """ +def _remote_endpoint() -> str: + if TTS_BACKEND != "remote": + raise RuntimeError(f"Unsupported TTS_BACKEND: {TTS_BACKEND!r}") + if not TTS_REMOTE_URL: + raise RuntimeError("TTS_REMOTE_URL is required when TTS_BACKEND=remote") + parsed = urllib.parse.urlparse(TTS_REMOTE_URL) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise RuntimeError("TTS_REMOTE_URL must be an absolute http(s) URL") + if parsed.username or parsed.password or parsed.query or parsed.fragment: + raise RuntimeError("TTS_REMOTE_URL must not contain credentials, query, or fragment") + root = TTS_REMOTE_URL.rstrip("/") + return root if root.endswith("/v1/tts") else f"{root}/v1/tts" + + +def _validate_wav_bytes(audio_bytes: bytes) -> None: + if len(audio_bytes) < 12 or audio_bytes[:4] != b"RIFF" or audio_bytes[8:12] != b"WAVE": + raise RuntimeError("Remote TTS returned a non-WAVE response") + try: + with wave.open(io.BytesIO(audio_bytes), "rb") as wav: + frame_count = wav.getnframes() + frame_width = wav.getnchannels() * wav.getsampwidth() + if frame_count <= 0 or wav.getframerate() <= 0 or frame_width <= 0: + raise RuntimeError("Remote TTS returned an empty WAV") + if len(wav.readframes(frame_count)) != frame_count * frame_width: + raise RuntimeError("Remote TTS returned a truncated WAV") + except (wave.Error, EOFError) as error: + raise RuntimeError("Remote TTS returned an invalid WAV") from error + + +def _write_wav_atomically(output_path: str, audio_bytes: bytes) -> None: + _validate_wav_bytes(audio_bytes) + destination = Path(output_path) + destination.parent.mkdir(parents=True, exist_ok=True) + temporary_path: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="wb", + dir=destination.parent, + prefix=f".{destination.name}.", + suffix=".tmp", + delete=False, + ) as temporary: + temporary_path = Path(temporary.name) + temporary.write(audio_bytes) + temporary.flush() + os.fsync(temporary.fileno()) + os.replace(temporary_path, destination) + temporary_path = None + finally: + if temporary_path is not None: + temporary_path.unlink(missing_ok=True) + + +def _read_limited(path: Path, max_bytes: int) -> bytes: + with path.open("rb") as source: + return source.read(max_bytes + 1) + + +async def _read_reference_audio(ref_audio: str) -> tuple[str, bytes]: + reference = Path(ref_audio) + if not reference.is_file(): + raise RuntimeError("Reference audio file does not exist") + try: + audio_bytes = await asyncio.to_thread(_read_limited, reference, TTS_REMOTE_MAX_BYTES) + except OSError as error: + raise RuntimeError("Reference audio could not be read") from error + if len(audio_bytes) > TTS_REMOTE_MAX_BYTES: + raise RuntimeError("Reference audio exceeds the remote TTS size limit") + _validate_wav_bytes(audio_bytes) + return reference.name, audio_bytes + + +async def _generate_remote( + text: str, + output_path: str, + instruct: Optional[str] = None, + ref_audio: Optional[str] = None, + ref_text: Optional[str] = None, + speed: float = 1.0, +) -> str: + data = {"text": text, "speed": str(speed)} + if instruct: + data["instruct"] = instruct + if ref_text: + data["ref_text"] = ref_text + + files = None + # OmniVoice cloning requires both fields. Preserve local compatibility when a caller supplies + # only ref_audio by following the instruct/generic path rather than leaking a local path. + if ref_audio and ref_text: + filename, audio_bytes = await _read_reference_audio(ref_audio) + files = {"ref_audio": (filename, audio_bytes, "audio/wav")} + + headers = {} + if TTS_REMOTE_TOKEN: + headers["authorization"] = f"Bearer {TTS_REMOTE_TOKEN}" + + timeout = httpx.Timeout(TTS_REMOTE_TIMEOUT_S) + try: + async with httpx.AsyncClient(timeout=timeout) as client: + async with client.stream( + "POST", + _remote_endpoint(), + data=data, + files=files, + headers=headers, + ) as response: + if response.status_code < 200 or response.status_code >= 300: + raise RuntimeError(f"Remote TTS failed with HTTP {response.status_code}") + content_type = response.headers.get("content-type", "").split(";", 1)[0].lower() + if content_type != "audio/wav": + raise RuntimeError("Remote TTS returned an unexpected content type") + audio = bytearray() + async for chunk in response.aiter_bytes(): + if len(audio) + len(chunk) > TTS_REMOTE_MAX_BYTES: + raise RuntimeError("Remote TTS response exceeds the size limit") + audio.extend(chunk) + except httpx.HTTPError as error: + raise RuntimeError("Remote TTS request failed") from error + + _write_wav_atomically(output_path, bytes(audio)) + logger.info("Remote TTS saved to %s", output_path) + return output_path + + async def generate_speech( text: str, output_path: str, @@ -81,9 +223,13 @@ async def generate_speech( ref_text: Optional[str] = None, speed: float = 1.0, ) -> str: - """Generate speech for text via subprocess. Returns path to WAV file.""" - Path(output_path).parent.mkdir(parents=True, exist_ok=True) + """Generate speech and return a local WAV path.""" + if TTS_BACKEND == "remote": + return await _generate_remote(text, output_path, instruct, ref_audio, ref_text, speed) + if TTS_BACKEND != "local": + raise RuntimeError(f"Unsupported TTS_BACKEND: {TTS_BACKEND!r}") + Path(output_path).parent.mkdir(parents=True, exist_ok=True) args = { "model": TTS_MODEL, "text": text, @@ -98,9 +244,8 @@ async def generate_speech( if ref_text: args["ref_text"] = ref_text - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() result = await loop.run_in_executor(None, _run_tts_subprocess, args) - if not result.get("ok"): raise RuntimeError(f"TTS failed: {result.get('error', 'unknown')}") @@ -109,10 +254,12 @@ async def generate_speech( def _run_tts_subprocess(args: dict) -> dict: - """Run TTS subprocess.""" + """Run one local TTS subprocess.""" proc = subprocess.run( [PYTHON_BIN, "-c", _TTS_SCRIPT, json.dumps(args)], - capture_output=True, text=True, timeout=120, + capture_output=True, + text=True, + timeout=120, ) if proc.returncode != 0: return {"ok": False, "error": proc.stderr[-500:] if proc.stderr else "unknown error"} @@ -122,6 +269,11 @@ def _run_tts_subprocess(args: dict) -> dict: return {"ok": False, "error": proc.stdout[-200:] + proc.stderr[-200:]} +def _wav_duration_seconds(path: str) -> float: + with wave.open(path, "rb") as wav: + return wav.getnframes() / wav.getframerate() + + async def generate_video_narration( scenes: list[dict], output_dir: str, @@ -130,15 +282,12 @@ async def generate_video_narration( ref_text: Optional[str] = None, speed: float = 1.0, ) -> list[dict]: - """Generate narration WAVs for scenes with narrator_text. - - Uses batch subprocess — loads model once for all scenes. - Returns list of result dicts. - """ + """Generate narration WAVs for scenes with narrator_text.""" out_dir = Path(output_dir) out_dir.mkdir(parents=True, exist_ok=True) + if TTS_BACKEND not in {"local", "remote"}: + raise RuntimeError(f"Unsupported TTS_BACKEND: {TTS_BACKEND!r}") - # Build batch items (only scenes with narrator_text) items = [] scene_map = {} for scene in scenes: @@ -150,17 +299,41 @@ async def generate_video_narration( continue wav_path = str(out_dir / f"scene_{display_order:03d}_{scene_id}.wav") - # Skip if WAV already exists and is non-trivial (>1KB) if Path(wav_path).exists() and Path(wav_path).stat().st_size > 1024: logger.info("Skipping scene %03d (WAV exists: %s)", display_order, wav_path) - scene_map[scene_id] = {"display_order": display_order, "narrator_text": narrator_text, "skipped": True, "wav_path": wav_path} + scene_map[scene_id] = { + "display_order": display_order, + "narrator_text": narrator_text, + "skipped": True, + "wav_path": wav_path, + } continue items.append({"id": scene_id, "text": narrator_text, "output": wav_path}) scene_map[scene_id] = {"display_order": display_order, "narrator_text": narrator_text} - # Run batch subprocess if there are items batch_results = {} - if items: + if items and TTS_BACKEND == "remote": + # Keep one request per scene so failures remain attributable and the remote service's + # single warm model is not driven concurrently. + for item in items: + try: + await generate_speech( + text=item["text"], + output_path=item["output"], + instruct=instruct, + ref_audio=ref_audio, + ref_text=ref_text, + speed=speed, + ) + batch_results[item["id"]] = { + "id": item["id"], + "ok": True, + "path": item["output"], + "duration": _wav_duration_seconds(item["output"]), + } + except Exception as error: + batch_results[item["id"]] = {"id": item["id"], "ok": False, "error": str(error)} + elif items: args = { "model": TTS_MODEL, "sample_rate": TTS_SAMPLE_RATE, @@ -174,12 +347,11 @@ async def generate_video_narration( if ref_text: args["ref_text"] = ref_text - loop = asyncio.get_event_loop() + loop = asyncio.get_running_loop() raw = await loop.run_in_executor(None, _run_batch_subprocess, args) - for r in raw: - batch_results[r["id"]] = r + for result in raw: + batch_results[result["id"]] = result - # Build final results for all scenes results = [] for scene in scenes: scene_id = scene.get("id") @@ -198,27 +370,27 @@ async def generate_video_narration( }) continue - sm = scene_map.get(scene_id, {}) - if sm.get("skipped"): + scene_result = scene_map.get(scene_id, {}) + if scene_result.get("skipped"): results.append({ "scene_id": scene_id, "display_order": display_order, "narrator_text": narrator_text, - "audio_path": sm["wav_path"], + "audio_path": scene_result["wav_path"], "duration": None, "status": "COMPLETED", "error": None, }) continue - br = batch_results.get(scene_id, {}) - if br.get("ok"): + batch_result = batch_results.get(scene_id, {}) + if batch_result.get("ok"): results.append({ "scene_id": scene_id, "display_order": display_order, "narrator_text": narrator_text, - "audio_path": br.get("path"), - "duration": br.get("duration"), + "audio_path": batch_result.get("path"), + "duration": batch_result.get("duration"), "status": "COMPLETED", "error": None, }) @@ -230,18 +402,20 @@ async def generate_video_narration( "audio_path": None, "duration": None, "status": "FAILED", - "error": br.get("error", "not processed"), + "error": batch_result.get("error", "not processed"), }) return results def _run_batch_subprocess(args: dict) -> list[dict]: - """Run batch TTS subprocess. Model loads once.""" - timeout = 180 + len(args.get("items", [])) * 45 # ~180s model load + ~45s per scene + """Run local batch TTS subprocess. Model loads once.""" + timeout = 180 + len(args.get("items", [])) * 45 proc = subprocess.run( [PYTHON_BIN, "-c", _TTS_BATCH_SCRIPT, json.dumps(args)], - capture_output=True, text=True, timeout=timeout, + capture_output=True, + text=True, + timeout=timeout, ) if proc.returncode != 0: error = proc.stderr[-500:] if proc.stderr else "unknown" diff --git a/requirements.txt b/requirements.txt index c2cd265b..be00c9db 100644 --- a/requirements.txt +++ b/requirements.txt @@ -6,3 +6,4 @@ pydantic>=2.5.0 aiohttp>=3.9.0 httpx>=0.25.0 anthropic>=0.40.0 +python-multipart>=0.0.9 diff --git a/run-omnivoice-api.sh b/run-omnivoice-api.sh new file mode 100755 index 00000000..630ffa64 --- /dev/null +++ b/run-omnivoice-api.sh @@ -0,0 +1,23 @@ +#!/usr/bin/env bash +# Start the standalone warm OmniVoice inference API. +set -euo pipefail + +FLOWKIT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PYTHON_BIN="${OMNIVOICE_PYTHON:-${TTS_PYTHON_BIN:-$HOME/.venvs/flowkit/bin/python}}" + +if [[ ! -x "$PYTHON_BIN" ]]; then + echo "OmniVoice Python not found at $PYTHON_BIN" >&2 + echo "Install FlowKit's Python 3.10+ environment or set OMNIVOICE_PYTHON." >&2 + exit 1 +fi + +if ! "$PYTHON_BIN" -c 'import multipart' >/dev/null 2>&1; then + echo "python-multipart is missing from $PYTHON_BIN" >&2 + echo "Install it with: $PYTHON_BIN -m pip install python-multipart" >&2 + exit 1 +fi + +cd "$FLOWKIT_DIR" +exec "$PYTHON_BIN" -m uvicorn agent.omnivoice_api:app \ + --host "${OMNIVOICE_API_HOST:-127.0.0.1}" \ + --port "${OMNIVOICE_API_PORT:-8200}" diff --git a/skills/fk-gen-narrator.md b/skills/fk-gen-narrator.md index 4e49f55d..5a874ea2 100644 --- a/skills/fk-gen-narrator.md +++ b/skills/fk-gen-narrator.md @@ -163,6 +163,12 @@ Ask user: "Review OK? Type 'yes' to generate TTS, or 'edit N' to modify scene N' **CRITICAL: Always pass BOTH `ref_audio` AND `ref_text` together.** Without `ref_text`, OmniVoice falls back to generic voice → each scene sounds different. +> **Backend is transparent here.** `/api/tts/generate` uses whatever backend FlowKit +> is configured with (`TTS_BACKEND=local` default, or `remote` for a GPU inference +> service). This skill is identical either way — no request changes. See +> `fk-gen-tts-template.md` and the README "Optional remote OmniVoice API" section for +> backend setup. + ### Proven workflow (per-scene via `/api/tts/generate`): The batch endpoint (`/api/videos/<VID>/narrate`) can timeout on large batches (40+ scenes). diff --git a/skills/fk-gen-tts-template.md b/skills/fk-gen-tts-template.md index 8b101a47..8be3892a 100644 --- a/skills/fk-gen-tts-template.md +++ b/skills/fk-gen-tts-template.md @@ -16,43 +16,55 @@ Source: https://github.com/tuannguyenhoangit-droid/OmniVoice > **Windows users:** Run all setup commands inside **WSL** or **Git Bash** (not CMD/PowerShell). The project's `setup.sh` and all bash scripts require a Unix shell. -**Step 1 — Install PyTorch** (in a fresh venv recommended): +**Step 1 — Install PyTorch** (in the FlowKit launcher venv, or a separate venv if you set +`TTS_PYTHON_BIN`): ```bash # macOS Apple Silicon (CPU — recommended for GLA, MPS produces gibberish) -pip install torch==2.8.0 torchaudio==2.8.0 +$HOME/.venvs/flowkit/bin/python -m pip install torch==2.8.0 torchaudio==2.8.0 # Linux / WSL with NVIDIA GPU -pip install torch==2.8.0+cu128 torchaudio==2.8.0+cu128 --extra-index-url https://download.pytorch.org/whl/cu128 +python -m pip install torch==2.8.0+cu128 torchaudio==2.8.0+cu128 \ + --extra-index-url https://download.pytorch.org/whl/cu128 # Linux / WSL CPU-only -pip install torch==2.8.0 torchaudio==2.8.0 +python -m pip install torch==2.8.0 torchaudio==2.8.0 ``` **Step 2 — Install OmniVoice** (choose one): ```bash # From PyPI (stable) -pip install omnivoice +$HOME/.venvs/flowkit/bin/python -m pip install omnivoice # From source -pip install git+https://github.com/k2-fsa/OmniVoice.git +$HOME/.venvs/flowkit/bin/python -m pip install \ + git+https://github.com/k2-fsa/OmniVoice.git # Dev install -git clone https://github.com/k2-fsa/OmniVoice.git && cd OmniVoice && pip install -e . +git clone https://github.com/k2-fsa/OmniVoice.git +cd OmniVoice +$HOME/.venvs/flowkit/bin/python -m pip install -e . ``` -**Step 3 — Point GLA to the right Python** (if OmniVoice is in a separate venv): +**Step 3 — Verify installation:** ```bash -export TTS_PYTHON_BIN=/path/to/omnivoice-venv/bin/python3 +$HOME/.venvs/flowkit/bin/python -c \ + 'from omnivoice import OmniVoice; print("OmniVoice OK")' ``` -If OmniVoice is installed in the same env as the agent, no extra config needed. +FlowKit's local TTS subprocess defaults to the interpreter that launched FlowKit +(`sys.executable`). `TTS_PYTHON_BIN` is optional and is only needed when OmniVoice +is installed in another venv. -**Verify installation:** +For remote GPU inference, run `agent/omnivoice_api.py` as a separate FastAPI service. +The FlowKit remote backend uploads `ref_audio` bytes and expects raw `audio/wav` bytes +back; it never sends a local filesystem path to the remote service. + +**HuggingFace mirror** (if model download is slow): ```bash -python3 -c "from omnivoice import OmniVoice; print('OK')" +export HF_ENDPOINT="https://hf-mirror.com" ``` **HuggingFace mirror** (if model download is slow): diff --git a/skills/fk-import-voice.md b/skills/fk-import-voice.md index 6772e0fa..5ce23da3 100644 --- a/skills/fk-import-voice.md +++ b/skills/fk-import-voice.md @@ -110,6 +110,6 @@ Play the output for user to verify voice quality matches the original. ## Notes - **Transcript accuracy matters** — `ref_text` must match the audio closely for good voice cloning. Always confirm with the user. -- **CPU only** — whisper large-v3 and OmniVoice both run on CPU. MPS produces artifacts. +- **CPU by default** — whisper large-v3 and local OmniVoice both run on CPU (MPS produces artifacts). For GPU speed, run OmniVoice as a remote service and set `TTS_BACKEND=remote` (see `fk-gen-tts-template.md`); transcription still runs locally on CPU. - **3-10s audio** — shorter clips lack enough voice characteristics; longer clips slow down cloning. - **One template per voice** — don't register multiple templates for the same voice. Update instead. diff --git a/tests/unit/test_flow_batch.py b/tests/unit/test_flow_batch.py index 63b8761d..47c75245 100644 --- a/tests/unit/test_flow_batch.py +++ b/tests/unit/test_flow_batch.py @@ -231,3 +231,26 @@ def test_every_resolved_video_model_is_one_flow_accepts(self): from agent.config import VIDEO_MODELS key = VIDEO_MODELS.get(tier, {}).get(gen, {}).get(aspect) assert fb.resolve_video_model(key) in fb.VIDEO_MODELS + + +class TestProjectRpcs: + def test_create_request_carries_the_captured_slots(self): + assert inner(fb.create_project_request("My Film")) == [ + "projects/*", [None, ["My Film"]], [None, fb.SURFACE_ID]] + + def test_delete_request_names_the_project_path(self): + assert inner(fb.delete_project_request("11111111-2222-3333-4444-555555555555")) == [ + "projects/11111111-2222-3333-4444-555555555555"] + + def test_read_created_id_takes_the_leading_uuid(self): + pid = "8e30afb8-92d0-4d0b-b652-f0a79faca9f5" + assert fb.read_created_project_id([pid, ["My Film"]]) == pid + + def test_read_created_id_rejects_a_trailing_newline(self): + r"""`$` would let "<uuid>\n" through; fullmatch must reject it.""" + with pytest.raises(fb.FlowBatchError): + fb.read_created_project_id(["8e30afb8-92d0-4d0b-b652-f0a79faca9f5\n", ["x"]]) + + def test_read_created_id_rejects_a_non_uuid_lead(self): + with pytest.raises(fb.FlowBatchError): + fb.read_created_project_id([["My Film"], "not-a-uuid"]) diff --git a/tests/unit/test_flow_client_batch.py b/tests/unit/test_flow_client_batch.py index 733f5244..40b972ea 100644 --- a/tests/unit/test_flow_client_batch.py +++ b/tests/unit/test_flow_client_batch.py @@ -311,12 +311,73 @@ async def test_create_project_hands_back_the_pinned_one(self, client): result = await client.create_project("My Film") assert result["data"]["projectId"] == PROJECT - async def test_create_project_without_a_pin_explains_itself(self, client, monkeypatch): + async def test_create_project_without_a_pin_creates_one(self, client, monkeypatch): + """Unpinned, the batch path mints a real project and hands back its uuid.""" import agent.services.flow_client as module monkeypatch.setattr(module, "FLOW_PROJECT_ID", "") + new_pid = "8e30afb8-92d0-4d0b-b652-f0a79faca9f5" + client.responses[fb.RPC_CREATE_PROJECT] = { + "data": envelope(fb.RPC_CREATE_PROJECT, [new_pid, ["My Film"]]) + } result = await client.create_project("My Film") - assert "NO_FLOW_PROJECT" in result["error"] - assert "FLOW_PROJECT_ID" in result["error"] + assert result["data"]["projectId"] == new_pid + + async def test_create_project_sends_the_captured_request(self, client, monkeypatch): + """The jHPbke envelope carries ["projects/*", [None, [title]], [None, 22]].""" + import agent.services.flow_client as module + monkeypatch.setattr(module, "FLOW_PROJECT_ID", "") + client.responses[fb.RPC_CREATE_PROJECT] = { + "data": envelope(fb.RPC_CREATE_PROJECT, ["8e30afb8-92d0-4d0b-b652-f0a79faca9f5", ["My Film"]]) + } + await client.create_project("My Film") + assert client.calls[0]["rpcid"] == fb.RPC_CREATE_PROJECT + inner = json.loads(json.loads(client.calls[0]["freq"])[0][0][1]) + assert inner == ["projects/*", [None, ["My Film"]], [None, fb.SURFACE_ID]] + + async def test_create_project_rejects_an_id_less_response(self, client, monkeypatch): + """A 200 whose lead slot is not a uuid is a failure, not a guess (accepted != used).""" + import agent.services.flow_client as module + monkeypatch.setattr(module, "FLOW_PROJECT_ID", "") + client.responses[fb.RPC_CREATE_PROJECT] = { + "data": envelope(fb.RPC_CREATE_PROJECT, ["not-a-uuid", ["My Film"]]) + } + result = await client.create_project("My Film") + assert "CREATE_PROJECT_FAILED" in result["error"] + + async def test_create_project_refuses_a_non_default_tool_on_batch(self, client, monkeypatch): + """The batch create has no tool slot, so it will not pretend to honour one.""" + import agent.services.flow_client as module + monkeypatch.setattr(module, "FLOW_PROJECT_ID", "") + result = await client.create_project("My Film", tool_name="WHISK") + assert "UNSUPPORTED_ON_BATCH_API" in result["error"] + assert not client.calls, "nothing should have been sent" + + async def test_create_project_uses_the_legacy_path_when_batch_is_off(self, client, monkeypatch): + """USE_BATCH_RPC=0 still routes to the (post-mortem) tRPC create, untouched.""" + import agent.services.flow_client as module + monkeypatch.setattr(module, "USE_BATCH_RPC", False) + seen = {} + async def fake_legacy(title, tool): + seen["args"] = (title, tool) + return {"status": 200, "data": {"projectId": "legacy"}} + client._legacy_create_project = fake_legacy + result = await client.create_project("My Film", tool_name="PINHOLE") + assert seen["args"] == ("My Film", "PINHOLE") + assert result["data"]["projectId"] == "legacy" + assert not client.calls, "the batch transport should be untouched" + + async def test_delete_project_calls_QI2zvc_with_the_project_path(self, client): + client.responses[fb.RPC_DELETE_PROJECT] = {"data": envelope(fb.RPC_DELETE_PROJECT, [])} + result = await client.delete_project(PROJECT) + assert client.calls[0]["rpcid"] == fb.RPC_DELETE_PROJECT + inner = json.loads(json.loads(client.calls[0]["freq"])[0][0][1]) + assert inner == [f"projects/{PROJECT}"] + assert result["data"]["projectId"] == PROJECT + + async def test_delete_project_surfaces_a_transport_error(self, client): + client.responses[fb.RPC_DELETE_PROJECT] = {"error": "NO_FLOW_TAB"} + result = await client.delete_project(PROJECT) + assert "DELETE_PROJECT_FAILED" in result["error"] async def test_credits_answers_the_configured_tier_rather_than_guessing(self, client): result = await client.get_credits() diff --git a/tests/unit/test_projects_route.py b/tests/unit/test_projects_route.py new file mode 100644 index 00000000..eeb64ccd --- /dev/null +++ b/tests/unit/test_projects_route.py @@ -0,0 +1,129 @@ +"""The project DELETE route's Flow-side gate. + +The route reaches through to Google Flow, so these lock down when it does and, +more importantly, when it must not: never for a project it does not already +track, never for the pinned shared project, and never leaving local and remote +out of step when the remote call fails. +""" +import pytest + +from agent.api import projects + + +class FakeRepo: + def __init__(self, exists=True): + self._exists = exists + self.deleted = [] + self.created = [] + + async def get_project(self, pid): + return {"id": pid} if self._exists else None + + async def create_project(self, **kw): + self.created.append(kw) + return {"id": kw.get("id")} + + async def delete_project(self, pid): + self.deleted.append(pid) + return True + + +class FakeClient: + def __init__(self, connected=True, error=None, pin="pinned-project"): + self.connected = connected + self._error = error + self._pin = pin + self.deleted = [] + + def flow_project_id(self, requested=None): + return requested or self._pin + + async def get_credits(self): + return {"status": 200, "data": {"userPaygateTier": "PAYGATE_TIER_TWO"}} + + async def delete_project(self, pid): + self.deleted.append(pid) + if self._error: + return {"error": self._error} + return {"status": 200, "data": {"projectId": pid}} + + +@pytest.fixture +def wire(monkeypatch): + """Point the route at fake collaborators; batch path with a pinned project.""" + def _wire(*, repo, client, batch=True, pin="pinned-project"): + monkeypatch.setattr(projects, "USE_BATCH_RPC", batch) + monkeypatch.setattr(projects, "FLOW_PROJECT_ID", pin) + monkeypatch.setattr(projects, "_get_repo", lambda: repo) + monkeypatch.setattr(projects, "get_flow_client", lambda: client) + return _wire + + +async def test_missing_project_is_404_before_any_remote_call(wire): + repo, client = FakeRepo(exists=False), FakeClient() + wire(repo=repo, client=client) + with pytest.raises(projects.HTTPException) as exc: + await projects.delete("nope") + assert exc.value.status_code == 404 + assert client.deleted == [], "a stray id must not reach Flow" + assert repo.deleted == [] + + +async def test_pinned_project_is_removed_locally_only(wire): + repo, client = FakeRepo(), FakeClient() + wire(repo=repo, client=client, pin="pinned-project") + result = await projects.delete("pinned-project") + assert result == {"ok": True} + assert client.deleted == [], "the shared project is never deleted on Flow" + assert repo.deleted == ["pinned-project"] + + +async def test_tracked_project_is_deleted_remote_first_then_local(wire): + repo, client = FakeRepo(), FakeClient() + wire(repo=repo, client=client) + result = await projects.delete("proj-1") + assert result == {"ok": True} + assert client.deleted == ["proj-1"] + assert repo.deleted == ["proj-1"] + + +async def test_remote_failure_keeps_the_local_row(wire): + repo, client = FakeRepo(), FakeClient(error="NO_FLOW_TAB") + wire(repo=repo, client=client) + with pytest.raises(projects.HTTPException) as exc: + await projects.delete("proj-1") + assert exc.value.status_code == 502 + assert repo.deleted == [], "local row stays put so the two do not drift" + + +async def test_offline_extension_deletes_nothing(wire): + repo, client = FakeRepo(), FakeClient(connected=False) + wire(repo=repo, client=client) + with pytest.raises(projects.HTTPException) as exc: + await projects.delete("proj-1") + assert exc.value.status_code == 503 + assert client.deleted == [] + assert repo.deleted == [] + + +async def test_legacy_path_deletes_locally_without_touching_flow(wire): + repo, client = FakeRepo(), FakeClient() + wire(repo=repo, client=client, batch=False) + result = await projects.delete("proj-1") + assert result == {"ok": True} + assert client.deleted == [] + assert repo.deleted == ["proj-1"] + + +async def test_create_reuses_an_existing_local_project_instead_of_reinserting(wire): + """A pinned/request project already in the DB is returned, not re-INSERTed. + + The bare INSERT in crud.create_project raises `UNIQUE constraint failed: + project.id` on a duplicate id, which surfaced as an uncaught 500. + """ + from agent.models.project import ProjectCreate + repo, client = FakeRepo(exists=True), FakeClient(pin="pinned-project") + wire(repo=repo, client=client, pin="pinned-project") + result = await projects.create(ProjectCreate(name="anything", material="realistic")) + assert result == {"id": "pinned-project"}, "returned the existing project row" + assert repo.created == [], "must not attempt a second INSERT of the same id" diff --git a/tests/unit/test_tts.py b/tests/unit/test_tts.py new file mode 100644 index 00000000..e9791290 --- /dev/null +++ b/tests/unit/test_tts.py @@ -0,0 +1,187 @@ +"""Focused tests for local-path and remote OmniVoice TTS contracts.""" + +import io +import wave + +import pytest +from agent import omnivoice_api +from agent.services import tts + + +def _wav_bytes() -> bytes: + output = io.BytesIO() + with wave.open(output, "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(24000) + wav.writeframes(b"\x00\x00" * 240) + return output.getvalue() + + +class _FakeResponse: + def __init__(self, body: bytes, status_code: int = 200, content_type: str = "audio/wav"): + self.status_code = status_code + self.headers = {"content-type": content_type} + self._body = body + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def aread(self): + return self._body + + async def aiter_bytes(self): + yield self._body + + +class _FakeClient: + response = _FakeResponse(_wav_bytes()) + calls = [] + + def __init__(self, **kwargs): + self.timeout = kwargs["timeout"] + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + def stream(self, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + return self.response + + +@pytest.mark.asyncio +async def test_remote_generation_uploads_reference_bytes_and_writes_local_wav(tmp_path, monkeypatch): + reference = tmp_path / "reference.wav" + reference.write_bytes(_wav_bytes()) + output = tmp_path / "scene.wav" + _FakeClient.calls = [] + + monkeypatch.setattr(tts, "TTS_BACKEND", "remote") + monkeypatch.setattr(tts, "TTS_REMOTE_URL", "https://tts.example.test") + monkeypatch.setattr(tts, "TTS_REMOTE_TOKEN", "secret") + monkeypatch.setattr(tts.httpx, "AsyncClient", _FakeClient) + + result = await tts._generate_remote( + text="Xin chao", + output_path=str(output), + ref_audio=str(reference), + ref_text="Mau transcript", + speed=1.1, + ) + + assert result == str(output) + + + assert output.read_bytes() == _wav_bytes() + method, url, request = _FakeClient.calls[0] + assert method == "POST" + assert url == "https://tts.example.test/v1/tts" + assert request["headers"] == {"authorization": "Bearer secret"} + assert request["data"] == {"text": "Xin chao", "speed": "1.1", "ref_text": "Mau transcript"} + assert request["files"]["ref_audio"][1] == _wav_bytes() + + +@pytest.mark.asyncio +async def test_remote_generation_does_not_replace_target_on_invalid_response(tmp_path, monkeypatch): + output = tmp_path / "scene.wav" + original = b"previous audio" + output.write_bytes(original) + + class InvalidClient(_FakeClient): + response = _FakeResponse(b"not wav", content_type="audio/wav") + + monkeypatch.setattr(tts, "TTS_BACKEND", "remote") + monkeypatch.setattr(tts, "TTS_REMOTE_URL", "http://127.0.0.1:8200") + monkeypatch.setattr(tts.httpx, "AsyncClient", InvalidClient) + + with pytest.raises(RuntimeError, match="non-WAVE"): + await tts._generate_remote("Xin chao", str(output)) + + assert output.read_bytes() == original + + +def test_write_wav_atomically_creates_valid_wav(tmp_path): + output = tmp_path / "nested" / "scene.wav" + + tts._write_wav_atomically(str(output), _wav_bytes()) + + assert output.exists() + with wave.open(str(output), "rb") as wav: + assert wav.getnframes() == 240 + assert wav.getframerate() == 24000 + + +def test_wav_validator_rejects_truncated_payload(): + with pytest.raises(RuntimeError, match="truncated"): + tts._validate_wav_bytes(_wav_bytes()[:-10]) + + +def test_standalone_api_auth_and_wav_response(monkeypatch): + from fastapi.testclient import TestClient + + monkeypatch.setattr(omnivoice_api, "OMNIVOICE_API_TOKEN", "secret") + monkeypatch.setattr(omnivoice_api, "_load_model", lambda: None) + monkeypatch.setattr(omnivoice_api, "_generate_wav_bytes", lambda *args: _wav_bytes()) + + with TestClient(omnivoice_api.app) as client: + unauthorized = client.post("/v1/tts", data={"text": "Xin chao"}, headers={}) + assert unauthorized.status_code == 401 + + authorized = client.post( + "/v1/tts", + data={"text": "Xin chao", "speed": "1.0", "ref_text": "Mau transcript"}, + files={"ref_audio": ("reference.wav", _wav_bytes(), "audio/wav")}, + headers={"Authorization": "Bearer secret"}, + ) + assert authorized.status_code == 200 + assert authorized.headers["content-type"] == "audio/wav" + assert authorized.content == _wav_bytes() + + +def test_standalone_api_rejects_oversized_request(monkeypatch): + from fastapi.testclient import TestClient + + monkeypatch.setattr(omnivoice_api, "OMNIVOICE_API_TOKEN", "") + monkeypatch.setattr(omnivoice_api, "OMNIVOICE_MAX_REQUEST_BYTES", 100) + monkeypatch.setattr(omnivoice_api, "_load_model", lambda: None) + + with TestClient(omnivoice_api.app) as client: + response = client.post("/v1/tts", data={"text": "x" * 1000}) + + assert response.status_code == 413 + + +def test_standalone_api_rejects_nonfinite_speed(monkeypatch): + from fastapi.testclient import TestClient + + monkeypatch.setattr(omnivoice_api, "OMNIVOICE_API_TOKEN", "") + monkeypatch.setattr(omnivoice_api, "_load_model", lambda: None) + + with TestClient(omnivoice_api.app) as client: + response = client.post("/v1/tts", data={"text": "Xin chao", "speed": "nan"}) + + assert response.status_code == 422 + + +@pytest.mark.asyncio +async def test_remote_generation_bounds_reference_read(tmp_path, monkeypatch): + reference = tmp_path / "reference.wav" + reference.write_bytes(b"x" * 11) + + monkeypatch.setattr(tts, "TTS_BACKEND", "remote") + monkeypatch.setattr(tts, "TTS_REMOTE_URL", "http://127.0.0.1:8200") + monkeypatch.setattr(tts, "TTS_REMOTE_MAX_BYTES", 10) + + with pytest.raises(RuntimeError, match="size limit"): + await tts._generate_remote( + "Xin chao", + str(tmp_path / "scene.wav"), + ref_audio=str(reference), + ref_text="Mau transcript", + )