video automations: post-download scan chain (parity with music's batch→scan→update)

Mirrors the music flow: a finished download batch → refresh the media server → (after
it indexes) pull the new media into the DB — so a downloaded movie/episode shows as
owned without waiting for the 6h scheduled scan.

Two event-based system automations (owned_by='video'):
  - 'Auto-Scan Video After Downloads' (video_batch_complete → video_scan_server)
  - 'Auto-Update Video Database After Scan' (video_library_scan_completed → video_update_database)

Pieces:
- core/video/download_events.py: a callback registry (core/video can't import the
  engine — isolation). The monitor publishes batch-complete; web_server bridges it to
  automation_engine.emit('video_batch_complete', …), like music's web_scan_manager.
- download_monitor: fires the batch-complete event once, when the last in-flight
  download finishes (none queued/downloading/searching left).
- video_scan_server handler (stage 1): refresh server, wait a debounce for indexing,
  then emit 'video_library_scan_completed' (mirrors music's time-based completion).
- video_update_database handler (stage 2): incremental read (newest-first, stop after
  25 consecutive known — same as music).
- blocks: 2 video triggers + 2 video actions (scope='video'); registration; seeds.

kettui: seam tests for both handlers (refresh/wait/emit, incremental read, error paths),
the event registry (idempotent register, isolated failures), and the monitor (fires once
on last completion, never while work remains). 298 automation tests + isolation green.

core/automation/blocks.py

@@ -123,6 +123,14 @@ TRIGGERS: list[dict] = [
     {"type": "webhook_received", "label": "Webhook Received", "icon": "globe", "scope": "both",
      "description": "When an external API request is received (POST /api/v1/request)", "available": True,
      "variables": ["query", "request_id", "source"]},
+
+    # ── Video side (scope='video') — the post-download scan chain triggers ──
+    {"type": "video_batch_complete", "label": "Video Download Batch Done", "icon": "check-circle", "scope": "video",
+     "description": "When a batch of video downloads finishes", "available": True,
+     "variables": ["completed"]},
+    {"type": "video_library_scan_completed", "label": "Video Library Scan Done", "icon": "hard-drive", "scope": "video",
+     "description": "When the media server finishes rescanning your video sections", "available": True,
+     "variables": ["server"]},
 ]
 
 
@@ -212,6 +220,21 @@ ACTIONS: list[dict] = [
                       {"value": "deep", "label": "Deep (also remove missing)"}],
           "default": "full"}
      ]},
+    # Post-download chain actions (two stages, like music's scan_library +
+    # start_database_update). Stage 1 nudges the server; stage 2 reads it in.
+    {"type": "video_scan_server", "label": "Scan Video Server", "icon": "refresh", "scope": "video",
+     "description": "Tell the media server to rescan your video sections, then fire 'Video Library Scan Done'", "available": True,
+     "config_fields": [
+         {"key": "debounce_seconds", "type": "number", "label": "Wait for indexing (sec)", "default": 120, "min": 10}
+     ]},
+    {"type": "video_update_database", "label": "Update Video Database", "icon": "database", "scope": "video",
+     "description": "Read newly-indexed media from the server into SoulSync (incremental)", "available": True,
+     "config_fields": [
+         {"key": "mode", "type": "select", "label": "Mode",
+          "options": [{"value": "incremental", "label": "Incremental (recent only)"},
+                      {"value": "full", "label": "Full (add + refresh)"}],
+          "default": "incremental"}
+     ]},
 ]
 
 

core/automation/handlers/registration.py

@@ -35,7 +35,9 @@ from core.automation.handlers.download_cleanup import (
 )
 from core.automation.handlers.run_script import auto_run_script
 from core.automation.handlers.search_and_download import auto_search_and_download
-from core.automation.handlers.video_scan_library import auto_video_scan_library
+from core.automation.handlers.video_scan_library import (
+    auto_video_scan_library, auto_video_scan_server, auto_video_update_database,
+)
 from core.automation.handlers.progress_callbacks import (
     progress_init,
     progress_finish,
@@ -175,6 +177,15 @@ def register_all(deps: AutomationDeps) -> None:
         'video_scan_library',
         lambda config: auto_video_scan_library(config, deps),
     )
