diff --git a/plugins/video_gen/google/__init__.py b/plugins/video_gen/google/__init__.py
new file mode 100644
index 000000000000..bbddcc8d4627
--- /dev/null
+++ b/plugins/video_gen/google/__init__.py
@@ -0,0 +1,406 @@
+"""Google Veo video generation plugin.
+
+Registers a single ``video_generate`` tool that wraps the Gemini API's
+``models/{veo}:predictLongRunning`` endpoint:
+
+  1. Submit prompt → get an operation name (long-running op)
+  2. Poll the op until ``done: true``
+  3. Follow the signed download URI to fetch the MP4 bytes
+  4. Save under ``$HERMES_HOME/cache/videos/`` and return the path
+
+Auth: ``GEMINI_API_KEY`` (preferred) or ``GOOGLE_API_KEY``. Veo requires
+Tier 1 (paid) on the Gemini API project; free-tier sees zero quota.
+
+Selection precedence (first hit wins):
+1. ``GOOGLE_VIDEO_MODEL`` env var
+2. ``video_gen.google.model`` in ``config.yaml``
+3. :data:`DEFAULT_MODEL` (``veo-3.0-fast-generate-001``)
+"""
+
+from __future__ import annotations
+
+import datetime
+import logging
+import os
+import time
+import uuid
+from pathlib import Path
+from typing import Any, Dict, Optional, Tuple
+
+import requests
+
+from tools.registry import tool_error, tool_result
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Model catalog
+# ---------------------------------------------------------------------------
+
+_MODELS: Dict[str, Dict[str, Any]] = {
+    "veo-3.0-fast-generate-001": {
+        "display": "Veo 3 Fast",
+        "speed": "~30-60s for 4s video",
+        "strengths": "Fast iteration, good quality",
+    },
+    "veo-3.0-generate-001": {
+        "display": "Veo 3",
+        "speed": "~60-120s for 8s video",
+        "strengths": "Default Veo 3 — balanced quality/cost",
+    },
+    "veo-2.0-generate-001": {
+        "display": "Veo 2",
+        "speed": "~60-120s",
+        "strengths": "Older generation, sometimes cheaper",
+    },
+    "veo-3.1-generate-preview": {
+        "display": "Veo 3.1 (preview)",
+        "speed": "~90s",
+        "strengths": "Newer preview, improved temporal coherence",
+    },
+    "veo-3.1-fast-generate-preview": {
+        "display": "Veo 3.1 Fast (preview)",
+        "speed": "~30-60s",
+        "strengths": "Newer preview, fast tier",
+    },
+    "veo-3.1-lite-generate-preview": {
+        "display": "Veo 3.1 Lite (preview)",
+        "speed": "~20-40s",
+        "strengths": "Newer preview, cheapest tier",
+    },
+}
+
+DEFAULT_MODEL = "veo-3.0-fast-generate-001"
+
+API_BASE = "https://generativelanguage.googleapis.com/v1beta"
+
+VALID_ASPECT_RATIOS: Tuple[str, ...] = ("16:9", "9:16", "1:1")
+DEFAULT_ASPECT_RATIO = "16:9"
+
+# Cap polling so the tool never blocks forever even if Google hangs the op.
+POLL_TIMEOUT_SECONDS = 600
+POLL_INTERVAL_SECONDS = 5
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _resolve_api_key() -> str:
+    return (
+        os.getenv("GEMINI_API_KEY", "").strip()
+        or os.getenv("GOOGLE_API_KEY", "").strip()
+    )
+
+
+def _load_video_config() -> Dict[str, Any]:
+    """Read ``video_gen.google`` from config.yaml (returns {} on failure)."""
+    try:
+        from hermes_cli.config import load_config
+
+        cfg = load_config()
+        section = cfg.get("video_gen") if isinstance(cfg, dict) else None
+        google_section = section.get("google") if isinstance(section, dict) else None
+        return google_section if isinstance(google_section, dict) else {}
+    except Exception as exc:
+        logger.debug("Could not load video_gen.google config: %s", exc)
+        return {}
+
+
+def _resolve_model() -> Tuple[str, Dict[str, Any]]:
+    env_override = os.environ.get("GOOGLE_VIDEO_MODEL")
+    if env_override and env_override in _MODELS:
+        return env_override, _MODELS[env_override]
+
+    cfg = _load_video_config()
+    candidate = cfg.get("model") if isinstance(cfg.get("model"), str) else None
+    if candidate and candidate in _MODELS:
+        return candidate, _MODELS[candidate]
+
+    return DEFAULT_MODEL, _MODELS[DEFAULT_MODEL]
+
+
+def _videos_cache_dir() -> Path:
+    """Return ``$HERMES_HOME/cache/videos/``, creating parents as needed."""
+    from hermes_constants import get_hermes_home
+
+    path = get_hermes_home() / "cache" / "videos"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def _save_mp4(content: bytes, *, prefix: str = "video") -> Path:
+    ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+    short = uuid.uuid4().hex[:8]
+    path = _videos_cache_dir() / f"{prefix}_{ts}_{short}.mp4"
+    path.write_bytes(content)
+    return path
+
+
+def _check_video_gen_available() -> bool:
+    """Tool gate — only advertised when an API key is present."""
+    return bool(_resolve_api_key())
+
+
+# ---------------------------------------------------------------------------
+# Veo API calls
+# ---------------------------------------------------------------------------
+
+
+def _start_op(
+    *, api_key: str, model_id: str, prompt: str,
+    aspect_ratio: str, duration_seconds: int,
+) -> Tuple[Optional[str], Optional[str]]:
+    """Submit the long-running op. Returns (op_name, error_message)."""
+    url = f"{API_BASE}/models/{model_id}:predictLongRunning"
+    payload: Dict[str, Any] = {
+        "instances": [{"prompt": prompt}],
+        "parameters": {
+            "aspectRatio": aspect_ratio,
+            "durationSeconds": duration_seconds,
+        },
+    }
+    try:
+        resp = requests.post(
+            url, params={"key": api_key}, json=payload, timeout=60
+        )
+    except requests.Timeout:
+        return None, "Veo submit timed out (60s)"
+    except requests.ConnectionError as exc:
+        return None, f"Connection error: {exc}"
+
+    if resp.status_code != 200:
+        try:
+            err_msg = resp.json().get("error", {}).get("message", resp.text[:300])
+        except Exception:
+            err_msg = resp.text[:300]
+        return None, f"Veo submit failed ({resp.status_code}): {err_msg}"
+
+    op_name = resp.json().get("name")
+    if not op_name:
+        return None, "Veo submit returned no operation name"
+    return op_name, None
+
+
+def _poll_op(
+    *, api_key: str, op_name: str
+) -> Tuple[Optional[Dict[str, Any]], Optional[str]]:
+    """Poll until done or timeout. Returns (final_op_body, error_message)."""
+    deadline = time.time() + POLL_TIMEOUT_SECONDS
+    url = f"{API_BASE}/{op_name}"
+    last_status_for_log = None
+
+    while time.time() < deadline:
+        try:
+            resp = requests.get(url, params={"key": api_key}, timeout=30)
+        except requests.RequestException as exc:
+            return None, f"Polling failed: {exc}"
+
+        if resp.status_code != 200:
+            return None, f"Polling returned {resp.status_code}: {resp.text[:200]}"
+
+        try:
+            body = resp.json()
+        except Exception as exc:
+            return None, f"Polling returned invalid JSON: {exc}"
+
+        if body.get("done"):
+            return body, None
+
+        # Optional progress log (Veo doesn't surface % progress today, but this
+        # helps if Google adds it later)
+        meta = body.get("metadata") or {}
+        progress = meta.get("progressPercent") or meta.get("progress")
+        if progress != last_status_for_log:
+            logger.debug("Veo op %s progress: %s", op_name, progress)
+            last_status_for_log = progress
+
+        time.sleep(POLL_INTERVAL_SECONDS)
+
+    return None, f"Veo op {op_name} did not complete within {POLL_TIMEOUT_SECONDS}s"
+
+
+def _extract_video_uri(op_body: Dict[str, Any]) -> Optional[str]:
+    """Pull the download URI out of a finished op body."""
+    err = op_body.get("error")
+    if err:
+        return None
+    resp = op_body.get("response") or {}
+    samples = (resp.get("generateVideoResponse") or {}).get("generatedSamples") or []
+    if not samples:
+        samples = resp.get("generatedSamples") or []
+    for s in samples:
+        v = s.get("video") or {}
+        uri = v.get("uri") or v.get("url")
+        if uri:
+            return uri
+    return None
+
+
+def _download_video(*, api_key: str, uri: str) -> Tuple[Optional[bytes], Optional[str]]:
+    """Download the MP4. Veo's URI returns a 302 → follow with the same key."""
+    try:
+        resp = requests.get(
+            uri,
+            headers={"x-goog-api-key": api_key},
+            timeout=120,
+            allow_redirects=True,
+        )
+    except requests.Timeout:
+        return None, "Video download timed out (120s)"
+    except requests.ConnectionError as exc:
+        return None, f"Connection error: {exc}"
+
+    if resp.status_code != 200:
+        return None, f"Video download failed ({resp.status_code}): {resp.text[:200]}"
+    if len(resp.content) < 1000:
+        # 95-byte JSON error masquerading as a download — guard against it.
+        return None, f"Video download returned suspiciously small body ({len(resp.content)} bytes)"
+    return resp.content, None
+
+
+# ---------------------------------------------------------------------------
+# Tool handler
+# ---------------------------------------------------------------------------
+
+
+VIDEO_GENERATE_SCHEMA = {
+    "name": "video_generate",
+    "description": (
+        "Generate a short video from a text prompt using Google Veo. "
+        "Long-running (~30s–2min). Returns an absolute path to a saved "
+        "MP4 file in the `video` field. Display with markdown "
+        "![description](path) and the gateway will deliver it. "
+        "Backend model is user-configured via "
+        "`video_gen.google.model` or `GOOGLE_VIDEO_MODEL` env var."
+    ),
+    "parameters": {
+        "type": "object",
+        "properties": {
+            "prompt": {
+                "type": "string",
+                "description": (
+                    "The text prompt describing the desired video. Be "
+                    "detailed: subject, action, camera motion, style, "
+                    "lighting. Veo follows cinematic prompt structure well."
+                ),
+            },
+            "aspect_ratio": {
+                "type": "string",
+                "enum": list(VALID_ASPECT_RATIOS),
+                "description": "Aspect ratio. 16:9 widescreen, 9:16 vertical, 1:1 square.",
+                "default": DEFAULT_ASPECT_RATIO,
+            },
+            "duration_seconds": {
+                "type": "integer",
+                "description": "Length of the clip in seconds (Veo 3: 4 or 8; Veo 2: 5–8).",
+                "default": 4,
+                "minimum": 2,
+                "maximum": 16,
+            },
+        },
+        "required": ["prompt"],
+    },
+}
+
+
+def _handle_video_generate(args: Dict[str, Any], **_kw: Any) -> str:
+    prompt = (args.get("prompt") or "").strip()
+    if not prompt:
+        return tool_error("prompt is required for video generation")
+
+    aspect_ratio = args.get("aspect_ratio") or DEFAULT_ASPECT_RATIO
+    if aspect_ratio not in VALID_ASPECT_RATIOS:
+        aspect_ratio = DEFAULT_ASPECT_RATIO
+
+    duration_seconds = int(args.get("duration_seconds") or 4)
+    duration_seconds = max(2, min(16, duration_seconds))
+
+    api_key = _resolve_api_key()
+    if not api_key:
+        return tool_error(
+            "GEMINI_API_KEY (or GOOGLE_API_KEY) not set. Get a key at "
+            "https://aistudio.google.com/apikey — Veo requires Tier 1 "
+            "(paid) on the project.",
+            error_type="auth_required",
+            provider="google",
+        )
+
+    model_id, meta = _resolve_model()
+
+    op_name, err = _start_op(
+        api_key=api_key,
+        model_id=model_id,
+        prompt=prompt,
+        aspect_ratio=aspect_ratio,
+        duration_seconds=duration_seconds,
+    )
+    if err:
+        return tool_error(err, error_type="api_error", provider="google",
+                          model=model_id, prompt=prompt)
+
+    op_body, err = _poll_op(api_key=api_key, op_name=op_name)
+    if err:
+        return tool_error(err, error_type="poll_error", provider="google",
+                          model=model_id, operation=op_name)
+
+    op_err = op_body.get("error") if op_body else None
+    if op_err:
+        return tool_error(
+            f"Veo op finished with error: {op_err.get('message', op_err)}",
+            error_type="api_error", provider="google",
+            model=model_id, operation=op_name,
+        )
+
+    uri = _extract_video_uri(op_body or {})
+    if not uri:
+        return tool_error(
+            "Veo op completed but no video URI in response",
+            error_type="empty_response", provider="google",
+            model=model_id, operation=op_name,
+        )
+
+    content, err = _download_video(api_key=api_key, uri=uri)
+    if err:
+        return tool_error(err, error_type="download_error", provider="google",
+                          model=model_id, operation=op_name)
+
+    try:
+        saved_path = _save_mp4(content, prefix=f"google_{model_id}")
+    except Exception as exc:
+        return tool_error(
+            f"Could not save video to cache: {exc}",
+            error_type="io_error", provider="google", model=model_id,
+        )
+
+    return tool_result(
+        success=True,
+        video=str(saved_path),
+        model=model_id,
+        prompt=prompt,
+        aspect_ratio=aspect_ratio,
+        duration_seconds=duration_seconds,
+        provider="google",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Plugin entry point
+# ---------------------------------------------------------------------------
+
+
+def register(ctx: Any) -> None:
+    """Plugin entry point — register the video_generate tool."""
+    ctx.register_tool(
+        name="video_generate",
+        toolset="video_gen",
+        schema=VIDEO_GENERATE_SCHEMA,
+        handler=_handle_video_generate,
+        check_fn=_check_video_gen_available,
+        requires_env=[],
+        is_async=False,
+        description="Generate a short video from a text prompt via Google Veo.",
+        emoji="🎬",
+    )
diff --git a/plugins/video_gen/google/plugin.yaml b/plugins/video_gen/google/plugin.yaml
new file mode 100644
index 000000000000..86dc9f4e3b39
--- /dev/null
+++ b/plugins/video_gen/google/plugin.yaml
@@ -0,0 +1,7 @@
+name: video_gen-google
+version: 1.0.0
+description: "Video generation via Google Veo (2.0, 3.0 fast/standard, 3.1 family). Registers a `video_generate` tool. Saves MP4 to $HERMES_HOME/cache/videos/."
+author: ItachiDevv
+kind: standalone
+requires_env:
+  - GEMINI_API_KEY
diff --git a/tests/plugins/video_gen/__init__.py b/tests/plugins/video_gen/__init__.py
new file mode 100644
index 000000000000..e69de29bb2d1
diff --git a/tests/plugins/video_gen/test_google_video.py b/tests/plugins/video_gen/test_google_video.py
new file mode 100644
index 000000000000..4eaa9f957df7
--- /dev/null
+++ b/tests/plugins/video_gen/test_google_video.py
@@ -0,0 +1,353 @@
+#!/usr/bin/env python3
+"""Tests for Google Veo video_generate plugin."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _fake_api_key(monkeypatch):
+    monkeypatch.setenv("GEMINI_API_KEY", "test-key-12345")
+    monkeypatch.delenv("GOOGLE_API_KEY", raising=False)
+
+
+@pytest.fixture(autouse=True)
+def _clear_model_overrides(monkeypatch):
+    monkeypatch.delenv("GOOGLE_VIDEO_MODEL", raising=False)
+
+
+@pytest.fixture(autouse=True)
+def _no_real_sleeps(monkeypatch):
+    """Polling loop calls time.sleep — short-circuit so tests stay fast."""
+    monkeypatch.setattr("plugins.video_gen.google.time.sleep", lambda *_a, **_k: None)
+
+
+# ---------------------------------------------------------------------------
+# Tool gate + model resolution
+# ---------------------------------------------------------------------------
+
+
+class TestGate:
+    def test_available_with_gemini_key(self, monkeypatch):
+        monkeypatch.setenv("GEMINI_API_KEY", "k")
+        from plugins.video_gen.google import _check_video_gen_available
+
+        assert _check_video_gen_available() is True
+
+    def test_available_with_google_api_key(self, monkeypatch):
+        monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+        monkeypatch.setenv("GOOGLE_API_KEY", "k")
+        from plugins.video_gen.google import _check_video_gen_available
+
+        assert _check_video_gen_available() is True
+
+    def test_unavailable_without_key(self, monkeypatch):
+        monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+        monkeypatch.delenv("GOOGLE_API_KEY", raising=False)
+        from plugins.video_gen.google import _check_video_gen_available
+
+        assert _check_video_gen_available() is False
+
+
+class TestModelResolution:
+    def test_default_is_veo3_fast(self):
+        from plugins.video_gen.google import _resolve_model
+
+        model_id, _ = _resolve_model()
+        assert model_id == "veo-3.0-fast-generate-001"
+
+    def test_env_override(self, monkeypatch):
+        monkeypatch.setenv("GOOGLE_VIDEO_MODEL", "veo-3.0-generate-001")
+        from plugins.video_gen.google import _resolve_model
+
+        model_id, _ = _resolve_model()
+        assert model_id == "veo-3.0-generate-001"
+
+    def test_env_override_unknown_falls_back(self, monkeypatch):
+        monkeypatch.setenv("GOOGLE_VIDEO_MODEL", "veo-99-impossible")
+        from plugins.video_gen.google import _resolve_model
+
+        model_id, _ = _resolve_model()
+        assert model_id == "veo-3.0-fast-generate-001"
+
+
+# ---------------------------------------------------------------------------
+# Helpers — operation submit / poll / extract / download
+# ---------------------------------------------------------------------------
+
+
+class TestSubmitOp:
+    def test_returns_op_name(self):
+        from plugins.video_gen.google import _start_op
+
+        mock_resp = MagicMock()
+        mock_resp.status_code = 200
+        mock_resp.json.return_value = {"name": "models/veo-3.0-fast-generate-001/operations/abc"}
+
+        with patch("plugins.video_gen.google.requests.post", return_value=mock_resp) as post:
+            name, err = _start_op(
+                api_key="k", model_id="veo-3.0-fast-generate-001",
+                prompt="x", aspect_ratio="16:9", duration_seconds=4,
+            )
+        assert err is None
+        assert name == "models/veo-3.0-fast-generate-001/operations/abc"
+
+        called_url = post.call_args.args[0]
+        assert ":predictLongRunning" in called_url
+        body = post.call_args.kwargs["json"]
+        assert body["instances"][0]["prompt"] == "x"
+        assert body["parameters"]["aspectRatio"] == "16:9"
+        assert body["parameters"]["durationSeconds"] == 4
+        assert post.call_args.kwargs["params"] == {"key": "k"}
+
+    def test_api_error_propagates(self):
+        from plugins.video_gen.google import _start_op
+
+        mock_resp = MagicMock()
+        mock_resp.status_code = 400
+        mock_resp.text = '{"error": {"message": "paid plan required"}}'
+        mock_resp.json.return_value = {"error": {"message": "paid plan required"}}
+
+        with patch("plugins.video_gen.google.requests.post", return_value=mock_resp):
+            _, err = _start_op(
+                api_key="k", model_id="veo-3.0-fast-generate-001",
+                prompt="x", aspect_ratio="16:9", duration_seconds=4,
+            )
+        assert err is not None
+        assert "paid plan" in err
+
+    def test_timeout(self):
+        import requests as req_lib
+        from plugins.video_gen.google import _start_op
+
+        with patch("plugins.video_gen.google.requests.post", side_effect=req_lib.Timeout()):
+            _, err = _start_op(
+                api_key="k", model_id="veo-3.0-fast-generate-001",
+                prompt="x", aspect_ratio="16:9", duration_seconds=4,
+            )
+        assert err is not None and "timed out" in err
+
+
+class TestPollOp:
+    def test_returns_body_when_done(self):
+        from plugins.video_gen.google import _poll_op
+
+        # Two pending responses, then a done one — verifies the loop polls.
+        responses = [
+            MagicMock(status_code=200, **{"json.return_value": {"done": False}}),
+            MagicMock(status_code=200, **{"json.return_value": {"done": False}}),
+            MagicMock(status_code=200, **{"json.return_value": {"done": True, "response": {}}}),
+        ]
+
+        with patch("plugins.video_gen.google.requests.get", side_effect=responses):
+            body, err = _poll_op(api_key="k", op_name="models/veo/operations/x")
+        assert err is None
+        assert body == {"done": True, "response": {}}
+
+    def test_polling_returns_non_200(self):
+        from plugins.video_gen.google import _poll_op
+
+        mock_resp = MagicMock(status_code=500, text="oops")
+        with patch("plugins.video_gen.google.requests.get", return_value=mock_resp):
+            body, err = _poll_op(api_key="k", op_name="models/veo/operations/x")
+        assert body is None
+        assert err is not None and "500" in err
+
+
+class TestExtractUri:
+    def test_finds_uri_in_generateVideoResponse(self):
+        from plugins.video_gen.google import _extract_video_uri
+
+        body = {
+            "response": {
+                "generateVideoResponse": {
+                    "generatedSamples": [{"video": {"uri": "https://x/file.mp4"}}]
+                }
+            }
+        }
+        assert _extract_video_uri(body) == "https://x/file.mp4"
+
+    def test_finds_uri_at_top_level(self):
+        from plugins.video_gen.google import _extract_video_uri
+
+        body = {"response": {"generatedSamples": [{"video": {"uri": "https://y/v.mp4"}}]}}
+        assert _extract_video_uri(body) == "https://y/v.mp4"
+
+    def test_returns_none_on_error(self):
+        from plugins.video_gen.google import _extract_video_uri
+
+        assert _extract_video_uri({"error": {"message": "x"}}) is None
+
+    def test_returns_none_when_missing(self):
+        from plugins.video_gen.google import _extract_video_uri
+
+        assert _extract_video_uri({"response": {}}) is None
+
+
+class TestDownload:
+    def test_success(self):
+        from plugins.video_gen.google import _download_video
+
+        mock_resp = MagicMock(status_code=200, content=b"\x00" * 5000)
+        with patch("plugins.video_gen.google.requests.get", return_value=mock_resp) as g:
+            content, err = _download_video(api_key="k", uri="https://x/v.mp4")
+        assert err is None
+        assert content == b"\x00" * 5000
+        # Auth via header, follow redirects
+        assert g.call_args.kwargs["headers"]["x-goog-api-key"] == "k"
+        assert g.call_args.kwargs["allow_redirects"] is True
+
+    def test_small_body_rejected(self):
+        from plugins.video_gen.google import _download_video
+
+        mock_resp = MagicMock(status_code=200, content=b'{"err":"x"}')
+        with patch("plugins.video_gen.google.requests.get", return_value=mock_resp):
+            content, err = _download_video(api_key="k", uri="https://x")
+        assert content is None
+        assert err is not None and "small" in err.lower()
+
+
+# ---------------------------------------------------------------------------
+# End-to-end handler
+# ---------------------------------------------------------------------------
+
+
+class TestHandler:
+    def test_full_happy_path(self, tmp_path, monkeypatch):
+        monkeypatch.setattr("plugins.video_gen.google._videos_cache_dir", lambda: tmp_path)
+        from plugins.video_gen.google import _handle_video_generate
+
+        with patch("plugins.video_gen.google._start_op", return_value=("ops/x", None)):
+            with patch("plugins.video_gen.google._poll_op", return_value=(
+                {"done": True, "response": {"generateVideoResponse": {"generatedSamples": [{"video": {"uri": "https://x/v.mp4"}}]}}},
+                None,
+            )):
+                with patch("plugins.video_gen.google._download_video", return_value=(b"\x00" * 5000, None)):
+                    raw = _handle_video_generate({"prompt": "a robot waves", "duration_seconds": 4})
+
+        result = json.loads(raw)
+        assert result["success"] is True
+        assert result["model"] == "veo-3.0-fast-generate-001"
+        assert result["aspect_ratio"] == "16:9"
+        assert result["duration_seconds"] == 4
+        # File actually saved
+        from pathlib import Path
+        saved = Path(result["video"])
+        assert saved.exists()
+        assert saved.stat().st_size == 5000
+
+    def test_missing_api_key(self, monkeypatch):
+        monkeypatch.delenv("GEMINI_API_KEY", raising=False)
+        monkeypatch.delenv("GOOGLE_API_KEY", raising=False)
+        from plugins.video_gen.google import _handle_video_generate
+
+        result = json.loads(_handle_video_generate({"prompt": "x"}))
+        assert "GEMINI_API_KEY" in result["error"]
+        assert result["error_type"] == "auth_required"
+
+    def test_empty_prompt(self):
+        from plugins.video_gen.google import _handle_video_generate
+
+        result = json.loads(_handle_video_generate({"prompt": ""}))
+        assert "prompt is required" in result["error"]
+
+    def test_invalid_aspect_ratio_normalised(self):
+        from plugins.video_gen.google import _handle_video_generate
+
+        captured = {}
+
+        def fake_start(**kw):
+            captured.update(kw)
+            return None, "stop here"
+
+        with patch("plugins.video_gen.google._start_op", side_effect=fake_start):
+            _handle_video_generate({"prompt": "x", "aspect_ratio": "21:9"})
+        # 21:9 is not in the enum — should be normalised to the default 16:9.
+        assert captured["aspect_ratio"] == "16:9"
+
+    def test_duration_clamped(self):
+        from plugins.video_gen.google import _handle_video_generate
+
+        captured = {}
+
+        def fake_start(**kw):
+            captured.update(kw)
+            return None, "stop"
+
+        with patch("plugins.video_gen.google._start_op", side_effect=fake_start):
+            _handle_video_generate({"prompt": "x", "duration_seconds": 99})
+        assert captured["duration_seconds"] == 16  # clamped to max
+
+        captured.clear()
+        with patch("plugins.video_gen.google._start_op", side_effect=fake_start):
+            _handle_video_generate({"prompt": "x", "duration_seconds": 1})
+        assert captured["duration_seconds"] == 2  # clamped to min
+
+    def test_submit_error_propagates(self):
+        from plugins.video_gen.google import _handle_video_generate
+
+        with patch("plugins.video_gen.google._start_op", return_value=(None, "submit failed: 400")):
+            result = json.loads(_handle_video_generate({"prompt": "x"}))
+        assert result["error_type"] == "api_error"
+        assert "submit failed" in result["error"]
+
+    def test_op_finished_with_error(self):
+        from plugins.video_gen.google import _handle_video_generate
+
+        with patch("plugins.video_gen.google._start_op", return_value=("ops/x", None)):
+            with patch("plugins.video_gen.google._poll_op", return_value=(
+                {"done": True, "error": {"message": "content policy"}}, None,
+            )):
+                result = json.loads(_handle_video_generate({"prompt": "x"}))
+        assert result["error_type"] == "api_error"
+        assert "content policy" in result["error"]
+
+    def test_no_uri_in_response(self):
+        from plugins.video_gen.google import _handle_video_generate
+
+        with patch("plugins.video_gen.google._start_op", return_value=("ops/x", None)):
+            with patch("plugins.video_gen.google._poll_op", return_value=(
+                {"done": True, "response": {}}, None,
+            )):
+                result = json.loads(_handle_video_generate({"prompt": "x"}))
+        assert result["error_type"] == "empty_response"
+
+
+# ---------------------------------------------------------------------------
+# Schema + registration
+# ---------------------------------------------------------------------------
+
+
+class TestSchema:
+    def test_schema_shape(self):
+        from plugins.video_gen.google import VIDEO_GENERATE_SCHEMA
+
+        assert VIDEO_GENERATE_SCHEMA["name"] == "video_generate"
+        params = VIDEO_GENERATE_SCHEMA["parameters"]["properties"]
+        assert "prompt" in params
+        assert params["aspect_ratio"]["enum"] == ["16:9", "9:16", "1:1"]
+        assert params["duration_seconds"]["minimum"] == 2
+        assert params["duration_seconds"]["maximum"] == 16
+
+
+class TestRegistration:
+    def test_register_calls_register_tool(self):
+        from plugins.video_gen.google import register
+
+        ctx = MagicMock()
+        register(ctx)
+        ctx.register_tool.assert_called_once()
+        kwargs = ctx.register_tool.call_args.kwargs
+        assert kwargs["name"] == "video_generate"
+        assert kwargs["toolset"] == "video_gen"
+        assert kwargs["emoji"] == "🎬"
+        assert kwargs["check_fn"] is not None