perf: read local image bytes off the event loop in vision providers

src/youtube_extension/integrations/cloud_ai/providers/azure_vision.py

@@ -8,6 +8,7 @@
 - Custom Vision (if configured)
 """
 
+import asyncio
 import logging
 from datetime import datetime
 from typing import Any, Optional
@@ -29,6 +30,12 @@
 logger = logging.getLogger(__name__)
 
 
+def _read_file_bytes(path: str) -> bytes:
+    """Read a file's bytes. Module-level so it can run in a worker thread."""
+    with open(path, 'rb') as handle:
+        return handle.read()
+
+
 class AzureVision(BaseCloudAI):
     """Microsoft Azure AI Vision integration."""
 
@@ -251,9 +258,8 @@ async def _prepare_image_input(self, image_url: str) -> Optional[bytes]:
             # For URL input, Azure can analyze directly
             return None
         else:
-            # For local files, read content
-            with open(image_url, 'rb') as image_file:
-                return image_file.read()
+            # Local file - read off the event loop
+            return await asyncio.to_thread(_read_file_bytes, image_url)
 
     async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs: Any) -> Any:
         """Run a blocking Azure SDK call in a worker thread, bounded by a
@@ -263,8 +269,6 @@ async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs
         ``asyncio.wait_for`` enforces the remaining budget so neither a slow read
         nor a slow poll can run past the OCR timeout. Raises ``CloudAIError`` on
         expiry."""
-        import asyncio
-
         remaining = deadline - asyncio.get_running_loop().time()
         if remaining <= 0:
             raise CloudAIError("Azure OCR operation timed out")
