feat(tools): add Google Lyria and direct Google Veo generation tools

tests/contracts/test_phase3_contracts.py

@@ -6,8 +6,11 @@ stage director skills, meta skills, and the animated-explainer pipeline.
 
 import sys
 import builtins
+import base64
+import os
 import shutil
 from pathlib import Path
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -32,10 +35,51 @@ from tools.audio.elevenlabs_tts import ElevenLabsTTS
 from tools.audio.openai_tts import OpenAITTS
 from tools.audio.piper_tts import PiperTTS
 from tools.audio.tts_selector import TTSSelector
+from tools.audio.google_tts import GoogleTTS
+from tools.graphics.google_imagen import GoogleImagen
+from tools.audio.google_music import GoogleMusic
+from tools.video.veo_video import VeoVideo
+
+
+# ---- Google Credentials ----
+
+
+class TestGoogleCredentials:
+    def test_get_genai_client_with_google_api_key(self):
+        from tools.google_credentials import get_genai_client
+        from google.genai import types
+
+        mock_client = MagicMock()
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "GOOGLE_API_KEY": "my_google_key",
+                    "GEMINI_API_KEY": "",
+                    "GOOGLE_GENAI_USE_VERTEXAI": "false",
+                },
+            ),
+            patch("google.genai.Client", return_value=mock_client) as mock_genai_client,
+        ):
+            # 1. Call with default options (None)
+            client = get_genai_client()
+            assert client is not None
+            kwargs = mock_genai_client.call_args[1]
+            assert kwargs["api_key"] == "my_google_key"
+            assert kwargs["http_options"] is None
+
+            # 2. Call with explicit options
+            my_opts = types.HttpOptions(timeout=12345)
+            client_custom = get_genai_client(http_options=my_opts)
+            assert client_custom is not None
+            kwargs_custom = mock_genai_client.call_args[1]
+            assert kwargs_custom["api_key"] == "my_google_key"
+            assert kwargs_custom["http_options"].timeout == 12345
 
 
 # ---- TTS Provider Tools ----
 
+
 class TestElevenLabsTTS:
     def test_identity(self):
         tool = ElevenLabsTTS()
@@ -92,6 +136,30 @@ class TestPiperTTS:
         assert PiperTTS().get_status() == ToolStatus.UNAVAILABLE
 
 
+class TestGoogleTTS:
+    def test_identity(self):
+        tool = GoogleTTS()
+        info = tool.get_info()
+        assert info["name"] == "google_tts"
+        assert info["tier"] == "voice"
+        assert info["capability"] == "tts"
+        assert info["provider"] == "google_tts"
+
+    def test_cost_estimate(self):
+        tool = GoogleTTS()
+        cost = tool.estimate_cost({"text": "Hello world, this is a test."})
+        assert cost > 0
+        assert cost < 0.01  # short text should be cheap
+
+    def test_capabilities(self):
+        tool = GoogleTTS()
+        assert "text_to_speech" in tool.capabilities
+        assert "voice_selection" in tool.capabilities
+
+
+# ---- Music Generation Tools ----
+
+
 class TestMusicGen:
     def test_identity(self):
         tool = MusicGen()
@@ -110,6 +178,451 @@ class TestMusicGen:
         assert "generate_background_music" in tool.capabilities
 
 
