diff --git a/extended/docs/proposals/0004-send-message-step-plugin/checklist.ruby.md b/extended/docs/proposals/0004-send-message-step-plugin/checklist.ruby.md
new file mode 100644
index 0000000000..60db08e308
--- /dev/null
+++ b/extended/docs/proposals/0004-send-message-step-plugin/checklist.ruby.md
@@ -0,0 +1,76 @@
+# 0004 Ruby Checklist
+
+## Baseline
+
+- [x] Plugin work lives under `extended/plugins/send-message-ruby/`.
+- [x] The subtree stands alone.
+- [x] No committed Slack credential appears anywhere.
+- [x] The plugin owns Kubernetes reads and Slack calls.
+
+## Phase 0: pin the contract
+
+- [x] Re-read `proposal.md`.
+- [x] Re-read `spec.md`.
+- [x] Re-read the `send-message` slice in proposal `0002`.
+- [x] Keep scope to Slack only.
+
+## Phase 1: create the tiny repo
+
+- [x] Create `extended/plugins/send-message-ruby/`.
+- [x] Add runtime source.
+- [x] Add image build.
+- [x] Add `plugin.yaml`.
+- [x] Add CRD manifests.
+- [x] Add RBAC manifests.
+- [x] Add smoke assets.
+
+## Phase 2: implement the runtime
+
+- [x] Implement `POST /api/v1/step.execute`.
+- [x] Enforce bearer auth from `/var/run/kargo/token`.
+- [x] Read `MessageChannel`.
+- [x] Read `ClusterMessageChannel`.
+- [x] Read referenced `Secret`.
+- [x] Send plaintext Slack payloads.
+- [x] Send encoded Slack payloads.
+- [x] Return `slack.threadTS`.
+
+## Phase 3: test the contract
+
+- [x] Add auth tests.
+- [x] Add channel lookup tests.
+- [x] Add Secret lookup tests.
+- [x] Add plaintext payload tests.
+- [x] Add encoded payload tests.
+- [x] Add XML decode tests.
+- [x] Add Slack failure tests.
+
+## Phase 4: smoke
+
+- [x] Add plugin-owned `smoke/smoke_test.rb`.
+- [x] Keep smoke orchestration in Ruby, not shell.
+- [x] Build the image.
+- [x] Load it into kind.
+- [x] Install CRDs and RBAC.
+- [x] Install StepPlugin `ConfigMap`.
+- [x] Create local-only test Secret.
+- [x] Create test `MessageChannel`.
+- [x] Run a `Stage` with `uses: send-message`.
+- [x] Assert `Succeeded`.
+- [x] Assert non-empty `slack.threadTS`.
+
+## Phase 5: mandatory radical simplification pass 1
+
+- [x] Ask "can I replace this gem with stdlib and make the repo smaller?"
+- [x] Ask "can I collapse this into fewer files and still keep it readable?"
+- [x] Ask "am I carrying Ruby framework habits into a tiny sidecar?"
+- [x] Delete anything that fails those checks.
+
+## Phase 6: mandatory radical simplification pass 2
+
+- [x] Re-run tests and smoke from a green tree.
+- [x] Ask "can I make this radically simpler?"
+- [x] Remove wrappers that only save a few obvious lines.
+- [x] Remove folder structure that makes the plugin look more official than
+      small.
+- [x] Stop only when the repo still reads like a small third-party plugin.
diff --git a/extended/docs/proposals/0004-send-message-step-plugin/plan.ruby.md b/extended/docs/proposals/0004-send-message-step-plugin/plan.ruby.md
new file mode 100644
index 0000000000..9065dcde63
--- /dev/null
+++ b/extended/docs/proposals/0004-send-message-step-plugin/plan.ruby.md
@@ -0,0 +1,93 @@
+# 0004 Ruby Plan
+
+## Goal
+
+- Show the same `send-message` plugin contract in a form that is tiny,
+  readable, and comfortable for a scripting-language user.
+- Keep all plugin-owned code under `extended/plugins/send-message-ruby/`.
+- Keep the subtree standalone enough to behave like its own Git repo.
+- Match the same Slack-only contract as `spec.md`.
+
+## Design Rule
+
+- Prefer stdlib unless a gem removes real complexity.
+- Prefer direct HTTPS to Kubernetes and Slack if that keeps the repo smaller.
+- Do not chase a Ruby-flavored framework stack.
+- If a gem exists only to save a small amount of obvious code, do not use it.
+
+## Runtime Shape
+
+- Runtime:
+  - Ruby
+- Suggested layout:
+  - `extended/plugins/send-message-ruby/`
+  - `server.rb`
+  - `smoke/smoke_test.rb`
+  - `Gemfile`
+  - `Dockerfile`
+  - `plugin.yaml`
+  - `manifests/`
+  - `smoke/`
+- Suggested server shape:
+  - one small HTTP server
+  - one request parser
+  - one Kubernetes reader
+  - one Slack sender
+
+## Minimal Dependency Target
+
+- Strong preference:
+  - stdlib `webrick`
+  - stdlib `json`
+  - stdlib `psych`
+  - stdlib `rexml`
+  - stdlib `net/http`
+- Avoid:
+  - Rails-adjacent stacks
+  - Sinatra unless it clearly removes code
+  - large Kubernetes or Slack client wrappers
+
+## Behavior
+
+- Implement `POST /api/v1/step.execute`.
+- Enforce bearer auth from `/var/run/kargo/token`.
+- Read `MessageChannel` and `ClusterMessageChannel` directly from Kubernetes.
+- Read referenced `Secret` directly from Kubernetes.
+- Send Slack messages directly or through a tiny client if one clearly helps.
+- Support:
+  - plaintext
+  - `json`
+  - `yaml`
+  - `xml`
+- Match the response contract in `spec.md`.
+
+## Tests
+
+- Keep tests inside the subtree.
+- Prefer a small test file and direct fixtures.
+- Cover:
+  - auth
+  - channel lookup
+  - Secret lookup
+  - plaintext payload shaping
+  - encoded payload shaping
+  - XML decode shape
+  - Slack error handling
+
+## Smoke
+
+- Own `smoke/smoke_test.rb` inside the subtree.
+- Assume Kargo already exists.
+- Use Ruby for the smoke orchestration too.
+- Build image, install manifests, create Secret and channel, run a Stage,
+  assert `Succeeded`, assert non-empty `slack.threadTS`.
+
+## Mandatory Simplify Passes
+
+- Simplify pass 1, before full smoke:
+  - ask "can I replace this gem with stdlib and make the repo smaller?"
+  - ask "can I collapse this into fewer files?"
+- Simplify pass 2, after green:
+  - ask "can I make this radically simpler?"
+  - remove any Ruby structure that exists to feel proper rather than to keep
+    the plugin clear
diff --git a/extended/plugins/send-message-ruby/Dockerfile b/extended/plugins/send-message-ruby/Dockerfile
new file mode 100644
index 0000000000..5c45c8a015
--- /dev/null
+++ b/extended/plugins/send-message-ruby/Dockerfile
@@ -0,0 +1,14 @@
+FROM ruby:3.4.7-alpine3.22
+
+RUN gem install --no-document webrick
+
+RUN addgroup -S plugin && adduser -S -G plugin -u 65532 plugin
+
+WORKDIR /app
+
+COPY send_message_plugin.rb /app/send_message_plugin.rb
+COPY lib /app/lib
+
+USER 65532:65532
+
+ENTRYPOINT ["ruby", "/app/send_message_plugin.rb"]
diff --git a/extended/plugins/send-message-ruby/lib/send_message_plugin.rb b/extended/plugins/send-message-ruby/lib/send_message_plugin.rb
new file mode 100644
index 0000000000..a4d2e6792d
--- /dev/null
+++ b/extended/plugins/send-message-ruby/lib/send_message_plugin.rb
@@ -0,0 +1,39 @@
+require "json"
+require "logger"
+require "net/http"
+require "openssl"
+require "psych"
+require "rexml/document"
+require "uri"
+require "webrick"
+
+require_relative "send_message_plugin/errors"
+require_relative "send_message_plugin/kubernetes_client"
+require_relative "send_message_plugin/payload_builder"
+require_relative "send_message_plugin/slack_client"
+require_relative "send_message_plugin/app"
+
+module SendMessagePlugin
+  STEP_EXECUTE_PATH = "/api/v1/step.execute"
+  AUTH_HEADER = "Authorization"
+  BEARER_PREFIX = "Bearer "
+  DEFAULT_TOKEN_PATH = "/var/run/kargo/token"
+  DEFAULT_SYSTEM_RESOURCES_NAMESPACE = "kargo-system-resources"
+  DEFAULT_SLACK_API_BASE_URL = "https://slack.com/api"
+
+  def self.run!
+    app = App.new
+    server = WEBrick::HTTPServer.new(
+      BindAddress: "0.0.0.0",
+      Port: Integer(ENV.fetch("PORT", "9765")),
+      AccessLog: [],
+      Logger: WEBrick::Log.new($stderr, WEBrick::Log::WARN)
+    )
+    server.mount_proc(STEP_EXECUTE_PATH) do |req, res|
+      app.serve(req, res)
+    end
+    trap("INT") { server.shutdown }
+    trap("TERM") { server.shutdown }
+    server.start
+  end
+end
diff --git a/extended/plugins/send-message-ruby/lib/send_message_plugin/app.rb b/extended/plugins/send-message-ruby/lib/send_message_plugin/app.rb
new file mode 100644
index 0000000000..8138d11f55
--- /dev/null
+++ b/extended/plugins/send-message-ruby/lib/send_message_plugin/app.rb
@@ -0,0 +1,175 @@
+module SendMessagePlugin
+  class App
+    def initialize(
+      token_path: DEFAULT_TOKEN_PATH,
+      system_resources_namespace: ENV.fetch(
+        "SYSTEM_RESOURCES_NAMESPACE",
+        DEFAULT_SYSTEM_RESOURCES_NAMESPACE
+      ),
+      slack_api_base_url: ENV.fetch(
+        "SLACK_API_BASE_URL",
+        DEFAULT_SLACK_API_BASE_URL
+      ),
+      kubernetes_client: nil,
+      slack_client: nil,
+      logger: nil
+    )
+      @token_path = token_path
+      @system_resources_namespace = system_resources_namespace
+      @slack_api_base_url = slack_api_base_url
+      @kubernetes_client = kubernetes_client || KubernetesClient.new
+      @slack_client = slack_client || SlackClient.new
+      @logger = logger || Logger.new($stderr, level: Logger::WARN)
+    end
+
+    def call(method:, path:, headers:, body:)
+      return [404, nil] if path != STEP_EXECUTE_PATH
+      return [405, nil] if method != "POST"
+
+      authorize!(headers[AUTH_HEADER] || headers[AUTH_HEADER.downcase])
+      request = JSON.parse(body)
+      [200, execute(request)]
+    rescue AuthError => error
+      [403, errored(error.message, error.message)]
+    rescue JSON::ParserError => error
+      [400, errored("invalid request body", error.message)]
+    rescue RequestError => error
+      [200, errored(error.message, error.message)]
+    rescue ExecutionError => error
+      [200, failed(error.message)]
+    rescue StandardError => error
+      [200, failed(error.message)]
+    end
+
+    def serve(req, res)
+      headers = req.header.each_with_object({}) do |(key, value), memo|
+        memo[key] = Array(value).first
+      end
+      status, response = call(
+        method: req.request_method,
+        path: req.path,
+        headers: headers,
+        body: req.body.to_s
+      )
+      if response.nil?
+        res.status = status
+        return
+      end
+
+      res.status = status
+      res["Content-Type"] = "application/json"
+      res.body = JSON.generate(response)
+    rescue StandardError => error
+      @logger.error(error.full_message)
+      res.status = 500
+      res["Content-Type"] = "application/json"
+      res.body = JSON.generate(failed(error.message))
+    end
+
+    private
+
+    def authorize!(header_value)
+      expected = File.read(@token_path).strip
+      raise AuthError, "missing bearer token" unless header_value&.start_with?(BEARER_PREFIX)
+
+      received = header_value.delete_prefix(BEARER_PREFIX).strip
+      raise AuthError, "invalid bearer token" if received != expected
+    rescue Errno::ENOENT => error
+      raise AuthError, "error reading auth token: #{error.message}"
+    end
+
+    def execute(request)
+      step = fetch_hash(request, "step", "step")
+      config = fetch_hash(step, "config", "step.config")
+      context = optional_hash(request["context"], "context")
+      raise RequestError, "unsupported step kind #{step["kind"].inspect}" unless step["kind"].to_s == "send-message"
+      raise RequestError, "step.config.message is required" if config["message"].nil?
+
+      channel_ref = fetch_hash(config, "channel", "step.config.channel")
+      channel_kind = channel_ref["kind"].to_s.strip
+      channel_name = channel_ref["name"].to_s.strip
+      raise RequestError, "step.config.channel.kind is required" if channel_kind.empty?
+      raise RequestError, "step.config.channel.name is required" if channel_name.empty?
+
+      channel, secret_namespace = lookup_channel(
+        project: context["project"].to_s.strip,
+        channel_kind: channel_kind,
+        channel_name: channel_name
+      )
+      secret = @kubernetes_client.get_secret(secret_namespace, channel.fetch("secret_name"))
+      token = secret["apiKey"].to_s
+      raise RequestError, 'Slack Secret is missing key "apiKey"' if token.empty?
+
+      payload, output_thread_ts = PayloadBuilder.new(
+        config: config,
+        channel: channel
+      ).build
+
+      slack_response = @slack_client.post_message(
+        api_base_url: @slack_api_base_url,
+        token: token,
+        payload: payload
+      )
+      unless slack_response["ok"]
+        detail = slack_response["error"].to_s
+        detail = "unknown_error" if detail.empty?
+        raise ExecutionError, "Slack API error: #{detail}"
+      end
+
+      output_thread_ts = slack_response["ts"].to_s if output_thread_ts.to_s.empty?
+      {
+        "status" => "Succeeded",
+        "output" => {
+          "slack" => {
+            "threadTS" => output_thread_ts
+          }
+        }
+      }
+    end
+
+    def lookup_channel(project:, channel_kind:, channel_name:)
+      case channel_kind
+      when "MessageChannel"
+        raise RequestError, "step context project is required for MessageChannel" if project.empty?
+
+        [@kubernetes_client.get_message_channel(project, channel_name), project]
+      when "ClusterMessageChannel"
+        [@kubernetes_client.get_cluster_message_channel(channel_name), @system_resources_namespace]
+      else
+        raise RequestError, "unsupported channel kind #{channel_kind.inspect}"
+      end
+    end
+
+    def fetch_hash(object, key, path)
+      value = object[key]
+      optional_hash(value, path).tap do |hash|
+        raise RequestError, "#{path} is required" if hash.empty? && !value.is_a?(Hash)
+      end
+    end
+
+    def optional_hash(value, path)
+      return {} if value.nil?
+      return value if value.is_a?(Hash)
+
+      raise RequestError, "#{path} must be an object"
+    end
+
+    def errored(message, error)
+      {
+        "status" => "Errored",
+        "message" => message,
+        "error" => error,
+        "terminal" => true
+      }
+    end
+
+    def failed(message)
+      {
+        "status" => "Failed",
+        "message" => message,
+        "error" => message,
+        "terminal" => true
+      }
+    end
+  end
+end
diff --git a/extended/plugins/send-message-ruby/lib/send_message_plugin/errors.rb b/extended/plugins/send-message-ruby/lib/send_message_plugin/errors.rb
new file mode 100644
index 0000000000..393ff1a2cf
--- /dev/null
+++ b/extended/plugins/send-message-ruby/lib/send_message_plugin/errors.rb
@@ -0,0 +1,10 @@
+module SendMessagePlugin
+  class RequestError < StandardError
+  end
+
+  class AuthError < RequestError
+  end
+
+  class ExecutionError < StandardError
+  end
+end
diff --git a/extended/plugins/send-message-ruby/lib/send_message_plugin/kubernetes_client.rb b/extended/plugins/send-message-ruby/lib/send_message_plugin/kubernetes_client.rb
new file mode 100644
index 0000000000..58a358f31e
--- /dev/null
+++ b/extended/plugins/send-message-ruby/lib/send_message_plugin/kubernetes_client.rb
@@ -0,0 +1,93 @@
+module SendMessagePlugin
+  class KubernetesClient
+    DEFAULT_KUBERNETES_HOST = "kubernetes.default.svc"
+    DEFAULT_KUBERNETES_PORT = "443"
+    DEFAULT_SERVICE_ACCOUNT_DIR = "/var/run/secrets/kubernetes.io/serviceaccount"
+
+    def initialize(
+      host: ENV.fetch("KUBERNETES_SERVICE_HOST", DEFAULT_KUBERNETES_HOST),
+      port: ENV.fetch("KUBERNETES_SERVICE_PORT_HTTPS", DEFAULT_KUBERNETES_PORT),
+      token_path: File.join(DEFAULT_SERVICE_ACCOUNT_DIR, "token"),
+      ca_path: File.join(DEFAULT_SERVICE_ACCOUNT_DIR, "ca.crt")
+    )
+      @base_uri = URI("https://#{host}:#{port}")
+      @token_path = token_path
+      @ca_path = ca_path
+    end
+
+    def get_message_channel(namespace, name)
+      payload = get_json("/apis/ee.kargo.akuity.io/v1alpha1/namespaces/#{namespace}/messagechannels/#{name}")
+      build_channel(payload)
+    rescue RequestError
+      raise
+    rescue StandardError => error
+      raise ExecutionError, "error getting MessageChannel: #{error.message}"
+    end
+
+    def get_cluster_message_channel(name)
+      payload = get_json("/apis/ee.kargo.akuity.io/v1alpha1/clustermessagechannels/#{name}")
+      build_channel(payload)
+    rescue RequestError
+      raise
+    rescue StandardError => error
+      raise ExecutionError, "error getting ClusterMessageChannel: #{error.message}"
+    end
+
+    def get_secret(namespace, name)
+      payload = get_json("/api/v1/namespaces/#{namespace}/secrets/#{name}")
+      decode_secret(payload)
+    rescue StandardError => error
+      raise ExecutionError, "error getting Slack Secret: #{error.message}"
+    end
+
+    private
+
+    def get_json(path)
+      request = Net::HTTP::Get.new(path)
+      request[AUTH_HEADER] = "#{BEARER_PREFIX}#{File.read(@token_path).strip}"
+
+      response = http_client.start do |http|
+        http.request(request)
+      end
+
+      parsed = response.body.to_s.empty? ? {} : JSON.parse(response.body)
+      return parsed if response.code.to_i < 400
+
+      detail = parsed["message"].to_s
+      detail = response.message if detail.empty?
+      raise ExecutionError, detail.empty? ? "HTTP #{response.code}" : detail
+    rescue Errno::ENOENT => error
+      raise ExecutionError, error.message
+    end
+
+    def http_client
+      @http_client ||= Net::HTTP.new(@base_uri.host, @base_uri.port).tap do |http|
+        http.use_ssl = true
+        http.ca_file = @ca_path if File.exist?(@ca_path)
+        http.verify_mode = OpenSSL::SSL::VERIFY_PEER
+      end
+    end
+
+    def build_channel(payload)
+      {
+        "secret_name" => payload.dig("spec", "secretRef", "name").to_s.strip,
+        "slack_channel_id" => payload.dig("spec", "slack", "channelID").to_s.strip,
+        "resource_kind" => payload["kind"].to_s,
+        "resource_name" => payload.dig("metadata", "name").to_s
+      }.tap do |channel|
+        if channel["secret_name"].empty?
+          raise RequestError, %(#{channel["resource_kind"]} "#{channel["resource_name"]}" does not define spec.secretRef.name)
+        end
+      end
+    end
+
+    def decode_secret(payload)
+      data = payload["data"]
+      return {} unless data.is_a?(Hash)
+
+      data.each_with_object({}) do |(key, value), decoded|
+        decoded[key] = value.nil? ? "" : value.unpack1("m0")
+      end
+    end
+  end
+end
diff --git a/extended/plugins/send-message-ruby/lib/send_message_plugin/payload_builder.rb b/extended/plugins/send-message-ruby/lib/send_message_plugin/payload_builder.rb
new file mode 100644
index 0000000000..a761dd1fab
--- /dev/null
+++ b/extended/plugins/send-message-ruby/lib/send_message_plugin/payload_builder.rb
@@ -0,0 +1,124 @@
+module SendMessagePlugin
+  class PayloadBuilder
+    def initialize(config:, channel:)
+      @config = config
+      @channel = channel
+    end
+
+    def build
+      encoding_type = @config["encodingType"].to_s.strip
+      return build_plaintext if encoding_type.empty?
+
+      payload = decode_encoded_payload(encoding_type)
+      raise RequestError, "Slack payload must decode to an object" unless payload.is_a?(Hash)
+
+      if payload["channel"].nil?
+        raise RequestError, missing_encoded_channel_id_message if @channel["slack_channel_id"].to_s.empty?
+
+        payload["channel"] = @channel["slack_channel_id"]
+      end
+      [payload, payload["thread_ts"].to_s]
+    end
+
+    private
+
+    def build_plaintext
+      slack_options = optional_hash(@config["slack"], "step.config.slack")
+      channel_id = slack_options["channelID"].to_s.strip
+      channel_id = @channel["slack_channel_id"] if channel_id.empty?
+      raise RequestError, missing_plaintext_channel_id_message if channel_id.to_s.empty?
+
+      payload = {
+        "channel" => channel_id,
+        "text" => @config["message"]
+      }
+      thread_ts = slack_options["threadTS"].to_s.strip
+      payload["thread_ts"] = thread_ts unless thread_ts.empty?
+      [payload, thread_ts]
+    end
+
+    def decode_encoded_payload(encoding_type)
+      case encoding_type
+      when "json"
+        JSON.parse(@config["message"])
+      when "yaml"
+        Psych.safe_load(@config["message"], aliases: false)
+      when "xml"
+        decode_xml_payload(@config["message"])
+      else
+        raise RequestError, "unsupported encodingType #{encoding_type.inspect}"
+      end
+    rescue JSON::ParserError, Psych::SyntaxError, REXML::ParseException => error
+      raise RequestError, "error decoding #{encoding_type.upcase} Slack payload: #{error.message}"
+    end
+
+    def decode_xml_payload(message)
+      document = REXML::Document.new(message)
+      root = document.root
+      raise RequestError, "error decoding XML Slack payload: empty document" if root.nil?
+
+      xml_root_to_hash(root)
+    end
+
+    def xml_root_to_hash(element)
+      payload = {}
+      element.attributes.each_attribute do |attribute|
+        payload[attribute.expanded_name] = attribute.value
+      end
+      element.elements.each do |child|
+        append_xml_value(payload, child.name, xml_element_value(child))
+      end
+
+      text = direct_text(element)
+      return payload if text.empty?
+
+      if payload.empty?
+        payload["text"] = text
+      else
+        payload["#text"] = text
+      end
+      payload
+    end
+
+    def xml_element_value(element)
+      return direct_text(element) if element.attributes.empty? && element.elements.empty?
+
+      payload = {}
+      element.attributes.each_attribute do |attribute|
+        payload[attribute.expanded_name] = attribute.value
+      end
+      element.elements.each do |child|
+        append_xml_value(payload, child.name, xml_element_value(child))
+      end
+      text = direct_text(element)
+      payload["#text"] = text unless text.empty?
+      payload
+    end
+
+    def direct_text(element)
+      element.children.grep(REXML::Text).map(&:value).join.strip
+    end
+
+    def append_xml_value(payload, key, value)
+      return payload[key] = value unless payload.key?(key)
+
+      existing = payload[key]
+      payload[key] = existing.is_a?(Array) ? existing + [value] : [existing, value]
+    end
+
+    def optional_hash(value, path)
+      return {} if value.nil?
+      return value if value.is_a?(Hash)
+
+      raise RequestError, "#{path} must be an object"
+    end
+
+    def missing_plaintext_channel_id_message
+      %(#{@channel["resource_kind"]} "#{@channel["resource_name"]}" does not define spec.slack.channelID and config.slack.channelID is empty)
+    end
+
+    def missing_encoded_channel_id_message
+      %(#{@channel["resource_kind"]} "#{@channel["resource_name"]}" does not define spec.slack.channelID)
+    end
+  end
+end
diff --git a/extended/plugins/send-message-ruby/lib/send_message_plugin/slack_client.rb b/extended/plugins/send-message-ruby/lib/send_message_plugin/slack_client.rb
new file mode 100644
index 0000000000..28e7e958a8
--- /dev/null
+++ b/extended/plugins/send-message-ruby/lib/send_message_plugin/slack_client.rb
@@ -0,0 +1,22 @@
+module SendMessagePlugin
+  class SlackClient
+    def post_message(api_base_url:, token:, payload:)
+      uri = URI("#{api_base_url.sub(%r{/\z}, "")}/chat.postMessage")
+      request = Net::HTTP::Post.new(uri)
+      request[AUTH_HEADER] = "#{BEARER_PREFIX}#{token}"
+      request["Content-Type"] = "application/json"
+      request.body = JSON.generate(payload)
+
+      response = Net::HTTP.start(uri.host, uri.port, use_ssl: uri.scheme == "https") do |http|
+        http.request(request)
+      end
+
+      parsed = response.body.to_s.empty? ? {} : JSON.parse(response.body)
+      if response.code.to_i >= 400
+        parsed["ok"] = false
+        parsed["error"] = response.message if parsed["error"].to_s.empty?
+      end
+      parsed
+    end
+  end
+end
diff --git a/extended/plugins/send-message-ruby/manifests/crds.yaml b/extended/plugins/send-message-ruby/manifests/crds.yaml
new file mode 100644
index 0000000000..008f9931f8
--- /dev/null
+++ b/extended/plugins/send-message-ruby/manifests/crds.yaml
@@ -0,0 +1,81 @@
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: messagechannels.ee.kargo.akuity.io
+spec:
+  group: ee.kargo.akuity.io
+  names:
+    kind: MessageChannel
+    plural: messagechannels
+    singular: messagechannel
+    listKind: MessageChannelList
+  scope: Namespaced
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        properties:
+          spec:
+            type: object
+            properties:
+              secretRef:
+                type: object
+                properties:
+                  name:
+                    type: string
+                required:
+                - name
+              slack:
+                type: object
+                properties:
+                  channelID:
+                    type: string
+                required:
+                - channelID
+            required:
+            - secretRef
+            - slack
+---
+apiVersion: apiextensions.k8s.io/v1
+kind: CustomResourceDefinition
+metadata:
+  name: clustermessagechannels.ee.kargo.akuity.io
+spec:
+  group: ee.kargo.akuity.io
+  names:
+    kind: ClusterMessageChannel
+    plural: clustermessagechannels
+    singular: clustermessagechannel
+    listKind: ClusterMessageChannelList
+  scope: Cluster
+  versions:
+  - name: v1alpha1
+    served: true
+    storage: true
+    schema:
+      openAPIV3Schema:
+        type: object
+        properties:
+          spec:
+            type: object
+            properties:
+              secretRef:
+                type: object
+                properties:
+                  name:
+                    type: string
+                required:
+                - name
+              slack:
+                type: object
+                properties:
+                  channelID:
+                    type: string
+                required:
+                - channelID
+            required:
+            - secretRef
+            - slack
diff --git a/extended/plugins/send-message-ruby/manifests/rbac.yaml b/extended/plugins/send-message-ruby/manifests/rbac.yaml
new file mode 100644
index 0000000000..df10af6b7f
--- /dev/null
+++ b/extended/plugins/send-message-ruby/manifests/rbac.yaml
@@ -0,0 +1,22 @@
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: send-message-step-plugin-reader
+rules:
+- apiGroups:
+  - ee.kargo.akuity.io
+  resources:
+  - messagechannels
+  - clustermessagechannels
+  verbs:
+  - get
+  - list
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - secrets
+  verbs:
+  - get
+  - list
+  - watch
diff --git a/extended/plugins/send-message-ruby/plugin.yaml b/extended/plugins/send-message-ruby/plugin.yaml
new file mode 100644
index 0000000000..e7f79935a1
--- /dev/null
+++ b/extended/plugins/send-message-ruby/plugin.yaml
@@ -0,0 +1,34 @@
+apiVersion: kargo-extended.code.org/v1alpha1
+kind: StepPlugin
+metadata:
+  name: send-message
+  namespace: kargo-system-resources
+spec:
+  sidecar:
+    automountServiceAccountToken: true
+    container:
+      name: send-message-step-plugin
+      image: send-message-step-plugin-ruby:dev
+      imagePullPolicy: IfNotPresent
+      env:
+      - name: SYSTEM_RESOURCES_NAMESPACE
+        value: kargo-system-resources
+      ports:
+      - containerPort: 9765
+      securityContext:
+        runAsNonRoot: true
+        runAsUser: 65532
+        allowPrivilegeEscalation: false
+        readOnlyRootFilesystem: true
+        capabilities:
+          drop:
+          - ALL
+      resources:
+        requests:
+          cpu: 50m
+          memory: 64Mi
+        limits:
+          cpu: 250m
+          memory: 128Mi
+  steps:
+  - kind: send-message
diff --git a/extended/plugins/send-message-ruby/send_message_plugin.rb b/extended/plugins/send-message-ruby/send_message_plugin.rb
new file mode 100644
index 0000000000..0272b466da
--- /dev/null
+++ b/extended/plugins/send-message-ruby/send_message_plugin.rb
@@ -0,0 +1,5 @@
+#!/usr/bin/env ruby
+
+require_relative "lib/send_message_plugin"
+
+SendMessagePlugin.run! if $PROGRAM_NAME == __FILE__
diff --git a/extended/plugins/send-message-ruby/smoke/README.md b/extended/plugins/send-message-ruby/smoke/README.md
new file mode 100644
index 0000000000..83cc923729
--- /dev/null
+++ b/extended/plugins/send-message-ruby/smoke/README.md
@@ -0,0 +1,23 @@
+# Local Smoke Notes
+
+- Primary smoke entrypoint:
+  - `extended/plugins/send-message-ruby/smoke/smoke_test.rb`
+- The script assumes a working Kargo cluster and CLI are already available.
+- Required env:
+  - `SEND_MESSAGE_SMOKE_PROJECT`
+  - `SEND_MESSAGE_SMOKE_WAREHOUSE`
+  - `SEND_MESSAGE_SMOKE_FREIGHT_NAME`
+  - `SEND_MESSAGE_SMOKE_SLACK_API_KEY`
+  - `SEND_MESSAGE_SMOKE_CHANNEL_ID`
+- Optional env:
+  - `KARGO_BIN`
+  - `KARGO_FLAGS`
+  - `SEND_MESSAGE_SMOKE_SYSTEM_RESOURCES_NAMESPACE`
+  - `SEND_MESSAGE_SMOKE_SECRET_NAME`
+  - `SEND_MESSAGE_SMOKE_CHANNEL_NAME`
+  - `SEND_MESSAGE_SMOKE_CONFIGMAP_NAME`
+  - `KUBECTL_BIN`
+  - `DOCKER_BIN`
+  - `KIND_BIN`
+- No committed file in this subtree contains a real token value or local token
+  source.
diff --git a/extended/plugins/send-message-ruby/smoke/smoke_runner.rb b/extended/plugins/send-message-ruby/smoke/smoke_runner.rb
new file mode 100644
index 0000000000..1e968dacd7
--- /dev/null
+++ b/extended/plugins/send-message-ruby/smoke/smoke_runner.rb
@@ -0,0 +1,452 @@
+require "fileutils"
+require "json"
+require "open3"
+require "tmpdir"
+require "yaml"
+
+module SendMessageRubySmoke
+  class SmokeFailure < StandardError
+  end
+
+  class Shell
+    def capture(*command, chdir: nil)
+      stdout, stderr, status =
+        if chdir
+          Open3.capture3(*command, chdir: chdir)
+        else
+          Open3.capture3(*command)
+        end
+      return stdout if status.success?
+
+      raise SmokeFailure, "#{command.join(' ')} failed\n#{stdout}#{stderr}"
+    end
+
+    def run_command(*command, chdir: nil)
+      output = capture(*command, chdir: chdir)
+      print output unless output.empty?
+      true
+    end
+  end
+
+  class SmokeRunner
+    REQUIRED_ENV = %w[
+      SEND_MESSAGE_SMOKE_PROJECT
+      SEND_MESSAGE_SMOKE_WAREHOUSE
+      SEND_MESSAGE_SMOKE_FREIGHT_NAME
+      SEND_MESSAGE_SMOKE_SLACK_API_KEY
+      SEND_MESSAGE_SMOKE_CHANNEL_ID
+    ].freeze
+
+    def initialize(shell: Shell.new)
+      @shell = shell
+      @plugin_dir = File.expand_path("..", __dir__)
+      @kargo_bin = ENV.fetch("KARGO_BIN", "kargo")
+      @kargo_flags = ENV.fetch("KARGO_FLAGS", "")
+      @kubectl_bin = ENV.fetch("KUBECTL_BIN", "kubectl")
+      @docker_bin = ENV.fetch("DOCKER_BIN", "docker")
+      @kind_bin = ENV.fetch("KIND_BIN", "kind")
+      @system_resources_namespace = ENV.fetch(
+        "SEND_MESSAGE_SMOKE_SYSTEM_RESOURCES_NAMESPACE",
+        "kargo-system-resources"
+      )
+      @secret_name = ENV.fetch(
+        "SEND_MESSAGE_SMOKE_SECRET_NAME",
+        "send-message-slack-token"
+      )
+      @channel_name = ENV.fetch(
+        "SEND_MESSAGE_SMOKE_CHANNEL_NAME",
+        "send-message-smoke"
+      )
+      @configmap_name = ENV.fetch(
+        "SEND_MESSAGE_SMOKE_CONFIGMAP_NAME",
+        "send-message-step-plugin"
+      )
+      @cluster_role_name = "send-message-step-plugin-reader"
+      @project = ENV["SEND_MESSAGE_SMOKE_PROJECT"]
+      @warehouse = ENV["SEND_MESSAGE_SMOKE_WAREHOUSE"]
+      @freight_name = ENV["SEND_MESSAGE_SMOKE_FREIGHT_NAME"]
+      @slack_api_key = ENV["SEND_MESSAGE_SMOKE_SLACK_API_KEY"]
+      @slack_channel_id = ENV["SEND_MESSAGE_SMOKE_CHANNEL_ID"]
+      @tmp_dir = nil
+      @stage_name = nil
+      @promotion_name = nil
+      @cluster_role_binding_name = nil
+    end
+
+    def run
+      require_env!
+      ensure_kind_context!
+      build_and_load_image
+      render_plugin_dir
+      build_plugin_configmap
+      install_manifests
+      create_test_resources
+      run_stage
+      poll_promotion
+      pass("send-message StepPlugin smoke promotion finished successfully")
+    ensure
+      cleanup
+    end
+
+    private
+
+    def require_env!
+      missing = REQUIRED_ENV.select { |name| ENV[name].to_s.empty? }
+      return if missing.empty?
+
+      noun = missing.length == 1 ? "is" : "are"
+      raise SmokeFailure, "#{missing.join(', ')} #{noun} required"
+    end
+
+    def ensure_kind_context!
+      context = capture(@kubectl_bin, "config", "current-context").strip
+      return if context.start_with?("kind-")
+
+      raise SmokeFailure, "send-message smoke requires a kind context, got: #{context}"
+    end
+
+    def build_and_load_image
+      kind_name = capture(@kubectl_bin, "config", "current-context").strip.delete_prefix("kind-")
+      @image_tag = "send-message-step-plugin-ruby:e2e-#{Time.now.to_i}"
+
+      info("Build send-message StepPlugin image")
+      run_command(@docker_bin, "build", "-t", @image_tag, @plugin_dir)
+      pass("Build send-message StepPlugin image")
+
+      info("Load send-message StepPlugin image into kind")
+      run_command(@kind_bin, "load", "docker-image", "--name", kind_name, @image_tag)
+      pass("Load send-message StepPlugin image into kind")
+    end
+
+    def render_plugin_dir
+      @tmp_dir = Dir.mktmpdir("send-message-stepplugin-ruby-")
+      plugin_yaml = File.read(File.join(@plugin_dir, "plugin.yaml"))
+      plugin_yaml = plugin_yaml.gsub("namespace: kargo-system-resources", "namespace: #{@system_resources_namespace}")
+      plugin_yaml = plugin_yaml.gsub("image: send-message-step-plugin-ruby:dev", "image: #{@image_tag}")
+      plugin_yaml = plugin_yaml.gsub("value: kargo-system-resources", "value: #{@system_resources_namespace}")
+      File.write(File.join(@tmp_dir, "plugin.yaml"), plugin_yaml)
+    end
+
+    def build_plugin_configmap
+      info("Build send-message StepPlugin ConfigMap")
+      run_command(@kargo_bin, "step-plugin", "build", ".", chdir: @tmp_dir)
+      pass("Build send-message StepPlugin ConfigMap")
+    end
+
+    def install_manifests
+      info("Install send-message CRDs")
+      run_command(@kubectl_bin, "apply", "-f", File.join(@plugin_dir, "manifests", "crds.yaml"))
+      pass("Install send-message CRDs")
+
+      info("Install send-message ClusterRole")
+      run_command(@kubectl_bin, "apply", "-f", File.join(@plugin_dir, "manifests", "rbac.yaml"))
+      pass("Install send-message ClusterRole")
+
+      @cluster_role_binding_name = "send-message-step-plugin-reader-#{@project}"
+      binding_path = File.join(@tmp_dir, "#{@cluster_role_binding_name}.yaml")
+      File.write(binding_path, YAML.dump(cluster_role_binding))
+      info("Bind send-message ClusterRole to test project default ServiceAccount")
+      run_command(@kubectl_bin, "apply", "-f", binding_path)
+      pass("Bind send-message ClusterRole to test project default ServiceAccount")
+
+      info("Install send-message StepPlugin ConfigMap in system resources namespace")
+      run_command(
+        @kubectl_bin,
+        "apply",
+        "-f",
+        File.join(@tmp_dir, "#{@configmap_name}-configmap.yaml")
+      )
+      pass("Install send-message StepPlugin ConfigMap in system resources namespace")
+    end
+
+    def create_test_resources
+      info("Create send-message Slack Secret")
+      run_command(
+        @kubectl_bin,
+        "apply",
+        "-f",
+        write_yaml("#{@secret_name}.yaml", secret_manifest)
+      )
+      pass("Create send-message Slack Secret")
+
+      info("Create send-message MessageChannel")
+      run_command(
+        @kubectl_bin,
+        "apply",
+        "-f",
+        write_yaml("#{@channel_name}.yaml", channel_manifest)
+      )
+      pass("Create send-message MessageChannel")
+    end
+
+    def run_stage
+      @stage_name = "smsgrb-#{Time.now.to_i}"
+      info("Create send-message StepPlugin smoke stage")
+      output = capture(
+        @kargo_bin,
+        "apply",
+        "-f",
+        write_yaml("#{@stage_name}.yaml", stage_manifest),
+        *split_flags
+      )
+      puts output
+      unless output.include?("stage.kargo.akuity.io/#{@stage_name}")
+        raise SmokeFailure, "send-message smoke stage apply output did not mention #{@stage_name}"
+      end
+      pass("Create send-message StepPlugin smoke stage")
+
+      info("Approve freight for send-message StepPlugin smoke stage")
+      run_command(
+        @kargo_bin,
+        "approve",
+        "--project=#{@project}",
+        "--freight=#{@freight_name}",
+        "--stage=#{@stage_name}",
+        *split_flags
+      )
+      pass("Approve freight for send-message StepPlugin smoke stage")
+
+      @promotion_name = "#{@stage_name}.manual"
+      info("Create send-message StepPlugin smoke promotion")
+      run_command(
+        @kubectl_bin,
+        "apply",
+        "-f",
+        write_yaml("#{@promotion_name}.yaml", promotion_manifest)
+      )
+      pass("Create send-message StepPlugin smoke promotion")
+    end
+
+    def poll_promotion
+      90.times do
+        promotion_resource = fetch_promotion
+        phase = promotion_resource.dig("status", "phase").to_s
+        case phase
+        when "Succeeded"
+          thread_ts = promotion_resource.dig("status", "stepExecutionMetadata", 0, "output", "slack", "threadTS")
+          thread_ts ||= promotion_resource.dig("status", "state", "step-1", "slack", "threadTS")
+          return unless thread_ts.to_s.empty?
+
+          raise SmokeFailure, "send-message smoke did not produce slack.threadTS output"
+        when "Failed", "Errored", "Aborted"
+          dump = capture(
+            @kubectl_bin,
+            "get",
+            "promotion.kargo.akuity.io",
+            promotion_resource.dig("metadata", "name"),
+            "-n",
+            @project,
+            "-o",
+            "yaml"
+          )
+          raise SmokeFailure, "send-message smoke promotion reached terminal phase #{phase}\n#{dump}"
+        end
+        sleep 2
+      end
+
+      dump = capture(
+        @kubectl_bin,
+        "get",
+        "promotion.kargo.akuity.io",
+        "-n",
+        @project,
+        "-o",
+        "yaml"
+      )
+      raise SmokeFailure, "send-message smoke promotion did not succeed in time\n#{dump}"
+    end
+
+    def fetch_promotion
+      return {} unless @promotion_name
+
+      payload = capture(
+        @kubectl_bin,
+        "get",
+        "promotion.kargo.akuity.io",
+        @promotion_name,
+        "-n",
+        @project,
+        "-o",
+        "json"
+      )
+      JSON.parse(payload)
+    rescue JSON::ParserError
+      {}
+    rescue SmokeFailure
+      {}
+    end
+
+    def cleanup
+      delete("stage.kargo.akuity.io", @stage_name, namespace: @project) if @stage_name
+      delete("promotion.kargo.akuity.io", @promotion_name, namespace: @project) if @promotion_name
+      delete("messagechannel.ee.kargo.akuity.io", @channel_name, namespace: @project)
+      delete("secret", @secret_name, namespace: @project)
+      delete("configmap", @configmap_name, namespace: @system_resources_namespace)
+      delete("clusterrolebinding", @cluster_role_binding_name) if @cluster_role_binding_name
+      delete("clusterrole", @cluster_role_name)
+      run_command(
+        @kubectl_bin,
+        "delete",
+        "-f",
+        File.join(@plugin_dir, "manifests", "crds.yaml"),
+        "--ignore-not-found"
+      )
+    rescue StandardError
+      nil
+    ensure
+      FileUtils.remove_entry(@tmp_dir) if @tmp_dir && File.exist?(@tmp_dir)
+    end
+
+    def delete(kind, name, namespace: nil)
+      return if name.to_s.empty?
+
+      command = [@kubectl_bin, "delete", kind, name]
+      command += ["-n", namespace] if namespace
+      command << "--ignore-not-found"
+      run_command(*command)
+    end
+
+    def cluster_role_binding
+      {
+        "apiVersion" => "rbac.authorization.k8s.io/v1",
+        "kind" => "ClusterRoleBinding",
+        "metadata" => {
+          "name" => @cluster_role_binding_name
+        },
+        "subjects" => [
+          {
+            "kind" => "ServiceAccount",
+            "name" => "default",
+            "namespace" => @project
+          }
+        ],
+        "roleRef" => {
+          "apiGroup" => "rbac.authorization.k8s.io",
+          "kind" => "ClusterRole",
+          "name" => @cluster_role_name
+        }
+      }
+    end
+
+    def secret_manifest
+      {
+        "apiVersion" => "v1",
+        "kind" => "Secret",
+        "metadata" => {
+          "name" => @secret_name,
+          "namespace" => @project
+        },
+        "type" => "Opaque",
+        "stringData" => {
+          "apiKey" => @slack_api_key
+        }
+      }
+    end
+
+    def channel_manifest
+      {
+        "apiVersion" => "ee.kargo.akuity.io/v1alpha1",
+        "kind" => "MessageChannel",
+        "metadata" => {
+          "name" => @channel_name,
+          "namespace" => @project
+        },
+        "spec" => {
+          "secretRef" => {
+            "name" => @secret_name
+          },
+          "slack" => {
+            "channelID" => @slack_channel_id
+          }
+        }
+      }
+    end
+
+    def stage_manifest
+      {
+        "apiVersion" => "kargo.akuity.io/v1alpha1",
+        "kind" => "Stage",
+        "metadata" => {
+          "name" => @stage_name,
+          "namespace" => @project
+        },
+        "spec" => {
+          "requestedFreight" => [
+            {
+              "origin" => {
+                "kind" => "Warehouse",
+                "name" => @warehouse
+              },
+              "sources" => {
+                "direct" => true
+              }
+            }
+          ],
+          "promotionTemplate" => {
+            "spec" => {
+              "steps" => [
+                {
+                  "uses" => "send-message",
+                  "config" => {
+                    "channel" => {
+                      "kind" => "MessageChannel",
+                      "name" => @channel_name
+                    },
+                    "message" => "send-message StepPlugin smoke from the Ruby implementation #{@stage_name}"
+                  }
+                }
+              ]
+            }
+          }
+        }
+      }
+    end
+
+    def promotion_manifest
+      {
+        "apiVersion" => "kargo.akuity.io/v1alpha1",
+        "kind" => "Promotion",
+        "metadata" => {
+          "name" => @promotion_name,
+          "namespace" => @project
+        },
+        "spec" => {
+          "stage" => @stage_name,
+          "freight" => @freight_name,
+          "steps" => stage_manifest.dig(
+            "spec",
+            "promotionTemplate",
+            "spec",
+            "steps"
+          )
+        }
+      }
+    end
+
+    def split_flags
+      @kargo_flags.split.reject(&:empty?)
+    end
+
+    def write_yaml(name, object)
+      path = File.join(@tmp_dir, name)
+      File.write(path, YAML.dump(object))
+      path
+    end
+
+    def capture(*command, chdir: nil)
+      @shell.capture(*command, chdir: chdir)
+    end
+
+    def run_command(*command, chdir: nil)
+      @shell.run_command(*command, chdir: chdir)
+    end
+
+    def info(message)
+      puts "[INFO] #{message}"
+    end
+
+    def pass(message)
+      puts "[PASS] #{message}"
+    end
+  end
+end
diff --git a/extended/plugins/send-message-ruby/smoke/smoke_test.rb b/extended/plugins/send-message-ruby/smoke/smoke_test.rb
new file mode 100644
index 0000000000..8824df687d
--- /dev/null
+++ b/extended/plugins/send-message-ruby/smoke/smoke_test.rb
@@ -0,0 +1,5 @@
+#!/usr/bin/env ruby
+
+require_relative "smoke_runner"
+
+SendMessageRubySmoke::SmokeRunner.new.run
diff --git a/extended/plugins/send-message-ruby/test/server_test.rb b/extended/plugins/send-message-ruby/test/server_test.rb
new file mode 100644
index 0000000000..3587a35144
--- /dev/null
+++ b/extended/plugins/send-message-ruby/test/server_test.rb
@@ -0,0 +1,501 @@
+require "json"
+require "logger"
+require "minitest/autorun"
+require "tmpdir"
+
+require_relative "../lib/send_message_plugin"
+
+class SendMessagePluginTest < Minitest::Test
+  def test_rejects_missing_bearer_token
+    app = build_app
+
+    status, response = app.call(
+      method: "POST",
+      path: SendMessagePlugin::STEP_EXECUTE_PATH,
+      headers: {},
+      body: JSON.generate(minimal_request)
+    )
+
+    assert_equal 403, status
+    assert_equal "Errored", response["status"]
+    assert_equal "missing bearer token", response["error"]
+  end
+
+  def test_rejects_invalid_bearer_token
+    app = build_app
+
+    status, response = app.call(
+      method: "POST",
+      path: SendMessagePlugin::STEP_EXECUTE_PATH,
+      headers: { "Authorization" => "Bearer wrong-token" },
+      body: JSON.generate(minimal_request)
+    )
+
+    assert_equal 403, status
+    assert_equal "Errored", response["status"]
+    assert_equal "invalid bearer token", response["error"]
+  end
+
+  def test_rejects_invalid_request_body
+    app = build_app
+
+    status, response = app.call(
+      method: "POST",
+      path: SendMessagePlugin::STEP_EXECUTE_PATH,
+      headers: { "Authorization" => "Bearer expected-token" },
+      body: "{not-json"
+    )
+
+    assert_equal 400, status
+    assert_equal "Errored", response["status"]
+    assert_equal "invalid request body", response["message"]
+  end
+
+  def test_uses_namespaced_message_channel
+    kube = FakeKubernetesClient.new(
+      message_channels: {
+        "demo/send" => {
+          "secret_name" => "slack-token",
+          "slack_channel_id" => "C123",
+          "resource_kind" => "MessageChannel",
+          "resource_name" => "send"
+        }
+      },
+      secrets: {
+        "demo/slack-token" => { "apiKey" => "xoxb-demo" }
+      }
+    )
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000100"
+    )
+    app = build_app(kube: kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C123",
+        "text" => "hello from plugin"
+      },
+      slack.last_payload
+    )
+    assert_equal(
+      "1712345678.000100",
+      response.dig("output", "slack", "threadTS")
+    )
+  end
+
+  def test_uses_cluster_message_channel
+    kube = FakeKubernetesClient.new(
+      cluster_message_channels: {
+        "send" => {
+          "secret_name" => "slack-token",
+          "slack_channel_id" => "C777",
+          "resource_kind" => "ClusterMessageChannel",
+          "resource_name" => "send"
+        }
+      },
+      secrets: {
+        "kargo-system-resources/slack-token" => { "apiKey" => "xoxb-demo" }
+      }
+    )
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000200"
+    )
+    app = build_app(kube: kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["channel"] = {
+      "kind" => "ClusterMessageChannel",
+      "name" => "send"
+    }
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C777",
+        "text" => "hello from plugin"
+      },
+      slack.last_payload
+    )
+  end
+
+  def test_plaintext_honors_slack_overrides
+    kube = standard_kube
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000300"
+    )
+    app = build_app(kube: kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["slack"] = {
+      "channelID" => "C999",
+      "threadTS" => "1700000000.000001"
+    }
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C999",
+        "text" => "hello from plugin",
+        "thread_ts" => "1700000000.000001"
+      },
+      slack.last_payload
+    )
+    assert_equal(
+      "1700000000.000001",
+      response.dig("output", "slack", "threadTS")
+    )
+  end
+
+  def test_supports_json_payloads
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000400"
+    )
+    app = build_app(kube: standard_kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["encodingType"] = "json"
+    request["step"]["config"]["message"] = <<~JSON
+      {"text":"rich","blocks":[{"type":"section"}]}
+    JSON
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C123",
+        "text" => "rich",
+        "blocks" => [
+          { "type" => "section" }
+        ]
+      },
+      slack.last_payload
+    )
+  end
+
+  def test_supports_yaml_payloads
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000410"
+    )
+    app = build_app(kube: standard_kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["encodingType"] = "yaml"
+    request["step"]["config"]["message"] = <<~YAML
+      text: rich
+      blocks:
+      - type: section
+    YAML
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C123",
+        "text" => "rich",
+        "blocks" => [
+          { "type" => "section" }
+        ]
+      },
+      slack.last_payload
+    )
+  end
+
+  def test_encoded_payload_ignores_outer_slack_overrides
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000450"
+    )
+    app = build_app(kube: standard_kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["encodingType"] = "json"
+    request["step"]["config"]["message"] = '{"text":"rich","thread_ts":"1700000000.000002"}'
+    request["step"]["config"]["slack"] = {
+      "channelID" => "C999",
+      "threadTS" => "1700000000.000001"
+    }
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C123",
+        "text" => "rich",
+        "thread_ts" => "1700000000.000002"
+      },
+      slack.last_payload
+    )
+    assert_equal(
+      "1700000000.000002",
+      response.dig("output", "slack", "threadTS")
+    )
+  end
+
+  def test_supports_xml_payloads
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000500"
+    )
+    app = build_app(kube: standard_kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["encodingType"] = "xml"
+    request["step"]["config"]["message"] = <<~XML
+      <message>
+        <text>rich</text>
+        <thread_ts>1700000000.000003</thread_ts>
+      </message>
+    XML
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C123",
+        "text" => "rich",
+        "thread_ts" => "1700000000.000003"
+      },
+      slack.last_payload
+    )
+    assert_equal(
+      "1700000000.000003",
+      response.dig("output", "slack", "threadTS")
+    )
+  end
+
+  def test_xml_decode_repeats_siblings_into_arrays
+    slack = FakeSlackClient.new(
+      "ok" => true,
+      "ts" => "1712345678.000510"
+    )
+    app = build_app(kube: standard_kube, slack: slack)
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["encodingType"] = "xml"
+    request["step"]["config"]["message"] = <<~XML
+      <message>
+        <text>rich</text>
+        <blocks>
+          <type>section</type>
+          <text>
+            <type>mrkdwn</type>
+            <text>*hello*</text>
+          </text>
+        </blocks>
+        <blocks>
+          <type>divider</type>
+        </blocks>
+      </message>
+    XML
+    response = execute_request(app, request)
+
+    assert_equal "Succeeded", response["status"]
+    assert_equal(
+      {
+        "channel" => "C123",
+        "text" => "rich",
+        "blocks" => [
+          {
+            "type" => "section",
+            "text" => {
+              "type" => "mrkdwn",
+              "text" => "*hello*"
+            }
+          },
+          {
+            "type" => "divider"
+          }
+        ]
+      },
+      slack.last_payload
+    )
+  end
+
+  def test_secret_lookup_failure_returns_failed
+    kube = FakeKubernetesClient.new(
+      message_channels: {
+        "demo/send" => {
+          "secret_name" => "slack-token",
+          "slack_channel_id" => "C123",
+          "resource_kind" => "MessageChannel",
+          "resource_name" => "send"
+        }
+      }
+    )
+    app = build_app(kube: kube, slack: FakeSlackClient.new("ok" => true, "ts" => "0"))
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    response = execute_request(app, request)
+
+    assert_equal "Failed", response["status"]
+    assert_match(/error getting Slack Secret/, response["error"])
+  end
+
+  def test_slack_rejection_returns_failed
+    app = build_app(
+      kube: standard_kube,
+      slack: FakeSlackClient.new(
+        "ok" => false,
+        "error" => "channel_not_found"
+      )
+    )
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    response = execute_request(app, request)
+
+    assert_equal "Failed", response["status"]
+    assert_equal "Slack API error: channel_not_found", response["error"]
+  end
+
+  def test_bad_yaml_is_errored
+    app = build_app(kube: standard_kube, slack: FakeSlackClient.new("ok" => true, "ts" => "0"))
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    request["step"]["config"]["encodingType"] = "yaml"
+    request["step"]["config"]["message"] = "text: [oops"
+    response = execute_request(app, request)
+
+    assert_equal "Errored", response["status"]
+    assert_match(/error decoding YAML Slack payload/, response["error"])
+  end
+
+  def test_slack_transport_error_returns_failed
+    app = build_app(
+      kube: standard_kube,
+      slack: FakeSlackClient.new(
+        { "ok" => true, "ts" => "0" },
+        error: StandardError.new("dial tcp timeout")
+      )
+    )
+
+    request = minimal_request
+    request["context"] = { "project" => "demo" }
+    response = execute_request(app, request)
+
+    assert_equal "Failed", response["status"]
+    assert_equal "dial tcp timeout", response["error"]
+  end
+
+  private
+
+  def build_app(kube: FakeKubernetesClient.new, slack: FakeSlackClient.new("ok" => true, "ts" => "0"))
+    token_dir = Dir.mktmpdir
+    token_path = File.join(token_dir, "token")
+    File.write(token_path, "expected-token")
+
+    SendMessagePlugin::App.new(
+      token_path: token_path,
+      system_resources_namespace: "kargo-system-resources",
+      kubernetes_client: kube,
+      slack_client: slack,
+      logger: Logger.new(File::NULL)
+    )
+  end
+
+  def execute_request(app, request)
+    status, response = app.call(
+      method: "POST",
+      path: SendMessagePlugin::STEP_EXECUTE_PATH,
+      headers: { "Authorization" => "Bearer expected-token" },
+      body: JSON.generate(request)
+    )
+
+    assert_equal 200, status
+    response
+  end
+
+  def minimal_request
+    {
+      "step" => {
+        "kind" => "send-message",
+        "config" => {
+          "channel" => {
+            "kind" => "MessageChannel",
+            "name" => "send"
+          },
+          "message" => "hello from plugin"
+        }
+      }
+    }
+  end
+
+  def standard_kube
+    FakeKubernetesClient.new(
+      message_channels: {
+        "demo/send" => {
+          "secret_name" => "slack-token",
+          "slack_channel_id" => "C123",
+          "resource_kind" => "MessageChannel",
+          "resource_name" => "send"
+        }
+      },
+      secrets: {
+        "demo/slack-token" => { "apiKey" => "xoxb-demo" }
+      }
+    )
+  end
+end
+
+class FakeKubernetesClient
+  def initialize(message_channels: {}, cluster_message_channels: {}, secrets: {})
+    @message_channels = message_channels
+    @cluster_message_channels = cluster_message_channels
+    @secrets = secrets
+  end
+
+  def get_message_channel(namespace, name)
+    @message_channels.fetch("#{namespace}/#{name}") do
+      raise SendMessagePlugin::ExecutionError, "message channel not found"
+    end
+  end
+
+  def get_cluster_message_channel(name)
+    @cluster_message_channels.fetch(name) do
+      raise SendMessagePlugin::ExecutionError, "cluster message channel not found"
+    end
+  end
+
+  def get_secret(namespace, name)
+    @secrets.fetch("#{namespace}/#{name}") do
+      raise SendMessagePlugin::ExecutionError, "error getting Slack Secret: secret not found"
+    end
+  end
+end
+
+class FakeSlackClient
+  attr_reader :last_payload
+
+  def initialize(response, error: nil)
+    @response = response
+    @error = error
+  end
+
+  def post_message(api_base_url:, token:, payload:)
+    @last_payload = payload
+    raise @error if @error
+
+    @response
+  end
+end