diff --git a/agent/api/projects.py b/agent/api/projects.py index 9a6f33c36..69fe112b6 100644 --- a/agent/api/projects.py +++ b/agent/api/projects.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel +from agent.utils.paths import file_url_to_path from agent.config import BASE_DIR from agent.models.project import Project, ProjectCreate, ProjectUpdate from agent.models.character import Character @@ -212,6 +213,7 @@ async def create(body: ProjectCreate): material=material_id, allow_music=create_data.get("allow_music", False), allow_voice=create_data.get("allow_voice", False), + video_model_family=create_data.get("video_model_family", "veo"), ) # Step 3: Create reference entities (characters, locations, assets) with profiles @@ -467,8 +469,8 @@ async def generate_thumbnail(pid: str, body: ThumbnailRequest): except aiohttp.ClientError as e: raise HTTPException(502, f"Failed to download image: {e}") from e elif gen_result.url.startswith("file://"): - src = Path(urlparse(gen_result.url).path) - if not src.is_file(): + src = file_url_to_path(gen_result.url) + if src is None or not src.is_file(): raise HTTPException(502, f"Provider returned missing file: {gen_result.url}") shutil.copy2(src, output_path) diff --git a/agent/api/videos.py b/agent/api/videos.py index bfc5be79a..2a5e56273 100644 --- a/agent/api/videos.py +++ b/agent/api/videos.py @@ -10,6 +10,7 @@ from fastapi import APIRouter, HTTPException from pydantic import BaseModel, Field +from agent.utils.paths import file_url_to_path from agent.config import OUTPUT_DIR from agent.db import crud from agent.models.video import Video, VideoCreate, VideoUpdate @@ -131,10 +132,10 @@ def _resolve_media_local(url: str, tmpdir: Path) -> Optional[str]: """Resolve a media URL to a local file path (downloading http(s) when needed).""" if not url: return None + local = file_url_to_path(url) + if local is not None: + return str(local) if local.is_file() else None parsed = urlparse(url) - if parsed.scheme == "file": - p = Path(parsed.path) - return str(p) if p.is_file() else None if parsed.scheme in ("http", "https"): dest = tmpdir / f"dl_{abs(hash(url)) % 10**8}{Path(parsed.path).suffix or '.mp4'}" try: diff --git a/agent/config.py b/agent/config.py index 9c72cfc25..e020edc81 100644 --- a/agent/config.py +++ b/agent/config.py @@ -76,6 +76,15 @@ VIDEO_POLL_INTERVAL = int(os.environ.get("VIDEO_POLL_INTERVAL", "10")) # polling interval for video/upscale status MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "5")) VIDEO_POLL_TIMEOUT = int(os.environ.get("VIDEO_POLL_TIMEOUT", "420")) +# Omni Flash defaults used by the worker when a project's video_model_family +# is "omni_flash". Duration must be 4/6/8/10; resolution 360p or 720p. +OMNI_FLASH_DURATION_S = int(os.environ.get("OMNI_FLASH_DURATION_S", "8")) +OMNI_FLASH_RESOLUTION = os.environ.get("OMNI_FLASH_RESOLUTION", "720p") +# Flow provider pacing. Every generate mints a reCAPTCHA inside the Flow tab; +# firing them back-to-back from a background tab degrades the session score +# until Google answers PUBLIC_ERROR_UNUSUAL_ACTIVITY. Slow down when that hits. +FLOW_MAX_CONCURRENT = int(os.environ.get("FLOW_MAX_CONCURRENT", "5")) +FLOW_COOLDOWN_S = float(os.environ.get("FLOW_COOLDOWN_S", "10")) API_COOLDOWN = int(os.environ.get("API_COOLDOWN", "10")) # DEPRECATED: per-provider cooldown_s in provider capabilities is authoritative MAX_CONCURRENT_REQUESTS = int(os.environ.get("MAX_CONCURRENT_REQUESTS", "5")) # DEPRECATED: per-provider max_concurrent in provider capabilities is authoritative STALE_PROCESSING_TIMEOUT = int(os.environ.get("STALE_PROCESSING_TIMEOUT", "600")) # 10 min diff --git a/agent/db/crud.py b/agent/db/crud.py index a9ae2b1ac..00184fc8a 100644 --- a/agent/db/crud.py +++ b/agent/db/crud.py @@ -20,7 +20,7 @@ def _validate_table(table: str) -> None: # Column whitelists per table — prevents SQL injection via kwargs keys _COLUMNS = { "character": {"name", "slug", "entity_type", "description", "image_prompt", "voice_description", "reference_image_url", "media_id", "updated_at"}, - "project": {"name", "description", "story", "thumbnail_url", "language", "status", "user_paygate_tier", "narrator_voice", "narrator_ref_audio", "material", "allow_music", "allow_voice", "updated_at"}, + "project": {"name", "description", "story", "thumbnail_url", "language", "status", "user_paygate_tier", "narrator_voice", "narrator_ref_audio", "material", "allow_music", "allow_voice", "video_model_family", "updated_at"}, "video": {"title", "description", "display_order", "status", "orientation", "vertical_url", "horizontal_url", "thumbnail_url", "duration", "target_duration_s", "resolution", "youtube_id", "privacy", "tags", "updated_at"}, "scene": {"prompt", "image_prompt", "video_prompt", "character_names", "parent_scene_id", "chain_type", @@ -116,13 +116,13 @@ async def list_characters() -> list[dict]: # ─── Project ──────────────────────────────────────────────── -async def create_project(name: str, description: str = None, story: str = None, language: str = "en", user_paygate_tier: str = "PAYGATE_TIER_ONE", id: str = None, material: str = None, allow_music: bool = False, allow_voice: bool = False) -> dict: +async def create_project(name: str, description: str = None, story: str = None, language: str = "en", user_paygate_tier: str = "PAYGATE_TIER_ONE", id: str = None, material: str = None, allow_music: bool = False, allow_voice: bool = False, video_model_family: str = "veo") -> dict: db = await get_db() pid, now = id or _uuid(), _now() async with _db_lock: await db.execute( - "INSERT INTO project (id,name,description,story,language,user_paygate_tier,material,allow_music,allow_voice,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)", - (pid, name, description, story, language, user_paygate_tier, material, int(allow_music), int(allow_voice), now, now)) + "INSERT INTO project (id,name,description,story,language,user_paygate_tier,material,allow_music,allow_voice,video_model_family,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", + (pid, name, description, story, language, user_paygate_tier, material, int(allow_music), int(allow_voice), video_model_family or "veo", now, now)) await db.commit() return await _get_with_db(db, "project", "id", pid) diff --git a/agent/db/schema.py b/agent/db/schema.py index 3f696daa9..eeef28d3e 100644 --- a/agent/db/schema.py +++ b/agent/db/schema.py @@ -38,6 +38,7 @@ material TEXT DEFAULT 'realistic', allow_music INTEGER NOT NULL DEFAULT 0, allow_voice INTEGER NOT NULL DEFAULT 0, + video_model_family TEXT NOT NULL DEFAULT 'veo', created_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), updated_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) ); @@ -322,6 +323,9 @@ async def init_db(): if "allow_voice" not in project_columns: await db.execute("ALTER TABLE project ADD COLUMN allow_voice INTEGER NOT NULL DEFAULT 0") logger.info("Migrated: added allow_voice column to project table") + if "video_model_family" not in project_columns: + await db.execute("ALTER TABLE project ADD COLUMN video_model_family TEXT NOT NULL DEFAULT 'veo'") + logger.info("Migrated: added video_model_family column to project table") # Migration: add orientation to video table + backfill from scene data cursor = await db.execute("PRAGMA table_info(video)") video_columns = {row[1] for row in await cursor.fetchall()} diff --git a/agent/models/project.py b/agent/models/project.py index fdc6489fd..38a34fd7d 100644 --- a/agent/models/project.py +++ b/agent/models/project.py @@ -1,7 +1,12 @@ from pydantic import BaseModel, Field, model_validator -from typing import Optional +from typing import Literal, Optional from agent.models.enums import ProjectStatus, PaygateTier, EntityType +# Which Flow video family the worker uses for this project's scenes. +# veo -> Veo 3.1 i2v (1080p). r2v / first+last are unported on batch. +# omni_flash -> Omni Flash (720p, 4/6/8/10s); all frame/ref modes are ported. +VideoModelFamily = Literal["veo", "omni_flash"] + class CharacterInput(BaseModel): """Reference entity stub provided at project creation time.""" @@ -25,6 +30,7 @@ class ProjectCreate(BaseModel): style: Optional[str] = None # deprecated: use material instead; "3D"→"3d_pixar", "photorealistic"→"realistic" allow_music: bool = False # when True, skip "no background music" suffix in video prompts allow_voice: bool = False # when True, keep character dialogue in video audio (suppress only music/narration) + video_model_family: VideoModelFamily = "veo" characters: Optional[list[CharacterInput]] = None @model_validator(mode="before") @@ -52,6 +58,7 @@ class ProjectUpdate(BaseModel): material: Optional[str] = None allow_music: Optional[bool] = None allow_voice: Optional[bool] = None + video_model_family: Optional[VideoModelFamily] = None class Project(BaseModel): @@ -66,6 +73,7 @@ class Project(BaseModel): material: Optional[str] = None allow_music: bool = False allow_voice: bool = False + video_model_family: str = "veo" narrator_voice: Optional[str] = None narrator_ref_audio: Optional[str] = None created_at: Optional[str] = None diff --git a/agent/sdk/models/project.py b/agent/sdk/models/project.py index 003f2ea9f..ec44238c5 100644 --- a/agent/sdk/models/project.py +++ b/agent/sdk/models/project.py @@ -29,6 +29,7 @@ class Project(DomainModel): material: Optional[str] = None allow_music: bool = False allow_voice: bool = False + video_model_family: str = "veo" narrator_voice: Optional[str] = None narrator_ref_audio: Optional[str] = None created_at: Optional[str] = None diff --git a/agent/sdk/persistence/sqlite_repository.py b/agent/sdk/persistence/sqlite_repository.py index 84f0c3e75..7662a7f83 100644 --- a/agent/sdk/persistence/sqlite_repository.py +++ b/agent/sdk/persistence/sqlite_repository.py @@ -34,6 +34,7 @@ def _row_to_project(self, row: dict[str, Any]) -> Project: material=row.get("material"), allow_music=bool(row.get("allow_music", 0)), allow_voice=bool(row.get("allow_voice", 0)), + video_model_family=row.get("video_model_family") or "veo", narrator_voice=row.get("narrator_voice"), narrator_ref_audio=row.get("narrator_ref_audio"), created_at=row.get("created_at"), @@ -192,6 +193,7 @@ async def save_project(self, project: Project) -> None: user_paygate_tier=project.user_paygate_tier, allow_music=int(project.allow_music), allow_voice=int(project.allow_voice), + video_model_family=project.video_model_family, ) async def create_project( @@ -206,6 +208,7 @@ async def create_project( material: Optional[str] = None, allow_music: bool = False, allow_voice: bool = False, + video_model_family: str = "veo", ) -> Project: row = await crud.create_project( name=name, @@ -217,6 +220,7 @@ async def create_project( material=material, allow_music=allow_music, allow_voice=allow_voice, + video_model_family=video_model_family, ) return self._row_to_project(row) diff --git a/agent/sdk/services/assistant_provider.py b/agent/sdk/services/assistant_provider.py index b8096d29d..199eb7ff3 100644 --- a/agent/sdk/services/assistant_provider.py +++ b/agent/sdk/services/assistant_provider.py @@ -44,6 +44,7 @@ from agent import config from agent.db import crud +from agent.utils.paths import file_url_to_path from agent.sdk.services.provider_base import ( KIND_AUDIO, KIND_EDIT_IMAGE, @@ -92,8 +93,8 @@ def _video_result(mid: str, url: str) -> dict: def _audio_result(url: str) -> dict: """TTS result: the finished audio URL plus a local path when file://.""" - parsed = urlparse(url or "") - local_path = str(Path(parsed.path).resolve()) if parsed.scheme == "file" else "" + local = file_url_to_path(url or "") + local_path = str(local.resolve()) if local is not None else "" return {"data": {"url": url, "audio_path": local_path or url}} @@ -204,8 +205,9 @@ async def register_existing_image( self, url: str, *, name: str = "", project_id: str = "" ) -> dict: """Mint a UUID media_id for an already-existing local image.""" - parsed = urlparse(url) - local = Path(parsed.path) if parsed.scheme == "file" else Path(url) + local = file_url_to_path(url) + if local is None: + local = Path(url) if not local.is_file(): return {"error": f"AssistantProvider: local image not found for '{name}': {url}"} mid = str(uuid.uuid4()) diff --git a/agent/sdk/services/flow_provider.py b/agent/sdk/services/flow_provider.py index 6a70e7f98..60cc25db0 100644 --- a/agent/sdk/services/flow_provider.py +++ b/agent/sdk/services/flow_provider.py @@ -18,6 +18,26 @@ from agent import config from agent.config import VIDEO_POLL_INTERVAL, VIDEO_POLL_TIMEOUT from agent.db import crud +from agent.services.omni_flash import ( + OMNI_FLASH_VALID_DURATIONS, + generate_omni_flash_first_frame_video, + generate_omni_flash_first_last_video, +) + + +def _omni_duration(requested) -> int: + """Round a requested clip length up to the next Omni Flash step (4/6/8/10s). + + Rounding *down* would clip narration; anything past the max is capped. + None/0 falls back to OMNI_FLASH_DURATION_S. + """ + if not requested: + return int(config.OMNI_FLASH_DURATION_S) + want = float(requested) + for step in OMNI_FLASH_VALID_DURATIONS: + if step >= want - 1e-6: + return step + return OMNI_FLASH_VALID_DURATIONS[-1] from agent.sdk.services.provider_base import ( KIND_EDIT_IMAGE, KIND_IMAGE, @@ -282,8 +302,8 @@ class FlowProvider(MediaProvider): generate_video_i2v=True, generate_video_r2v=True, upscale=True, - max_concurrent=5, - cooldown_s=10.0, + max_concurrent=config.FLOW_MAX_CONCURRENT, + cooldown_s=config.FLOW_COOLDOWN_S, ) def __init__(self, client) -> None: @@ -377,15 +397,36 @@ async def _run_video(self, job: ProviderJob) -> dict: operations = [{"operation": {"name": existing_op}, "status": "MEDIA_GENERATION_STATUS_PENDING"}] return await _poll_operations(self._client, operations) - submit_result = await self._client.generate_video( - start_image_media_id=ex.get("start_media_id", ""), - prompt=job.prompt, - project_id=ex.get("project_id", "0"), - scene_id=ex.get("scene_id", ""), - aspect_ratio=ex.get("aspect", "VIDEO_ASPECT_RATIO_PORTRAIT"), - end_image_media_id=ex.get("end_media_id"), - user_paygate_tier=ex.get("tier", "PAYGATE_TIER_TWO"), - ) + # The project's video_model_family picks the Flow family. Omni Flash + # returns the same batch-operation shape as Veo, so polling and the + # retry-repoll guard above are shared. + if ex.get("model_family") == "omni_flash": + common = dict( + start_image_media_id=ex.get("start_media_id", ""), + prompt=job.prompt, + project_id=ex.get("project_id", "0"), + scene_id=ex.get("scene_id", ""), + duration_s=_omni_duration(ex.get("duration_s")), + resolution=ex.get("resolution") or config.OMNI_FLASH_RESOLUTION, + aspect_ratio=ex.get("aspect", "VIDEO_ASPECT_RATIO_PORTRAIT"), + user_paygate_tier=ex.get("tier", "PAYGATE_TIER_TWO"), + ) + end_id = ex.get("end_media_id") + if end_id: + submit_result = await generate_omni_flash_first_last_video( + end_image_media_id=end_id, **common) + else: + submit_result = await generate_omni_flash_first_frame_video(**common) + else: + submit_result = await self._client.generate_video( + start_image_media_id=ex.get("start_media_id", ""), + prompt=job.prompt, + project_id=ex.get("project_id", "0"), + scene_id=ex.get("scene_id", ""), + aspect_ratio=ex.get("aspect", "VIDEO_ASPECT_RATIO_PORTRAIT"), + end_image_media_id=ex.get("end_media_id"), + user_paygate_tier=ex.get("tier", "PAYGATE_TIER_TWO"), + ) if _is_error(submit_result): logger.error("[DEBUG] Video gen submit_result IS_ERROR: %s", str(submit_result)[:2000]) diff --git a/agent/sdk/services/operations.py b/agent/sdk/services/operations.py index 57c7a1000..435b337ca 100644 --- a/agent/sdk/services/operations.py +++ b/agent/sdk/services/operations.py @@ -275,6 +275,7 @@ async def generate_scene_video(self, scene: dict, orientation: str, tier = project.get("user_paygate_tier", "PAYGATE_TIER_TWO") if project else "PAYGATE_TIER_TWO" pid = scene.get("_project_id", "0") end_id = scene.get(f"{prefix}_end_scene_media_id") + model_family = (project.get("video_model_family") if project else None) or "veo" # Chain scenes with end_image: prefer transition_prompt (describes motion between frames) if end_id and scene.get("transition_prompt"): @@ -298,6 +299,11 @@ async def generate_scene_video(self, scene: dict, orientation: str, "start_media_id": image_media_id, "end_media_id": end_id, "request_id": request_id, + "model_family": model_family, + # Per-scene clip length (seconds). Omni Flash honours it (rounded + # up to 4/6/8/10); Veo ignores it. Lets a scene whose narration + # outruns the default be rendered longer instead of padded in post. + "duration_s": scene.get("duration"), "project_name": project.get("name") if project else "", "display_order": scene.get("display_order", 0), }, diff --git a/agent/services/tts.py b/agent/services/tts.py index 14a729856..5d8fcbea3 100644 --- a/agent/services/tts.py +++ b/agent/services/tts.py @@ -7,21 +7,48 @@ from pathlib import Path from typing import Optional -from agent.config import TTS_MODEL, TTS_SAMPLE_RATE +from agent.config import TTS_DEVICE, TTS_MODEL, TTS_SAMPLE_RATE 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") +# Shared by both inline scripts. OmniVoice.generate() returns a list of 1-D +# numpy arrays; torchaudio.save wants a 2-D (channels, samples) tensor, so +# the raw item cannot be handed over as-is ("Expected 2D Tensor, got 1D"). +# +# OmniVoice also stops at the very last voiced sample: no release tail, so the +# final vowel is clipped and a cut placed at wav end lands mid-sound. Fade the +# last TAIL_FADE_S down and append TAIL_PAD_S of silence so narration ends +# cleanly and downstream "trim to narrator length" leaves a natural breath. +TAIL_FADE_S = 0.06 +TAIL_PAD_S = 0.4 +_TO_WAV_TENSOR = f""" +TAIL_FADE_S = {TAIL_FADE_S} +TAIL_PAD_S = {TAIL_PAD_S} +def _to_wav_tensor(item, sample_rate): + w = item if isinstance(item, torch.Tensor) else torch.as_tensor(item) + w = w.detach().cpu().float() + w = w.unsqueeze(0) if w.ndim == 1 else w + n = w.shape[1] + fade = min(int(TAIL_FADE_S * sample_rate), n) + if fade > 0: + w = w.clone() + w[:, n - fade:] *= torch.linspace(1.0, 0.0, fade) + pad = torch.zeros(w.shape[0], int(TAIL_PAD_S * sample_rate)) + return torch.cat([w, pad], dim=1) +""" + # Inline script template for TTS generation via subprocess _TTS_SCRIPT = """ import sys, json, torch, torchaudio args = json.loads(sys.argv[1]) from omnivoice import OmniVoice - -model = OmniVoice.from_pretrained(args["model"], device_map="cpu", dtype=torch.float32) +""" + _TO_WAV_TENSOR + """ +model = OmniVoice.from_pretrained( + args["model"], device_map=args.get("device", "cpu"), dtype=torch.float32) kwargs = {"text": args["text"]} if args.get("ref_audio") and args.get("ref_text"): @@ -33,7 +60,7 @@ kwargs["speed"] = args["speed"] audio = model.generate(**kwargs) -torchaudio.save(args["output"], audio[0], args["sample_rate"]) +torchaudio.save(args["output"], _to_wav_tensor(audio[0], args["sample_rate"]), args["sample_rate"]) print(json.dumps({"ok": True, "path": args["output"]})) """ @@ -44,8 +71,9 @@ args = json.loads(sys.argv[1]) from omnivoice import OmniVoice - -model = OmniVoice.from_pretrained(args["model"], device_map="cpu", dtype=torch.float32) +""" + _TO_WAV_TENSOR + """ +model = OmniVoice.from_pretrained( + args["model"], device_map=args.get("device", "cpu"), dtype=torch.float32) results = [] for item in args["items"]: @@ -61,7 +89,7 @@ audio = model.generate(**kwargs) Path(item["output"]).parent.mkdir(parents=True, exist_ok=True) - torchaudio.save(item["output"], audio[0], args["sample_rate"]) + torchaudio.save(item["output"], _to_wav_tensor(audio[0], args["sample_rate"]), args["sample_rate"]) info = torchaudio.info(item["output"]) duration = info.num_frames / info.sample_rate @@ -90,6 +118,7 @@ async def generate_speech( "output": output_path, "sample_rate": TTS_SAMPLE_RATE, "speed": speed, + "device": TTS_DEVICE, } if instruct: args["instruct"] = instruct @@ -166,6 +195,7 @@ async def generate_video_narration( "sample_rate": TTS_SAMPLE_RATE, "speed": speed, "items": items, + "device": TTS_DEVICE, } if instruct: args["instruct"] = instruct diff --git a/agent/services/video_reviewer.py b/agent/services/video_reviewer.py index 0fe22d213..0318c2ee7 100644 --- a/agent/services/video_reviewer.py +++ b/agent/services/video_reviewer.py @@ -38,6 +38,7 @@ PROVIDER_BINARIES, resolve_role, ) +from agent.utils.paths import file_url_to_path logger = logging.getLogger(__name__) @@ -149,11 +150,12 @@ def _local_media_path(url: str) -> Path | None: """ if not url: return None - parsed = urlparse(url) - if parsed.scheme == "file": - p = Path(parsed.path) + p = file_url_to_path(url) + if p is not None: return p if p.is_file() else None - if not parsed.scheme: + parsed = urlparse(url) + # A Windows drive letter ("C:") parses as a scheme; it is still a path. + if not parsed.scheme or len(parsed.scheme) == 1: p = Path(url) return p if p.is_file() else None return None diff --git a/agent/utils/paths.py b/agent/utils/paths.py index fae79d7bc..a41909ae6 100644 --- a/agent/utils/paths.py +++ b/agent/utils/paths.py @@ -1,9 +1,30 @@ """Centralized path resolver for project output directories and scene files.""" from pathlib import Path +from urllib.parse import urlparse +from urllib.request import url2pathname from agent.config import OUTPUT_DIR +def file_url_to_path(url: str | None) -> "Path | None": + """Turn a ``file://`` URL into a local ``Path``; ``None`` for anything else. + + Accepts both the RFC 8089 form ``Path.as_uri()`` produces + (``file:///C:/x/y.png``, percent-encoded) and the naive + ``"file://" + str(path)`` form the pipeline stores. The latter is fine on + POSIX (``file:///tmp/y.png``) but on Windows ``urlparse`` files + ``C:\\x\\y.png`` under *netloc* with an empty *path*, so reading only + ``parsed.path`` made every local clip look missing. + """ + if not url: + return None + parsed = urlparse(url) + if parsed.scheme != "file": + return None + raw = (parsed.netloc or "") + parsed.path + return Path(url2pathname(raw)) + + def project_dir(project_slug: str) -> Path: """Return the output directory for a given project slug.""" return OUTPUT_DIR / project_slug diff --git a/tests/unit/test_cli_providers.py b/tests/unit/test_cli_providers.py index a6f02f1e4..a1e848989 100644 --- a/tests/unit/test_cli_providers.py +++ b/tests/unit/test_cli_providers.py @@ -415,7 +415,8 @@ async def test_claude_agy_providers_include_read_the_images_at(self, provider, r assert "BASE_PROMPT" in captured["prompt"] # The sheets live outside the server cwd, so the directory holding them # is named to the CLI rather than left to be discovered. - assert captured["add_dirs"] == ("/tmp",) + # str(Path) so the expectation is spelled the way the OS spells it. + assert captured["add_dirs"] == (str(Path("/tmp")),) assert result == {"dimensions": {}, "errors": [], "usable_segments": []} @pytest.mark.asyncio @@ -473,7 +474,7 @@ async def test_single_sheet_wrapper_wording_matches_original_singular_form(self) new=AsyncMock(side_effect=capture_runner(captured))): await _analyze_cli(contact_sheets, 9, 4.0, {}, timestamped=True) - assert captured["prompt"].startswith("Read the image at /tmp/sheet_00.jpg.") + assert captured["prompt"].startswith(f"Read the image at {contact_sheets[0]}.") assert "Read the images at:" not in captured["prompt"] assert "sequential contact sheets" not in captured["prompt"] assert "It is a contact sheet of 9 video frames at 4.0fps with timestamps." in captured["prompt"] diff --git a/tests/unit/test_file_url_paths.py b/tests/unit/test_file_url_paths.py new file mode 100644 index 000000000..0a5d446f1 --- /dev/null +++ b/tests/unit/test_file_url_paths.py @@ -0,0 +1,48 @@ +"""file:// URL <-> local path conversion must work on every OS. + +The pipeline stores local media as ``"file://" + str(path)``. On POSIX that +yields ``file:///tmp/x.mp4`` which ``urlparse().path`` reads back fine. On +Windows it yields ``file://C:\\Temp\\x.mp4`` — ``urlparse`` puts ``C:\\Temp\\x.mp4`` +in *netloc* and leaves ``path`` empty, so every ``Path(parsed.path)`` caller +concluded the file did not exist and fell through to an HTTP download. +""" +import sys +from pathlib import Path + +import pytest + +from agent.utils.paths import file_url_to_path + + +def test_naive_file_url_from_str_path_roundtrips(tmp_path): + f = tmp_path / "clip.mp4" + f.write_bytes(b"x") + assert file_url_to_path(f"file://{f}") == f + + +def test_rfc8089_file_url_from_as_uri_roundtrips(tmp_path): + f = tmp_path / "x y.png" # space -> %20 in as_uri() + f.write_bytes(b"x") + assert file_url_to_path(f.as_uri()) == f + + +def test_posix_style_file_url_is_accepted(): + p = file_url_to_path("file:///tmp/out.png") + assert p == Path("/tmp/out.png") + + +def test_non_file_urls_and_empty_return_none(): + assert file_url_to_path("https://example.com/a.mp4") is None + assert file_url_to_path("") is None + assert file_url_to_path(None) is None + + +def test_plain_path_is_not_a_file_url(tmp_path): + # Callers that accept bare paths handle that branch themselves. + assert file_url_to_path(str(tmp_path / "a.mp4")) is None + + +@pytest.mark.skipif(sys.platform != "win32", reason="Windows drive-letter form") +def test_windows_backslash_form_keeps_drive_letter(): + p = file_url_to_path(r"file://C:\Windows\Temp\x.png") + assert p == Path(r"C:\Windows\Temp\x.png") diff --git a/tests/unit/test_tts_scripts.py b/tests/unit/test_tts_scripts.py new file mode 100644 index 000000000..fac639c3a --- /dev/null +++ b/tests/unit/test_tts_scripts.py @@ -0,0 +1,133 @@ +"""The inline TTS scripts must save what OmniVoice.generate() actually returns. + +OmniVoice >= 0.2 returns ``list[numpy.ndarray]`` (1-D float32 per item). +``torchaudio.save`` wants a 2-D ``(channels, samples)`` tensor, so handing it +``audio[0]`` raw fails with ``Expected 2D Tensor, got 1D`` — every TTS call +died at the very last step, after the model had generated fine. + +These tests exec the script templates against a fake ``omnivoice`` module so +they run without torch/omnivoice installed in the agent venv. +""" +import json +import sys +import types + +import pytest + +np = pytest.importorskip("numpy") + +from agent.services import tts as tts_mod # noqa: E402 + + +class _FakeTorchaudio: + saved = [] + + class info: # noqa: N801 — mimics torchaudio.info(path) + def __init__(self, path): + self.num_frames = 24000 + self.sample_rate = 24000 + + @classmethod + def save(cls, path, tensor, sample_rate): + if getattr(tensor, "ndim", None) != 2: + raise ValueError(f"Expected 2D Tensor, got {tensor.ndim}D.") + cls.saved.append((path, tuple(tensor.shape), sample_rate)) + + +class _FakeOmniVoice: + kwargs_seen = {} + + @classmethod + def from_pretrained(cls, model, **kwargs): + cls.kwargs_seen = kwargs + return cls() + + def generate(self, **kwargs): + return [np.zeros(2400, dtype=np.float32)] + + +@pytest.fixture +def fake_modules(monkeypatch): + torch = pytest.importorskip("torch") + _FakeTorchaudio.saved = [] + monkeypatch.setitem(sys.modules, "torchaudio", _FakeTorchaudio) + ov = types.ModuleType("omnivoice") + ov.OmniVoice = _FakeOmniVoice + monkeypatch.setitem(sys.modules, "omnivoice", ov) + return torch + + +def _run(script: str, args: dict, capsys): + sys.argv = ["tts", json.dumps(args)] + exec(compile(script, "", "exec"), {"__name__": "__main__"}) + return json.loads(capsys.readouterr().out.strip().splitlines()[-1]) + + +def test_single_script_saves_1d_numpy_output_as_2d(fake_modules, tmp_path, capsys): + out = tmp_path / "a.wav" + result = _run(tts_mod._TTS_SCRIPT, { + "model": "m", "text": "xin chào", "output": str(out), "sample_rate": 24000, + }, capsys) + assert result["ok"] is True + assert _FakeTorchaudio.saved == [(str(out), (1, 2400 + int(tts_mod.TAIL_PAD_S * 24000)), 24000)] + + +def test_batch_script_saves_1d_numpy_output_as_2d(fake_modules, tmp_path, capsys): + out = tmp_path / "s" / "b.wav" + results = _run(tts_mod._TTS_BATCH_SCRIPT, { + "model": "m", "sample_rate": 24000, + "items": [{"id": "s1", "text": "xin chào", "output": str(out)}], + }, capsys) + assert results == [{"id": "s1", "ok": True, "path": str(out), "duration": 1.0}] + assert _FakeTorchaudio.saved == [(str(out), (1, 2400 + int(tts_mod.TAIL_PAD_S * 24000)), 24000)] + + +def test_scripts_honour_device_arg(fake_modules, tmp_path, capsys): + _run(tts_mod._TTS_SCRIPT, { + "model": "m", "text": "x", "output": str(tmp_path / "a.wav"), + "sample_rate": 24000, "device": "cuda", + }, capsys) + assert _FakeOmniVoice.kwargs_seen["device_map"] == "cuda" + + +def test_generate_speech_passes_configured_device(monkeypatch, tmp_path): + captured = {} + monkeypatch.setattr(tts_mod, "TTS_DEVICE", "cuda") + monkeypatch.setattr(tts_mod, "_run_tts_subprocess", + lambda args: captured.update(args) or {"ok": True}) + import asyncio + asyncio.run(tts_mod.generate_speech("hi", str(tmp_path / "o.wav"))) + assert captured["device"] == "cuda" + + +class _FakeOmniVoiceHot(_FakeOmniVoice): + """Returns audio that is still loud at the very last sample — OmniVoice + stops exactly where the voice stops, so the final vowel gets clipped.""" + + def generate(self, **kwargs): + return [np.full(2400, 0.5, dtype=np.float32)] + + +def test_saved_wav_has_silent_tail_and_fade(fake_modules, tmp_path, capsys, monkeypatch): + torch = fake_modules + captured = {} + + def save(path, tensor, sample_rate): + captured["t"] = tensor + captured["sr"] = sample_rate + monkeypatch.setattr(_FakeTorchaudio, "save", staticmethod(save)) + ov = types.ModuleType("omnivoice"); ov.OmniVoice = _FakeOmniVoiceHot + monkeypatch.setitem(sys.modules, "omnivoice", ov) + + _run(tts_mod._TTS_SCRIPT, {"model": "m", "text": "hi", "output": str(tmp_path / "o.wav"), + "sample_rate": 24000, "ref_audio": None, "ref_text": None, + "instruct": None, "language": "vi"}, capsys) + t = captured["t"] + sr = captured["sr"] + assert t.shape[0] == 1 + assert t.shape[1] >= 2400 + int(0.3 * sr), "no silence appended after the voice" + assert float(t[0, -1].abs()) < 1e-6, "tail is not silent" + # the original last voiced sample must have been faded down, not hard-cut + assert float(t[0, 2399].abs()) < 0.05, "no fade-out before the pad" + # voice body (before the 60ms fade window) untouched + assert abs(float(t[0, 500]) - 0.5) < 1e-6 diff --git a/tests/unit/test_video_model_family.py b/tests/unit/test_video_model_family.py new file mode 100644 index 000000000..01f83f5b7 --- /dev/null +++ b/tests/unit/test_video_model_family.py @@ -0,0 +1,157 @@ +"""The worker must honour a project's video_model_family. + +Before this, the queue path (GENERATE_VIDEO -> FlowProvider._run_video) was +hardwired to Veo; Omni Flash existed only on the ad-hoc /api/flow/generate-video +endpoint, which does not write results back to the scene. A project that +chooses Omni Flash (720p, cheaper, and the only family with r2v / first+last +on the batch transport) must get it through the normal pipeline so review, +concat and retry-repoll keep working unchanged. +""" +from unittest.mock import AsyncMock, patch + +import pytest + +from agent.sdk.services.flow_provider import FlowProvider +from agent.sdk.services.provider_base import KIND_VIDEO, ProviderJob + + +def _job(**extra): + base = {"project_id": "p1", "tier": "PAYGATE_TIER_TWO", "aspect": "VIDEO_ASPECT_RATIO_LANDSCAPE", + "scene_id": "s1", "start_media_id": "img-1", "end_media_id": None, "request_id": ""} + base.update(extra) + return ProviderJob(job_id="j", kind=KIND_VIDEO, prompt="walk", orientation="HORIZONTAL", extra=base) + + +@pytest.mark.asyncio +async def test_omni_flash_family_routes_to_omni_first_frame(): + client = AsyncMock() + client.generate_video = AsyncMock() + submitted = {"data": {"operations": [{"operation": {"name": "op-omni"}, + "status": "MEDIA_GENERATION_STATUS_PENDING"}]}} + with patch("agent.sdk.services.flow_provider.generate_omni_flash_first_frame_video", + new=AsyncMock(return_value=submitted)) as omni, \ + patch("agent.sdk.services.flow_provider._poll_operations", + new=AsyncMock(return_value={"data": {"ok": 1}})) as poll: + result = await FlowProvider(client)._run_video(_job(model_family="omni_flash")) + + client.generate_video.assert_not_called() + kw = omni.await_args.kwargs + assert kw["start_image_media_id"] == "img-1" + assert kw["aspect_ratio"] == "VIDEO_ASPECT_RATIO_LANDSCAPE" + assert kw["duration_s"] == 8 and kw["resolution"] == "720p" + assert poll.await_args.args[1][0]["operation"]["name"] == "op-omni" + assert result == {"data": {"ok": 1}} + + +@pytest.mark.asyncio +async def test_omni_flash_with_end_frame_uses_first_last(): + client = AsyncMock() + submitted = {"data": {"operations": [{"operation": {"name": "op-fl"}, + "status": "MEDIA_GENERATION_STATUS_PENDING"}]}} + with patch("agent.sdk.services.flow_provider.generate_omni_flash_first_last_video", + new=AsyncMock(return_value=submitted)) as fl, \ + patch("agent.sdk.services.flow_provider._poll_operations", + new=AsyncMock(return_value={"data": {}})): + await FlowProvider(client)._run_video(_job(model_family="omni_flash", end_media_id="img-2")) + assert fl.await_args.kwargs["end_image_media_id"] == "img-2" + + +@pytest.mark.asyncio +async def test_default_family_is_still_veo(): + client = AsyncMock() + client.generate_video = AsyncMock( + return_value={"data": {"operations": [{"operation": {"name": "op-veo"}}]}}) + with patch("agent.sdk.services.flow_provider._poll_operations", + new=AsyncMock(return_value={"data": {}})): + await FlowProvider(client)._run_video(_job()) + client.generate_video.assert_called_once() + + +@pytest.mark.asyncio +async def test_generate_scene_video_passes_project_family(): + """operations.generate_scene_video reads the project's family into job.extra.""" + from agent.sdk.services.operations import OperationService + + captured = {} + + async def fake_run(self, job, provider=None): + captured.update(job.extra) + return {"data": {}} + + scene = {"id": "s1", "_project_id": "p1", "horizontal_image_media_id": "img-1", + "horizontal_image_url": "file:///x.png", "video_prompt": "walk", "display_order": 1} + with patch("agent.sdk.services.operations.crud") as crud, \ + patch("agent.sdk.services.operations._build_video_prompt", new=AsyncMock(return_value="p")), \ + patch.object(OperationService, "_run_provider_job", fake_run), \ + patch.object(OperationService, "_resolve_chain_end_url", new=AsyncMock(return_value=None)): + crud.get_project = AsyncMock(return_value={"user_paygate_tier": "PAYGATE_TIER_TWO", + "video_model_family": "omni_flash"}) + await OperationService.__new__(OperationService).generate_scene_video(scene, "HORIZONTAL") + assert captured["model_family"] == "omni_flash" + + +@pytest.mark.asyncio +async def test_project_update_persists_video_model_family(test_db): + """crud._update whitelists columns; the new one must be allowed or PATCH is a silent no-op.""" + from agent.db import crud + + p = await crud.create_project("x") + assert p["video_model_family"] == "veo" + row = await crud._update("project", "id", p["id"], video_model_family="omni_flash") + assert row["video_model_family"] == "omni_flash" + + +def test_flow_provider_rate_limits_come_from_env(monkeypatch): + """A session that fires generates every 10s trips Flow's UNUSUAL_ACTIVITY guard; + operators must be able to slow the worker without editing code.""" + import importlib + from agent import config + monkeypatch.setenv("FLOW_MAX_CONCURRENT", "1") + monkeypatch.setenv("FLOW_COOLDOWN_S", "45") + importlib.reload(config) + import agent.sdk.services.flow_provider as fp + importlib.reload(fp) + try: + assert fp.FlowProvider.capabilities.max_concurrent == 1 + assert fp.FlowProvider.capabilities.cooldown_s == 45.0 + finally: + monkeypatch.delenv("FLOW_MAX_CONCURRENT"); monkeypatch.delenv("FLOW_COOLDOWN_S") + importlib.reload(config); importlib.reload(fp) + + +@pytest.mark.asyncio +async def test_generate_scene_video_passes_scene_duration_for_omni(): + """scene.duration (seconds) must reach the Omni Flash call so a scene whose + narration outruns the 8s default can be rendered at 10s instead of being + padded with a frozen last frame in post.""" + from agent.sdk.services.operations import OperationService + + captured = {} + + async def fake_run(self, job, provider=None): + captured.update(job.extra) + return {"data": {}} + + scene = {"id": "s1", "_project_id": "p1", "horizontal_image_media_id": "img-1", + "horizontal_image_url": "file:///x.png", "video_prompt": "walk", "display_order": 1, + "duration": 10} + with patch("agent.sdk.services.operations.crud") as crud, \ + patch("agent.sdk.services.operations._build_video_prompt", new=AsyncMock(return_value="p")), \ + patch.object(OperationService, "_run_provider_job", fake_run), \ + patch.object(OperationService, "_resolve_chain_end_url", new=AsyncMock(return_value=None)): + crud.get_project = AsyncMock(return_value={"user_paygate_tier": "PAYGATE_TIER_TWO", + "video_model_family": "omni_flash"}) + await OperationService.__new__(OperationService).generate_scene_video(scene, "HORIZONTAL") + assert captured["duration_s"] == 10 + + +@pytest.mark.asyncio +async def test_omni_flash_rounds_duration_up_to_supported_step(): + """Omni Flash only accepts 4/6/8/10s; a 8.8s request must become 10, never 8.""" + client = AsyncMock() + submitted = {"data": {"operations": [{"operation": {"name": "op"}, "status": "MEDIA_GENERATION_STATUS_PENDING"}]}} + with patch("agent.sdk.services.flow_provider.generate_omni_flash_first_frame_video", + new=AsyncMock(return_value=submitted)) as omni, \ + patch("agent.sdk.services.flow_provider._poll_operations", new=AsyncMock(return_value={"data": {}})): + await FlowProvider(client)._run_video(_job(model_family="omni_flash", duration_s=8.8)) + assert omni.await_args.kwargs["duration_s"] == 10