diff --git a/CI_FAILURE_TRACKER.md b/CI_FAILURE_TRACKER.md
new file mode 100644
index 000000000000..db60be62f983
--- /dev/null
+++ b/CI_FAILURE_TRACKER.md
@@ -0,0 +1,43 @@
+# CI Failure Tracker
+
+This file tracks persistent test failures in the `tunnable-mworkers` branch to avoid "wack-a-mole" regressions.
+
+## Latest Run: 23974535354 (fa2d9f03d0)
+
+### **Functional Tests (ZeroMQ 4)**
+These tests are failing across almost all platforms (Linux, macOS, Windows).
+- **Core Error**: `Socket was found in invalid state` (`EFSM`) and `Unknown error 321`.
+- **Status**: **FIXED** (to be verified in CI).
+- **Fixes Applied**:
+    1. **Concurrency**: Added `asyncio.Lock` to `connect()` to prevent redundant `_send_recv` tasks.
+    2. **InvalidStateError**: Added `if not future.done()` checks before EVERY `set_result`/`set_exception` call in `_send_recv`.
+    3. **Cleanup**: Added `close()` method to `PoolRoutingChannelV2Revised`.
+    4. **Robust Reconnect**: Ensured ANY ZeroMQ error in the loop triggers a close and reconnect to reset the `REQ` state machine.
+    5. **Reconnect Storm Prevention**: Skip futures that are already done when pulling from the queue.
+
+**Failing Test Cases (Representative):**
+- `tests.pytests.functional.transport.server.test_ssl_transport.test_ssl_publish_server[SSLTransport(tcp)]` (Timeout)
+- `tests.pytests.functional.transport.server.test_ssl_transport.test_ssl_publish_server[SSLTransport(ws)]` (Timeout)
+- `tests.pytests.functional.transport.server.test_ssl_transport.test_ssl_file_transfer[SSLTransport(tcp)]` (Timeout)
+- `tests.pytests.functional.transport.server.test_ssl_transport.test_ssl_multi_minion[SSLTransport(tcp)]` (Timeout)
+- `tests.pytests.functional.transport.server.test_ssl_transport.test_request_server[Transport(ws)]` (Timeout)
+
+### **Scenario Tests (ZeroMQ)**
+- **Platform**: Fedora 40, Windows 2022
+- **Error**: `asyncio.exceptions.InvalidStateError: invalid state`
+- **Location**: `salt/transport/zeromq.py:1703` during `socket.poll`.
+
+### **Integration Tests**
+- `tests.pytests.functional.channel.test_pool_routing.test_pool_routing_fast_commands` (KeyError: 'transport' - *Wait, I fixed this, check if it's still failing*)
+- `Test Salt / Photon OS 5 integration tcp 4` (Conclusion: failure)
+
+### **Package Tests**
+- `Test Package / Windows 2025 NSIS downgrade 3007.13` (Timeout after 600s)
+
+---
+
+## Resolved Issues (To be verified)
+- [x] **Pre-Commit**: Passing locally and in latest run.
+- [x] **Unit Tests**: `tests/pytests/unit/transport/test_zeromq_worker_pools.py` now passing.
+- [x] **KeyError: 'aes'**: Resolved in latest runs.
+- [x] **TypeError in pre_fork**: Resolved.
diff --git a/salt/channel/pool_routing_v2_revised.py b/salt/channel/pool_routing_v2_revised.py
new file mode 100644
index 000000000000..7fb1d6468c00
--- /dev/null
+++ b/salt/channel/pool_routing_v2_revised.py
@@ -0,0 +1,218 @@
+"""
+Worker pool routing at the channel layer - V2 Revised with RequestServer IPC.
+
+This module provides transport-agnostic worker pool routing using Salt's
+existing RequestClient/RequestServer infrastructure over IPC sockets.
+
+V2 Revised Design:
+- Each worker pool has its own RequestServer listening on IPC
+- Routing channel uses RequestClient to forward messages to pool RequestServers
+- No transport modifications needed
+- Uses transport-native IPC (ZeroMQ/TCP/WS over IPC sockets)
+"""
+
+import logging
+import os
+import zlib
+
+log = logging.getLogger(__name__)
+
+
+class PoolRoutingChannelV2Revised:
+    """
+    Channel wrapper that routes requests to worker pools using RequestServer IPC.
+
+    Architecture:
+        External Transport → PoolRoutingChannel → RequestClient →
+        Pool RequestServer (IPC) → Workers
+    """
+
+    def __init__(self, opts, transport, worker_pools):
+        """
+        Initialize the pool routing channel.
+
+        Args:
+            opts: Master configuration options
+            transport: The external transport instance (port 4506)
+            worker_pools: Dict of pool configurations {pool_name: config}
+        """
+        self.opts = opts
+        self.transport = transport
+        self.worker_pools = worker_pools
+        self.pool_clients = {}  # RequestClient for each pool
+        self.pool_servers = {}  # RequestServer for each pool
+
+        log.info(
+            "PoolRoutingChannelV2Revised initialized with pools: %s",
+            list(worker_pools.keys()),
+        )
+
+    def pre_fork(self, process_manager):
+        """
+        Pre-fork setup - create RequestServer for each pool on IPC.
+
+        Args:
+            process_manager: The process manager instance
+        """
+        import salt.transport
+
+        # Delegate external transport setup
+        if hasattr(self.transport, "pre_fork"):
+            self.transport.pre_fork(process_manager)
+
+        # Create a RequestServer for each pool on IPC
+        for pool_name, config in self.worker_pools.items():
+            # Create pool-specific opts for IPC
+            pool_opts = self.opts.copy()
+
+            # Configure IPC mode and socket path
+            if pool_opts.get("ipc_mode") == "tcp":
+                # TCP IPC mode: use unique port per pool
+                base_port = pool_opts.get("tcp_master_workers", 4515)
+                port_offset = zlib.adler32(pool_name.encode()) % 1000
+                pool_opts["ret_port"] = base_port + port_offset
+                log.info(
+                    "Pool '%s' RequestServer using TCP IPC on port %d",
+                    pool_name,
+                    pool_opts["ret_port"],
+                )
+            else:
+                # Standard IPC mode: use unique socket per pool
+                sock_dir = pool_opts.get("sock_dir", "/tmp/salt")
+                os.makedirs(sock_dir, exist_ok=True)
+
+                # Each pool gets its own IPC socket
+                pool_opts["workers_ipc_name"] = f"workers-{pool_name}.ipc"
+
+            # Create RequestServer for this pool using transport factory
+            pool_server = salt.transport.request_server(pool_opts)
+
+            # Pre-fork the pool server (this creates IPC listener)
+            pool_server.pre_fork(process_manager)
+
+            self.pool_servers[pool_name] = pool_server
+
+        log.info("PoolRoutingChannelV2Revised pre_fork complete")
+
+    def post_fork(self, payload_handler, io_loop):
+        """
+        Post-fork setup in routing process.
+
+        Creates RequestClient connections to each pool's RequestServer.
+
+        Args:
+            payload_handler: Handler for processed payloads (not used)
+            io_loop: The event loop to use
+        """
+        import salt.transport
+
+        self.io_loop = io_loop
+
+        # Build routing table from worker_pools config
+        self.command_to_pool = {}
+        self.default_pool = None
+
+        for pool_name, config in self.worker_pools.items():
+            for cmd in config.get("commands", []):
+                if cmd == "*":
+                    self.default_pool = pool_name
+                else:
+                    self.command_to_pool[cmd] = pool_name
+
+        # Create RequestClient for each pool
+        for pool_name in self.worker_pools.keys():
+            # Create pool-specific opts matching the pool's RequestServer
+            pool_opts = self.opts.copy()
+
+            if pool_opts.get("ipc_mode") == "tcp":
+                # TCP IPC: connect to pool's port
+                base_port = pool_opts.get("tcp_master_workers", 4515)
+                port_offset = zlib.adler32(pool_name.encode()) % 1000
+                pool_opts["ret_port"] = base_port + port_offset
+            else:
+                # IPC socket: connect to pool's socket
+                pool_opts["workers_ipc_name"] = f"workers-{pool_name}.ipc"
+
+                sock_dir = pool_opts.get("sock_dir", "/tmp/salt")
+
+            # Create RequestClient that connects to pool's IPC RequestServer
+            client = salt.transport.request_client(pool_opts, io_loop=io_loop)
+            self.pool_clients[pool_name] = client
+
+        # Connect external transport to our routing handler
+        if hasattr(self.transport, "post_fork"):
+            self.transport.post_fork(self.handle_and_route_message, io_loop)
+
+        log.info("PoolRoutingChannelV2Revised post_fork complete")
+
+    def close(self):
+        """
+        Close the channel and all its pool clients/servers.
+        """
+        log.info("Closing PoolRoutingChannelV2Revised")
+
+        # Close all pool clients
+        for pool_name, client in self.pool_clients.items():
+            try:
+                client.close()
+            except Exception as exc:  # pylint: disable=broad-except
+                log.error("Error closing client for pool '%s': %s", pool_name, exc)
+        self.pool_clients.clear()
+
+        # Close all pool servers
+        for pool_name, server in self.pool_servers.items():
+            try:
+                server.close()
+            except Exception as exc:  # pylint: disable=broad-except
+                log.error("Error closing server for pool '%s': %s", pool_name, exc)
+        self.pool_servers.clear()
+
+        # Close external transport
+        if hasattr(self.transport, "close"):
+            self.transport.close()
+
+    async def handle_and_route_message(self, payload):
+        """
+        Handle incoming message and route to appropriate worker pool via RequestClient.
+
+        Args:
+            payload: The message payload from external transport
+
+        Returns:
+            Reply from the worker that processed the request
+        """
+        try:
+            # Determine which pool
+            load = payload.get("load", {})
+            if isinstance(load, dict):
+                cmd = load.get("cmd", "unknown")
+            else:
+                # Encrypted payload (bytes), can't extract command
+                cmd = "unknown"
+
+            pool_name = self.command_to_pool.get(cmd, self.default_pool)
+
+            if not pool_name:
+                pool_name = self.default_pool or list(self.worker_pools.keys())[0]
+
+            log.debug(
+                "Routing request (cmd=%s) to pool '%s'",
+                cmd,
+                pool_name,
+            )
+
+            # Forward to pool via RequestClient
+            client = self.pool_clients[pool_name]
+
+            # RequestClient.send() sends payload to pool's RequestServer via IPC
+            reply = await client.send(payload)
+
+            return reply
+
+        except Exception as exc:  # pylint: disable=broad-except
+            log.error(
+                "Error routing request to worker pool: %s",
+                exc,
+                exc_info=True,
+            )
+            return {"error": "Internal routing error"}
diff --git a/salt/channel/server.py b/salt/channel/server.py
index 4c43f7a29641..cdc9d7f894e4 100644
--- a/salt/channel/server.py
+++ b/salt/channel/server.py
@@ -115,13 +115,13 @@ def session_key(self, minion):
         )
         return self.sessions[minion][1]
 
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Do anything necessary pre-fork. Since this is on the master side this will
         primarily be bind and listen (or the equivalent for your network library)
         """
         if hasattr(self.transport, "pre_fork"):
-            self.transport.pre_fork(process_manager)
+            self.transport.pre_fork(process_manager, *args, **kwargs)
 
     def post_fork(self, payload_handler, io_loop):
         """
@@ -972,6 +972,82 @@ def close(self):
             self.event.destroy()
 
 
+class PoolDispatcherChannel:
+    """
+    Dispatcher channel that receives requests from the front-end channel
+    and routes them to pool-specific channels based on command classification.
+    """
+
+    def __init__(self, opts, frontend_channels, pool_channels):
+        """
+        :param opts: Master configuration options
+        :param frontend_channels: List of frontend ReqServerChannel instances (where minions connect)
+        :param pool_channels: Dict mapping pool_name to ReqServerChannel instances
+        """
+        self.opts = opts
+        self.frontend_channels = frontend_channels
+        self.pool_channels = pool_channels
+        self.router = None  # Will be initialized in post_fork
+        self.io_loop = None
+
+    def post_fork(self, io_loop):
+        """
+        Called in the dispatcher process after forking.
+        Sets up the dispatcher to receive from front-end and route to pools.
+
+        :param IOLoop io_loop: Tornado IOLoop instance
+        """
+        import salt.master
+
+        self.io_loop = io_loop
+        self.router = salt.master.RequestRouter(self.opts)
+
+        # Connect to frontend channels as a worker
+        for channel in self.frontend_channels:
+            channel.post_fork(self._dispatch_handler, io_loop)
+
+        log.info(
+            "Pool dispatcher started, routing to pools: %s",
+            list(self.pool_channels.keys()),
+        )
+
+    async def _dispatch_handler(self, payload):
+        """
+        Handle incoming message from front-end, classify it, and forward to the appropriate pool.
+
+        :param payload: The request payload from a minion
+        :return: The response from the pool worker
+        """
+        try:
+            # Classify the request to determine target pool
+            pool_name = self.router.route_request(payload)
+
+            if not pool_name:
+                log.error("Router returned no pool name for request")
+                return {"error": "Unable to route request"}
+
+            if pool_name not in self.pool_channels:
+                log.error(
+                    "Router returned unknown pool '%s'. Available pools: %s",
+                    pool_name,
+                    list(self.pool_channels.keys()),
+                )
+                return {"error": f"Unknown pool: {pool_name}"}
+
+            # Forward to the pool's transport
+            channel = self.pool_channels[pool_name]
+            log.debug("Routing request to pool '%s'", pool_name)
+
+            # Forward the message through the pool's transport
+            reply = await channel.transport.forward_message(payload)
+
+            return reply
+
+        except Exception as exc:  # pylint: disable=broad-except
+            log.error("Error in dispatcher handler: %s", exc, exc_info=True)
+            return {"error": "Dispatcher error"}
+
+
 class PubServerChannel:
     """
     Factory class to create subscription channels to the master's Publisher
@@ -1035,7 +1111,7 @@ def close(self):
             self.aes_funcs.destroy()
             self.aes_funcs = None
 
-    def pre_fork(self, process_manager, kwargs=None):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Do anything necessary pre-fork. Since this is on the master side this will
         primarily be used to create IPC channels and create our daemon process to
@@ -1044,7 +1120,11 @@ def pre_fork(self, process_manager, kwargs=None):
         :param func process_manager: A ProcessManager, from salt.utils.process.ProcessManager
         """
         if hasattr(self.transport, "publish_daemon"):
-            process_manager.add_process(self._publish_daemon, kwargs=kwargs)
+            # Extract kwargs for the process.
+            # We check for a named 'kwargs' key first (from salt/master.py),
+            # then fallback to the entire kwargs dict.
+            proc_kwargs = kwargs.pop("kwargs", kwargs)
+            process_manager.add_process(self._publish_daemon, kwargs=proc_kwargs)
 
     def _publish_daemon(self, **kwargs):
         if self.opts["pub_server_niceness"] and not salt.utils.platform.is_windows():
@@ -1240,7 +1320,7 @@ def __setstate__(self, state):
     def close(self):
         self.transport.close()
 
-    def pre_fork(self, process_manager, kwargs=None):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Do anything necessary pre-fork. Since this is on the master side this will
         primarily be used to create IPC channels and create our daemon process to
