diff --git a/CHANGELOG.md b/CHANGELOG.md
index 597458b..c4acf2b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,15 @@
 # Changelog
 
+## [0.3.8] - 2026-04-25
+
+### Added
+- Configurable cluster-control payload signing via `extensions.node.cluster_control.auth.mode` (`auto`, `required`, `disabled`), with timestamp freshness and nonce format validation when signing is active.
+
+### Changed
+- `ClusterControl` queues now read durability and retention settings from `extensions.node.cluster_control.queue`; defaults remain durable and non-auto-delete so per-node commands survive node restarts.
+- Beat no longer runs immediately at actor construction, avoiding startup reconciliation before extension boot completes.
+- `update_gem` now installs through `Legion::Extensions::GemSource` and calls extension-scoped reload instead of full daemon reload.
+
 ## [0.3.7] - 2026-03-31
 
 ### Fixed
diff --git a/README.md b/README.md
index 2ec95e0..132d0c1 100644
--- a/README.md
+++ b/README.md
@@ -68,6 +68,37 @@ Each node periodically calls `beat` to broadcast its presence. On startup, nodes
 
 Dynamic config changes (`update_settings`) and gem upgrades (`update_gem`) can be pushed to individual nodes or broadcast to all nodes at runtime. Both operations publish an `UpdateResult` message to `node.<name>.update_result` so the outcome can be observed cluster-wide.
 
+### Cluster Control Settings
+
+Cluster-control broadcasts are configurable under `extensions.node.cluster_control`. By default, auth mode is `auto`: messages are signed and verified when a shared secret is configured, and unsigned messages are allowed when no secret exists for simpler local or homelab deployments.
+
+```json
+{
+  "extensions": {
+    "node": {
+      "cluster_control": {
+        "auth": {
+          "mode": "auto",
+          "timestamp_skew_seconds": 300,
+          "nonce_bytes": 16
+        },
+        "queue": {
+          "durable": true,
+          "exclusive": false,
+          "auto_delete": false,
+          "queue_type": "classic",
+          "expires_ms": 604800000,
+          "message_ttl_ms": 86400000,
+          "max_length": 1000
+        }
+      }
+    }
+  }
+}
+```
+
+Set `auth.mode` to `required` to force HMAC signatures, or `disabled` to run without cluster-control signatures even when a shared secret is present.
+
 ## Transport
 
 - **Exchange**: `node` (topic exchange)
diff --git a/lib/legion/extensions/node.rb b/lib/legion/extensions/node.rb
index 8f372bc..183bfae 100644
--- a/lib/legion/extensions/node.rb
+++ b/lib/legion/extensions/node.rb
@@ -1,12 +1,17 @@
 # frozen_string_literal: true
 
 require 'legion/extensions/node/version'
+require 'legion/extensions/node/config'
 require 'legion/extensions/node/helpers/rabbitmq'
 
 module Legion
   module Extensions
     module Node
       extend Legion::Extensions::Core if Legion::Extensions.const_defined? :Core, false
+
+      def self.default_settings
+        Config.default_settings
+      end
     end
   end
 end
diff --git a/lib/legion/extensions/node/actors/beat.rb b/lib/legion/extensions/node/actors/beat.rb
index a63ee39..e168b62 100644
--- a/lib/legion/extensions/node/actors/beat.rb
+++ b/lib/legion/extensions/node/actors/beat.rb
@@ -22,7 +22,7 @@ def generate_task?
           end
 
           def run_now?
-            true
+            false
           end
 
           def time
diff --git a/lib/legion/extensions/node/actors/cluster_control.rb b/lib/legion/extensions/node/actors/cluster_control.rb
index ce1b044..aa38ef5 100644
--- a/lib/legion/extensions/node/actors/cluster_control.rb
+++ b/lib/legion/extensions/node/actors/cluster_control.rb
@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require 'legion/extensions/node/control_auth'
+
 module Legion
   module Extensions
     module Node
@@ -28,6 +30,11 @@ def check_subtask?
           def generate_task?
             false
           end
+
+          def process_message(message, metadata, delivery_info)
+            verified_message = Legion::Extensions::Node::ControlAuth.verify!(message)
+            super(verified_message, metadata, delivery_info)
+          end
         end
       end
     end
