fix: resolve file:// URLs correctly on Windows

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)
 

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:

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

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)
 

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()}

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

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

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)
 

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())

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])

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),
             },