diff --git a/src/youtube_extension/backend/services/performance_monitor.py b/src/youtube_extension/backend/services/performance_monitor.py
index 26f6f1e64..1c89fccb0 100644
--- a/src/youtube_extension/backend/services/performance_monitor.py
+++ b/src/youtube_extension/backend/services/performance_monitor.py
@@ -270,7 +270,8 @@ async def record_metric(self,
 
     async def _store_metric(self, metric: PerformanceMetric):
         """Store metric in database"""
-        try:
+
+        def _write() -> None:
             conn = sqlite3.connect(self.db_path)
             cursor = conn.cursor()
 
@@ -290,6 +291,8 @@ async def _store_metric(self, metric: PerformanceMetric):
             conn.commit()
             conn.close()
 
+        try:
+            await asyncio.to_thread(_write)
         except Exception as e:
             logger.error(f"Failed to store metric in database: {e}")
 
@@ -353,7 +356,8 @@ async def _trigger_alert(self, alert: PerformanceAlert):
 
     async def _store_alert(self, alert: PerformanceAlert):
         """Store alert in database"""
-        try:
+
+        def _write() -> None:
             conn = sqlite3.connect(self.db_path)
             cursor = conn.cursor()
 
@@ -375,6 +379,8 @@ async def _store_alert(self, alert: PerformanceAlert):
             conn.commit()
             conn.close()
 
+        try:
+            await asyncio.to_thread(_write)
         except Exception as e:
             logger.error(f"Failed to store alert in database: {e}")
 
@@ -498,7 +504,8 @@ async def _cleanup_old_metrics(self):
 
     async def _basic_cleanup(self):
         """Basic cleanup fallback when service is not available"""
-        try:
+
+        def _purge() -> None:
             cutoff_date = (datetime.now(timezone.utc) - timedelta(days=30)).isoformat()
 
             conn = sqlite3.connect(self.db_path)
@@ -518,8 +525,9 @@ async def _basic_cleanup(self):
             conn.commit()
             conn.close()
 
+        try:
+            await asyncio.to_thread(_purge)
             logger.info("Basic cleanup completed successfully")
-
         except Exception as e:
             logger.error(f"Error in basic cleanup: {e}")
 
@@ -562,7 +570,8 @@ async def trigger_manual_cleanup(self) -> dict[str, Any]:
 
     async def get_current_performance_summary(self) -> dict[str, dict[str, float]]:
         """Get current performance summary (last hour averages)"""
-        try:
+
+        def _query() -> dict[str, dict[str, float]]:
             conn = sqlite3.connect(self.db_path)
             cursor = conn.cursor()
 
@@ -582,6 +591,8 @@ async def get_current_performance_summary(self) -> dict[str, dict[str, float]]:
             conn.close()
             return dict(results)
 
+        try:
+            return await asyncio.to_thread(_query)
         except Exception as e:
             logger.error(f"Error getting performance summary: {e}")
             return {}
@@ -658,7 +669,8 @@ async def _get_target_progress(self) -> dict[str, Any]:
 
     async def _get_recent_metrics_summary(self) -> dict[str, Any]:
         """Get recent metrics summary"""
-        try:
+
+        def _query() -> dict[str, Any]:
             conn = sqlite3.connect(self.db_path)
             cursor = conn.cursor()
 
@@ -683,6 +695,8 @@ async def _get_recent_metrics_summary(self) -> dict[str, Any]:
             conn.close()
             return metrics_summary
 
+        try:
+            return await asyncio.to_thread(_query)
         except Exception as e:
             logger.error(f"Error getting recent metrics summary: {e}")
             return {}
@@ -798,7 +812,8 @@ async def run_performance_benchmark(self,
 
     async def _store_benchmark_result(self, result: dict[str, Any]):
         """Store benchmark result in database"""
-        try:
+
+        def _write() -> None:
             conn = sqlite3.connect(self.db_path)
             cursor = conn.cursor()
 
@@ -824,6 +839,8 @@ async def _store_benchmark_result(self, result: dict[str, Any]):
             conn.commit()
             conn.close()
 
+        try:
+            await asyncio.to_thread(_write)
         except Exception as e:
             logger.error(f"Failed to store benchmark result: {e}")
 
diff --git a/tests/unit/test_performance_monitor.py b/tests/unit/test_performance_monitor.py
index 682c578f8..b5fcf603c 100644
--- a/tests/unit/test_performance_monitor.py
+++ b/tests/unit/test_performance_monitor.py
@@ -17,14 +17,20 @@
 sys.modules['psutil'] = _real_psutil
 sys.modules.pop('youtube_extension.backend.services.performance_monitor', None)
 
+import asyncio
+import contextlib
+import sqlite3
+import time
 from collections import deque
 from datetime import datetime, timezone
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "src"))
 
