diff --git a/src/napari_plugin_manager/_tests/test_qt_plugin_dialog.py b/src/napari_plugin_manager/_tests/test_qt_plugin_dialog.py
index 48a1510..5f2a216 100644
--- a/src/napari_plugin_manager/_tests/test_qt_plugin_dialog.py
+++ b/src/napari_plugin_manager/_tests/test_qt_plugin_dialog.py
@@ -666,3 +666,26 @@ def test_import_plugins(plugin_dialog, tmp_path, qtbot):
     path.write_text('requests\npackaging\n')
     with qtbot.waitSignal(plugin_dialog.installer.allFinished, timeout=60_000):
         plugin_dialog.import_plugins(str(path))
+
+
+def test_query_status(plugin_dialog, monkeypatch):
+    status, description = plugin_dialog.query_status()
+    assert status == qt_plugin_dialog.Status.COMPLETED
+    assert not description
+
+    monkeypatch.setattr(
+        plugin_dialog.installer,
+        '_queue',
+        ['mock'],
+    )
+    status, description = plugin_dialog.query_status()
+    assert status == qt_plugin_dialog.Status.BUSY
+    assert description
+
+    monkeypatch.setattr(
+        plugin_dialog.installer,
+        '_queue',
+        ['mock', 'other-mock'],
+    )
+    assert status == qt_plugin_dialog.Status.BUSY
+    assert description
diff --git a/src/napari_plugin_manager/base_qt_package_installer.py b/src/napari_plugin_manager/base_qt_package_installer.py
index c32d8c5..0c3e569 100644
--- a/src/napari_plugin_manager/base_qt_package_installer.py
+++ b/src/napari_plugin_manager/base_qt_package_installer.py
@@ -477,7 +477,7 @@ def cancel_all(self) -> None:
                 process.finished.disconnect(self._on_process_finished)
                 process.errorOccurred.disconnect(self._on_error_occurred)
 
-            self._end_process(process)
+            self._end_process(process, wait_for_finished=True)
 
         self._queue.clear()
         self._current_process = None
@@ -588,7 +588,7 @@ def _process_queue(self) -> None:
             process.start()
             self._current_process = process
 
-    def _end_process(self, process: QProcess) -> None:
+    def _end_process(self, process: QProcess, wait_for_finished=False) -> None:
         if os.name == 'nt':
             # TODO: this might be too agressive and won't allow rollbacks!
             # investigate whether we can also do .terminate()
@@ -596,6 +596,9 @@ def _end_process(self, process: QProcess) -> None:
         else:
             process.terminate()
 
