perf(pipeline): harden transcript action workflow cleanup

src/youtube_extension/services/workflows/transcript_action_workflow.py

@@ -849,13 +849,7 @@ async def _fallback_transcript_with_gemini(
                 if file_result.error:
                     errors.append(file_result.error)
             finally:
-                if video_path.exists():
-                    try:
-                        video_path.unlink()
-                    except OSError:
-                        pass
-                if temp_root and temp_root.exists():
-                    shutil.rmtree(temp_root, ignore_errors=True)
+                await self._cleanup_download_artifacts(video_path, temp_root)
 
         error_message = errors[0] if errors else "Gemini transcription failed"
         logger.warning("Gemini transcription fallback failed: %s", error_message)
@@ -867,6 +861,65 @@ async def _fallback_transcript_with_gemini(
             "error": error_message,
         }
 
+    @staticmethod
+    async def _cleanup_download_artifacts(
+        video_path: Path | None,
+        temp_root: Path | None,
+    ) -> None:
+        """Remove downloaded video artifacts without blocking the event loop.
+
+        The Gemini file fallback downloads a whole video into a temporary tree.
+        Because the format chain may fall back to separate video/audio streams,
+        that tree can hold the merged output plus unmerged ``.fNNN`` fragments,
+        so the removal is unbounded disk work. It therefore runs in a worker
+        thread using the same ``to_thread`` idiom as ``_download_video_file``.
+
+        The await is shielded because this runs from a ``finally`` block. The
+        previous inline implementation was synchronous and so uncancellable,
+        which meant cleanup always completed; an unshielded await would let a
+        cancellation delivered during the ``finally`` skip cleanup and leak the
+        tree. Shielding preserves that "always cleans up" property while still
+        yielding the loop, and still re-raises ``CancelledError`` to the caller.
+
+        ``CancelledError`` is deliberately allowed to propagate rather than
+        being suppressed in favour of any exception already in flight. Python
+        chains the in-flight exception onto it as ``__context__``, so no
+        diagnostic information is lost, whereas swallowing it would report a
+        cancelled task as ``cancelled() is False`` and defeat
+        ``asyncio.timeout``. The caller in ``_extract_transcript`` catches
+        ``Exception``, so a suppressed cancellation would be downgraded to a
+        per-source error and the pipeline would keep issuing network calls
+        after the request was abandoned.
+        """
+
+        def _cleanup() -> None:
+            # Both branches are total. This runs from a ``finally``, so raising
+            # here would replace whatever exception is already propagating.
+            #
+            # Note the absence of ``exists()`` probes: ``exists()`` performs a
+            # stat and can itself raise, which is exactly the masking this must
+            # avoid. ``unlink`` already raises ``FileNotFoundError`` for absent
+            # paths and ``rmtree`` tolerates them.
+            #
+            # The guards catch ``Exception``, not ``OSError``, because neither
+            # call is OSError-total. A path holding a NUL byte makes ``unlink``
+            # raise ``ValueError: embedded null character``, and makes
+            # ``rmtree`` raise the same from its internal ``lstat`` despite
+            # ``ignore_errors=True`` -- that flag only suppresses ``OSError``.
+            # ``CancelledError`` is a ``BaseException``, so it still propagates.
+            if video_path is not None:
+                try:
+                    video_path.unlink()
+                except Exception:  # noqa: BLE001 - must not mask in-flight error
+                    logger.debug("Cleanup failed for %s", video_path, exc_info=True)
+            if temp_root is not None:
+                try:
+                    shutil.rmtree(temp_root, ignore_errors=True)
+                except Exception:  # noqa: BLE001 - must not mask in-flight error
+                    logger.debug("Cleanup failed for %s", temp_root, exc_info=True)
+
+        await asyncio.shield(asyncio.to_thread(_cleanup))
+
     async def _record_metric(
         self,
         name: str,

tests/unit/test_transcript_action_workflow.py

@@ -1,6 +1,11 @@
 from __future__ import annotations
 
+import asyncio
 import datetime
+import pathlib
+import shutil
+import threading
+import time
 from dataclasses import asdict
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
@@ -1189,3 +1194,217 @@ async def test_success_via_youtube_url(self):
 
         assert result["text"] == "Gemini transcript"
         assert result["source"] == "gemini_video"
+
+
+# ---------------------------------------------------------------------------
+# _cleanup_download_artifacts
+# ---------------------------------------------------------------------------
+
+class TestCleanupDownloadArtifactsOffEventLoop:
+    """The Gemini fallback must not delete downloaded video trees inline.
+
+    ``_fallback_transcript_with_gemini`` downloads a full video into a
+    temporary tree, so the ``finally`` cleanup is unbounded disk work. These
+    tests pin that work to a worker thread rather than the event loop.
+    """
+
+    async def test_cleanup_runs_off_event_loop(self, monkeypatch, tmp_path):
+        loop_thread = threading.get_ident()
+        seen: dict[str, int] = {}
+        calls = {"unlink": 0, "rmtree": 0}
+
+        real_unlink = pathlib.Path.unlink
+        real_rmtree = shutil.rmtree
+
+        def recording_unlink(self, *args, **kwargs):
+            seen["unlink"] = threading.get_ident()
+            calls["unlink"] += 1
+            return real_unlink(self, *args, **kwargs)
+
+        def recording_rmtree(path, *args, **kwargs):
+            seen["rmtree"] = threading.get_ident()
+            calls["rmtree"] += 1
+            return real_rmtree(path, *args, **kwargs)
+
+        monkeypatch.setattr(pathlib.Path, "unlink", recording_unlink)
+        monkeypatch.setattr(shutil, "rmtree", recording_rmtree)
+
+        temp_root = tmp_path / "gemini_video_abc"
+        temp_root.mkdir()
+        video_path = temp_root / "auJzb1D-fag.mp4"
+        video_path.write_bytes(b"video-bytes")
+
+        await TranscriptActionWorkflow._cleanup_download_artifacts(
+            video_path, temp_root
+        )
+
+        # Guard against a vacuous pass: both primitives must really have run.
+        assert calls == {"unlink": 1, "rmtree": 1}
+        assert seen["unlink"] != loop_thread
+        assert seen["rmtree"] != loop_thread
+
+    async def test_cleanup_does_not_block_event_loop(self, monkeypatch, tmp_path):
+        started = threading.Event()
+        release = threading.Event()
+
+        def blocking_rmtree(path, *args, **kwargs):
+            started.set()
+            # Fail instead of hanging CI if the loop never gets to resume.
+            assert release.wait(timeout=10), "event loop blocked during cleanup"
+
+        monkeypatch.setattr(shutil, "rmtree", blocking_rmtree)
+
+        temp_root = tmp_path / "gemini_video_abc"
+        temp_root.mkdir()
+
+        cleanup = asyncio.create_task(
+            TranscriptActionWorkflow._cleanup_download_artifacts(None, temp_root)
+        )
+
+        # Poll against a wall-clock deadline rather than a fixed iteration
+        # count: a loaded CI box may take a while to hand the cleanup closure a
+        # worker thread, and a fixed budget would fail for scheduling reasons
+        # rather than for the behaviour under test. The polling itself is the
+        # assertion -- each completed tick is one turn of the event loop taken
+        # while rmtree is parked -- so this cannot be replaced by a blocking
+        # wait without destroying what the test proves.
+        deadline = time.monotonic() + 30.0
+        ticks = 0
+        while not started.is_set():
+            assert time.monotonic() < deadline, "cleanup never started"
+            await asyncio.sleep(0.01)
+            ticks += 1
+
+        # Reaching here while rmtree is still parked proves the loop kept
+        # running concurrently with the deletion.
+        assert ticks >= 1, "event loop never yielded while cleanup was running"
+        assert not cleanup.done()
+
+        release.set()
+        await asyncio.wait_for(cleanup, timeout=10)
+
+    async def test_cleanup_removes_artifacts(self, tmp_path):
+        temp_root = tmp_path / "gemini_video_abc"
+        temp_root.mkdir()
+        video_path = temp_root / "auJzb1D-fag.mp4"
+        video_path.write_bytes(b"video-bytes")
+        fragment = temp_root / "auJzb1D-fag.f140.m4a"
+        fragment.write_bytes(b"audio-bytes")
+
+        await TranscriptActionWorkflow._cleanup_download_artifacts(
+            video_path, temp_root
+        )
+
+        assert not video_path.exists()
+        assert not fragment.exists()
+        assert not temp_root.exists()
+
+    async def test_cleanup_survives_missing_paths(self, tmp_path):
+        # Mirrors the pre-existing guards: absent artifacts are not an error.
+        await TranscriptActionWorkflow._cleanup_download_artifacts(
+            tmp_path / "gone.mp4", tmp_path / "gone_dir"
+        )
+        await TranscriptActionWorkflow._cleanup_download_artifacts(None, None)
+
+    async def test_cleanup_completes_when_task_cancelled(self, monkeypatch, tmp_path):
+        # The replaced inline code was synchronous and therefore uncancellable,
+        # so cleanup always ran. The shielded await must preserve that.
+        started = threading.Event()
+        finished = threading.Event()
+        real_rmtree = shutil.rmtree
+
+        def slow_rmtree(path, *args, **kwargs):
+            started.set()
+            time.sleep(0.3)
+            real_rmtree(path, *args, **kwargs)
+            finished.set()
+
+        monkeypatch.setattr(shutil, "rmtree", slow_rmtree)
+
+        temp_root = tmp_path / "gemini_video_abc"
+        temp_root.mkdir()
+        (temp_root / "auJzb1D-fag.mp4").write_bytes(b"video-bytes")
+
+        task = asyncio.create_task(
+            TranscriptActionWorkflow._cleanup_download_artifacts(None, temp_root)
+        )
+
+        # Wall-clock deadline rather than a fixed iteration budget: a loaded
+        # CI runner can stretch each sleep well past 10ms.
+        deadline = time.monotonic() + 30.0
+        while not started.is_set():
+            assert time.monotonic() < deadline, "cleanup never started"
+            await asyncio.sleep(0.01)
+
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+        assert finished.wait(timeout=10), "shielded cleanup did not finish"
+        assert not temp_root.exists()
+
+        # Let the shielded inner task settle before the loop closes.
+        await asyncio.sleep(0.1)
+
+    async def test_cleanup_does_not_mask_in_flight_exception(
+        self, monkeypatch, tmp_path
+    ):
+        """A cleanup failure must never replace the exception being propagated.
+
+        The helper runs from a ``finally``. If it raised, it would discard the
+        real error and report a spurious filesystem fault instead. Both removal
+        primitives are therefore total.
+        """
+
+        def exploding_unlink(self, *args, **kwargs):
+            raise PermissionError("read-only filesystem")
+
+        def exploding_stat(self, *args, **kwargs):
+            raise OSError("stat exploded")
+
+        monkeypatch.setattr(pathlib.Path, "unlink", exploding_unlink)
+        # ``Path.exists()`` is implemented via ``stat()``. Patching stat proves
+        # the helper never probes a path in a way that could itself raise.
+        # ``shutil.rmtree`` is left real: its ``ignore_errors=True`` is the
+        # documented mechanism that makes the directory removal total, so
+        # patching it away would test a guarantee the code never claimed.
+        monkeypatch.setattr(pathlib.Path, "stat", exploding_stat)
+
+        class Boom(Exception):
+            pass
+
+        async def failing_operation():
+            try:
+                raise Boom("the real error")
+            finally:
+                await TranscriptActionWorkflow._cleanup_download_artifacts(
+                    tmp_path / "video.mp4", tmp_path / "absent_tree"
+                )
+
+        # The original exception survives; no OSError leaks out of cleanup.
+        with pytest.raises(Boom, match="the real error"):
+            await failing_operation()
+
+    async def test_cleanup_is_total_for_non_oserror_failures(self):
+        """A NUL byte in either path must not escape as ``ValueError``.
+
+        ``shutil.rmtree(..., ignore_errors=True)`` only suppresses ``OSError``:
+        a NUL byte makes its internal ``lstat`` raise ``ValueError``, and
+        ``Path.unlink`` raises the same directly. Because the helper runs from
+        a ``finally``, either escape would replace the in-flight exception --
+        the exact defect the removal of the ``exists()`` probes fixed. No
+        mocking is used, so this exercises the real stdlib behaviour.
+        """
+        nul_video = pathlib.Path("/tmp/eventrelay-nul\x00.mp4")
+        nul_root = pathlib.Path("/tmp/eventrelay-nul\x00-dir")
+
+        # Premise: the bare calls really are not OSError-total.
+        with pytest.raises(ValueError, match="null"):
+            nul_video.unlink()
+        with pytest.raises(ValueError, match="null"):
+            shutil.rmtree(nul_root, ignore_errors=True)
+
+        # Contract: the helper swallows both and returns normally.
+        await TranscriptActionWorkflow._cleanup_download_artifacts(
+            nul_video, nul_root
+        )