+    # Post-download chain: scan the server, then (on the scan-done event) update the DB.
+    engine.register_action_handler(
+        'video_scan_server',
+        lambda config: auto_video_scan_server(config, deps),
+    )
+    engine.register_action_handler(
+        'video_update_database',
+        lambda config: auto_video_update_database(config, deps),
+    )
 
     # Progress + history callbacks: the engine invokes these around
     # each handler run. Lift the closures from

core/automation/handlers/video_scan_library.py

@@ -25,6 +25,7 @@ phase string with a generic 'completed' label.
 
 from __future__ import annotations
 
+import time
 from typing import Any, Callable, Dict, Optional
 
 from core.automation.deps import AutomationDeps
@@ -143,3 +144,78 @@ def auto_video_scan_library(
             log_type='error',
         )
         return {'status': 'error', 'error': str(e), '_manages_own_progress': True}
+
+
+# ── post-download chain (video twin of music's scan_library → start_database_update) ──
+# Split into two stages so the calendar/library reflect a download promptly, mirroring
+# the music side: tell the server to rescan, then (after it has had time to index) read
+# the new state into video.db. Stage 1 emits 'video_library_scan_completed' on a debounce
+# (like music's web_scan_manager time-based completion); stage 2 listens for that.
+
+def auto_video_scan_server(
+    config: Dict[str, Any],
+    deps: AutomationDeps,
+    *,
+    server_refresh: Optional[Callable[[], Dict[str, Any]]] = None,
+    sleep: Optional[Callable[[float], None]] = None,
+    emit: Optional[Callable[[str, Dict[str, Any]], None]] = None,
+) -> Dict[str, Any]:
+    """Stage 1: tell the media server to rescan the video sections, wait a debounce
+    for it to index, then fire 'video_library_scan_completed' so the DB-update twin
+    reads the fresh state. (Video twin of music's scan_library.)"""
+    server_refresh = server_refresh or _default_server_refresh
+    sleep = sleep or time.sleep
+    emit = emit or deps.engine.emit
+    automation_id = config.get('_automation_id')
+    try:
+        debounce = int(config.get('debounce_seconds') or 120)
+    except (TypeError, ValueError):
+        debounce = 120
+    try:
+        deps.update_progress(automation_id, phase='Asking media server to rescan…', progress=20,
+                             log_line='Triggering server-side video scan', log_type='info')
+        refresh = server_refresh() or {}
+        if not refresh.get('ok'):
+            deps.update_progress(automation_id,
+                                 log_line='Server scan trigger unavailable: ' + str(refresh.get('error') or 'unknown'),
+                                 log_type='warning')
+        deps.update_progress(automation_id, phase=f'Waiting for the server to index ({debounce}s)…', progress=55)
+        sleep(debounce)
+        emit('video_library_scan_completed', {'server': refresh.get('server') or ''})
+        deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
+                             log_line='Server scan done — updating the database', log_type='success')
+        return {'status': 'completed', '_manages_own_progress': True}
+    except Exception as e:  # noqa: BLE001
+        deps.update_progress(automation_id, status='error', phase='Error', log_line=str(e), log_type='error')
+        return {'status': 'error', 'error': str(e), '_manages_own_progress': True}
+
+
+def auto_video_update_database(
+    config: Dict[str, Any],
+    deps: AutomationDeps,
+    *,
+    run_video_scan: Optional[Callable[[str], Dict[str, Any]]] = None,
+) -> Dict[str, Any]:
+    """Stage 2: read the (now-rescanned) server into video.db — INCREMENTAL by default
+    (newest-first, stop after N consecutive known). Video twin of start_database_update."""
+    run_video_scan = run_video_scan or _default_run_video_scan
+    automation_id = config.get('_automation_id')
+    mode = config.get('mode') or 'incremental'
+    try:
+        deps.update_progress(automation_id, phase='Reading new media into SoulSync…', progress=40)
+        result = run_video_scan(mode) or {}
+        if result.get('state') == 'error':
+            err = result.get('error') or 'Video database update failed'
+            deps.update_progress(automation_id, status='error', phase='Error', log_line=err, log_type='error')
+            return {'status': 'error', 'error': err, '_manages_own_progress': True}
+        movies = int(result.get('movies', 0) or 0)
+        shows = int(result.get('shows', 0) or 0)
+        episodes = int(result.get('episodes', 0) or 0)
+        deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
+                             log_line=f'Video database updated: {movies} movies, {shows} shows, {episodes} episodes',
+                             log_type='success')
+        return {'status': 'completed', '_manages_own_progress': True,
+                'movies': movies, 'shows': shows, 'episodes': episodes}
+    except Exception as e:  # noqa: BLE001
+        deps.update_progress(automation_id, status='error', phase='Error', log_line=str(e), log_type='error')
+        return {'status': 'error', 'error': str(e), '_manages_own_progress': True}