+        if wait_for_finished:
+            process.waitForFinished()
+
         if self._output_widget:
             self._output_widget.append(
                 trans._('\nTask was cancelled by the user.')
@@ -654,6 +657,10 @@ def _on_process_done(
             )
 
         if item is not None:
+            if not isinstance(exit_code, int):
+                exit_code = 0
+                if error:
+                    exit_code = 1
             self.processFinished.emit(
                 {
                     'exit_code': exit_code,
diff --git a/src/napari_plugin_manager/base_qt_plugin_dialog.py b/src/napari_plugin_manager/base_qt_plugin_dialog.py
index 2a71ec3..f453d19 100644
--- a/src/napari_plugin_manager/base_qt_plugin_dialog.py
+++ b/src/napari_plugin_manager/base_qt_plugin_dialog.py
@@ -1,8 +1,10 @@
 import contextlib
 import importlib.metadata
 import os
+import uuid
 import webbrowser
-from collections.abc import Iterable, Sequence
+from collections.abc import Callable, Iterable, Sequence
+from enum import Enum, auto
 from functools import partial
 from logging import getLogger
 from typing import (
@@ -67,6 +69,15 @@
 log = getLogger(__name__)
 
 
+class Status(Enum):
+    PENDING = auto()
+    BUSY = auto()
+    COMPLETED = auto()
+    CANCELLED = auto()
+    FAILED = auto()
+    STARTED_FAILED = auto()
+
+
 class PackageMetadataProtocol(Protocol):
     """
     Protocol class defining the minimum atributtes/properties needed for package metadata.
@@ -1048,6 +1059,7 @@ def __init__(self, parent: QDialog = None, prefix=None) -> None:
         self._plugin_data = []  # Store all plugin data
         self._filter_texts = []
         self._filter_idxs_cache = set()
+        self._task_status_id = None
         self.worker = None
         self._plugin_data_map = {}
         self._add_items_timer = QTimer(self)
@@ -1130,6 +1142,25 @@ def _update_theme(self, event: Any) -> None:
         """
         raise NotImplementedError
 
+    def _register_task_status(self) -> None:
+        status, description = self.query_status()
+
+        if self._task_status_id is not None:
+            self._update_task_status(status, description=description)
+            return
+
+        self._task_status_id = self.register_task_status(
+            status, description, cancel_callback=self.installer.cancel_all
+        )
+
+    def _update_task_status(
+        self, status: Status = Status.COMPLETED, description: str = ''
+    ):
+        if self._task_status_id is not None:
+            self.update_task_status(
+                self._task_status_id, status, description=description
+            )
+
     def _on_installer_start(self) -> None:
         """Updates dialog buttons and status when installing a plugin."""
         self.cancel_all_btn.setVisible(True)
@@ -1137,6 +1168,7 @@ def _on_installer_start(self) -> None:
         self.process_success_indicator.hide()
         self.process_error_indicator.hide()
         self.refresh_button.setDisabled(True)
+        self._register_task_status()
 
     def _on_process_finished(
         self, process_finished_data: ProcessFinishedData
@@ -1203,18 +1235,21 @@ def _on_installer_all_finished(self, exit_codes: Iterable[int]) -> None:
         self.close_btn.setDisabled(False)
         self.refresh_button.setDisabled(False)
 
+        if sum(exit_codes) > 0:
+            message = self._trans(
+                'Plugin Manager: process completed with errors\n'
+            )
+            status = Status.FAILED
+            show_message = self._show_warning
+        else:
+            message = self._trans('Plugin Manager: process completed\n')
+            status = Status.COMPLETED
+            show_message = self._show_info
+
         if not self.isVisible():
-            if sum(exit_codes) > 0:
-                self._show_warning(
-                    self._trans(
-                        'Plugin Manager: process completed with errors\n'
-                    )
-                )
-            else:
-                self._show_info(
-                    self._trans('Plugin Manager: process completed\n')
-                )
+            show_message(message)
 
+        self._update_task_status(status=status, description=message)
         self.search()
 
     def _add_to_installed(
@@ -1791,7 +1826,7 @@ def search(self, text: str | None = None, skip=False) -> None:
         if len(text.strip()) == 0:
             self.installed_list.filter('')
             self.available_widget.setCurrentWidget(self.available_message)
-            self._plugin_queue = None
+            self._plugin_queue = []
             self._add_items_timer.stop()
             self._plugins_found = 0
         else:
@@ -1810,7 +1845,7 @@ def search(self, text: str | None = None, skip=False) -> None:
                 self._plugins_found = len(items)
                 self._add_items_timer.start()
             else:
-                self._plugin_queue = None
+                self._plugin_queue = []
                 self._add_items_timer.stop()
                 self._plugins_found = 0
 
@@ -1829,6 +1864,7 @@ def refresh(self, clear_cache: bool = False) -> None:
         self._plugin_queue = []
         self._plugin_data = []
         self._plugin_data_map = {}
+        self._latest_status = None
 
         self.installed_list.clear()
         self.available_list.clear()
@@ -1886,4 +1922,30 @@ def import_plugins(self, fpath: str) -> None:
         plugins = [p for p in plugins if p]
         self._install_packages(plugins)
 
+    def register_task_status(
+        self,
+        status: Status,
+        description: str,
+        cancel_callback: Callable | None = None,
+    ) -> uuid.UUID:
+        """Register a task status for the plugin manager."""
+        raise NotImplementedError
+
+    def update_task_status(
+        self, task_status_id: uuid.UUID, status: Status, description: str = ''
+    ) -> bool:
+        """Update task status for the plugin manager."""
+        raise NotImplementedError
+
+    def query_status(self) -> tuple[Status, str]:
+        """
+        Return the current status of plugins installations.
+
+        Returns
+        -------
+        A tuple containing the current status (`Status`) and a description.
+
+        """
+        raise NotImplementedError
+
     # endregion - Public methods
diff --git a/src/napari_plugin_manager/qt_plugin_dialog.py b/src/napari_plugin_manager/qt_plugin_dialog.py
index e27a50a..12899b6 100644
--- a/src/napari_plugin_manager/qt_plugin_dialog.py
+++ b/src/napari_plugin_manager/qt_plugin_dialog.py
@@ -1,4 +1,6 @@
 import sys
+import uuid
+from collections.abc import Callable
 from pathlib import Path
 
 import napari.plugins
@@ -37,6 +39,12 @@
 from napari_plugin_manager.qt_package_installer import NapariInstallerQueue
 from napari_plugin_manager.utils import is_conda_package
 
+try:
+    from napari.utils.task_status import Status
+except ImportError:
+    from napari_plugin_manager.base_qt_plugin_dialog import Status
+
+
 # Scaling factor for each list widget item when expanding.
 STYLES_PATH = Path(__file__).parent / 'styles.qss'
 DISMISS_WARN_PYPI_INSTALL_DLG = False
@@ -267,6 +275,49 @@ def _show_warning(self, warning: str) -> None:
     def _trans(self, text: str, **kwargs) -> str:
         return trans._(text, **kwargs)
 
+    def register_task_status(
+        self,
+        task_status: Status,
+        description: str,
+        cancel_callback: Callable | None = None,
+    ) -> uuid.UUID:
+        window = getattr(self._parent, '_window', None)
+        if window and hasattr(window, '_register_task_status'):
+            return window._register_task_status(
+                'napari-plugin-manager',
+                task_status,
+                description,
+                cancel_callback=cancel_callback,
+            )
+        return None
+
+    def update_task_status(
+        self,
+        task_status_id: uuid.UUID,
+        status: Status,
+        description: str = '',
+    ) -> bool:
+        window = getattr(self._parent, '_window', None)
+        if window and hasattr(window, '_update_task_status'):
+            return window._update_task_status(
+                task_status_id, status, description=description
+            )
+        return False
+
+    def query_status(self) -> tuple[Status, str]:
+        if self.installer.hasJobs():
+            task_status = Status.BUSY
+            description = trans._n(
+                'The plugin manager is currently busy with {n} task.',
+                'The plugin manager is currently busy with {n} tasks.',
+                n=self.installer.currentJobs(),
+            )
+        else:
+            task_status = Status.COMPLETED
+            description = ''
+
+        return task_status, description
+
 
 if __name__ == '__main__':
     from qtpy.QtWidgets import QApplication