+class TestGoogleMusic:
+    def test_identity(self):
+        tool = GoogleMusic()
+        info = tool.get_info()
+        assert info["name"] == "google_music"
+        assert info["tier"] == "generate"
+        assert info["capability"] == "music_generation"
+        assert info["provider"] == "google"
+
+    def test_duration_validation(self):
+        tool = GoogleMusic()
+        mock_client = MagicMock()
+
+        mock_interaction = MagicMock()
+        mock_interaction.status = "completed"
+        mock_interaction.output_audio.data = base64.b64encode(b"audio_bytes").decode(
+            "utf-8"
+        )
+        mock_client.interactions.create.return_value = mock_interaction
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+            patch("pathlib.Path.write_bytes"),
+        ):
+            # 1. duration > 184 and auto_fix is True -> coerced to 184
+            inputs = {
+                "prompt": "melodic pop",
+                "duration_seconds": 200,
+                "auto_fix": True,
+                "output_path": "test_out.mp3",
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+            assert res.data["duration_seconds"] == 184.0
+
+            # 2. duration > 184 and auto_fix is False -> raises error
+            inputs = {
+                "prompt": "melodic pop",
+                "duration_seconds": 200,
+                "auto_fix": False,
+            }
+            res = tool.execute(inputs)
+            assert res.success is False
+            assert res.error is not None
+            assert "maximum duration is 184" in res.error
+
+    def test_execute_success_convenience_extraction(self, tmp_path):
+        tool = GoogleMusic()
+        mock_client = MagicMock()
+
+        mock_interaction = MagicMock()
+        mock_interaction.status = "completed"
+        mock_interaction.output_audio.data = base64.b64encode(
+            b"my_fake_google_lyria_audio"
+        ).decode("utf-8")
+        mock_client.interactions.create.return_value = mock_interaction
+
+        output_file = tmp_path / "test_music.mp3"
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+        ):
+            inputs = {
+                "prompt": "atmospheric electronic ambient beat",
+                "duration_seconds": 30,
+                "output_path": str(output_file),
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+            assert res.data["provider"] == "google"
+            assert res.data["model"] == "lyria-3-pro-preview"
+            assert res.data["output"] == str(output_file)
+            assert output_file.read_bytes() == b"my_fake_google_lyria_audio"
+
+    def test_execute_success_fallback_extraction(self, tmp_path):
+        tool = GoogleMusic()
+        mock_client = MagicMock()
+
+        mock_interaction = MagicMock()
+        mock_interaction.status = "completed"
+        del mock_interaction.output_audio
+
+        mock_part = MagicMock()
+        mock_part.type = "audio"
+        mock_part.data = base64.b64encode(b"raw_step_audio").decode("utf-8")
+
+        mock_step = MagicMock()
+        mock_step.type = "model_output"
+        mock_step.content = [mock_part]
+        mock_interaction.steps = [mock_step]
+
+        mock_client.interactions.create.return_value = mock_interaction
+
+        output_file = tmp_path / "test_music_step.mp3"
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+        ):
+            inputs = {
+                "prompt": "jazz piano solo",
+                "duration_seconds": 30,
+                "output_path": str(output_file),
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+            assert res.data["output"] == str(output_file)
+            assert output_file.read_bytes() == b"raw_step_audio"
+
+    @patch("os.path.exists")
+    @patch("requests.get")
+    def test_multimodal_image(self, mock_get, mock_exists, tmp_path):
+        tool = GoogleMusic()
+        mock_client = MagicMock()
+
+        mock_interaction = MagicMock()
+        mock_interaction.status = "completed"
+        mock_interaction.output_audio.data = base64.b64encode(b"audio_bytes").decode(
+            "utf-8"
+        )
+        mock_client.interactions.create.return_value = mock_interaction
+
+        mock_exists.return_value = True
+
+        mock_resp = MagicMock()
+        mock_resp.headers = {"Content-Type": "image/jpeg"}
+        mock_resp.content = b"url_image_bytes"
+        mock_get.return_value = mock_resp
+
+        local_image = tmp_path / "ref.png"
+        with open(local_image, "wb") as f:
+            f.write(b"local_image_bytes")
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+        ):
+            # 1. Local image path
+            inputs = {
+                "prompt": "music inspired by image",
+                "image_path": str(local_image),
+                "output_path": str(tmp_path / "out1.mp3"),
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+
+            called_input = mock_client.interactions.create.call_args[1]["input"]
+            assert len(called_input) == 2
+            assert called_input[0] == {
+                "type": "text",
+                "text": "music inspired by image\n\n[Target Duration: 30 seconds]",
+            }
+            assert called_input[1]["type"] == "image"
+            assert called_input[1]["mime_type"] == "image/png"
+            assert called_input[1]["data"] == base64.b64encode(
+                b"local_image_bytes"
+            ).decode("utf-8")
+
+            # 2. Remote image URL
+            inputs = {
+                "prompt": "music inspired by url",
+                "image_url": "https://example.com/art.jpg",
+                "output_path": str(tmp_path / "out2.mp3"),
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+
+            called_input = mock_client.interactions.create.call_args[1]["input"]
+            assert len(called_input) == 2
+            assert called_input[0] == {
+                "type": "text",
+                "text": "music inspired by url\n\n[Target Duration: 30 seconds]",
+            }
+            assert called_input[1]["type"] == "image"
+            assert called_input[1]["mime_type"] == "image/jpeg"
+            assert called_input[1]["data"] == base64.b64encode(
+                b"url_image_bytes"
+            ).decode("utf-8")
+
+    def test_minimum_duration_validation(self, caplog):
+        import logging
+
+        tool = GoogleMusic()
+        mock_client = MagicMock()
+
+        mock_interaction = MagicMock()
+        mock_interaction.status = "completed"
+        mock_interaction.output_audio.data = base64.b64encode(b"audio_bytes").decode(
+            "utf-8"
+        )
+        mock_client.interactions.create.return_value = mock_interaction
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+            patch("pathlib.Path.write_bytes"),
+            caplog.at_level(logging.WARNING),
+        ):
+            # 1. duration < 5 and auto_fix is True -> coerced to 5 with warning logged
+            inputs = {
+                "prompt": "melodic pop",
+                "duration_seconds": 3,
+                "auto_fix": True,
+                "output_path": "test_out.mp3",
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+            assert res.data["duration_seconds"] == 5.0
+
+            warnings = [
+                rec.message
+                for rec in caplog.records
+                if "minimum duration" in rec.message
+            ]
+            assert len(warnings) == 1
+            assert "minimum duration of 5 seconds" in warnings[0]
+
+            # Clear records for next check
+            caplog.clear()
+
+            # 2. duration < 5 and auto_fix is False -> raises error
+            inputs = {
+                "prompt": "melodic pop",
+                "duration_seconds": 3,
+                "auto_fix": False,
+            }
+            res = tool.execute(inputs)
+            assert res.success is False
+            assert res.error is not None
+            assert "minimum duration is 5" in res.error
+
+    def test_missing_image_path_error(self, tmp_path):
+        tool = GoogleMusic()
+        with patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}):
+            inputs = {
+                "prompt": "music with missing image",
+                "image_path": str(tmp_path / "does_not_exist.png"),
+                "output_path": str(tmp_path / "out.mp3"),
+            }
+            res = tool.execute(inputs)
+            assert res.success is False
+            assert res.error is not None
+            assert "Failed to load visual conditioning image" in res.error
+            assert "Local reference image not found" in res.error
+
+
+# ---- Image Generation Tools ----
+
+
+class TestGoogleImagen:
+    def test_identity(self):
+        tool = GoogleImagen()
+        info = tool.get_info()
+        assert info["name"] == "google_imagen"
+        assert info["tier"] == "generate"
+        assert info["capability"] == "image_generation"
+        assert info["provider"] == "google_imagen"
+
+    def test_capabilities(self):
+        tool = GoogleImagen()
+        assert "text_to_image" in tool.capabilities
+
+
+# ---- Video Generation Tools ----
+
+
+class TestVeoVideo:
+    def test_identity(self):
+        tool = VeoVideo()
+        info = tool.get_info()
+        assert info["name"] == "veo_video"
+        assert info["tier"] == "generate"
+        assert info["capability"] == "video_generation"
+        assert info["provider"] == "veo"
+
+    def test_backend_auto_detect(self):
+        tool = VeoVideo()
+
+        with patch.dict(os.environ, {"GEMINI_API_KEY": "test_key", "FAL_KEY": ""}):
+            if "FAL_KEY" in os.environ:
+                del os.environ["FAL_KEY"]
+            if "FAL_AI_API_KEY" in os.environ:
+                del os.environ["FAL_AI_API_KEY"]
+            assert tool._get_google_credentials_status() is True
+            assert not tool._get_fal_api_key()
+            assert tool.get_status() == ToolStatus.AVAILABLE
+
+        with patch.dict(
+            os.environ,
+            {"GEMINI_API_KEY": "", "GOOGLE_API_KEY": "", "FAL_KEY": "test_fal_key"},
+        ):
+            assert tool._get_google_credentials_status() is False
+            assert tool._get_fal_api_key() == "test_fal_key"
+            assert tool.get_status() == ToolStatus.AVAILABLE
+
+    @patch("tools.video._shared.probe_output")
+    def test_duration_coercion(self, mock_probe):
+        tool = VeoVideo()
+        mock_probe.return_value = {"width": 1920, "height": 1080, "duration": 8.0}
+
+        mock_client = MagicMock()
+        mock_client._api_client.vertexai = False
+        mock_operation = MagicMock()
+        mock_operation.done = True
+        mock_operation.error = None
+
+        mock_video_result = MagicMock()
+        mock_video_result.video = MagicMock()
+        mock_operation.response.generated_videos = [mock_video_result]
+        mock_client.models.generate_videos.return_value = mock_operation
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+        ):
+            # auto_fix = True -> coerced to 8s
+            inputs = {
+                "prompt": "Test prompt",
+                "backend": "google",
+                "resolution": "1080p",
+                "duration": "4s",
+                "auto_fix": True,
+                "output_path": "test_out.mp4",
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+
+            called_config = mock_client.models.generate_videos.call_args[1]["config"]
+            assert called_config.duration_seconds == 8
+
+    @patch("tools.video._shared.probe_output")
+    @patch("PIL.Image.open")
+    @patch("os.path.exists")
+    @patch("requests.get")
+    def test_operations_mapping(
+        self, mock_req_get, mock_exists, mock_img_open, mock_probe
+    ):
+        tool = VeoVideo()
+        mock_probe.return_value = {"width": 1920, "height": 1080, "duration": 8.0}
+        mock_exists.return_value = True
+
+        mock_img = MagicMock()
+        mock_img.format = "PNG"
+        mock_img_open.return_value = mock_img
+
+        mock_resp = MagicMock()
+        mock_resp.content = b"fake_image_bytes"
+        mock_req_get.return_value = mock_resp
+
+        mock_client = MagicMock()
+        mock_client._api_client.vertexai = False
+        mock_operation = MagicMock()
+        mock_operation.done = True
+        mock_operation.error = None
+        mock_video_result = MagicMock()
+        mock_video_result.video = MagicMock()
+        mock_operation.response.generated_videos = [mock_video_result]
+        mock_client.models.generate_videos.return_value = mock_operation
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+        ):
+            # text_to_video
+            inputs = {
+                "prompt": "Test text to video",
+                "backend": "google",
+                "operation": "text_to_video",
+                "duration": "8s",
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+
+            called_kwargs = mock_client.models.generate_videos.call_args[1]
+            assert called_kwargs["image"] is None
+
+            # image_to_video
+            inputs = {
+                "prompt": "Test image to video",
+                "backend": "google",
+                "operation": "image_to_video",
+                "image_path": "local_img.png",
+                "duration": "8s",
+            }
+            res = tool.execute(inputs)
+            assert res.success is True
+
+            called_kwargs = mock_client.models.generate_videos.call_args[1]
+            assert called_kwargs["image"] is not None
+
+    def test_vertex_ai_mode_rejection(self):
+        tool = VeoVideo()
+        mock_client = MagicMock()
+        mock_client.vertexai = True
+        if hasattr(mock_client, "_api_client"):
+            delattr(mock_client, "_api_client")
+
+        with (
+            patch.dict(os.environ, {"GEMINI_API_KEY": "test_key"}),
+            patch("google.genai.Client", return_value=mock_client),
+        ):
+            inputs = {
+                "prompt": "cinematic shot",
+                "backend": "google",
+            }
+            res = tool.execute(inputs)
+            assert res.success is False
+            assert res.error is not None
+            assert "only supported using the Gemini Developer API" in res.error
+
+    def test_missing_local_image_paths(self):
+        tool = VeoVideo()
+        with patch.dict(
+            os.environ,
+            {"GEMINI_API_KEY": "test_key", "GOOGLE_GENAI_USE_VERTEXAI": "false"},
+        ):
+            inputs = {
+                "prompt": "cinematic shot",
+                "backend": "google",
+                "operation": "image_to_video",
+                "image_path": "non_existent_file_path_12345.png",
+            }
+            res = tool.execute(inputs)
+            assert res.success is False
+            assert "Local input image not found" in res.error
+
+    def test_missing_reference_image_paths(self):
+        tool = VeoVideo()
+        with patch.dict(
+            os.environ,
+            {"GEMINI_API_KEY": "test_key", "GOOGLE_GENAI_USE_VERTEXAI": "false"},
+        ):
+            inputs = {
+                "prompt": "cinematic shot",
+                "backend": "google",
+                "operation": "reference_to_video",
+                "reference_image_paths": ["non_existent_reference_12345.png"],
+            }
+            res = tool.execute(inputs)
+            assert res.success is False
+            assert "Local reference image not found" in res.error
+
+
 class TestNewToolsRegistry:
     def test_all_register(self):
         reg = ToolRegistry()
@@ -135,7 +648,9 @@ class TestCapabilityMetadata:
         info = tool.get_info()
         assert info["capability"] == "tts"
         assert info["provider"] == "elevenlabs"
-        assert info["usage_location"].endswith("tools\\audio\\elevenlabs_tts.py") or info["usage_location"].endswith("tools/audio/elevenlabs_tts.py")
+        assert info["usage_location"].endswith(
+            "tools\\audio\\elevenlabs_tts.py"
+        ) or info["usage_location"].endswith("tools/audio/elevenlabs_tts.py")
         assert "related_skills" in info
         assert "fallback_tools" in info
 
@@ -151,7 +666,9 @@ class TestCapabilityMetadata:
             "piper_tts",
             "tts_selector",
         }
