diff --git a/.rubocop_todo.yml b/.rubocop_todo.yml
index f5a31b5e..58636a71 100644
--- a/.rubocop_todo.yml
+++ b/.rubocop_todo.yml
@@ -17,6 +17,6 @@ Metrics/BlockLength:
 # Offense count: 1
 # Configuration parameters: CountComments, CountAsOne.
 Metrics/ClassLength:
-  Max: 111
+  Max: 200
   Exclude:
     - 'lib/valkey.rb'
diff --git a/AGENTS.md b/AGENTS.md
index 9ee0adcc..49bf81cb 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -287,7 +287,7 @@ cargo fmt --manifest-path ./Cargo.toml --all
 ### Never Commit
 
 - Secrets, `.env` credentials, production URLs
-- Debug `puts` in production code paths (the native Pub/Sub callback in `lib/valkey/glide/pubsub.rb` must never `puts` or block: it runs on a Rust thread under a borrowed GVL)
+- Debug `puts` in production code paths (the native Pub/Sub callback in `lib/valkey/glide/pubsub_receiver.rb` must never `puts` or block: it runs on a Rust thread under a borrowed GVL)
 
 ## Project Structure (Essential)
 
@@ -297,7 +297,7 @@ valkey-glide-ruby/
 ├── lib/valkey/
 │   ├── bindings.rb
 │   ├── native/{arch}-{os}/libglide_ffi.{so,dylib}   # bundled per-platform lib (packaged during CD)
