feat(downloader): NF26 — yt-dlp auto-update active-download guard

maybe_auto_update_ytdlp() used to fire fire-and-forget; yt-dlp -U
atomically replaces the binary, and on Windows an in-flight
subprocess.Popen([YTDLP_PATH, ...]) could race the replace with
file-in-use errors.

New optional active_count_fn parameter — when supplied and returns
> 0, the update is deferred and the next 24h throttle window picks
it up. Caller in the GUI server-start path now passes
self.dl_manager.active_count so the check consults the live queue
without coupling the function to the manager instance.

Probe failures (raising callable) fall through to 'proceed with
warning' — failure mode of an under-construction probe is at least
as bad as racing the self-replace.

Back-compat preserved: calling maybe_auto_update_ytdlp(config)
without the new arg still works exactly as before.

Touches:
  astra_downloader/astra_downloader.py — signature + guard at the
    top of the function (lines 981-1020 region); call site at the
    GUI server-start hook (line 3758) now passes the active_count
    callable
  astra_downloader/test_astra_downloader.py — new
    AutoUpdateActiveDownloadGuardTests class with four cases:
      fires_when_zero / defers_when_positive /
      proceeds_when_probe_raises / back_compat_no_arg

Closes NF26 from RESEARCH_FEATURE_PLAN.

Verified: 92/92 Python tests pass (+4 new).

Author: SysAdminDoc

CHANGELOG.md

@@ -6,6 +6,25 @@ All notable changes to Astra Deck are documented here. Versions are listed newes
 
 ## [Unreleased]
 
+- **astra_downloader: yt-dlp auto-update active-download guard (NF26).**
+  `maybe_auto_update_ytdlp()` (astra_downloader.py:981) used to fire
+  fire-and-forget; `yt-dlp.exe -U` atomically replaces the binary, and
+  on Windows an in-flight `subprocess.Popen([YTDLP_PATH, ...])` could
+  race the replace with file-in-use errors. New optional
+  `active_count_fn` parameter — when supplied and returns > 0, the
+  update is deferred and the next 24h throttle window picks it up.
+  Caller in the GUI server-start path now passes
+  `self.dl_manager.active_count` so the check consults the live queue
+  without coupling the function to the manager instance. Probe
+  failures (raising callable) fall through to "proceed with warning"
+  — failure mode of an under-construction probe is at least as bad
+  as racing the self-replace. Back-compat preserved: calling
+  `maybe_auto_update_ytdlp(config)` without the new arg still works
+  exactly as before. Pinned by four new `AutoUpdateActiveDownloadGuardTests`
+  in `test_astra_downloader.py`:
+  fires-when-zero / defers-when-positive / proceeds-when-probe-raises
+  / back-compat-no-arg. 92/92 Python tests pass (+4 new).
+
 - **check-versions: SETTINGS_VERSION parity gate (NF25).** The product
   version is enforced across 5 sources by `scripts/check-versions.js`;
   the schema-version namespace was not. Drift caught: `popup.js`

RESEARCH_FEATURE_PLAN.md

@@ -37,11 +37,6 @@ Each item carries: priority, complexity, why, evidence, touches, acceptance, ver
 
 ### Settings / downloader hardening (P0 batch from 2026-05-25 audit)
 
-- **P0 / M — yt-dlp auto-update active-download guard (NF26)**
-  - Why: `astra_downloader.py:981-1018` `maybe_auto_update_ytdlp()` fires fire-and-forget at `:3758` with no `active_count() == 0` guard. Windows file-locking can fail an in-flight download mid-`-U`.
-  - Touches: `astra_downloader/astra_downloader.py`, `astra_downloader/test_astra_downloader.py`.
-  - Acceptance: pytest with `active_count` mock returning >0 → update deferred; ==0 → update fires.
-
 - **P0 / M — Deno cutoff hard-gate on `/download` (NF27)**
   - Why: yt-dlp ≥ 2026.04.01 requires Deno. The `denoRuntime` probe at `astra_downloader.py:751-804` reports state but `/download` doesn't consult it; yt-dlp returns empty format lists with an opaque error.
   - Touches: `astra_downloader.py` (`/download` handler), `extension/ytkit.js` MediaDLManager error path.

astra_downloader/astra_downloader.py