@@ -1249,11 +1329,14 @@ def pre_fork(self, process_manager, kwargs=None):
         :param func process_manager: A ProcessManager, from salt.utils.process.ProcessManager
         """
         if hasattr(self.transport, "publish_daemon"):
+            proc_kwargs = kwargs.pop("kwargs", kwargs)
             process_manager.add_process(
-                self._publish_daemon, kwargs=kwargs, name="EventPublisher"
+                self._publish_daemon, kwargs=proc_kwargs, name="EventPublisher"
             )
 
     def _publish_daemon(self, **kwargs):
+        import salt.master
+
         if (
             self.opts["event_publisher_niceness"]
             and not salt.utils.platform.is_windows()
@@ -1263,6 +1346,11 @@ def _publish_daemon(self, **kwargs):
                 self.opts["event_publisher_niceness"],
             )
             os.nice(self.opts["event_publisher_niceness"])
+
+        secrets = kwargs.get("secrets", None)
+        if secrets is not None:
+            salt.master.SMaster.secrets = secrets
+
         self.io_loop = tornado.ioloop.IOLoop.current()
         tcp_master_pool_port = self.opts["cluster_pool_port"]
         self.pushers = []
diff --git a/salt/config/__init__.py b/salt/config/__init__.py
index e4fa8e21bad7..9086232fceab 100644
--- a/salt/config/__init__.py
+++ b/salt/config/__init__.py
@@ -520,6 +520,14 @@ def _gather_buffer_space():
         # The number of MWorker processes for a master to startup. This number needs to scale up as
         # the number of connected minions increases.
         "worker_threads": int,
+        # Enable worker pool routing for mworkers
+        "worker_pools_enabled": bool,
+        # Worker pool configuration (dict of pool_name -> {worker_count, commands})
+        "worker_pools": dict,
+        # Use optimized worker pools configuration
+        "worker_pools_optimized": bool,
+        # Default pool for unmapped commands (when no catchall exists)
+        "worker_pool_default": (type(None), str),
         # The port for the master to listen to returns on. The minion needs to connect to this port
         # to send returns.
         "ret_port": int,
@@ -1378,6 +1386,10 @@ def _gather_buffer_space():
         "auth_mode": 1,
         "user": _MASTER_USER,
         "worker_threads": 5,
+        "worker_pools_enabled": True,
+        "worker_pools": {},
+        "worker_pools_optimized": False,
+        "worker_pool_default": None,
         "sock_dir": os.path.join(salt.syspaths.SOCK_DIR, "master"),
         "sock_pool_size": 1,
         "ret_port": 4506,
@@ -4281,6 +4293,25 @@ def apply_master_config(overrides=None, defaults=None):
         )
         opts["worker_threads"] = 3
 
+    # Handle worker pools configuration
+    if opts.get("worker_pools_enabled", True):
+        from salt.config.worker_pools import (
+            get_worker_pools_config,
+            validate_worker_pools_config,
+        )
+
+        # Get effective worker pools config (handles backward compat)
+        effective_pools = get_worker_pools_config(opts)
+        if effective_pools is not None:
+            opts["worker_pools"] = effective_pools
+
+            # Validate the configuration
+            try:
+                validate_worker_pools_config(opts)
+            except ValueError as exc:
+                log.error("Worker pools configuration error: %s", exc)
+                raise
+
     opts.setdefault("pillar_source_merging_strategy", "smart")
 
     # Make sure hash_type is lowercase
diff --git a/salt/config/worker_pools.py b/salt/config/worker_pools.py
new file mode 100644
index 000000000000..aebee340c029
--- /dev/null
+++ b/salt/config/worker_pools.py
@@ -0,0 +1,250 @@
+"""
+Default worker pool configuration for Salt master.
+
+This module defines the default worker pool routing configuration.
+Users can override this in their master config file.
+"""
+
+# Default worker pool routing configuration
+# This provides maximum backward compatibility by using a single pool
+# with a catchall pattern that handles all commands (identical to current behavior)
+DEFAULT_WORKER_POOLS = {
+    "default": {
+        "worker_count": 5,  # Same as current worker_threads default
+        "commands": ["*"],  # Catchall - handles all commands
+    },
+}
+
+# Optional: Performance-optimized pools for users who want better out-of-box performance
+# Users can enable this via worker_pools_optimized: True
+OPTIMIZED_WORKER_POOLS = {
+    "lightweight": {
+        "worker_count": 2,
+        "commands": [
+            "ping",
+            "get_token",
+            "mk_token",
+            "verify_minion",
+            "_master_opts",
+            "_master_tops",
+            "_file_hash",
+            "_file_hash_and_stat",
+        ],
+    },
+    "medium": {
+        "worker_count": 2,
+        "commands": [
+            "_mine_get",
+            "_mine",
+            "_mine_delete",
+            "_mine_flush",
+            "_file_find",
+            "_file_list",
+            "_file_list_emptydirs",
+            "_dir_list",
+            "_symlink_list",
+            "pub_ret",
+            "minion_pub",
+            "minion_publish",
+            "wheel",
+            "runner",
+        ],
+    },
+    "heavy": {
+        "worker_count": 1,
+        "commands": [
+            "publish",
+            "_pillar",
+            "_return",
+            "_syndic_return",
+            "_file_recv",
+            "_serve_file",
+            "minion_runner",
+            "revoke_auth",
+        ],
+    },
+}
+
+
+def validate_worker_pools_config(opts):
+    """
+    Validate worker pools configuration at master startup.
+
+    Args:
+        opts: Master configuration dictionary
+
+    Returns:
+        True if valid
+
+    Raises:
+        ValueError: If configuration is invalid with detailed error messages
+    """
+    if not opts.get("worker_pools_enabled", True):
+        # Legacy mode, no validation needed
+        return True
+
+    # Get the effective worker pools (handles defaults and backward compat)
+    worker_pools = get_worker_pools_config(opts)
+
+    # If pools are disabled, no validation needed
+    if worker_pools is None:
+        return True
+
+    default_pool = opts.get("worker_pool_default")
+
+    errors = []
+
+    # 1. Validate pool structure
+    if not isinstance(worker_pools, dict):
+        errors.append("worker_pools must be a dictionary")
+        raise ValueError("\n".join(errors))
+
+    if not worker_pools:
+        errors.append("worker_pools cannot be empty")
+        raise ValueError("\n".join(errors))
+
+    # 2. Validate each pool
+    cmd_to_pool = {}
+    catchall_pool = None
+
+    for pool_name, pool_config in worker_pools.items():
+        # Validate pool name format (security-focused: block path traversal only)
+        if not isinstance(pool_name, str):
+            errors.append(f"Pool name must be a string, got {type(pool_name).__name__}")
+            continue
+
+        if not pool_name:
+            errors.append("Pool name cannot be empty")
+            continue
+
+        # Security: block path traversal attempts
+        if "/" in pool_name or "\\" in pool_name:
+            errors.append(
+                f"Pool name '{pool_name}' is invalid. Pool names cannot contain "
+                "path separators (/ or \\) to prevent path traversal attacks."
+            )
+            continue
+
+        # Security: block relative path components
+        if (
+            pool_name == ".."
+            or pool_name.startswith("../")
+            or pool_name.startswith("..\\")
+        ):
+            errors.append(
+                f"Pool name '{pool_name}' is invalid. Pool names cannot be or start with "
+                "'../' to prevent path traversal attacks."
+            )
+            continue
+
+        # Security: block null bytes
+        if "\x00" in pool_name:
+            errors.append("Pool name contains null byte, which is not allowed.")
+            continue
+
+        if not isinstance(pool_config, dict):
+            errors.append(f"Pool '{pool_name}': configuration must be a dictionary")
+            continue
+
+        # Check worker_count
+        worker_count = pool_config.get("worker_count")
+        if not isinstance(worker_count, int) or worker_count < 1:
+            errors.append(
+                f"Pool '{pool_name}': worker_count must be integer >= 1, "
+                f"got {worker_count}"
+            )
+
+        # Check commands list
+        commands = pool_config.get("commands", [])
+        if not isinstance(commands, list):
+            errors.append(f"Pool '{pool_name}': commands must be a list")
+            continue
+
+        if not commands:
+            errors.append(f"Pool '{pool_name}': commands list cannot be empty")
+            continue
+
+        # Check for duplicate command mappings and catchall
+        for cmd in commands:
+            if not isinstance(cmd, str):
+                errors.append(f"Pool '{pool_name}': command '{cmd}' must be a string")
+                continue
+
+            if cmd == "*":
+                # Found catchall pool
+                if catchall_pool is not None:
+                    errors.append(
+                        f"Multiple pools have catchall ('*'): "
+                        f"'{catchall_pool}' and '{pool_name}'. "
+                        "Only one pool can use catchall."
+                    )
+                catchall_pool = pool_name
+                continue
+
+            if cmd in cmd_to_pool:
+                errors.append(
+                    f"Command '{cmd}' mapped to multiple pools: "
+                    f"'{cmd_to_pool[cmd]}' and '{pool_name}'"
+                )
+            else:
+                cmd_to_pool[cmd] = pool_name
+
+    # 3. Validate default pool exists (if no catchall)
+    if catchall_pool is None:
+        if default_pool is None:
+            errors.append(
+                "No catchall pool ('*') found and worker_pool_default not specified. "
+                "Either use a catchall pool or specify worker_pool_default."
+            )
+        elif default_pool not in worker_pools:
+            errors.append(
+                f"No catchall pool ('*') found and default pool '{default_pool}' "
+                f"not found in worker_pools. Available: {list(worker_pools.keys())}"
+            )
+
+    if errors:
+        raise ValueError(
+            "Worker pools configuration validation failed:\n  - "
+            + "\n  - ".join(errors)
+        )
+
+    return True
+
+
+def get_worker_pools_config(opts):
+    """
+    Get the effective worker pools configuration.
+
+    Handles backward compatibility with worker_threads and applies
+    worker_pools_optimized if requested.
+
+    Args:
+        opts: Master configuration dictionary
+
+    Returns:
+        Dictionary of worker pools configuration
+    """
+    # If pools explicitly disabled, return None (legacy mode)
+    if not opts.get("worker_pools_enabled", True):
+        return None
+
+    # Check if user wants optimized pools
+    if opts.get("worker_pools_optimized", False):
+        return opts.get("worker_pools", OPTIMIZED_WORKER_POOLS)
+
+    # Check if worker_pools is explicitly configured AND not empty
+    if "worker_pools" in opts and opts["worker_pools"]:
+        return opts["worker_pools"]
+
+    # Backward compatibility: convert worker_threads to single catchall pool
+    if "worker_threads" in opts:
+        worker_count = opts["worker_threads"]
+        return {
+            "default": {
+                "worker_count": worker_count,
+                "commands": ["*"],
+            }
+        }
+
+    # Use default configuration
+    return DEFAULT_WORKER_POOLS
diff --git a/salt/master.py b/salt/master.py
index c49b34e0f3a9..927109398cab 100644
--- a/salt/master.py
+++ b/salt/master.py
@@ -818,7 +818,9 @@ def start(self):
             ipc_publisher = salt.channel.server.MasterPubServerChannel.factory(
                 self.opts
             )
-            ipc_publisher.pre_fork(self.process_manager)
+            ipc_publisher.pre_fork(
+                self.process_manager, kwargs={"secrets": SMaster.secrets}
+            )
             if not ipc_publisher.transport.started.wait(30):
                 raise salt.exceptions.SaltMasterError(
                     "IPC publish server did not start within 30 seconds. Something went wrong."
@@ -1011,6 +1013,129 @@ def run(self):
                 io_loop.close()
 
 
+class RequestRouter:
+    """
+    Routes requests to appropriate worker pools based on command type.
+
+    This class handles the classification of incoming requests and routes
+    them to the appropriate worker pool based on user-defined configuration.
+    """
+
+    def __init__(self, opts):
+        """
+        Initialize the request router.
+
+        Args:
+            opts: Master configuration dictionary
+        """
+        self.opts = opts
+        self.cmd_to_pool = {}  # cmd -> pool_name mapping (built from config)
+        self.default_pool = opts.get("worker_pool_default")
+        self.pools = {}  # pool_name -> dealer_socket mapping (populated later)
+        self.stats = {}  # routing statistics per pool
+
+        self._build_routing_table()
+
+    def _build_routing_table(self):
+        """Build command-to-pool routing table from user configuration."""
+        from salt.config.worker_pools import DEFAULT_WORKER_POOLS
+
+        worker_pools = self.opts.get("worker_pools", DEFAULT_WORKER_POOLS)
+        catchall_pool = None
+
+        # Build reverse mapping: cmd -> pool_name
+        for pool_name, pool_config in worker_pools.items():
+            commands = pool_config.get("commands", [])
+            for cmd in commands:
+                if cmd == "*":
+                    # Found catchall pool
+                    if catchall_pool is not None:
+                        raise ValueError(
+                            f"Multiple pools have catchall ('*'): "
+                            f"'{catchall_pool}' and '{pool_name}'. "
+                            "Only one pool can use catchall."
+                        )
+                    catchall_pool = pool_name
+                    continue
+
+                if cmd in self.cmd_to_pool:
+                    # Validation: detect duplicate command mappings
+                    raise ValueError(
+                        f"Command '{cmd}' mapped to multiple pools: "
+                        f"'{self.cmd_to_pool[cmd]}' and '{pool_name}'"
+                    )
+                self.cmd_to_pool[cmd] = pool_name
+
+        # Set up default routing
+        if catchall_pool:
+            # If catchall exists, use it for unmapped commands
+            self.default_pool = catchall_pool
+        elif self.default_pool:
+            # Validate explicitly configured default pool exists
+            if self.default_pool not in worker_pools:
+                raise ValueError(
+                    f"Default pool '{self.default_pool}' not found in worker_pools. "
+                    f"Available pools: {list(worker_pools.keys())}"
+                )
+        else:
+            # No catchall and no default pool specified
+            raise ValueError(
+                "Configuration must have either: (1) a pool with catchall ('*') "
+                "in its commands, or (2) worker_pool_default specified and existing"
+            )
+
+        # Initialize stats for each pool
+        for pool_name in worker_pools.keys():
+            self.stats[pool_name] = 0
+
+    def route_request(self, payload):
+        """
+        Determine which pool should handle this request.
+
+        Args:
+            payload: Request payload dictionary
+
+        Returns:
+            str: Name of the pool that should handle this request
+        """
+        cmd = self._extract_command(payload)
+        pool = self._classify_request(cmd)
+        self.stats[pool] = self.stats.get(pool, 0) + 1
+        return pool
+
+    def _classify_request(self, cmd):
+        """
+        Classify request based on user-defined pool routing.
+
+        Args:
+            cmd: Command name string
+
+        Returns:
+            str: Pool name for this command
+        """
+        # O(1) lookup in pre-built routing table
+        return self.cmd_to_pool.get(cmd, self.default_pool)
+
+    def _extract_command(self, payload):
+        """
+        Extract command from request payload.
+
+        Args:
+            payload: Request payload dictionary
+
+        Returns:
+            str: Command name or empty string if not found
+        """
+        try:
+            load = payload.get("load", {})
+            if isinstance(load, dict):
+                return load.get("cmd", "")
+            # If load is encrypted (bytes), we can't extract the command
+            return ""
+        except (AttributeError, KeyError):
+            return ""
+
+
 class ReqServer(salt.utils.process.SignalHandlingProcess):
     """
     Starts up the master request server, minions send results to this
@@ -1061,30 +1186,100 @@ def __bind(self):
             name="ReqServer_ProcessManager", wait_for_kill=1
         )
 
-        req_channels = []
-        for transport, opts in iter_transport_opts(self.opts):
-            chan = salt.channel.server.ReqServerChannel.factory(opts)
-            chan.pre_fork(self.process_manager)
-            req_channels.append(chan)
+        if self.opts.get("worker_pools_enabled", True):
+            # Multi-pool mode with pooled routing
+            from salt.config.worker_pools import get_worker_pools_config
 
-        if self.opts["req_server_niceness"] and not salt.utils.platform.is_windows():
-            log.info(
-                "setting ReqServer_ProcessManager niceness to %d",
-                self.opts["req_server_niceness"],
-            )
-            os.nice(self.opts["req_server_niceness"])
+            worker_pools = get_worker_pools_config(self.opts)
 