-        assert {tool.name for tool in reg.get_by_provider("elevenlabs")} == {"elevenlabs_tts"}
+        assert {tool.name for tool in reg.get_by_provider("elevenlabs")} == {
+            "elevenlabs_tts"
+        }
 
     def test_registry_catalog_views(self):
         reg = ToolRegistry()
@@ -173,6 +690,7 @@ class TestCapabilityMetadata:
 
 # ---- Animated Explainer Pipeline ----
 
+
 class TestAnimatedExplainerManifest:
     def test_loads(self):
         manifest = load_pipeline("animated-explainer")
@@ -182,7 +700,16 @@ class TestAnimatedExplainerManifest:
     def test_all_stages_present(self):
         manifest = load_pipeline("animated-explainer")
         stage_names = get_stage_order(manifest)
-        expected = ["research", "proposal", "script", "scene_plan", "assets", "edit", "compose", "publish"]
+        expected = [
+            "research",
+            "proposal",
+            "script",
+            "scene_plan",
+            "assets",
+            "edit",
+            "compose",
+            "publish",
+        ]
         assert stage_names == expected
 
     def test_every_stage_has_skill(self):
@@ -197,7 +724,9 @@ class TestAnimatedExplainerManifest:
         manifest = load_pipeline("animated-explainer")
         for stage in manifest["stages"]:
             focus = get_stage_review_focus(manifest, stage["name"])
-            assert len(focus) >= 3, f"Stage {stage['name']} needs more review focus items"
+            assert len(focus) >= 3, (
+                f"Stage {stage['name']} needs more review focus items"
+            )
 
     def test_required_tools_complete(self):
         manifest = load_pipeline("animated-explainer")
@@ -221,6 +750,7 @@ class TestAnimatedExplainerManifest:
 
 # ---- Style Playbooks ----
 
+
 class TestStylePlaybooks:
     def test_all_listed(self):
         playbooks = list_playbooks()
@@ -228,13 +758,17 @@ class TestStylePlaybooks:
         assert "flat-motion-graphics" in playbooks
         assert "minimalist-diagram" in playbooks
 
-    @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
+    @pytest.mark.parametrize(
+        "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]
+    )
     def test_loads_and_validates(self, name):
         pb = load_playbook(name)
         assert pb["identity"]["name"]
         assert pb["identity"]["category"]
 
-    @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
+    @pytest.mark.parametrize(
+        "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]
+    )
     def test_has_required_sections(self, name):
         pb = load_playbook(name)
         assert "visual_language" in pb
@@ -245,7 +779,9 @@ class TestStylePlaybooks:
         assert "quality_rules" in pb
         assert len(pb["quality_rules"]) >= 3
 
-    @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
+    @pytest.mark.parametrize(
+        "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]
+    )
     def test_color_palette_complete(self, name):
         pb = load_playbook(name)
         palette = pb["visual_language"]["color_palette"]
@@ -254,7 +790,9 @@ class TestStylePlaybooks:
         assert "background" in palette
         assert "text" in palette
 
-    @pytest.mark.parametrize("name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"])
+    @pytest.mark.parametrize(
+        "name", ["clean-professional", "flat-motion-graphics", "minimalist-diagram"]
+    )
     def test_pacing_rules_present(self, name):
         pb = load_playbook(name)
         pacing = pb["motion"]["pacing_rules"]
@@ -268,60 +806,75 @@ class TestStylePlaybooks:
         # compatible_playbooks is a dict with recommended/also_works lists
         playbook_names = compat.get("recommended", []) + compat.get("also_works", [])
         for name in playbook_names:
-            assert name in available, f"Manifest references unavailable playbook: {name}"
+            assert name in available, (
+                f"Manifest references unavailable playbook: {name}"
+            )
 
 
 # ---- Skills Existence ----
 
+
 class TestSkillsExist:
     SKILLS_DIR = PROJECT_ROOT / "skills"
 
-    @pytest.mark.parametrize("skill_path", [
-        "pipelines/explainer/idea-director.md",
-        "pipelines/explainer/script-director.md",
-        "pipelines/explainer/scene-director.md",
-        "pipelines/explainer/asset-director.md",
-        "pipelines/explainer/edit-director.md",
-        "pipelines/explainer/compose-director.md",
-        "pipelines/explainer/publish-director.md",
-    ])
+    @pytest.mark.parametrize(
+        "skill_path",
+        [
+            "pipelines/explainer/idea-director.md",
+            "pipelines/explainer/script-director.md",
+            "pipelines/explainer/scene-director.md",
+            "pipelines/explainer/asset-director.md",
+            "pipelines/explainer/edit-director.md",
+            "pipelines/explainer/compose-director.md",
+            "pipelines/explainer/publish-director.md",
+        ],
+    )
     def test_director_skills_exist(self, skill_path):
         full_path = self.SKILLS_DIR / skill_path
         assert full_path.exists(), f"Missing director skill: {skill_path}"
         content = full_path.read_text(encoding="utf-8")
         assert len(content) > 500, f"Skill too short to be useful: {skill_path}"
 
-    @pytest.mark.parametrize("skill_path", [
-        "meta/reviewer.md",
-        "meta/checkpoint-protocol.md",
-        "meta/skill-creator.md",
-    ])
+    @pytest.mark.parametrize(
+        "skill_path",
+        [
+            "meta/reviewer.md",
+            "meta/checkpoint-protocol.md",
+            "meta/skill-creator.md",
+        ],
+    )
     def test_meta_skills_exist(self, skill_path):
         full_path = self.SKILLS_DIR / skill_path
         assert full_path.exists(), f"Missing meta skill: {skill_path}"
         content = full_path.read_text(encoding="utf-8")
         assert len(content) > 500, f"Skill too short to be useful: {skill_path}"
 
-    @pytest.mark.parametrize("skill_path", [
-        "pipelines/explainer/idea-director.md",
-        "pipelines/explainer/script-director.md",
-        "pipelines/explainer/scene-director.md",
-        "pipelines/explainer/asset-director.md",
-        "pipelines/explainer/edit-director.md",
-        "pipelines/explainer/compose-director.md",
-        "pipelines/explainer/publish-director.md",
-    ])
+    @pytest.mark.parametrize(
+        "skill_path",
+        [
+            "pipelines/explainer/idea-director.md",
+            "pipelines/explainer/script-director.md",
+            "pipelines/explainer/scene-director.md",
+            "pipelines/explainer/asset-director.md",
+            "pipelines/explainer/edit-director.md",
+            "pipelines/explainer/compose-director.md",
+            "pipelines/explainer/publish-director.md",
+        ],
+    )
     def test_director_skills_have_required_sections(self, skill_path):
         content = (self.SKILLS_DIR / skill_path).read_text(encoding="utf-8")
         assert "## When to Use" in content
         assert "## Process" in content or "## Protocol" in content
         assert "Self-Evaluate" in content or "self-evaluate" in content.lower()
 