@@ -978,18 +978,48 @@ def mark_ytdlp_update_check(config):
         write_persistent_log(f"Could not persist yt-dlp update timestamp: {e}")
 
 
-def maybe_auto_update_ytdlp(config):
+def maybe_auto_update_ytdlp(config, active_count_fn=None):
     """Background-run yt-dlp -U when more than a day has passed.
 
     Fire-and-forget in a daemon thread so startup isn't blocked. The exit code
     is logged (previously swallowed entirely).
+
+    v4.47.0 NF26: when ``active_count_fn`` is provided and returns > 0, the
+    update is deferred. yt-dlp's ``-U`` flag atomically replaces the binary,
+    and on Windows the in-flight ``subprocess.Popen([YTDLP_PATH, ...])`` of
+    an active download can race the replace with file-in-use errors. Callers
+    in the GUI / server path pass ``dl_manager.active_count`` so the check
+    consults the live queue without coupling this function to the manager
+    instance.
+
+    The check is racy by design — we accept that a download started during
+    the millisecond between ``active_count_fn()`` returning 0 and ``-U``
+    spawning the replacement process can still race. Mitigation in practice
+    is that ``-U`` runs ahead of any user download (server-start hook), and
+    a download starting in that micro-window is so unlikely we don't pay the
+    cost of a hard cross-process lock.
     """
     if not YTDLP_PATH.exists():
         return
     if not config.get("AutoUpdateYtDlp", True):
         return
     if not should_check_ytdlp_update(config):
         return
+    if active_count_fn is not None:
+        try:
+            in_flight = int(active_count_fn() or 0)
+        except Exception as e:  # noqa: BLE001
+            # reason: active_count_fn is caller-supplied; if it raises we
+            # must NOT block the update, since the caller's failure mode
+            # is at least as bad as racing a self-replace.
+            write_persistent_log(f"yt-dlp auto-update active-count probe failed: {e}")
+            in_flight = 0
+        if in_flight > 0:
+            write_persistent_log(
+                f"yt-dlp auto-update deferred — {in_flight} active download(s); "
+                f"next check at the configured 24h throttle."
+            )
+            return
 
     def run():
         try:
@@ -3755,7 +3785,10 @@ def run():
         # Auto-update yt-dlp — throttled (once per 24h) so we don't re-run
         # it on every single launch. Logs exit code instead of silently
         # discarding it.
-        maybe_auto_update_ytdlp(self.config)
+        #
+        # v4.47.0 NF26: pass the manager's active_count so an in-flight
+        # download isn't raced by a yt-dlp.exe self-replace.
+        maybe_auto_update_ytdlp(self.config, self.dl_manager.active_count)
 
     def _stop_server(self):
         if self.server_obj:

astra_downloader/test_astra_downloader.py

@@ -693,6 +693,130 @@ def get(self, key, default=None):
         self.assertTrue(ad.should_check_ytdlp_update(_C()))
 
 
