From d59257edf154ebba372745878bee220a9c77a017 Mon Sep 17 00:00:00 2001
From: Khanh <khanh@local>
Date: Tue, 29 Sep 2026 19:44:19 +0700
Subject: [PATCH 1/6] fix: resolve file:// URLs correctly on Windows

The pipeline stores local media as "file://" + str(path). On Windows that
is file://C:\...\x.mp4, which urlparse splits into netloc="C:\..." and an
empty path, so every Path(parsed.path) caller decided the clip was missing
and fell through to an HTTP download that then failed.

Add agent.utils.paths.file_url_to_path (netloc+path via url2pathname, so
both as_uri() and the naive form work) and use it in video_reviewer,
api/videos, api/projects and assistant_provider. Make the two
test_cli_providers assertions spell /tmp the way the host OS does.

Windows: 12 failed -> 0 failed (445 passed).
---
 agent/api/projects.py                    |  5 ++-
 agent/api/videos.py                      |  7 ++--
 agent/sdk/services/assistant_provider.py | 10 +++--
 agent/services/video_reviewer.py         | 10 +++--
 agent/utils/paths.py                     | 21 +++++++++++
 tests/unit/test_cli_providers.py         |  5 ++-
 tests/unit/test_file_url_paths.py        | 48 ++++++++++++++++++++++++
 7 files changed, 91 insertions(+), 15 deletions(-)
 create mode 100644 tests/unit/test_file_url_paths.py

diff --git a/agent/api/projects.py b/agent/api/projects.py
index 9a6f33c36..f1000c823 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
@@ -467,8 +468,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/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/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")

From 5e5dec2a2b27b719b72541f4cb712f57e5ccbeda Mon Sep 17 00:00:00 2001
From: Khanh <khanh@local>
Date: Tue, 29 Sep 2026 21:11:49 +0700
Subject: [PATCH 2/6] fix(tts): save OmniVoice output as 2-D tensor, honour
 TTS_DEVICE
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

OmniVoice.generate() returns list[numpy.ndarray] (1-D). torchaudio.save
needs (channels, samples), so every TTS call failed at the last step with
'Expected 2D Tensor, got 1D' after the model had generated fine. Wrap the
item via a shared _to_wav_tensor helper in both inline scripts.

Also pass config.TTS_DEVICE through to device_map instead of hardcoding
cpu — TTS_DEVICE already existed in config but was never used. On an RTX
3060 a template takes ~23s including model load vs minutes on CPU.
---
 agent/services/tts.py          |  28 ++++++---
 tests/unit/test_tts_scripts.py | 100 +++++++++++++++++++++++++++++++++
 2 files changed, 121 insertions(+), 7 deletions(-)
 create mode 100644 tests/unit/test_tts_scripts.py

diff --git a/agent/services/tts.py b/agent/services/tts.py
index 14a729856..23b3bf574 100644
--- a/agent/services/tts.py
+++ b/agent/services/tts.py
@@ -7,21 +7,32 @@
 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").
+_TO_WAV_TENSOR = """
+def _to_wav_tensor(item):
+    w = item if isinstance(item, torch.Tensor) else torch.as_tensor(item)
+    w = w.detach().cpu().float()
+    return w.unsqueeze(0) if w.ndim == 1 else w
+"""
+
 # 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 +44,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"])
 print(json.dumps({"ok": True, "path": args["output"]}))
 """
 
@@ -44,8 +55,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 +73,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"])
 
         info = torchaudio.info(item["output"])
         duration = info.num_frames / info.sample_rate
@@ -90,6 +102,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 +179,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/tests/unit/test_tts_scripts.py b/tests/unit/test_tts_scripts.py
new file mode 100644
index 000000000..f378d37ec
--- /dev/null
+++ b/tests/unit/test_tts_scripts.py
@@ -0,0 +1,100 @@
+"""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, "<tts-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), 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), 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"

From 328df2a6b7c548b855cf3c8de7f4057af387d7f6 Mon Sep 17 00:00:00 2001
From: Khanh <khanh@local>
Date: Wed, 30 Sep 2026 12:55:34 +0700
Subject: [PATCH 3/6] feat: per-project video_model_family so the worker can
 use Omni Flash

The queue path (GENERATE_VIDEO -> FlowProvider._run_video) was hardwired
to Veo. Omni Flash was reachable only through the ad-hoc
/api/flow/generate-video endpoint, which does not persist results to the
scene, so choosing Omni meant leaving the pipeline (no review, no concat,
no retry-repoll).