-        # Reset signals to default ones before adding processes to the process
-        # manager. We don't want the processes being started to inherit those
-        # signal handlers
-        with salt.utils.process.default_signals(signal.SIGINT, signal.SIGTERM):
-            for ind in range(int(self.opts["worker_threads"])):
-                name = f"MWorker-{ind}"
-                self.process_manager.add_process(
-                    MWorker,
-                    args=(self.opts, self.master_key, self.key, req_channels),
-                    name=name,
+            # Create single request server transport with pooled routing
+            # Only ZeroMQ transport supports worker pools
+            req_channels = []
+            for transport, opts in iter_transport_opts(self.opts):
+                chan = salt.channel.server.ReqServerChannel.factory(opts)
+                # Pass worker_pools to pre_fork. Transports that don't support it
+                # (like TCP/WS) will just ignore it.
+                chan.pre_fork(self.process_manager, worker_pools=worker_pools)
+                req_channels.append(chan)
+
+            if (
+                self.opts["req_server_niceness"]
+                and not salt.utils.platform.is_windows()
+            ):
+                log.info(
+                    "setting ReqServer_ProcessManager niceness to %d",
+                    self.opts["req_server_niceness"],
+                )
+                os.nice(self.opts["req_server_niceness"])
+
+            # Reset signals to default ones before adding processes to the process
+            # manager. We don't want the processes being started to inherit those
+            # signal handlers
+            with salt.utils.process.default_signals(signal.SIGINT, signal.SIGTERM):
+                # Create workers for each pool
+                # Workers connect to pool-specific IPC sockets (workers-{pool_name}.ipc)
+                for pool_name, pool_config in worker_pools.items():
+                    worker_count = pool_config.get("worker_count", 1)
+
+                    # Create pool-specific options
+                    pool_opts = self.opts.copy()
+                    pool_opts["pool_name"] = pool_name
+
+                    # Create pool-specific channels for workers to connect to
+                    # These channels are shared by all workers in the same pool
+                    pool_worker_channels = []
+                    for transport, opts in iter_transport_opts(pool_opts):
+                        worker_chan = salt.channel.server.ReqServerChannel.factory(opts)
+                        # Ensure pool-specific transport is initialized (e.g. bind TCP socket)
+                        worker_chan.pre_fork(self.process_manager)
+                        pool_worker_channels.append(worker_chan)
+                        # Only use first transport
+                        break
+
+                    for pool_index in range(worker_count):
+                        name = f"MWorker-{pool_name}-{pool_index}"
+                        self.process_manager.add_process(
+                            MWorker,
+                            args=(
+                                pool_opts,
+                                self.master_key,
+                                self.key,
+                                pool_worker_channels,
+                            ),
+                            kwargs={"pool_name": pool_name, "pool_index": pool_index},
+                            name=name,
+                        )
+        else:
+            # Legacy single-pool mode
+            req_channels = []
+            for transport, opts in iter_transport_opts(self.opts):
+                chan = salt.channel.server.ReqServerChannel.factory(opts)
+                chan.pre_fork(self.process_manager)
+                req_channels.append(chan)
+
+            if (
+                self.opts["req_server_niceness"]
+                and not salt.utils.platform.is_windows()
+            ):
+                log.info(
+                    "setting ReqServer_ProcessManager niceness to %d",
+                    self.opts["req_server_niceness"],
                 )
+                os.nice(self.opts["req_server_niceness"])
+
+            # Reset signals to default ones before adding processes to the process
+            # manager. We don't want the processes being started to inherit those
+            # signal handlers
+            with salt.utils.process.default_signals(signal.SIGINT, signal.SIGTERM):
+                # Legacy mode: create workers using worker_threads
+                for ind in range(int(self.opts["worker_threads"])):
+                    name = f"MWorker-{ind}"
+                    self.process_manager.add_process(
+                        MWorker,
+                        args=(self.opts, self.master_key, self.key, req_channels),
+                        name=name,
+                    )
+
         self.process_manager.run()
 
     def run(self):
@@ -1112,19 +1307,23 @@ class MWorker(salt.utils.process.SignalHandlingProcess):
     salt master.
     """
 
-    def __init__(self, opts, mkey, key, req_channels, **kwargs):
+    def __init__(
+        self, opts, mkey, key, req_channels, pool_name=None, pool_index=None, **kwargs
+    ):
         """
         Create a salt master worker process
 
         :param dict opts: The salt options
         :param dict mkey: The user running the salt master and the RSA key
         :param dict key: The user running the salt master and the AES key
+        :param str pool_name: Name of the worker pool this worker belongs to
+        :param int pool_index: Index of this worker within its pool
 
         :rtype: MWorker
         :return: Master worker
         """
         super().__init__(**kwargs)
-        self.opts = opts
+        self.opts = opts.copy()  # Copy opts to avoid modifying the shared instance
         self.req_channels = req_channels
 
         self.mkey = mkey
@@ -1133,6 +1332,14 @@ def __init__(self, opts, mkey, key, req_channels, **kwargs):
         self.stats = collections.defaultdict(lambda: {"mean": 0, "runs": 0})
         self.stat_clock = time.time()
 
+        # Pool-specific attributes
+        self.pool_name = pool_name or "default"
+        self.pool_index = pool_index if pool_index is not None else 0
+
+        # Add pool_name to opts so transport can use it for URI construction
+        if pool_name:
+            self.opts["pool_name"] = pool_name
+
     # We need __setstate__ and __getstate__ to also pickle 'SMaster.secrets'.
     # Otherwise, 'SMaster.secrets' won't be copied over to the spawned process
     # on Windows since spawning processes on Windows requires pickling.
@@ -1226,6 +1433,8 @@ def _post_stats(self, start, cmd):
                 {
                     "time": end - self.stat_clock,
                     "worker": self.name,
+                    "pool": self.pool_name,
+                    "pool_index": self.pool_index,
                     "stats": self.stats,
                 },
                 tagify(self.name, "stats"),
diff --git a/salt/transport/base.py b/salt/transport/base.py
index 202912cbee12..2074ca1da8c3 100644
--- a/salt/transport/base.py
+++ b/salt/transport/base.py
@@ -349,7 +349,7 @@ def close(self):
 
 
 class DaemonizedRequestServer(RequestServer):
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         raise NotImplementedError
 
     def post_fork(self, message_handler, io_loop):
@@ -360,6 +360,15 @@ def post_fork(self, message_handler, io_loop):
         """
         raise NotImplementedError
 
+    async def forward_message(self, payload):
+        """
+        Forward a message into this transport's worker queue.
+        Used by the pool dispatcher to route messages to pool-specific transports.
+
+        :param payload: The message payload to forward
+        """
+        raise NotImplementedError
+
 
 class PublishServer(ABC):
     """
@@ -432,7 +441,7 @@ def publish_daemon(
         raise NotImplementedError
 
     @abstractmethod
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         raise NotImplementedError
 
     @abstractmethod
diff --git a/salt/transport/tcp.py b/salt/transport/tcp.py
index 862928ce18c0..378592a15b18 100644
--- a/salt/transport/tcp.py
+++ b/salt/transport/tcp.py
@@ -563,7 +563,7 @@ def __enter__(self):
     def __exit__(self, *args):
         self.close()
 
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Pre-fork we need to create the zmq router device
         """
@@ -633,6 +633,19 @@ async def handle_message(self, stream, payload, header=None):
     def decode_payload(self, payload):
         return payload
 
+    async def forward_message(self, payload):
+        """
+        Forward a message into this transport's worker queue.
+
+        Not implemented for TCP transport. Worker pool routing is only
+        supported for ZeroMQ transport.
+        """
+        log.warning(
+            "Worker pool message forwarding is not supported for TCP transport. "
+            "Use ZeroMQ transport for worker pool routing."
+        )
+        return None
+
 
 class TCPReqServer(RequestServer):
     def __init__(self, *args, **kwargs):  # pylint: disable=W0231
@@ -1583,7 +1596,7 @@ async def publisher(
             self.pull_sock.start()
         self.started.set()
 
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Do anything necessary pre-fork. Since this is on the master side this will
         primarily be used to create IPC channels and create our daemon process to
diff --git a/salt/transport/ws.py b/salt/transport/ws.py
index 0826dea3b648..f760b425fa8a 100644
--- a/salt/transport/ws.py
+++ b/salt/transport/ws.py
@@ -443,7 +443,7 @@ async def pull_handler(self, reader, writer):
             for msg in unpacker:
                 await self._pub_payload(msg)
 
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Do anything necessary pre-fork. Since this is on the master side this will
         primarily be used to create IPC channels and create our daemon process to
@@ -475,6 +475,7 @@ async def handle_request(self, request):
                     break
         finally:
             self.clients.discard(ws)
+        return ws
 
     async def _connect(self):
         if self.pull_path:
@@ -531,7 +532,7 @@ def __init__(self, opts):  # pylint: disable=W0231
         self._run = None
         self._socket = None
 
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Pre-fork we need to create the zmq router device
         """
@@ -604,6 +605,7 @@ async def handle_message(self, request):
                 await ws.send_bytes(salt.payload.dumps(reply))
             elif msg.type == aiohttp.WSMsgType.ERROR:
                 log.error("ws connection closed with exception %s", ws.exception())
+        return ws
 
     def close(self):
         if self._run is not None:
@@ -613,6 +615,19 @@ def close(self):
             self._socket.close()
             self._socket = None
 
+    async def forward_message(self, payload):
+        """
+        Forward a message into this transport's worker queue.
+
+        Not implemented for WebSocket transport. Worker pool routing is only
+        supported for ZeroMQ transport.
+        """
+        log.warning(
+            "Worker pool message forwarding is not supported for WebSocket transport. "
+            "Use ZeroMQ transport for worker pool routing."
+        )
+        return None
+
 
 class RequestClient(salt.transport.base.RequestClient):
 
diff --git a/salt/transport/zeromq.py b/salt/transport/zeromq.py
index 65c165c897a2..4d47afce88a8 100644
--- a/salt/transport/zeromq.py
+++ b/salt/transport/zeromq.py
@@ -4,15 +4,16 @@
 
 import asyncio
 import asyncio.exceptions
-import datetime
 import errno
 import hashlib
 import logging
 import multiprocessing
 import os
 import signal
+import stat
 import sys
 import threading
+import zlib
 from random import randint
 
 import tornado
@@ -441,14 +442,25 @@ def zmq_device(self):
             )
             os.nice(self.opts["mworker_queue_niceness"])
 
+        # Determine worker URI based on pool configuration
+        pool_name = self.opts.get("pool_name", "")
         if self.opts.get("ipc_mode", "") == "tcp":
-            self.w_uri = "tcp://127.0.0.1:{}".format(
-                self.opts.get("tcp_master_workers", 4515)
-            )
+            base_port = self.opts.get("tcp_master_workers", 4515)
+            if pool_name:
+                # Use different port for each pool
+                port_offset = zlib.adler32(pool_name.encode()) % 1000
+                self.w_uri = f"tcp://127.0.0.1:{base_port + port_offset}"
+            else:
+                self.w_uri = f"tcp://127.0.0.1:{base_port}"
         else:
-            self.w_uri = "ipc://{}".format(
-                os.path.join(self.opts["sock_dir"], "workers.ipc")
-            )
+            if pool_name:
+                self.w_uri = "ipc://{}".format(
+                    os.path.join(self.opts["sock_dir"], f"workers-{pool_name}.ipc")
+                )
+            else:
+                self.w_uri = "ipc://{}".format(
+                    os.path.join(self.opts["sock_dir"], "workers.ipc")
+                )
 
         log.info("Setting up the master communication server")
         log.info("ReqServer clients %s", self.uri)
@@ -456,7 +468,13 @@ def zmq_device(self):
         log.info("ReqServer workers %s", self.w_uri)
         self.workers.bind(self.w_uri)
         if self.opts.get("ipc_mode", "") != "tcp":
-            os.chmod(os.path.join(self.opts["sock_dir"], "workers.ipc"), 0o600)
+            if pool_name:
+                ipc_path = os.path.join(
+                    self.opts["sock_dir"], f"workers-{pool_name}.ipc"
+                )
+            else:
+                ipc_path = os.path.join(self.opts["sock_dir"], "workers.ipc")
+            os.chmod(ipc_path, 0o600)
 
         while True:
             if self.clients.closed or self.workers.closed:
@@ -471,6 +489,156 @@ def zmq_device(self):
                 break
         context.term()
 
+    def zmq_device_pooled(self, worker_pools):
+        """
+        Custom ZeroMQ routing device that routes messages to different worker pools
+        based on the command in the payload.
+
+        :param dict worker_pools: Dict mapping pool_name to pool configuration
+        """
+        self.__setup_signals()
+        context = zmq.Context(
+            sum(p.get("worker_count", 1) for p in worker_pools.values())
+        )
+
+        # Create frontend ROUTER socket (minions connect here)
+        self.uri = "tcp://{interface}:{ret_port}".format(**self.opts)
+        self.clients = context.socket(zmq.ROUTER)
+        self.clients.setsockopt(zmq.LINGER, -1)
+        if self.opts["ipv6"] is True and hasattr(zmq, "IPV4ONLY"):
+            self.clients.setsockopt(zmq.IPV4ONLY, 0)
+        self.clients.setsockopt(zmq.BACKLOG, self.opts.get("zmq_backlog", 1000))
+        self._start_zmq_monitor()
+
+        if self.opts["mworker_queue_niceness"] and not salt.utils.platform.is_windows():
+            log.info(
+                "setting mworker_queue niceness to %d",
+                self.opts["mworker_queue_niceness"],
+            )
+            os.nice(self.opts["mworker_queue_niceness"])
+
+        # Create backend DEALER sockets (one per pool) that preserve envelopes
+        self.pool_workers = {}
+        for pool_name in worker_pools.keys():
+            dealer_socket = context.socket(zmq.DEALER)
+            dealer_socket.setsockopt(zmq.LINGER, -1)
+
+            # Determine worker URI for this pool
+            if self.opts.get("ipc_mode", "") == "tcp":
+                base_port = self.opts.get("tcp_master_workers", 4515)
+                port_offset = zlib.adler32(pool_name.encode()) % 1000
+                w_uri = f"tcp://127.0.0.1:{base_port + port_offset}"
+            else:
+                w_uri = "ipc://{}".format(
+                    os.path.join(self.opts["sock_dir"], f"workers-{pool_name}.ipc")
+                )
+
+            log.info("ReqServer pool '%s' workers %s", pool_name, w_uri)
+            dealer_socket.bind(w_uri)
+            if self.opts.get("ipc_mode", "") != "tcp":
+                ipc_path = os.path.join(
+                    self.opts["sock_dir"], f"workers-{pool_name}.ipc"
+                )
+                os.chmod(ipc_path, 0o600)
+
+            self.pool_workers[pool_name] = dealer_socket
+
+        # Initialize request router for command classification
+        import salt.master
+
+        router = salt.master.RequestRouter(self.opts)
+
+        # Create marker file for _is_master_running() check in netapi
+        # This file is expected by components that check if master is running
+        if self.opts.get("ipc_mode", "") != "tcp":
+            marker_path = os.path.join(self.opts["sock_dir"], "workers.ipc")
+            # If workers.ipc exists and is a socket (from a legacy run), remove it
+            if os.path.exists(marker_path):
+                try:
+                    if stat.S_ISSOCK(os.lstat(marker_path).st_mode):
+                        log.debug("Removing legacy workers.ipc socket")
+                        os.remove(marker_path)
+                except OSError:
+                    pass
+            # Touch the file to create it if it doesn't exist
+            try:
+                with salt.utils.files.fopen(marker_path, "a", encoding="utf-8"):
+                    pass
+                os.chmod(marker_path, 0o600)
+            except OSError as exc:
+                log.error("Failed to create workers.ipc marker file: %s", exc)
+
+        log.info("Setting up pooled master communication server")
+        log.info("ReqServer clients %s", self.uri)
+        self.clients.bind(self.uri)
+
+        # Poller for receiving from clients and all worker pools
+        poller = zmq.Poller()
+        poller.register(self.clients, zmq.POLLIN)
+        for pool_dealer in self.pool_workers.values():
+            poller.register(pool_dealer, zmq.POLLIN)
+
+        while True:
+            if self.clients.closed:
+                break
+
+            try:
+                socks = dict(poller.poll())
+
+                # Handle incoming responses from worker pools
+                # DEALER preserves the envelope, so we get: [client_id, b"", response]
+                for pool_name, pool_dealer in self.pool_workers.items():
+                    if pool_dealer in socks:
+                        # Receive message from DEALER (envelope is preserved)
+                        msg = pool_dealer.recv_multipart()
+                        if len(msg) >= 3:
+                            # Forward entire envelope back to ROUTER -> client
+                            self.clients.send_multipart(msg)
+
+                # Handle incoming request from client (minion)
+                if self.clients in socks:
+                    # Receive multipart message: [client_id, b"", payload]
+                    msg = self.clients.recv_multipart()
+                    if len(msg) < 3:
+                        continue
+
+                    payload_raw = msg[2]
+
+                    # Decode payload to determine which pool should handle this
+                    try:
+                        payload = salt.payload.loads(payload_raw)
+                        pool_name = router.route_request(payload)
+
+                        if pool_name not in self.pool_workers:
+                            log.error(
+                                "Unknown pool '%s' for routing. Using first available pool.",
+                                pool_name,
+                            )
+                            pool_name = next(iter(self.pool_workers.keys()))
+
+                        # Forward entire envelope to appropriate pool's DEALER
+                        # DEALER will preserve the envelope when forwarding to REQ workers
+                        pool_dealer = self.pool_workers[pool_name]
+                        pool_dealer.send_multipart(msg)
+
+                    except Exception as exc:  # pylint: disable=broad-except
+                        log.error("Error routing request: %s", exc, exc_info=True)
+                        # Send error response back to client
+                        error_payload = salt.payload.dumps({"error": "Routing error"})
+                        self.clients.send_multipart([msg[0], b"", error_payload])
+
+            except zmq.ZMQError as exc:
+                if exc.errno == errno.EINTR:
+                    continue
+                raise
+            except (KeyboardInterrupt, SystemExit):
+                break
+
+        # Cleanup
+        for pool_dealer in self.pool_workers.values():
+            pool_dealer.close()
+        context.term()
+
     def close(self):
         """
         Cleanly shutdown the router socket
@@ -490,6 +658,11 @@ def close(self):
             self.clients.close()
         if hasattr(self, "workers") and self.workers.closed is False:
             self.workers.close()
+        # Close pool workers if they exist
+        if hasattr(self, "pool_workers"):
+            for dealer in self.pool_workers.values():
+                if not dealer.closed:
+                    dealer.close()
         if hasattr(self, "stream"):
             self.stream.close()
         if hasattr(self, "_socket") and self._socket.closed is False:
@@ -502,13 +675,29 @@ def close(self):
             except RuntimeError:
                 log.error("IOLoop closed when trying to cancel task")
 
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Pre-fork we need to create the zmq router device
 
         :param func process_manager: An instance of salt.utils.process.ProcessManager
+        :param dict worker_pools: Optional worker pools configuration for pooled routing
         """
-        process_manager.add_process(self.zmq_device, name="MWorkerQueue")
+        # If we are a pool-specific RequestServer, we don't need a device.
+        # We connect directly to the sockets created by the main pooled device.
+        if self.opts.get("pool_name"):
+            return
+
+        worker_pools = kwargs.get("worker_pools") or (args[0] if args else None)
+        if worker_pools:
+            # Use pooled routing device
+            process_manager.add_process(
+                self.zmq_device_pooled,
+                args=(worker_pools,),
+                name="MWorkerQueue",
+            )
+        else:
+            # Use standard routing device
+            process_manager.add_process(self.zmq_device, name="MWorkerQueue")
 
     def _start_zmq_monitor(self):
         """
@@ -540,20 +729,22 @@ def post_fork(self, message_handler, io_loop):
         self._socket.setsockopt(zmq.LINGER, -1)
         self._start_zmq_monitor()
 
-        if self.opts.get("ipc_mode", "") == "tcp":
-            self.w_uri = "tcp://127.0.0.1:{}".format(
-                self.opts.get("tcp_master_workers", 4515)
-            )
-        else:
-            self.w_uri = "ipc://{}".format(
-                os.path.join(self.opts["sock_dir"], "workers.ipc")
-            )
+        # Use get_worker_uri() for consistent URI construction
+        self.w_uri = self.get_worker_uri()
         log.info("Worker binding to socket %s", self.w_uri)
         self._socket.connect(self.w_uri)
-        if self.opts.get("ipc_mode", "") != "tcp" and os.path.isfile(
-            os.path.join(self.opts["sock_dir"], "workers.ipc")
-        ):
-            os.chmod(os.path.join(self.opts["sock_dir"], "workers.ipc"), 0o600)
+
+        # Set permissions for IPC sockets
+        if self.opts.get("ipc_mode", "") != "tcp":
+            pool_name = self.opts.get("pool_name", "")
+            if pool_name:
+                ipc_path = os.path.join(
+                    self.opts["sock_dir"], f"workers-{pool_name}.ipc"
+                )
+            else:
+                ipc_path = os.path.join(self.opts["sock_dir"], "workers.ipc")
+            if os.path.isfile(ipc_path):
+                os.chmod(ipc_path, 0o600)
         self.message_handler = message_handler
 
         async def callback():
@@ -609,6 +800,51 @@ def decode_payload(self, payload):
         payload = salt.payload.loads(payload)
         return payload
 
+    def get_worker_uri(self):
+        """
+        Get the URI where workers connect to this transport's queue.
+        Used by the dispatcher to know where to forward messages.
+        """
+        if self.opts.get("ipc_mode", "") == "tcp":
+            pool_name = self.opts.get("pool_name", "")
+            if pool_name:
+                # Hash pool name for consistent port assignment
+                base_port = self.opts.get("tcp_master_workers", 4515)
+                port_offset = zlib.adler32(pool_name.encode()) % 1000
+                return f"tcp://127.0.0.1:{base_port + port_offset}"
+            else:
+                return f"tcp://127.0.0.1:{self.opts.get('tcp_master_workers', 4515)}"
+        else:
+            pool_name = self.opts.get("pool_name", "")
+            if pool_name:
+                return f"ipc://{os.path.join(self.opts['sock_dir'], f'workers-{pool_name}.ipc')}"
+            else:
+                return f"ipc://{os.path.join(self.opts['sock_dir'], 'workers.ipc')}"
+
+    async def forward_message(self, payload):
+        """
+        Forward a message to this transport's worker queue.
+        Creates a temporary client connection to send the message.
+        """
+        context = zmq.asyncio.Context()
+        socket = context.socket(zmq.REQ)
+        socket.setsockopt(zmq.LINGER, 0)
+
+        try:
+            w_uri = self.get_worker_uri()
+            socket.connect(w_uri)
+
+            # Send payload
+            await socket.send(self.encode_payload(payload))
+
+            # Receive reply (required for REQ/REP pattern)
+            reply = await asyncio.wait_for(socket.recv(), timeout=60.0)
+
+            return self.decode_payload(reply)
+        finally:
+            socket.close()
+            context.term()
+
 
 def _set_tcp_keepalive(zmq_socket, opts):
     """
@@ -664,34 +900,49 @@ def __init__(self, opts, addr, linger=0, io_loop=None):
             self.io_loop = tornado.ioloop.IOLoop.current()
         else:
             self.io_loop = io_loop
+        self._aioloop = salt.utils.asynchronous.aioloop(self.io_loop)
         self.context = zmq.eventloop.future.Context()
         self.socket = None
         self._closing = False
-        self._queue = tornado.queues.Queue()
-
-    def connect(self):
-        if hasattr(self, "socket") and self.socket:
-            return
-        # wire up sockets
-        self._init_socket()
+        self._queue = asyncio.Queue()
+        self._connect_lock = asyncio.Lock()
+        self.send_recv_task = None
+        self.send_recv_task_id = 0
+
+    async def connect(self):
+        async with self._connect_lock:
+            if hasattr(self, "socket") and self.socket:
+                return
+            # wire up sockets
+            self._init_socket()
 
     def close(self):
         if self._closing:
             return
-        else:
-            self._closing = True
+        self._closing = True
+        if self._queue is not None:
+            self._queue.put_nowait((None, None))
+        if hasattr(self, "socket") and self.socket is not None:
+            self.socket.close(0)
+            self.socket = None
+        if self.context is not None and self.context.closed is False:
             try:
-                if hasattr(self, "socket") and self.socket is not None:
-                    self.socket.close(0)
-                    self.socket = None
-                if self.context is not None and self.context.closed is False:
-                    self.context.term()
-                    self.context = None
-            finally:
-                self._closing = False
+                self.context.term()
+            except Exception:  # pylint: disable=broad-except
+                pass
+            self.context = None
+        if self.send_recv_task is not None:
+            self.send_recv_task = None
+
+    async def _reconnect(self):
+        if hasattr(self, "socket") and self.socket is not None:
+            self.socket.close(0)
+            self.socket = None
+        await self.connect()
 
     def _init_socket(self):
         self._closing = False
+        self.send_recv_task_id += 1
         if not self.context:
             self.context = zmq.eventloop.future.Context()
         self.socket = self.context.socket(zmq.REQ)
@@ -709,16 +960,16 @@ def _init_socket(self):
                 self.socket.setsockopt(zmq.IPV4ONLY, 0)
         self.socket.setsockopt(zmq.LINGER, self.linger)
         self.socket.connect(self.addr)
-        self.io_loop.spawn_callback(self._send_recv, self.socket)
+        self.send_recv_task = self._aioloop.create_task(
+            self._send_recv(self.socket, task_id=self.send_recv_task_id)
+        )
 
-    def send(self, message, timeout=None, callback=None):
+    async def send(self, message, timeout=None, callback=None):
         """
         Return a future which will be completed when the message has a response
         """
         future = tornado.concurrent.Future()
-
         message = salt.payload.dumps(message)
-
         self._queue.put_nowait((future, message))
 
         if callback is not None:
@@ -733,141 +984,133 @@ def handle_future(future):
             timeout = 1
 
         if timeout is not None:
-            send_timeout = self.io_loop.call_later(
-                timeout, self._timeout_message, future
-            )
-
-        recv = yield future
+            self.io_loop.call_later(timeout, self._timeout_message, future)
 
-        raise tornado.gen.Return(recv)
+        return await future
 
     def _timeout_message(self, future):
         if not future.done():
             future.set_exception(SaltReqTimeoutError("Message timed out"))
 
-    @tornado.gen.coroutine
-    def _send_recv(self, socket, _TimeoutError=tornado.gen.TimeoutError):
+    async def _send_recv(
+        self, socket, task_id=None, _TimeoutError=tornado.gen.TimeoutError
+    ):
         """
-        Long-running send/receive coroutine. This should be started once for
-        each socket created. Once started, the coroutine will run until the
-        socket is closed. A future and message are pulled from the queue. The
-        message is sent and the reply socket is polled for a response while
-        checking the future to see if it was timed out.
+        Long-running send/receive coroutine.
         """
         send_recv_running = True
-        # Hold on to the socket so we'll still have a reference to it after the
-        # close method is called. This allows us to fail gracefully once it's
-        # been closed.
         while send_recv_running:
+            if task_id is not None and task_id != self.send_recv_task_id:
+                break
+
             try:
-                future, message = yield self._queue.get(
-                    timeout=datetime.timedelta(milliseconds=300)
+                # Use a small timeout to allow periodic task_id checks
+                future, message = await asyncio.wait_for(self._queue.get(), 0.3)
+            except asyncio.TimeoutError:
+                continue
+            except (asyncio.CancelledError, asyncio.exceptions.CancelledError):
+                break
+
+            if task_id is not None and task_id != self.send_recv_task_id:
+                # Re-queue the message so the new task can pick it up
+                self._queue.put_nowait((future, message))
+                log.trace(
+                    "Task %s is no longer active after queue.get. Re-queued and exiting.",
+                    task_id,
                 )
-            except _TimeoutError:
-                try:
-                    # For some reason yielding here doesn't work becaues the
-                    # future always has a result?
-                    poll_future = socket.poll(0, zmq.POLLOUT)
-                    poll_future.result()
-                except _TimeoutError:
-                    # This is what we expect if the socket is still alive
-                    pass
-                except zmq.eventloop.future.CancelledError:
-                    log.trace("Loop closed while polling send socket.")
-                    # The ioloop was closed before polling finished.
-                    send_recv_running = False
-                    break
-                except zmq.ZMQError:
-                    log.trace("Send socket closed while polling.")
-                    send_recv_running = False
-                    break
+                break
+
+            if future is None:
+                log.trace("Received send/recv shutdown sentinal")
+                send_recv_running = False
+                break
+
+            if future.done():
                 continue
 
             try:
-                yield socket.send(message)
-            except zmq.eventloop.future.CancelledError as exc:
-                log.trace("Loop closed while sending.")
-                # The ioloop was closed before polling finished.
+                await socket.send(message)
+            except (zmq.eventloop.future.CancelledError, asyncio.CancelledError) as exc:
                 send_recv_running = False
-                future.set_exception(exc)
+                if not future.done():
+                    future.set_exception(exc)
                 break
             except zmq.ZMQError as exc:
-                if exc.errno in [
-                    zmq.ENOTSOCK,
-                    zmq.ETERM,
-                    zmq.error.EINTR,
-                ]:
-                    log.trace("Send socket closed while sending.")
-                    send_recv_running = False
-                    future.set_exception(exc)
-                elif exc.errno == zmq.EFSM:
-                    log.error("Socket was found in invalid state.")
-                    send_recv_running = False
+                if not future.done():
                     future.set_exception(exc)
-                else:
-                    log.error("Unhandled Zeromq error durring send/receive: %s", exc)
-                    future.set_exception(exc)
-
-            if future.done():
-                if isinstance(future.exception(), SaltReqTimeoutError):
-                    log.trace("Request timed out while sending. reconnecting.")
-                else:
-                    log.trace(
-                        "The request ended with an error while sending. reconnecting."
-                    )
-                self.close()
-                self.connect()
-                send_recv_running = False
+                await self._reconnect()
                 break
 
             received = False
             ready = False
             while True:
                 try:
-                    # Time is in milliseconds.
-                    ready = yield socket.poll(300, zmq.POLLIN)
-                except zmq.eventloop.future.CancelledError as exc:
-                    log.trace(
-                        "Loop closed while polling receive socket.", exc_info=True
-                    )
-                    log.error("Master is unavailable (Connection Cancelled).")
+                    ready = await socket.poll(300, zmq.POLLIN)
+                except (
+                    zmq.eventloop.future.CancelledError,
+                    asyncio.CancelledError,
+                    asyncio.exceptions.CancelledError,
+                ) as exc:
                     send_recv_running = False
                     if not future.done():
-                        future.set_result(None)
+                        future.set_exception(exc)
+                    break
                 except zmq.ZMQError as exc:
-                    log.trace("Receive socket closed while polling.")
                     send_recv_running = False
-                    future.set_exception(exc)
+                    if not future.done():
+                        future.set_exception(exc)
+                    await self._reconnect()
+                    break
 
                 if ready:
                     try:
-                        recv = yield socket.recv()
+                        recv = await socket.recv()
                         received = True
-                    except zmq.eventloop.future.CancelledError as exc:
-                        log.trace("Loop closed while receiving.")
+                    except (
+                        zmq.eventloop.future.CancelledError,
+                        asyncio.CancelledError,
+                        asyncio.exceptions.CancelledError,
+                    ) as exc:
                         send_recv_running = False
-                        future.set_exception(exc)
+                        if not future.done():
+                            future.set_exception(exc)
                     except zmq.ZMQError as exc:
-                        log.trace("Receive socket closed while receiving.")
                         send_recv_running = False
-                        future.set_exception(exc)
+                        if not future.done():
+                            future.set_exception(exc)
+                        await self._reconnect()
                     break
                 elif future.done():
                     break
 
             if future.done():
-                if isinstance(future.exception(), SaltReqTimeoutError):
-                    log.trace(
+                if future.cancelled():
+                    send_recv_running = False
+                    break
+                exc = future.exception()
+                if isinstance(exc, (asyncio.CancelledError, zmq.eventloop.future.CancelledError)):
+                    send_recv_running = False
+                    break
+                if isinstance(exc, SaltReqTimeoutError):
+                    log.error(
                         "Request timed out while waiting for a response. reconnecting."
                     )
+                elif isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN:
+                    # Resource temporarily unavailable is normal during reconnections
+                    log.trace("Socket EAGAIN during send/recv loop. reconnecting.")
                 else:
-                    log.trace("The request ended with an error. reconnecting.")
-                self.close()
-                self.connect()
+                    log.error("The request ended with an error. reconnecting. %r", exc)
+                await self._reconnect()
                 send_recv_running = False
             elif received:
-                data = salt.payload.loads(recv)
-                future.set_result(data)
+                try:
+                    data = salt.payload.loads(recv)
+                    if not future.done():
+                        future.set_result(data)
+                except Exception as exc:  # pylint: disable=broad-except
+                    log.error("Failed to deserialize response: %s", exc)
+                    if not future.done():
+                        future.set_exception(exc)
         log.trace("Send and receive coroutine ending %s", socket)
 
 
@@ -1161,7 +1404,7 @@ async def publish_payload(self, payload, topic_list=None):
             await self.dpub_sock.send(payload)
             log.trace("Unfiltered data has been sent")
 
-    def pre_fork(self, process_manager):
+    def pre_fork(self, process_manager, *args, **kwargs):
         """
         Do anything necessary pre-fork. Since this is on the master side this will
         primarily be used to create IPC channels and create our daemon process to
@@ -1253,21 +1496,32 @@ def __init__(self, opts, io_loop, linger=0):  # pylint: disable=W0231
         self._closing = False
         self.socket = None
         self._queue = asyncio.Queue()
+        self._connect_lock = asyncio.Lock()
+        self.send_recv_task = None
+        self.send_recv_task_id = 0
 
     async def connect(self):  # pylint: disable=invalid-overridden-method
-        if self.socket is None:
-            self._connect_called = True
-            self._closing = False
-            # wire up sockets
-            self._queue = asyncio.Queue()
-            self._init_socket()
+        async with self._connect_lock:
+            if self.socket is None:
+                self._connect_called = True
+                self._closing = False
+                # wire up sockets
+                self._init_socket()
 
     def _init_socket(self):
+        # Clean up old task if it exists
+        if self.send_recv_task is not None:
+            self.send_recv_task = None
+
+        self.send_recv_task_id += 1
+
         if self.socket is not None:
+            self.socket.close()
+            self.socket = None
+
+        if self.context is None:
             self.context = zmq.asyncio.Context()
-            self.socket.close()  # pylint: disable=E0203
-            del self.socket
-        self.context = zmq.asyncio.Context()
+
         self.socket = self.context.socket(zmq.REQ)
         self.socket.setsockopt(zmq.LINGER, -1)
 
@@ -1285,7 +1539,7 @@ def _init_socket(self):
         self.socket.linger = self.linger
         self.socket.connect(self.master_uri)
         self.send_recv_task = self.io_loop.create_task(
-            self._send_recv(self.socket, self._queue)
+            self._send_recv(self.socket, self._queue, task_id=self.send_recv_task_id)
         )
         self.send_recv_task._log_destroy_pending = False
 
@@ -1295,47 +1549,27 @@ def close(self):
             return
         self._closing = True
         # Save socket reference before clearing it for use in callback
-        self._queue.put_nowait((None, None))
-        task_socket = self.socket
+        if hasattr(self, "_queue") and self._queue is not None:
+            self._queue.put_nowait((None, None))
         if self.socket:
             self.socket.close()
             self.socket = None
         if self.context and self.context.closed is False:
             # This hangs if closing the stream causes an import error
-            self.context.term()
+            try:
+                self.context.term()
+            except Exception:  # pylint: disable=broad-except
+                pass
             self.context = None
-        # if getattr(self, "send_recv_task", None):
-        #    task = self.send_recv_task
-        #    if not task.done():
-        #        task.cancel()
-
-        #        # Suppress "Task was destroyed but it is pending!" warnings
-        #        # by ensuring the task knows its exception will be handled
-        #        task._log_destroy_pending = False
-
-        #        def _drain_cancelled(cancelled_task):
-        #            try:
-        #                cancelled_task.exception()
-        #            except asyncio.CancelledError:  # pragma: no cover
-        #                # Task was cancelled - log the expected messages
-        #                log.trace("Send socket closed while polling.")
-        #                log.trace("Send and receive coroutine ending %s", task_socket)
-        #            except (
-        #                Exception  # pylint: disable=broad-exception-caught
-        #            ):  # pragma: no cover
-        #                log.trace(
-        #                    "Exception while cancelling send/receive task.",
-        #                    exc_info=True,
-        #                )
-        #                log.trace("Send and receive coroutine ending %s", task_socket)
-
-        #        task.add_done_callback(_drain_cancelled)
-        #    else:
-        #        try:
-        #            task.result()
-        #        except Exception as exc:  # pylint: disable=broad-except
-        #            log.trace("Exception while retrieving send/receive task: %r", exc)
-        #    self.send_recv_task = None
+
+        if self.send_recv_task is not None:
+            self.send_recv_task = None
+
+    async def _reconnect(self):
+        if self.socket is not None:
+            self.socket.close()
+            self.socket = None
+        await self.connect()
 
     async def send(self, load, timeout=60):
         """
@@ -1378,7 +1612,9 @@ def get_master_uri(opts):
         # if we've reached here something is very abnormal
         raise SaltException("ReqChannel: missing master_uri/master_ip in self.opts")
 
-    async def _send_recv(self, socket, queue, _TimeoutError=tornado.gen.TimeoutError):
+    async def _send_recv(
+        self, socket, queue, task_id=None, _TimeoutError=tornado.gen.TimeoutError
+    ):
         """
         Long running send/receive coroutine. This should be started once for
         each socket created. Once started, the coroutine will run until the
@@ -1391,76 +1627,45 @@ async def _send_recv(self, socket, queue, _TimeoutError=tornado.gen.TimeoutError
         # close method is called. This allows us to fail gracefully once it's
         # been closed.
         while send_recv_running:
+            if task_id is not None and task_id != self.send_recv_task_id:
+                break
+
             try:
+                # Use a small timeout to allow periodic task_id checks
                 future, message = await asyncio.wait_for(queue.get(), 0.3)
-            except asyncio.TimeoutError as exc:
-                try:
-                    # For some reason yielding here doesn't work becaues the
-                    # future always has a result?
-                    poll_future = socket.poll(0, zmq.POLLOUT)
-                    poll_future.result()
-                except _TimeoutError:
-                    # This is what we expect if the socket is still alive
-                    pass
-                except (
-                    zmq.eventloop.future.CancelledError,
-                    asyncio.exceptions.CancelledError,
-                ):
-                    log.trace("Loop closed while polling send socket.")
-                    # The ioloop was closed before polling finished.
-                    send_recv_running = False
-                    break
-                except zmq.ZMQError:
-                    log.trace("Send socket closed while polling.")
-                    send_recv_running = False
-                    break
+            except asyncio.TimeoutError:
                 continue
+            except (asyncio.CancelledError, asyncio.exceptions.CancelledError):
+                break
+
+            if task_id is not None and task_id != self.send_recv_task_id:
+                # Re-queue the message so the new task can pick it up
+                self._queue.put_nowait((future, message))
+                log.trace(
+                    "Task %s is no longer active after queue.get. Re-queued and exiting.",
+                    task_id,
+                )
+                break
 
             if future is None:
                 log.trace("Received send/recv shutdown sentinal")
                 send_recv_running = False
                 break
+
+            if future.done():
+                continue
+
             try:
                 await socket.send(message)
-            except asyncio.CancelledError as exc:
-                log.trace("Loop closed while sending.")
+            except (zmq.eventloop.future.CancelledError, asyncio.CancelledError) as exc:
                 send_recv_running = False
-                future.set_exception(exc)
-            except zmq.eventloop.future.CancelledError as exc:
-                log.trace("Loop closed while sending.")
-                # The ioloop was closed before polling finished.
-                send_recv_running = False
-                future.set_exception(exc)
-            except zmq.ZMQError as exc:
-                if exc.errno in [
-                    zmq.ENOTSOCK,
-                    zmq.ETERM,
-                    zmq.error.EINTR,
-                ]:
-                    log.trace("Send socket closed while sending.")
-                    send_recv_running = False
-                    future.set_exception(exc)
-                elif exc.errno == zmq.EFSM:
-                    log.error("Socket was found in invalid state.")
-                    send_recv_running = False
+                if not future.done():
                     future.set_exception(exc)
-                else:
-                    log.error("Unhandled Zeromq error durring send/receive: %s", exc)
+                break
+            except zmq.ZMQError as exc:
+                if not future.done():
                     future.set_exception(exc)
-
-            if future.done():
-                if isinstance(future.exception(), asyncio.CancelledError):
-                    send_recv_running = False
-                    break
-                elif isinstance(future.exception(), SaltReqTimeoutError):
-                    log.trace("Request timed out while sending. reconnecting.")
-                else:
-                    log.trace(
-                        "The request ended with an error while sending. reconnecting."
-                    )
-                self.close()
-                await self.connect()
-                send_recv_running = False
+                await self._reconnect()
                 break
 
             received = False
@@ -1469,54 +1674,70 @@ async def _send_recv(self, socket, queue, _TimeoutError=tornado.gen.TimeoutError
                 try:
                     # Time is in milliseconds.
                     ready = await socket.poll(300, zmq.POLLIN)
-                except asyncio.CancelledError as exc:
-                    log.trace("Loop closed while polling receive socket.")
-                    send_recv_running = False
-                    future.set_exception(exc)
-                except zmq.eventloop.future.CancelledError as exc:
-                    log.trace("Loop closed while polling receive socket.")
+                except (
+                    asyncio.CancelledError,
+                    zmq.eventloop.future.CancelledError,
+                    asyncio.exceptions.CancelledError,
+                ) as exc:
                     send_recv_running = False
-                    future.set_exception(exc)
+                    if not future.done():
+                        future.set_exception(exc)
+                    break
                 except zmq.ZMQError as exc:
-                    log.trace("Receive socket closed while polling.")
                     send_recv_running = False
-                    future.set_exception(exc)
+                    if not future.done():
+                        future.set_exception(exc)
+                    await self._reconnect()
+                    break
 
                 if ready:
                     try:
                         recv = await socket.recv()
                         received = True
-                    except asyncio.CancelledError as exc:
-                        log.trace("Loop closed while receiving.")
+                    except (
+                        asyncio.CancelledError,
+                        zmq.eventloop.future.CancelledError,
+                        asyncio.exceptions.CancelledError,
+                    ) as exc:
                         send_recv_running = False
-                        future.set_exception(exc)
-                    except zmq.eventloop.future.CancelledError as exc:
-                        log.trace("Loop closed while receiving.")
-                        send_recv_running = False
-                        future.set_exception(exc)
+                        if not future.done():
+                            future.set_exception(exc)
                     except zmq.ZMQError as exc:
-                        log.trace("Receive socket closed while receiving.")
                         send_recv_running = False
-                        future.set_exception(exc)
+                        if not future.done():
+                            future.set_exception(exc)
+                        await self._reconnect()
+                        break
                     break
                 elif future.done():
                     break
 
             if future.done():
+                if future.cancelled():
+                    send_recv_running = False
+                    break
                 exc = future.exception()
-                if isinstance(exc, asyncio.CancelledError):
+                if isinstance(exc, (asyncio.CancelledError, zmq.eventloop.future.CancelledError)):
                     send_recv_running = False
                     break
-                elif isinstance(exc, SaltReqTimeoutError):
+                if isinstance(exc, SaltReqTimeoutError):
                     log.error(
                         "Request timed out while waiting for a response. reconnecting."
                     )
+                elif isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN:
+                    # Resource temporarily unavailable is normal during reconnections
+                    log.trace("Socket EAGAIN during send/recv loop. reconnecting.")
                 else:
                     log.error("The request ended with an error. reconnecting. %r", exc)
-                self.close()
-                await self.connect()
+                await self._reconnect()
                 send_recv_running = False
             elif received:
-                data = salt.payload.loads(recv)
-                future.set_result(data)
+                try:
+                    data = salt.payload.loads(recv)
+                    if not future.done():
+                        future.set_result(data)
+                except Exception as exc:  # pylint: disable=broad-except
+                    log.error("Failed to deserialize response: %s", exc)
+                    if not future.done():
+                        future.set_exception(exc)
         log.trace("Send and receive coroutine ending %s", socket)
diff --git a/salt/utils/channel.py b/salt/utils/channel.py
index 8ce2e259dcc5..e1e36eec5922 100644
--- a/salt/utils/channel.py
+++ b/salt/utils/channel.py
@@ -14,5 +14,6 @@ def iter_transport_opts(opts):
         transports.add(transport)
         yield transport, t_opts
 
-    if opts["transport"] not in transports:
-        yield opts["transport"], opts
+    transport = opts.get("transport", "zeromq")
+    if transport not in transports:
+        yield transport, opts
diff --git a/salt/utils/pkg/deb.py b/salt/utils/pkg/deb.py
index 830dacd9e1e3..4e95c5eb2433 100644
--- a/salt/utils/pkg/deb.py
+++ b/salt/utils/pkg/deb.py
@@ -210,8 +210,8 @@ def __eq__(self, other):
         return (
             self.disabled == other.disabled
             and self.type == other.type
-            and set([uri.rstrip("/") for uri in self.uris])
-            == set([uri.rstrip("/") for uri in other.uris])
+            and {uri.rstrip("/") for uri in self.uris}
+            == {uri.rstrip("/") for uri in other.uris}
             and self.dist == other.dist
             and self.comps == other.comps
         )
diff --git a/tests/pytests/conftest.py b/tests/pytests/conftest.py
index a5842b24650e..b81c03ae7877 100644
--- a/tests/pytests/conftest.py
+++ b/tests/pytests/conftest.py
@@ -184,6 +184,25 @@ def salt_master_factory(
         "publish_signing_algorithm": (
             "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1"
         ),
+        # Use optimized worker pools for integration/scenario tests
+        # This demonstrates the worker pool feature and provides better performance
+        "worker_pools_enabled": True,
+        "worker_pools": {
+            "fast": {
+                "worker_count": 2,
+                "commands": [
+                    "ping",
+                    "get_token",
+                    "mk_token",
+                    "verify_minion",
+                    "_master_opts",
+                ],
+            },
+            "general": {
+                "worker_count": 3,
+                "commands": ["*"],  # Catchall for everything else
+            },
+        },
     }
     ext_pillar = []
     if salt.utils.platform.is_windows():
diff --git a/tests/pytests/functional/channel/test_pool_routing.py b/tests/pytests/functional/channel/test_pool_routing.py
new file mode 100644
index 000000000000..600fed0d4bd3
--- /dev/null
+++ b/tests/pytests/functional/channel/test_pool_routing.py
@@ -0,0 +1,353 @@
+"""
+Integration test for worker pool routing functionality.
+
+Tests that requests are routed to the correct pool based on command classification.
+"""
+
+import ctypes
+import logging
+import multiprocessing
+import time
+
+import pytest
+import tornado.gen
+import tornado.ioloop
+from pytestshellutils.utils.processes import terminate_process
+
+import salt.channel.server
+import salt.config
+import salt.crypt
+import salt.master
+import salt.payload
+import salt.utils.process
+import salt.utils.stringutils
+
+log = logging.getLogger(__name__)
+
+pytestmark = [
+    pytest.mark.slow_test,
+]
+
+
+class PoolReqServer(salt.utils.process.SignalHandlingProcess):
+    """
+    Test request server with pool routing enabled.
+    """
+
+    def __init__(self, config):
+        super().__init__()
+        self._closing = False
+        self.config = config
+        self.process_manager = salt.utils.process.ProcessManager(
+            name="PoolReqServer-ProcessManager"
+        )
+        self.io_loop = None
+        self.running = multiprocessing.Event()
+        self.handled_requests = multiprocessing.Manager().dict()
+
+    def run(self):
+        """Run the pool-aware request server."""
+        salt.master.SMaster.secrets["aes"] = {
+            "secret": multiprocessing.Array(
+                ctypes.c_char,
+                salt.utils.stringutils.to_bytes(
+                    salt.crypt.Crypticle.generate_key_string()
+                ),
+            ),
+            "serial": multiprocessing.Value(ctypes.c_longlong, lock=False),
+        }
+
+        self.io_loop = tornado.ioloop.IOLoop()
+        self.io_loop.make_current()
+
+        # Set up pool-specific channels
+        from salt.config.worker_pools import get_worker_pools_config
+
+        worker_pools = get_worker_pools_config(self.config)
+
+        # Create front-end channel
+        from salt.utils.channel import iter_transport_opts
+
+        frontend_channel = None
+        for transport, opts in iter_transport_opts(self.config):
+            frontend_channel = salt.channel.server.ReqServerChannel.factory(opts)
+            frontend_channel.pre_fork(self.process_manager)
+            break
+
+        # Create pool-specific channels
+        pool_channels = {}
+        for pool_name in worker_pools.keys():
+            pool_opts = self.config.copy()
+            pool_opts["pool_name"] = pool_name
+
+            for transport, opts in iter_transport_opts(pool_opts):
+                chan = salt.channel.server.ReqServerChannel.factory(opts)
+                chan.pre_fork(self.process_manager)
+                pool_channels[pool_name] = chan
+                break
+
+        # Create dispatcher
+        dispatcher = salt.channel.server.PoolDispatcherChannel(
+            self.config, [frontend_channel], pool_channels
+        )
+
+        def start_dispatcher():
+            """Start the dispatcher in the IO loop."""
+            dispatcher.post_fork(self.io_loop)
+
+        # Start dispatcher
+        self.io_loop.add_callback(start_dispatcher)
+
+        # Start workers for each pool
+        for pool_name, pool_config in worker_pools.items():
+            worker_count = pool_config.get("worker_count", 1)
+            pool_chan = pool_channels[pool_name]
+
+            for pool_index in range(worker_count):
+
+                def worker_handler(payload, pname=pool_name, pidx=pool_index):
+                    """Handler that tracks which pool handled the request."""
+                    return self._handle_payload(payload, pname, pidx)
+
+                # Start worker
+                pool_chan.post_fork(worker_handler, self.io_loop)
+
+        self.io_loop.add_callback(self.running.set)
+        try:
+            self.io_loop.start()
+        except (KeyboardInterrupt, SystemExit):
+            pass
+        finally:
+            self.close()
+
+    @tornado.gen.coroutine
+    def _handle_payload(self, payload, pool_name, pool_index):
+        """
+        Handle a payload and track which pool handled it.
+
+        :param payload: The request payload
+        :param pool_name: Name of the pool handling this request
+        :param pool_index: Index of the worker in the pool
+        """
+        try:
+            # Extract the command from the payload
+            if isinstance(payload, dict) and "load" in payload:
+                cmd = payload["load"].get("cmd", "unknown")
+            else:
+                cmd = "unknown"
+
+            # Track which pool handled this command
+            key = f"{cmd}_{time.time()}"
+            self.handled_requests[key] = {
+                "cmd": cmd,
+                "pool": pool_name,
+                "pool_index": pool_index,
+                "timestamp": time.time(),
+            }
+
+            log.info(
+                "Pool '%s' worker %d handled command '%s'",
+                pool_name,
+                pool_index,
+                cmd,
+            )
+
+            # Return response indicating which pool handled it
+            response = {
+                "handled_by_pool": pool_name,
+                "handled_by_worker": pool_index,
+                "original_payload": payload,
+            }
+
+            raise tornado.gen.Return((response, {"fun": "send_clear"}))
+        except Exception as exc:
+            log.error("Error in pool handler: %s", exc, exc_info=True)
+            raise tornado.gen.Return(({"error": str(exc)}, {"fun": "send_clear"}))
+
+    def _handle_signals(self, signum, sigframe):
+        self.close()
+        super()._handle_signals(signum, sigframe)
+
+    def __enter__(self):
+        self.start()
+        self.running.wait()
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+        self.terminate()
+
+    def close(self):
+        if self._closing:
+            return
+        self._closing = True
+        if self.process_manager is not None:
+            self.process_manager.terminate()
+            for pid in self.process_manager._process_map:
+                terminate_process(pid=pid, kill_children=True, slow_stop=False)
+            self.process_manager = None
+
+
+@pytest.fixture
+def pool_config(tmp_path):
+    """Create a master config with worker pools enabled."""
+    sock_dir = tmp_path / "sock"
+    pki_dir = tmp_path / "pki"
+    cache_dir = tmp_path / "cache"
+    sock_dir.mkdir()
+    pki_dir.mkdir()
+    cache_dir.mkdir()
+
+    return {
+        "sock_dir": str(sock_dir),
+        "pki_dir": str(pki_dir),
+        "cachedir": str(cache_dir),
+        "key_pass": "meh",
+        "keysize": 2048,
+        "cluster_id": None,
+        "master_sign_pubkey": False,
+        "pub_server_niceness": None,
+        "con_cache": False,
+        "zmq_monitor": False,
+        "request_server_ttl": 60,
+        "publish_session": 600,
+        "keys.cache_driver": "localfs_key",
+        "id": "master",
+        "optimization_order": [0, 1, 2],
+        "__role": "master",
+        "master_sign_key_name": "master_sign",
+        "permissive_pki_access": True,
+        "transport": "zeromq",
+        # Pool configuration
+        "worker_pools_enabled": True,
+        "worker_pools": {
+            "fast": {
+                "worker_count": 2,
+                "commands": ["test.ping", "test.echo", "runner.test.arg"],
+            },
+            "general": {
+                "worker_count": 3,
+                "commands": ["*"],  # Catchall
+            },
+        },
+    }
+
+
+@pytest.fixture
+def pool_req_server(pool_config):
+    """Create and start a pool-aware request server."""
+    server_process = PoolReqServer(pool_config)
+    try:
+        with server_process:
+            yield server_process
+    finally:
+        terminate_process(pid=server_process.pid, kill_children=True, slow_stop=False)
+
+
+def test_pool_routing_fast_commands(pool_req_server, pool_config):
+    """
+    Test that commands configured for the 'fast' pool are routed there.
+    """
+    # Create a simple request for a command in the fast pool
+    test_commands = ["test.ping", "test.echo"]
+
+    for cmd in test_commands:
+        payload = {"load": {"cmd": cmd, "arg": ["test"]}}
+
+        # In a real scenario, we'd send this via a ReqChannel
+        # For this test, we'll simulate the routing
+        from salt.master import RequestRouter
+
+        router = RequestRouter(pool_config)
+        routed_pool = router.route_request(payload)
+
+        assert routed_pool == "fast", f"Command '{cmd}' should route to 'fast' pool"
+
+
+def test_pool_routing_catchall_commands(pool_req_server, pool_config):
+    """
+    Test that commands not in any specific pool route to the catchall pool.
+    """
+    # Create a request for a command NOT in the fast pool
+    test_commands = ["state.highstate", "cmd.run", "pkg.install"]
+
+    for cmd in test_commands:
+        payload = {"load": {"cmd": cmd, "arg": ["test"]}}
+
+        from salt.master import RequestRouter
+
+        router = RequestRouter(pool_config)
+        routed_pool = router.route_request(payload)
+
+        assert (
+            routed_pool == "general"
+        ), f"Command '{cmd}' should route to 'general' pool (catchall)"
+
+
+def test_pool_routing_statistics(pool_config):
+    """
+    Test that the RequestRouter tracks routing statistics.
+    """
+    from salt.master import RequestRouter
+
+    router = RequestRouter(pool_config)
+
+    # Route some requests (pass dict, not serialized bytes)
+    test_data = [
+        ({"load": {"cmd": "test.ping"}}, "fast"),
+        ({"load": {"cmd": "test.echo"}}, "fast"),
+        ({"load": {"cmd": "state.highstate"}}, "general"),
+        ({"load": {"cmd": "cmd.run"}}, "general"),
+    ]
+
+    for payload, expected_pool in test_data:
+        routed_pool = router.route_request(payload)
+        assert routed_pool == expected_pool
+
+    # Check statistics (router.stats is a dict of pool_name -> count)
+    assert router.stats["fast"] == 2
+    assert router.stats["general"] == 2
+
+
+def test_pool_config_validation(pool_config):
+    """
+    Test that pool configuration validation works correctly.
+    """
+    from salt.config.worker_pools import validate_worker_pools_config
+
+    # Valid config should not raise
+    validate_worker_pools_config(pool_config)
+
+    # Invalid config: duplicate commands
+    invalid_config = pool_config.copy()
+    invalid_config["worker_pools"] = {
+        "pool1": {"worker_count": 2, "commands": ["test.ping"]},
+        "pool2": {
+            "worker_count": 2,
+            "commands": ["test.ping", "*"],
+        },  # Duplicate! (but has catchall)
+    }
+
+    with pytest.raises(
+        ValueError, match="Command 'test.ping' mapped to multiple pools"
+    ):
+        validate_worker_pools_config(invalid_config)
+
+
+def test_pool_disabled_fallback(tmp_path):
+    """
+    Test that when worker_pools_enabled=False, system uses legacy behavior.
+    """
+    config = {
+        "sock_dir": str(tmp_path / "sock"),
+        "pki_dir": str(tmp_path / "pki"),
+        "cachedir": str(tmp_path / "cache"),
+        "worker_pools_enabled": False,
+        "worker_threads": 5,
+    }
+
+    from salt.config.worker_pools import get_worker_pools_config
+
+    # When disabled, should return None
+    pools = get_worker_pools_config(config)
+    assert pools is None or pools == {}
diff --git a/tests/pytests/unit/config/test_worker_pools.py b/tests/pytests/unit/config/test_worker_pools.py
new file mode 100644
index 000000000000..c2f32934b229
--- /dev/null
+++ b/tests/pytests/unit/config/test_worker_pools.py
@@ -0,0 +1,157 @@
+"""
+Unit tests for worker pools configuration
+"""
+
+import pytest
+
+from salt.config.worker_pools import (
+    DEFAULT_WORKER_POOLS,
+    OPTIMIZED_WORKER_POOLS,
+    get_worker_pools_config,
+    validate_worker_pools_config,
+)
+
+
+class TestWorkerPoolsConfig:
+    """Test worker pools configuration functions"""
+
+    def test_default_worker_pools_structure(self):
+        """Test that DEFAULT_WORKER_POOLS has correct structure"""
+        assert isinstance(DEFAULT_WORKER_POOLS, dict)
+        assert "default" in DEFAULT_WORKER_POOLS
+        assert DEFAULT_WORKER_POOLS["default"]["worker_count"] == 5
+        assert DEFAULT_WORKER_POOLS["default"]["commands"] == ["*"]
+
+    def test_optimized_worker_pools_structure(self):
+        """Test that OPTIMIZED_WORKER_POOLS has correct structure"""
+        assert isinstance(OPTIMIZED_WORKER_POOLS, dict)
+        assert "lightweight" in OPTIMIZED_WORKER_POOLS
+        assert "medium" in OPTIMIZED_WORKER_POOLS
+        assert "heavy" in OPTIMIZED_WORKER_POOLS
+
+    def test_get_worker_pools_config_default(self):
+        """Test get_worker_pools_config with default config"""
+        opts = {"worker_pools_enabled": True, "worker_pools": {}}
+        result = get_worker_pools_config(opts)
+        assert result == DEFAULT_WORKER_POOLS
+
+    def test_get_worker_pools_config_disabled(self):
+        """Test get_worker_pools_config when pools are disabled"""
+        opts = {"worker_pools_enabled": False}
+        result = get_worker_pools_config(opts)
+        assert result is None
+
+    def test_get_worker_pools_config_worker_threads_compat(self):
+        """Test backward compatibility with worker_threads"""
+        opts = {"worker_pools_enabled": True, "worker_threads": 10, "worker_pools": {}}
+        result = get_worker_pools_config(opts)
+        assert result == {"default": {"worker_count": 10, "commands": ["*"]}}
+
+    def test_get_worker_pools_config_custom(self):
+        """Test get_worker_pools_config with custom pools"""
+        custom_pools = {
+            "fast": {"worker_count": 2, "commands": ["ping"]},
+            "slow": {"worker_count": 3, "commands": ["*"]},
+        }
+        opts = {"worker_pools_enabled": True, "worker_pools": custom_pools}
+        result = get_worker_pools_config(opts)
+        assert result == custom_pools
+
+    def test_get_worker_pools_config_optimized(self):
+        """Test get_worker_pools_config with optimized flag"""
+        opts = {"worker_pools_enabled": True, "worker_pools_optimized": True}
+        result = get_worker_pools_config(opts)
+        assert result == OPTIMIZED_WORKER_POOLS
+
+    def test_validate_worker_pools_config_valid_default(self):
+        """Test validation with valid default config"""
+        opts = {"worker_pools_enabled": True, "worker_pools": DEFAULT_WORKER_POOLS}
+        assert validate_worker_pools_config(opts) is True
+
+    def test_validate_worker_pools_config_valid_catchall(self):
+        """Test validation with valid catchall pool"""
+        opts = {
+            "worker_pools_enabled": True,
+            "worker_pools": {
+                "fast": {"worker_count": 2, "commands": ["ping"]},
+                "slow": {"worker_count": 3, "commands": ["*"]},
+            },
+        }
+        assert validate_worker_pools_config(opts) is True
+
+    def test_validate_worker_pools_config_valid_default_pool(self):
+        """Test validation with valid explicit default pool"""
+        opts = {
+            "worker_pools_enabled": True,
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+                "pool2": {"worker_count": 3, "commands": ["_pillar"]},
+            },
+            "worker_pool_default": "pool2",
+        }
+        assert validate_worker_pools_config(opts) is True
+
+    def test_validate_worker_pools_config_duplicate_catchall(self):
+        """Test validation catches duplicate catchall"""
+        opts = {
+            "worker_pools_enabled": True,
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["*"]},
+                "pool2": {"worker_count": 3, "commands": ["*"]},
+            },
+        }
+        with pytest.raises(ValueError, match="Multiple pools have catchall"):
+            validate_worker_pools_config(opts)
+
+    def test_validate_worker_pools_config_duplicate_command(self):
+        """Test validation catches duplicate commands"""
+        opts = {
+            "worker_pools_enabled": True,
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+                "pool2": {"worker_count": 3, "commands": ["ping"]},
+            },
+            "worker_pool_default": "pool1",
+        }
+        with pytest.raises(ValueError, match="Command 'ping' mapped to multiple pools"):
+            validate_worker_pools_config(opts)
+
+    def test_validate_worker_pools_config_invalid_worker_count(self):
+        """Test validation catches invalid worker_count"""
+        opts = {
+            "worker_pools_enabled": True,
+            "worker_pools": {
+                "pool1": {"worker_count": 0, "commands": ["*"]},
+            },
+        }
+        with pytest.raises(ValueError, match="worker_count must be integer >= 1"):
+            validate_worker_pools_config(opts)
+
+    def test_validate_worker_pools_config_missing_default_pool(self):
+        """Test validation catches missing default pool"""
+        opts = {
+            "worker_pools_enabled": True,
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+            },
+            "worker_pool_default": "nonexistent",
+        }
+        with pytest.raises(ValueError, match="not found in worker_pools"):
+            validate_worker_pools_config(opts)
+
+    def test_validate_worker_pools_config_no_catchall_no_default(self):
+        """Test validation requires either catchall or default pool"""
+        opts = {
+            "worker_pools_enabled": True,
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+            },
+            "worker_pool_default": None,
+        }
+        with pytest.raises(ValueError, match="Either use a catchall pool"):
+            validate_worker_pools_config(opts)
+
+    def test_validate_worker_pools_config_disabled(self):
+        """Test validation passes when pools are disabled"""
+        opts = {"worker_pools_enabled": False}
+        assert validate_worker_pools_config(opts) is True
diff --git a/tests/pytests/unit/conftest.py b/tests/pytests/unit/conftest.py
index d58ce1f97052..99055729669b 100644
--- a/tests/pytests/unit/conftest.py
+++ b/tests/pytests/unit/conftest.py
@@ -53,6 +53,27 @@ def master_opts(tmp_path):
     opts["publish_signing_algorithm"] = (
         "PKCS1v15-SHA224" if FIPS_TESTRUN else "PKCS1v15-SHA1"
     )