+class AutoUpdateActiveDownloadGuardTests(unittest.TestCase):
+    """v4.47.0 NF26 — yt-dlp's ``-U`` flag atomically replaces the binary.
+    On Windows an in-flight ``subprocess.Popen([YTDLP_PATH, ...])`` of an
+    active download can race the replace with file-in-use errors. The
+    guard takes a caller-supplied ``active_count_fn`` and defers the
+    update when the function reports any in-flight downloads.
+    """
+
+    def _fake_config(self, last_check_stamp=""):
+        # Empty stamp triggers should_check_ytdlp_update -> True.
+        class _C:
+            def __init__(self, stamp):
+                self._d = {
+                    "AutoUpdateYtDlp": True,
+                    "LastYtDlpUpdateCheck": stamp,
+                }
+            def get(self, key, default=None):
+                return self._d.get(key, default)
+            def set(self, key, value):
+                self._d[key] = value
+            def save(self):
+                pass
+        return _C(last_check_stamp)
+
+    def test_update_fires_when_no_active_downloads(self):
+        # active_count_fn returning 0 must NOT defer the update; the
+        # threading.Thread must be spawned (we mock it to track the spawn
+        # without actually running yt-dlp).
+        spawned = {"count": 0}
+        orig_thread = ad.threading.Thread
+        try:
+            class _FakeThread:
+                def __init__(self, target=None, daemon=None):
+                    self._target = target
+                def start(self):
+                    spawned["count"] += 1
+                    # Don't actually run -U; this tests the guard, not the
+                    # subprocess invocation.
+            ad.threading.Thread = _FakeThread
+            # YTDLP_PATH must exist for the guard to fall through to the
+            # active-count check; patch its existence check.
+            ad.YTDLP_PATH = type(ad.YTDLP_PATH)(ad.YTDLP_PATH)
+            with mock.patch.object(ad.YTDLP_PATH.__class__, 'exists', return_value=True):
+                ad.maybe_auto_update_ytdlp(self._fake_config(), active_count_fn=lambda: 0)
+        finally:
+            ad.threading.Thread = orig_thread
+        self.assertEqual(spawned["count"], 1,
+                         "Update thread must spawn when active_count == 0")
+
+    def test_update_defers_when_active_downloads_in_flight(self):
+        # active_count_fn returning > 0 must defer the update; no thread
+        # should spawn.
+        spawned = {"count": 0}
+        log_lines = []
+        orig_thread = ad.threading.Thread
+        orig_log = ad.write_persistent_log
+        try:
+            class _FakeThread:
+                def __init__(self, target=None, daemon=None):
+                    pass
+                def start(self):
+                    spawned["count"] += 1
+            ad.threading.Thread = _FakeThread
+            ad.write_persistent_log = lambda msg: log_lines.append(msg)
+            with mock.patch.object(ad.YTDLP_PATH.__class__, 'exists', return_value=True):
+                ad.maybe_auto_update_ytdlp(self._fake_config(), active_count_fn=lambda: 3)
+        finally:
+            ad.threading.Thread = orig_thread
+            ad.write_persistent_log = orig_log
+        self.assertEqual(spawned["count"], 0,
+                         "Update thread must NOT spawn when active_count > 0")
+        self.assertTrue(any("auto-update deferred" in line for line in log_lines),
+                        f"Defer log line must surface in persistent log; got {log_lines!r}")
+        self.assertTrue(any("3 active download" in line for line in log_lines),
+                        f"Defer log must include the count; got {log_lines!r}")
+
+    def test_update_proceeds_when_active_count_fn_raises(self):
+        # Caller-supplied probe failure must not block the update.
+        # Failure mode of an under-construction probe is at least as bad
+        # as racing a self-replace, so we prefer "proceed with warning".
+        spawned = {"count": 0}
+        log_lines = []
+        orig_thread = ad.threading.Thread
+        orig_log = ad.write_persistent_log
+        try:
+            class _FakeThread:
+                def __init__(self, target=None, daemon=None):
+                    pass
+                def start(self):
+                    spawned["count"] += 1
+            ad.threading.Thread = _FakeThread
+            ad.write_persistent_log = lambda msg: log_lines.append(msg)
+
+            def broken_probe():
+                raise RuntimeError("probe boom")
+            with mock.patch.object(ad.YTDLP_PATH.__class__, 'exists', return_value=True):
+                ad.maybe_auto_update_ytdlp(self._fake_config(), active_count_fn=broken_probe)
+        finally:
+            ad.threading.Thread = orig_thread
+            ad.write_persistent_log = orig_log
+        self.assertEqual(spawned["count"], 1,
+                         "Update thread must still spawn when probe raises")
+        self.assertTrue(any("probe failed" in line for line in log_lines),
+                        f"Probe-failure log line must surface; got {log_lines!r}")
+
+    def test_update_without_active_count_fn_proceeds(self):
+        # Back-compat: existing callers without the new arg must still work.
+        spawned = {"count": 0}
+        orig_thread = ad.threading.Thread
+        try:
+            class _FakeThread:
+                def __init__(self, target=None, daemon=None):
+                    pass
+                def start(self):
+                    spawned["count"] += 1
+            ad.threading.Thread = _FakeThread
+            with mock.patch.object(ad.YTDLP_PATH.__class__, 'exists', return_value=True):
+                ad.maybe_auto_update_ytdlp(self._fake_config())
+        finally:
+            ad.threading.Thread = orig_thread
+        self.assertEqual(spawned["count"], 1,
+                         "Update must still fire when no active_count_fn provided")
+
+
 class NoArchiveLockTests(unittest.TestCase):
     """v1.3.0 removed the download-archive lock so re-downloads always
     run. These tests pin the invariants so the lock can't be silently