+from youtube_extension.backend.services import performance_monitor as perf_mod
 from youtube_extension.backend.services.performance_monitor import (
     PerformanceAlert,
     PerformanceMetric,
@@ -1002,3 +1008,139 @@ def my_sync_func():
 
         result = my_sync_func()
         assert result == "hello"
+
+
+# ===========================================================================
+# SQLite I/O must not block the event loop
+# ===========================================================================
+
+
+class TestSqliteDoesNotBlockEventLoop:
+    """`sqlite3` is fully synchronous: connect + INSERT + commit (fsync) + close.
+
+    `PerformanceMonitor.record_metric` runs on live request paths
+    (`api/v1/router.py:1165` and `:1183`), so every one of those calls used to
+    stall the event loop for the duration of the write.  These tests measure
+    whether an independent coroutine keeps getting scheduled while the database
+    work is in flight.
+    """
+
+    _DB_SECONDS = 0.15
+    _HEARTBEAT_INTERVAL = 0.01
+
+    @pytest.fixture
+    def monitor(self, tmp_path):
+        return PerformanceMonitor(db_path=str(tmp_path / "perf.db"))
+
+    def _slow_connect(self, real_connect):
+        """Wrap sqlite3.connect so the *synchronous* work takes a visible time."""
+
+        def _connect(*args, **kwargs):
+            time.sleep(self._DB_SECONDS)
+            return real_connect(*args, **kwargs)
+
+        return _connect
+
+    async def _count_heartbeats_during(self, coro):
+        ticks = 0
+        stop = False
+
+        async def heartbeat():
+            nonlocal ticks
+            while not stop:
+                await asyncio.sleep(self._HEARTBEAT_INTERVAL)
+                ticks += 1
+
+        beat = asyncio.create_task(heartbeat())
+        await asyncio.sleep(0)  # let the heartbeat reach its first await first
+        try:
+            result = await coro
+        finally:
+            stop = True
+            beat.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await beat
+        return result, ticks
+
+    async def test_store_metric_does_not_block_event_loop(self, monitor):
+        metric = PerformanceMetric(
+            component="api",
+            metric_name="api_response_time",
+            value=12.5,
+            timestamp=datetime.now(timezone.utc),
+            unit="ms",
+            tags={},
+        )
+        real = perf_mod.sqlite3.connect
+        with patch.object(perf_mod.sqlite3, "connect", self._slow_connect(real)):
+            _, ticks = await self._count_heartbeats_during(monitor._store_metric(metric))
+        assert ticks > 0, "event loop was blocked for the whole sqlite write"
+
+    async def test_record_metric_does_not_block_event_loop(self, monitor):
+        """The public request-path entry point, not just the private writer."""
+        real = perf_mod.sqlite3.connect
+        with patch.object(perf_mod.sqlite3, "connect", self._slow_connect(real)):
+            _, ticks = await self._count_heartbeats_during(
+                monitor.record_metric("api", "api_response_time", 12.5)
+            )
+        assert ticks > 0, "record_metric blocked the event loop"
+
+    async def test_read_path_does_not_block_event_loop(self, monitor):
+        real = perf_mod.sqlite3.connect
+        with patch.object(perf_mod.sqlite3, "connect", self._slow_connect(real)):
+            _, ticks = await self._count_heartbeats_during(
+                monitor.get_current_performance_summary()
+            )
+        assert ticks > 0, "the read path blocked the event loop"
+
+    # --- preserved-behaviour guards (pass under old and new code alike) -----
+
+    async def test_store_metric_still_writes_the_row(self, monitor):
+        metric = PerformanceMetric(
+            component="api",
+            metric_name="api_response_time",
+            value=42.0,
+            timestamp=datetime.now(timezone.utc),
+            unit="ms",
+            tags={"route": "/x"},
+        )
+        await monitor._store_metric(metric)
+
+        conn = sqlite3.connect(monitor.db_path)
+        rows = conn.execute(
+            "SELECT component, metric_name, value, unit FROM performance_metrics"
+        ).fetchall()
+        conn.close()
+        assert rows == [("api", "api_response_time", 42.0, "ms")]
+
+    async def test_summary_reflects_stored_metrics(self, monitor):
+        for value in (10.0, 20.0):
+            await monitor._store_metric(
+                PerformanceMetric(
+                    component="api",
+                    metric_name="api_response_time",
+                    value=value,
+                    timestamp=datetime.now(timezone.utc),
+                    unit="ms",
+                    tags={},
+                )
+            )
+        summary = await monitor.get_current_performance_summary()
+        assert summary["api"]["api_response_time"] == 15.0
+
+    async def test_store_metric_swallows_database_errors(self, monitor):
+        """Error handling must still absorb failures rather than propagate."""
+        metric = PerformanceMetric(
+            component="api",
+            metric_name="api_response_time",
+            value=1.0,
+            timestamp=datetime.now(timezone.utc),
+            unit="ms",
+            tags={},
+        )
+
+        def _boom(*_args, **_kwargs):
+            raise sqlite3.OperationalError("disk I/O error")
+
+        with patch.object(perf_mod.sqlite3, "connect", _boom):
+            await monitor._store_metric(metric)  # must not raise