diff --git a/app/library/downloads/status_tracker.py b/app/library/downloads/status_tracker.py
index 5a29a77c..a836f081 100644
--- a/app/library/downloads/status_tracker.py
+++ b/app/library/downloads/status_tracker.py
@@ -56,6 +56,26 @@ class StatusTracker:
         self._terminator_sent: bool = False
         self._candidate_filepath: Path | None = None
         self.update_task: asyncio.Task | None = None
+        self._last_emit_time: float = 0.0
+        self._last_status: str | None = None
+        self._pending_emit: bool = False
+        self._emit_interval: float = 0.5
+
+    def _emit_update(self, force: bool = False) -> None:
+        now = time.monotonic()
+        status_changed = self.info.status != self._last_status
+
+        if force or status_changed or (now - self._last_emit_time) >= self._emit_interval:
+            self._notify.emit(Events.ITEM_UPDATED, data=self.info)
+            self._last_emit_time = now
+            self._last_status = self.info.status
+            self._pending_emit = False
+        else:
+            self._pending_emit = True
+
+    def flush_pending(self) -> None:
+        if self._pending_emit:
+            self._emit_update(force=True)
 
     async def _finalize_file(self, filepath: Path) -> None:
         """
@@ -235,7 +255,7 @@ class StatusTracker:
             await self._finalize_file(Path(final_name))
             self.info.status = "finished"
 
-        self._notify.emit(Events.ITEM_UPDATED, data=self.info)
+        self._emit_update()
 
     async def progress_update(self) -> None:
         """
@@ -248,6 +268,7 @@ class StatusTracker:
                 self.update_task = asyncio.get_running_loop().run_in_executor(None, self.status_queue.get)
                 status = await self.update_task
                 if status is None or isinstance(status, Terminator):
+                    self.flush_pending()
                     return
                 await self.process_status_update(status)
             except (asyncio.CancelledError, OSError, FileNotFoundError, EOFError, BrokenPipeError, ConnectionError):
@@ -295,6 +316,8 @@ class StatusTracker:
             except (queue.Empty, BrokenPipeError, ConnectionRefusedError, EOFError, OSError):
                 continue
 
+        self.flush_pending()
+
     def cancel_update_task(self) -> None:
         """Cancel the progress update task if it's running."""
         try:
diff --git a/app/tests/test_download.py b/app/tests/test_download.py
index 22228ce1..3bcc96be 100644
--- a/app/tests/test_download.py
+++ b/app/tests/test_download.py
@@ -1211,6 +1211,74 @@ class TestStatusTracker:
         assert 1 == len(queue.items), "Should add terminator to queue"
         assert isinstance(queue.items[0], Terminator), "Should add Terminator instance"
 
+    @pytest.mark.asyncio
+    async def test_throttle_skips_rapid_progress_updates(self, mock_config: dict) -> None:
+        st = StatusTracker(**mock_config)
+        emit_calls: list = []
+        st._notify = Mock()
+        st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a))
+
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100})
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 200})
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 300})
+
+        item_updated_calls = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED]
+        assert len(item_updated_calls) == 1, "Should throttle rapid progress updates to one emission"
+        assert st._pending_emit is True, "Should mark pending emit for throttled updates"
+
+    @pytest.mark.asyncio
+    async def test_throttle_always_emits_on_status_change(self, mock_config: dict) -> None:
+        st = StatusTracker(**mock_config)
+        emit_calls: list = []
+        st._notify = Mock()
+        st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a))
+
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100})
+        await st.process_status_update({"id": "test-id", "status": "error", "error": "fail"})
+
+        item_updated_calls = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED]
+        assert len(item_updated_calls) == 2, "Should emit immediately when status changes"
+
+    @pytest.mark.asyncio
+    async def test_flush_pending_sends_throttled_update(self, mock_config: dict) -> None:
+        st = StatusTracker(**mock_config)
+        emit_calls: list = []
+        st._notify = Mock()
+        st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a))
+
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100})
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 200})
+
+        before = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED]
+        assert len(before) == 1, "Second update should be throttled"
+
+        st.flush_pending()
+
+        after = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED]
+        assert len(after) == 2, "flush_pending should emit the throttled update"
+        assert st._pending_emit is False, "Should clear pending flag after flush"
+
+    def test_flush_pending_noop_when_no_pending(self, mock_config: dict) -> None:
+        st = StatusTracker(**mock_config)
+        st._notify = Mock()
+        st._notify.emit = Mock()
+        st.flush_pending()
+        st._notify.emit.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_throttle_disabled_when_interval_is_zero(self, mock_config: dict) -> None:
+        st = StatusTracker(**mock_config)
+        st._emit_interval = 0.0  # disable throttle
+        emit_calls: list = []
+        st._notify = Mock()
+        st._notify.emit = Mock(side_effect=lambda *a, **kw: emit_calls.append(a))
+
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 100})
+        await st.process_status_update({"id": "test-id", "status": "downloading", "downloaded_bytes": 200})
+
+        item_updated_calls = [c for c in emit_calls if c[0] == Events.ITEM_UPDATED]
+        assert len(item_updated_calls) == 2, "Should emit all updates when interval is 0"
+
 
 class TestQueueManager:
     @staticmethod