-│   ├── glide/pubsub.rb   # all Pub/Sub logic; internal, wired into Valkey
+│   ├── glide/pubsub_receiver.rb   # internal; Pub/Sub message queue + FFI push handler
 │   ├── commands.rb       # requires + includes all command modules
 │   ├── commands/*.rb     # 20 command-family modules
 │   ├── opentelemetry.rb
diff --git a/lib/valkey.rb b/lib/valkey.rb
index 54bc4f16..08dbd0ce 100644
--- a/lib/valkey.rb
+++ b/lib/valkey.rb
@@ -15,10 +15,12 @@
 require "valkey/commands"
 require "valkey/errors"
 require "valkey/future"
-require "valkey/glide/pubsub"
 require "valkey/pipeline"
 require "valkey/opentelemetry"
 require "valkey/route"
+require "valkey/glide/pubsub_message"
+require "valkey/glide/pubsub_state"
+require "valkey/glide/pubsub_receiver"
 
 class Valkey
   include Utils
@@ -29,6 +31,9 @@ class Valkey
   # `GLIDE_NAME=GlideRuby`, which is only glide-core's fallback.
   DEFAULT_LIB_NAME = "GlideRuby"
 
+  # The RESP protocol specified.
+  attr_reader :protocol
+
   # Resolves the effective `CLIENT SETINFO LIB-NAME` value, composing `base(tag)`.
   # An empty override or tag means "not configured" and is omitted. Character
   # validity is glide-core's (see valkey-io/valkey-glide#6891).
@@ -113,6 +118,8 @@ def initialize(options = {})
       options = url_options.merge(options.except(:url))
     end
 
+    @protocol = options[:protocol]
+
     # Extract connection parameters
     host = options[:host] || "127.0.0.1"
     port = options[:port] || 6379
@@ -318,16 +325,12 @@ def initialize(options = {})
       }
     end
 
-    pubsub_config = Glide::PubSub.parse_config(options[:pubsub], protocol: options[:protocol])
+    pubsub_config = parse_pubsub_configs(options[:pubsub], protocol: options[:protocol])
     json_options.merge!(pubsub_config)
 
-    @pubsub = Glide::PubSub.new(
-      self,
-      cluster_mode: options[:cluster_mode] ? true : false,
-      protocol: options[:protocol]
-    )
-    json_str = json_options.empty? ? nil : JSON.generate(json_options)
+    @pubsub_receiver = Glide::PubSubReceiver.new
 
+    json_str = json_options.empty? ? nil : JSON.generate(json_options)
     # Create client using URI-based FFI function
     client_type = Bindings::ClientType.new
     client_type[:tag] = 1 # SyncClient
@@ -336,7 +339,7 @@ def initialize(options = {})
       uri_str,
       json_str,
       client_type,
-      @pubsub.ffi_handler
+      @pubsub_receiver.ffi_handler
     )
 
     res = Bindings::ConnectionResponse.new(response_ptr)
@@ -380,7 +383,7 @@ def close
 
       # Closed before the native handle goes away, so a thread blocked in
       # get_message wakes with nil instead of hanging on a dead client.
-      @pubsub&.close
+      @pubsub_receiver&.close
       # Fork safety: freeing a handle owned by another process aborts this one.
       # The parent still frees it on its own close.
       return if @pid != Process.pid
@@ -860,4 +863,42 @@ def convert_response(res, &block)
       response
     end
   end
+
+  # Parses and validates the `pubsub:` option into the connection JSON.
+  #
+  # @example pubsub_configs:
+  #   {
+  #     subscriptions: {
+  #       exact:   ["news", "alerts"],  # exact matches
+  #       pattern: ["news.*"],          # glob patterns
+  #       sharded: ["shard-chan"]       # cluster mode
+  #     },
+  #     callback: ->(message, context) { ... },  # callback handler
+  #     context: my_app_state                   # callback context
+  #   }
+  def parse_pubsub_configs(pubsub_configs, protocol: nil)
+    subscriptions = (pubsub_configs || {})[:subscriptions] || {}
+    return {} if subscriptions.empty?
+
+    validate_pubsub_subscriptions!(subscriptions, protocol: protocol)
+
+    { "pubsub_subscriptions" => pubsub_subscriptions_to_ffi(subscriptions) }
+  end
+
+  def validate_pubsub_subscriptions!(subscriptions, protocol:)
+    unknown_modes = subscriptions.keys - SUBSCRIPTION_MODES.keys
+    raise ArgumentError, unknown_pubsub_mode_message(unknown_modes) if unknown_modes.any?
+    raise Resp3RequiredError, protocol unless RESP3_VALUES.include?(protocol)
+  end
+
+  def pubsub_subscriptions_to_ffi(subscriptions)
+    subscriptions
+      .transform_keys { |mode| SUBSCRIPTION_MODES.fetch(mode).to_s }
+      .transform_values { |channels| Array(channels).map(&:to_s) }
+  end
+
+  def unknown_pubsub_mode_message(unknown_modes)
+    "Unknown Pub/Sub subscription mode(s): #{unknown_modes.join(', ')}. " \
+      "Valid modes are: #{SUBSCRIPTION_MODES.keys.join(', ')}"
+  end
 end
diff --git a/lib/valkey/commands/pubsub_commands.rb b/lib/valkey/commands/pubsub_commands.rb
index 7f7a0429..0acd051c 100644
--- a/lib/valkey/commands/pubsub_commands.rb
+++ b/lib/valkey/commands/pubsub_commands.rb
@@ -31,6 +31,12 @@ module Commands
     # @see https://valkey.io/commands/#pubsub
     #
     module PubSubCommands
+      # Subscription mode mapped to the integer key glide-core expects
+      SUBSCRIPTION_MODES = { exact: 0, pattern: 1, sharded: 2 }.freeze
+
+      # PubSub requires RESP3
+      RESP3_VALUES = [:resp3, "resp3", 3].freeze
+
       # Subscribe to exact channels, waiting for the server to confirm the subscription.
       #
       # @example Subscribe to channels
@@ -46,7 +52,15 @@ module PubSubCommands
       #
       # @see https://valkey.io/commands/subscribe/
       def subscribe(*channels, timeout_ms: 0)
-        @pubsub.subscribe(*channels, timeout_ms: timeout_ms)
+        validate_resp3!
+        # glide-core already rejects an empty list with this message, but as
+        # ErrorKind::ClientError, which surfaces here as the too-generic
+        # Valkey::CommandError.
+        # TODO: push this upstream once glide-core reports it as an argument
+        # error, then drop the check here.
+        raise ArgumentError, "No channels provided for subscription" if channels.empty?
+
+        send_command(RequestType::SUBSCRIBE_BLOCKING, channels.map(&:to_s) + [parse_timeout(timeout_ms)])
       end
 
       # Unsubscribe from exact channels, waiting for the server to confirm the change.
@@ -67,7 +81,9 @@ def subscribe(*channels, timeout_ms: 0)
       #
       # @see https://valkey.io/commands/unsubscribe/
       def unsubscribe(*channels, timeout_ms: 0)
-        @pubsub.unsubscribe(*channels, timeout_ms: timeout_ms)
+        validate_resp3!
+
+        send_command(RequestType::UNSUBSCRIBE_BLOCKING, channels.map(&:to_s) + [parse_timeout(timeout_ms)])
       end
 
       # Subscribe to channel patterns, waiting for the server to confirm the subscription.
@@ -84,9 +100,7 @@ def unsubscribe(*channels, timeout_ms: 0)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/psubscribe/
-      def psubscribe(*patterns, timeout_ms: 0)
-        @pubsub.psubscribe(*patterns, timeout_ms: timeout_ms)
-      end
+      def psubscribe(*patterns, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Unsubscribe from channel patterns, waiting for the server to confirm the change.
       #
@@ -105,9 +119,7 @@ def psubscribe(*patterns, timeout_ms: 0)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/punsubscribe/
-      def punsubscribe(*patterns, timeout_ms: 0)
-        @pubsub.punsubscribe(*patterns, timeout_ms: timeout_ms)
-      end
+      def punsubscribe(*patterns, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Subscribe to sharded channels, waiting for the server to confirm the subscription.
       #
@@ -125,9 +137,7 @@ def punsubscribe(*patterns, timeout_ms: 0)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/ssubscribe/
-      def ssubscribe(*channels, timeout_ms: 0)
-        @pubsub.ssubscribe(*channels, timeout_ms: timeout_ms)
-      end
+      def ssubscribe(*channels, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Unsubscribe from sharded channels, waiting for the server to confirm the change.
       #
@@ -148,9 +158,7 @@ def ssubscribe(*channels, timeout_ms: 0)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/sunsubscribe/
-      def sunsubscribe(*channels, timeout_ms: 0)
-        @pubsub.sunsubscribe(*channels, timeout_ms: timeout_ms)
-      end
+      def sunsubscribe(*channels, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Subscribe to exact channels without waiting for the server to confirm.
       #
@@ -166,9 +174,7 @@ def sunsubscribe(*channels, timeout_ms: 0)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/subscribe/
-      def subscribe_lazy(*channels)
-        @pubsub.subscribe_lazy(*channels)
-      end
+      def subscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Unsubscribe from exact channels without waiting for the server to confirm.
       #
@@ -183,9 +189,7 @@ def subscribe_lazy(*channels)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/unsubscribe/
-      def unsubscribe_lazy(*channels)
-        @pubsub.unsubscribe_lazy(*channels)
-      end
+      def unsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Subscribe to channel patterns without waiting for the server to confirm.
       #
@@ -201,9 +205,7 @@ def unsubscribe_lazy(*channels)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/psubscribe/
-      def psubscribe_lazy(*patterns)
-        @pubsub.psubscribe_lazy(*patterns)
-      end
+      def psubscribe_lazy(*patterns) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Unsubscribe from channel patterns without waiting for the server to confirm.
       #
@@ -218,9 +220,7 @@ def psubscribe_lazy(*patterns)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/punsubscribe/
-      def punsubscribe_lazy(*patterns)
-        @pubsub.punsubscribe_lazy(*patterns)
-      end
+      def punsubscribe_lazy(*patterns) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Subscribe to sharded channels without waiting for the server to confirm.
       #
@@ -237,9 +237,7 @@ def punsubscribe_lazy(*patterns)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/ssubscribe/
-      def ssubscribe_lazy(*channels)
-        @pubsub.ssubscribe_lazy(*channels)
-      end
+      def ssubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Unsubscribe from sharded channels without waiting for the server to confirm.
       #
@@ -256,9 +254,7 @@ def ssubscribe_lazy(*channels)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/sunsubscribe/
-      def sunsubscribe_lazy(*channels)
-        @pubsub.sunsubscribe_lazy(*channels)
-      end
+      def sunsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Publish a message on a Pub/Sub channel.
       #
@@ -285,37 +281,9 @@ def sunsubscribe_lazy(*channels)
       # @see https://valkey.io/commands/publish/
       # @see https://valkey.io/commands/spublish/
       def publish(message, channel, sharded: false)
-        @pubsub.publish(message, channel, sharded: sharded)
-      end
+        raise NotImplementedError, "Sharded publish is not implemented yet" if sharded
 
-      # Get the next Pub/Sub message, blocking until one is available.
-      #
-      # @example Consume messages until the client is closed
-      #   while (message = valkey.get_pubsub_message)
-      #     handle(message.channel, message.message)
-      #   end
-      #
-      # @return [Valkey::Glide::PubSub::Message, nil] the message, or `nil` once the client is closed.
-      #   `#pattern` is set only when the push was a `PMESSAGE`
-      # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3
-      def get_pubsub_message
-        @pubsub.get_message
-      end
-
-      # Get the next Pub/Sub message if one is already queued. Never blocks.
-      #
-      # @example Poll for a message
-      #   valkey.try_get_pubsub_message
-      #     # => #<struct Valkey::Glide::PubSub::Message message="hi", channel="channel1", pattern=nil>
-      # @example Poll when nothing is queued
-      #   valkey.try_get_pubsub_message
-      #     # => nil
-      #
-      # @return [Valkey::Glide::PubSub::Message, nil] the message, or `nil` when the queue is empty or the
-      #   client is closed. `#pattern` is set only when the push was a `PMESSAGE`
-      # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3
-      def try_get_pubsub_message
-        @pubsub.try_get_message
+        send_command(RequestType::PUBLISH, [channel.to_s, message.to_s])
       end
 
       # Get this connection's subscription state: what the client asked for and what the server confirmed.
@@ -327,12 +295,10 @@ def try_get_pubsub_message
       #   state.actual_subscriptions
       #     # => {exact: ["channel1"], pattern: [], sharded: ["shard1"]}
       #
-      # @return [Valkey::Glide::PubSub::SubscriptionState] both hashes are keyed `:exact`, `:pattern` and
-      #   `:sharded`, mapping to `Array<String>`; standalone connections omit `:sharded`
+      # @return [Valkey::Glide::PubSubState] both hashes are keyed `:exact`, `:pattern`
+      #   and `:sharded`, mapping to `Array<String>`; standalone connections omit `:sharded`
       # @raise [NotImplementedError] this method is not implemented yet
-      def get_subscriptions
-        @pubsub.get_subscriptions
-      end
+      def get_subscriptions = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # List the currently active channels, that is, the channels with at least one subscriber.
       #
@@ -351,9 +317,7 @@ def get_subscriptions
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/pubsub-channels/
-      def pubsub_channels(pattern = nil)
-        @pubsub.pubsub_channels(pattern)
-      end
+      def pubsub_channels(pattern = nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Get the number of unique patterns that are subscribed to by clients.
       #
@@ -370,9 +334,7 @@ def pubsub_channels(pattern = nil)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/pubsub-numpat/
-      def pubsub_numpat
-        @pubsub.pubsub_numpat
-      end
+      def pubsub_numpat = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Get the number of subscribers for the specified channels, exclusive of clients subscribed to patterns.
       #
@@ -391,9 +353,7 @@ def pubsub_numpat
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/pubsub-numsub/
-      def pubsub_numsub(*channels)
-        @pubsub.pubsub_numsub(*channels)
-      end
+      def pubsub_numsub(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # List the currently active sharded channels, that is, the ones with at least one subscriber.
       #
@@ -414,9 +374,7 @@ def pubsub_numsub(*channels)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/pubsub-shardchannels/
-      def pubsub_shardchannels(pattern = nil)
-        @pubsub.pubsub_shardchannels(pattern)
-      end
+      def pubsub_shardchannels(pattern = nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")
 
       # Get the number of subscribers for the specified sharded channels, exclusive of clients subscribed to
       # patterns.
@@ -436,8 +394,55 @@ def pubsub_shardchannels(pattern = nil)
       # @raise [NotImplementedError] this method is not implemented yet
       #
       # @see https://valkey.io/commands/pubsub-shardnumsub/
-      def pubsub_shardnumsub(*channels)
-        @pubsub.pubsub_shardnumsub(*channels)
+      def pubsub_shardnumsub(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
+
+      # Get the next Pub/Sub message, blocking until one is available.
+      #
+      # @example Consume messages until the client is closed
+      #   while (message = valkey.get_pubsub_message)
+      #     handle(message.channel, message.message)
+      #   end
+      #
+      # @return [Valkey::Glide::PubSubMessage, nil] the message, or `nil` once the client is closed.
+      #   `#pattern` is set only when the push was a `PMESSAGE`
+      # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3
+      def get_pubsub_message
+        validate_resp3!
+        @pubsub_receiver.pop
+      end
+
+      # Get the next Pub/Sub message if one is already queued. Never blocks.
+      #
+      # @example Poll for a message
+      #   valkey.try_get_pubsub_message
+      #     # => #<struct Valkey::Glide::PubSubMessage message="hi", channel="channel1", pattern=nil>
+      # @example Poll when nothing is queued
+      #   valkey.try_get_pubsub_message
+      #     # => nil
+      #
+      # @return [Valkey::Glide::PubSubMessage, nil] the message, or `nil` when the queue is empty or the
+      #   client is closed. `#pattern` is set only when the push was a `PMESSAGE`
+      # @raise [Valkey::Resp3RequiredError] GLIDE Pub/Sub requires RESP3
+      def try_get_pubsub_message
+        validate_resp3!
+        @pubsub_receiver.try_pop
+      end
+
+      private
+
+      def validate_resp3!
+        raise Resp3RequiredError, protocol unless RESP3_VALUES.include?(protocol)
+      end
+
+      # glide-core takes the timeout as the last command argument, in whole
+      # milliseconds, and reads a zero as "no deadline".
+      def parse_timeout(timeout_ms)
+        valid = timeout_ms.is_a?(Numeric) && !timeout_ms.negative?
+        raise ArgumentError, "Timeout must be a non-negative number, got: #{timeout_ms.inspect}" unless valid
+        return "0" if timeout_ms.zero?
+
+        # Handling floats.
+        [timeout_ms.to_i, 1].max.to_s
       end
     end
   end
diff --git a/lib/valkey/glide/pubsub.rb b/lib/valkey/glide/pubsub.rb
deleted file mode 100644
index 546da4c9..00000000
--- a/lib/valkey/glide/pubsub.rb
+++ /dev/null
@@ -1,537 +0,0 @@
-# frozen_string_literal: true
-
-class Valkey
-  # GLIDE-specific internals. Nothing here is public API.
-  module Glide
-    # The wrapper around a Valkey client to handle all its PubSub functionalities.
-    #
-    # Pub/Sub requires the RESP3 protocol, which `parse_config` enforces on the
-    # connect-time `pubsub:` option:
-    #
-    #   pubsub: {
-    #     subscriptions: {
-    #       exact: ["news"],           # channel names
-    #       pattern: ["news.*"],       # channel name patterns
-    #       sharded: ["shard-chan"]    # sharded channels, cluster mode only
-    #     }
-    #   }
-    #
-    # There are three ways to receive messages:
-    #
-    # * Inline: messages are queued and read with {#get_message} or
-    #   {#try_get_message}.
-    # * Callback: a `callback:` proc receives every message instead, along with
-    #   an arbitrary `context:`. Not supported yet.
-    # * Lazy: the `_lazy` subscribe and unsubscribe methods return without
-    #   waiting for the server to confirm the change; read back
-    #   {#get_subscriptions} for the subscriptions the server actually has.
-    #
-    # @see https://valkey.io/docs/topics/pubsub/
-    # @api private
-    class PubSub
-      # Push kinds as delivered by the FFI handler's `kind` argument.
-      # Mirrors `PushKind` in valkey-glide/ffi/src/lib.rs; keep in sync there.
-      module PushKind
-        DISCONNECTION = 0
-        OTHER = 1
-        INVALIDATE = 2
-        MESSAGE = 3
-        PMESSAGE = 4
-        SMESSAGE = 5
-        UNSUBSCRIBE = 6
-        PUNSUBSCRIBE = 7
-        SUNSUBSCRIBE = 8
-        SUBSCRIBE = 9
-        PSUBSCRIBE = 10
-        SSUBSCRIBE = 11
-
-        # The only kinds that carry a payload for the user.
-        MESSAGE_KINDS = [MESSAGE, PMESSAGE, SMESSAGE].freeze
-      end
-
-      # One delivered push: the incoming `message`, the `channel` that carried
-      # it, and the `pattern` that matched it. `pattern` is set only for
-      # PMESSAGE; exact and sharded pushes leave it nil.
-      Message = Struct.new(:message, :channel, :pattern)
-
-      # A connection's subscriptions, as returned by {#get_subscriptions}. Both
-      # fields are a `Hash` keyed by `:exact`, `:pattern` and `:sharded`, each
-      # mapping to an `Array<String>`. `desired_subscriptions` is what the
-      # client asked for, `actual_subscriptions` is what the server currently
-      # has. Standalone connections omit `:sharded` entirely.
-      SubscriptionState = Struct.new(:desired_subscriptions, :actual_subscriptions)
-
-      # Subscription mode to the integer key glide-core expects in the
-      # connection JSON: `:exact` uses exact channel names, `:pattern` uses
-      # glob-style channel name patterns, `:sharded` uses sharded Pub/Sub and is
-      # cluster-only.
-      SUBSCRIPTION_MODES = { exact: 0, pattern: 1, sharded: 2 }.freeze
-
-      # PubSub requires RESP3
-      RESP3_VALUES = [:resp3, "resp3", 3].freeze
-
-      class << self
-        # Parses and validates the `pubsub:` option.
-        #
-        # @example pubsub_configs:
-        #   {
-        #     subscriptions: {
-        #       exact:   ["news", "alerts"],  # exact matches
-        #       pattern: ["news.*"],          # glob patterns
-        #       sharded: ["shard-chan"]       # cluster mode
-        #     },
-        #     callback: ->(message, context) { ... },  # callback handler
-        #     context: my_app_state                   # callback context
-        #   }
-        #
-        # @param pubsub_configs [Hash, nil] the `pubsub:` options: the channels
-        #   and patterns to subscribe to at connection time under
-        #   `:subscriptions`, keyed by mode, an optional `:callback` to accept
-        #   the incoming messages and an arbitrary `:context` passed to it.
-        # @param protocol [Symbol, String, Integer, nil] the `protocol:` option.
-        #   Subscriptions require RESP3.
-        # @return [Hash] the subscriptions keyed by the integer mode glide-core
-        #   expects, or an empty Hash when nothing is subscribed to.
-        # @raise [ArgumentError] on an unknown subscription mode.
-        # @raise [Valkey::Resp3RequiredError] on a protocol other than RESP3.
-        # @see https://valkey.io/docs/topics/pubsub/
-        def parse_config(pubsub_configs, protocol: nil)
-          pubsub_configs ||= {}
-          subscriptions = pubsub_configs[:subscriptions] || {}
-
-          validate!(subscriptions, protocol: protocol)
-
-          to_ffi(pubsub_configs)
-        end
-
-        private
-
-        # Converts the configuration to what the ffi expects.
-        def to_ffi(pubsub_configs)
-          subscriptions = pubsub_configs[:subscriptions] || {}
-          return {} if subscriptions.empty?
-
-          mapped = subscriptions
-                   .transform_keys { |mode| SUBSCRIPTION_MODES.fetch(mode).to_s }
-                   .transform_values { |channels| Array(channels).map(&:to_s) }
-
-          { "pubsub_subscriptions" => mapped }
-        end
-
-        def validate!(subscriptions, protocol:)
-          return if subscriptions.empty?
-
-          unknown_modes = subscriptions.keys - SUBSCRIPTION_MODES.keys
-          raise ArgumentError, unknown_mode_message(unknown_modes) if unknown_modes.any?
-
-          return if RESP3_VALUES.include?(protocol)
-
-          raise Resp3RequiredError, protocol
-        end
-
-        def unknown_mode_message(unknown_modes)
-          "Unknown Pub/Sub subscription mode(s): #{unknown_modes.join(', ')}. " \
-            "Valid modes are: #{SUBSCRIPTION_MODES.keys.join(', ')}"
-        end
-      end
-
-      # @param client [Valkey] A valkey client connection.
-      # @param cluster_mode [Boolean] The client cluster mode.
-      # @param protocol [Symbol, String, Integer, nil] the client's protocol.
-      def initialize(client, cluster_mode:, protocol: nil)
-        @client = client
-        @cluster_mode = cluster_mode
-        @protocol = protocol
-        @message_queue = Thread::Queue.new
-
-        # The handler proc for receiving messages from the FFI.
-        @ffi_handler = build_ffi_handler
-      end
-
-      attr_reader :ffi_handler
-
-      # Gets a Pub/Sub message, blocking until one is available.
-      #
-      # @return [Message, nil] the next message, or nil once the message queue
-      #   is closed.
-      # @raise [Valkey::Resp3RequiredError]
-      def get_message
-        validate_resp3!
-        @message_queue.pop
-      end
-
-      # Gets a Pub/Sub message without blocking.
-      #
-      # @return [Message, nil] the next message, or nil when the message queue
-      #   is empty or closed.
-      # @raise [Valkey::Resp3RequiredError]
-      def try_get_message
-        validate_resp3!
-        @message_queue.pop(true)
-      rescue ThreadError
-        nil
-      end
-
-      # Releases the Pub/Sub resources, closing the message queue. Callers
-      # blocked in {#get_message} are woken with nil.
-      def close
-        @message_queue.close
-      end
-
-      # Subscribes to exact channels (blocking). Updates the client's desired
-      # subscription state and waits for the server's confirmation.
-      #
-      # @param channels [Array<String>] the channel names to subscribe to. An
-      #   empty list is rejected with "No channels provided for subscription".
-      # @param timeout_ms [Integer, Float] maximum time in milliseconds to wait
-      #   for the server to confirm. `0` blocks indefinitely.
-      # @return [void] once the server has confirmed the subscription.
-      # @raise [ArgumentError] on a negative timeout_ms or an empty channel list.
-      # @raise [Valkey::Resp3RequiredError]
-      # @raise [Valkey::TimeoutError] when the timeout expires before the
-      #   server confirms.
-      # @see https://valkey.io/commands/subscribe/
-      def subscribe(*channels, timeout_ms: 0)
-        validate_resp3!
-        # glide-core already rejects an empty list with this message, but as
-        # ErrorKind::ClientError, which surfaces here as the too-generic
-        # Valkey::CommandError.
-        # TODO: push this upstream once glide-core reports it as an argument
-        # error, then drop the check here.
-        raise ArgumentError, "No channels provided for subscription" if channels.empty?
-
-        @client.send_command(RequestType::SUBSCRIBE_BLOCKING, channels.map(&:to_s) + [timeout_argument(timeout_ms)])
-      end
-
-      # Unsubscribes from exact channels (blocking). Updates the client's
-      # desired subscription state and waits for the server's confirmation.
-      #
-      # @param channels [Array<String>] the channel names to unsubscribe from.
-      #   Empty unsubscribes from every exact channel.
-      # @param timeout_ms [Integer, Float] maximum time in milliseconds to wait
-      #   for the server to confirm. `0` blocks indefinitely.
-      # @return [void] once the server has confirmed the unsubscription.
-      # @raise [ArgumentError] on a negative timeout_ms.
-      # @raise [Valkey::Resp3RequiredError]  GLIDE Pub/Sub requires RESP3
-      # @raise [Valkey::TimeoutError] when the timeout expires before the
-      #   server confirms.
-      # @see https://valkey.io/commands/unsubscribe/
-      def unsubscribe(*channels, timeout_ms: 0)
-        validate_resp3!
-
-        @client.send_command(RequestType::UNSUBSCRIBE_BLOCKING, channels.map(&:to_s) + [timeout_argument(timeout_ms)])
-      end
-
-      # Subscribes to channel patterns (blocking). Updates the client's desired
-      # subscription state and waits for the server's confirmation.
-      #
-      # @param patterns [Array<String>] the glob-style patterns to subscribe to,
-      #   for example `"news.*"`. An empty list is rejected with "No channels
-      #   provided for subscription".
-      # @param timeout_ms [Integer, Float] maximum time in milliseconds to wait
-      #   for the server to confirm. `0` blocks indefinitely.
-      # @return [void] once the server has confirmed the subscription.
-      # @raise [ArgumentError] on a negative timeout_ms.
-      # @raise [Valkey::TimeoutError] when the timeout expires before the
-      #   server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/psubscribe/
-      def psubscribe(*patterns, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Unsubscribes from channel patterns (blocking). Updates the client's
-      # desired subscription state and waits for the server's confirmation.
-      #
-      # @param patterns [Array<String>] the patterns to unsubscribe from. Empty
-      #   unsubscribes from every pattern.
-      # @param timeout_ms [Integer, Float] maximum time in milliseconds to wait
-      #   for the server to confirm. `0` blocks indefinitely.
-      # @return [void] once the server has confirmed the unsubscription.
-      # @raise [ArgumentError] on a negative timeout_ms.
-      # @raise [Valkey::TimeoutError] when the timeout expires before the
-      #   server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/punsubscribe/
-      def punsubscribe(*patterns, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Subscribes to sharded channels (blocking). Updates the client's desired
-      # subscription state and waits for the server's confirmation. Requires
-      # `cluster_mode: true`; sharded Pub/Sub has no standalone equivalent.
-      #
-      # Since: Valkey version 7.0.0.
-      #
-      # @param channels [Array<String>] the sharded channel names to subscribe
-      #   to. An empty list is rejected with "No channels provided for
-      #   subscription".
-      # @param timeout_ms [Integer, Float] maximum time in milliseconds to wait
-      #   for the server to confirm. `0` blocks indefinitely.
-      # @return [void] once the server has confirmed the subscription.
-      # @raise [ArgumentError] on a negative timeout_ms.
-      # @raise [Valkey::TimeoutError] when the timeout expires before the
-      #   server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/ssubscribe/
-      def ssubscribe(*channels, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Unsubscribes from sharded channels (blocking). Updates the client's
-      # desired subscription state and waits for the server's confirmation.
-      # Requires `cluster_mode: true`; sharded Pub/Sub has no standalone
-      # equivalent.
-      #
-      # Since: Valkey version 7.0.0.
-      #
-      # @param channels [Array<String>] the sharded channel names to unsubscribe
-      #   from. Empty unsubscribes from every sharded channel.
-      # @param timeout_ms [Integer, Float] maximum time in milliseconds to wait
-      #   for the server to confirm. `0` blocks indefinitely.
-      # @return [void] once the server has confirmed the unsubscription.
-      # @raise [ArgumentError] on a negative timeout_ms.
-      # @raise [Valkey::TimeoutError] when the timeout expires before the
-      #   server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/sunsubscribe/
-      def sunsubscribe(*channels, timeout_ms: 0) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Subscribes to exact channels (non-blocking). Updates the client's
-      # desired subscription state without waiting for the server's
-      # confirmation, and returns as soon as the local state is updated. The
-      # client subscribes asynchronously in the background. Use
-      # {#get_subscriptions} to verify the actual server-side subscription
-      # state.
-      #
-      # @param channels [Array<String>] the channel names to subscribe to. An
-      #   empty list is rejected with "No channels provided for subscription".
-      # @return [void] immediately, before the server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/subscribe/
-      def subscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Unsubscribes from exact channels (non-blocking). Updates the client's
-      # desired subscription state without waiting for the server's
-      # confirmation, and returns as soon as the local state is updated. Use
-      # {#get_subscriptions} to verify the actual server-side subscription
-      # state.
-      #
-      # @param channels [Array<String>] the channel names to unsubscribe from.
-      #   Empty unsubscribes from every exact channel.
-      # @return [void] immediately, before the server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/unsubscribe/
-      def unsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Subscribes to channel patterns (non-blocking). Updates the client's
-      # desired subscription state without waiting for the server's
-      # confirmation, and returns as soon as the local state is updated. The
-      # client subscribes asynchronously in the background. Use
-      # {#get_subscriptions} to verify the actual server-side subscription
-      # state.
-      #
-      # @param patterns [Array<String>] the glob-style patterns to subscribe to,
-      #   for example `"news.*"`. An empty list is rejected with "No channels
-      #   provided for subscription".
-      # @return [void] immediately, before the server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/psubscribe/
-      def psubscribe_lazy(*patterns) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Unsubscribes from channel patterns (non-blocking). Updates the client's
-      # desired subscription state without waiting for the server's
-      # confirmation, and returns as soon as the local state is updated. Use
-      # {#get_subscriptions} to verify the actual server-side subscription
-      # state.
-      #
-      # @param patterns [Array<String>] the patterns to unsubscribe from. Empty
-      #   unsubscribes from every pattern.
-      # @return [void] immediately, before the server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/punsubscribe/
-      def punsubscribe_lazy(*patterns) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Subscribes to sharded channels (non-blocking). Updates the client's
-      # desired subscription state without waiting for the server's
-      # confirmation, and returns as soon as the local state is updated. The
-      # client subscribes asynchronously in the background. Use
-      # {#get_subscriptions} to verify the actual server-side subscription
-      # state. Requires `cluster_mode: true`; sharded Pub/Sub has no standalone
-      # equivalent.
-      #
-      # Since: Valkey version 7.0.0.
-      #
-      # @param channels [Array<String>] the sharded channel names to subscribe
-      #   to. An empty list is rejected with "No channels provided for
-      #   subscription".
-      # @return [void] immediately, before the server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/ssubscribe/
-      def ssubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Unsubscribes from sharded channels (non-blocking). Updates the client's
-      # desired subscription state without waiting for the server's
-      # confirmation, and returns as soon as the local state is updated. Use
-      # {#get_subscriptions} to verify the actual server-side subscription
-      # state. Requires `cluster_mode: true`; sharded Pub/Sub has no standalone
-      # equivalent.
-      #
-      # Since: Valkey version 7.0.0.
-      #
-      # @param channels [Array<String>] the sharded channel names to unsubscribe
-      #   from. Empty unsubscribes from every sharded channel.
-      # @return [void] immediately, before the server confirms.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/sunsubscribe/
-      def sunsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Publishes a message on a Pub/Sub channel. Aggregates the PUBLISH and
-      # SPUBLISH functionalities, the mode selected by `sharded`. In both modes
-      # the request is routed using the hashed channel as key. The message comes
-      # first, matching the other GLIDE clients.
-      #
-      # @param message [String] the message to publish.
-      # @param channel [String] the channel to publish the message on.
-      # @param sharded [Boolean] use sharded Pub/Sub mode. Available since
-      #   Valkey version 7.0, and requires `cluster_mode: true`.
-      # @return [Integer] the number of subscriptions that received the message.
-      #   In cluster mode that is the count on the node the request was routed
-      #   to; in standalone it is the count on the primary node, which does not
-      #   include subscriptions configured on replicas.
-      # @raise [NotImplementedError] when `sharded` is true; sharded publish is
-      #   not implemented yet.
-      # @see https://valkey.io/commands/publish/
-      # @see https://valkey.io/commands/spublish/
-      def publish(message, channel, sharded: false)
-        raise NotImplementedError, "Sharded publish is not implemented yet" if sharded
-
-        @client.send_command(RequestType::PUBLISH, [channel.to_s, message.to_s])
-      end
-
-      # The connection's subscription state: what the client asked for, and what
-      # the server currently has. The way to confirm the outcome of a lazy
-      # subscribe or unsubscribe call.
-      #
-      # @return [SubscriptionState] the desired and actual subscriptions.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      def get_subscriptions = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Lists the currently active channels, that is the channels with at least
-      # one exact subscriber. The command is routed to all nodes and the
-      # responses are aggregated into a single array.
-      #
-      # @param pattern [String, nil] a glob-style pattern to match active
-      #   channels against. `nil` returns all active channels.
-      # @return [Array<String>] the active channels matching the given pattern.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/pubsub-channels/
-      def pubsub_channels(pattern = nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Counts the unique patterns clients are subscribed to. That is the total
-      # number of unique patterns across all clients, not the number of clients
-      # subscribed to patterns. The command is routed to all nodes and the
-      # responses are aggregated into their sum.
-      #
-      # @return [Integer] the number of unique patterns.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/pubsub-numpat/
-      def pubsub_numpat = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Counts the subscribers of the given channels, exclusive of clients
-      # subscribed to patterns. The command is routed to all nodes and the
-      # responses are aggregated into a single Hash.
-      #
-      # @param channels [Array<String>] the channels to query for the number of
-      #   subscribers. Calling this without channels is valid and returns an
-      #   empty Hash.
-      # @return [Hash{String => Integer}] the channel names mapped to their
-      #   number of subscribers.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/pubsub-numsub/
-      def pubsub_numsub(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Lists the currently active sharded channels, that is the sharded
-      # channels with at least one subscriber. The command is routed to all nodes
-      # and the responses are aggregated into a single array.
-      #
-      # Since: Valkey version 7.0.0.
-      #
-      # @param pattern [String, nil] a glob-style pattern to match active sharded
-      #   channels against. `nil` returns all active sharded channels.
-      # @return [Array<String>] the active sharded channels matching the given
-      #   pattern.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/pubsub-shardchannels/
-      def pubsub_shardchannels(pattern = nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Counts the subscribers of the given sharded channels, exclusive of
-      # clients subscribed to patterns. The command is routed to all nodes and
-      # the responses are aggregated into a single Hash.
-      #
-      # Since: Valkey version 7.0.0.
-      #
-      # @param channels [Array<String>] the sharded channels to query for the
-      #   number of subscribers. Calling this without channels is valid and
-      #   returns an empty Hash.
-      # @return [Hash{String => Integer}] the sharded channel names mapped to
-      #   their number of subscribers.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/pubsub-shardnumsub/
-      def pubsub_shardnumsub(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      # Raw PUBSUB passthrough, dispatching to the matching `pubsub_*` method.
-      #
-      # @param subcommand [String, Symbol] one of `channels`, `numpat`,
-      #   `numsub`, `shardchannels` or `shardnumsub`.
-      # @param args [Array] the arguments of the subcommand.
-      # @return [Array<String>, Integer, Hash{String => Integer}] whatever the
-      #   target method returns.
-      # @raise [NotImplementedError] this method is not implemented yet.
-      # @see https://valkey.io/commands/#pubsub
-      def pubsub(subcommand, *args) = raise(NotImplementedError, "#{__method__} is not implemented yet")
-
-      private
-
-      # Builds the proc handed to the FFI.
-      #
-      # Runs on a Rust thread the Ruby runtime did not create, on a single push
-      # worker, with the GVL borrowed. Keep it thin: copy out, enqueue, return.
-      # No user code, no FFI re-entry, no blocking I/O -- anything slow here
-      # stalls every message behind it. The pointers are freed when it returns,
-      # so the copy has to happen synchronously.
-      #
-      # The reads are length-driven so a payload with an embedded NUL survives.
-      def build_ffi_handler
-        lambda do |_client_ptr, kind, message_ptr, message_size, channel_ptr, channel_size, pattern_ptr, pattern_size|
-          next unless PushKind::MESSAGE_KINDS.include?(kind)
-
-          pattern = pattern_ptr.null? ? nil : pattern_ptr.read_string(pattern_size)
-          message = message_ptr.read_string(message_size)
-          deliver(Message.new(message, channel_ptr.read_string(channel_size), pattern))
-        rescue StandardError
-          # TODO: Log the swallowed error once a logger binding exists.
-          nil
-        end
-      end
-
-      def validate_resp3!
-        raise Resp3RequiredError, @protocol unless RESP3_VALUES.include?(@protocol)
-      end
-
-      # glide-core takes the timeout as the last command argument, in whole
-      # milliseconds, and reads a zero as "no deadline".
-      def timeout_argument(timeout_ms)
-        valid = timeout_ms.is_a?(Numeric) && !timeout_ms.negative?
-        raise ArgumentError, "Timeout must be a non-negative number, got: #{timeout_ms.inspect}" unless valid
-        return "0" if timeout_ms.zero?
-
-        # Handling floats.
-        [timeout_ms.to_i, 1].max.to_s
-      end
-
-      # Single delivery point, so push mode is added by branching here and
-      # nothing else changes.
-      #
-      # TODO: unfinished -- callback branch:
-      #   @callback.arity == 1 ? call(msg) : call(msg, @context).
-      def deliver(message)
-        @message_queue.push(message)
-      end
-    end
-  end
-end
diff --git a/lib/valkey/glide/pubsub_message.rb b/lib/valkey/glide/pubsub_message.rb
new file mode 100644
index 00000000..447caa92
--- /dev/null
+++ b/lib/valkey/glide/pubsub_message.rb
@@ -0,0 +1,23 @@
+# frozen_string_literal: true
+
+class Valkey
+  # GLIDE-specific public types, returned by APIs that have no redis-rb
+  # equivalent. Internal members carry `@api private`.
+  module Glide
+    # A message delivered on a subscribed channel.
+    #
+    # @!attribute [rw] message
+    #   @return [String] the published payload. Binary safe: embedded NUL bytes
+    #     are preserved.
+    # @!attribute [rw] channel
+    #   @return [String] the channel the message was published to. For a
+    #     pattern subscription this is the concrete channel that matched, not
+    #     the pattern.
+    # @!attribute [rw] pattern
+    #   @return [String, nil] the pattern the subscription matched on, set only
+    #     when the push was a `PMESSAGE`; `nil` for exact and sharded pushes.
+    #
+    # @see https://valkey.io/docs/topics/pubsub/
+    PubSubMessage = Struct.new(:message, :channel, :pattern)
+  end
+end
diff --git a/lib/valkey/glide/pubsub_receiver.rb b/lib/valkey/glide/pubsub_receiver.rb
new file mode 100644
index 00000000..88f2ec67
--- /dev/null
+++ b/lib/valkey/glide/pubsub_receiver.rb
@@ -0,0 +1,88 @@
+# frozen_string_literal: true
+
+class Valkey
+  module Glide
+    # @api private
+    class PubSubReceiver
+      # Push kinds as delivered by the FFI PubSub handler's `kind` argument.
+      # Mirrors `PushKind` in valkey-glide/ffi/src/lib.rs; keep in sync there.
+      #
+      # @api private
+      module PushKind
+        DISCONNECTION = 0
+        OTHER = 1
+        INVALIDATE = 2
+        MESSAGE = 3
+        PMESSAGE = 4
+        SMESSAGE = 5
+        UNSUBSCRIBE = 6
+        PUNSUBSCRIBE = 7
+        SUNSUBSCRIBE = 8
+        SUBSCRIBE = 9
+        PSUBSCRIBE = 10
+        SSUBSCRIBE = 11
+
+        # The only kinds that carry a payload for the user.
+        MESSAGE_KINDS = [MESSAGE, PMESSAGE, SMESSAGE].freeze
+      end
+
+      def initialize
+        @message_queue = Thread::Queue.new
+
+        @ffi_handler = build_ffi_handler
+      end
+
+      attr_reader :ffi_handler
+
+      def pop
+        @message_queue.pop
+      end
+
+      def try_pop
+        @message_queue.pop(true)
+      rescue ThreadError
+        # TODO: Log debug here
+        nil
+      end
+
+      def close
+        @message_queue.close
+      end
+
+      private
+
+      # Builds the proc handed to the FFI.
+      #
+      # Runs on a Rust thread the Ruby runtime did not create, on a single push
+      # worker, with the GVL borrowed. Keep it thin: copy out, enqueue, return.
+      # No user code, no FFI re-entry, no blocking I/O -- anything slow here
+      # stalls every message behind it. The pointers are freed when it returns,
+      # so the copy has to happen synchronously.
+      #
+      # The reads are length-driven so a payload with an embedded NUL survives.
+      def build_ffi_handler
+        lambda do |_client_ptr, kind, message_ptr, message_size, channel_ptr, channel_size, pattern_ptr, pattern_size|
+          next unless PushKind::MESSAGE_KINDS.include?(kind)
+
+          pattern = pattern_ptr.null? ? nil : pattern_ptr.read_string(pattern_size)
+          message = message_ptr.read_string(message_size)
+          deliver(
+            PubSubMessage.new(message, channel_ptr.read_string(channel_size), pattern)
+          )
+        rescue StandardError
+          # TODO: Log the swallowed error once a logger binding exists.
+          nil
+        end
+      end
+
+      # Single delivery point, so push mode is added by branching here and
+      # nothing else changes.
+      #
+      # TODO: unfinished -- callback branch:
+      #   @callback.arity == 1 ? call(msg) : call(msg, @context).
+      def deliver(message)
+        @message_queue.push(message)
+      end
+    end
+  end
+end
diff --git a/lib/valkey/glide/pubsub_state.rb b/lib/valkey/glide/pubsub_state.rb
new file mode 100644
index 00000000..af6f9d9d
--- /dev/null
+++ b/lib/valkey/glide/pubsub_state.rb
@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+class Valkey
+  module Glide
+    # A connection's subscriptions
+    #
+    # @!attribute [rw] desired_subscriptions
+    #   @return [Hash{Symbol => Array<String>}] the subscriptions the client
+    #     asked for, keyed `:exact`, `:pattern` and `:sharded`. Standalone
+    #     connections omit `:sharded`.
+    # @!attribute [rw] actual_subscriptions
+    #   @return [Hash{Symbol => Array<String>}] the subscriptions the server has
+    #     confirmed, keyed the same way.
+    #
+    # @see https://valkey.io/docs/topics/pubsub/
+    PubSubState = Struct.new(:desired_subscriptions, :actual_subscriptions)
+  end
+end
diff --git a/lib/valkey/pipeline.rb b/lib/valkey/pipeline.rb
index 6590396b..26692d9a 100644
--- a/lib/valkey/pipeline.rb
+++ b/lib/valkey/pipeline.rb
@@ -185,5 +185,22 @@ def call_v(argv)
       super
     end
     # rubocop:enable Lint/UselessMethodDefinition
+
+    # Subscriptions outlive a batch and pushes arrive out of band, so neither
+    # can be expressed as one queued reply.
+    PUBSUB_UNSUPPORTED = %i[
+      subscribe unsubscribe psubscribe punsubscribe ssubscribe sunsubscribe
+      subscribe_lazy unsubscribe_lazy psubscribe_lazy punsubscribe_lazy
+      ssubscribe_lazy sunsubscribe_lazy
+      get_subscriptions get_pubsub_message try_get_pubsub_message
+      pubsub_channels pubsub_numpat pubsub_numsub
+      pubsub_shardchannels pubsub_shardnumsub
+    ].freeze
+
+    PUBSUB_UNSUPPORTED.each do |name|
+      define_method(name) do |*, **|
+        raise ArgumentError, "#{name} is not supported inside pipelined/multi"
+      end
+    end
   end
 end
diff --git a/test/unit/glide/pubsub_receiver_test.rb b/test/unit/glide/pubsub_receiver_test.rb
new file mode 100644
index 00000000..146abbec
--- /dev/null
+++ b/test/unit/glide/pubsub_receiver_test.rb
@@ -0,0 +1,84 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class TestPubSubReceiverUnit < Minitest::Test
+  Kind = Valkey::Glide::PubSubReceiver::PushKind
+
+  def setup
+    @receiver = Valkey::Glide::PubSubReceiver.new
+  end
+
+  def teardown
+    @receiver&.close
+  end
+
+  def test_ffi_handler_queues_messages
+    push(Kind::MESSAGE, message: "exact", channel: "news")
+    push(Kind::PMESSAGE, message: "pattern", channel: "news.tech", pattern: "news.*")
+    push(Kind::SMESSAGE, message: "sharded", channel: "shard-chan")
+
+    expected = [
+      ["exact", "news", nil],
+      ["pattern", "news.tech", "news.*"],
+      ["sharded", "shard-chan", nil]
+    ]
+
+    received = 3.times.map { @receiver.try_pop.to_a }
+
+    assert_equal expected, received
+  end
+
+  def test_ffi_handler_drops_non_message_kinds
+    non_message_kinds = [
+      Kind::DISCONNECTION, Kind::OTHER, Kind::INVALIDATE,
+      Kind::SUBSCRIBE, Kind::PSUBSCRIBE, Kind::SSUBSCRIBE,
+      Kind::UNSUBSCRIBE, Kind::PUNSUBSCRIBE, Kind::SUNSUBSCRIBE
+    ]
+
+    non_message_kinds.each do |kind|
+      push(kind, message: "phantom", channel: "news")
+
+      assert_nil @receiver.try_pop, "kind #{kind} must not queue a message"
+    end
+  end
+
+  def test_ffi_handler_with_embedded_nul
+    payload = "before\0after"
+
+    push(Kind::MESSAGE, message: payload, channel: "news")
+
+    assert_equal payload, @receiver.try_pop.message
+  end
+
+  # This is a sanity check ensuring that FFI handler is not recreated, otherwise it won't work across the FFI.
+  def test_ffi_handler_is_retained
+    assert_same @receiver.ffi_handler, @receiver.ffi_handler
+  end
+
+  private
+
+  # Calls the retained FFI handler the way the Rust push worker does, with real
+  # buffers and the byte lengths alongside them.
+  def push(kind, message: nil, channel: nil, pattern: nil)
+    message_pointer, message_length = buffer_for(message)
+    channel_pointer, channel_length = buffer_for(channel)
+    pattern_pointer, pattern_length = buffer_for(pattern)
+
+    @receiver.ffi_handler.call(
+      0, kind,
+      message_pointer, message_length,
+      channel_pointer, channel_length,
+      pattern_pointer, pattern_length
+    )
+  end
+
+  def buffer_for(value)
+    return [FFI::Pointer::NULL, 0] if value.nil?
+
+    bytes = value.b
+    buffer = FFI::MemoryPointer.new(:char, bytes.bytesize)
+    buffer.put_bytes(0, bytes)
+    [buffer, bytes.bytesize]
+  end
+end
diff --git a/test/unit/glide/pubsub_test.rb b/test/unit/glide/pubsub_test.rb
index 4886760c..28824603 100644
--- a/test/unit/glide/pubsub_test.rb
+++ b/test/unit/glide/pubsub_test.rb
@@ -3,22 +3,26 @@
 require "test_helper"
 require "timeout"
 
-# Unit tests for Valkey::Glide::PubSub.
+# Unit tests for Valkey::Commands::PubSubCommands.
 # TODO: https://github.com/valkey-io/valkey-glide-ruby/issues/135.
-class TestGlidePubSubUnit < Minitest::Test
-  Kind = Valkey::Glide::PubSub::PushKind
+class TestPubSubCommandsUnit < Minitest::Test
+  Kind = Valkey::Glide::PubSubReceiver::PushKind
 
-  # Stands in for the Valkey client so dispatch can be asserted without a
-  # server: records what would have gone over the wire and replies with a
-  # canned value.
-  class RecordingClient
+  # A real client with the connection left out, so the mixin's own methods run
+  # unmodified while dispatch is asserted without a server: records what would
+  # have gone over the wire and replies with a canned value.
+  class RecordingClient < Valkey
     SentCommand = Struct.new(:request_type, :args)
 
     attr_reader :sent_commands
 
-    def initialize(response: nil)
+    def initialize(response: nil, protocol: :resp3) # rubocop:disable Lint/MissingSuper
       @sent_commands = []
       @response = response
+      @protocol = protocol
+      @pubsub_receiver = Valkey::Glide::PubSubReceiver.new
+      @close_lock = Mutex.new
+      @pid = Process.pid
     end
 
     def send_command(request_type, args = [], **_options)
@@ -34,7 +38,7 @@ def last_command
   # Pub/Sub requires RESP3, so the shared instance declares it; the protocol
   # guard itself is exercised with purpose-built instances further down.
   def setup
-    @pubsub = Valkey::Glide::PubSub.new(nil, cluster_mode: false, protocol: :resp3)
+    @pubsub = RecordingClient.new
   end
 
   def teardown
@@ -45,7 +49,8 @@ def teardown
   # before the handler exists. `deliver` is private because nothing outside the
   # handler may call it in production.
   def deliver(message:, channel:, pattern: nil)
-    @pubsub.send(:deliver, Valkey::Glide::PubSub::Message.new(message, channel, pattern))
+    receiver_for(@pubsub).send(:deliver,
+                               Valkey::Glide::PubSubMessage.new(message, channel, pattern))
   end
 
   def test_message_kinds_cover_only_payload_carrying_pushes
@@ -55,33 +60,27 @@ def test_message_kinds_cover_only_payload_carrying_pushes
   end
 
   def test_message_carries_message_channel_and_pattern
-    msg = Valkey::Glide::PubSub::Message.new("hello", "news.tech", "news.*")
+    msg = Valkey::Glide::PubSubMessage.new("hello", "news.tech", "news.*")
 
     assert_equal "hello", msg.message
     assert_equal "news.tech", msg.channel
     assert_equal "news.*", msg.pattern
   end
 
-  # Rust holds the function pointer for the client's lifetime, so the proc has
-  # to be the same retained object every time, not a fresh one per call.
-  def test_ffi_handler_is_retained
-    assert_same @pubsub.ffi_handler, @pubsub.ffi_handler
+  def test_try_get_pubsub_message_returns_nil_when_empty
+    assert_nil @pubsub.try_get_pubsub_message
   end
 
-  def test_try_get_message_returns_nil_when_empty
-    assert_nil @pubsub.try_get_message
-  end
-
-  def test_get_message_returns_messages_in_delivery_order
+  def test_get_pubsub_message_returns_messages_in_delivery_order
     3.times { |i| deliver(message: "m#{i}", channel: "news") }
 
-    received = 3.times.map { @pubsub.get_message.message }
+    received = 3.times.map { @pubsub.get_pubsub_message.message }
 
     assert_equal %w[m0 m1 m2], received
   end
 
   def test_close_wakes_a_blocked_reader_with_nil
-    reader = Thread.new { @pubsub.get_message }
+    reader = Thread.new { @pubsub.get_pubsub_message }
     # Let the reader reach the blocking pop before the queue closes.
     sleep 0.05
     @pubsub.close
@@ -89,31 +88,31 @@ def test_close_wakes_a_blocked_reader_with_nil
     assert_nil Timeout.timeout(2) { reader.value }
   end
 
-  def test_try_get_message_returns_nil_once_closed
+  def test_try_get_pubsub_message_returns_nil_once_closed
     @pubsub.close
 
-    assert_nil @pubsub.try_get_message
+    assert_nil @pubsub.try_get_pubsub_message
   end
 
   # --- Pub/Sub connection options -----------------------------------------
 
   def test_pubsub_explicit_resp2_raises
     error = assert_raises(Valkey::Resp3RequiredError) do
-      Valkey::Glide::PubSub.parse_config({ subscriptions: { exact: ["news"] } }, protocol: :resp2)
+      parse_pubsub_configs.call({ subscriptions: { exact: ["news"] } }, protocol: :resp2)
     end
     assert_match(/RESP3/, error.message)
   end
 
   def test_pubsub_omitted_protocol_raises
     error = assert_raises(Valkey::Resp3RequiredError) do
-      Valkey::Glide::PubSub.parse_config({ subscriptions: { exact: ["news"] } })
+      parse_pubsub_configs.call({ subscriptions: { exact: ["news"] } })
     end
     assert_match(/RESP3/, error.message)
   end
 
   def test_pubsub_explicit_nil_protocol_raises
     error = assert_raises(Valkey::Resp3RequiredError) do
-      Valkey::Glide::PubSub.parse_config({ subscriptions: { exact: ["news"] } }, protocol: nil)
+      parse_pubsub_configs.call({ subscriptions: { exact: ["news"] } }, protocol: nil)
     end
     assert_match(/RESP3/, error.message)
   end
@@ -123,7 +122,7 @@ def test_pubsub_parse_config_ok
       subscriptions: { exact: ["news", :symbols], pattern: ["news.*"], sharded: ["news.shard"] }
     }
 
-    parsed = Valkey::Glide::PubSub.parse_config(pubsub_config, protocol: :resp3)
+    parsed = parse_pubsub_configs.call(pubsub_config, protocol: :resp3)
 
     expected = { "pubsub_subscriptions" => { "0" => %w[news symbols], "1" => ["news.*"], "2" => ["news.shard"] } }
 
@@ -131,67 +130,26 @@ def test_pubsub_parse_config_ok
   end
 
   def test_pubsub_parse_config_nil
-    config = Valkey::Glide::PubSub.parse_config(nil)
-    expected = Valkey::Glide::PubSub.parse_config({})
+    config = parse_pubsub_configs.call(nil)
+    expected = parse_pubsub_configs.call({})
     assert_equal config, expected
   end
 
   def test_pubsub_unknown_subscription_mode_raises
     error = assert_raises(ArgumentError) do
-      Valkey::Glide::PubSub.parse_config({ subscriptions: { unknown_mode: ["news.*"] } })
+      parse_pubsub_configs.call({ subscriptions: { unknown_mode: ["news.*"] } })
     end
 
     assert_equal "Unknown Pub/Sub subscription mode(s): unknown_mode. Valid modes are: exact, pattern, sharded",
                  error.message
   end
 
-  # --- FFI push handler ----------------------------------------------------
-
-  def test_ffi_handler_queues_messages
-    push(Kind::MESSAGE, message: "exact", channel: "news")
-    push(Kind::PMESSAGE, message: "pattern", channel: "news.tech", pattern: "news.*")
-    push(Kind::SMESSAGE, message: "sharded", channel: "shard-chan")
-
-    expected = [
-      ["exact", "news", nil],
-      ["pattern", "news.tech", "news.*"],
-      ["sharded", "shard-chan", nil]
-    ]
-
-    received = 3.times.map { @pubsub.try_get_message.to_a }
-
-    assert_equal expected, received
-  end
-
-  def test_ffi_handler_drops_non_message_kinds
-    non_message_kinds = [
-      Kind::DISCONNECTION, Kind::OTHER, Kind::INVALIDATE,
-      Kind::SUBSCRIBE, Kind::PSUBSCRIBE, Kind::SSUBSCRIBE,
-      Kind::UNSUBSCRIBE, Kind::PUNSUBSCRIBE, Kind::SUNSUBSCRIBE
-    ]
-
-    non_message_kinds.each do |kind|
-      push(kind, message: "phantom", channel: "news")
-
-      assert_nil @pubsub.try_get_message, "kind #{kind} must not queue a message"
-    end
-  end
-
-  def test_ffi_handler_with_embedded_nul
-    payload = "before\0after"
-
-    push(Kind::MESSAGE, message: payload, channel: "news")
-
-    assert_equal payload, @pubsub.try_get_message.message
-  end
-
   # --- Command dispatch ----------------------------------------------------
 
   def test_subscribe
     client = RecordingClient.new
-    pubsub = build_pubsub(client)
 
-    pubsub.subscribe("news", "alerts", timeout_ms: 2000)
+    client.subscribe("news", "alerts", timeout_ms: 2000)
 
     assert_equal Valkey::RequestType::SUBSCRIBE_BLOCKING, client.last_command.request_type
     assert_equal %w[news alerts 2000], client.last_command.args
@@ -199,17 +157,16 @@ def test_subscribe
 
   def test_subscribe_with_default
     client = RecordingClient.new
-    build_pubsub(client).subscribe("news")
+    client.subscribe("news")
 
     assert_equal %w[news 0], client.last_command.args
   end
 
   def test_subscribe_fractional_milliseconds
     client = RecordingClient.new
-    pubsub = build_pubsub(client)
 
-    pubsub.subscribe("news", timeout_ms: 1500.6)
-    pubsub.subscribe("news", timeout_ms: 250.2)
+    client.subscribe("news", timeout_ms: 1500.6)
+    client.subscribe("news", timeout_ms: 250.2)
 
     assert_equal %w[news 1500], client.sent_commands[0].args
     assert_equal %w[news 250], client.sent_commands[1].args
@@ -218,7 +175,7 @@ def test_subscribe_fractional_milliseconds
   def test_subscribe_sub_milliseconds_timeout
     client = RecordingClient.new
 
-    build_pubsub(client).subscribe("news", timeout_ms: 0.4)
+    client.subscribe("news", timeout_ms: 0.4)
 
     assert_equal %w[news 1], client.last_command.args
   end
@@ -226,14 +183,14 @@ def test_subscribe_sub_milliseconds_timeout
   def test_subscribe_rejects_a_negative_timeout
     client = RecordingClient.new
 
-    assert_raises(ArgumentError) { build_pubsub(client).subscribe("news", timeout_ms: -1) }
+    assert_raises(ArgumentError) { client.subscribe("news", timeout_ms: -1) }
     assert_empty client.sent_commands
   end
 
   def test_subscribe_without_channels_raises
     client = RecordingClient.new
 
-    error = assert_raises(ArgumentError) { build_pubsub(client).subscribe }
+    error = assert_raises(ArgumentError) { client.subscribe }
 
     assert_equal "No channels provided for subscription", error.message
     assert_empty client.sent_commands
@@ -242,7 +199,7 @@ def test_subscribe_without_channels_raises
   def test_unsubscribe
     client = RecordingClient.new
 
-    build_pubsub(client).unsubscribe("news", timeout_ms: 3000)
+    client.unsubscribe("news", timeout_ms: 3000)
 
     assert_equal Valkey::RequestType::UNSUBSCRIBE_BLOCKING, client.last_command.request_type
     assert_equal %w[news 3000], client.last_command.args
@@ -251,7 +208,7 @@ def test_unsubscribe
   def test_unsubscribe_default
     client = RecordingClient.new
 
-    build_pubsub(client).unsubscribe
+    client.unsubscribe
 
     assert_equal Valkey::RequestType::UNSUBSCRIBE_BLOCKING, client.last_command.request_type
     assert_equal %w[0], client.last_command.args
@@ -260,14 +217,14 @@ def test_unsubscribe_default
   def test_unsubscribe_rejects_a_negative_timeout
     client = RecordingClient.new
 
-    assert_raises(ArgumentError) { build_pubsub(client).unsubscribe("news", timeout_ms: -0.5) }
+    assert_raises(ArgumentError) { client.unsubscribe("news", timeout_ms: -0.5) }
     assert_empty client.sent_commands
   end
 
   def test_publish_works_without_resp3
-    client = RecordingClient.new(response: 0)
+    client = RecordingClient.new(response: 0, protocol: nil)
 
-    Valkey::Glide::PubSub.new(client, cluster_mode: false).publish("hello", "news")
+    client.publish("hello", "news")
 
     assert_equal %w[news hello], client.last_command.args
   end
@@ -276,11 +233,11 @@ def test_publish_works_without_resp3
 
   def test_subscription_methods_reject_a_non_resp3_protocol
     [nil, :resp2, "resp2", 2].each do |protocol|
-      pubsub = build_pubsub(RecordingClient.new, protocol: protocol)
+      client = RecordingClient.new(protocol: protocol)
 
-      guarded_calls(pubsub).each do |name, call|
+      guarded_calls(client).each do |name, call|
         # Timeout so a missing guard fails the assertion instead of blocking in
-        # get_message forever.
+        # get_pubsub_message forever.
         error = assert_raises(Valkey::Resp3RequiredError, "#{name} must reject protocol #{protocol.inspect}") do
           Timeout.timeout(2) { call.call }
         end
@@ -291,58 +248,41 @@ def test_subscription_methods_reject_a_non_resp3_protocol
 
   def test_subscription_methods_accept_every_resp3_spelling
     [:resp3, "resp3", 3].each do |protocol|
-      pubsub = build_pubsub(RecordingClient.new, protocol: protocol)
+      client = RecordingClient.new(protocol: protocol)
 
-      pubsub.subscribe("news")
-      pubsub.unsubscribe
+      client.subscribe("news")
+      client.unsubscribe
 
-      assert_nil pubsub.try_get_message, "protocol #{protocol.inspect} must be accepted"
-      assert_equal "hello", queued_message(pubsub).message
+      assert_nil client.try_get_pubsub_message, "protocol #{protocol.inspect} must be accepted"
+      assert_equal "hello", queued_message(client).message
     end
   end
 
   private
 
-  def build_pubsub(client, protocol: :resp3)
-    Valkey::Glide::PubSub.new(client, cluster_mode: false, protocol: protocol)
-  end
-
-  # get_message blocks, so it is only called on a queue that already holds one.
-  def queued_message(pubsub)
-    push(Kind::MESSAGE, message: "hello", channel: "news", pubsub: pubsub)
-    pubsub.get_message
+  def receiver_for(client)
+    client.instance_variable_get(:@pubsub_receiver)
   end
 
-  def guarded_calls(pubsub)
-    {
-      subscribe: -> { pubsub.subscribe("news") },
-      unsubscribe: -> { pubsub.unsubscribe },
-      get_message: -> { pubsub.get_message },
-      try_get_message: -> { pubsub.try_get_message }
-    }
+  # Bound via reflection because the parser is private on the client.
+  def parse_pubsub_configs
+    @pubsub.method(:parse_pubsub_configs)
   end
 
-  # Calls the retained FFI handler the way the Rust push worker does, with real
-  # buffers and the byte lengths alongside them.
-  def push(kind, message: nil, channel: nil, pattern: nil, pubsub: nil)
-    message_pointer, message_length = buffer_for(message)
-    channel_pointer, channel_length = buffer_for(channel)
-    pattern_pointer, pattern_length = buffer_for(pattern)
-
-    (pubsub || @pubsub).ffi_handler.call(
-      0, kind,
-      message_pointer, message_length,
-      channel_pointer, channel_length,
-      pattern_pointer, pattern_length
-    )
+  # get_pubsub_message blocks, so it is only called on a queue that already
+  # holds one.
+  def queued_message(client)
+    receiver_for(client).send(:deliver,
+                              Valkey::Glide::PubSubMessage.new("hello", "news", nil))
+    client.get_pubsub_message
   end
 
-  def buffer_for(value)
-    return [FFI::Pointer::NULL, 0] if value.nil?
-
-    bytes = value.b
-    buffer = FFI::MemoryPointer.new(:char, bytes.bytesize)
-    buffer.put_bytes(0, bytes)
-    [buffer, bytes.bytesize]
+  def guarded_calls(client)
+    {
+      subscribe: -> { client.subscribe("news") },
+      unsubscribe: -> { client.unsubscribe },
+      get_pubsub_message: -> { client.get_pubsub_message },
+      try_get_pubsub_message: -> { client.try_get_pubsub_message }
+    }
   end
 end
diff --git a/test/unit/pipeline_test.rb b/test/unit/pipeline_test.rb
index 68613d63..74ea3494 100644
--- a/test/unit/pipeline_test.rb
+++ b/test/unit/pipeline_test.rb
@@ -41,4 +41,14 @@ def test_abort_futures_marks_only_unresolved_futures
     assert_equal "OK", future_a.value
     assert_raises(Valkey::FutureAborted) { future_b.value }
   end
+
+  def test_pubsub_commands_raise_argument_error
+    pipeline = Valkey::Pipeline.new
+
+    Valkey::Pipeline::PUBSUB_UNSUPPORTED.each do |name|
+      error = assert_raises(ArgumentError, "#{name} must be rejected") { pipeline.public_send(name) }
+
+      assert_equal "#{name} is not supported inside pipelined/multi", error.message
+    end
+  end
 end