-    @pytest.mark.parametrize("skill_path", [
-        "meta/reviewer.md",
-        "meta/checkpoint-protocol.md",
-        "meta/skill-creator.md",
-    ])
+    @pytest.mark.parametrize(
+        "skill_path",
+        [
+            "meta/reviewer.md",
+            "meta/checkpoint-protocol.md",
+            "meta/skill-creator.md",
+        ],
+    )
     def test_meta_skills_have_required_sections(self, skill_path):
         content = (self.SKILLS_DIR / skill_path).read_text(encoding="utf-8")
         assert "## When to Use" in content
@@ -330,6 +883,7 @@ class TestSkillsExist:
 
 # ---- Remotion Scaffold ----
 
+
 class TestRemotionScaffold:
     REMOTION_DIR = PROJECT_ROOT / "remotion-composer"
 
@@ -354,17 +908,23 @@ class TestRemotionScaffold:
 
 # ---- Video Compose Operations ----
 
+
 class TestVideoComposeOperations:
     def test_render_operation_exists(self):
+        from typing import Any
         from tools.video.video_compose import VideoCompose
+
         tool = VideoCompose()
-        ops = tool.input_schema["properties"]["operation"]["enum"]
+        schema: Any = tool.input_schema
+        ops = schema["properties"]["operation"]["enum"]
         assert "render" in ops
         assert "remotion_render" in ops
 
     def test_render_rejects_missing_inputs(self):
         from tools.video.video_compose import VideoCompose
+
         tool = VideoCompose()
         result = tool.execute({"operation": "render"})
         assert not result.success
+        assert result.error is not None
         assert "edit_decisions" in result.error

tools/audio/google_music.py

@@ -0,0 +1,333 @@
+"""Generate music using Google Lyria via Google GenAI SDK.
+
+Generate background music and audio tracks for video production using lyria-3-pro-preview.
+"""
+
+from __future__ import annotations
+
+import base64
+import mimetypes
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+from tools.base_tool import (
+    BaseTool,
+    Determinism,
+    ExecutionMode,
+    ResourceProfile,
+    RetryPolicy,
+    ToolResult,
+    ToolRuntime,
+    ToolStability,
+    ToolStatus,
+    ToolTier,
+)
+
+
+class GoogleMusic(BaseTool):
+    name = "google_music"
+    version = "0.1.0"
+    tier = ToolTier.GENERATE
+    capability = "music_generation"
+    provider = "google"
+    stability = ToolStability.EXPERIMENTAL
+    execution_mode = ExecutionMode.SYNC
+    determinism = Determinism.STOCHASTIC
+    runtime = ToolRuntime.API
+
+    dependencies = []
+    install_instructions = (
+        "Configure Google credentials:\n"
+        "  - Set GEMINI_API_KEY (or GOOGLE_API_KEY) in environment.\n"
+        "  - Or set GOOGLE_APPLICATION_CREDENTIALS for Vertex AI service account."
+    )
+    fallback_tools = ["music_gen"]
+    agent_skills = ["music"]
+
+    capabilities = [
+        "generate_background_music",
+    ]
+    supports = {
+        "instrumental": True,
+        "vocals": True,
+        "custom_lyrics": True,
+        "style_control": True,
+        "long_form": True,
+    }
+    best_for = [
+        "high-quality instrumental background music",
+        "genre-specific music guided by rich text prompts",
+        "Google ecosystem integration",
+    ]
+    not_good_for = [
+        "offline generation",
+        "sub-5-second sound effects",
+    ]
+
+    input_schema = {
+        "type": "object",
+        "required": ["prompt"],
+        "properties": {
+            "prompt": {
+                "type": "string",
+                "description": "Music description (mood, genre, instruments, tempo)",
+            },
+            "duration_seconds": {
+                "type": "number",
+                "minimum": 5,
+                "maximum": 184,
+                "default": 30,
+                "description": "Target duration in seconds (model hard limit is 184s)",
+            },
+            "image_url": {
+                "type": "string",
+                "description": "Reference image URL for visual music conditioning",
+            },
+            "image_path": {
+                "type": "string",
+                "description": "Local reference image path for visual music conditioning",
+            },
+            "auto_fix": {"type": "boolean", "default": True},
+            "output_path": {
+                "type": "string",
+                "default": "music_output.mp3",
+                "description": "Path where the generated MP3 file should be written",
+            },
+        },
+    }
+
+    resource_profile = ResourceProfile(
+        cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=50, network_required=True
+    )
+    retry_policy = RetryPolicy(
+        max_retries=2, retryable_errors=["rate_limit", "timeout"]
+    )
+    idempotency_key_fields = ["prompt", "duration_seconds", "image_url", "image_path"]
+    side_effects = [
+        "writes audio file to output_path",
+        "calls Google Gemini/Vertex API",
+    ]
+    user_visible_verification = [
+        "Listen to generated music for style and quality",
+    ]
+
+    def _get_google_credentials_status(self) -> bool:
+        """Check whether Google API keys or Vertex AI service account credentials are set."""
+        from tools.google_credentials import has_google_credentials
+
+        return has_google_credentials()
+
+    def get_status(self) -> ToolStatus:
+        """Determine whether the tool is available based on configured credentials."""
+        if self._get_google_credentials_status():
+            return ToolStatus.AVAILABLE
+        return ToolStatus.UNAVAILABLE
+
+    def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        """Estimate the generation cost in USD."""
+        # Lyria 3 Pro is a flat $0.08 per generation request
+        return 0.08
+
+    def execute(self, inputs: dict[str, Any]) -> ToolResult:
+        """Execute the music generation tool using the Google GenAI SDK."""
+        if not self._get_google_credentials_status():
+            return ToolResult(
+                success=False,
+                error="No Google credentials configured. " + self.install_instructions,
+            )
+
+        start = time.time()
+
+        try:
+            import requests
+            from google.genai import types
+            from tools.google_credentials import get_genai_client, GOOGLE_API_TIMEOUT_MS
+
+            http_options = types.HttpOptions(timeout=GOOGLE_API_TIMEOUT_MS)
+            client = get_genai_client(http_options=http_options)
+        except ImportError as e:
+            return ToolResult(
+                success=False,
+                error=f"Failed to import required Google libraries: {e}. Run 'uv pip install google-genai requests'",
+            )
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                error=f"Failed to initialize Google GenAI Client: {e}",
+            )
+
+        prompt = inputs["prompt"]
+        duration = float(inputs.get("duration_seconds", 30))
+        auto_fix = inputs.get("auto_fix", True)
+        output_path = inputs.get("output_path", "music_output.mp3")
+
+        # Ensure minimum duration of 5 seconds
+        if duration < 5:
+            if auto_fix:
+                import logging
+
+                logging.getLogger(__name__).warning(
+                    "Lyria 3 Pro requires a minimum duration of 5 seconds. Coercing duration_seconds to 5.0."
+                )
+                duration = 5.0
+            else:
+                return ToolResult(
+                    success=False,
+                    error="lyria-3-pro-preview minimum duration is 5 seconds.",
+                )
+
+        # Cap at 184 seconds
+        if duration > 184:
+            if auto_fix:
+                import logging
+
+                logging.getLogger(__name__).warning(
+                    "Lyria 3 Pro supports up to 184 seconds of audio. Coercing duration_seconds to 184."
+                )
+                duration = 184.0
+            else:
+                return ToolResult(
+                    success=False,
+                    error="lyria-3-pro-preview maximum duration is 184 seconds.",
+                )
+
+        # Helper to load reference image bytes + mime type
+        def _get_image_data(
+            url: str | None, path: str | None
+        ) -> tuple[str, str] | None:
+            if path:
+                if not os.path.exists(path):
+                    raise FileNotFoundError(f"Local reference image not found: {path}")
+                img_bytes = Path(path).read_bytes()
+                mime, _ = mimetypes.guess_type(path)
+                if not mime:
+                    mime = "image/png"
+                b64 = base64.b64encode(img_bytes).decode("utf-8")
+                return b64, mime
+            if url:
+                resp = requests.get(url, timeout=30)
+                resp.raise_for_status()
+                mime = resp.headers.get("Content-Type")
+                if not mime or "image" not in mime:
+                    mime = "image/png"
+                b64 = base64.b64encode(resp.content).decode("utf-8")
+                return b64, mime
+            return None
+
+        # Build payload input incorporating target duration instructions
+        timed_prompt = f"{prompt}\n\n[Target Duration: {int(duration)} seconds]"
+        input_list: list[dict[str, Any]] = [{"type": "text", "text": timed_prompt}]
+        try:
+            image_data = _get_image_data(
+                inputs.get("image_url"), inputs.get("image_path")
+            )
+            if image_data:
+                b64, mime = image_data
+                input_list.append({"type": "image", "mime_type": mime, "data": b64})
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                error=f"Failed to load visual conditioning image: {e}",
+            )
+
+        model_name = "lyria-3-pro-preview"
+
+        try:
+            # Create parent dirs if needed
+            Path(output_path).parent.mkdir(parents=True, exist_ok=True)
+
+            interaction = client.interactions.create(model=model_name, input=input_list)
+
+            if hasattr(interaction, "status") and interaction.status in (
+                "failed",
+                "cancelled",
+            ):
+                return ToolResult(
+                    success=False,
+                    error=f"Google Lyria music generation failed. Status: {interaction.status}",
+                )
+
+            audio_data = None
+            if hasattr(interaction, "output_audio") and interaction.output_audio:
+                audio_data = getattr(interaction.output_audio, "data", None)
+
+            # Fall back to outputs list
+            outputs = getattr(interaction, "outputs", None)
+            if not audio_data and isinstance(outputs, list):
+                for output in outputs:
+                    if hasattr(output, "inline_data") and output.inline_data:
+                        audio_data = getattr(output.inline_data, "data", None)
+                        if audio_data:
+                            break
+
+            # Fall back to step traversal
+            steps = getattr(interaction, "steps", None)
+            if not audio_data and isinstance(steps, list):
+                for step in steps:
+                    if (
+                        hasattr(step, "type")
+                        and step.type == "model_output"
+                        and hasattr(step, "content")
+                        and step.content
+                    ):
+                        for content_part in step.content:
+                            if (
+                                hasattr(content_part, "type")
+                                and content_part.type == "audio"
+                            ):
+                                audio_data = getattr(content_part, "data", None)
+                                break
+                        if audio_data:
+                            break
+
+            if not audio_data:
+                return ToolResult(
+                    success=False,
+                    error=f"No audio data returned by model {model_name}.",
+                )
+
+            # Decode and save output file
+            if isinstance(audio_data, str):
+                audio_bytes = base64.b64decode(audio_data)
+            else:
+                # If it's already bytes, it could be raw audio or base64 bytes
+                if audio_data.startswith(b"ID3") or (
+                    len(audio_data) > 2
+                    and audio_data[0] == 0xFF
+                    and (audio_data[1] & 0xE0) == 0xE0
+                ):
+                    audio_bytes = audio_data
+                else:
+                    try:
+                        audio_bytes = base64.b64decode(audio_data)
+                    except Exception:
+                        audio_bytes = audio_data
+
+            Path(output_path).write_bytes(audio_bytes)
+
+        except Exception as e:
+            return ToolResult(
+                success=False, error=f"Google Lyria music generation failed: {e}"
+            )
+
+        duration_seconds = round(time.time() - start, 2)
+        cost_usd = self.estimate_cost(inputs)
+
+        return ToolResult(
+            success=True,
+            data={
+                "provider": "google",
+                "model": model_name,
+                "prompt": prompt,
+                "duration_seconds": duration,
+                "output": str(output_path),
+                "output_path": str(output_path),
+                "format": "mp3",
+            },
+            artifacts=[str(output_path)],
+            cost_usd=cost_usd,
+            duration_seconds=duration_seconds,
+            model=model_name,
+        )