Add project.video_model_family ('veo' | 'omni_flash', default veo, with
schema migration), thread it through ProjectCreate/ProjectUpdate, the SDK
model/repository, and into job.extra. _run_video dispatches on it:
omni_flash -> generate_omni_flash_first_frame_video / _first_last_video,
which return the same batch-operation shape, so polling and the
already-submitted guard are shared. Defaults come from
OMNI_FLASH_DURATION_S / OMNI_FLASH_RESOLUTION.
---
 agent/api/projects.py                      |   1 +
 agent/config.py                            |   4 +
 agent/db/crud.py                           |   8 +-
 agent/db/schema.py                         |   4 +
 agent/models/project.py                    |  10 +-
 agent/sdk/models/project.py                |   1 +
 agent/sdk/persistence/sqlite_repository.py |   4 +
 agent/sdk/services/flow_provider.py        |  43 +++++++--
 agent/sdk/services/operations.py           |   2 +
 tests/unit/test_video_model_family.py      | 101 +++++++++++++++++++++
 10 files changed, 164 insertions(+), 14 deletions(-)
 create mode 100644 tests/unit/test_video_model_family.py

diff --git a/agent/api/projects.py b/agent/api/projects.py
index f1000c823..69fe112b6 100644
--- a/agent/api/projects.py
+++ b/agent/api/projects.py
@@ -213,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
diff --git a/agent/config.py b/agent/config.py
index 9c72cfc25..d5875ddd3 100644
--- a/agent/config.py
+++ b/agent/config.py
@@ -76,6 +76,10 @@
 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")
 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/flow_provider.py b/agent/sdk/services/flow_provider.py