core/automation_engine.py

@@ -161,6 +161,24 @@ SYSTEM_AUTOMATIONS = [
         # app startup; post-download freshness is handled by the event chain instead.
         'initial_delay': 6 * 60 * 60,
     },
+    # Post-download chain (video twin of music's batch_complete → scan_library →
+    # library_scan_completed → start_database_update). Event-based, so a finished
+    # video download refreshes the server then pulls the new media into video.db.
+    {
+        'name': 'Auto-Scan Video After Downloads',
+        'trigger_type': 'video_batch_complete',
+        'trigger_config': {},
+        'action_type': 'video_scan_server',
+        'owned_by': 'video',
+    },
+    {
+        'name': 'Auto-Update Video Database After Scan',
+        'trigger_type': 'video_library_scan_completed',
+        'trigger_config': {},
+        'action_type': 'video_update_database',
+        'action_config': {'mode': 'incremental'},
+        'owned_by': 'video',
+    },
 ]
 
 

core/video/download_events.py

@@ -0,0 +1,40 @@
+"""Video download events — a tiny publish/subscribe bridge.
+
+ISOLATION: ``core/video`` must not import the automation engine (that's music-side
+shared infra). So when a batch of video downloads finishes, the monitor publishes
+to this callback registry instead of touching the engine directly. The shared side
+(web_server) registers a callback that forwards to
+``automation_engine.emit('video_batch_complete', …)`` — the same one-way bridge the
+music ``web_scan_manager`` uses for ``library_scan_completed``.
+"""
+
+from __future__ import annotations
+
+from typing import Callable
+
+from utils.logging_config import get_logger
+
+logger = get_logger("video_download_events")
+
+_batch_complete_callbacks: list[Callable[[dict], None]] = []
+
+
+def register_batch_complete_callback(cb: Callable[[dict], None]) -> None:
+    """Subscribe to 'a batch of video downloads just finished'. Idempotent."""
+    if cb not in _batch_complete_callbacks:
+        _batch_complete_callbacks.append(cb)
+
+
+def notify_batch_complete(data: dict | None = None) -> None:
+    """Fire every registered batch-complete callback (best-effort; one failing
+    subscriber never blocks the others or the monitor)."""
+    payload = data or {}
+    for cb in list(_batch_complete_callbacks):
+        try:
+            cb(payload)
+        except Exception:
+            logger.exception("video batch-complete callback failed")
+
+
+def _reset_for_tests() -> None:
+    _batch_complete_callbacks.clear()

core/video/download_monitor.py

@@ -234,11 +234,14 @@ def _tick(db) -> None:
     download_dir = str(config_manager.get("soulseek.download_path", "") or "")
     transfers = list_downloads()
     live_ids = set()
+    completed_now = 0
     for dl in active:
         live_ids.add(dl["id"])
         upd = process_download(dl, transfers, download_dir, lister=_walk, mover=_move)
         if not upd:
             continue
+        if upd.get("status") == "completed":
+            completed_now += 1
         if upd.get("_missing"):
             n = _misses.get(dl["id"], 0) + 1
             _misses[dl["id"]] = n
@@ -258,6 +261,16 @@ def _tick(db) -> None:
             logger.exception("video download %s: failed to persist update", dl.get("id"))
     for k in [k for k in _misses if k not in live_ids]:
         _misses.pop(k, None)
+    # Batch complete: we finished ≥1 download this tick AND nothing is left in
+    # flight (queued/downloading/searching). Fires once, on the transition to
+    # empty — the next tick early-returns. Publishes to the event bridge so the
+    # 'Auto-Scan Video After Downloads' automation can refresh the server.
+    if completed_now and not db.get_active_video_downloads():
+        try:
+            from core.video.download_events import notify_batch_complete
+            notify_batch_complete({"completed": completed_now})
+        except Exception:
+            logger.exception("video monitor: batch-complete notify failed")
 
 
 def _run(db_provider) -> None:

tests/automation/test_automation_blocks.py

@@ -125,6 +125,19 @@ def test_music_scope_matches_legacy_full_lists_minus_video():
         assert {b['type'] for b in music[key]} == expected
 
 
+def test_post_download_chain_blocks_are_video_scoped():
+    music = blocks.blocks_for_scope('music')
+    video = blocks.blocks_for_scope('video')
+    trig = {t['type'] for t in video['triggers']}
+    act = {a['type'] for a in video['actions']}
+    assert {'video_batch_complete', 'video_library_scan_completed'} <= trig
+    assert {'video_scan_server', 'video_update_database'} <= act
+    mtrig = {t['type'] for t in music['triggers']}
+    mact = {a['type'] for a in music['actions']}
+    assert not ({'video_batch_complete', 'video_library_scan_completed'} & mtrig)
+    assert not ({'video_scan_server', 'video_update_database'} & mact)
+
+
 def test_video_scan_library_block_shape():
     action = next(a for a in blocks.ACTIONS if a['type'] == 'video_scan_library')
     assert action['scope'] == 'video'

tests/automation/test_handler_registration.py

@@ -55,6 +55,8 @@ EXPECTED_ACTION_NAMES = frozenset({
     'search_and_download',
     # Video side (isolated app, shared engine).
     'video_scan_library',
+    'video_scan_server',
+    'video_update_database',
 })
 
 # Action names that MUST register a guard (duplicate-run prevention).

tests/automation/test_handler_video_scan_library.py

@@ -173,3 +173,65 @@ class TestHandlerNeverRaises:
             {}, deps, server_refresh=_refresh_ok(), run_video_scan=_scan_done(),
         )
         assert result['status'] == 'completed'
+
+
+# ── post-download chain: video_scan_server (stage 1) ───────────────────────
+from core.automation.handlers.video_scan_library import (  # noqa: E402
+    auto_video_scan_server, auto_video_update_database)
+
+
+class TestScanServerStage:
+    def test_refreshes_waits_then_emits_scan_done(self):
+        deps = _RecordingDeps()
+        events = []
+        slept = []
+        r = auto_video_scan_server(
+            {'_automation_id': 'a', 'debounce_seconds': 90}, deps,
+            server_refresh=_refresh_ok(), sleep=lambda s: slept.append(s),
+            emit=lambda ev, data: events.append((ev, data)))
+        assert r['status'] == 'completed'
+        assert slept == [90]                                  # waited the debounce
+        assert events == [('video_library_scan_completed', {'server': ''})]
+
+    def test_default_debounce_and_server_unavailable_still_emits(self):
+        deps = _RecordingDeps()
+        events = []
+        auto_video_scan_server(
+            {'_automation_id': 'a'}, deps,
+            server_refresh=lambda: {'ok': False, 'error': 'no server'},
+            sleep=lambda s: None, emit=lambda ev, data: events.append(ev))
+        assert events == ['video_library_scan_completed']     # fires even if refresh failed
+        assert 'warning' in deps.log_types()
+
+    def test_never_raises(self):
+        deps = _RecordingDeps()
+        r = auto_video_scan_server(
+            {'_automation_id': 'a'}, deps,
+            server_refresh=lambda: (_ for _ in ()).throw(RuntimeError('boom')),
+            sleep=lambda s: None, emit=lambda *a: None)
+        assert r['status'] == 'error' and r['error'] == 'boom'
+
+
+# ── post-download chain: video_update_database (stage 2) ───────────────────
+class TestUpdateDatabaseStage:
+    def test_incremental_read_returns_counts(self):
+        deps = _RecordingDeps()
+        r = auto_video_update_database({'_automation_id': 'a'}, deps, run_video_scan=_scan_done(2, 1, 5))
+        assert r == {'status': 'completed', '_manages_own_progress': True,
+                     'movies': 2, 'shows': 1, 'episodes': 5}
+
+    def test_defaults_to_incremental_mode(self):
+        seen = {}
+
+        def _scan(mode):
+            seen['mode'] = mode
+            return {'state': 'done'}
+
+        auto_video_update_database({'_automation_id': 'a'}, _RecordingDeps(), run_video_scan=_scan)
+        assert seen['mode'] == 'incremental'
+
+    def test_scan_error_propagates(self):
+        deps = _RecordingDeps()
+        r = auto_video_update_database({'_automation_id': 'a'}, deps,
+                                       run_video_scan=lambda m: {'state': 'error', 'error': 'no server'})
+        assert r['status'] == 'error' and r['error'] == 'no server'