tools/audio/google_tts.py

@@ -24,7 +24,11 @@ from tools.base_tool import (
     ToolStatus,
     ToolTier,
 )
-from tools.google_credentials import get_access_token, service_account_configured
+from tools.google_credentials import (
+    get_access_token,
+    service_account_configured,
+    has_google_credentials,
+)
 
 
 class GoogleTTS(BaseTool):
@@ -121,8 +125,17 @@ class GoogleTTS(BaseTool):
     resource_profile = ResourceProfile(
         cpu_cores=1, ram_mb=256, vram_mb=0, disk_mb=50, network_required=True
     )
-    retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"])
-    idempotency_key_fields = ["text", "input_type", "voice", "language_code", "speaking_rate", "pitch"]
+    retry_policy = RetryPolicy(
+        max_retries=2, retryable_errors=["rate_limit", "timeout"]
+    )
+    idempotency_key_fields = [
+        "text",
+        "input_type",
+        "voice",
+        "language_code",
+        "speaking_rate",
+        "pitch",
+    ]
     side_effects = ["writes audio file to output_path", "calls Google Cloud TTS API"]
     user_visible_verification = ["Listen to generated audio for natural speech quality"]
 
@@ -141,7 +154,7 @@ class GoogleTTS(BaseTool):
     def get_status(self) -> ToolStatus:
         # Available via either an API key or a service-account JSON. Both paths
         # are honoured by execute() — so this no longer over-reports.
-        if self._get_api_key() or service_account_configured():
+        if has_google_credentials():
             return ToolStatus.AVAILABLE
         return ToolStatus.UNAVAILABLE
 
@@ -226,7 +239,11 @@ class GoogleTTS(BaseTool):
 
         if input_type == "ssml":
             stripped = text.strip()
-            ssml = stripped if stripped.startswith("<speak") else f"<speak>{stripped}</speak>"
+            ssml = (
+                stripped
+                if stripped.startswith("<speak")
+                else f"<speak>{stripped}</speak>"
+            )
             synthesis_input = {"ssml": ssml}
         else:
             synthesis_input = {"text": text}
@@ -252,7 +269,7 @@ class GoogleTTS(BaseTool):
         params: dict[str, str] = {}
         if bearer_token:
             headers["Authorization"] = f"Bearer {bearer_token}"