@@ -277,8 +281,6 @@ async def _await_ocr_call(self, deadline: float, func: Any, *args: Any, **kwargs
 
     async def _perform_ocr(self, image_url: str, image_stream: Optional[bytes]) -> dict[str, Any]:
         """Perform OCR using Azure Read API."""
-        import asyncio
-
         from azure.cognitiveservices.vision.computervision.models import (
             OperationStatusCodes,
         )
@@ -322,8 +324,6 @@ async def _perform_ocr(self, image_url: str, image_stream: Optional[bytes]) -> d
 
     async def _perform_ocr_stream(self, image_stream) -> dict[str, Any]:
         """Perform OCR on image stream."""
-        import asyncio
-
         from azure.cognitiveservices.vision.computervision.models import (
             OperationStatusCodes,
         )

src/youtube_extension/integrations/cloud_ai/providers/google_cloud.py

@@ -31,6 +31,12 @@
 logger = logging.getLogger(__name__)
 
 
+def _read_file_bytes(path: str) -> bytes:
+    """Read a file's bytes. Module-level so it can run in a worker thread."""
+    with open(path, 'rb') as handle:
+        return handle.read()
+
+
 class GoogleCloudAI(BaseCloudAI):
     """Google Cloud Video Intelligence and Vision API integration."""
 
@@ -181,9 +187,8 @@ async def analyze_image(self, image_url: str,
             if image_url.startswith(('http://', 'https://')):
                 image.source.image_uri = image_url
             else:
-                # For local files
-                with open(image_url, 'rb') as image_file:
-                    image.content = image_file.read()
+                # Local file - read off the event loop
+                image.content = await asyncio.to_thread(_read_file_bytes, image_url)
 
             # Prepare features
             features = self._prepare_vision_features(analysis_types)

tests/unit/test_azure_vision_provider.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import builtins
 import sys
+import threading
 import types as _types
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -1194,3 +1196,92 @@ def test_cost_scales_with_analysis_types(self):
         cost1 = provider.estimate_cost(60.0, [AnalysisType.OCR])
         cost2 = provider.estimate_cost(60.0, [AnalysisType.OCR, AnalysisType.FACE_DETECTION])
         assert cost2 > cost1
+
+
+# ===========================================================================
+# Local image reads must not block the event loop
+# ===========================================================================
+
+
+class _ThreadRecordingHandle:
+    """File-object proxy that records the calling thread on every ``read``."""
+
+    def __init__(self, handle, threads):
+        self._handle = handle
+        self._threads = threads
+
+    def read(self, *args, **kwargs):
+        self._threads.append(threading.get_ident())
+        return self._handle.read(*args, **kwargs)
+
+    def __enter__(self):
+        self._handle.__enter__()
+        return self
+
+    def __exit__(self, *exc_info):
+        return self._handle.__exit__(*exc_info)
+
+    def __getattr__(self, name):
+        return getattr(self._handle, name)
+
+
+class _ThreadRecordingOpen:
+    """Wrap ``builtins.open`` and record which thread *reads* a target path.
+
+    Off-loop execution is asserted by *thread identity* rather than elapsed
+    wall-clock time, which is flaky on loaded CI runners. The recording hooks
+    ``read()`` on the returned handle rather than ``open()`` itself, so a
+    regression that opens the file on a worker thread but reads its bytes back
+    on the event loop is still caught. Only the target path is wrapped so
+    unrelated ``open`` traffic (logging, coverage) cannot contaminate the
+    result.
+    """
+
+    def __init__(self, target):
+        self._real_open = builtins.open
+        self._target = str(target)
+        self.threads: list[int] = []
+
+    def __call__(self, file, *args, **kwargs):
+        handle = self._real_open(file, *args, **kwargs)
+        if str(file) == self._target:
+            return _ThreadRecordingHandle(handle, self.threads)
+        return handle
+
+
+class TestAzureVisionImageReadOffEventLoop:
+    async def test_local_file_read_runs_on_worker_thread(self, tmp_path):
+        provider = _make_provider()
+        img_file = tmp_path / "frame.jpg"
+        img_file.write_bytes(b"\xff\xd8\xff\xe0")
+        recorder = _ThreadRecordingOpen(img_file)
+        loop_thread = threading.get_ident()
+
+        with patch("builtins.open", recorder):
+            result = await provider._prepare_image_input(str(img_file))
+
+        assert result == b"\xff\xd8\xff\xe0"
+        assert recorder.threads, "expected the provider to read the local image file"
+        assert loop_thread not in recorder.threads, (
+            "local image bytes were read on the event loop thread; the read must "
+            "be offloaded to a worker thread"
+        )
+
+    async def test_http_url_performs_no_disk_read(self, tmp_path):
+        """The URL branch must remain untouched: Azure fetches it directly."""
+        provider = _make_provider()
+        decoy = tmp_path / "unused.jpg"
+        decoy.write_bytes(b"\x00")
+        recorder = _ThreadRecordingOpen(decoy)
+
+        with patch("builtins.open", recorder):
+            result = await provider._prepare_image_input("https://example.com/img.jpg")
+
+        assert result is None
+        assert recorder.threads == []
+
+    async def test_missing_file_still_raises_file_not_found(self, tmp_path):
+        """Offloading must not swallow or re-wrap I/O errors."""
+        provider = _make_provider()
+        with pytest.raises(FileNotFoundError):
+            await provider._prepare_image_input(str(tmp_path / "does-not-exist.jpg"))

tests/unit/test_google_cloud_provider.py

@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import asyncio
+import builtins
 import sys
+import threading
 import types as _types
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -1115,3 +1117,144 @@ def test_zero_duration_zero_cost(self):
         provider = _make_provider()
         cost = provider.estimate_cost(0.0, [AnalysisType.LABEL_DETECTION])
         assert cost == pytest.approx(0.0)
+
+
+# ===========================================================================
+# Local image reads must not block the event loop
+# ===========================================================================
+
+
+class _ThreadRecordingHandle:
+    """File-object proxy that records the calling thread on every ``read``."""
+
+    def __init__(self, handle, threads):
+        self._handle = handle
+        self._threads = threads
+
+    def read(self, *args, **kwargs):
+        self._threads.append(threading.get_ident())
+        return self._handle.read(*args, **kwargs)
+
+    def __enter__(self):
+        self._handle.__enter__()
+        return self
+
+    def __exit__(self, *exc_info):
+        return self._handle.__exit__(*exc_info)
+
+    def __getattr__(self, name):
+        return getattr(self._handle, name)
+
+
+class _ThreadRecordingOpen:
+    """Wrap ``builtins.open`` and record which thread *reads* a target path.
+
+    Off-loop execution is asserted by *thread identity* rather than elapsed
+    wall-clock time, which is flaky on loaded CI runners. The recording hooks
+    ``read()`` on the returned handle rather than ``open()`` itself, so a
+    regression that opens the file on a worker thread but reads its bytes back
+    on the event loop is still caught. Only the target path is wrapped so
+    unrelated ``open`` traffic (logging, coverage) cannot contaminate the
+    result.
+    """
+
+    def __init__(self, target):
+        self._real_open = builtins.open
+        self._target = str(target)
+        self.threads: list[int] = []
+
+    def __call__(self, file, *args, **kwargs):
+        handle = self._real_open(file, *args, **kwargs)
+        if str(file) == self._target:
+            return _ThreadRecordingHandle(handle, self.threads)
+        return handle
+
+
+class TestGoogleCloudImageReadOffEventLoop:
+    def _vision_modules(self):
+        mock_image_instance = MagicMock()
+        mock_image_instance.source = MagicMock()
+        mock_image_cls = MagicMock(return_value=mock_image_instance)
+        mock_feature_type = MagicMock()
+        mock_feature_type.LABEL_DETECTION = "LABEL_DETECTION"
+        mock_feature_cls = MagicMock()
+        mock_feature_cls.Type = mock_feature_type
+        mock_vision = MagicMock()
+        mock_vision.Image = mock_image_cls
+        mock_vision.Feature = mock_feature_cls
+        return mock_vision
+
+    def _patched_modules(self, mock_vision):
+        return patch.dict("sys.modules", {
+            "google": _types.ModuleType("google"),
+            "google.cloud": _types.ModuleType("google.cloud"),
+            "google.cloud.vision": mock_vision,
+        })
+
+    def _client(self):
+        client = AsyncMock()
+        client.annotate_image = AsyncMock(return_value=_make_vision_response())
+        return client
+
+    async def test_local_file_read_runs_on_worker_thread(self, tmp_path):
+        provider = _make_provider()
+        provider._vision_client = self._client()
+        img_file = tmp_path / "frame.jpg"
+        img_file.write_bytes(b"\x89PNG\r\n")
+        mock_vision = self._vision_modules()
+        recorder = _ThreadRecordingOpen(img_file)
+        loop_thread = threading.get_ident()
+
+        with self._patched_modules(mock_vision), patch("builtins.open", recorder):
+            await provider.analyze_image(str(img_file), [AnalysisType.LABEL_DETECTION])
+
+        assert mock_vision.Image.return_value.content == b"\x89PNG\r\n"
+        assert recorder.threads, "expected the provider to read the local image file"
+        assert loop_thread not in recorder.threads, (
+            "local image bytes were read on the event loop thread; the read must "
+            "be offloaded to a worker thread"
+        )
+
+    async def test_http_url_performs_no_disk_read(self, tmp_path):
+        """The URI branch must remain untouched: Vision fetches it directly."""
+        provider = _make_provider()
+        provider._vision_client = self._client()
+        decoy = tmp_path / "unused.jpg"
+        decoy.write_bytes(b"\x00")
+        mock_vision = self._vision_modules()
+        recorder = _ThreadRecordingOpen(decoy)
+
+        with self._patched_modules(mock_vision), patch("builtins.open", recorder):
+            await provider.analyze_image(
+                "https://example.com/img.jpg", [AnalysisType.LABEL_DETECTION]
+            )
+
+        assert recorder.threads == []
+        assert mock_vision.Image.return_value.source.image_uri == "https://example.com/img.jpg"
+
+    async def test_missing_local_file_wrapped_in_cloud_ai_error(self, tmp_path):
+        """A missing local image surfaces as ``CloudAIError`` from the public API.
+
+        Unlike Azure's private ``_prepare_image_input`` (which propagates
+        ``FileNotFoundError``), Google's public ``analyze_image`` catches it in
+        its broad ``except Exception`` and re-raises as ``CloudAIError``. Pin
+        that wrapper contract so moving the read off the loop cannot silently
+        alter how a missing file is reported.
+        """
+        provider = _make_provider()
+        provider._vision_client = self._client()
+        missing = tmp_path / "does-not-exist.jpg"
+        mock_vision = self._vision_modules()
+
+        with self._patched_modules(mock_vision):
+            with pytest.raises(CloudAIError) as exc_info:
+                await provider.analyze_image(
+                    str(missing), [AnalysisType.LABEL_DETECTION]
+                )
+
+        # The provider re-raises without ``from e``, so the original error is
+        # carried on ``__context__`` (implicit chaining), not ``__cause__``.
+        assert isinstance(exc_info.value.__context__, FileNotFoundError), (
+            "offloading must preserve the underlying I/O error in the exception chain"
+        )
+        assert "No such file or directory" in str(exc_info.value)