diff --git a/lib/galaxy/managers/sse_dispatch.py b/lib/galaxy/managers/sse_dispatch.py
index c5d0e46a402..41a9370f8af 100644
--- a/lib/galaxy/managers/sse_dispatch.py
+++ b/lib/galaxy/managers/sse_dispatch.py
@@ -15,6 +15,7 @@ from collections.abc import Callable
 from typing import (
     Any,
     Optional,
+    Protocol,
 )
 
 from cachetools import TTLCache
@@ -29,6 +30,17 @@ from galaxy.queues import all_control_queues_for_declare
 from galaxy.web.statsd_client import VanillaGalaxyStatsdClient
 from galaxy.web_stack import ApplicationStack
 
+
+class ControlTaskLike(Protocol):
+    """Structural type for the dispatcher's control-task collaborator.
+
+    The dispatcher only calls ``send_task(**kwargs)``. Typed as a Protocol so
+    tests can pass lightweight fakes (``FakeControlTask``, ``NoopControlTask``)
+    without subclassing ``ControlTask``.
+    """
+
+    def send_task(self, **kwargs: Any) -> Any: ...
+
 log = logging.getLogger(__name__)
 
 
@@ -56,7 +68,7 @@ class SSEEventDispatcher:
         application_stack: ApplicationStack,
         statsd_client: Optional[VanillaGalaxyStatsdClient] = None,
         clock: Callable[[], float] = time.monotonic,
-        control_task_factory: Callable[[GalaxyQueueWorker], ControlTask] = ControlTask,
+        control_task_factory: Callable[[GalaxyQueueWorker], ControlTaskLike] = ControlTask,
         queues_provider: Optional[Callable[[], list[Queue]]] = None,
     ) -> None:
         self._queue_worker = queue_worker
diff --git a/lib/galaxy_test/base/sse.py b/lib/galaxy_test/base/sse.py
index 73b3c39e2b6..d98ab7c6c52 100644
--- a/lib/galaxy_test/base/sse.py
+++ b/lib/galaxy_test/base/sse.py
@@ -9,10 +9,8 @@ thread instead of silently swallowing them.
 
 import queue
 import threading
-from typing import (
-    Callable,
-    Optional,
-)
+from collections.abc import Callable
+from typing import Optional
 
 import requests