-        else:
+        elif api_key:
             params["key"] = api_key
 
         response = requests.post(

tools/google_credentials.py

@@ -13,10 +13,15 @@ surfaces as an actionable runtime error rather than a hard import failure.
 from __future__ import annotations
 
 import os
+from typing import Any
 
 # Broad scope that covers Cloud Text-to-Speech and Vertex AI prediction.
 CLOUD_PLATFORM_SCOPE = "https://www.googleapis.com/auth/cloud-platform"
 
+# Shared constants for long-running Google/Vertex AI generation calls (e.g. music, video)
+GOOGLE_API_TIMEOUT_SECONDS = 600
+GOOGLE_API_TIMEOUT_MS = GOOGLE_API_TIMEOUT_SECONDS * 1000
+
 
 def service_account_configured() -> bool:
     """True when GOOGLE_APPLICATION_CREDENTIALS points to an existing file."""
@@ -24,6 +29,41 @@ def service_account_configured() -> bool:
     return bool(path and os.path.exists(path))
 
 
+def has_google_credentials() -> bool:
+    """True when GOOGLE_API_KEY, GEMINI_API_KEY, or service account is configured."""
+    return bool(
+        os.environ.get("GOOGLE_API_KEY")
+        or os.environ.get("GEMINI_API_KEY")
+        or service_account_configured()
+    )
+
+
+def get_genai_client(http_options: Any | None = None) -> Any:
+    """Lazily import and initialize the Google GenAI Client based on configured credentials."""
+    from google import genai
+
+    api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
+    use_vertex = os.environ.get("GOOGLE_GENAI_USE_VERTEXAI", "").lower() in (
+        "true",
+        "1",
+    ) or os.environ.get("GOOGLE_GENAI_USE_ENTERPRISE", "").lower() in ("true", "1")
+
+    if use_vertex or (not api_key and service_account_configured()):
+        kwargs = {
+            "vertexai": True,
+            "location": os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1"),
+            "http_options": http_options,
+        }
+        project_id = resolve_project_id()
+        if project_id:
+            kwargs["project"] = project_id
+        return genai.Client(**kwargs)
+    else:
+        if api_key:
+            return genai.Client(api_key=api_key, http_options=http_options)
+        return genai.Client(http_options=http_options)
+
+
 def resolve_project_id(creds_project_id: str | None = None) -> str | None:
     """Resolve the GCP project id from env vars, falling back to the key file's.
 
@@ -77,4 +117,12 @@ def get_access_token(scopes: list[str] | None = None) -> tuple[str, str | None]:
             f"Failed to load/refresh service-account credentials from {path}: {exc}"
         ) from exc
 
-    return creds.token, getattr(creds, "project_id", None)
+    token = creds.token
+    if not token or not isinstance(token, str):
+        raise RuntimeError(
+            "Service-account credentials did not yield a valid access token."
+        )
+
+    project_id = getattr(creds, "project_id", None)
+    ret_project_id = str(project_id) if project_id is not None else None
+    return token, ret_project_id

tools/graphics/google_imagen.py

@@ -24,6 +24,7 @@ from tools.google_credentials import (
     get_access_token,
     resolve_project_id,
     service_account_configured,
+    has_google_credentials,
 )
 
 # Aspect ratio to approximate pixel dimensions (for cost/reporting only)
@@ -93,7 +94,10 @@ class GoogleImagen(BaseTool):
         "type": "object",
         "required": ["prompt"],
         "properties": {
-            "prompt": {"type": "string", "description": "Image description (max 480 tokens)"},
+            "prompt": {
+                "type": "string",
+                "description": "Image description (max 480 tokens)",
+            },
             "aspect_ratio": {
                 "type": "string",
                 "enum": ["1:1", "3:4", "4:3", "9:16", "16:9"],
@@ -131,9 +135,14 @@ class GoogleImagen(BaseTool):
     resource_profile = ResourceProfile(
         cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=100, network_required=True
     )
-    retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"])
+    retry_policy = RetryPolicy(
+        max_retries=2, retryable_errors=["rate_limit", "timeout"]
+    )
     idempotency_key_fields = ["prompt", "aspect_ratio", "model"]
-    side_effects = ["writes image file to output_path", "calls Google Generative AI API"]
+    side_effects = [
+        "writes image file to output_path",
+        "calls Google Generative AI API",
+    ]
     user_visible_verification = ["Inspect generated image for relevance and quality"]
 
     def _get_api_key(self) -> str | None:
@@ -141,7 +150,7 @@ class GoogleImagen(BaseTool):
 
     def get_status(self) -> ToolStatus:
         # API key -> AI Studio endpoint; service-account JSON -> Vertex AI.
-        if self._get_api_key() or service_account_configured():
+        if has_google_credentials():
             return ToolStatus.AVAILABLE
         return ToolStatus.UNAVAILABLE
 
@@ -188,6 +197,7 @@ class GoogleImagen(BaseTool):
         prompt = inputs["prompt"]
 
         import logging
+
         logger = logging.getLogger(__name__)
 
         # Resolve aspect ratio: explicit > derived from width/height > default
@@ -198,7 +208,8 @@ class GoogleImagen(BaseTool):
             aspect_ratio = _dims_to_aspect_ratio(inputs["width"], inputs["height"])
             logger.info(
                 "google_imagen: remapped %s to nearest supported aspect ratio %s",
-                requested_ratio, aspect_ratio,
+                requested_ratio,
+                aspect_ratio,
             )
         else:
             aspect_ratio = "1:1"
@@ -228,7 +239,7 @@ class GoogleImagen(BaseTool):
             )
             headers = {
                 "Content-Type": "application/json",
-                "x-goog-api-key": api_key,
+                "x-goog-api-key": api_key or "",
             }
 
         try:
@@ -246,11 +257,11 @@ class GoogleImagen(BaseTool):
 
             predictions = data.get("predictions", [])
             if not predictions:
-                return ToolResult(success=False, error="No images returned from Imagen API")
+                return ToolResult(
+                    success=False, error="No images returned from Imagen API"
+                )
 
-            image_bytes = base64.b64decode(
-                predictions[0]["bytesBase64Encoded"]
-            )
+            image_bytes = base64.b64decode(predictions[0]["bytesBase64Encoded"])
 
             output_path = Path(inputs.get("output_path", "generated_image.png"))
             output_path.parent.mkdir(parents=True, exist_ok=True)

tools/video/veo_video.py

@@ -1,6 +1,6 @@
-"""Google Veo 3.1 video generation via fal.ai API.
+"""Generate video using Google Veo 3.1 via fal.ai or Google GenAI API.
 
-Supports text-to-video, image-to-video, reference-to-video, and first/last-frame
+Support text-to-video, image-to-video, reference-to-video, and first/last-frame
 interpolation so agents can preserve visual consistency instead of relying only on
 raw text prompts.
 """
@@ -14,6 +14,8 @@ import time
 from pathlib import Path
 from typing import Any
 
+from tools.google_credentials import GOOGLE_API_TIMEOUT_SECONDS
+
 from tools.base_tool import (
     BaseTool,
     Determinism,
@@ -30,7 +32,7 @@ from tools.base_tool import (
 
 class VeoVideo(BaseTool):
     name = "veo_video"
-    version = "0.1.0"
+    version = "0.2.0"
     tier = ToolTier.GENERATE
     capability = "video_generation"
     provider = "veo"
@@ -41,12 +43,21 @@ class VeoVideo(BaseTool):
 
     dependencies = []
     install_instructions = (
-        "Set FAL_KEY or FAL_AI_API_KEY to your fal.ai API key.\n"
-        "  Get one at https://fal.ai/dashboard/keys"
+        "Configure at least one backend API key:\n"
+        "  - Direct Google GenAI backend: Set GEMINI_API_KEY (or GOOGLE_API_KEY).\n"
+        "    Get a key at https://aistudio.google.com/\n"
+        "    Or set GOOGLE_APPLICATION_CREDENTIALS for Vertex AI service account.\n"
+        "  - FAL.ai backend: Set FAL_KEY (or FAL_AI_API_KEY).\n"
+        "    Get one at https://fal.ai/dashboard/keys"
     )
     agent_skills = ["ai-video-gen"]
 
-    capabilities = ["text_to_video", "image_to_video", "reference_to_video", "first_last_frame_to_video"]
+    capabilities = [
+        "text_to_video",
+        "image_to_video",
+        "reference_to_video",
+        "first_last_frame_to_video",
+    ]
     supports = {
         "text_to_video": True,
         "image_to_video": True,
@@ -69,21 +80,31 @@ class VeoVideo(BaseTool):
         "required": ["prompt"],
         "properties": {
             "prompt": {"type": "string"},
+            "backend": {
+                "type": "string",
+                "enum": ["auto", "google", "fal"],
+                "default": "auto",
+                "description": "API backend provider to use for generation",
+            },
             "operation": {
                 "type": "string",
-                "enum": ["text_to_video", "image_to_video", "reference_to_video", "first_last_frame_to_video"],
+                "enum": [
+                    "text_to_video",
+                    "image_to_video",
+                    "reference_to_video",
+                    "first_last_frame_to_video",
+                ],
                 "default": "text_to_video",
             },
             "model_variant": {
                 "type": "string",
-                "enum": ["veo3", "veo3/fast", "veo3.1", "veo3.1/fast"],
                 "default": "veo3.1",
+                "description": "Model variant for FAL (e.g. veo3.1) or custom model for Google",
             },
             "duration": {
                 "type": "string",
-                "enum": ["4s", "6s", "8s"],
                 "default": "8s",
-                "description": "Duration in seconds",
+                "description": "Duration (e.g., '4s', '6s', '8s')",
             },
             "aspect_ratio": {
                 "type": "string",
@@ -108,8 +129,14 @@ class VeoVideo(BaseTool):
                 "enum": ["1", "2", "3", "4", "5", "6"],
                 "default": "4",
             },
-            "image_url": {"type": "string", "description": "Reference image URL for image_to_video"},
-            "image_path": {"type": "string", "description": "Local reference image path for image_to_video"},
+            "image_url": {
+                "type": "string",
+                "description": "Reference image URL for image_to_video",
+            },
+            "image_path": {
+                "type": "string",
+                "description": "Local reference image path for image_to_video",
+            },
             "reference_image_urls": {
                 "type": "array",
                 "items": {"type": "string"},
@@ -131,26 +158,56 @@ class VeoVideo(BaseTool):
     resource_profile = ResourceProfile(
         cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True
     )
-    retry_policy = RetryPolicy(max_retries=2, retryable_errors=["rate_limit", "timeout"])
+    retry_policy = RetryPolicy(
+        max_retries=2, retryable_errors=["rate_limit", "timeout"]
+    )
     idempotency_key_fields = ["prompt", "model_variant", "operation", "duration"]
-    side_effects = ["writes video file to output_path", "calls fal.ai API"]
+    side_effects = ["writes video file to output_path", "calls fal.ai or Google APIs"]
     user_visible_verification = [
         "Watch generated clip for visual quality and motion",
         "Listen for audio synchronization and quality",
     ]
 
-    def _get_api_key(self) -> str | None:
+    def _get_google_credentials_status(self) -> bool:
+        """Check whether Google API keys or Vertex AI service account credentials are set."""
+        from tools.google_credentials import has_google_credentials
+
+        return has_google_credentials()
+
+    def _get_fal_api_key(self) -> str | None:
+        """Retrieve the FAL API key from environment variables."""
         return os.environ.get("FAL_KEY") or os.environ.get("FAL_AI_API_KEY")
 
     def get_status(self) -> ToolStatus:
-        if self._get_api_key():
+        """Determine whether the tool is available based on configured credentials."""
+        if self._get_google_credentials_status() or self._get_fal_api_key():
             return ToolStatus.AVAILABLE
         return ToolStatus.UNAVAILABLE
 
     def estimate_cost(self, inputs: dict[str, Any]) -> float:
+        """Estimate the generation cost in USD based on input parameters."""
+        # Determine active backend using inputs and environment
+        backend = inputs.get("backend", "auto")
+        if backend == "auto":
+            if self._get_google_credentials_status():
+                backend = "google"
+            elif self._get_fal_api_key():
+                backend = "fal"
+            else:
+                backend = "google"
+
+        duration_text = str(inputs.get("duration", "8s")).lower().replace("s", "")
+        try:
+            duration = int(duration_text)
+        except ValueError:
+            duration = 8
+
+        if backend == "google":
+            # Standard Google Veo is $0.40 per second
+            return round(duration * 0.40, 4)
+
+        # FAL cost estimation
         variant = inputs.get("model_variant", "veo3.1")
-        duration_text = str(inputs.get("duration", "8s")).replace("s", "")
-        duration = int(duration_text)
         resolution = inputs.get("resolution", "1080p")
         generate_audio = bool(inputs.get("generate_audio", True))
 
@@ -168,6 +225,14 @@ class VeoVideo(BaseTool):
         return (audio_per_second if generate_audio else base_per_second) * duration
 
     def estimate_runtime(self, inputs: dict[str, Any]) -> float:
+        """Estimate the expected runtime in seconds."""
+        backend = inputs.get("backend", "auto")
+        if backend == "auto":
+            backend = "google" if self._get_google_credentials_status() else "fal"
+
+        if backend == "google":
+            return 90.0
+
         variant = inputs.get("model_variant", "veo3.1")
         if "fast" in variant:
             return 45.0
@@ -175,6 +240,7 @@ class VeoVideo(BaseTool):
 
     @staticmethod
     def _file_to_data_uri(path_str: str) -> str:
+        """Convert a local file into a base64-encoded Data URI."""
         path = Path(path_str)
         if not path.exists():
             raise FileNotFoundError(f"Input file not found: {path}")
@@ -184,7 +250,10 @@ class VeoVideo(BaseTool):
         encoded = base64.b64encode(path.read_bytes()).decode("ascii")
         return f"data:{mime_type};base64,{encoded}"
 
-    def _normalize_file_input(self, url_value: str | None, path_value: str | None) -> str | None:
+    def _normalize_file_input(
+        self, url_value: str | None, path_value: str | None
+    ) -> str | None:
+        """Normalize file input by converting local file paths to Data URIs or returning URLs."""
         if url_value:
             return url_value
         if path_value:
@@ -192,7 +261,285 @@ class VeoVideo(BaseTool):
         return None
 
     def execute(self, inputs: dict[str, Any]) -> ToolResult:
-        api_key = self._get_api_key()
+        """Execute the video generation tool using the selected backend."""
+        backend = inputs.get("backend", "auto")
+        if backend == "auto":
+            if self._get_google_credentials_status():
+                backend = "google"
+            elif self._get_fal_api_key():
+                backend = "fal"
+            else:
+                return ToolResult(
+                    success=False,
+                    error="No backend credentials configured. "
+                    + self.install_instructions,
+                )
+
+        if backend == "google":
+            return self._execute_google(inputs)
+        return self._execute_fal(inputs)
+
+    def _execute_google(self, inputs: dict[str, Any]) -> ToolResult:
+        """Execute the generation request using the Google GenAI SDK backend."""
+        start = time.time()
+        try:
+            from google.genai import types
+            from PIL import Image
+            from io import BytesIO
+            import requests
+            from tools.google_credentials import get_genai_client, GOOGLE_API_TIMEOUT_MS
+
+            http_options = types.HttpOptions(timeout=GOOGLE_API_TIMEOUT_MS)
+            client = get_genai_client(http_options=http_options)
+        except ImportError as e:
+            return ToolResult(
+                success=False,
+                error=f"Failed to import required Google libraries: {e}. Run 'uv pip install google-genai pillow requests'",
+            )
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                error=f"Failed to initialize Google GenAI Client: {e}",
+            )
+
+        is_vertex = getattr(client, "vertexai", None)
+        if is_vertex is None or not isinstance(is_vertex, bool):
+            is_vertex = getattr(client, "_api_client", None) and getattr(
+                client._api_client, "vertexai", False
+            )
+
+        if is_vertex:
+            return ToolResult(
+                success=False,
+                error="Google Veo video generation via google-genai is only supported using the Gemini Developer API (API key) backend. "
+                "Please configure GEMINI_API_KEY/GOOGLE_API_KEY or use the FAL.ai backend.",
+            )
+
+        prompt = inputs["prompt"]
+        operation = inputs.get("operation", "text_to_video")
+        model_variant = inputs.get("model_variant", "veo3.1")
+        auto_fix = inputs.get("auto_fix", True)
+
+        # Map to the official preview model unless a custom model name is provided
+        if model_variant in {"veo3", "veo3/fast", "veo3.1", "veo3.1/fast"}:
+            if is_vertex:
+                model_name = "veo-3.1-generate-001"
+            else:
+                model_name = "veo-3.1-generate-preview"
+        else:
+            model_name = model_variant
+
+        duration_text = str(inputs.get("duration", "8s")).lower().replace("s", "")
+        try:
+            duration_seconds = int(duration_text)
+        except ValueError:
+            duration_seconds = 8
+
+        aspect_ratio = inputs.get("aspect_ratio", "16:9")
+        resolution = inputs.get("resolution", "1080p")
+
+        # Validate/Auto-Fix duration based on 1080p/4K or reference-to-video rules
+        needs_8s = (resolution in {"1080p", "4k"}) or (
+            operation == "reference_to_video"
+        )
+        if needs_8s and duration_seconds != 8:
+            if auto_fix:
+                import logging
+
+                logging.getLogger(__name__).warning(
+                    f"Google Veo 3.1 requires 8 seconds duration when using "
+                    f"resolution={resolution} or operation={operation}. Coercing duration to 8s."
+                )
+                duration_seconds = 8
+            else:
+                return ToolResult(
+                    success=False,
+                    error=f"Google Veo 3.1 requires duration to be 8 seconds when resolution is {resolution} or operation is {operation}.",
+                )
+
+        # Construct generation configuration
+        config = types.GenerateVideosConfig(
+            aspect_ratio=aspect_ratio,
+            duration_seconds=duration_seconds,
+            resolution=resolution,
+            number_of_videos=1,
+        )
+
+        if inputs.get("generate_audio") is not None:
+            config.generate_audio = inputs["generate_audio"]
+        if inputs.get("negative_prompt"):
+            config.negative_prompt = inputs["negative_prompt"]
+        if inputs.get("seed") is not None:
+            config.seed = inputs["seed"]
+
+        def _get_image(url: str | None, path: str | None) -> Image.Image | None:
+            if path:
+                if not os.path.exists(path):
+                    raise FileNotFoundError(f"Local input image not found: {path}")
+                return Image.open(path)
+            if url:
+                resp = requests.get(url, timeout=30)
+                resp.raise_for_status()
+                return Image.open(BytesIO(resp.content))
+            return None
+
+        def _to_sdk_image(pil_img: Image.Image) -> types.Image:
+            buf = BytesIO()
+            fmt = pil_img.format or "PNG"
+            try:
+                pil_img.save(buf, format=fmt)
+            except KeyError:
+                pil_img.save(buf, format="PNG")
+                fmt = "PNG"
+            return types.Image(
+                image_bytes=buf.getvalue(),
+                mime_type=f"image/{fmt.lower()}",
+            )
+
+        # Build execution input args
+        sdk_image = None
+        try:
+            if operation == "image_to_video":
+                image_obj = _get_image(
+                    inputs.get("image_url"), inputs.get("image_path")
+                )
+                if not image_obj:
+                    return ToolResult(
+                        success=False,
+                        error="image_to_video requires image_url or image_path",
+                    )
+                sdk_image = _to_sdk_image(image_obj)
+
+            elif operation == "first_last_frame_to_video":
+                image_obj = _get_image(
+                    inputs.get("first_frame_url"), inputs.get("first_frame_path")
+                )
+                last_image = _get_image(
+                    inputs.get("last_frame_url"), inputs.get("last_frame_path")
+                )
+                if not image_obj or not last_image:
+                    return ToolResult(
+                        success=False,
+                        error="first_last_frame_to_video requires first_frame_url/path and last_frame_url/path",
+                    )
+                config.last_frame = _to_sdk_image(last_image)
+                sdk_image = _to_sdk_image(image_obj)
+
+            elif operation == "reference_to_video":
+                ref_images = []
+                image_urls = list(inputs.get("reference_image_urls") or [])
+                image_paths = list(inputs.get("reference_image_paths") or [])
+
+                for path in image_paths:
+                    if not os.path.exists(path):
+                        raise FileNotFoundError(
+                            f"Local reference image not found: {path}"
+                        )
+                    ref_images.append(
+                        types.VideoGenerationReferenceImage(
+                            image=_to_sdk_image(Image.open(path)),
+                            reference_type=types.VideoGenerationReferenceType.ASSET,
+                        )
+                    )
+                for url in image_urls:
+                    resp = requests.get(url, timeout=30)
+                    resp.raise_for_status()
+                    ref_images.append(
+                        types.VideoGenerationReferenceImage(
+                            image=_to_sdk_image(Image.open(BytesIO(resp.content))),
+                            reference_type=types.VideoGenerationReferenceType.ASSET,
+                        )
+                    )
+                if not ref_images:
+                    return ToolResult(
+                        success=False,
+                        error="reference_to_video requires reference_image_urls or reference_image_paths",
+                    )
+                config.reference_images = ref_images
+
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                error=f"Failed to load inputs for operation {operation}: {e}",
+            )
+
+        try:
+            # Submit generation request
+            operation_handle = client.models.generate_videos(
+                model=model_name, prompt=prompt, image=sdk_image, config=config
+            )
+
+            # Poll for completion with safety timeout
+            poll_interval = 5
+            deadline = time.time() + GOOGLE_API_TIMEOUT_SECONDS
+            while not operation_handle.done:
+                if time.time() >= deadline:
+                    return ToolResult(
+                        success=False,
+                        error=f"Veo video generation timed out after {GOOGLE_API_TIMEOUT_SECONDS} seconds.",
+                    )
+                time.sleep(poll_interval)
+                operation_handle = client.operations.get(operation_handle)
+
+            if operation_handle.error:
+                return ToolResult(
+                    success=False,
+                    error=f"Veo direct API error: {operation_handle.error}",
+                )
+
+            # Download and save final file
+            response = operation_handle.response
+            if not response or not response.generated_videos:
+                return ToolResult(
+                    success=False,
+                    error="No video generation response received.",
+                )
+            video_result = response.generated_videos[0]
+            video_asset = video_result.video
+            if not video_asset:
+                return ToolResult(
+                    success=False,
+                    error="No video asset returned in the response.",
+                )
+            client.files.download(file=video_asset)
+
+            output_path = Path(inputs.get("output_path", "veo_output.mp4"))
+            output_path.parent.mkdir(parents=True, exist_ok=True)
+            video_asset.save(str(output_path))
+
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                error=f"Veo direct API generation failed: {e}",
+            )
+
+        from tools.video._shared import probe_output
+
+        probed = probe_output(output_path)
+
+        return ToolResult(
+            success=True,
+            data={
+                "provider": "veo",
+                "gateway": "google",
+                "model": model_name,
+                "prompt": prompt,
+                "operation": operation,
+                "aspect_ratio": aspect_ratio,
+                "output": str(output_path),
+                "output_path": str(output_path),
+                "format": "mp4",
+                **probed,
+            },
+            artifacts=[str(output_path)],
+            cost_usd=self.estimate_cost(inputs),
+            duration_seconds=round(time.time() - start, 2),
+            model=model_name,
+        )
+
+    def _execute_fal(self, inputs: dict[str, Any]) -> ToolResult:
+        """Execute the generation request using the fal.ai API backend."""
+        api_key = self._get_fal_api_key()
         if not api_key:
             return ToolResult(
                 success=False,
@@ -207,7 +554,11 @@ class VeoVideo(BaseTool):
         duration = inputs.get("duration", "8s")
 
         # Current fal Veo 3.1 image-guided endpoints only accept 8-second clips.
-        if variant == "veo3.1" and operation in {"reference_to_video", "first_last_frame_to_video"} and duration != "8s":
+        if (
+            variant == "veo3.1"
+            and operation in {"reference_to_video", "first_last_frame_to_video"}
+            and duration != "8s"
+        ):
             return ToolResult(
                 success=False,
                 error=(
@@ -244,7 +595,9 @@ class VeoVideo(BaseTool):
             payload["safety_tolerance"] = inputs["safety_tolerance"]
 
         if operation == "image_to_video":
-            image_value = self._normalize_file_input(inputs.get("image_url"), inputs.get("image_path"))
+            image_value = self._normalize_file_input(
+                inputs.get("image_url"), inputs.get("image_path")
+            )
             if not image_value:
                 return ToolResult(
                     success=False,
@@ -297,9 +650,16 @@ class VeoVideo(BaseTool):
             status_url = queue_data["status_url"]
             response_url = queue_data["response_url"]
 
-            # Poll until complete
+            # Poll until complete with safety timeout
+            poll_interval = 5
+            deadline = time.time() + GOOGLE_API_TIMEOUT_SECONDS
             while True:
-                time.sleep(5)
+                if time.time() >= deadline:
+                    return ToolResult(
+                        success=False,
+                        error=f"Veo video generation timed out on FAL.ai after {GOOGLE_API_TIMEOUT_SECONDS} seconds.",
+                    )
+                time.sleep(poll_interval)
                 status_resp = requests.get(status_url, headers=headers, timeout=15)
                 status_resp.raise_for_status()
                 status = status_resp.json().get("status", "UNKNOWN")
@@ -332,15 +692,23 @@ class VeoVideo(BaseTool):
         except Exception as e:
             return ToolResult(success=False, error=f"Veo video generation failed: {e}")
 
+        from tools.video._shared import probe_output
+
+        probed = probe_output(output_path)
+
         return ToolResult(
             success=True,
             data={
                 "provider": "veo",
+                "gateway": "fal",
                 "model": f"fal-ai/{model_path}",
                 "prompt": inputs["prompt"],
-                "output": str(output_path),
-                "has_audio": inputs.get("generate_audio", True),
                 "operation": operation,
+                "aspect_ratio": inputs.get("aspect_ratio", "16:9"),
+                "output": str(output_path),
+                "output_path": str(output_path),
+                "format": "mp4",
+                **probed,
             },
             artifacts=[str(output_path)],
             cost_usd=self.estimate_cost(inputs),