tests/test_video_download_events.py

@@ -0,0 +1,79 @@
+"""Video download → batch-complete event bridge + the monitor detection that fires it.
+
+The isolated monitor (core/video) publishes 'batch complete' to a callback registry;
+web_server bridges that to automation_engine.emit('video_batch_complete', …). These
+pin the publish side: the registry, and the monitor firing exactly once when the last
+download finishes (never while work is still in flight).
+"""
+
+from __future__ import annotations
+
+import core.video.download_monitor as mon
+import core.video.download_events as events
+
+
+def setup_function(_):
+    events._reset_for_tests()
+
+
+# ── the event registry ─────────────────────────────────────────────────────
+
+def test_register_and_notify_fires_callbacks():
+    got = []
+    events.register_batch_complete_callback(lambda d: got.append(d))
+    events.notify_batch_complete({"completed": 3})
+    assert got == [{"completed": 3}]
+
+
+def test_register_is_idempotent_and_one_failure_is_isolated():
+    got = []
+    cb = lambda d: got.append(d)
+    events.register_batch_complete_callback(cb)
+    events.register_batch_complete_callback(cb)          # dup → ignored
+    events.register_batch_complete_callback(lambda d: (_ for _ in ()).throw(RuntimeError("boom")))
+    events.notify_batch_complete({})                     # must not raise despite the bad cb
+    assert got == [{}]                                   # the good cb fired exactly once
+
+
+# ── the monitor fires it on the last completion ────────────────────────────
+
+class _FakeDb:
+    def __init__(self, active):
+        self._active = list(active)
+
+    def get_active_video_downloads(self):
+        return list(self._active)
+
+    def update_video_download(self, dl_id, **kw):
+        if kw.get("status") in ("completed", "failed", "cancelled"):
+            self._active = [d for d in self._active if d["id"] != dl_id]
+
+
+def _patch(monkeypatch, result):
+    monkeypatch.setattr(mon, "list_downloads", lambda: [])
+    monkeypatch.setattr(mon, "process_download", lambda dl, *a, **k: dict(result))
+
+
+def test_tick_fires_batch_complete_when_last_download_finishes(monkeypatch):
+    fired = []
+    events.register_batch_complete_callback(lambda d: fired.append(d))
+    _patch(monkeypatch, {"status": "completed", "progress": 100.0})
+    db = _FakeDb([{"id": 1, "status": "downloading", "filename": "a.mkv"}])
+    mon._tick(db)
+    assert fired == [{"completed": 1}]                   # one download done, none left → fired once
+
+
+def test_tick_does_not_fire_while_a_download_is_still_active(monkeypatch):
+    fired = []
+    events.register_batch_complete_callback(lambda d: fired.append(d))
+    # one completes, one is still downloading → batch NOT done yet
+    monkeypatch.setattr(mon, "list_downloads", lambda: [])
+
+    def _proc(dl, *a, **k):
+        return {"status": "completed", "progress": 100.0} if dl["id"] == 1 else {"progress": 50.0}
+
+    monkeypatch.setattr(mon, "process_download", _proc)
+    db = _FakeDb([{"id": 1, "status": "downloading", "filename": "a.mkv"},
+                  {"id": 2, "status": "downloading", "filename": "b.mkv"}])
+    mon._tick(db)
+    assert fired == []                                   # 2 still active → no batch-complete

web_server.py

@@ -1259,6 +1259,17 @@ def _register_automation_handlers():
     )
     _register_extracted_handlers(_automation_deps)
 
+    # Bridge the isolated video download monitor's batch-complete signal into the
+    # automation engine (core/video can't import the engine). Mirrors how the music
+    # web_scan_manager forwards library_scan_completed. Fires the 'Auto-Scan Video
+    # After Downloads' automation when a batch of video downloads finishes.
+    if automation_engine is not None:
+        try:
+            from core.video.download_events import register_batch_complete_callback
+            register_batch_complete_callback(
+                lambda data: automation_engine.emit('video_batch_complete', data or {}))
+        except Exception:
+            logger.exception("Could not wire video batch-complete -> automation engine")
 
     logger.info("Automation action handlers registered")