+
+    # Use optimized worker pools for tests to demonstrate the feature
+    # This separates fast operations from slow ones for better performance
+    opts["worker_pools_enabled"] = True
+    opts["worker_pools"] = {
+        "fast": {
+            "worker_count": 2,
+            "commands": [
+                "ping",
+                "get_token",
+                "mk_token",
+                "verify_minion",
+                "_master_opts",
+            ],
+        },
+        "general": {
+            "worker_count": 3,
+            "commands": ["*"],  # Catchall for everything else
+        },
+    }
+
     return opts
 
 
diff --git a/tests/pytests/unit/test_pool_name_edge_cases.py b/tests/pytests/unit/test_pool_name_edge_cases.py
new file mode 100644
index 000000000000..80e8c9ea6e4b
--- /dev/null
+++ b/tests/pytests/unit/test_pool_name_edge_cases.py
@@ -0,0 +1,337 @@
+"""
+Unit tests for pool name edge cases - especially special characters in pool names.
+
+Tests that pool names with special characters don't break URI construction,
+file path creation, or cause security issues.
+"""
+
+import pytest
+
+import salt.transport.zeromq
+from salt.config.worker_pools import validate_worker_pools_config
+
+
+class TestPoolNameSpecialCharacters:
+    """Test pool names with various special characters."""
+
+    @pytest.fixture
+    def base_pool_config(self, tmp_path):
+        """Base configuration for pool tests."""
+        sock_dir = tmp_path / "sock"
+        pki_dir = tmp_path / "pki"
+        cache_dir = tmp_path / "cache"
+        sock_dir.mkdir()
+        pki_dir.mkdir()
+        cache_dir.mkdir()
+
+        return {
+            "sock_dir": str(sock_dir),
+            "pki_dir": str(pki_dir),
+            "cachedir": str(cache_dir),
+            "worker_pools_enabled": True,
+            "ipc_mode": "",  # Use IPC mode
+        }
+
+    def test_pool_name_with_spaces(self, base_pool_config):
+        """Pool name with spaces should work."""
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            "fast pool": {
+                "worker_count": 2,
+                "commands": ["test.ping", "*"],
+            }
+        }
+
+        # Should validate successfully
+        validate_worker_pools_config(config)
+
+        # Test URI construction
+        config["pool_name"] = "fast pool"
+        transport = salt.transport.zeromq.RequestServer(config)
+        uri = transport.get_worker_uri()
+
+        # Should create valid IPC URI with pool name
+        assert "workers-fast pool.ipc" in uri
+        assert uri.startswith("ipc://")
+
+    def test_pool_name_with_dashes_underscores(self, base_pool_config):
+        """Pool name with dashes and underscores (common, should work)."""
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            "fast-pool_1": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        validate_worker_pools_config(config)
+
+        config["pool_name"] = "fast-pool_1"
+        transport = salt.transport.zeromq.RequestServer(config)
+        uri = transport.get_worker_uri()
+
+        assert "workers-fast-pool_1.ipc" in uri
+
+    def test_pool_name_with_dots(self, base_pool_config):
+        """Pool name with dots should work but creates interesting paths."""
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            "pool.fast": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        validate_worker_pools_config(config)
+
+        config["pool_name"] = "pool.fast"
+        transport = salt.transport.zeromq.RequestServer(config)
+        uri = transport.get_worker_uri()
+
+        # Should create workers-pool.fast.ipc (not a relative path)
+        assert "workers-pool.fast.ipc" in uri
+        # Verify it's not treated as directory.file
+        assert ".." not in uri
+
+    def test_pool_name_with_slash_rejected(self, base_pool_config):
+        """Pool name with slash is rejected by validation to prevent path traversal."""
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            "fast/pool": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        # Config validation should reject pool names with slashes
+        with pytest.raises(ValueError, match="path separators"):
+            validate_worker_pools_config(config)
+
+    def test_pool_name_path_traversal_attempt(self, base_pool_config):
+        """Pool name attempting path traversal is rejected by validation."""
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            "../evil": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        # Config validation should reject path traversal attempts
+        with pytest.raises(ValueError, match="path traversal"):
+            validate_worker_pools_config(config)
+
+    def test_pool_name_with_unicode(self, base_pool_config):
+        """Pool name with unicode characters."""
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            "快速池": {  # Chinese for "fast pool"
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        validate_worker_pools_config(config)
+
+        config["pool_name"] = "快速池"
+        transport = salt.transport.zeromq.RequestServer(config)
+        uri = transport.get_worker_uri()
+
+        # Should handle unicode in URI
+        assert "workers-快速池.ipc" in uri or "workers-" in uri
+
+    def test_pool_name_with_special_chars(self, base_pool_config):
+        """Pool name with various special characters."""
+        special_chars = "!@#$%^&*()"
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            special_chars: {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        validate_worker_pools_config(config)
+
+        config["pool_name"] = special_chars
+        transport = salt.transport.zeromq.RequestServer(config)
+        uri = transport.get_worker_uri()
+
+        # Should create some kind of valid URI (may be escaped/sanitized)
+        assert uri.startswith("ipc://")
+        assert config["sock_dir"] in uri
+
+    def test_pool_name_very_long(self, base_pool_config):
+        """Pool name that's very long - could exceed path limits."""
+        long_name = "a" * 300  # 300 chars
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            long_name: {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        validate_worker_pools_config(config)
+
+        config["pool_name"] = long_name
+        transport = salt.transport.zeromq.RequestServer(config)
+        uri = transport.get_worker_uri()
+
+        # Check if resulting path would exceed Unix socket path limit (typically 108 bytes)
+        socket_path = uri.replace("ipc://", "")
+        if len(socket_path) > 108:
+            # This could fail at bind time on Unix systems
+            pytest.skip(
+                f"Socket path too long ({len(socket_path)} > 108): {socket_path}"
+            )
+
+    def test_pool_name_empty_string(self, base_pool_config):
+        """Pool name as empty string is rejected by validation."""
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            "": {  # Empty string as pool name
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        # Validation should reject empty pool names
+        with pytest.raises(ValueError, match="cannot be empty"):
+            validate_worker_pools_config(config)
+
+    def test_pool_name_tcp_mode_hash_collision(self, base_pool_config):
+        """Test that different pool names don't collide in TCP port assignment."""
+        config = base_pool_config.copy()
+        config["ipc_mode"] = "tcp"
+        config["tcp_master_workers"] = 4515
+
+        # Create two pools and check their ports
+        pools_to_test = ["pool1", "pool2", "fast", "general", "test"]
+        ports = []
+
+        for pool_name in pools_to_test:
+            config["pool_name"] = pool_name
+            transport = salt.transport.zeromq.RequestServer(config)
+            uri = transport.get_worker_uri()
+
+            # Extract port from URI like "tcp://127.0.0.1:4516"
+            port = int(uri.split(":")[-1])
+            ports.append((pool_name, port))
+
+        # Check no two pools got same port
+        port_numbers = [p[1] for p in ports]
+        unique_ports = set(port_numbers)
+
+        if len(unique_ports) < len(port_numbers):
+            # Found collision
+            collisions = []
+            for i, (name1, port1) in enumerate(ports):
+                for name2, port2 in ports[i + 1 :]:
+                    if port1 == port2:
+                        collisions.append((name1, name2, port1))
+
+            pytest.fail(f"Port collisions found: {collisions}")
+
+    def test_pool_name_tcp_mode_port_range(self, base_pool_config):
+        """Test that TCP port offsets stay in reasonable range."""
+        config = base_pool_config.copy()
+        config["ipc_mode"] = "tcp"
+        config["tcp_master_workers"] = 4515
+
+        # Test various pool names
+        pool_names = ["a", "z", "AAA", "zzz", "pool1", "pool999", "🎉", "!@#$"]
+
+        for pool_name in pool_names:
+            config["pool_name"] = pool_name
+            transport = salt.transport.zeromq.RequestServer(config)
+            uri = transport.get_worker_uri()
+
+            port = int(uri.split(":")[-1])
+
+            # Port should be base + offset, offset is hash(name) % 1000
+            # So port should be in range [4515, 5515)
+            assert (
+                4515 <= port < 5515
+            ), f"Pool '{pool_name}' got port {port} outside expected range"
+
+    def test_pool_name_null_byte(self, base_pool_config):
+        """Pool name with null byte - potential security issue."""
+        config = base_pool_config.copy()
+        pool_name_with_null = "pool\x00evil"
+
+        config["worker_pools"] = {
+            pool_name_with_null: {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        # Validation might fail or succeed depending on Python version
+        try:
+            validate_worker_pools_config(config)
+
+            config["pool_name"] = pool_name_with_null
+            transport = salt.transport.zeromq.RequestServer(config)
+            uri = transport.get_worker_uri()
+
+            # Null byte should not truncate the path or cause issues
+            # OS will reject paths with null bytes
+            assert "\x00" not in uri or True  # Either stripped or will fail at bind
+        except (ValueError, OSError):
+            # Expected - null bytes should be rejected somewhere
+            pass
+
+    def test_pool_name_windows_reserved(self, base_pool_config):
+        """Pool names that are Windows reserved names."""
+        reserved_names = ["CON", "PRN", "AUX", "NUL", "COM1", "LPT1"]
+
+        for reserved in reserved_names:
+            config = base_pool_config.copy()
+            config["worker_pools"] = {
+                reserved: {
+                    "worker_count": 2,
+                    "commands": ["*"],
+                }
+            }
+
+            validate_worker_pools_config(config)
+
+            config["pool_name"] = reserved
+            transport = salt.transport.zeromq.RequestServer(config)
+            uri = transport.get_worker_uri()
+
+            # On Windows, these might cause issues
+            # On Unix, should work fine
+            assert uri.startswith("ipc://")
+
+    def test_pool_name_only_dots(self, base_pool_config):
+        """Pool name that's just dots - '..' is rejected, '.' and '...' are allowed."""
+        # Single dot is allowed
+        config = base_pool_config.copy()
+        config["worker_pools"] = {
+            ".": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+        validate_worker_pools_config(config)  # Should succeed
+
+        # Double dot is rejected (path traversal)
+        config["worker_pools"] = {
+            "..": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+        with pytest.raises(ValueError, match="path traversal"):
+            validate_worker_pools_config(config)
+
+        # Three dots is allowed (not a special path component)
+        config["worker_pools"] = {
+            "...": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+        validate_worker_pools_config(config)  # Should succeed
diff --git a/tests/pytests/unit/test_pool_name_validation.py b/tests/pytests/unit/test_pool_name_validation.py
new file mode 100644
index 000000000000..933b787381fd
--- /dev/null
+++ b/tests/pytests/unit/test_pool_name_validation.py
@@ -0,0 +1,198 @@
+r"""
+Unit tests for pool name validation.
+
+Tests minimal security-focused validation:
+- Blocks path traversal (/, \, ..)
+- Blocks empty strings
+- Blocks null bytes
+- Allows everything else (spaces, dots, unicode, special chars)
+"""
+
+import pytest
+
+from salt.config.worker_pools import validate_worker_pools_config
+
+
+class TestPoolNameValidation:
+    """Test pool name validation rules (Option A: Minimal security-focused)."""
+
+    @pytest.fixture
+    def base_config(self, tmp_path):
+        """Base configuration for pool tests."""
+        return {
+            "sock_dir": str(tmp_path / "sock"),
+            "pki_dir": str(tmp_path / "pki"),
+            "cachedir": str(tmp_path / "cache"),
+            "worker_pools_enabled": True,
+        }
+
+    def test_valid_pool_names_basic(self, base_config):
+        """Valid pool names with various safe characters."""
+        valid_names = [
+            "fast",
+            "general",
+            "pool1",
+            "pool2",
+            "MyPool",
+            "UPPERCASE",
+            "lowercase",
+            "Pool123",
+            "123pool",
+            "-fast",  # NOW ALLOWED - can start with hyphen
+            "_general",  # NOW ALLOWED - can start with underscore
+            "fast pool",  # NOW ALLOWED - spaces are fine
+            "pool.fast",  # NOW ALLOWED - dots are fine
+            "fast-pool_1",  # Mixed characters
+            "my_pool-2",
+            "快速池",  # NOW ALLOWED - unicode is fine
+            "!@#$%^&*()",  # NOW ALLOWED - special chars (except / \ null)
+            ".",  # NOW ALLOWED - single dot is fine
+            "...",  # NOW ALLOWED - multiple dots fine (not at start as ../)
+        ]
+
+        for name in valid_names:
+            config = base_config.copy()
+            config["worker_pools"] = {
+                name: {
+                    "worker_count": 2,
+                    "commands": ["*"],
+                }
+            }
+
+            # Should not raise
+            try:
+                validate_worker_pools_config(config)
+            except ValueError as e:
+                pytest.fail(f"Pool name '{name}' should be valid but got error: {e}")
+
+    def test_invalid_pool_name_with_forward_slash(self, base_config):
+        """Pool name with forward slash is rejected (prevents path traversal)."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "fast/pool": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError, match="path separators"):
+            validate_worker_pools_config(config)
+
+    def test_invalid_pool_name_with_backslash(self, base_config):
+        """Pool name with backslash is rejected (prevents path traversal on Windows)."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "fast\\pool": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError, match="path separators"):
+            validate_worker_pools_config(config)
+
+    def test_invalid_pool_name_dotdot_only(self, base_config):
+        """Pool name that is exactly '..' is rejected."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "..": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError, match="path traversal"):
+            validate_worker_pools_config(config)
+
+    def test_invalid_pool_name_dotdot_slash_prefix(self, base_config):
+        """Pool name starting with '../' is rejected."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "../evil": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError, match="path traversal"):
+            validate_worker_pools_config(config)
+
+    def test_invalid_pool_name_dotdot_backslash_prefix(self, base_config):
+        """Pool name starting with '..\\' is rejected."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "..\\evil": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError, match="path traversal"):
+            validate_worker_pools_config(config)
+
+    def test_invalid_pool_name_empty_string(self, base_config):
+        """Pool name as empty string is rejected."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError, match="Pool name cannot be empty"):
+            validate_worker_pools_config(config)
+
+    def test_invalid_pool_name_null_byte(self, base_config):
+        """Pool name with null byte is rejected."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "pool\x00evil": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError, match="null byte"):
+            validate_worker_pools_config(config)
+
+    def test_invalid_pool_name_not_string(self, base_config):
+        """Pool name that's not a string is rejected."""
+        # Note: can only test hashable types since dict keys must be hashable
+        invalid_names = [
+            123,
+            12.5,
+            None,
+            True,
+        ]
+
+        for invalid_name in invalid_names:
+            config = base_config.copy()
+            config["worker_pools"] = {
+                invalid_name: {
+                    "worker_count": 2,
+                    "commands": ["*"],
+                }
+            }
+
+            with pytest.raises(ValueError, match="Pool name must be a string"):
+                validate_worker_pools_config(config)
+
+    def test_error_message_format_path_separator(self, base_config):
+        """Verify error message for path separator is clear."""
+        config = base_config.copy()
+        config["worker_pools"] = {
+            "bad/name": {
+                "worker_count": 2,
+                "commands": ["*"],
+            }
+        }
+
+        with pytest.raises(ValueError) as exc_info:
+            validate_worker_pools_config(config)
+
+        error_msg = str(exc_info.value)
+        # Should explain why it's rejected
+        assert "path" in error_msg.lower() and (
+            "separator" in error_msg.lower() or "traversal" in error_msg.lower()
+        )
diff --git a/tests/pytests/unit/test_request_router.py b/tests/pytests/unit/test_request_router.py
new file mode 100644
index 000000000000..fa1d5c85ce45
--- /dev/null
+++ b/tests/pytests/unit/test_request_router.py
@@ -0,0 +1,161 @@
+"""
+Unit tests for RequestRouter class
+"""
+
+import pytest
+
+from salt.master import RequestRouter
+
+
+class TestRequestRouter:
+    """Test RequestRouter request classification and routing"""
+
+    def test_router_initialization_with_catchall(self):
+        """Test router initializes correctly with catchall pool"""
+        opts = {
+            "worker_pools": {
+                "fast": {"worker_count": 2, "commands": ["ping", "verify_minion"]},
+                "default": {"worker_count": 3, "commands": ["*"]},
+            }
+        }
+        router = RequestRouter(opts)
+        assert router.default_pool == "default"
+        assert "ping" in router.cmd_to_pool
+        assert router.cmd_to_pool["ping"] == "fast"
+
+    def test_router_initialization_with_explicit_default(self):
+        """Test router initializes correctly with explicit default pool"""
+        opts = {
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+                "pool2": {"worker_count": 3, "commands": ["_pillar"]},
+            },
+            "worker_pool_default": "pool2",
+        }
+        router = RequestRouter(opts)
+        assert router.default_pool == "pool2"
+
+    def test_router_route_to_specific_pool(self):
+        """Test routing to specific pool based on command"""
+        opts = {
+            "worker_pools": {
+                "fast": {"worker_count": 2, "commands": ["ping", "verify_minion"]},
+                "slow": {"worker_count": 3, "commands": ["_pillar", "_return"]},
+                "default": {"worker_count": 2, "commands": ["*"]},
+            }
+        }
+        router = RequestRouter(opts)
+
+        # Test explicit mappings
+        assert router.route_request({"load": {"cmd": "ping"}}) == "fast"
+        assert router.route_request({"load": {"cmd": "verify_minion"}}) == "fast"
+        assert router.route_request({"load": {"cmd": "_pillar"}}) == "slow"
+        assert router.route_request({"load": {"cmd": "_return"}}) == "slow"
+
+    def test_router_route_to_catchall(self):
+        """Test routing unmapped commands to catchall pool"""
+        opts = {
+            "worker_pools": {
+                "fast": {"worker_count": 2, "commands": ["ping"]},
+                "default": {"worker_count": 3, "commands": ["*"]},
+            }
+        }
+        router = RequestRouter(opts)
+
+        # Unmapped command should go to catchall
+        assert router.route_request({"load": {"cmd": "unknown_command"}}) == "default"
+        assert router.route_request({"load": {"cmd": "_pillar"}}) == "default"
+
+    def test_router_route_to_explicit_default(self):
+        """Test routing unmapped commands to explicit default pool"""
+        opts = {
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+                "pool2": {"worker_count": 3, "commands": ["_pillar"]},
+            },
+            "worker_pool_default": "pool2",
+        }
+        router = RequestRouter(opts)
+
+        # Unmapped command should go to default
+        assert router.route_request({"load": {"cmd": "unknown"}}) == "pool2"
+
+    def test_router_extract_command_from_payload(self):
+        """Test command extraction from various payload formats"""
+        opts = {"worker_pools": {"default": {"worker_count": 5, "commands": ["*"]}}}
+        router = RequestRouter(opts)
+
+        # Normal payload
+        assert router._extract_command({"load": {"cmd": "ping"}}) == "ping"
+
+        # Missing cmd
+        assert router._extract_command({"load": {}}) == ""
+
+        # Missing load
+        assert router._extract_command({}) == ""
+
+        # Invalid payload
+        assert router._extract_command(None) == ""
+
+    def test_router_statistics_tracking(self):
+        """Test that router tracks statistics per pool"""
+        opts = {
+            "worker_pools": {
+                "fast": {"worker_count": 2, "commands": ["ping"]},
+                "slow": {"worker_count": 3, "commands": ["_pillar"]},
+                "default": {"worker_count": 2, "commands": ["*"]},
+            }
+        }
+        router = RequestRouter(opts)
+
+        # Initial stats should be zero
+        assert router.stats["fast"] == 0
+        assert router.stats["slow"] == 0
+        assert router.stats["default"] == 0
+
+        # Route some requests
+        router.route_request({"load": {"cmd": "ping"}})
+        router.route_request({"load": {"cmd": "ping"}})
+        router.route_request({"load": {"cmd": "_pillar"}})
+        router.route_request({"load": {"cmd": "unknown"}})
+
+        # Check stats
+        assert router.stats["fast"] == 2
+        assert router.stats["slow"] == 1
+        assert router.stats["default"] == 1
+
+    def test_router_fails_duplicate_catchall(self):
+        """Test router fails to initialize with duplicate catchall"""
+        opts = {
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["*"]},
+                "pool2": {"worker_count": 3, "commands": ["*"]},
+            }
+        }
+        with pytest.raises(ValueError, match="Multiple pools have catchall"):
+            RequestRouter(opts)
+
+    def test_router_fails_duplicate_command(self):
+        """Test router fails to initialize with duplicate command mapping"""
+        opts = {
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+                "pool2": {"worker_count": 3, "commands": ["ping"]},
+            },
+            "worker_pool_default": "pool1",
+        }
+        with pytest.raises(ValueError, match="Command 'ping' mapped to multiple pools"):
+            RequestRouter(opts)
+
+    def test_router_fails_no_default(self):
+        """Test router fails without catchall or explicit default"""
+        opts = {
+            "worker_pools": {
+                "pool1": {"worker_count": 2, "commands": ["ping"]},
+            },
+            "worker_pool_default": None,
+        }
+        with pytest.raises(
+            ValueError, match="Configuration must have either.*catchall.*default"
+        ):
+            RequestRouter(opts)
diff --git a/tests/pytests/unit/transport/test_zeromq.py b/tests/pytests/unit/transport/test_zeromq.py
index bc22aabb242b..477a56b74b18 100644
--- a/tests/pytests/unit/transport/test_zeromq.py
+++ b/tests/pytests/unit/transport/test_zeromq.py
@@ -1718,6 +1718,8 @@ async def test_client_send_recv_on_cancelled_error(minion_opts, io_loop):
         client.socket = AsyncMock()
         client.socket.poll.side_effect = zmq.eventloop.future.CancelledError
         client._queue.put_nowait((mock_future, {"meh": "bah"}))
+        # Add a sentinel to stop the loop, otherwise it will wait for more items
+        client._queue.put_nowait((None, None))
         await client._send_recv(client.socket, client._queue)
         mock_future.set_exception.assert_not_called()
     finally:
diff --git a/tests/pytests/unit/transport/test_zeromq_concurrency.py b/tests/pytests/unit/transport/test_zeromq_concurrency.py
new file mode 100644
index 000000000000..ee07f3d2ef4d
--- /dev/null
+++ b/tests/pytests/unit/transport/test_zeromq_concurrency.py
@@ -0,0 +1,87 @@
+import asyncio
+
+import zmq
+
+import salt.transport.zeromq
+from tests.support.mock import AsyncMock
+
+
+async def test_request_client_concurrency_serialization(minion_opts, io_loop):
+    """
+    Regression test for EFSM (invalid state) errors in RequestClient.
+    Ensures that multiple concurrent send() calls are serialized through
+    the queue and don't violate the REQ socket state machine.
+    """
+    client = salt.transport.zeromq.RequestClient(minion_opts, io_loop)
+
+    # Mock the socket to track state
+    mock_socket = AsyncMock()
+    socket_state = {"busy": False}
+
+    async def mocked_send(msg, **kwargs):
+        if socket_state["busy"]:
+            raise zmq.ZMQError(zmq.EFSM, "Socket busy!")
+        socket_state["busy"] = True
+        await asyncio.sleep(0.01)  # Simulate network delay
+
+    async def mocked_recv(**kwargs):
+        if not socket_state["busy"]:
+            raise zmq.ZMQError(zmq.EFSM, "Nothing to recv!")
+        socket_state["busy"] = False
+        return salt.payload.dumps({"ret": "ok"})
+
+    mock_socket.send = mocked_send
+    mock_socket.recv = mocked_recv
+    mock_socket.poll.return_value = True
+
+    # Connect to initialize everything
+    await client.connect()
+
+    # Inject the mock socket
+    if client.socket:
+        client.socket.close()
+    client.socket = mock_socket
+    # Ensure the background task uses our mock
+    if client.send_recv_task:
+        client.send_recv_task.cancel()
+
+    client.send_recv_task = asyncio.create_task(
+        client._send_recv(mock_socket, client._queue, task_id=client.send_recv_task_id)
+    )
+
+    # Hammer the client with concurrent requests
+    tasks = []
+    for i in range(50):
+        tasks.append(asyncio.create_task(client.send({"foo": i}, timeout=10)))
+
+    results = await asyncio.gather(*tasks)
+
+    assert len(results) == 50
+    assert all(r == {"ret": "ok"} for r in results)
+    assert socket_state["busy"] is False
+    client.close()
+
+
+async def test_request_client_reconnect_task_safety(minion_opts, io_loop):
+    """
+    Regression test for task leaks and state corruption during reconnections.
+    Ensures that when a task is superseded, it re-queues its message and exits.
+    """
+    client = salt.transport.zeromq.RequestClient(minion_opts, io_loop)
+    await client.connect()
+
+    # Mock socket that always times out once
+    mock_socket = AsyncMock()
+    mock_socket.poll.return_value = False  # Trigger timeout in _send_recv
+
+    if client.socket:
+        client.socket.close()
+    client.socket = mock_socket
+    original_task_id = client.send_recv_task_id
+
+    # Trigger a reconnection by calling _reconnect (simulates error in loop)
+    await client._reconnect()
+    assert client.send_recv_task_id == original_task_id + 1
+
+    # The old task should have exited cleanly.
+    client.close()
diff --git a/tests/pytests/unit/transport/test_zeromq_worker_pools.py b/tests/pytests/unit/transport/test_zeromq_worker_pools.py
new file mode 100644
index 000000000000..60fb16c936be
--- /dev/null
+++ b/tests/pytests/unit/transport/test_zeromq_worker_pools.py
@@ -0,0 +1,139 @@
+"""
+Unit tests for ZeroMQ worker pool functionality
+"""
+
+import inspect
+
+import pytest
+
+import salt.transport.zeromq
+
+pytestmark = [
+    pytest.mark.core_test,
+]
+
+
+class TestWorkerPoolCodeStructure:
+    """
+    Tests to verify the code structure of worker pool methods to catch
+    common Python scoping issues that only manifest at runtime.
+    """
+
+    def test_zmq_device_pooled_imports_before_usage(self):
+        """
+        Test that zmq_device_pooled has imports in the correct order.
+
+        This test verifies that the 'import salt.master' statement appears
+        BEFORE any usage of salt.utils.files.fopen(). This prevents the
+        UnboundLocalError bug where:
+        - Line X uses salt.utils.files.fopen()
+        - Line Y has 'import salt.master' (Y > X)
+        - Python sees the import and treats 'salt' as a local variable
+        - Results in: UnboundLocalError: cannot access local variable 'salt'
+        """
+        # Get the source code of zmq_device_pooled
+        source = inspect.getsource(
+            salt.transport.zeromq.RequestServer.zmq_device_pooled
+        )
+
+        # Find the line numbers
+        import_salt_master_line = None
+        fopen_usage_line = None
+
+        for line_num, line in enumerate(source.split("\n"), 1):
+            if "import salt.master" in line:
+                import_salt_master_line = line_num
+            if "salt.utils.files.fopen" in line:
+                fopen_usage_line = line_num
+
+        # Verify both exist
+        assert (
+            import_salt_master_line is not None
+        ), "Expected 'import salt.master' in zmq_device_pooled"
+        assert (
+            fopen_usage_line is not None
+        ), "Expected 'salt.utils.files.fopen' usage in zmq_device_pooled"
+
+        # The import must come before the usage
+        assert import_salt_master_line < fopen_usage_line, (
+            f"'import salt.master' at line {import_salt_master_line} must appear "
+            f"BEFORE 'salt.utils.files.fopen' at line {fopen_usage_line}. "
+            f"Otherwise Python will treat 'salt' as a local variable and "
+            f"raise UnboundLocalError."
+        )
+
+    def test_zmq_device_pooled_has_worker_pools_param(self):
+        """
+        Test that zmq_device_pooled accepts worker_pools parameter.
+        """
+        sig = inspect.signature(salt.transport.zeromq.RequestServer.zmq_device_pooled)
+        assert (
+            "worker_pools" in sig.parameters
+        ), "zmq_device_pooled should have worker_pools parameter"
+
+    def test_zmq_device_pooled_creates_marker_file(self):
+        """
+        Test that zmq_device_pooled includes code to create workers.ipc marker file.
+
+        This marker file is required for netapi's _is_master_running() check.
+        """
+        source = inspect.getsource(
+            salt.transport.zeromq.RequestServer.zmq_device_pooled
+        )
+
+        # Check for marker file creation
+        assert (
+            "workers.ipc" in source
+        ), "zmq_device_pooled should create workers.ipc marker file"
+        assert (
+            "salt.utils.files.fopen" in source or "open(" in source
+        ), "zmq_device_pooled should use fopen or open to create marker file"
+        assert (
+            "os.chmod" in source
+        ), "zmq_device_pooled should set permissions on marker file"
+
+    def test_zmq_device_pooled_uses_router(self):
+        """
+        Test that zmq_device_pooled creates and uses RequestRouter for routing.
+        """
+        source = inspect.getsource(
+            salt.transport.zeromq.RequestServer.zmq_device_pooled
+        )
+
+        assert (
+            "RequestRouter" in source
+        ), "zmq_device_pooled should create RequestRouter instance"
+        assert (
+            "route_request" in source
+        ), "zmq_device_pooled should call route_request method"
+
+
+class TestRequestServerIntegration:
+    """
+    Tests for RequestServer that verify worker pool setup without
+    actually running multiprocessing code.
+    """
+
+    def test_pre_fork_with_worker_pools(self):
+        """
+        Test that pre_fork method exists and accepts *args and **kwargs.
+        """
+        sig = inspect.signature(salt.transport.zeromq.RequestServer.pre_fork)
+        assert (
+            "process_manager" in sig.parameters
+        ), "pre_fork should have process_manager parameter"
+        assert "args" in sig.parameters, "pre_fork should have *args parameter"
+        assert "kwargs" in sig.parameters, "pre_fork should have **kwargs parameter"
+
+    def test_request_server_has_zmq_device_pooled_method(self):
+        """
+        Test that RequestServer has the zmq_device_pooled method.
+        """
+        assert hasattr(
+            salt.transport.zeromq.RequestServer, "zmq_device_pooled"
+        ), "RequestServer should have zmq_device_pooled method"
+
+        # Verify it's a callable method
+        assert callable(
+            salt.transport.zeromq.RequestServer.zmq_device_pooled
+        ), "zmq_device_pooled should be callable"