feat: add Gemini agentic video adapter

pyproject.toml

@@ -66,7 +66,7 @@ dependencies = [
     "PyYAML>=6.0.0",
     "requests>=2.31.0",
     "Pillow>=10.0.0",
-    "google-genai>=1.0.0",
+    "google-genai>=2.21.0",
     "google-generativeai>=0.3.0",
     "opentelemetry-distro>=0.40b0",
     "opentelemetry-exporter-otlp>=1.20.0",

requirements.txt

@@ -55,7 +55,7 @@ psutil>=5.9.0
 cachetools>=5.0.0
 
 # Google AI & Cloud
-google-genai>=1.0.0
+google-genai>=2.21.0
 google-generativeai>=0.3.0
 google-cloud-aiplatform>=1.133.0
 google-cloud-vision>=3.7.0

src/integration/gemini_agentic_video.py

@@ -0,0 +1,149 @@
+"""Gemini agentic video-understanding adapter.
+
+This module isolates the Gemini Interactions API from EventRelay's existing
+``generateContent`` integration.  It enables targeted, server-side inspection
+of transcripts, frames, and audio without changing the production path until
+benchmark evidence supports promotion.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from dataclasses import dataclass
+from typing import Any, Literal, Sequence
+from urllib.parse import urlsplit
+
+ProcessingMode = Literal["agentic", "static"]
+
+
+@dataclass(frozen=True)
+class VideoInput:
+    """One video reference and its independently selected processing mode."""
+
+    uri: str
+    processing: ProcessingMode = "agentic"
+    mime_type: str | None = None
+
+
+@dataclass(frozen=True)
+class AgenticVideoReceipt:
+    """Stable execution receipt retained by EventRelay after analysis."""
+
+    output_text: str
+    total_tokens: int | None
+    model: str
+    sources: tuple[str, ...]
+    processing_modes: tuple[ProcessingMode, ...]
+
+
+class GeminiAgenticVideoService:
+    """Run Gemini's Think -> Act -> Observe video-analysis loop."""
+
+    DEFAULT_MODEL = "gemini-3.7-flash"
+
+    def __init__(
+        self,
+        api_key: str | None = None,
+        *,
+        client: Any | None = None,
+        model: str | None = None,
+    ) -> None:
+        self.model = model or os.getenv(
+            "GEMINI_AGENTIC_VIDEO_MODEL", self.DEFAULT_MODEL
+        )
+        if client is not None:
+            self._client = client
+            return
+
+        from google import genai
+
+        resolved_key = api_key or os.getenv("GEMINI_API_KEY")
+        self._client = (
+            genai.Client(api_key=resolved_key) if resolved_key else genai.Client()
+        )
+
+    @staticmethod
+    def build_input(videos: Sequence[VideoInput], prompt: str) -> list[dict[str, str]]:
+        """Build the documented Interactions API input without materializing media."""
+        if not videos:
+            raise ValueError("At least one video reference is required")
+        if not prompt.strip():
+            raise ValueError("A non-empty analysis prompt is required")
+
+        items: list[dict[str, str]] = []
+        for video in videos:
+            if not GeminiAgenticVideoService._is_supported_uri(video.uri):
+                raise ValueError("Video URI must be a supported YouTube or file URI")
+            item = {
+                "type": "video",
+                "uri": video.uri,
+                "processing": video.processing,
+            }
+            if video.mime_type:
+                item["mime_type"] = video.mime_type
+            items.append(item)
+        items.append({"type": "text", "text": prompt})
+        return items
+
+    @staticmethod
+    def _is_supported_uri(uri: str) -> bool:
+        if not uri.strip() or uri != uri.strip():
+            return False
+        try:
+            parsed = urlsplit(uri)
+            hostname = (parsed.hostname or "").lower()
+        except ValueError:
+            return False
+
+        if parsed.scheme == "https":
+            return (
+                bool(parsed.path)
+                and (
+                    hostname == "youtu.be"
+                    or hostname == "youtube.com"
+                    or hostname.endswith(".youtube.com")
+                )
+            )
+        if parsed.scheme == "gs":
+            return bool(parsed.netloc and parsed.path)
+        if parsed.scheme == "file":
+            return bool(parsed.path)
+        return False
+
+    async def analyze(
+        self,
+        videos: Sequence[VideoInput],
+        prompt: str,
+        *,
+        model: str | None = None,
+    ) -> AgenticVideoReceipt:
+        """Analyze referenced media and return a durable, comparable receipt."""
+        selected_model = model or self.model
+        request_input = self.build_input(videos, prompt)
+        response = await asyncio.to_thread(
+            self._client.interactions.create,
+            model=selected_model,
+            input=request_input,
+        )
+
+        usage = getattr(response, "usage", None)
+        total_tokens = getattr(usage, "total_tokens", None)
+        return AgenticVideoReceipt(
+            output_text=self._response_text(response),
+            total_tokens=int(total_tokens) if total_tokens is not None else None,
+            model=selected_model,
+            sources=tuple(video.uri for video in videos),
+            processing_modes=tuple(video.processing for video in videos),
+        )
+
+    @staticmethod
+    def _response_text(response: Any) -> str:
+        output_text = getattr(response, "output_text", None)
+        if output_text is not None:
+            return str(output_text)
+        return "".join(
+            str(getattr(output, "text", ""))
+            for output in (getattr(response, "outputs", None) or [])
+            if getattr(output, "text", None) is not None
+        )

tests/unit/test_gemini_agentic_video.py

@@ -0,0 +1,93 @@
+from types import SimpleNamespace
+
+import pytest
+from google.genai._gaos.types.interactions.interaction import Interaction
+from google.genai._gaos.types.interactions.modeloutputstep import ModelOutputStep
+from google.genai._gaos.types.interactions.textcontent import TextContent
+from google.genai._gaos.types.interactions.usage import Usage
+
+from src.integration.gemini_agentic_video import (
+    GeminiAgenticVideoService,
+    VideoInput,
+)
+
+
+class FakeInteractions:
+    def __init__(self) -> None:
+        self.request = None
+
+    def create(self, **kwargs):
+        self.request = kwargs
+        return Interaction(
+            id="interaction-123",
+            created="2026-09-09T20:00:00Z",
+            status="completed",
+            updated="2026-09-09T20:00:01Z",
+            steps=[
+                ModelOutputStep(content=[TextContent(text="grounded result")])
+            ],
+            usage=Usage(total_tokens=321),
+        )
+
+
+@pytest.mark.asyncio
+async def test_agentic_youtube_request_returns_execution_receipt():
+    interactions = FakeInteractions()
+    client = SimpleNamespace(interactions=interactions)
+    service = GeminiAgenticVideoService(client=client)
+
+    receipt = await service.analyze(
+        [VideoInput("https://youtu.be/auJzb1D-fag")],
+        "Find the implementation steps and their timestamps.",
+    )
+
+    assert interactions.request == {
+        "model": "gemini-3.7-flash",
+        "input": [
+            {
+                "type": "video",
+                "uri": "https://youtu.be/auJzb1D-fag",
+                "processing": "agentic",
+            },
+            {
+                "type": "text",
+                "text": "Find the implementation steps and their timestamps.",
+            },
+        ],
+    }
+    assert receipt.output_text == "grounded result"
+    assert receipt.total_tokens == 321
+    assert receipt.sources == ("https://youtu.be/auJzb1D-fag",)
+    assert receipt.processing_modes == ("agentic",)
+
+
+def test_mixed_mode_keeps_each_video_processing_policy():
+    request_input = GeminiAgenticVideoService.build_input(
+        [
+            VideoInput("gs://bucket/reference.mp4", "agentic", "video/mp4"),
+            VideoInput("gs://bucket/clip.mp4", "static", "video/mp4"),
+        ],
+        "Locate the clip in the reference recording.",
+    )
+
+    assert request_input[0]["processing"] == "agentic"
+    assert request_input[1]["processing"] == "static"
+    assert request_input[0]["mime_type"] == "video/mp4"
+
+
+@pytest.mark.parametrize(
+    "uri",
+    ["x", "https://example.com/video.mp4", "ftp://youtu.be/auJzb1D-fag"],
+)
+def test_malformed_video_uri_fails_before_calling_provider(uri):
+    with pytest.raises(ValueError):
+        GeminiAgenticVideoService.build_input([VideoInput(uri)], "question")
+
+
+@pytest.mark.parametrize(
+    ("videos", "prompt"),
+    [([], "question"), ([VideoInput("")], "question"), ([VideoInput("x")], " ")],
+)
+def test_invalid_requests_fail_before_calling_provider(videos, prompt):
+    with pytest.raises(ValueError):
+        GeminiAgenticVideoService.build_input(videos, prompt)