diff --git a/lib/legion/extensions/node/config.rb b/lib/legion/extensions/node/config.rb
new file mode 100644
index 0000000..b195714
--- /dev/null
+++ b/lib/legion/extensions/node/config.rb
@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+
+module Legion
+  module Extensions
+    module Node
+      module Config
+        DEFAULT_SETTINGS = {
+          cluster_control: {
+            auth:  {
+              mode:                   'auto',
+              timestamp_skew_seconds: 300,
+              nonce_bytes:            16
+            },
+            queue: {
+              durable:        true,
+              exclusive:      false,
+              auto_delete:    false,
+              queue_type:     'classic',
+              expires_ms:     604_800_000,
+              message_ttl_ms: 86_400_000,
+              max_length:     1000
+            }
+          }
+        }.freeze
+
+        module_function
+
+        def default_settings
+          deep_dup(DEFAULT_SETTINGS)
+        end
+
+        def cluster_control
+          deep_merge(default_settings[:cluster_control], extension_settings[:cluster_control] || {})
+        end
+
+        def control_auth
+          cluster_control[:auth] || {}
+        end
+
+        def control_queue
+          cluster_control[:queue] || {}
+        end
+
+        def extension_settings
+          return {} unless defined?(Legion::Settings) && Legion::Settings.respond_to?(:dig)
+
+          settings = Legion::Settings.dig(:extensions, :node)
+          settings.is_a?(Hash) ? symbolize_keys(settings) : {}
+        rescue StandardError => e
+          log.debug("node settings lookup failed: #{e.message}")
+          {}
+        end
+
+        def deep_merge(base, override)
+          base_hash = symbolize_keys(base)
+          override_hash = symbolize_keys(override)
+          base_hash.merge(override_hash) do |_key, old_value, new_value|
+            old_value.is_a?(Hash) && new_value.is_a?(Hash) ? deep_merge(old_value, new_value) : new_value
+          end
+        end
+
+        def deep_dup(value)
+          case value
+          when Hash
+            value.each_with_object({}) { |(key, val), result| result[key] = deep_dup(val) }
+          when Array
+            value.map { |item| deep_dup(item) }
+          else
+            value
+          end
+        end
+
+        def symbolize_keys(value)
+          case value
+          when Hash
+            value.each_with_object({}) { |(key, val), result| result[key.to_sym] = symbolize_keys(val) }
+          when Array
+            value.map { |item| symbolize_keys(item) }
+          else
+            value
+          end
+        end
+
+        def log
+          Legion::Logging.respond_to?(:logger) ? Legion::Logging.logger : Legion::Logging
+        end
+      end
+    end
+  end
+end
diff --git a/lib/legion/extensions/node/control_auth.rb b/lib/legion/extensions/node/control_auth.rb
new file mode 100644
index 0000000..8b3accb
--- /dev/null
+++ b/lib/legion/extensions/node/control_auth.rb
@@ -0,0 +1,186 @@
+# frozen_string_literal: true
+
+require 'openssl'
+require 'securerandom'
+require 'json'
+require 'legion/extensions/node/config'
+
+module Legion
+  module Extensions
+    module Node
+      module ControlAuth
+        AUTH_MODES = %w[auto required disabled].freeze
+
+        class UnauthorizedControlMessage < StandardError; end
+
+        module_function
+
+        def sign(payload)
+          normalized = deep_symbolize(payload).dup
+          return normalized unless sign_control_messages?
+
+          control = {
+            sender:    node_name,
+            timestamp: Time.now.utc.to_i,
+            nonce:     SecureRandom.hex(nonce_bytes)
+          }
+          normalized[:control] = control.merge(signature: signature_for(normalized.merge(control: control)))
+          normalized
+        end
+
+        def verify!(payload)
+          normalized = deep_symbolize(payload)
+          raise UnauthorizedControlMessage, 'missing control signature' unless normalized.is_a?(Hash)
+          return normalized unless verify_control_messages?
+
+          control = normalized[:control]
+          raise UnauthorizedControlMessage, 'missing control signature' unless control.is_a?(Hash)
+
+          signature = control[:signature].to_s
+          raise UnauthorizedControlMessage, 'missing control signature' if signature.empty?
+          raise UnauthorizedControlMessage, 'stale control message' unless fresh_timestamp?(control[:timestamp])
+          raise UnauthorizedControlMessage, 'invalid control nonce' unless valid_nonce?(control[:nonce])
+
+          unsigned_control = control.dup
+          unsigned_control.delete(:signature)
+          expected = signature_for(normalized.merge(control: unsigned_control))
+          raise UnauthorizedControlMessage, 'invalid control signature' unless secure_compare(signature, expected)
+
+          normalized
+        end
+
+        def sign_control_messages?
+          return false if auth_disabled?
+          return true if auth_required?
+
+          !secret.to_s.empty?
+        end
+
+        def verify_control_messages?
+          return false if auth_disabled?
+          return true if auth_required?
+
+          !secret.to_s.empty?
+        end
+
+        def auth_mode
+          configured = auth_settings[:mode] || auth_settings[:enabled]
+          mode = case configured
+                 when true then 'required'
+                 when false then 'disabled'
+                 else configured.to_s
+                 end
+          AUTH_MODES.include?(mode) ? mode : 'auto'
+        end
+
+        def auth_required?
+          auth_mode == 'required'
+        end
+
+        def auth_disabled?
+          auth_mode == 'disabled'
+        end
+
+        def auth_settings
+          Legion::Extensions::Node::Config.control_auth
+        end
+
+        def secret
+          configured_secret || ENV.fetch('LEGION_NODE_CONTROL_SECRET', nil)
+        end
+
+        def configured_secret
+          if defined?(Legion::Settings) && Legion::Settings.respond_to?(:dig)
+            auth_settings[:secret] ||
+              Legion::Settings.dig(:cluster, :control_secret) ||
+              Legion::Settings.dig(:crypt, :cluster_secret)
+          end
+        rescue StandardError => e
+          log.debug("control secret lookup failed: #{e.message}")
+          nil
+        end
+
+        def signature_for(payload)
+          value = secret
+          raise UnauthorizedControlMessage, 'cluster control secret is not configured' if value.to_s.empty?
+
+          OpenSSL::HMAC.hexdigest('SHA256', value.to_s, canonical_json(payload))
+        end
+
+        def canonical_json(value)
+          ::JSON.generate(canonicalize(value))
+        end
+
+        def canonicalize(value)
+          case value
+          when Hash
+            value.each_with_object({}) do |(key, val), result|
+              result[key.to_s] = canonicalize(val)
+            end.sort.to_h
+          when Array
+            value.map { |item| canonicalize(item) }
+          else
+            value
+          end
+        end
+
+        def deep_symbolize(value)
+          case value
+          when Hash
+            value.each_with_object({}) do |(key, val), result|
+              result[key.to_sym] = deep_symbolize(val)
+            end
+          when Array
+            value.map { |item| deep_symbolize(item) }
+          else
+            value
+          end
+        end
+
+        def fresh_timestamp?(timestamp)
+          ts = Integer(timestamp)
+          (Time.now.utc.to_i - ts).abs <= timestamp_skew_seconds
+        rescue ArgumentError, TypeError => e
+          log.debug("control timestamp validation failed: #{e.message}")
+          false
+        end
+
+        def timestamp_skew_seconds
+          Integer(auth_settings[:timestamp_skew_seconds])
+        rescue ArgumentError, TypeError => e
+          log.debug("control timestamp skew setting invalid: #{e.message}")
+          300
+        end
+
+        def nonce_bytes
+          bytes = Integer(auth_settings[:nonce_bytes])
+          bytes.positive? ? bytes : 16
+        rescue ArgumentError, TypeError => e
+          log.debug("control nonce byte setting invalid: #{e.message}")
+          16
+        end
+
+        def valid_nonce?(nonce)
+          nonce.to_s.match?(/\A[0-9a-f]{#{nonce_bytes * 2}}\z/i)
+        end
+
+        def secure_compare(left, right)
+          return false unless left.bytesize == right.bytesize
+
+          left.bytes.zip(right.bytes).reduce(0) { |acc, (a, b)| acc | (a ^ b) }.zero?
+        end
+
+        def node_name
+          (Legion::Settings[:client]&.fetch(:name, nil) if defined?(Legion::Settings)) || 'unknown'
+        rescue StandardError => e
+          log.debug("node name lookup failed: #{e.message}")
+          'unknown'
+        end
+
+        def log
+          Legion::Logging.respond_to?(:logger) ? Legion::Logging.logger : Legion::Logging
+        end
+      end
+    end
+  end
+end
diff --git a/lib/legion/extensions/node/runners/node.rb b/lib/legion/extensions/node/runners/node.rb
index a899996..e507fa8 100644
--- a/lib/legion/extensions/node/runners/node.rb
+++ b/lib/legion/extensions/node/runners/node.rb
@@ -28,8 +28,10 @@ def update_gem(extension:, version: nil, reload: true, **_opts)
             gem_name = "lex-#{name}"
             log.debug "update_gem: installing #{gem_name} #{version || 'latest'}"
 
-            Gem.install(gem_name, version)
-            Legion.reload if reload
+            install_result = install_extension_gem(gem_name, version)
+            raise Gem::InstallError, install_result[:output].to_s unless install_result[:success]
+
+            reload_extension(gem_name) if reload
 
             publish_update_result(action: 'update_gem', status: 'success', detail: "#{gem_name} #{version || 'latest'}")
           rescue StandardError => e
@@ -146,6 +148,20 @@ def killswitch(extension:, **_opts)
 
           private
 
+          def install_extension_gem(gem_name, version)
+            require 'legion/extensions/gem_source' unless defined?(Legion::Extensions::GemSource) # rubocop:disable Legion/HelperMigration/RequireDefinedGuard
+
+            Legion::Extensions::GemSource.install_gem(gem_name, version: version)
+          rescue LoadError => e
+            { success: false, output: "GemSource unavailable: #{e.message}" }
+          end
+
+          def reload_extension(gem_name)
+            raise 'extension-scoped reload is unavailable' unless defined?(Legion::Extensions) && Legion::Extensions.respond_to?(:reload_extension)
+
+            Legion::Extensions.reload_extension(gem_name)
+          end
+
           def publish_update_result(action:, status:, detail: nil, error: nil)
             Legion::Extensions::Node::Transport::Messages::UpdateResult.new(
               action:    action,
diff --git a/lib/legion/extensions/node/transport/messages/cluster_killswitch.rb b/lib/legion/extensions/node/transport/messages/cluster_killswitch.rb
index bb03006..e2ae86a 100644
--- a/lib/legion/extensions/node/transport/messages/cluster_killswitch.rb
+++ b/lib/legion/extensions/node/transport/messages/cluster_killswitch.rb
@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require 'legion/extensions/node/control_auth'
+
 module Legion
   module Extensions
     module Node
@@ -25,11 +27,11 @@ def encrypt?
             end
 
             def message
-              {
+              Legion::Extensions::Node::ControlAuth.sign(
                 function: 'update_settings',
                 settings: { extensions: { blocked: [@options[:extension]] } },
                 restart:  true
-              }
+              )
             end
 
             def validate
diff --git a/lib/legion/extensions/node/transport/messages/cluster_settings.rb b/lib/legion/extensions/node/transport/messages/cluster_settings.rb
index 7728e34..932a559 100644
--- a/lib/legion/extensions/node/transport/messages/cluster_settings.rb
+++ b/lib/legion/extensions/node/transport/messages/cluster_settings.rb
@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require 'legion/extensions/node/control_auth'
+
 module Legion
   module Extensions
     module Node
@@ -23,11 +25,11 @@ def encrypt?
             end
 
             def message
-              {
+              Legion::Extensions::Node::ControlAuth.sign(
                 function: 'update_settings',
                 settings: @options[:settings],
                 restart:  @options.fetch(:restart, false)
-              }
+              )
             end
 
             def validate
diff --git a/lib/legion/extensions/node/transport/queues/cluster_control.rb b/lib/legion/extensions/node/transport/queues/cluster_control.rb
index 4f39b09..64b000d 100644
--- a/lib/legion/extensions/node/transport/queues/cluster_control.rb
+++ b/lib/legion/extensions/node/transport/queues/cluster_control.rb
@@ -1,5 +1,7 @@
 # frozen_string_literal: true
 
+require 'legion/extensions/node/config'
+
 module Legion
   module Extensions
     module Node
@@ -11,8 +13,36 @@ def queue_name
             end
 
             def queue_options
-              { durable: false, exclusive: false, auto_delete: true,
-                arguments: { 'x-queue-type': 'classic' } }
+              options = Legion::Extensions::Node::Config.control_queue
+              { durable:     truthy?(options[:durable], default: true),
+                exclusive:   truthy?(options[:exclusive], default: false),
+                auto_delete: truthy?(options[:auto_delete], default: false),
+                arguments:   queue_arguments(options) }
+            end
+
+            private
+
+            def queue_arguments(options)
+              {
+                'x-queue-type':  options[:queue_type],
+                'x-expires':     positive_integer(options[:expires_ms]),
+                'x-message-ttl': positive_integer(options[:message_ttl_ms]),
+                'x-max-length':  positive_integer(options[:max_length])
+              }.compact
+            end
+
+            def positive_integer(value)
+              return value if value.is_a?(Integer) && value.positive?
+              return nil unless value.to_s.match?(/\A\d+\z/)
+
+              parsed = value.to_i
+              parsed.positive? ? parsed : nil
+            end
+
+            def truthy?(value, default:)
+              return default if value.nil?
+
+              value == true || value.to_s == 'true'
             end
           end
         end
diff --git a/lib/legion/extensions/node/version.rb b/lib/legion/extensions/node/version.rb
index 4383ccc..19325ed 100644
--- a/lib/legion/extensions/node/version.rb
+++ b/lib/legion/extensions/node/version.rb
@@ -3,7 +3,7 @@
 module Legion
   module Extensions
     module Node
-      VERSION = '0.3.7'
+      VERSION = '0.3.8'
     end
   end
 end
diff --git a/spec/legion/extensions/node/actors/beat_spec.rb b/spec/legion/extensions/node/actors/beat_spec.rb
index e931eef..1b54f0f 100644
--- a/spec/legion/extensions/node/actors/beat_spec.rb
+++ b/spec/legion/extensions/node/actors/beat_spec.rb
@@ -66,8 +66,8 @@ def call(method_name)
   end
 
   describe '#run_now?' do
-    it 'returns true' do
-      expect(call(:run_now?)).to be true
+    it 'returns false so heartbeat/reconcile does not run before extension boot completes' do
+      expect(call(:run_now?)).to be false
     end
   end
 
diff --git a/spec/legion/extensions/node/actors/cluster_control_spec.rb b/spec/legion/extensions/node/actors/cluster_control_spec.rb
index a63926a..974e3d5 100644
--- a/spec/legion/extensions/node/actors/cluster_control_spec.rb
+++ b/spec/legion/extensions/node/actors/cluster_control_spec.rb
@@ -8,6 +8,10 @@ module Extensions
       module Actors
         class Subscription
           def initialize(**); end
+
+          def process_message(message, _metadata, _delivery_info)
+            [:processed, message]
+          end
         end
       end
     end
@@ -85,4 +89,17 @@ def call(method_name)
       expect(call(:generate_task?)).to be false
     end
   end
+
+  describe '#process_message' do
+    it 'verifies the incoming payload before dispatching to the subscription handler' do
+      raw_message = { function: 'update_settings' }
+      verified_message = { function: 'update_settings', control: { verified: true } }
+      allow(Legion::Extensions::Node::ControlAuth).to receive(:verify!)
+        .with(raw_message).and_return(verified_message)
+
+      result = instance.process_message(raw_message, double('metadata'), double('delivery_info'))
+
+      expect(result).to eq([:processed, verified_message])
+    end
+  end
 end
diff --git a/spec/legion/extensions/node/config_spec.rb b/spec/legion/extensions/node/config_spec.rb
new file mode 100644
index 0000000..3a9d761
--- /dev/null
+++ b/spec/legion/extensions/node/config_spec.rb
@@ -0,0 +1,61 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require 'legion/extensions/node/config'
+
+RSpec.describe Legion::Extensions::Node::Config do
+  before do
+    allow(Legion::Settings).to receive(:dig).and_return(nil)
+  end
+
+  describe '.default_settings' do
+    it 'returns a defensive copy of the node defaults' do
+      settings = described_class.default_settings
+      settings[:cluster_control][:auth][:mode] = 'disabled'
+
+      expect(described_class.default_settings[:cluster_control][:auth][:mode]).to eq('auto')
+    end
+  end
+
+  describe '.control_auth' do
+    it 'deep merges string-keyed settings over defaults' do
+      allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return(
+        'cluster_control' => {
+          'auth' => {
+            'mode'                   => 'required',
+            'timestamp_skew_seconds' => 60
+          }
+        }
+      )
+
+      expect(described_class.control_auth).to include(
+        mode:                   'required',
+        timestamp_skew_seconds: 60,
+        nonce_bytes:            16
+      )
+    end
+  end
+
+  describe '.control_queue' do
+    it 'deep merges partial queue overrides over defaults' do
+      allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return(
+        cluster_control: {
+          queue: {
+            durable:    false,
+            expires_ms: nil
+          }
+        }
+      )
+
+      expect(described_class.control_queue).to include(
+        durable:        false,
+        exclusive:      false,
+        auto_delete:    false,
+        queue_type:     'classic',
+        expires_ms:     nil,
+        message_ttl_ms: 86_400_000,
+        max_length:     1000
+      )
+    end
+  end
+end
diff --git a/spec/legion/extensions/node/control_auth_spec.rb b/spec/legion/extensions/node/control_auth_spec.rb
new file mode 100644
index 0000000..f414313
--- /dev/null
+++ b/spec/legion/extensions/node/control_auth_spec.rb
@@ -0,0 +1,100 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+require 'legion/extensions/node/control_auth'
+
+RSpec.describe Legion::Extensions::Node::ControlAuth do
+  before do
+    allow(Legion::Settings).to receive(:dig).and_return(nil)
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return({})
+    allow(Legion::Settings).to receive(:dig).with(:cluster, :control_secret).and_return('shared-secret')
+    allow(Legion::Settings).to receive(:[]).with(:client).and_return({ name: 'sender-node' })
+  end
+
+  it 'signs and verifies cluster control payloads when a shared secret is configured' do
+    payload = described_class.sign(function: 'update_settings', settings: { extensions: { blocked: ['bad'] } })
+
+    expect(payload[:control]).to include(:sender, :timestamp, :nonce, :signature)
+    expect(described_class.verify!(payload)).to eq(payload)
+  end
+
+  it 'passes unsigned payloads in auto mode when no shared secret is configured' do
+    allow(Legion::Settings).to receive(:dig).with(:cluster, :control_secret).and_return(nil)
+
+    payload = described_class.sign(function: 'update_settings')
+
+    expect(payload).not_to include(:control)
+    expect(described_class.verify!(payload)).to eq(payload)
+  end
+
+  it 'rejects unsigned payloads when a shared secret is configured' do
+    expect { described_class.verify!(function: 'update_settings') }
+      .to raise_error(Legion::Extensions::Node::ControlAuth::UnauthorizedControlMessage)
+  end
+
+  it 'requires signed payloads when auth mode is required' do
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return(
+      cluster_control: { auth: { mode: 'required' } }
+    )
+    allow(Legion::Settings).to receive(:dig).with(:cluster, :control_secret).and_return(nil)
+
+    expect { described_class.verify!(function: 'update_settings') }
+      .to raise_error(Legion::Extensions::Node::ControlAuth::UnauthorizedControlMessage)
+  end
+
+  it 'does not sign or verify payloads when auth mode is disabled' do
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return(
+      cluster_control: { auth: { mode: 'disabled' } }
+    )
+
+    payload = described_class.sign(function: 'update_settings', settings: { feature: true })
+    tampered = payload.merge(settings: { feature: false })
+
+    expect(payload).not_to include(:control)
+    expect(described_class.verify!(tampered)).to eq(tampered)
+  end
+
+  it 'rejects malformed payloads predictably' do
+    expect { described_class.verify!(nil) }
+      .to raise_error(Legion::Extensions::Node::ControlAuth::UnauthorizedControlMessage, 'missing control signature')
+  end
+
+  it 'rejects payloads without a valid nonce' do
+    payload = described_class.sign(function: 'update_settings', settings: { feature: true })
+    payload[:control] = payload[:control].merge(nonce: '')
+
+    expect { described_class.verify!(payload) }
+      .to raise_error(Legion::Extensions::Node::ControlAuth::UnauthorizedControlMessage, 'invalid control nonce')
+  end
+
+  it 'uses configured nonce byte length' do
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return(
+      cluster_control: { auth: { nonce_bytes: 8 } }
+    )
+
+    payload = described_class.sign(function: 'update_settings', settings: { feature: true })
+
+    expect(payload[:control][:nonce]).to match(/\A[0-9a-f]{16}\z/i)
+    expect(described_class.verify!(payload)).to eq(payload)
+  end
+
+  it 'uses configured timestamp skew' do
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return(
+      cluster_control: { auth: { timestamp_skew_seconds: 1 } }
+    )
+    allow(Time).to receive(:now).and_return(Time.utc(2026, 1, 1, 0, 0, 0))
+    payload = described_class.sign(function: 'update_settings', settings: { feature: true })
+    allow(Time).to receive(:now).and_return(Time.utc(2026, 1, 1, 0, 0, 2))
+
+    expect { described_class.verify!(payload) }
+      .to raise_error(Legion::Extensions::Node::ControlAuth::UnauthorizedControlMessage, 'stale control message')
+  end
+
+  it 'rejects tampered payloads' do
+    payload = described_class.sign(function: 'update_settings', settings: { feature: true })
+    payload[:settings] = { feature: false }
+
+    expect { described_class.verify!(payload) }
+      .to raise_error(Legion::Extensions::Node::ControlAuth::UnauthorizedControlMessage)
+  end
+end
diff --git a/spec/legion/extensions/node/runners/node_spec.rb b/spec/legion/extensions/node/runners/node_spec.rb
index 08dec1a..4a1c1d1 100644
--- a/spec/legion/extensions/node/runners/node_spec.rb
+++ b/spec/legion/extensions/node/runners/node_spec.rb
@@ -314,31 +314,39 @@ def error(*); end
   describe '#update_gem' do
     before do
       @settings_store[:client] = { name: 'test-node' }
+      gem_source = Module.new do
+        def self.install_gem(*) = { success: true, output: 'installed' }
+      end
+      stub_const('Legion::Extensions::GemSource', gem_source)
+      allow(Legion::Extensions::GemSource).to receive(:install_gem).and_return({ success: true, output: 'installed' })
       allow(Gem).to receive(:install)
       allow(Legion).to receive(:reload)
+      allow(Legion::Extensions).to receive(:reload_extension).and_return(true)
       allow(Legion::Extensions::Node::Transport::Messages::UpdateResult).to receive(:new).and_return(
         instance_double(Legion::Extensions::Node::Transport::Messages::UpdateResult, publish: nil)
       )
     end
 
-    it 'calls Gem.install with the normalized extension name and version' do
+    it 'installs through the configured GemSource with the normalized extension name and version' do
       runner.update_gem(extension: 'github', version: '0.3.0')
-      expect(Gem).to have_received(:install).with('lex-github', '0.3.0')
+      expect(Legion::Extensions::GemSource).to have_received(:install_gem).with('lex-github', version: '0.3.0')
+      expect(Gem).not_to have_received(:install)
     end
 
     it 'normalizes extension name that already has lex- prefix' do
       runner.update_gem(extension: 'lex-github', version: '0.3.0')
-      expect(Gem).to have_received(:install).with('lex-github', '0.3.0')
+      expect(Legion::Extensions::GemSource).to have_received(:install_gem).with('lex-github', version: '0.3.0')
     end
 
     it 'passes nil version for latest' do
       runner.update_gem(extension: 'github')
-      expect(Gem).to have_received(:install).with('lex-github', nil)
+      expect(Legion::Extensions::GemSource).to have_received(:install_gem).with('lex-github', version: nil)
     end
 
-    it 'calls Legion.reload when reload is true' do
+    it 'calls extension-scoped reload when reload is true' do
       runner.update_gem(extension: 'github', reload: true)
-      expect(Legion).to have_received(:reload)
+      expect(Legion::Extensions).to have_received(:reload_extension).with('lex-github')
+      expect(Legion).not_to have_received(:reload)
     end
 
     it 'does not call Legion.reload when reload is false' do
@@ -353,8 +361,8 @@ def error(*); end
       runner.update_gem(extension: 'github', version: '0.3.0')
     end
 
-    context 'when Gem.install raises an error' do
-      before { allow(Gem).to receive(:install).and_raise(Gem::InstallError, 'network timeout') }
+    context 'when install_gem returns a failed result' do
+      before { allow(Legion::Extensions::GemSource).to receive(:install_gem).and_return({ success: false, output: 'network timeout' }) }
 
       it 'does not crash' do
         expect { runner.update_gem(extension: 'github') }.not_to raise_error
diff --git a/spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb b/spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb
index 1c6c0c8..1a92f81 100644
--- a/spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb
+++ b/spec/legion/extensions/node/transport/messages/cluster_killswitch_spec.rb
@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'spec_helper'
+require 'legion/extensions/node/control_auth'
 require 'legion/extensions/node/transport/exchanges/cluster_control'
 
 CLUSTER_KILLSWITCH_ROUTING_KEY = 'settings.extensions.blocked'
@@ -28,11 +29,11 @@ def encrypt?
   end
 
   def message
-    {
+    Legion::Extensions::Node::ControlAuth.sign(
       function: 'update_settings',
       settings: { extensions: { blocked: [@options[:extension]] } },
       restart:  true
-    }
+    )
   end
 
   def validate
@@ -45,6 +46,13 @@ def validate
 RSpec.describe 'ClusterKillswitch message' do
   subject(:msg) { CLUSTER_KILLSWITCH_TEST_CLASS.new(extension: 'my-ext') }
 
+  before do
+    allow(Legion::Settings).to receive(:dig).and_return(nil)
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return({})
+    allow(Legion::Settings).to receive(:dig).with(:cluster, :control_secret).and_return('shared-secret')
+    allow(Legion::Settings).to receive(:[]).with(:client).and_return({ name: 'sender-node' })
+  end
+
   describe 'routing key constant' do
     it 'is settings.extensions.blocked' do
       expect(CLUSTER_KILLSWITCH_ROUTING_KEY).to eq('settings.extensions.blocked')
@@ -87,6 +95,27 @@ def validate
     it 'wraps extension in a blocked array' do
       expect(msg.message[:settings][:extensions][:blocked]).to eq(['my-ext'])
     end
+
+    it 'includes a verifiable control signature envelope' do
+      payload = msg.message
+
+      expect(payload[:control]).to include(:sender, :timestamp, :nonce, :signature)
+      expect(Legion::Extensions::Node::ControlAuth.verify!(payload)).to eq(payload)
+    end
+
+    it 'rejects tampered signed payloads' do
+      payload = msg.message
+      payload[:settings] = { extensions: { blocked: ['other-ext'] } }
+
+      expect { Legion::Extensions::Node::ControlAuth.verify!(payload) }
+        .to raise_error(Legion::Extensions::Node::ControlAuth::UnauthorizedControlMessage)
+    end
+
+    it 'omits the control signature envelope in auto mode when no secret is configured' do
+      allow(Legion::Settings).to receive(:dig).with(:cluster, :control_secret).and_return(nil)
+
+      expect(msg.message).not_to include(:control)
+    end
   end
 
   describe '#validate' do
diff --git a/spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb b/spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb
index 2e84130..add950b 100644
--- a/spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb
+++ b/spec/legion/extensions/node/transport/messages/cluster_settings_spec.rb
@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'spec_helper'
+require 'legion/extensions/node/control_auth'
 require 'legion/extensions/node/transport/exchanges/cluster_control'
 
 # Mirror ClusterSettings message logic in a proxy class to avoid superclass mismatch.
@@ -26,11 +27,11 @@ def encrypt?
   end
 
   def message
-    {
+    Legion::Extensions::Node::ControlAuth.sign(
       function: 'update_settings',
       settings: @options[:settings],
       restart:  @options.fetch(:restart, false)
-    }
+    )
   end
 
   def validate
@@ -45,6 +46,13 @@ def validate
 
   subject(:msg) { CLUSTER_SETTINGS_TEST_CLASS.new(settings: settings_hash) }
 
+  before do
+    allow(Legion::Settings).to receive(:dig).and_return(nil)
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return({})
+    allow(Legion::Settings).to receive(:dig).with(:cluster, :control_secret).and_return('shared-secret')
+    allow(Legion::Settings).to receive(:[]).with(:client).and_return({ name: 'sender-node' })
+  end
+
   describe '#routing_key' do
     it 'defaults to "settings"' do
       expect(msg.routing_key).to eq('settings')
@@ -91,6 +99,29 @@ def validate
       m = CLUSTER_SETTINGS_TEST_CLASS.new(settings: settings_hash, restart: true)
       expect(m.message[:restart]).to be true
     end
+
+    it 'includes a verifiable control signature envelope' do
+      payload = msg.message
+
+      expect(payload[:control]).to include(:sender, :timestamp, :nonce, :signature)
+      expect(Legion::Extensions::Node::ControlAuth.verify!(payload)).to eq(payload)
+    end
+
+    it 'changes the signature when the settings payload changes' do
+      allow(SecureRandom).to receive(:hex).with(16).and_return('a' * 32)
+      allow(Time).to receive(:now).and_return(Time.utc(2026, 1, 1))
+
+      first = CLUSTER_SETTINGS_TEST_CLASS.new(settings: { feature: true }).message
+      second = CLUSTER_SETTINGS_TEST_CLASS.new(settings: { feature: false }).message
+
+      expect(first[:control][:signature]).not_to eq(second[:control][:signature])
+    end
+
+    it 'omits the control signature envelope in auto mode when no secret is configured' do
+      allow(Legion::Settings).to receive(:dig).with(:cluster, :control_secret).and_return(nil)
+
+      expect(msg.message).not_to include(:control)
+    end
   end
 
   describe '#validate' do
diff --git a/spec/legion/extensions/node/transport/queues/cluster_control_spec.rb b/spec/legion/extensions/node/transport/queues/cluster_control_spec.rb
index 0fb102b..2cdeebe 100644
--- a/spec/legion/extensions/node/transport/queues/cluster_control_spec.rb
+++ b/spec/legion/extensions/node/transport/queues/cluster_control_spec.rb
@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'spec_helper'
+require 'legion/extensions/node/config'
 
 # Mirror ClusterControl queue logic in a proxy class to avoid superclass mismatch
 # with Legion::Transport::Queue which is not available in the test environment.
@@ -10,8 +11,36 @@ def queue_name
   end
 
   def queue_options
-    { durable: false, exclusive: false, auto_delete: true,
-      arguments: { 'x-queue-type': 'classic' } }
+    options = Legion::Extensions::Node::Config.control_queue
+    { durable:     truthy?(options[:durable], default: true),
+      exclusive:   truthy?(options[:exclusive], default: false),
+      auto_delete: truthy?(options[:auto_delete], default: false),
+      arguments:   queue_arguments(options) }
+  end
+
+  private
+
+  def queue_arguments(options)
+    {
+      'x-queue-type':  options[:queue_type],
+      'x-expires':     positive_integer(options[:expires_ms]),
+      'x-message-ttl': positive_integer(options[:message_ttl_ms]),
+      'x-max-length':  positive_integer(options[:max_length])
+    }.compact
+  end
+
+  def positive_integer(value)
+    return value if value.is_a?(Integer) && value.positive?
+    return nil unless value.to_s.match?(/\A\d+\z/)
+
+    parsed = value.to_i
+    parsed.positive? ? parsed : nil
+  end
+
+  def truthy?(value, default:)
+    return default if value.nil?
+
+    value == true || value.to_s == 'true'
   end
 end
 
@@ -19,6 +48,8 @@ def queue_options
   subject(:queue) { CLUSTER_CONTROL_QUEUE_TEST_CLASS.new }
 
   before do
+    allow(Legion::Settings).to receive(:dig).and_return(nil)
+    allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return({})
     allow(Legion::Settings).to receive(:[]).with(:client).and_return({ name: 'test-node' })
   end
 
@@ -29,20 +60,53 @@ def queue_options
   end
 
   describe '#queue_options' do
-    it 'is not durable' do
-      expect(queue.queue_options[:durable]).to be false
+    it 'is durable so offline nodes can receive queued desired-state commands after restart' do
+      expect(queue.queue_options[:durable]).to be true
     end
 
     it 'is not exclusive' do
       expect(queue.queue_options[:exclusive]).to be false
     end
 
-    it 'auto-deletes when node disconnects' do
-      expect(queue.queue_options[:auto_delete]).to be true
+    it 'does not auto-delete when node disconnects' do
+      expect(queue.queue_options[:auto_delete]).to be false
     end
 
     it 'uses classic queue type' do
       expect(queue.queue_options[:arguments][:'x-queue-type']).to eq('classic')
     end
+
+    it 'expires abandoned per-node queues and stale commands' do
+      expect(queue.queue_options[:arguments]).to include(
+        'x-expires':     604_800_000,
+        'x-message-ttl': 86_400_000,
+        'x-max-length':  1000
+      )
+    end
+
+    it 'uses configured queue durability and retention overrides' do
+      allow(Legion::Settings).to receive(:dig).with(:extensions, :node).and_return(
+        cluster_control: {
+          queue: {
+            durable:        false,
+            auto_delete:    true,
+            expires_ms:     nil,
+            message_ttl_ms: 30_000,
+            max_length:     25
+          }
+        }
+      )
+
+      expect(queue.queue_options).to eq(
+        durable:     false,
+        exclusive:   false,
+        auto_delete: true,
+        arguments:   {
+          'x-queue-type':  'classic',
+          'x-message-ttl': 30_000,
+          'x-max-length':  25
+        }
+      )
+    end
   end
 end