diff --git a/.agents/skills/ai-video-gen/SKILL.md b/.agents/skills/ai-video-gen/SKILL.md index da1f6e9d..56686803 100644 --- a/.agents/skills/ai-video-gen/SKILL.md +++ b/.agents/skills/ai-video-gen/SKILL.md @@ -1,7 +1,7 @@ --- name: ai-video-gen description: | - Generate AI videos from text prompts using multiple provider gateways. Use when: (1) Generating videos from text descriptions, (2) Creating AI-generated video clips for content production, (3) Image-to-video generation with a reference image, (4) Choosing between video generation providers (VEO, Kling, Sora, Runway, Seedance, MiniMax). Supports HeyGen API, fal.ai API, and Kling official direct API. + Generate AI videos from text prompts using multiple provider gateways. Use when: (1) Generating videos from text descriptions, (2) Creating AI-generated video clips for content production, (3) Image-to-video generation with a reference image, (4) Choosing between video generation providers (VEO, Kling, Sora, Runway, Seedance, MiniMax, Gemini Omni). Supports gateways: HeyGen API, fal.ai API, Kling official direct API, and the Gemini API (Gemini Omni Flash). allowed-tools: mcp__heygen__* metadata: openclaw: @@ -10,17 +10,22 @@ metadata: - HEYGEN_API_KEY - FAL_KEY - KLING_API_KEY + - GEMINI_API_KEY + - GOOGLE_API_KEY --- # Video Generation (Multi-Gateway) -Generate AI videos from text prompts. Supports multiple providers via three API paths: +Generate AI videos from text prompts. Supports multiple providers via four API paths: | Gateway | Env Variable | Providers | Tool | |---------|-------------|-----------|------| | **fal.ai** | `FAL_KEY` | **Seedance 2.0** (standard + fast), Kling v3/v2.1, MiniMax, VEO | `seedance_video`, `kling_video`, `minimax_video`, `veo_video` | | **HeyGen** | `HEYGEN_API_KEY` | VEO 3.1, Kling Pro, Sora v2, Runway Gen-4, Seedance Pro / Lite (1.x) | `heygen_video` | | **Kling Official** | `KLING_API_KEY` | Kling official Classic, Turbo, and basic Omni video | `kling_official_video` | +| **Gemini API** | `GEMINI_API_KEY` / `GOOGLE_API_KEY` | Gemini Omni Flash (generation + conversational editing) | `gemini_omni_video` | + +**Iterative editing — Gemini Omni.** When the brief calls for *refining an existing clip* (add/remove objects, restyle, change lighting or on-screen text) rather than regenerating, Gemini Omni Flash is the only provider in the fleet with stateful multi-turn editing. See Layer 3 `gemini-omni` for the authoritative prompting guide (reference-image tags, timecode syntax, edit-prompt rules) before writing any prompt for it. **Preferred premium default — Seedance 2.0.** When any premium gateway is configured (`FAL_KEY` → `seedance_video`, or HeyGen's Video Agent / Avatar Shots path), Seedance 2.0 is the preferred default for cinematic, trailer, and high-fidelity clip work. It is the only model in the fleet with **single-pass native synchronized audio, multi-shot generation, director-level camera control, and lip-sync from quoted dialogue**, and it ranks #1 on Artificial Analysis Elo as of early 2026. Switch off it only when the user has a specific reason (budget, provider preference, stylistic fit like VEO for photoreal landscape or Kling for specific anime look). See Layer 3 `seedance-2-0` for the authoritative prompting and parameter guide. @@ -33,8 +38,9 @@ Use whichever configured gateway best matches the user's available providers and - **HeyGen:** Set `HEYGEN_API_KEY` to access the multi-model gateway. - **fal.ai:** Set `FAL_KEY` to access Kling, MiniMax, and Veo through fal.ai. - **Kling Official:** Set `KLING_API_KEY` to access Kling's official direct API via `provider="kling_official"`. +- **Gemini API:** Set `GEMINI_API_KEY` or `GOOGLE_API_KEY` to access Gemini Omni video generation and conversational editing. -Do not describe either gateway as the default or top choice without checking the registry and current task fit first. +Do not describe any gateway as the default or top choice without checking the registry and current task fit first. fal.ai Kling (`kling_video`, `provider="kling"`) and Kling Official (`kling_official_video`, `provider="kling_official"`) are different paths. Do not reuse fal.ai queue URLs, `FAL_KEY`, or image upload behavior when the official provider is selected. diff --git a/.agents/skills/gemini-omni/SKILL.md b/.agents/skills/gemini-omni/SKILL.md new file mode 100644 index 00000000..1587c66b --- /dev/null +++ b/.agents/skills/gemini-omni/SKILL.md @@ -0,0 +1,113 @@ +--- +name: gemini-omni +description: | + Generate and conversationally edit short videos with Google Gemini Omni Flash (`gemini-omni-flash-preview`). Use when: (1) iterating on a clip with natural-language edits instead of regenerating ("make the phone invisible, keep everything else the same"), (2) generating 3-10s 720p clips with synthesized audio, rendered on-screen text, or timecoded beats, (3) binding reference images to roles with / prompt tags, (4) editing an existing uploaded video. Accessed via the `gemini_omni_video` tool using the project's GEMINI_API_KEY/GOOGLE_API_KEY — the same key as Imagen and Google TTS. +allowed-tools: Bash, Read, Write +metadata: + openclaw: + requires: + env_any: + - GEMINI_API_KEY + - GOOGLE_API_KEY +--- + +# Gemini Omni Flash (Google DeepMind) + +Gemini Omni is Google DeepMind's video generation **and editing** model family, announced at I/O 2026. The first model, **Gemini Omni Flash** (`gemini-omni-flash-preview`, developer access since June 30, 2026), generates 3-10 second clips at 720p/24fps with synthesized audio via the Gemini **Interactions API**. Its differentiator in the OpenMontage fleet is **stateful conversational editing**: each generation returns an `interaction_id`, and a follow-up call with `previous_interaction_id` edits that video in place — no other wrapped provider can refine a clip without regenerating it. + +OpenMontage wraps it as `gemini_omni_video` (native Gemini API, no gateway). It shares `GOOGLE_API_KEY`/`GEMINI_API_KEY` with `google_imagen` and `google_tts` — one key, three capabilities. Paid tier only: ~$0.10 per second of output video (billed as 5,792 output tokens/sec at $17.50/1M). + +## When to pick it (and when not) + +| Use it for | Prefer another provider for | +|---|---| +| Iterative refinement — generate, review, then edit the same clip in layers | One-shot cinematic hero clips (→ Seedance 2.0, see `seedance-2-0`) | +| Editing an existing/uploaded clip (restyle, add/remove objects, change text) | Clips longer than 10s or above 720p | +| On-screen rendered text and word-by-word text beats | Seed-reproducible generations (no seed support) | +| Reference-image-bound subjects/styles via prompt tags | First/last-frame interpolation (→ `veo_video`) | +| Timecode-scheduled multi-beat clips from one prompt | Non-English narration (English only fully supported) | + +Route through `video_selector` for generation operations. **Editing (`edit_video`) is a direct-tool operation** — call `gemini_omni_video` from the registry, because the multi-turn interaction state lives outside the selector's model. + +## Generation prompting + +Describe **scene + camera + lighting + motion + audio**. Official example: + +> Continuous, unbroken handheld shot of a fluffy tabby cat sitting on a sunny windowsill, looking out into a leafy garden. The cat's tail twitches slowly, and its ears rotate slightly toward ambient noises. Sunbeams illuminate dust motes in the air. + +- **Force a single shot** explicitly: "In a single continuous shot," / "No scene cuts." Otherwise the model may cut between scenes. +- **Negatives go in prose** — there is no `negative_prompt` parameter: "No dialogue," "No extra sound effects." +- **No sampler controls**: system instructions, temperature, top_p, and seeds are all unsupported. The prompt is the only lever. +- **Meta-prompt for quality**: "Consider micro-detail, expression and timing to create a very rich, detailed but entirely natural scene." + +### Timecode syntax + +Schedule beats with bracketed ranges or natural language — this maps directly onto OpenMontage scene-plan timings: + +``` +[0-3s] A person is walking [3-6s] They stop and turn around +``` + +> "After 3 seconds, a woman enters the scene." / "At 5s the chorus starts in the background audio." + +### Audio and on-screen text + +Audio is synthesized automatically; direct it in the prompt: "Include calm background music," "The audio is a low tinny radio broadcast in the background." Rendered text works and can be timed: + +> One word on the screen at a time: 'did, you, know, that, Omni, can, do, awesome, text?' Each word appears for 1s. + +## Reference images (`` / `` tags) + +Pass local images via `reference_image_paths` (they are sent in order), then bind them to roles **inside the prompt** with tags. `` indexes from 0 in the order supplied: + +``` +in the style of a woman is walking +``` + +``` +[0-3s] A studio fashion sequence. Starting with woman , she is +holding [3-6s] Then we see the man holding +``` + +- `` makes an image the opening frame: ` a woman is walking`. +- Use high-resolution images; describe the intended motion specifically rather than "make it move." +- Say what each image *is* (product / character / style / background reference) — the model decides usage from context. + +## Conversational editing (the differentiator) + +**Editing prompts are the opposite of generation prompts: short and surgical.** Overly descriptive edit prompts cause unintended changes. + +1. Generate the base clip (subject + scene + motion). The tool returns `interaction_id` in its result data. +2. Pass it back as `previous_interaction_id` with `operation="edit_video"` and describe **only the delta**. +3. Append **"Keep everything else the same."** to pin unmentioned elements. +4. Refine in layers — one turn for lighting, one for camera, one for action, one for audio. + +Official good/bad pairs: + +| Avoid | Instead | +|---|---| +| "In the video of the man sitting on the sofa, please add a small black cat..." | "Add a cat that jumps onto his lap, he begins to pet it. Keep everything else the same." | +| "Please remove the cell phone... and fill in the background so it looks like..." | "Make the phone invisible. Keep everything else the same." | + +Other working edit prompts: "Make this video anime" / "Put a fashionable hat on this person" / "Change the lighting to be more dramatic" / "Change the text on the sign to say 'Omni Flash'". + +**Gotcha — `store`:** editing via `previous_interaction_id` only works if the *prior* call kept the interaction server-side (`store` defaults to true in `gemini_omni_video`). Set `store=false` only for one-shot generations you will never edit. + +**Editing uploaded videos:** pass `input_video_path` instead of `previous_interaction_id`; the tool uploads it via the Files API. Unavailable in the EEA, Switzerland, and the UK (editing *generated* videos works everywhere). + +## Hard limitations (preview) + +- Output: 3-10s, 720p, 24fps, MP4 with audio; aspect ratio `16:9` or `9:16`. All output carries an invisible SynthID watermark. +- No seed, negative prompt, temperature, top_p, or system instructions. +- No video extension or first/last-frame interpolation; no voice editing. +- Audio reference inputs unsupported. Video references ≤3s are accepted by the schema but **not processed correctly** — don't rely on them. +- Multi-video prompting unsupported; may degrade output. +- English fully supported; other languages untested. +- Images of minors (EEA/CH/UK) and certain recognizable people are blocked for upload/editing. + +## Sources + +- Generation & editing guide: https://ai.google.dev/gemini-api/docs/omni +- Model card: https://ai.google.dev/gemini-api/docs/models/gemini-omni-flash +- Pricing: https://ai.google.dev/gemini-api/docs/pricing +- Announcement: https://blog.google/innovation-and-ai/models-and-research/gemini-models/gemini-omni/ diff --git a/.claude/skills/ai-video-gen/SKILL.md b/.claude/skills/ai-video-gen/SKILL.md index feb155ac..b801209b 100644 --- a/.claude/skills/ai-video-gen/SKILL.md +++ b/.claude/skills/ai-video-gen/SKILL.md @@ -1,7 +1,7 @@ --- name: ai-video-gen description: | - Generate AI videos from text prompts using multiple provider gateways. Use when: (1) Generating videos from text descriptions, (2) Creating AI-generated video clips for content production, (3) Image-to-video generation with a reference image, (4) Choosing between video generation providers (VEO, Kling, Sora, Runway, Seedance, MiniMax). Supports two gateways: HeyGen API and fal.ai API. + Generate AI videos from text prompts using multiple provider gateways. Use when: (1) Generating videos from text descriptions, (2) Creating AI-generated video clips for content production, (3) Image-to-video generation with a reference image, (4) Choosing between video generation providers (VEO, Kling, Sora, Runway, Seedance, MiniMax, Gemini Omni). Supports gateways: HeyGen API, fal.ai API, and the Gemini API (Gemini Omni Flash). allowed-tools: mcp__heygen__* metadata: openclaw: @@ -9,6 +9,8 @@ metadata: env_any: - HEYGEN_API_KEY - FAL_KEY + - GEMINI_API_KEY + - GOOGLE_API_KEY --- # Video Generation (Multi-Gateway) @@ -19,6 +21,9 @@ Generate AI videos from text prompts. Supports multiple providers via two API ga |---------|-------------|-----------|------| | **fal.ai** | `FAL_KEY` | **Seedance 2.0** (standard + fast), Kling v3/v2.1, MiniMax, VEO | `seedance_video`, `kling_video`, `minimax_video`, `veo_video` | | **HeyGen** | `HEYGEN_API_KEY` | VEO 3.1, Kling Pro, Sora v2, Runway Gen-4, Seedance Pro / Lite (1.x) | `heygen_video` | +| **Gemini API** | `GEMINI_API_KEY` / `GOOGLE_API_KEY` | Gemini Omni Flash (generation + conversational editing) | `gemini_omni_video` | + +**Iterative editing — Gemini Omni.** When the brief calls for *refining an existing clip* (add/remove objects, restyle, change lighting or on-screen text) rather than regenerating, Gemini Omni Flash is the only provider in the fleet with stateful multi-turn editing. See Layer 3 `gemini-omni` for the authoritative prompting guide (reference-image tags, timecode syntax, edit-prompt rules) before writing any prompt for it. **Preferred premium default — Seedance 2.0.** When any premium gateway is configured (`FAL_KEY` → `seedance_video`, or HeyGen's Video Agent / Avatar Shots path), Seedance 2.0 is the preferred default for cinematic, trailer, and high-fidelity clip work. It is the only model in the fleet with **single-pass native synchronized audio, multi-shot generation, director-level camera control, and lip-sync from quoted dialogue**, and it ranks #1 on Artificial Analysis Elo as of early 2026. Switch off it only when the user has a specific reason (budget, provider preference, stylistic fit like VEO for photoreal landscape or Kling for specific anime look). See Layer 3 `seedance-2-0` for the authoritative prompting and parameter guide. diff --git a/.env.example b/.env.example index 6bed6c41..fa0dc047 100644 --- a/.env.example +++ b/.env.example @@ -21,9 +21,11 @@ KLING_API_KEY= # Official Kling API key; enables video, image, TTS KLING_API_BASE_URL= # Optional endpoint override; leave blank for default https://api-singapore.klingai.com # Mainland China accounts can use https://api-beijing.klingai.com -# --- Google (one key unlocks image gen + TTS) --- -GOOGLE_API_KEY= # Google Imagen images, Google Cloud TTS (700+ voices, 50+ languages) +# --- Google (one key unlocks image gen + TTS + video) --- +GOOGLE_API_KEY= # Google Imagen images, Google Cloud TTS (700+ voices, 50+ languages), + # Gemini Omni video (generation + conversational editing, paid tier) # Get one at https://aistudio.google.com/apikey +# GEMINI_API_KEY= # Alias for GOOGLE_API_KEY (takes precedence when both are set) # Alternative to the API key: service-account JSON auth. # TTS uses Cloud Text-to-Speech; Imagen routes to Vertex AI. GOOGLE_APPLICATION_CREDENTIALS= # path to a service-account JSON key file diff --git a/AGENT_GUIDE.md b/AGENT_GUIDE.md index 1abfd8e5..feb67699 100644 --- a/AGENT_GUIDE.md +++ b/AGENT_GUIDE.md @@ -676,7 +676,7 @@ The `.agents/skills/` directory is large. When you're not coming in through a to | **Animation knowledge (generic)** | `gsap-core`, `gsap-timeline`, `gsap-plugins` (SplitText / MorphSVG / DrawSVG / MotionPath / Flip / CustomEase), `gsap-utils`, `gsap-react`, `gsap-performance`, `gsap-scrolltrigger`, `gsap-frameworks`, `framer-motion` (Disney 12 principles), `lottie-bodymovin` (Lottie export) | | **Character animation** | `character-rigging`, `svg-character-animation`, `pose-library-design`, `canvas-procedural-animation`, `character-animation-qa` | | **Image generation** | `bfl-api`, `flux-best-practices` | -| **Video generation** | `seedance-2-0` (preferred premium default — cinematic, trailer, multi-shot, synced audio, lip-sync), `ai-video-gen`, `ltx2` | +| **Video generation** | `seedance-2-0` (preferred premium default — cinematic, trailer, multi-shot, synced audio, lip-sync), `gemini-omni` (conversational video editing, reference tags, timecoded beats), `ai-video-gen`, `ltx2` | | **Audio** | `elevenlabs`, `music`, `sound-effects`, `acestep`, `text-to-speech`, `setup-api-key` | | **Avatar / lip-sync** | `avatar-video`, `heygen`, `create-video`, `faceswap`, `video-translate`, `speech-to-text`, `agents` | | **Capture** | `playwright-recording` (browser flows), `ffmpeg` (post) | diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index 830e932f..8940e690 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -33,8 +33,8 @@ Everything you need to know about every provider in OpenMontage — setup instru PEXELS_API_KEY= # Stock photos + videos PIXABAY_API_KEY= # Stock photos + videos -# GOOGLE (one key, two tools, generous free tier) -GOOGLE_API_KEY= # Google TTS + Google Imagen +# GOOGLE (one key, multiple tools, generous TTS free tier) +GOOGLE_API_KEY= # Google TTS + Imagen + Lyria music + Gemini Omni/Veo video # VOICE + MUSIC ELEVENLABS_API_KEY= # TTS, music, sound effects (10K chars/month free) @@ -292,12 +292,12 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon --- -### Google — TTS + Imagen (Shared Key) +### Google — TTS + Imagen + Music + Video (Shared Key) -> **One key, two tools.** Google Cloud TTS has 700+ voices in 50+ languages — the strongest localization option. Imagen 4 generates high-quality images. +> **One key, five tools.** Google Cloud TTS has 700+ voices in 50+ languages — the strongest localization option. Imagen 4 generates high-quality images. Google Lyria generates high-quality background music. Gemini Omni Flash supports conversational video editing, and direct Veo generation covers premium short video clips. -**Tools unlocked:** `google_tts`, `google_imagen` -**Env var:** `GOOGLE_API_KEY` +**Tools unlocked:** `google_tts`, `google_imagen`, `google_music`, `gemini_omni_video`, `veo_video` +**Env var:** `GOOGLE_API_KEY` (or `GEMINI_API_KEY` — either works; `GEMINI_API_KEY` takes precedence) #### Setup @@ -305,14 +305,14 @@ Doubao Speech 2.0 is billed by character package or usage in Volcengine. OpenMon 2. Navigate to [aistudio.google.com/apikey](https://aistudio.google.com/apikey) 3. Click **Create API Key**, select a Google Cloud project 4. Copy the key -5. Add to `.env`: `GOOGLE_API_KEY=AIza...` +5. Add to `.env`: `GOOGLE_API_KEY=AIza...` (or `GEMINI_API_KEY=AIza...`) **For TTS specifically**, you also need to enable the Text-to-Speech API: 1. Visit [console.cloud.google.com/apis/library/texttospeech.googleapis.com](https://console.cloud.google.com/apis/library/texttospeech.googleapis.com) 2. Click **Enable** 3. Make sure your API key's restrictions allow the Text-to-Speech API -**For Imagen**, enable the Generative Language API: +**For Imagen, Lyria Music, Gemini Omni video, and direct Veo video**, enable the Generative Language API: 1. Visit [console.cloud.google.com/apis/library/generativelanguage.googleapis.com](https://console.cloud.google.com/apis/library/generativelanguage.googleapis.com) 2. Click **Enable** @@ -338,7 +338,23 @@ The free tiers apply *independently* — you get 1M Standard AND 1M WaveNet AND **Free tier for Imagen:** None. Paid tier only. -**New account bonus:** Google Cloud offers **$300 in free credits** for new accounts (90-day trial), applicable to both TTS and Imagen. +#### Gemini Omni Video Pricing + +| Model | Price | Notes | +|-------|-------|-------| +| `gemini-omni-flash-preview` | ~$0.10 per second of video | Billed as 5,792 output tokens/sec of 720p video at $17.50/1M tokens | + +Generates 3–10 second clips at 720p/24fps with synthesized audio, plus stateful conversational editing (`edit_video` via `previous_interaction_id`). **Paid tier only — no free tier.** A typical 8-second clip costs ~$0.80; each edit turn generates a new clip and bills again. + +#### Google Music (Lyria) Pricing + +| Model | Price per generation request | +|-------|-----------------------------| +| `lyria-3-pro-preview` | $0.08 (flat rate, up to 184s duration) | + +**Free tier for Music:** None. Paid tier only. + +**New account bonus:** Google Cloud offers **$300 in free credits** for new accounts (90-day trial), applicable to TTS, Imagen, Music, Gemini Omni video, and direct Veo video. #### Google TTS Voice Types @@ -811,7 +827,7 @@ These tools require only FFmpeg or Python packages — no GPU, no API key. | **Pexels** | `PEXELS_API_KEY` | `pexels_image`, `pexels_video` | Free | | **Pixabay** | `PIXABAY_API_KEY` | `pixabay_image`, `pixabay_video` | Free | | **Piper** | — (install only) | `piper_tts` | Free | -| **Google** | `GOOGLE_API_KEY` | `google_tts`, `google_imagen` | Free tier + paid | +| **Google** | `GOOGLE_API_KEY` (or `GEMINI_API_KEY`) | `google_tts`, `google_imagen`, `google_music`, `gemini_omni_video`, `veo_video` | Free tier (TTS) + paid | | **ElevenLabs** | `ELEVENLABS_API_KEY` | `elevenlabs_tts`, `music_gen` | Free tier + paid | | **fal.ai** | `FAL_KEY` | `flux_image`, `recraft_image`, `kling_video`, `veo_video`, `minimax_video` | Pay-as-you-go | | **Kling Official** | `KLING_API_KEY` | `kling_official_video`, `kling_official_image`, `kling_tts`, `kling_avatar`, `kling_lip_sync` | Pay-as-you-go | @@ -834,9 +850,9 @@ How many providers cover each capability: | Capability | Cloud Providers | Local Providers | Free Options | |-----------|----------------|-----------------|--------------| | **Image Generation** | FLUX, Kling Official, Grok, Google Imagen, GPT Image 2, Recraft | Local Diffusion | Pexels, Pixabay (stock) | -| **Video Generation** | Grok, Kling Official, Kling via fal.ai, Runway, Veo, Higgsfield, MiniMax, HeyGen | WAN, Hunyuan, CogVideo, LTX | Pexels, Pixabay (stock) | +| **Video Generation** | Grok, Kling Official, Kling via fal.ai, Runway, Veo, Gemini Omni, Higgsfield, MiniMax, HeyGen | WAN, Hunyuan, CogVideo, LTX | Pexels, Pixabay (stock) | | **Text-to-Speech** | ElevenLabs, Google TTS, Kling Official, OpenAI | Piper | Piper, Google free tier, ElevenLabs free tier | -| **Music Generation** | ElevenLabs, Suno | — | ElevenLabs free tier | +| **Music Generation** | ElevenLabs, Suno, Google Lyria | — | ElevenLabs free tier | | **Post-Production** | — | FFmpeg (compose, stitch, trim, mix, enhance, grade) | All free | | **Analysis** | — | WhisperX, Scene Detect, Frame Sampler, CLIP/BLIP-2 | All free | | **Enhancement** | — | Upscale, BG Remove, Face Enhance, Face Restore | All free | diff --git a/requirements.txt b/requirements.txt index 60d2884c..1a26d96e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,6 +7,7 @@ Pillow>=10.0 numpy>=1.24 requests>=2.31 google-auth>=2.0 # service-account auth for Google TTS + Imagen (Vertex AI) +google-genai>=1.0.0 # Gemini Developer API clients for Lyria, Veo, and Gemini Omni openai>=2.44.0 # Videos API support for Sora 2 # Backlot - the living storyboard (local board server) diff --git a/setup.py b/setup.py index d074c788..38de3eb8 100644 --- a/setup.py +++ b/setup.py @@ -13,6 +13,7 @@ setup( "python-dotenv>=1.0", "Pillow>=10.0", "requests>=2.31", + "google-genai>=1.0.0", "openai>=2.44.0", ], ) diff --git a/tests/contracts/test_character_animation_pipeline.py b/tests/contracts/test_character_animation_pipeline.py index 4c765936..0227ebd3 100644 --- a/tests/contracts/test_character_animation_pipeline.py +++ b/tests/contracts/test_character_animation_pipeline.py @@ -10,6 +10,7 @@ sys.path.insert(0, str(PROJECT_ROOT)) from lib.pipeline_loader import get_required_tools, get_stage_order, load_pipeline from schemas.artifacts import ARTIFACT_NAMES, validate_artifact +from tools.base_tool import ToolResult from tools.character.character_animation import ( ActionTimelineCompiler, CharacterAnimationReviewer, @@ -235,12 +236,7 @@ def test_character_style_is_normalized_for_schema(tmp_path): } -def test_character_renderer_can_handoff_to_video_compose(tmp_path): - hyperframes = HyperFramesCompose() - runtime = hyperframes._runtime_check() - if not runtime["runtime_available"]: - pytest.skip("HyperFrames runtime is required for character render handoff") - +def test_character_renderer_can_handoff_to_video_compose(tmp_path, monkeypatch): character_design = CharacterSpecGenerator().execute( {"characters": [{"id": "mouse_lead", "role": "lead", "body_type": "mouse with tail"}]} ).data["character_design"] @@ -286,6 +282,21 @@ def test_character_renderer_can_handoff_to_video_compose(tmp_path): assert Path(render_result.data["composition_path"]).exists() output_path = tmp_path / "renders" / "final.mp4" + captured_handoff = {} + + def fake_hyperframes_execute(self, inputs): + captured_handoff.update(inputs) + Path(inputs["output_path"]).write_bytes(b"fake mp4") + return ToolResult(success=True, data={"output": inputs["output_path"]}) + + monkeypatch.setattr(VideoCompose, "_hyperframes_available", lambda self: True) + monkeypatch.setattr( + VideoCompose, + "_run_final_review", + lambda self, *args, **kwargs: {"status": "pass", "issues_found": []}, + ) + monkeypatch.setattr(HyperFramesCompose, "execute", fake_hyperframes_execute) + compose_result = VideoCompose().execute( { "operation": "render", @@ -301,3 +312,8 @@ def test_character_renderer_can_handoff_to_video_compose(tmp_path): assert compose_result.success, compose_result.error assert output_path.exists() + assert captured_handoff["operation"] == "render" + assert captured_handoff["workspace_path"] == render_result.data["hyperframes_workspace"] + assert captured_handoff["output_path"] == str(output_path) + assert captured_handoff["edit_decisions"]["render_runtime"] == "hyperframes" + assert captured_handoff["skip_contrast"] is True diff --git a/tests/contracts/test_phase3_contracts.py b/tests/contracts/test_phase3_contracts.py index 19cecf94..34811155 100644 --- a/tests/contracts/test_phase3_contracts.py +++ b/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() @@ -174,6 +691,7 @@ class TestCapabilityMetadata: # ---- Animated Explainer Pipeline ---- + class TestAnimatedExplainerManifest: def test_loads(self): manifest = load_pipeline("animated-explainer") @@ -183,7 +701,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): @@ -198,7 +725,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") @@ -222,6 +751,7 @@ class TestAnimatedExplainerManifest: # ---- Style Playbooks ---- + class TestStylePlaybooks: def test_all_listed(self): playbooks = list_playbooks() @@ -229,13 +759,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 @@ -246,7 +780,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"] @@ -255,7 +791,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"] @@ -269,60 +807,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 @@ -331,6 +884,7 @@ class TestSkillsExist: # ---- Remotion Scaffold ---- + class TestRemotionScaffold: REMOTION_DIR = PROJECT_ROOT / "remotion-composer" @@ -355,17 +909,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 diff --git a/tests/tools/test_gemini_omni_video.py b/tests/tools/test_gemini_omni_video.py new file mode 100644 index 00000000..6f2b794c --- /dev/null +++ b/tests/tools/test_gemini_omni_video.py @@ -0,0 +1,285 @@ +"""Regression coverage for first-class Gemini Omni provider discovery and execution.""" + +from __future__ import annotations + +import base64 +import json +import sys +import types +from pathlib import Path + +import pytest + +from tools.base_tool import ToolStatus + + +class FakeResponse: + def __init__(self, json_data=None, content=b"", ok=True, status_code=200, headers=None, text=""): + self._json = json_data + self.content = content + self.ok = ok + self.status_code = status_code + self.headers = headers or {} + self.text = text or (json.dumps(json_data) if json_data is not None else "") + + def json(self): + return self._json + + def raise_for_status(self): + if not self.ok: + raise RuntimeError(f"HTTP {self.status_code}") + + +def _install_fake_requests(monkeypatch, post_responses, get_responses): + """Inject a fake requests module; returns the recorded calls.""" + calls = {"post": [], "get": []} + + fake = types.ModuleType("requests") + + def fake_post(url, headers=None, json=None, data=None, timeout=None, params=None): + calls["post"].append({"url": url, "headers": headers, "json": json, "data": data}) + return post_responses.pop(0) + + def fake_get(url, headers=None, timeout=None, params=None): + calls["get"].append({"url": url, "headers": headers, "params": params}) + return get_responses.pop(0) + + fake.post = fake_post + fake.get = fake_get + monkeypatch.setitem(sys.modules, "requests", fake) + return calls + + +@pytest.fixture() +def gemini_env(monkeypatch): + monkeypatch.delenv("GOOGLE_API_KEY", raising=False) + monkeypatch.setenv("GEMINI_API_KEY", "test-gemini-key") + + +def test_gemini_omni_is_discovered_as_video_provider(): + from tools.tool_registry import ToolRegistry + + registry = ToolRegistry() + registry.discover() + + tool = registry.get("gemini_omni_video") + assert tool is not None + assert tool.provider == "gemini_omni" + assert tool.capability == "video_generation" + # Ranking + selector-filter contract: without these the scorer buries the + # editing capability and the selector drops it from image_to_video routing. + assert tool.quality_score == 0.85 + assert tool.supports["image_to_video"] is True + assert tool.supports["reference_to_video"] is True + assert tool.supports["conversational_editing"] is True + assert "gemini-omni" in tool.agent_skills + + +def test_gemini_omni_is_routed_by_video_selector(): + from tools.video.video_selector import VideoSelector + + provider_names = [t.name for t in VideoSelector()._providers()] + assert "gemini_omni_video" in provider_names + + +def test_gemini_omni_status_tracks_google_api_keys(monkeypatch): + from tools.video.gemini_omni_video import GeminiOmniVideo + + monkeypatch.delenv("GEMINI_API_KEY", raising=False) + monkeypatch.delenv("GOOGLE_API_KEY", raising=False) + assert GeminiOmniVideo().get_status() == ToolStatus.UNAVAILABLE + + monkeypatch.setenv("GOOGLE_API_KEY", "test-google-key") + assert GeminiOmniVideo().get_status() == ToolStatus.AVAILABLE + + +def test_gemini_omni_cost_estimate_clamps_duration_hint(gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + tool = GeminiOmniVideo() + assert tool.estimate_cost({"prompt": "x"}) == pytest.approx(0.80) + assert tool.estimate_cost({"prompt": "x", "duration": "5s"}) == pytest.approx(0.50) + assert tool.estimate_cost({"prompt": "x", "duration": "30"}) == pytest.approx(1.00) + + +def test_gemini_omni_text_to_video_via_uri_delivery(monkeypatch, tmp_path, gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + calls = _install_fake_requests( + monkeypatch, + post_responses=[ + FakeResponse({"id": "int_123", "output_video": {"uri": "files/vid-123"}}), + ], + get_responses=[ + FakeResponse({"state": "ACTIVE"}), + FakeResponse(content=b"fake omni mp4"), + ], + ) + + output_path = tmp_path / "clip.mp4" + result = GeminiOmniVideo().execute( + { + "prompt": "A marble rolling on a track, single continuous shot.", + "aspect_ratio": "9:16", + "output_path": str(output_path), + } + ) + + assert result.success, result.error + assert output_path.read_bytes() == b"fake omni mp4" + assert result.data["interaction_id"] == "int_123" + assert result.data["editable"] is True + + payload = calls["post"][0]["json"] + assert payload["model"] == "gemini-omni-flash-preview" + assert payload["input"] == "A marble rolling on a track, single continuous shot." + assert payload["response_format"] == {"type": "video", "aspect_ratio": "9:16", "delivery": "uri"} + assert calls["post"][0]["headers"]["x-goog-api-key"] == "test-gemini-key" + assert calls["get"][1]["url"].endswith("files/vid-123:download") + assert calls["get"][1]["params"] == {"alt": "media"} + + +def test_gemini_omni_uri_delivery_handles_full_download_url(monkeypatch, tmp_path, gemini_env): + """The API may return a full .../files/:download?alt=media URL, not just files/.""" + from tools.video.gemini_omni_video import GeminiOmniVideo + + full_url = ( + "https://generativelanguage.googleapis.com/v1beta/files/vid-456:download?alt=media" + ) + calls = _install_fake_requests( + monkeypatch, + post_responses=[FakeResponse({"id": "int_5", "output_video": {"uri": full_url}})], + get_responses=[ + FakeResponse({"state": "ACTIVE"}), + FakeResponse(content=b"full url mp4"), + ], + ) + + output_path = tmp_path / "full.mp4" + result = GeminiOmniVideo().execute({"prompt": "A sunset.", "output_path": str(output_path)}) + + assert result.success, result.error + assert output_path.read_bytes() == b"full url mp4" + assert calls["get"][0]["url"].endswith("/files/vid-456") + assert calls["get"][1]["url"].endswith("/files/vid-456:download") + + +@pytest.mark.parametrize( + "uri", + [ + "files/vid-456", + "files/vid-456/", + "v1beta/files/vid-456", + "https://generativelanguage.googleapis.com/v1beta/files/vid-456", + "https://generativelanguage.googleapis.com/v1beta/files/vid-456:download?alt=media", + ], +) +def test_gemini_omni_file_id_extraction_covers_documented_uri_shapes(uri): + from tools.video.gemini_omni_video import GeminiOmniVideo + + assert GeminiOmniVideo._file_id_from_uri(uri) == "vid-456" + + +def test_gemini_omni_inline_data_response_is_handled(monkeypatch, tmp_path, gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + inline = base64.b64encode(b"inline mp4").decode("ascii") + calls = _install_fake_requests( + monkeypatch, + post_responses=[FakeResponse({"id": "int_9", "output_video": {"data": inline}})], + get_responses=[], + ) + + output_path = tmp_path / "inline.mp4" + result = GeminiOmniVideo().execute({"prompt": "A sunset.", "output_path": str(output_path)}) + + assert result.success, result.error + assert output_path.read_bytes() == b"inline mp4" + assert calls["get"] == [] + + +def test_gemini_omni_edit_turn_sends_previous_interaction_id(monkeypatch, tmp_path, gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + inline = base64.b64encode(b"edited mp4").decode("ascii") + calls = _install_fake_requests( + monkeypatch, + post_responses=[FakeResponse({"id": "int_2", "output_video": {"data": inline}})], + get_responses=[], + ) + + result = GeminiOmniVideo().execute( + { + "prompt": "Make the violin invisible. Keep everything else the same.", + "operation": "edit_video", + "previous_interaction_id": "int_1", + "output_path": str(tmp_path / "edit.mp4"), + } + ) + + assert result.success, result.error + assert calls["post"][0]["json"]["previous_interaction_id"] == "int_1" + + +def test_gemini_omni_edit_without_source_is_rejected(gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + result = GeminiOmniVideo().execute({"prompt": "Make it anime", "operation": "edit_video"}) + assert not result.success + assert "previous_interaction_id" in result.error + + +def test_gemini_omni_image_to_video_sends_typed_parts(monkeypatch, tmp_path, gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + ref = tmp_path / "cat.png" + ref.write_bytes(b"png bytes") + inline = base64.b64encode(b"cat mp4").decode("ascii") + calls = _install_fake_requests( + monkeypatch, + post_responses=[FakeResponse({"id": "int_3", "output_video": {"data": inline}})], + get_responses=[], + ) + + result = GeminiOmniVideo().execute( + { + "prompt": "A cat playfully batting at yarn.", + "operation": "image_to_video", + "reference_image_path": str(ref), + "output_path": str(tmp_path / "cat.mp4"), + } + ) + + assert result.success, result.error + parts = calls["post"][0]["json"]["input"] + assert parts[0]["type"] == "image" + assert parts[0]["mime_type"] == "image/png" + assert parts[0]["data"] == base64.b64encode(b"png bytes").decode("ascii") + assert parts[-1] == {"type": "text", "text": "A cat playfully batting at yarn."} + + +def test_gemini_omni_image_to_video_requires_reference(gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + result = GeminiOmniVideo().execute({"prompt": "x", "operation": "image_to_video"}) + assert not result.success + assert "reference_image_path" in result.error + + +def test_gemini_omni_store_false_marks_result_not_editable(monkeypatch, tmp_path, gemini_env): + from tools.video.gemini_omni_video import GeminiOmniVideo + + inline = base64.b64encode(b"oneshot mp4").decode("ascii") + calls = _install_fake_requests( + monkeypatch, + post_responses=[FakeResponse({"id": "int_4", "output_video": {"data": inline}})], + get_responses=[], + ) + + result = GeminiOmniVideo().execute( + {"prompt": "A sunset.", "store": False, "output_path": str(tmp_path / "s.mp4")} + ) + + assert result.success, result.error + assert result.data["editable"] is False + assert calls["post"][0]["json"]["store"] is False diff --git a/tools/audio/google_music.py b/tools/audio/google_music.py new file mode 100644 index 00000000..defb56fd --- /dev/null +++ b/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, + ) diff --git a/tools/audio/google_tts.py b/tools/audio/google_tts.py index b6218a1c..424e4b30 100644 --- a/tools/audio/google_tts.py +++ b/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("{stripped}" + ssml = ( + stripped + if stripped.startswith("{stripped}" + ) 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( diff --git a/tools/google_credentials.py b/tools/google_credentials.py index 6336bc24..71158c73 100644 --- a/tools/google_credentials.py +++ b/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 diff --git a/tools/graphics/google_imagen.py b/tools/graphics/google_imagen.py index 72eebba0..a222374c 100644 --- a/tools/graphics/google_imagen.py +++ b/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) diff --git a/tools/video/gemini_omni_video.py b/tools/video/gemini_omni_video.py new file mode 100644 index 00000000..097c6795 --- /dev/null +++ b/tools/video/gemini_omni_video.py @@ -0,0 +1,452 @@ +"""Google Gemini Omni Flash video generation and conversational editing. + +Calls the Gemini Interactions API (``POST /v1beta/interactions``) directly with +the project's Google API key — the same key that unlocks Imagen images and +Cloud TTS. Gemini Omni Flash generates 3-10 second 720p/24fps clips with +synthesized audio, and is the only provider in the fleet with stateful +conversational editing: pass ``previous_interaction_id`` and describe only the +delta ("Make the violin invisible. Keep everything else the same."). + +Reference images bind to roles via inline prompt tags (````, +````) and beats can be scheduled with timecode syntax +(``[0-3s] ... [3-6s] ...``). See the Layer 3 skill ``gemini-omni`` for the +authoritative prompting guide — read it before writing prompts. +""" + +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, +) + +_BASE_URL = "https://generativelanguage.googleapis.com/v1beta" +_UPLOAD_URL = "https://generativelanguage.googleapis.com/upload/v1beta/files" +_DEFAULT_MODEL = "gemini-omni-flash-preview" +# Billed at 5,792 output tokens per second of 720p video, $17.50/1M tokens +# (ai.google.dev/gemini-api/docs/pricing) — effectively ~$0.10 per second. +_COST_PER_SECOND = 0.10 +_DEFAULT_DURATION_SECONDS = 8 +_POLL_INTERVAL_SECONDS = 5 +_MAX_POLL_SECONDS = 900 + + +class GeminiOmniVideo(BaseTool): + name = "gemini_omni_video" + version = "0.1.0" + tier = ToolTier.GENERATE + capability = "video_generation" + provider = "gemini_omni" + stability = ToolStability.EXPERIMENTAL + execution_mode = ExecutionMode.SYNC + determinism = Determinism.STOCHASTIC + runtime = ToolRuntime.API + + dependencies = [] + install_instructions = ( + "Set GEMINI_API_KEY or GOOGLE_API_KEY to a Google AI Studio API key.\n" + " Get one at https://aistudio.google.com/apikey\n" + " Gemini Omni Flash is paid-tier only (no free tier); ~$0.10 per second of video." + ) + agent_skills = ["gemini-omni", "ai-video-gen"] + + capabilities = ["text_to_video", "image_to_video", "reference_to_video", "edit_video"] + supports = { + "text_to_video": True, + "image_to_video": True, + "reference_to_video": True, + "edit_video": True, + "conversational_editing": True, + "native_audio": True, + "text_rendering": True, + "timecode_control": True, + # Preview limitations — no sampler controls of any kind. + "seed": False, + "negative_prompt": False, + "first_last_frame_to_video": False, + } + best_for = [ + "iterative natural-language video editing (edit a clip without regenerating it)", + "reference-image-driven clips via / prompt tags", + "fast 3-10s clips with synced audio, rendered text, and timecoded beats from one Google key", + ] + not_good_for = [ + "clips longer than 10 seconds or above 720p", + "seed-reproducible output or negative-prompt control", + "offline generation", + ] + fallback_tools = ["veo_video", "sora_video", "kling_video", "minimax_video"] + # Conversational editing + native audio are unique in the fleet, but preview + # output is capped at 720p/10s — below seedance (0.95) and grok/runway (0.9) + # on raw generation fidelity. Without a quality_score the scorer would only + # count supports/stability flags and bury the editing capability entirely. + # See lib/scoring.py. + quality_score = 0.85 + + input_schema = { + "type": "object", + "required": ["prompt"], + "properties": { + "prompt": { + "type": "string", + "description": ( + "Video description, or for edit_video the change to apply. " + "Supports / tags and [0-3s] timecodes — " + "see the gemini-omni skill." + ), + }, + "operation": { + "type": "string", + "enum": ["text_to_video", "image_to_video", "reference_to_video", "edit_video"], + "default": "text_to_video", + }, + "aspect_ratio": { + "type": "string", + "enum": ["16:9", "9:16"], + "default": "16:9", + }, + "duration": { + "type": "string", + "description": ( + "Duration hint in seconds (3-10). The model chooses the actual length; " + "this only shapes the prompt-independent cost estimate." + ), + }, + "reference_image_path": { + "type": "string", + "description": "Local reference image (jpg/png) for image_to_video.", + }, + "reference_image_paths": { + "type": "array", + "items": {"type": "string"}, + "description": "Local reference images, bound in the prompt as , , ...", + }, + "previous_interaction_id": { + "type": "string", + "description": ( + "Interaction id from a prior gemini_omni_video result — edits that video " + "in place (edit_video). Requires the prior call to have used store=true." + ), + }, + "input_video_path": { + "type": "string", + "description": ( + "Local video to edit (edit_video). Uploaded via the Files API. " + "Editing uploaded videos is unavailable in the EEA, Switzerland, and the UK." + ), + }, + "store": { + "type": "boolean", + "default": True, + "description": ( + "Keep the interaction server-side so the result can be edited in later turns " + "via previous_interaction_id. Set false only for one-shot generations." + ), + }, + "output_path": {"type": "string"}, + }, + } + + resource_profile = ResourceProfile( + cpu_cores=1, ram_mb=512, vram_mb=0, disk_mb=500, network_required=True + ) + retry_policy = RetryPolicy(max_retries=1, retryable_errors=["rate_limit", "timeout"]) + idempotency_key_fields = ["prompt", "operation", "aspect_ratio", "previous_interaction_id"] + side_effects = [ + "writes video file to output_path", + "calls the Gemini Interactions API", + "stores the interaction server-side when store=true (enables later edits)", + ] + user_visible_verification = [ + "Watch generated clip for visual quality, motion, and prompt adherence", + "Listen for synthesized audio quality and any requested dialogue/music", + "After an edit turn, confirm unmentioned elements were preserved", + ] + + @staticmethod + def _get_api_key() -> str | None: + return os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY") + + def get_status(self) -> ToolStatus: + if self._get_api_key(): + return ToolStatus.AVAILABLE + return ToolStatus.UNAVAILABLE + + @staticmethod + def _duration_hint(inputs: dict[str, Any]) -> int: + raw = str(inputs.get("duration") or _DEFAULT_DURATION_SECONDS).strip().lower() + raw = raw[:-1] if raw.endswith("s") else raw + try: + seconds = int(float(raw)) + except ValueError: + seconds = _DEFAULT_DURATION_SECONDS + return max(3, min(10, seconds)) + + def estimate_cost(self, inputs: dict[str, Any]) -> float: + return _COST_PER_SECOND * self._duration_hint(inputs) + + def estimate_runtime(self, inputs: dict[str, Any]) -> float: + return 180.0 + + @staticmethod + def _image_part(path_str: str) -> dict[str, Any]: + path = Path(path_str) + if not path.exists(): + raise FileNotFoundError(f"Reference image not found: {path}") + mime_type, _ = mimetypes.guess_type(path.name) + if not mime_type or not mime_type.startswith("image/"): + mime_type = "image/png" + return { + "type": "image", + "data": base64.b64encode(path.read_bytes()).decode("ascii"), + "mime_type": mime_type, + } + + def _upload_video_file(self, requests_mod: Any, api_key: str, path_str: str) -> str: + """Upload a local video via the Files API (resumable) and return its URI.""" + path = Path(path_str) + if not path.exists(): + raise FileNotFoundError(f"Input video not found: {path}") + mime_type, _ = mimetypes.guess_type(path.name) + if not mime_type or not mime_type.startswith("video/"): + mime_type = "video/mp4" + video_bytes = path.read_bytes() + + start_resp = requests_mod.post( + _UPLOAD_URL, + headers={ + "x-goog-api-key": api_key, + "X-Goog-Upload-Protocol": "resumable", + "X-Goog-Upload-Command": "start", + "X-Goog-Upload-Header-Content-Length": str(len(video_bytes)), + "X-Goog-Upload-Header-Content-Type": mime_type, + "Content-Type": "application/json", + }, + json={"file": {"display_name": path.name}}, + timeout=30, + ) + start_resp.raise_for_status() + upload_url = start_resp.headers.get("X-Goog-Upload-URL") + if not upload_url: + raise RuntimeError("Files API did not return an upload URL") + + upload_resp = requests_mod.post( + upload_url, + headers={ + "X-Goog-Upload-Command": "upload, finalize", + "X-Goog-Upload-Offset": "0", + "Content-Length": str(len(video_bytes)), + }, + data=video_bytes, + timeout=300, + ) + upload_resp.raise_for_status() + file_info = upload_resp.json().get("file", {}) + + # Wait until the uploaded video is processed before referencing it. + deadline = time.time() + _MAX_POLL_SECONDS + while str(file_info.get("state", "")).upper() == "PROCESSING": + if time.time() > deadline: + raise TimeoutError("Uploaded video did not finish processing in time") + time.sleep(_POLL_INTERVAL_SECONDS) + status_resp = requests_mod.get( + f"{_BASE_URL}/{file_info.get('name')}", + headers={"x-goog-api-key": api_key}, + timeout=15, + ) + status_resp.raise_for_status() + file_info = status_resp.json() + if str(file_info.get("state", "")).upper() == "FAILED": + raise RuntimeError("Files API failed to process the uploaded video") + + uri = file_info.get("uri") + if not uri: + raise RuntimeError(f"Files API response missing uri: {file_info}") + return uri + + @staticmethod + def _extract_output_video(data: dict[str, Any]) -> dict[str, Any] | None: + """Find the output video payload ({'data': b64} or {'uri': files/...}).""" + for key in ("output_video", "outputVideo"): + video = data.get(key) + if isinstance(video, dict) and (video.get("data") or video.get("uri")): + return video + # REST responses may also carry the video inside steps[].content[]. + for step in data.get("steps") or []: + for item in step.get("content") or []: + if isinstance(item, dict) and (item.get("data") or item.get("uri")): + if "video" in str(item.get("type", "")).lower() or item.get("mime_type", "").startswith("video/"): + return item + if item.get("data") or str(item.get("uri", "")).startswith("files/"): + return item + return None + + @staticmethod + def _file_id_from_uri(uri: str) -> str: + """Extract the bare file id from any documented URI shape. + + The API may return ``files/``, a full + ``https://.../v1beta/files/`` resource URI, or a ready-made + download URL ``.../files/:download?alt=media``. Polling and + download both need just ````. + """ + path = uri.split("?", 1)[0].rstrip("/") + marker = "files/" + idx = path.rfind(marker) + tail = path[idx + len(marker):] if idx != -1 else path.split("/")[-1] + return tail.split(":", 1)[0] + + def _download_via_uri(self, requests_mod: Any, api_key: str, uri: str) -> bytes: + """Poll a Files API entry until ACTIVE, then download its bytes.""" + file_id = self._file_id_from_uri(uri) + headers = {"x-goog-api-key": api_key} + deadline = time.time() + _MAX_POLL_SECONDS + while True: + status_resp = requests_mod.get( + f"{_BASE_URL}/files/{file_id}", headers=headers, timeout=15 + ) + status_resp.raise_for_status() + state = str(status_resp.json().get("state", "")).upper() + if state == "ACTIVE": + break + if state == "FAILED": + raise RuntimeError("Gemini Omni video generation failed during processing") + if time.time() > deadline: + raise TimeoutError("Timed out waiting for Gemini Omni video to become ACTIVE") + time.sleep(_POLL_INTERVAL_SECONDS) + + download_resp = requests_mod.get( + f"{_BASE_URL}/files/{file_id}:download", + params={"alt": "media"}, + headers=headers, + timeout=300, + ) + download_resp.raise_for_status() + return download_resp.content + + def execute(self, inputs: dict[str, Any]) -> ToolResult: + api_key = self._get_api_key() + if not api_key: + return ToolResult( + success=False, + error="GEMINI_API_KEY / GOOGLE_API_KEY not set. " + self.install_instructions, + ) + + import requests + + start = time.time() + operation = inputs.get("operation", "text_to_video") + prompt = str(inputs["prompt"]).strip() + aspect_ratio = inputs.get("aspect_ratio", "16:9") + previous_interaction_id = inputs.get("previous_interaction_id") + + if operation == "edit_video" and not previous_interaction_id and not inputs.get("input_video_path"): + return ToolResult( + success=False, + error="edit_video requires previous_interaction_id (edit a generated clip) or input_video_path (edit an uploaded clip)", + ) + + reference_paths = list(inputs.get("reference_image_paths") or []) + if inputs.get("reference_image_path"): + reference_paths.insert(0, inputs["reference_image_path"]) + if operation in {"image_to_video", "reference_to_video"} and not reference_paths: + return ToolResult( + success=False, + error=f"{operation} requires reference_image_path or reference_image_paths", + ) + + try: + parts: list[dict[str, Any]] = [self._image_part(p) for p in reference_paths] + if inputs.get("input_video_path"): + video_uri = self._upload_video_file(requests, api_key, inputs["input_video_path"]) + parts.append({"type": "document", "uri": video_uri}) + except Exception as e: + return ToolResult(success=False, error=f"Gemini Omni input preparation failed: {e}") + + payload: dict[str, Any] = { + "model": _DEFAULT_MODEL, + # Plain string for text-only turns (the documented minimal form), + # typed parts when images or an uploaded video ride along. + "input": prompt if not parts else parts + [{"type": "text", "text": prompt}], + # uri delivery avoids the ~4MB inline-payload ceiling; inline data in + # the response is still handled below if the API returns it anyway. + "response_format": { + "type": "video", + "aspect_ratio": aspect_ratio, + "delivery": "uri", + }, + } + if previous_interaction_id: + payload["previous_interaction_id"] = previous_interaction_id + if inputs.get("store") is False: + payload["store"] = False + + try: + resp = requests.post( + f"{_BASE_URL}/interactions", + headers={"x-goog-api-key": api_key, "Content-Type": "application/json"}, + json=payload, + timeout=600, + ) + if not resp.ok: + detail = resp.text[:1000] + return ToolResult( + success=False, + error=f"Gemini Omni interaction failed ({resp.status_code}): {detail}", + ) + data = resp.json() + + interaction_id = data.get("id") + video = self._extract_output_video(data) + if not video: + return ToolResult( + success=False, + error=f"Gemini Omni response did not include an output video: {str(data)[:1000]}", + ) + + if video.get("data"): + video_bytes = base64.b64decode(video["data"]) + else: + video_bytes = self._download_via_uri(requests, api_key, str(video["uri"])) + + output_path = Path(inputs.get("output_path", "gemini_omni_output.mp4")) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(video_bytes) + except Exception as e: + return ToolResult(success=False, error=f"Gemini Omni video generation failed: {e}") + + editable = inputs.get("store") is not False + return ToolResult( + success=True, + data={ + "provider": self.provider, + "model": _DEFAULT_MODEL, + "prompt": prompt, + "operation": operation, + "output": str(output_path), + "aspect_ratio": aspect_ratio, + "has_audio": True, + # Feed this back as previous_interaction_id to edit this clip. + "interaction_id": interaction_id, + "editable": editable, + }, + artifacts=[str(output_path)], + cost_usd=self.estimate_cost(inputs), + duration_seconds=round(time.time() - start, 2), + model=_DEFAULT_MODEL, + ) diff --git a/tools/video/sora_video.py b/tools/video/sora_video.py index c2d4a5aa..6c9060c6 100644 --- a/tools/video/sora_video.py +++ b/tools/video/sora_video.py @@ -65,7 +65,7 @@ class SoraVideo(BaseTool): "4, 8, or 12 second social-video clips that OpenMontage can stitch and compose", ] not_good_for = ["offline generation", "long continuous scenes", "projects without Sora API access"] - fallback_tools = ["veo_video", "seedance_video", "kling_video", "minimax_video"] + fallback_tools = ["veo_video", "gemini_omni_video", "seedance_video", "kling_video", "minimax_video"] input_schema = { "type": "object", diff --git a/tools/video/veo_video.py b/tools/video/veo_video.py index 522128d5..de4efea2 100644 --- a/tools/video/veo_video.py +++ b/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, @@ -62,28 +73,38 @@ class VeoVideo(BaseTool): "ambient sound and music generation built in", ] not_good_for = ["budget projects", "offline generation", "quick iteration"] - fallback_tools = ["kling_video", "minimax_video", "wan_video"] + fallback_tools = ["gemini_omni_video", "kling_video", "minimax_video", "wan_video"] input_schema = { "type": "object", "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), diff --git a/tools/video/video_selector.py b/tools/video/video_selector.py index 350d6738..57889bb8 100644 --- a/tools/video/video_selector.py +++ b/tools/video/video_selector.py @@ -20,7 +20,7 @@ class VideoSelector(BaseTool): provider = "selector" stability = ToolStability.BETA runtime = ToolRuntime.HYBRID - agent_skills = ["ai-video-gen", "create-video", "ltx2"] + agent_skills = ["ai-video-gen", "create-video", "ltx2", "gemini-omni"] # Operations that REQUIRE motion: an image-only tool (image_selector) is not # an acceptable last-resort fallback for these, so fallback_tools_for() drops it.