feat: route watch through the Gemini engine when a key is present

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

skills/watch/scripts/watch.py

@@ -16,7 +16,8 @@ from pathlib import Path
 SCRIPT_DIR = Path(__file__).parent.resolve()
 sys.path.insert(0, str(SCRIPT_DIR))
 
-from config import ConfigError, frame_cap, get_config  # noqa: E402
+from config import ConfigError, frame_cap, get_config, load_gemini_key, resolve_engine  # noqa: E402
+import gemini  # noqa: E402
 from download import download, fetch_captions, is_url, auth_args  # noqa: E402
 from frames import MAX_FPS, auto_fps, auto_fps_focus, extract_at_timestamps, extract_keyframes, extract_scene_or_uniform, format_time, get_metadata, merge_frames, parse_time, parse_timestamps, validate_controls  # noqa: E402
 from transcribe import filter_range, format_transcript, parse_vtt  # noqa: E402
@@ -24,6 +25,67 @@ from whisper import load_api_key, transcribe_video  # noqa: E402
 from runtime import configure_stdio  # noqa: E402
 
 
+LOCAL_ONLY_FLAGS = (("--detail", "detail"), ("--fps", "fps"), ("--max-frames", "max_frames"),
+                    ("--timestamps", "timestamps"), ("--whisper", "whisper"), ("--sub-lang", "sub_lang"))
+
+
+def run_gemini(args, config, key, start_sec, end_sec, auth) -> int:
+    """Google watches the video. No captions, frames, or Whisper; no fallback on failure."""
+    ignored = [flag for flag, name in LOCAL_ONLY_FLAGS if getattr(args, name) is not None]
+    ignored += [flag for flag, on in (("--no-whisper", args.no_whisper), ("--no-dedup", args.no_dedup)) if on]
+    clip = (start_sec, end_sec) if start_sec is not None or end_sec is not None else None
+    uploaded, warning, result, error, sent = None, None, None, None, "URL sent to Google"
+    try:
+        if gemini.is_youtube(args.source):
+            video = {"uri": args.source}
+        else:
+            parent = Path(args.out_dir).expanduser().resolve() if args.out_dir else None
+            if parent:
+                parent.mkdir(parents=True, exist_ok=True)
+            work = Path(tempfile.mkdtemp(prefix="watch-", dir=parent))
+            print("[watch] downloading media…" if is_url(args.source) else "[watch] using local file…", file=sys.stderr)
+            media = download(args.source, work / "download", **(auth if is_url(args.source) else {}))
+            print("[watch] uploading to the Gemini Files API…", file=sys.stderr)
+            uploaded = gemini.upload_file(Path(media["video_path"]), key)
+            video = {"uri": uploaded["uri"], "mime_type": uploaded["mime_type"]}
+            sent = "video uploaded to Google, deleted after the answer"
+        print(f"[watch] asking {config['gemini_model']}…", file=sys.stderr)
+        result = gemini.ask(video, args.question, model=config["gemini_model"], key=key,
+                            clip=clip, timeout=config["gemini_timeout"])
+    except SystemExit as exc:
+        error = str(exc)
+    finally:
+        if uploaded:
+            warning = gemini.delete_file(uploaded["name"], key)
+
+    print()
+    print("# watch: video report")
+    print()
+    print(f"- **Source:** {args.source} ({sent})")
+    print(f"- **Engine:** {config['gemini_model']} ({result['processing'] if result else 'failed'})")
+    if clip:
+        print(f"- **Focus range:** {format_time(start_sec or 0)} → {format_time(end_sec) if end_sec is not None else 'end'}")
+    if ignored:
+        print(f"- **Ignored local options:** {', '.join(ignored)} (these only apply with --engine local)")
+    if result and result["total_tokens"] is not None:
+        print(f"- **Gemini tokens:** {result['total_tokens']}")
+    if warning:
+        print(f"- **Cleanup warning:** {warning}")
+    print()
+    if error:
+        print("## Unavailable evidence")
+        print()
+        print(f"- {error}")
+        return 1
+    print("## Answer (from Gemini)")
+    print()
+    print("_These are Gemini's observations of the video, not frames you viewed yourself. "
+          "Relay them as such; rerun with `--engine local` to inspect frames directly._")
+    print()
+    print(result["text"])
+    return 0
+
+
 def main() -> int:
     ap = argparse.ArgumentParser(
         prog="watch",
@@ -72,6 +134,11 @@ def main() -> int:
     cookies = ap.add_mutually_exclusive_group()
     cookies.add_argument("--cookies", default=None, help="Explicit cookie file (yt-dlp may update this jar)")
     cookies.add_argument("--cookies-from-browser", default=None, help="Explicit yt-dlp browser selector")
+    ap.add_argument("--engine", choices=["auto", "gemini", "local"], default=None,
+                    help="gemini: Google watches the video and answers (needs GEMINI_API_KEY). "
+                         "local: frames + transcript on this machine. Default auto: gemini when a key exists.")
+    ap.add_argument("--question", default=None,
+                    help="The user's question. Sent to Gemini on a gemini run; unused by the local engine.")
     args = ap.parse_args()
     if args.no_whisper and args.whisper:
         ap.error("--no-whisper conflicts with --whisper")
@@ -89,6 +156,10 @@ def main() -> int:
     auth_args(cookies_file, cookies_browser)  # Validate even before starting network work.
     auth = {"cookies_file": cookies_file, "cookies_from_browser": cookies_browser}
 
+    gemini_key = load_gemini_key()
+    if resolve_engine(args.engine or config["engine"], bool(gemini_key)) == "gemini":
+        return run_gemini(args, config, gemini_key, start_sec, end_sec, auth)
+
     parent = Path(args.out_dir).expanduser().resolve() if args.out_dir else None
     if parent:
         parent.mkdir(parents=True, exist_ok=True)

tests/test_watch.py

@@ -188,3 +188,84 @@ def test_cli_no_whisper_overrides_invalid_saved_backend(monkeypatch):
     monkeypatch.setenv('WATCH_WHISPER_BACKEND', 'invalid-old-preference')
     monkeypatch.setattr(sys, 'argv', ['watch', 'video.mp4', '--detail', 'transcript', '--no-whisper'])
     assert watch.main() == 0
+
+
+def _gemini_answer(monkeypatch, seen):
+    import gemini
+    def fake_ask(video, question, **kw):
+        seen.update(video=video, question=question, **kw)
+        return {'text': 'At 00:10 it turns blue.', 'model': kw['model'], 'processing': 'agentic', 'total_tokens': 4348}
+    monkeypatch.setattr(gemini, 'ask', fake_ask)
+
+
+def test_youtube_with_key_uses_gemini_and_touches_nothing_local(monkeypatch, capsys):
+    seen = {}
+    _gemini_answer(monkeypatch, seen)
+    monkeypatch.setenv('GEMINI_API_KEY', 'unit-key')
+    for name in ('fetch_captions', 'download', 'get_metadata', 'transcribe_video'):
+        monkeypatch.setattr(watch, name, lambda *a, _n=name, **kw: pytest.fail(f'{_n} must not run on a Gemini YouTube run'))
+    monkeypatch.setattr(sys, 'argv', ['watch', 'https://youtu.be/abc', '--question', 'When does it change?', '--detail', 'efficient'])
+    assert watch.main() == 0
+    out = capsys.readouterr().out
+    assert seen['video'] == {'uri': 'https://youtu.be/abc'} and seen['question'] == 'When does it change?'
+    assert seen['clip'] is None and seen['model'] == 'gemini-3.7-flash'
+    assert '**Engine:** gemini-3.7-flash (agentic)' in out
+    assert '**Ignored local options:** --detail' in out
+    assert '## Answer (from Gemini)' in out and 'At 00:10 it turns blue.' in out
+    assert 'unit-key' not in out
+
+
+def test_local_file_is_uploaded_asked_then_deleted(monkeypatch, capsys, static_clip):
+    import gemini
+    seen, deleted = {}, []
+    _gemini_answer(monkeypatch, seen)
+    monkeypatch.setenv('GEMINI_API_KEY', 'unit-key')
+    monkeypatch.setattr(gemini, 'upload_file', lambda path, key, **kw: {'name': 'files/abc', 'uri': 'https://g/files/abc', 'mime_type': 'video/mp4'})
+    monkeypatch.setattr(gemini, 'delete_file', lambda name, key: deleted.append(name))
+    monkeypatch.setattr(sys, 'argv', ['watch', str(static_clip), '--start', '0:20', '--end', '0:30'])
+    assert watch.main() == 0
+    assert seen['video'] == {'uri': 'https://g/files/abc', 'mime_type': 'video/mp4'}
+    assert seen['clip'] == (20.0, 30.0)
+    assert deleted == ['files/abc']
+    assert 'uploaded to Google' in capsys.readouterr().out
+
+
+def test_upload_is_deleted_even_when_the_question_fails(monkeypatch, capsys, static_clip):
+    import gemini
+    deleted = []
+    monkeypatch.setenv('GEMINI_API_KEY', 'unit-key')
+    monkeypatch.setattr(gemini, 'upload_file', lambda path, key, **kw: {'name': 'files/abc', 'uri': 'u', 'mime_type': 'video/mp4'})
+    monkeypatch.setattr(gemini, 'delete_file', lambda name, key: deleted.append(name))
+    def boom(*a, **kw):
+        raise SystemExit('Gemini quota: slow down. No local fallback was attempted; rerun with --engine local')
+    monkeypatch.setattr(gemini, 'ask', boom)
+    monkeypatch.setattr(watch, 'get_metadata', lambda *a, **kw: pytest.fail('no silent fallback to the local pipeline'))
+    monkeypatch.setattr(sys, 'argv', ['watch', str(static_clip)])
+    assert watch.main() == 1
+    assert deleted == ['files/abc']
+    out = capsys.readouterr().out
+    assert 'Gemini quota' in out and '--engine local' in out and '## Frames' not in out
+
+
+def test_engine_local_never_contacts_gemini(monkeypatch, static_clip, capsys):
+    import gemini
+    monkeypatch.setenv('GEMINI_API_KEY', 'unit-key')
+    for name in ('ask', 'upload_file'):
+        monkeypatch.setattr(gemini, name, lambda *a, **kw: pytest.fail('local engine must not contact Google'))
+    monkeypatch.setattr(sys, 'argv', ['watch', str(static_clip), '--engine', 'local', '--no-whisper'])
+    assert watch.main() == 0
+    assert '## Frames' in capsys.readouterr().out
+
+
+def test_no_key_keeps_the_local_pipeline(monkeypatch, static_clip, capsys):
+    monkeypatch.setattr(sys, 'argv', ['watch', str(static_clip), '--no-whisper'])
+    assert watch.main() == 0
+    assert '## Frames' in capsys.readouterr().out
+
+
+def test_explicit_gemini_without_key_fails_before_any_work(monkeypatch):
+    from config import ConfigError
+    monkeypatch.setattr(watch, 'download', lambda *a, **kw: pytest.fail('must fail before downloading'))
+    monkeypatch.setattr(sys, 'argv', ['watch', 'https://vimeo.com/1', '--engine', 'gemini'])
+    with pytest.raises(ConfigError, match='GEMINI_API_KEY'):
+        watch.main()