index 6a70e7f98..09318fad4 100644
--- a/agent/sdk/services/flow_provider.py
+++ b/agent/sdk/services/flow_provider.py
@@ -18,6 +18,10 @@
 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 (
+    generate_omni_flash_first_frame_video,
+    generate_omni_flash_first_last_video,
+)
 from agent.sdk.services.provider_base import (
     KIND_EDIT_IMAGE,
     KIND_IMAGE,
@@ -377,15 +381,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=int(ex.get("duration_s") or config.OMNI_FLASH_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..eb20e10f5 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,7 @@ 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,
                 "project_name": project.get("name") if project else "",
                 "display_order": scene.get("display_order", 0),
             },
diff --git a/tests/unit/test_video_model_family.py b/tests/unit/test_video_model_family.py
new file mode 100644
index 000000000..9d63a991c
--- /dev/null
+++ b/tests/unit/test_video_model_family.py
@@ -0,0 +1,101 @@
+"""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"

From 0b790bcb7b3ab56816dd64a43543c6ead0e99a37 Mon Sep 17 00:00:00 2001
From: Khanh <khanh@local>
Date: Wed, 30 Sep 2026 14:32:46 +0700
Subject: [PATCH 4/6] feat: FLOW_MAX_CONCURRENT / FLOW_COOLDOWN_S env for
 provider pacing

Each generate mints a reCAPTCHA in the Flow tab. A worker firing ~30 of
them 10s apart from a background tab degrades the session score until
Flow returns PUBLIC_ERROR_UNUSUAL_ACTIVITY on every call, while the same
account still generates fine by hand. Expose the provider's max_concurrent
and cooldown_s via env so operators can slow down without editing code;
defaults unchanged.
---
 agent/config.py                       |  5 +++++
 agent/sdk/services/flow_provider.py   |  4 ++--
 tests/unit/test_video_model_family.py | 18 ++++++++++++++++++
 3 files changed, 25 insertions(+), 2 deletions(-)

diff --git a/agent/config.py b/agent/config.py
index d5875ddd3..e020edc81 100644
--- a/agent/config.py
+++ b/agent/config.py
@@ -80,6 +80,11 @@
 # 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/sdk/services/flow_provider.py b/agent/sdk/services/flow_provider.py
index 09318fad4..243efae93 100644
--- a/agent/sdk/services/flow_provider.py
+++ b/agent/sdk/services/flow_provider.py
@@ -286,8 +286,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:
diff --git a/tests/unit/test_video_model_family.py b/tests/unit/test_video_model_family.py
index 9d63a991c..9c3942781 100644
--- a/tests/unit/test_video_model_family.py
+++ b/tests/unit/test_video_model_family.py
@@ -99,3 +99,21 @@ async def test_project_update_persists_video_model_family(test_db):
     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)

From cb1fd333ba08d3bff7bf71e238f4bde803f456f1 Mon Sep 17 00:00:00 2001
From: Khanh <khanh@local>
Date: Fri, 2 Oct 2026 12:06:53 +0700
Subject: [PATCH 5/6] fix(tts): fade and pad the OmniVoice tail so narration
 does not end mid-vowel

OmniVoice returns audio that stops at the last voiced sample: zero release
tail, final vowel clipped. Any pipeline that trims video to narrator length
then cuts exactly where the voice is still ringing, which reads as an abrupt
scene change. Fade the last 60 ms and append 0.4 s of silence inside the
save path so every wav ends cleanly and downstream trims get a breath.
---
 agent/services/tts.py          | 26 +++++++++++++++++++-----
 tests/unit/test_tts_scripts.py | 37 ++++++++++++++++++++++++++++++++--
 2 files changed, 56 insertions(+), 7 deletions(-)

diff --git a/agent/services/tts.py b/agent/services/tts.py
index 23b3bf574..5d8fcbea3 100644
--- a/agent/services/tts.py
+++ b/agent/services/tts.py
@@ -17,11 +17,27 @@
 # 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").
-_TO_WAV_TENSOR = """
-def _to_wav_tensor(item):
+#
+# 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()
-    return w.unsqueeze(0) if w.ndim == 1 else w
+    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
@@ -44,7 +60,7 @@ def _to_wav_tensor(item):
     kwargs["speed"] = args["speed"]
 
 audio = model.generate(**kwargs)
-torchaudio.save(args["output"], _to_wav_tensor(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"]}))
 """
 
@@ -73,7 +89,7 @@ def _to_wav_tensor(item):
 
         audio = model.generate(**kwargs)
         Path(item["output"]).parent.mkdir(parents=True, exist_ok=True)
-        torchaudio.save(item["output"], _to_wav_tensor(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
diff --git a/tests/unit/test_tts_scripts.py b/tests/unit/test_tts_scripts.py
index f378d37ec..fac639c3a 100644
--- a/tests/unit/test_tts_scripts.py
+++ b/tests/unit/test_tts_scripts.py
@@ -69,7 +69,7 @@ def test_single_script_saves_1d_numpy_output_as_2d(fake_modules, tmp_path, capsy
         "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), 24000)]
+    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):
@@ -79,7 +79,7 @@ def test_batch_script_saves_1d_numpy_output_as_2d(fake_modules, tmp_path, capsys
         "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), 24000)]
+    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):
@@ -98,3 +98,36 @@ def test_generate_speech_passes_configured_device(monkeypatch, tmp_path):
     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

From 2f2c5d84281f712452f661c6c25fba49b0f4f3b1 Mon Sep 17 00:00:00 2001
From: Khanh <khanh@local>
Date: Fri, 2 Oct 2026 12:28:59 +0700
Subject: [PATCH 6/6] feat: honour scene.duration for Omni Flash clip length
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A scene whose narration runs 7.6s cannot breathe inside an 8s clip; the
only post fix is a frozen last frame, which reads as a glitch. Pass
scene.duration into job.extra and round it UP to the next supported Omni
Flash step (4/6/8/10) in the provider — rounding down would clip the
voice. Veo ignores the field. Defaults unchanged.
---
 agent/sdk/services/flow_provider.py   | 18 ++++++++++++-
 agent/sdk/services/operations.py      |  4 +++
 tests/unit/test_video_model_family.py | 38 +++++++++++++++++++++++++++
 3 files changed, 59 insertions(+), 1 deletion(-)

diff --git a/agent/sdk/services/flow_provider.py b/agent/sdk/services/flow_provider.py
index 243efae93..60cc25db0 100644
--- a/agent/sdk/services/flow_provider.py
+++ b/agent/sdk/services/flow_provider.py
@@ -19,9 +19,25 @@
 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,
@@ -390,7 +406,7 @@ async def _run_video(self, job: ProviderJob) -> dict:
                 prompt=job.prompt,
                 project_id=ex.get("project_id", "0"),
                 scene_id=ex.get("scene_id", ""),
-                duration_s=int(ex.get("duration_s") or config.OMNI_FLASH_DURATION_S),
+                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"),
diff --git a/agent/sdk/services/operations.py b/agent/sdk/services/operations.py
index eb20e10f5..435b337ca 100644
--- a/agent/sdk/services/operations.py
+++ b/agent/sdk/services/operations.py
@@ -300,6 +300,10 @@ async def generate_scene_video(self, scene: dict, orientation: str,
                 "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/tests/unit/test_video_model_family.py b/tests/unit/test_video_model_family.py
index 9c3942781..01f83f5b7 100644
--- a/tests/unit/test_video_model_family.py
+++ b/tests/unit/test_video_model_family.py
@@ -117,3 +117,41 @@ def test_flow_provider_rate_limits_come_from_env(monkeypatch):
     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