diff --git a/extended/docs/proposals/0004-send-message-step-plugin/implementation_checklist.md b/extended/docs/proposals/0004-send-message-step-plugin/implementation_checklist.md
new file mode 100644
index 0000000000..07027a24b7
--- /dev/null
+++ b/extended/docs/proposals/0004-send-message-step-plugin/implementation_checklist.md
@@ -0,0 +1,94 @@
+# 0004 Implementation Checklist
+
+Update this as implementation teaches us things.
+
+## Baseline
+
+- [x] Plugin work lives under `extended/plugins/send-message`.
+- [x] The plugin subtree does not rely on repo resources outside itself.
+- [x] Slack credentials stay local-only and out of git.
+- [x] The host does not own channel lookup, Secret reads, or Slack API calls.
+
+## Phase 0: read and pin the seams
+
+- [x] Read proposal 0004 again before starting code.
+- [x] Re-read the `send-message` section in proposal 0002 spec.
+- [x] Read the current StepPlugin host runtime and e2e harness seams.
+- [x] Record any host-side gap found during that read.
+
+## Phase 1: create the standalone subtree
+
+- [x] Create `extended/plugins/send-message/`.
+- [x] Add plugin-owned runtime source there.
+- [x] Add plugin-owned image build there.
+- [x] Add plugin-owned tests there.
+- [x] Add plugin-owned CRD manifests there.
+- [x] Add plugin-owned RBAC manifests there.
+- [x] Add plugin-owned smoke assets or smoke helper there.
+
+## Phase 2: implement the runtime
+
+- [x] Implement `POST /api/v1/step.execute`.
+- [x] Enforce bearer-token auth from `/var/run/kargo/token`.
+- [x] Return `403` on bad auth.
+- [x] Reject unsupported methods cleanly.
+- [x] Parse the v1 step-config subset.
+- [x] Resolve `MessageChannel` from the Project namespace.
+- [x] Resolve `ClusterMessageChannel` from the system-resources namespace.
+- [x] Resolve referenced Secrets from the correct namespace.
+- [x] Read Slack token from Secret key `apiKey`.
+- [x] Send the Slack message from the plugin.
+- [x] Return `slack.threadTS` output.
+
+## Phase 3: own the OSS CRDs and RBAC
+
+- [x] Add Slack-only `MessageChannel` CRD manifest.
+- [x] Add Slack-only `ClusterMessageChannel` CRD manifest.
+- [x] Keep the API group `ee.kargo.akuity.io/v1alpha1`.
+- [x] Add plugin-owned RBAC manifests for channel and Secret reads.
+- [x] Keep RBAC setup out of host code unless a real host gap is proven.
+
+## Phase 4: tests inside the subtree
+
+- [x] Add runtime tests for auth.
+- [x] Add runtime tests for namespaced channel lookup.
+- [x] Add runtime tests for cluster-scoped channel lookup.
+- [x] Add runtime tests for referenced Secret lookup.
+- [x] Add runtime tests for Slack request shaping.
+- [x] Add runtime tests for `slack.channelID` override.
+- [x] Add runtime tests for `slack.threadTS` override and output.
+- [x] Add runtime tests for missing channel or Secret failures.
+- [x] Add runtime tests for Slack API failure handling.
+
+## Phase 5: local smoke path
+
+- [x] Build the plugin image from `extended/plugins/send-message`.
+- [x] Load the image into the local kind cluster.
+- [x] Keep kube access on an isolated `KUBECONFIG`.
+- [x] Keep the user's existing kube context untouched.
+- [x] Install plugin CRDs and RBAC.
+- [x] Generate and install the StepPlugin `ConfigMap`.
+- [x] Inject a local-only Slack token into a cluster `Secret`.
+- [x] Create a test `MessageChannel` or `ClusterMessageChannel`.
+- [x] Run a `Stage` with `uses: send-message`.
+- [x] Prove promotion success in-cluster.
+- [x] Prove `slack.threadTS` output is populated.
+- [x] Verify the message appeared in the target Slack channel.
+
+## Phase 6: repo harness integration
+
+- [x] Extend `extended/tests/e2e_stepplugins.sh` to support the real
+      `send-message` smoke path.
+- [x] Keep the committed harness credential-free.
+- [x] Gate Slack smoke on a local env var for the token.
+- [x] Keep any non-`extended/` edit to a tiny hook only, if one is needed.
+
+## Phase Post-Green: Minimize Diff Of Files Outside ./extended Against Kargo Upstream
+
+- [x] Fetch `upstream`.
+- [x] Review every edited file outside `extended/`, if any, against
+      `upstream/main`.
+- [x] Move more logic behind `extended/` helpers if that shrinks the outside
+      diff safely.
+- [x] Re-run matching tests after each cleanup pass.
+- [x] Stop only when no obvious outside-`extended/` shrink remains.
diff --git a/extended/docs/proposals/0004-send-message-step-plugin/implementation_notes.md b/extended/docs/proposals/0004-send-message-step-plugin/implementation_notes.md
new file mode 100644
index 0000000000..9c2f11bf0c
--- /dev/null
+++ b/extended/docs/proposals/0004-send-message-step-plugin/implementation_notes.md
@@ -0,0 +1,71 @@
+# 0004 Implementation Notes
+
+- Plugin subtree:
+  - `extended/plugins/send-message/`
+  - standalone `go.mod`
+  - standalone `Dockerfile`
+  - standalone `plugin.yaml`
+  - standalone CRDs and RBAC under `manifests/`
+  - standalone smoke helper under `smoke/`
+
+- Runtime contract implemented in the plugin:
+  - `POST /api/v1/step.execute`
+  - bearer auth from `/var/run/kargo/token`
+  - direct Kubernetes API reads from projected service-account credentials
+  - direct Slack API call from the plugin
+  - encoded `json`, `yaml`, and `xml` payload support
+
+- V1 scope shipped:
+  - `send-message`
+  - `MessageChannel`
+  - `ClusterMessageChannel`
+  - Slack only
+  - no SMTP
+
+- Secret and channel rules implemented:
+  - `secretRef.name` resolves in the Project namespace for `MessageChannel`
+  - `secretRef.name` resolves in the system-resources namespace for
+    `ClusterMessageChannel`
+  - Slack token key is `apiKey`
+  - `spec.slack.channelID` is required in CRD schema
+
+- Smoke harness knobs:
+  - `STEPPLUGIN_SEND_MESSAGE_SMOKE=true`
+  - `STEPPLUGIN_SEND_MESSAGE_CHANNEL_ID=<channel-id>`
+  - `STEPPLUGIN_SEND_MESSAGE_SLACK_API_KEY=<token>`
+- Plugin-local smoke entrypoint:
+  - `extended/plugins/send-message/smoke/smoke-test.sh`
+  - repo harness calls that script rather than owning the full orchestration
+
+- Host gap found during smoke:
+  - the StepPlugin auth-token mount was not readable by non-root sidecars
+  - fix was in `extended/pkg/stepplugin/agent/command_bridge.go`
+  - auth directory mode is now `0755`
+  - auth file mode is now `0444`
+
+- Smoke-result note:
+  - plugin response includes `output.slack.threadTS`
+  - Promotion state observed in-cluster stored the value under
+    `status.state.step-1.slack.threadTS`
+  - the smoke harness accepts either that path or
+    `status.stepExecutionMetadata[0].output.slack.threadTS`
+
+- Encoded-message note:
+  - when `encodingType` is set, `config.slack.channelID` and
+    `config.slack.threadTS` are ignored
+  - encoded body uses Slack-native field names such as `channel` and
+    `thread_ts`
+  - if encoded body omits `channel`, plugin fills it from channel resource
+    `spec.slack.channelID`
+
+- XML note:
+  - no public exact upstream XML example was found
+  - this implementation treats XML as a Kargo-owned alternate serialization of
+    the same Slack payload object used for `json` and `yaml`
+  - root name is ignored
+  - repeated sibling names become arrays
+
+- Non-plugin repo fix found during smoke:
+  - `pkg/cli/cmd/promote/promote.go` assumed wrapped REST payloads
+  - local smoke showed direct-object and direct-list payloads
+  - decoder helpers now accept both shapes
diff --git a/extended/docs/proposals/0004-send-message-step-plugin/implementation_plan.md b/extended/docs/proposals/0004-send-message-step-plugin/implementation_plan.md
new file mode 100644
index 0000000000..e6cdaea447
--- /dev/null
+++ b/extended/docs/proposals/0004-send-message-step-plugin/implementation_plan.md
@@ -0,0 +1,175 @@
+# 0004 Implementation Plan
+
+## Goal
+
+- Deliver a real Slack-only `send-message` StepPlugin under
+  `extended/plugins/send-message`.
+- Keep that subtree self-contained enough to behave like its own Git repo.
+- Do not rely on code, files, build helpers, tests, or other resources outside
+  `extended/plugins/send-message` for the plugin implementation itself.
+- Use the StepPlugin host slice from proposal
+  [0002-kargo-executor-plugins-from-argo-workflows](../0002-kargo-executor-plugins-from-argo-workflows/proposal.md)
+  without widening host responsibilities.
+- Keep channel lookup and referenced Secret reads in the plugin, not the host.
+- Use local-only Slack credentials for smoke testing. Do not commit any real
+  token or local token source.
+- Make the plugin match the documented Slack subset, including encoded-message
+  behavior, except for SMTP which stays out of scope.
+
+## Implementation Shape
+
+- Put all plugin-owned source, manifests, tests, and smoke assets under:
+  - `extended/plugins/send-message/`
+- Treat that subtree as a standalone plugin repo:
+  - own runtime source
+  - own image build
+  - own CRDs
+  - own RBAC manifests
+  - own tests
+  - own smoke-test helpers
+  - own smoke entrypoint script
+- It is acceptable for repo-level smoke harness code outside that subtree to
+  call into the subtree's smoke entrypoint.
+- It is not acceptable for plugin runtime code inside that subtree to import or
+  shell out to helpers elsewhere in this repo.
+
+## Plugin Runtime
+
+- Build a plugin-owned runtime image from `extended/plugins/send-message/`.
+- Implement `POST /api/v1/step.execute`.
+- Enforce the StepPlugin bearer-token contract:
+  - read `/var/run/kargo/token`
+  - require `Authorization: Bearer ...`
+  - return `403` on bad auth
+- Read Kubernetes credentials from the mounted service account projection when
+  present.
+- Use direct Kubernetes API reads from the plugin:
+  - `MessageChannel` in the Project namespace
+  - `ClusterMessageChannel` cluster-scoped
+  - referenced `Secret` in the Project namespace or system-resources namespace
+- Call Slack from the plugin runtime, not through the host.
+
+## V1 Behavior
+
+- Scope:
+  - `send-message`
+  - `MessageChannel`
+  - `ClusterMessageChannel`
+  - Slack only
+- Do not implement SMTP.
+- Support this step-config subset in v1:
+  - `channel.kind`
+  - `channel.name`
+  - `message`
+  - `encodingType`
+  - `slack.channelID`
+  - `slack.threadTS`
+- Support this output subset in v1:
+  - `slack.threadTS`
+- Plaintext behavior:
+  - `message` is sent as Slack `text`
+  - `slack.channelID` overrides channel resource `spec.slack.channelID`
+  - `slack.threadTS` is sent as Slack `thread_ts`
+  - output `slack.threadTS` is `config.slack.threadTS` if set, else Slack
+    response `ts`
+- Encoded-message behavior:
+  - support `json`, `yaml`, and `xml`
+  - when `encodingType` is set, treat the body as the Slack payload object
+  - ignore `config.slack.channelID` and `config.slack.threadTS`
+  - if encoded body omits `channel`, fill it from channel resource
+    `spec.slack.channelID`
+  - if encoded body sets `thread_ts`, return that value in output
+  - otherwise return Slack response `ts` in output
+- XML behavior:
+  - XML is a Kargo-owned alternate serialization of the same Slack payload
+    object used for `json` and `yaml`
+  - root element name is ignored
+  - attributes become keys
+  - repeated sibling element names become arrays
+  - nested elements become nested objects
+- `MessageChannel` lookup rules:
+  - resource kind is `MessageChannel`
+  - resource namespace is the Project namespace from the step context
+  - `secretRef.name` resolves in the same namespace
+  - `spec.slack.channelID` is required
+  - Slack token key is `apiKey`
+- `ClusterMessageChannel` lookup rules:
+  - resource kind is `ClusterMessageChannel`
+  - `secretRef.name` resolves in the system-resources namespace configured for
+    the plugin install
+  - `spec.slack.channelID` is required
+  - Slack token key is `apiKey`
+
+## CRDs And RBAC
+
+- The plugin subtree owns the OSS CRD manifests for:
+  - `MessageChannel`
+  - `ClusterMessageChannel`
+- Use the existing Akuity API group:
+  - `ee.kargo.akuity.io/v1alpha1`
+- Keep schemas minimal but structurally valid for the Slack-only slice.
+- Ship plugin-owned RBAC manifests under the subtree.
+- For the local smoke path:
+  - bind only the test Project default `ServiceAccount`
+  - grant channel reads and referenced Secret reads needed by the plugin
+  - keep that setup in the smoke assets or smoke helper, not in host code
+
+## Tests
+
+- Keep plugin tests under `extended/plugins/send-message/`.
+- Cover at least:
+  - bearer-token auth
+  - `MessageChannel` lookup
+  - `ClusterMessageChannel` lookup
+  - referenced Secret lookup
+  - Slack request shaping for plain text
+  - `slack.channelID` override
+  - `slack.threadTS` override
+  - `slack.threadTS` output
+  - encoded-body behavior ignores `config.slack.*`
+  - `xml` decode path
+  - failure path when channel or Secret is missing
+  - failure path when Slack returns an error
+
+## Smoke Test
+
+- Put the primary smoke script in:
+  - `extended/plugins/send-message/smoke/smoke-test.sh`
+- That script assumes Kargo is already available and takes its inputs from env.
+- Keep the committed smoke path credential-free.
+- Expect a local env var for the Slack API token during local smoke testing.
+- Build the plugin image from `extended/plugins/send-message/`.
+- Load that image into the local kind cluster without touching the user's
+  global kube context.
+- Install plugin CRDs, RBAC, and generated StepPlugin `ConfigMap`.
+- Create either a `MessageChannel` or `ClusterMessageChannel` test resource and
+  the referenced Secret in-cluster from the local-only token source.
+- Run a `Stage` with `uses: send-message`.
+- Prove in-cluster success from `Promotion` status, including non-empty
+  `slack.threadTS`.
+- For interactive local verification, also confirm the message appeared in the
+  target Slack channel.
+- The repo harness may call this script at the right point in e2e flow, but the
+  smoke orchestration lives in the plugin subtree.
+
+## Host Changes
+
+- Prefer zero host changes.
+- If the plugin proves a host gap, keep any host fix thin and behind
+  `extended/`.
+- Do not move channel lookup, Secret reads, or Slack API calls into the host.
+- Host gap actually found:
+  - non-root sidecars could not read `/var/run/kargo/token`
+  - keep the fix thin in `extended/pkg/stepplugin/agent/command_bridge.go`
+  - auth directory mode must permit traversal by non-root sidecars
+  - auth file mode must permit read by non-root sidecars
+
+## Phase Post-Green: Minimize Diff Of Files Outside ./extended Against Kargo Upstream
+
+1. Get the feature green first.
+2. Fetch `upstream/main`.
+3. Review every edited file outside `extended/`, if any.
+4. Move logic behind `extended/` helpers where that shrinks the conflict
+   surface.
+5. Re-run the matching tests after each cleanup pass.
+6. Stop when no obvious helper extraction or edit-block reduction remains.
diff --git a/extended/docs/proposals/0004-send-message-step-plugin/proposal.md b/extended/docs/proposals/0004-send-message-step-plugin/proposal.md
index be493341d7..bffdfd96e4 100644
--- a/extended/docs/proposals/0004-send-message-step-plugin/proposal.md
+++ b/extended/docs/proposals/0004-send-message-step-plugin/proposal.md
@@ -20,8 +20,6 @@ Date: 2026-03-24
   `extended/plugins/send-message`.
 - Use that subtree to test whether a third party can implement the plugin
   without merge permissions in the main repo.
-- Match the public Kargo `send-message` step shape for the Slack subset as
-  closely as public docs allow.
 - Keep channel lookup and referenced Secret reads in the plugin, not the host.
 - Do not implement SMTP in this slice.
 
diff --git a/extended/docs/proposals/0004-send-message-step-plugin/spec.md b/extended/docs/proposals/0004-send-message-step-plugin/spec.md
new file mode 100644
index 0000000000..4ca9c4beb3
--- /dev/null
+++ b/extended/docs/proposals/0004-send-message-step-plugin/spec.md
@@ -0,0 +1,341 @@
+# 0004 Spec
+
+## Scope
+
+- plugin dir:
+  - `extended/plugins/send-message`
+- step kind:
+  - `send-message`
+- channel resources:
+  - `MessageChannel`
+  - `ClusterMessageChannel`
+- transport:
+  - Slack only
+- out of scope:
+  - SMTP
+
+## Separate-Repo Rule
+
+- Treat `extended/plugins/send-message` as if it were its own Git repo.
+- Plugin runtime, manifests, tests, image build, and smoke assets live under
+  that subtree.
+- The plugin runtime must not require code, files, build helpers, or tests from
+  elsewhere in `kargo-extended`.
+- Repo-level harness code may call the subtree's smoke entrypoint, but the
+  plugin must own the smoke procedure.
+
+## Runtime Contract
+
+- HTTP endpoint:
+  - `POST /api/v1/step.execute`
+- auth:
+  - sidecar reads `/var/run/kargo/token`
+  - request must send `Authorization: Bearer <token>`
+  - bad or missing token returns `403`
+- Kubernetes access:
+  - plugin reads projected service-account credentials
+  - plugin reads the Kubernetes API directly
+- host responsibilities:
+  - mount workdir
+  - mount auth token
+  - optionally mount projected service-account credentials
+  - pass opaque step config
+- host does not:
+  - resolve channels
+  - resolve referenced Secrets
+  - call Slack
+  - reshape encoded message bodies
+
+## Step Config
+
+Supported config keys:
+
+- `channel.kind`
+- `channel.name`
+- `message`
+- `encodingType`
+- `slack.channelID`
+- `slack.threadTS`
+
+Channel reference:
+
+- `channel.kind` must be `MessageChannel` or `ClusterMessageChannel`
+- `channel.name` names the referenced channel resource
+
+## Channel Rules
+
+`MessageChannel`:
+
+- namespaced
+- resource namespace is the step context Project namespace
+- `secretRef.name` resolves in the same namespace
+- `spec.slack.channelID` is required
+- Secret key is `apiKey`
+
+`ClusterMessageChannel`:
+
+- cluster-scoped
+- `secretRef.name` resolves in `SYSTEM_RESOURCES_NAMESPACE`
+- `spec.slack.channelID` is required
+- Secret key is `apiKey`
+
+## Plaintext Behavior
+
+- `encodingType` omitted or empty means plaintext
+- plaintext mode sends Slack `text = config.message`
+- plaintext mode resolves `channel` in this order:
+  - `config.slack.channelID`, if set
+  - else `channel.spec.slack.channelID`
+- plaintext mode sets `thread_ts = config.slack.threadTS` if set
+- plaintext output:
+  - `slack.threadTS = config.slack.threadTS` if set
+  - else `slack.threadTS = Slack response ts`
+
+Plaintext example:
+
+```yaml
+channel:
+  kind: MessageChannel
+  name: send-message-smoke
+message: hello from plugin
+slack:
+  channelID: C999
+  threadTS: "1700000000.000001"
+```
+
+Produces Slack payload:
+
+```json
+{
+  "channel": "C999",
+  "text": "hello from plugin",
+  "thread_ts": "1700000000.000001"
+}
+```
+
+## Encoded Message Behavior
+
+- supported encodings:
+  - `json`
+  - `yaml`
+  - `xml`
+- encoded mode means `message` is the Slack body, not plain text
+- encoded mode owns the Slack body shape
+- encoded body uses Slack field names, not Kargo config names
+- when `encodingType` is set:
+  - ignore `config.slack.channelID`
+  - ignore `config.slack.threadTS`
+- if encoded body omits `channel`, fill it from `channel.spec.slack.channelID`
+- encoded output:
+  - `slack.threadTS = payload.thread_ts` if set
+  - else `slack.threadTS = Slack response ts`
+
+Encoded body keys may include, for example:
+
+- `channel`
+- `thread_ts`
+- `text`
+- `blocks`
+- `icon_emoji`
+- `icon_url`
+
+JSON example:
+
+```yaml
+channel:
+  kind: MessageChannel
+  name: send-message-smoke
+encodingType: json
+message: |
+  {
+    "text": "rich",
+    "thread_ts": "1700000000.000002",
+    "blocks": [
+      {
+        "type": "section",
+        "text": {
+          "type": "mrkdwn",
+          "text": "*hello*"
+        }
+      }
+    ]
+  }
+slack:
+  channelID: C999
+  threadTS: "1700000000.000001"
+```
+
+Produces Slack payload:
+
+```json
+{
+  "channel": "<channel.spec.slack.channelID>",
+  "text": "rich",
+  "thread_ts": "1700000000.000002",
+  "blocks": [
+    {
+      "type": "section",
+      "text": {
+        "type": "mrkdwn",
+        "text": "*hello*"
+      }
+    }
+  ]
+}
+```
+
+The outer `slack.*` keys are ignored in this mode.
+The encoded body did not set `channel`, so the plugin filled it from the
+channel resource.
+
+## XML Mapping
+
+- There is no known Slack-standard XML format here.
+- XML is a Kargo-owned alternate serialization of the same logical Slack body
+  object used for `json` and `yaml`.
+- The implementation mapping is:
+  - root element name is ignored
+  - root attributes become top-level keys
+  - child element names become object keys
+  - repeated sibling element names become arrays
+  - leaf elements become strings
+  - nested elements become nested objects
+  - mixed content stores text under `#text`
+
+Simple XML example:
+
+```xml
+<message>
+  <channel>C1234567890</channel>
+  <text>hello</text>
+  <thread_ts>1700000000.000003</thread_ts>
+</message>
+```
+
+Decodes to:
+
+```json
+{
+  "channel": "C1234567890",
+  "text": "hello",
+  "thread_ts": "1700000000.000003"
+}
+```
+
+Array/object XML example:
+
+```xml
+<message>
+  <text>rich</text>
+  <blocks>
+    <type>section</type>
+    <text>
+      <type>mrkdwn</type>
+      <text>*hello*</text>
+    </text>
+  </blocks>
+  <blocks>
+    <type>divider</type>
+  </blocks>
+</message>
+```
+
+Decodes to:
+
+```json
+{
+  "text": "rich",
+  "blocks": [
+    {
+      "type": "section",
+      "text": {
+        "type": "mrkdwn",
+        "text": "*hello*"
+      }
+    },
+    {
+      "type": "divider"
+    }
+  ]
+}
+```
+
+## Response Contract
+
+Success:
+
+- status `Succeeded`
+- output key `slack.threadTS`
+
+Failure classes:
+
+- status `Failed`:
+  - Slack API rejects the request
+  - Kubernetes lookup needed for execution fails after request validation
+- status `Errored`:
+  - request shape is invalid
+  - auth is bad or missing
+  - config cannot be decoded
+  - required inputs are missing
+
+## CRDs And RBAC
+
+- API group:
+  - `ee.kargo.akuity.io/v1alpha1`
+- plugin-owned CRDs:
+  - `MessageChannel`
+  - `ClusterMessageChannel`
+- plugin-owned RBAC grants:
+  - read `MessageChannel`
+  - read `ClusterMessageChannel`
+  - read referenced `Secret`
+- local smoke binds the plugin reader role only to the test Project default
+  `ServiceAccount`
+
+## Smoke Contract
+
+- primary smoke entrypoint:
+  - `extended/plugins/send-message/smoke/smoke-test.sh`
+- script assumes Kargo is already available
+- script requires:
+  - `SEND_MESSAGE_SMOKE_PROJECT`
+  - `SEND_MESSAGE_SMOKE_WAREHOUSE`
+  - `SEND_MESSAGE_SMOKE_FREIGHT_NAME`
+  - `SEND_MESSAGE_SMOKE_SLACK_API_KEY`
+  - `SEND_MESSAGE_SMOKE_CHANNEL_ID`
+- script may override:
+  - `SEND_MESSAGE_SMOKE_SYSTEM_RESOURCES_NAMESPACE`
+  - `SEND_MESSAGE_SMOKE_SECRET_NAME`
+  - `SEND_MESSAGE_SMOKE_CHANNEL_NAME`
+  - `SEND_MESSAGE_SMOKE_CONFIGMAP_NAME`
+  - `KARGO_BIN`
+  - `KARGO_FLAGS`
+  - `KUBECTL_BIN`
+  - `DOCKER_BIN`
+  - `KIND_BIN`
+  - `JQ_BIN`
+- script behavior:
+  - require a `kind-*` kube context
+  - build the plugin image
+  - load the image into `kind`
+  - render the plugin build dir
+  - build the StepPlugin discovery `ConfigMap`
+  - install CRDs and RBAC
+  - create test Secret and `MessageChannel`
+  - create a test `Stage`
+  - approve and promote the requested freight
+  - wait for `Succeeded`
+  - assert non-empty `slack.threadTS`
+  - clean up test resources on exit
+- repo harness may call this script, but this script is the plugin-owned smoke
+  path
+
+## Host Prerequisite Found During 0004
+
+- Non-root third-party sidecars must be able to read `/var/run/kargo/token`.
+- The base host slice was missing that property.
+- The current host bridge for that requirement lives under
+  `extended/pkg/stepplugin/agent/`.
+- Current required auth mount behavior for plugin sidecars:
+  - auth directory mode `0755`
+  - auth file mode `0444`
diff --git a/extended/plugins/send-message-go/Dockerfile b/extended/plugins/send-message-go/Dockerfile
new file mode 100644
index 0000000000..a5fa74294c
--- /dev/null
+++ b/extended/plugins/send-message-go/Dockerfile
@@ -0,0 +1,19 @@
+FROM golang:1.25.1-alpine AS build
+
+WORKDIR /src
+
+COPY go.mod go.sum ./
+RUN go mod download
+
+COPY . .
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 \
+    go build -trimpath -ldflags='-s -w' \
+    -o /out/send-message-plugin ./cmd/send-message-plugin
+
+FROM alpine:3.22
+
+RUN apk add --no-cache ca-certificates
+
+COPY --from=build /out/send-message-plugin /usr/local/bin/send-message-plugin
+
+ENTRYPOINT ["/usr/local/bin/send-message-plugin"]
diff --git a/extended/plugins/send-message-go/cmd/send-message-plugin/main.go b/extended/plugins/send-message-go/cmd/send-message-plugin/main.go
new file mode 100644
index 0000000000..9f8bfbeb78
--- /dev/null
+++ b/extended/plugins/send-message-go/cmd/send-message-plugin/main.go
@@ -0,0 +1,31 @@
+package main
+
+import (
+	"log"
+	"net/http"
+	"os"
+	"time"
+
+	"github.com/code-dot-org/kargo-send-message-step-plugin/internal/plugin"
+)
+
+func main() {
+	logger := log.New(os.Stderr, "", log.LstdFlags|log.LUTC)
+	server := plugin.NewServer(plugin.Options{
+		Logger: logger,
+	})
+
+	addr := ":9765"
+	httpServer := &http.Server{
+		Addr:              addr,
+		Handler:           server.Handler(),
+		ReadHeaderTimeout: 5 * time.Second,
+		ReadTimeout:       30 * time.Second,
+		WriteTimeout:      30 * time.Second,
+		IdleTimeout:       60 * time.Second,
+	}
+	logger.Printf("send-message step plugin listening on %s", addr)
+	if err := httpServer.ListenAndServe(); err != nil {
+		logger.Fatalf("send-message step plugin server exited: %v", err)
+	}
+}
diff --git a/extended/plugins/send-message-go/go.mod b/extended/plugins/send-message-go/go.mod
new file mode 100644
index 0000000000..53df35e2cd
--- /dev/null
+++ b/extended/plugins/send-message-go/go.mod
@@ -0,0 +1,13 @@
+module github.com/code-dot-org/kargo-send-message-step-plugin
+
+go 1.25.1
+
+require (
+	github.com/stretchr/testify v1.11.1
+	gopkg.in/yaml.v3 v3.0.1
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+)
diff --git a/extended/plugins/send-message-go/go.sum b/extended/plugins/send-message-go/go.sum
new file mode 100644
index 0000000000..c4c1710c47
--- /dev/null
+++ b/extended/plugins/send-message-go/go.sum
@@ -0,0 +1,10 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
diff --git a/extended/plugins/send-message-go/internal/plugin/plugin.go b/extended/plugins/send-message-go/internal/plugin/plugin.go
new file mode 100644
index 0000000000..e2f84db96f
--- /dev/null
+++ b/extended/plugins/send-message-go/internal/plugin/plugin.go
@@ -0,0 +1,752 @@
+package plugin
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/base64"
+	"encoding/json"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"gopkg.in/yaml.v3"
+)
+
+const (
+	stepExecutePath = "/api/v1/step.execute"
+	authHeader      = "Authorization"
+	bearerPrefix    = "Bearer "
+
+	//nolint:gosec // Fixed in-cluster mount path, not a hardcoded credential.
+	defaultExpectedAuthPath = "/var/run/kargo/token"
+	serviceAccountDir       = "/var/run/secrets/kubernetes.io/serviceaccount"
+	serviceAccountToken     = "token"
+	serviceAccountCA        = "ca.crt"
+	defaultSystemNS         = "kargo-system-resources"
+	defaultSlackAPIBase     = "https://slack.com/api"
+	defaultKubernetesURL    = "https://kubernetes.default.svc"
+	defaultHTTPTimeout      = 30 * time.Second
+)
+
+type Options struct {
+	Logger                   *log.Logger
+	ExpectedTokenPath        string
+	SystemResourcesNamespace string
+	SlackAPIBaseURL          string
+	KubernetesClient         KubernetesClient
+	SlackClient              SlackClient
+}
+
+type Server struct {
+	logger                   *log.Logger
+	expectedTokenPath        string
+	systemResourcesNamespace string
+	slackAPIBaseURL          string
+	kubeClient               KubernetesClient
+	slackClient              SlackClient
+}
+
+type KubernetesClient interface {
+	GetMessageChannel(
+		ctx context.Context,
+		namespace string,
+		name string,
+	) (*ChannelResource, error)
+	GetClusterMessageChannel(
+		ctx context.Context,
+		name string,
+	) (*ChannelResource, error)
+	GetSecret(
+		ctx context.Context,
+		namespace string,
+		name string,
+	) (map[string]string, error)
+}
+
+type SlackClient interface {
+	PostMessage(
+		ctx context.Context,
+		apiBaseURL string,
+		token string,
+		payload map[string]any,
+	) (*SlackPostMessageResponse, error)
+}
+
+type StepExecuteRequest struct {
+	Context StepContext `json:"context"`
+	Step    Step        `json:"step"`
+}
+
+type StepContext struct {
+	Project string `json:"project,omitempty"`
+}
+
+type Step struct {
+	Kind   string            `json:"kind"`
+	Config SendMessageConfig `json:"config"`
+}
+
+type SendMessageConfig struct {
+	Channel      ChannelRef   `json:"channel"`
+	Message      string       `json:"message"`
+	EncodingType string       `json:"encodingType,omitempty"`
+	Slack        SlackOptions `json:"slack,omitempty"`
+}
+
+type ChannelRef struct {
+	Kind string `json:"kind"`
+	Name string `json:"name"`
+}
+
+type SlackOptions struct {
+	ChannelID string `json:"channelID,omitempty"`
+	ThreadTS  string `json:"threadTS,omitempty"`
+}
+
+type StepExecuteResponse struct {
+	Status   string         `json:"status"`
+	Message  string         `json:"message,omitempty"`
+	Output   map[string]any `json:"output,omitempty"`
+	Error    string         `json:"error,omitempty"`
+	Terminal bool           `json:"terminal,omitempty"`
+}
+
+type ChannelResource struct {
+	SecretName     string
+	SlackChannelID string
+	ResourceKind   string
+	ResourceName   string
+	ResourceNS     string
+}
+
+type SlackPostMessageResponse struct {
+	OK    bool   `json:"ok"`
+	Error string `json:"error,omitempty"`
+	TS    string `json:"ts,omitempty"`
+}
+
+type requestError struct {
+	message string
+}
+
+func (e requestError) Error() string {
+	return e.message
+}
+
+func NewServer(opts Options) *Server {
+	logger := opts.Logger
+	if logger == nil {
+		logger = log.New(io.Discard, "", 0)
+	}
+
+	expectedTokenPath := opts.ExpectedTokenPath
+	if expectedTokenPath == "" {
+		expectedTokenPath = defaultExpectedAuthPath
+	}
+
+	systemNS := opts.SystemResourcesNamespace
+	if systemNS == "" {
+		systemNS = strings.TrimSpace(os.Getenv("SYSTEM_RESOURCES_NAMESPACE"))
+	}
+	if systemNS == "" {
+		systemNS = defaultSystemNS
+	}
+
+	slackBaseURL := opts.SlackAPIBaseURL
+	if slackBaseURL == "" {
+		slackBaseURL = strings.TrimSpace(os.Getenv("SLACK_API_BASE_URL"))
+	}
+	if slackBaseURL == "" {
+		slackBaseURL = defaultSlackAPIBase
+	}
+
+	kubeClient := opts.KubernetesClient
+	if kubeClient == nil {
+		kubeClient = MustNewInClusterKubernetesClient()
+	}
+	slackClient := opts.SlackClient
+	if slackClient == nil {
+		slackClient = RealSlackClient{}
+	}
+
+	return &Server{
+		logger:                   logger,
+		expectedTokenPath:        expectedTokenPath,
+		systemResourcesNamespace: systemNS,
+		slackAPIBaseURL:          slackBaseURL,
+		kubeClient:               kubeClient,
+		slackClient:              slackClient,
+	}
+}
+
+func (s *Server) Handler() http.Handler {
+	return http.HandlerFunc(s.handle)
+}
+
+func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
+	if r.URL.Path != stepExecutePath {
+		http.NotFound(w, r)
+		return
+	}
+	if r.Method != http.MethodPost {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+		return
+	}
+
+	if err := s.authorize(r.Header.Get(authHeader)); err != nil {
+		s.writeJSON(w, http.StatusForbidden, StepExecuteResponse{
+			Status:   "Errored",
+			Message:  err.Error(),
+			Error:    err.Error(),
+			Terminal: true,
+		})
+		return
+	}
+
+	var req StepExecuteRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		s.writeJSON(w, http.StatusBadRequest, StepExecuteResponse{
+			Status:   "Errored",
+			Message:  "invalid request body",
+			Error:    err.Error(),
+			Terminal: true,
+		})
+		return
+	}
+
+	resp := s.execute(r.Context(), req)
+	s.writeJSON(w, http.StatusOK, resp)
+}
+
+func (s *Server) execute(
+	ctx context.Context,
+	req StepExecuteRequest,
+) StepExecuteResponse {
+	if req.Step.Kind != "send-message" {
+		return StepExecuteResponse{
+			Status:   "Errored",
+			Message:  fmt.Sprintf("unsupported step kind %q", req.Step.Kind),
+			Error:    fmt.Sprintf("unsupported step kind %q", req.Step.Kind),
+			Terminal: true,
+		}
+	}
+
+	channel, secretNS, err := s.lookupChannel(ctx, req)
+	if err != nil {
+		if isRequestError(err) {
+			return erroredResponse(err)
+		}
+		return failedResponse(err)
+	}
+
+	secret, err := s.kubeClient.GetSecret(ctx, secretNS, channel.SecretName)
+	if err != nil {
+		return failedResponse(fmt.Errorf("error getting Slack Secret: %w", err))
+	}
+	token := secret["apiKey"]
+	if token == "" {
+		return failedResponse(errors.New("slack Secret is missing key \"apiKey\""))
+	}
+
+	payload, outputThreadTS, err := buildSlackPayload(req.Step.Config, channel)
+	if err != nil {
+		if isRequestError(err) {
+			return erroredResponse(err)
+		}
+		return failedResponse(err)
+	}
+
+	slackResp, err := s.slackClient.PostMessage(
+		ctx,
+		s.slackAPIBaseURL,
+		token,
+		payload,
+	)
+	if err != nil {
+		return failedResponse(fmt.Errorf("error posting Slack message: %w", err))
+	}
+	if !slackResp.OK {
+		return failedResponse(fmt.Errorf("slack API error: %s", slackResp.Error))
+	}
+	if outputThreadTS == "" {
+		outputThreadTS = slackResp.TS
+	}
+
+	return StepExecuteResponse{
+		Status: "Succeeded",
+		Output: map[string]any{
+			"slack": map[string]any{
+				"threadTS": outputThreadTS,
+			},
+		},
+	}
+}
+
+func (s *Server) lookupChannel(
+	ctx context.Context,
+	req StepExecuteRequest,
+) (*ChannelResource, string, error) {
+	ref := req.Step.Config.Channel
+	switch ref.Kind {
+	case "MessageChannel":
+		if req.Context.Project == "" {
+			return nil, "", requestError{
+				message: "step context project is required for MessageChannel",
+			}
+		}
+		channel, err := s.kubeClient.GetMessageChannel(ctx, req.Context.Project, ref.Name)
+		if err != nil {
+			return nil, "", err
+		}
+		return channel, req.Context.Project, nil
+	case "ClusterMessageChannel":
+		channel, err := s.kubeClient.GetClusterMessageChannel(ctx, ref.Name)
+		if err != nil {
+			return nil, "", err
+		}
+		return channel, s.systemResourcesNamespace, nil
+	default:
+		return nil, "", requestError{
+			message: fmt.Sprintf("unsupported channel kind %q", ref.Kind),
+		}
+	}
+}
+
+func buildSlackPayload(
+	cfg SendMessageConfig,
+	channel *ChannelResource,
+) (map[string]any, string, error) {
+	var payload map[string]any
+	switch strings.TrimSpace(cfg.EncodingType) {
+	case "":
+		channelID := strings.TrimSpace(cfg.Slack.ChannelID)
+		if channelID == "" {
+			channelID = strings.TrimSpace(channel.SlackChannelID)
+		}
+		if channelID == "" {
+			return nil, "", requestError{
+				message: fmt.Sprintf(
+					"%s %q does not define spec.slack.channelID and config.slack.channelID is empty",
+					channel.ResourceKind,
+					channel.ResourceName,
+				),
+			}
+		}
+		payload = map[string]any{
+			"channel": channelID,
+			"text":    cfg.Message,
+		}
+		threadTS := strings.TrimSpace(cfg.Slack.ThreadTS)
+		if threadTS != "" {
+			payload["thread_ts"] = threadTS
+		}
+		return payload, threadTS, nil
+	case "json":
+		if err := json.Unmarshal([]byte(cfg.Message), &payload); err != nil {
+			return nil, "", requestError{
+				message: fmt.Sprintf("error decoding JSON Slack payload: %v", err),
+			}
+		}
+	case "yaml":
+		if err := yaml.Unmarshal([]byte(cfg.Message), &payload); err != nil {
+			return nil, "", requestError{
+				message: fmt.Sprintf("error decoding YAML Slack payload: %v", err),
+			}
+		}
+	case "xml":
+		var err error
+		payload, err = decodeXMLSlackPayload(cfg.Message)
+		if err != nil {
+			return nil, "", err
+		}
+	default:
+		return nil, "", requestError{
+			message: fmt.Sprintf("unsupported encodingType %q", cfg.EncodingType),
+		}
+	}
+
+	if payload == nil {
+		return nil, "", requestError{message: "Slack payload must decode to an object"}
+	}
+
+	channelID := strings.TrimSpace(channel.SlackChannelID)
+	if channelID == "" {
+		return nil, "", requestError{
+			message: fmt.Sprintf(
+				"%s %q does not define spec.slack.channelID",
+				channel.ResourceKind,
+				channel.ResourceName,
+			),
+		}
+	}
+	if _, ok := payload["channel"]; !ok {
+		payload["channel"] = channelID
+	}
+	threadTS := strings.TrimSpace(stringValue(payload["thread_ts"]))
+	return payload, threadTS, nil
+}
+
+type xmlPayloadNode struct {
+	XMLName xml.Name
+	Attrs   []xml.Attr       `xml:",any,attr"`
+	Nodes   []xmlPayloadNode `xml:",any"`
+	Text    string           `xml:",chardata"`
+}
+
+func decodeXMLSlackPayload(message string) (map[string]any, error) {
+	var root xmlPayloadNode
+	if err := xml.Unmarshal([]byte(message), &root); err != nil {
+		return nil, requestError{
+			message: fmt.Sprintf("error decoding XML Slack payload: %v", err),
+		}
+	}
+
+	payload := xmlNodeToObject(root)
+	if payload == nil {
+		return nil, requestError{message: "Slack payload must decode to an object"}
+	}
+	return payload, nil
+}
+
+func xmlNodeToObject(node xmlPayloadNode) map[string]any {
+	payload := map[string]any{}
+	for _, attr := range node.Attrs {
+		payload[attr.Name.Local] = attr.Value
+	}
+	for _, child := range node.Nodes {
+		appendXMLValue(payload, child.XMLName.Local, xmlNodeValue(child))
+	}
+
+	text := strings.TrimSpace(node.Text)
+	if text != "" {
+		if len(payload) == 0 {
+			payload["text"] = text
+		} else {
+			payload["#text"] = text
+		}
+	}
+	return payload
+}
+
+func xmlNodeValue(node xmlPayloadNode) any {
+	if len(node.Attrs) == 0 && len(node.Nodes) == 0 {
+		return strings.TrimSpace(node.Text)
+	}
+	return xmlNodeToObject(node)
+}
+
+func appendXMLValue(payload map[string]any, key string, value any) {
+	if existing, ok := payload[key]; ok {
+		switch typed := existing.(type) {
+		case []any:
+			payload[key] = append(typed, value)
+		default:
+			payload[key] = []any{typed, value}
+		}
+		return
+	}
+	payload[key] = value
+}
+
+func stringValue(value any) string {
+	switch typed := value.(type) {
+	case string:
+		return typed
+	default:
+		return ""
+	}
+}
+
+func failedResponse(err error) StepExecuteResponse {
+	return StepExecuteResponse{
+		Status:   "Failed",
+		Message:  err.Error(),
+		Error:    err.Error(),
+		Terminal: true,
+	}
+}
+
+func erroredResponse(err error) StepExecuteResponse {
+	return StepExecuteResponse{
+		Status:   "Errored",
+		Message:  err.Error(),
+		Error:    err.Error(),
+		Terminal: true,
+	}
+}
+
+func isRequestError(err error) bool {
+	var target requestError
+	return errors.As(err, &target)
+}
+
+func (s *Server) authorize(authHeaderValue string) error {
+	//nolint:gosec // Path is plugin configuration or the fixed auth mount path.
+	expected, err := os.ReadFile(s.expectedTokenPath)
+	if err != nil {
+		return fmt.Errorf("error reading auth token: %w", err)
+	}
+	headerValue := strings.TrimSpace(authHeaderValue)
+	if !strings.HasPrefix(headerValue, bearerPrefix) {
+		return errors.New("missing bearer token")
+	}
+	received := strings.TrimSpace(strings.TrimPrefix(headerValue, bearerPrefix))
+	if received != strings.TrimSpace(string(expected)) {
+		return errors.New("invalid bearer token")
+	}
+	return nil
+}
+
+func (s *Server) writeJSON(w http.ResponseWriter, statusCode int, response StepExecuteResponse) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(statusCode)
+	if err := json.NewEncoder(w).Encode(response); err != nil {
+		s.logger.Printf("error writing response: %v", err)
+	}
+}
+
+type RealSlackClient struct{}
+
+func (RealSlackClient) PostMessage(
+	ctx context.Context,
+	apiBaseURL string,
+	token string,
+	payload map[string]any,
+) (*SlackPostMessageResponse, error) {
+	body, err := json.Marshal(payload)
+	if err != nil {
+		return nil, err
+	}
+	req, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodPost,
+		strings.TrimRight(apiBaseURL, "/")+"/chat.postMessage",
+		strings.NewReader(string(body)),
+	)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Authorization", bearerPrefix+token)
+	req.Header.Set("Content-Type", "application/json")
+
+	client := &http.Client{Timeout: defaultHTTPTimeout}
+	//nolint:gosec // The request target is the configured Slack API base URL.
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	var result SlackPostMessageResponse
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return nil, err
+	}
+	if resp.StatusCode >= http.StatusBadRequest {
+		if result.Error == "" {
+			result.Error = resp.Status
+		}
+		result.OK = false
+	}
+	return &result, nil
+}
+
+type inClusterKubernetesClient struct {
+	baseURL string
+	client  *http.Client
+	token   string
+}
+
+func MustNewInClusterKubernetesClient() KubernetesClient {
+	client, err := NewInClusterKubernetesClient()
+	if err != nil {
+		panic(err)
+	}
+	return client
+}
+
+func NewInClusterKubernetesClient() (KubernetesClient, error) {
+	tokenBytes, err := os.ReadFile(filepath.Join(serviceAccountDir, serviceAccountToken))
+	if err != nil {
+		return nil, fmt.Errorf("error reading service account token: %w", err)
+	}
+	caBytes, err := os.ReadFile(filepath.Join(serviceAccountDir, serviceAccountCA))
+	if err != nil {
+		return nil, fmt.Errorf("error reading service account ca: %w", err)
+	}
+
+	pool := x509.NewCertPool()
+	if !pool.AppendCertsFromPEM(caBytes) {
+		return nil, errors.New("error loading service account ca")
+	}
+
+	baseURL := defaultKubernetesURL
+	if host := strings.TrimSpace(os.Getenv("KUBERNETES_SERVICE_HOST")); host != "" {
+		port := strings.TrimSpace(os.Getenv("KUBERNETES_SERVICE_PORT_HTTPS"))
+		if port == "" {
+			port = strings.TrimSpace(os.Getenv("KUBERNETES_SERVICE_PORT"))
+		}
+		if port == "" {
+			port = "443"
+		}
+		baseURL = "https://" + host + ":" + port
+	}
+
+	return &inClusterKubernetesClient{
+		baseURL: baseURL,
+		client: &http.Client{
+			Timeout: defaultHTTPTimeout,
+			Transport: &http.Transport{
+				TLSClientConfig: &tls.Config{
+					MinVersion: tls.VersionTLS12,
+					RootCAs:    pool,
+				},
+			},
+		},
+		token: strings.TrimSpace(string(tokenBytes)),
+	}, nil
+}
+
+func (c *inClusterKubernetesClient) GetMessageChannel(
+	ctx context.Context,
+	namespace string,
+	name string,
+) (*ChannelResource, error) {
+	var response struct {
+		Metadata struct {
+			Name      string `json:"name"`
+			Namespace string `json:"namespace"`
+		} `json:"metadata"`
+		Spec struct {
+			SecretRef struct {
+				Name string `json:"name"`
+			} `json:"secretRef"`
+			Slack struct {
+				ChannelID string `json:"channelID"`
+			} `json:"slack"`
+		} `json:"spec"`
+	}
+	err := c.getJSON(
+		ctx,
+		fmt.Sprintf(
+			"/apis/ee.kargo.akuity.io/v1alpha1/namespaces/%s/messagechannels/%s",
+			namespace,
+			name,
+		),
+		&response,
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &ChannelResource{
+		SecretName:     response.Spec.SecretRef.Name,
+		SlackChannelID: response.Spec.Slack.ChannelID,
+		ResourceKind:   "MessageChannel",
+		ResourceName:   response.Metadata.Name,
+		ResourceNS:     response.Metadata.Namespace,
+	}, nil
+}
+
+func (c *inClusterKubernetesClient) GetClusterMessageChannel(
+	ctx context.Context,
+	name string,
+) (*ChannelResource, error) {
+	var response struct {
+		Metadata struct {
+			Name string `json:"name"`
+		} `json:"metadata"`
+		Spec struct {
+			SecretRef struct {
+				Name string `json:"name"`
+			} `json:"secretRef"`
+			Slack struct {
+				ChannelID string `json:"channelID"`
+			} `json:"slack"`
+		} `json:"spec"`
+	}
+	err := c.getJSON(
+		ctx,
+		fmt.Sprintf(
+			"/apis/ee.kargo.akuity.io/v1alpha1/clustermessagechannels/%s",
+			name,
+		),
+		&response,
+	)
+	if err != nil {
+		return nil, err
+	}
+	return &ChannelResource{
+		SecretName:     response.Spec.SecretRef.Name,
+		SlackChannelID: response.Spec.Slack.ChannelID,
+		ResourceKind:   "ClusterMessageChannel",
+		ResourceName:   response.Metadata.Name,
+	}, nil
+}
+
+func (c *inClusterKubernetesClient) GetSecret(
+	ctx context.Context,
+	namespace string,
+	name string,
+) (map[string]string, error) {
+	var response struct {
+		Data map[string]string `json:"data"`
+	}
+	err := c.getJSON(
+		ctx,
+		fmt.Sprintf("/api/v1/namespaces/%s/secrets/%s", namespace, name),
+		&response,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	decoded := make(map[string]string, len(response.Data))
+	for key, value := range response.Data {
+		raw, err := base64.StdEncoding.DecodeString(value)
+		if err != nil {
+			return nil, fmt.Errorf("error decoding secret key %q: %w", key, err)
+		}
+		decoded[key] = string(raw)
+	}
+	return decoded, nil
+}
+
+func (c *inClusterKubernetesClient) getJSON(
+	ctx context.Context,
+	path string,
+	out any,
+) error {
+	req, err := http.NewRequestWithContext(
+		ctx,
+		http.MethodGet,
+		strings.TrimRight(c.baseURL, "/")+path,
+		nil,
+	)
+	if err != nil {
+		return err
+	}
+	req.Header.Set(authHeader, bearerPrefix+c.token)
+
+	//nolint:gosec // The request target is the in-cluster Kubernetes API.
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusNotFound {
+		return fmt.Errorf("resource not found at %s", path)
+	}
+	if resp.StatusCode >= http.StatusBadRequest {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("kubernetes API error %s: %s", resp.Status, strings.TrimSpace(string(body)))
+	}
+	return json.NewDecoder(resp.Body).Decode(out)
+}
diff --git a/extended/plugins/send-message-go/internal/plugin/plugin_test.go b/extended/plugins/send-message-go/internal/plugin/plugin_test.go
new file mode 100644
index 0000000000..2587140c7e
--- /dev/null
+++ b/extended/plugins/send-message-go/internal/plugin/plugin_test.go
@@ -0,0 +1,609 @@
+package plugin
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestHandlerRejectsMissingBearerToken(t *testing.T) {
+	srv := newTestServer(t, nil, nil)
+	req := httptest.NewRequest(
+		http.MethodPost,
+		stepExecutePath,
+		bytes.NewReader(mustJSON(t, minimalRequest())),
+	)
+	rec := httptest.NewRecorder()
+
+	srv.Handler().ServeHTTP(rec, req)
+
+	require.Equal(t, http.StatusForbidden, rec.Code)
+	var resp StepExecuteResponse
+	require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp))
+	require.Equal(t, "Errored", resp.Status)
+}
+
+func TestHandlerRejectsInvalidBearerToken(t *testing.T) {
+	srv := newTestServer(t, nil, nil)
+	req := httptest.NewRequest(
+		http.MethodPost,
+		stepExecutePath,
+		bytes.NewReader(mustJSON(t, minimalRequest())),
+	)
+	req.Header.Set(authHeader, bearerPrefix+"wrong-token")
+	rec := httptest.NewRecorder()
+
+	srv.Handler().ServeHTTP(rec, req)
+
+	require.Equal(t, http.StatusForbidden, rec.Code)
+	var resp StepExecuteResponse
+	require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp))
+	require.Equal(t, "Errored", resp.Status)
+	require.Equal(t, "invalid bearer token", resp.Error)
+}
+
+func TestHandlerRejectsInvalidRequestBody(t *testing.T) {
+	srv := newTestServer(t, nil, nil)
+	req := httptest.NewRequest(
+		http.MethodPost,
+		stepExecutePath,
+		bytes.NewBufferString("{not-json"),
+	)
+	req.Header.Set(authHeader, bearerPrefix+"expected-token")
+	rec := httptest.NewRecorder()
+
+	srv.Handler().ServeHTTP(rec, req)
+
+	require.Equal(t, http.StatusBadRequest, rec.Code)
+	var resp StepExecuteResponse
+	require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp))
+	require.Equal(t, "Errored", resp.Status)
+	require.Equal(t, "invalid request body", resp.Message)
+}
+
+func TestExecuteUsesNamespacedMessageChannel(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK: true,
+			TS: "1712345678.000100",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.Channel = ChannelRef{
+		Kind: "MessageChannel",
+		Name: "send",
+	}
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Succeeded", resp.Status)
+	require.Equal(t, "1712345678.000100", slackThreadTS(t, resp))
+	require.Equal(
+		t,
+		map[string]any{
+			"channel": "C123",
+			"text":    "hello from plugin",
+		},
+		slack.lastPayload,
+	)
+}
+
+func TestExecuteUsesClusterMessageChannel(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		clusterMessageChannels: map[string]*ChannelResource{
+			"send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C777",
+				ResourceKind:   "ClusterMessageChannel",
+				ResourceName:   "send",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"kargo-system-resources/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK: true,
+			TS: "1712345678.000200",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.Channel = ChannelRef{
+		Kind: "ClusterMessageChannel",
+		Name: "send",
+	}
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Succeeded", resp.Status)
+	require.Equal(
+		t,
+		map[string]any{
+			"channel": "C777",
+			"text":    "hello from plugin",
+		},
+		slack.lastPayload,
+	)
+}
+
+func TestExecuteHonorsSlackOverrides(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK: true,
+			TS: "1712345678.000300",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.Slack.ChannelID = "C999"
+	req.Step.Config.Slack.ThreadTS = "1700000000.000001"
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Succeeded", resp.Status)
+	require.Equal(
+		t,
+		map[string]any{
+			"channel":   "C999",
+			"text":      "hello from plugin",
+			"thread_ts": "1700000000.000001",
+		},
+		slack.lastPayload,
+	)
+	require.Equal(
+		t,
+		"1700000000.000001",
+		slackThreadTS(t, resp),
+	)
+}
+
+func TestExecuteSupportsJSONEncoding(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK: true,
+			TS: "1712345678.000400",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.EncodingType = "json"
+	req.Step.Config.Message = `{"text":"rich","blocks":[{"type":"section"}]}`
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Succeeded", resp.Status)
+	require.Equal(
+		t,
+		map[string]any{
+			"channel": "C123",
+			"text":    "rich",
+			"blocks": []any{
+				map[string]any{"type": "section"},
+			},
+		},
+		slack.lastPayload,
+	)
+}
+
+func TestExecuteSupportsYAMLEncoding(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK: true,
+			TS: "1712345678.000410",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.EncodingType = "yaml"
+	req.Step.Config.Message = "text: rich\nblocks:\n- type: section\n"
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Succeeded", resp.Status)
+	require.Equal(
+		t,
+		map[string]any{
+			"channel": "C123",
+			"text":    "rich",
+			"blocks": []any{
+				map[string]any{"type": "section"},
+			},
+		},
+		slack.lastPayload,
+	)
+}
+
+func TestExecuteEncodedPayloadIgnoresSlackConfigOverrides(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK: true,
+			TS: "1712345678.000450",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.EncodingType = "json"
+	req.Step.Config.Message = `{"text":"rich","thread_ts":"1700000000.000002"}`
+	req.Step.Config.Slack.ChannelID = "C999"
+	req.Step.Config.Slack.ThreadTS = "1700000000.000001"
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Succeeded", resp.Status)
+	require.Equal(
+		t,
+		map[string]any{
+			"channel":   "C123",
+			"text":      "rich",
+			"thread_ts": "1700000000.000002",
+		},
+		slack.lastPayload,
+	)
+	require.Equal(
+		t,
+		"1700000000.000002",
+		slackThreadTS(t, resp),
+	)
+}
+
+func TestExecuteFailsWhenSlackErrors(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK:    false,
+			Error: "channel_not_found",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Failed", resp.Status)
+	require.Contains(t, resp.Error, "channel_not_found")
+}
+
+func TestExecuteFailsWhenSecretMissing(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{},
+	}
+	srv := newTestServer(t, kube, &fakeSlackClient{})
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Failed", resp.Status)
+	require.Contains(t, resp.Error, "error getting Slack Secret")
+}
+
+func TestExecuteSupportsXMLEncoding(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	slack := &fakeSlackClient{
+		response: &SlackPostMessageResponse{
+			OK: true,
+			TS: "1712345678.000500",
+		},
+	}
+	srv := newTestServer(t, kube, slack)
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.EncodingType = "xml"
+	req.Step.Config.Message = `
+<message>
+  <text>rich</text>
+  <thread_ts>1700000000.000003</thread_ts>
+</message>`
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Succeeded", resp.Status)
+	require.Equal(
+		t,
+		map[string]any{
+			"channel":   "C123",
+			"text":      "rich",
+			"thread_ts": "1700000000.000003",
+		},
+		slack.lastPayload,
+	)
+	require.Equal(
+		t,
+		"1700000000.000003",
+		slackThreadTS(t, resp),
+	)
+}
+
+func TestExecuteReturnsErroredForBadYAML(t *testing.T) {
+	kube := &fakeKubernetesClient{
+		messageChannels: map[string]*ChannelResource{
+			"demo/send": {
+				SecretName:     "slack-token",
+				SlackChannelID: "C123",
+				ResourceKind:   "MessageChannel",
+				ResourceName:   "send",
+				ResourceNS:     "demo",
+			},
+		},
+		secrets: map[string]map[string]string{
+			"demo/slack-token": {"apiKey": "xoxb-demo"},
+		},
+	}
+	srv := newTestServer(t, kube, &fakeSlackClient{})
+
+	req := minimalRequest()
+	req.Context.Project = "demo"
+	req.Step.Config.EncodingType = "yaml"
+	req.Step.Config.Message = "text: [oops"
+	resp := executeRequest(t, srv, req)
+
+	require.Equal(t, "Errored", resp.Status)
+	require.Contains(t, resp.Error, "error decoding YAML Slack payload")
+}
+
+type fakeKubernetesClient struct {
+	messageChannels        map[string]*ChannelResource
+	clusterMessageChannels map[string]*ChannelResource
+	secrets                map[string]map[string]string
+}
+
+func (f *fakeKubernetesClient) GetMessageChannel(
+	_ context.Context,
+	namespace string,
+	name string,
+) (*ChannelResource, error) {
+	if channel, ok := f.messageChannels[namespace+"/"+name]; ok {
+		return channel, nil
+	}
+	return nil, errors.New("message channel not found")
+}
+
+func (f *fakeKubernetesClient) GetClusterMessageChannel(
+	_ context.Context,
+	name string,
+) (*ChannelResource, error) {
+	if channel, ok := f.clusterMessageChannels[name]; ok {
+		return channel, nil
+	}
+	return nil, errors.New("cluster message channel not found")
+}
+
+func (f *fakeKubernetesClient) GetSecret(
+	_ context.Context,
+	namespace string,
+	name string,
+) (map[string]string, error) {
+	if secret, ok := f.secrets[namespace+"/"+name]; ok {
+		return secret, nil
+	}
+	return nil, errors.New("secret not found")
+}
+
+type fakeSlackClient struct {
+	lastPayload map[string]any
+	response    *SlackPostMessageResponse
+	err         error
+}
+
+func (f *fakeSlackClient) PostMessage(
+	_ context.Context,
+	_ string,
+	_ string,
+	payload map[string]any,
+) (*SlackPostMessageResponse, error) {
+	f.lastPayload = payload
+	if f.err != nil {
+		return nil, f.err
+	}
+	if f.response == nil {
+		return &SlackPostMessageResponse{OK: true, TS: "0"}, nil
+	}
+	return f.response, nil
+}
+
+func newTestServer(
+	t *testing.T,
+	kube KubernetesClient,
+	slack SlackClient,
+) *Server {
+	t.Helper()
+
+	dir := t.TempDir()
+	tokenPath := filepath.Join(dir, "token")
+	require.NoError(t, os.WriteFile(tokenPath, []byte("expected-token"), 0o600))
+
+	if kube == nil {
+		kube = &fakeKubernetesClient{}
+	}
+	if slack == nil {
+		slack = &fakeSlackClient{}
+	}
+
+	return NewServer(Options{
+		Logger:                   log.New(io.Discard, "", 0),
+		ExpectedTokenPath:        tokenPath,
+		SystemResourcesNamespace: "kargo-system-resources",
+		KubernetesClient:         kube,
+		SlackClient:              slack,
+	})
+}
+
+func executeRequest(
+	t *testing.T,
+	srv *Server,
+	req StepExecuteRequest,
+) StepExecuteResponse {
+	t.Helper()
+
+	httpReq := httptest.NewRequest(
+		http.MethodPost,
+		stepExecutePath,
+		bytes.NewReader(mustJSON(t, req)),
+	)
+	httpReq.Header.Set(authHeader, bearerPrefix+"expected-token")
+	rec := httptest.NewRecorder()
+	srv.Handler().ServeHTTP(rec, httpReq)
+	require.Equal(t, http.StatusOK, rec.Code)
+
+	var resp StepExecuteResponse
+	require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &resp))
+	return resp
+}
+
+func slackThreadTS(t *testing.T, resp StepExecuteResponse) string {
+	t.Helper()
+
+	require.Contains(t, resp.Output, "slack")
+	slackOutput, ok := resp.Output["slack"].(map[string]any)
+	require.True(t, ok)
+	threadTS, ok := slackOutput["threadTS"].(string)
+	require.True(t, ok)
+	return threadTS
+}
+
+func minimalRequest() StepExecuteRequest {
+	return StepExecuteRequest{
+		Step: Step{
+			Kind: "send-message",
+			Config: SendMessageConfig{
+				Channel: ChannelRef{
+					Kind: "MessageChannel",
+					Name: "send",
+				},
+				Message: "hello from plugin",
+			},
+		},
+	}
+}
+
+func mustJSON(t *testing.T, value any) []byte {
+	t.Helper()
+	data, err := json.Marshal(value)
+	require.NoError(t, err)
+	return data
+}
diff --git a/extended/plugins/send-message-go/manifests/crds.yaml b/extended/plugins/send-message-go/manifests/crds.yaml
new file mode 100644
index 0000000000..008f9931f8
--- /dev/null
+++ b/extended/plugins/send-message-go/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-go/manifests/rbac.yaml b/extended/plugins/send-message-go/manifests/rbac.yaml
new file mode 100644
index 0000000000..df10af6b7f
--- /dev/null
+++ b/extended/plugins/send-message-go/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-go/plugin.yaml b/extended/plugins/send-message-go/plugin.yaml
new file mode 100644
index 0000000000..76126e0a04
--- /dev/null
+++ b/extended/plugins/send-message-go/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: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-go/smoke/README.md b/extended/plugins/send-message-go/smoke/README.md
new file mode 100644
index 0000000000..a93e84ebbd
--- /dev/null
+++ b/extended/plugins/send-message-go/smoke/README.md
@@ -0,0 +1,17 @@
+# Local Smoke Notes
+
+- Primary smoke entrypoint:
+  - `extended/plugins/send-message-go/smoke/smoke-test.sh`
+- 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`
+- No committed file in this subtree contains a real token value or local token
+  source.
diff --git a/extended/plugins/send-message-go/smoke/render-plugin-dir.sh b/extended/plugins/send-message-go/smoke/render-plugin-dir.sh
new file mode 100755
index 0000000000..eecbfa6d96
--- /dev/null
+++ b/extended/plugins/send-message-go/smoke/render-plugin-dir.sh
@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+if [[ $# -ne 3 ]]; then
+    echo "usage: $0 OUT_DIR IMAGE SYSTEM_RESOURCES_NAMESPACE" >&2
+    exit 1
+fi
+
+OUT_DIR="$1"
+IMAGE="$2"
+SYSTEM_RESOURCES_NAMESPACE="$3"
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PLUGIN_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+mkdir -p "$OUT_DIR"
+cp "$PLUGIN_DIR/plugin.yaml" "$OUT_DIR/plugin.yaml"
+
+python3 - "$OUT_DIR/plugin.yaml" "$IMAGE" "$SYSTEM_RESOURCES_NAMESPACE" <<'PY'
+import pathlib
+import sys
+
+plugin_yaml = pathlib.Path(sys.argv[1])
+image = sys.argv[2]
+system_ns = sys.argv[3]
+
+data = plugin_yaml.read_text()
+data = data.replace("namespace: kargo-system-resources", f"namespace: {system_ns}")
+data = data.replace("image: send-message-step-plugin:dev", f"image: {image}")
+data = data.replace("value: kargo-system-resources", f"value: {system_ns}")
+plugin_yaml.write_text(data)
+PY
diff --git a/extended/plugins/send-message-go/smoke/smoke-test.sh b/extended/plugins/send-message-go/smoke/smoke-test.sh
new file mode 100755
index 0000000000..59564cbcef
--- /dev/null
+++ b/extended/plugins/send-message-go/smoke/smoke-test.sh
@@ -0,0 +1,285 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+log_info() {
+    printf '[INFO] %s\n' "$*"
+}
+
+log_pass() {
+    printf '[PASS] %s\n' "$*"
+}
+
+log_fail() {
+    printf '[FAIL] %s\n' "$*" >&2
+}
+
+require_env() {
+    local name="$1"
+    if [[ -z "${!name:-}" ]]; then
+        log_fail "$name is required"
+        exit 1
+    fi
+}
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PLUGIN_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+KARGO_BIN="${KARGO_BIN:-kargo}"
+KARGO_FLAGS="${KARGO_FLAGS:-}"
+KUBECTL_BIN="${KUBECTL_BIN:-kubectl}"
+DOCKER_BIN="${DOCKER_BIN:-docker}"
+KIND_BIN="${KIND_BIN:-kind}"
+JQ_BIN="${JQ_BIN:-jq}"
+
+SYSTEM_RESOURCES_NS="${SEND_MESSAGE_SMOKE_SYSTEM_RESOURCES_NAMESPACE:-kargo-system-resources}"
+SECRET_NAME="${SEND_MESSAGE_SMOKE_SECRET_NAME:-send-message-slack-token}"
+CHANNEL_NAME="${SEND_MESSAGE_SMOKE_CHANNEL_NAME:-send-message-smoke}"
+CONFIGMAP_NAME="${SEND_MESSAGE_SMOKE_CONFIGMAP_NAME:-send-message-step-plugin}"
+CLUSTER_ROLE_NAME="send-message-step-plugin-reader"
+
+require_env SEND_MESSAGE_SMOKE_PROJECT
+require_env SEND_MESSAGE_SMOKE_WAREHOUSE
+require_env SEND_MESSAGE_SMOKE_FREIGHT_NAME
+require_env SEND_MESSAGE_SMOKE_SLACK_API_KEY
+require_env SEND_MESSAGE_SMOKE_CHANNEL_ID
+
+TEST_PROJECT="$SEND_MESSAGE_SMOKE_PROJECT"
+TEST_WAREHOUSE="$SEND_MESSAGE_SMOKE_WAREHOUSE"
+SMOKE_FREIGHT_NAME="$SEND_MESSAGE_SMOKE_FREIGHT_NAME"
+
+PLUGIN_BUILD_DIR=""
+STAGE_NAME=""
+CLUSTER_ROLE_BINDING_NAME=""
+
+cleanup() {
+    if [[ -n "$STAGE_NAME" ]]; then
+        "$KUBECTL_BIN" delete stage.kargo.akuity.io "$STAGE_NAME" \
+            -n "$TEST_PROJECT" \
+            --ignore-not-found >/dev/null 2>&1 || true
+    fi
+
+    "$KUBECTL_BIN" delete messagechannel.ee.kargo.akuity.io "$CHANNEL_NAME" \
+        -n "$TEST_PROJECT" \
+        --ignore-not-found >/dev/null 2>&1 || true
+    "$KUBECTL_BIN" delete secret "$SECRET_NAME" \
+        -n "$TEST_PROJECT" \
+        --ignore-not-found >/dev/null 2>&1 || true
+    "$KUBECTL_BIN" delete configmap "$CONFIGMAP_NAME" \
+        -n "$SYSTEM_RESOURCES_NS" \
+        --ignore-not-found >/dev/null 2>&1 || true
+
+    if [[ -n "$CLUSTER_ROLE_BINDING_NAME" ]]; then
+        "$KUBECTL_BIN" delete clusterrolebinding "$CLUSTER_ROLE_BINDING_NAME" \
+            --ignore-not-found >/dev/null 2>&1 || true
+    fi
+
+    "$KUBECTL_BIN" delete clusterrole "$CLUSTER_ROLE_NAME" \
+        --ignore-not-found >/dev/null 2>&1 || true
+    "$KUBECTL_BIN" delete -f "$PLUGIN_DIR/manifests/crds.yaml" \
+        --ignore-not-found >/dev/null 2>&1 || true
+
+    if [[ -n "$PLUGIN_BUILD_DIR" ]]; then
+        rm -rf "$PLUGIN_BUILD_DIR"
+    fi
+}
+
+trap cleanup EXIT
+
+current_context="$("$KUBECTL_BIN" config current-context)"
+if [[ "$current_context" != kind-* ]]; then
+    log_fail "send-message smoke requires a kind context, got: $current_context"
+    exit 1
+fi
+
+kind_name="${current_context#kind-}"
+image_tag="send-message-step-plugin:e2e-$(date +%s)"
+PLUGIN_BUILD_DIR="$(mktemp -d "/tmp/send-message-stepplugin-e2e-XXXXXX")"
+
+log_info "Build send-message StepPlugin image"
+"$DOCKER_BIN" build -t "$image_tag" "$PLUGIN_DIR"
+log_pass "Build send-message StepPlugin image"
+
+log_info "Load send-message StepPlugin image into kind"
+"$KIND_BIN" load docker-image --name "$kind_name" "$image_tag"
+log_pass "Load send-message StepPlugin image into kind"
+
+log_info "Render send-message StepPlugin build dir"
+"$SCRIPT_DIR/render-plugin-dir.sh" "$PLUGIN_BUILD_DIR" "$image_tag" "$SYSTEM_RESOURCES_NS"
+log_pass "Render send-message StepPlugin build dir"
+
+log_info "Build send-message StepPlugin ConfigMap"
+(
+    cd "$PLUGIN_BUILD_DIR"
+    "$KARGO_BIN" step-plugin build .
+)
+log_pass "Build send-message StepPlugin ConfigMap"
+
+log_info "Install send-message CRDs"
+"$KUBECTL_BIN" apply -f "$PLUGIN_DIR/manifests/crds.yaml"
+log_pass "Install send-message CRDs"
+
+log_info "Install send-message ClusterRole"
+"$KUBECTL_BIN" apply -f "$PLUGIN_DIR/manifests/rbac.yaml"
+log_pass "Install send-message ClusterRole"
+
+CLUSTER_ROLE_BINDING_NAME="send-message-step-plugin-reader-${TEST_PROJECT}"
+cat > "/tmp/${CLUSTER_ROLE_BINDING_NAME}.yaml" <<EOF
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: ${CLUSTER_ROLE_BINDING_NAME}
+subjects:
+- kind: ServiceAccount
+  name: default
+  namespace: ${TEST_PROJECT}
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: ${CLUSTER_ROLE_NAME}
+EOF
+log_info "Bind send-message ClusterRole to test project default ServiceAccount"
+"$KUBECTL_BIN" apply -f "/tmp/${CLUSTER_ROLE_BINDING_NAME}.yaml"
+log_pass "Bind send-message ClusterRole to test project default ServiceAccount"
+
+log_info "Install send-message StepPlugin ConfigMap in system resources namespace"
+"$KUBECTL_BIN" apply -f "$PLUGIN_BUILD_DIR/${CONFIGMAP_NAME}-configmap.yaml"
+log_pass "Install send-message StepPlugin ConfigMap in system resources namespace"
+
+cat > "/tmp/${SECRET_NAME}.yaml" <<EOF
+apiVersion: v1
+kind: Secret
+metadata:
+  name: ${SECRET_NAME}
+  namespace: ${TEST_PROJECT}
+type: Opaque
+stringData:
+  apiKey: ${SEND_MESSAGE_SMOKE_SLACK_API_KEY}
+EOF
+log_info "Create send-message Slack Secret"
+"$KUBECTL_BIN" apply -f "/tmp/${SECRET_NAME}.yaml"
+log_pass "Create send-message Slack Secret"
+
+cat > "/tmp/${CHANNEL_NAME}.yaml" <<EOF
+apiVersion: ee.kargo.akuity.io/v1alpha1
+kind: MessageChannel
+metadata:
+  name: ${CHANNEL_NAME}
+  namespace: ${TEST_PROJECT}
+spec:
+  secretRef:
+    name: ${SECRET_NAME}
+  slack:
+    channelID: ${SEND_MESSAGE_SMOKE_CHANNEL_ID}
+EOF
+log_info "Create send-message MessageChannel"
+"$KUBECTL_BIN" apply -f "/tmp/${CHANNEL_NAME}.yaml"
+log_pass "Create send-message MessageChannel"
+
+STAGE_NAME="smsg-go-$(date +%s)"
+message_text="send-message GO implementation smoke ${STAGE_NAME}"
+cat > "/tmp/${STAGE_NAME}.yaml" <<EOF
+apiVersion: kargo.akuity.io/v1alpha1
+kind: Stage
+metadata:
+  name: ${STAGE_NAME}
+  namespace: ${TEST_PROJECT}
+spec:
+  requestedFreight:
+  - origin:
+      kind: Warehouse
+      name: ${TEST_WAREHOUSE}
+    sources:
+      direct: true
+  promotionTemplate:
+    spec:
+      steps:
+      - uses: send-message
+        config:
+          channel:
+            kind: MessageChannel
+            name: ${CHANNEL_NAME}
+          message: "${message_text}"
+EOF
+
+log_info "Create send-message StepPlugin smoke stage"
+stage_output="$("$KARGO_BIN" apply -f "/tmp/${STAGE_NAME}.yaml" $KARGO_FLAGS)"
+printf '%s\n' "$stage_output"
+if ! grep -Fq "stage.kargo.akuity.io/${STAGE_NAME}" <<<"$stage_output"; then
+    log_fail "send-message smoke stage apply output did not mention ${STAGE_NAME}"
+    exit 1
+fi
+log_pass "Create send-message StepPlugin smoke stage"
+
+log_info "Approve freight for send-message StepPlugin smoke stage"
+"$KARGO_BIN" approve \
+    --project="$TEST_PROJECT" \
+    --freight="$SMOKE_FREIGHT_NAME" \
+    --stage="$STAGE_NAME" \
+    $KARGO_FLAGS
+log_pass "Approve freight for send-message StepPlugin smoke stage"
+
+log_info "Promote freight through send-message StepPlugin smoke stage"
+"$KARGO_BIN" promote \
+    --project="$TEST_PROJECT" \
+    --freight="$SMOKE_FREIGHT_NAME" \
+    --stage="$STAGE_NAME" \
+    $KARGO_FLAGS
+log_pass "Promote freight through send-message StepPlugin smoke stage"
+
+promotion_name=""
+phase=""
+thread_ts=""
+for _ in $(seq 1 90); do
+    promotion_json="$(
+        "$KUBECTL_BIN" get promotion.kargo.akuity.io \
+            -n "$TEST_PROJECT" \
+            -o json 2>/dev/null |
+            "$JQ_BIN" -r --arg stage "$STAGE_NAME" '
+                [.items[] | select(.spec.stage == $stage)]
+                | sort_by(.metadata.creationTimestamp)
+                | last // {}
+            '
+    )" || true
+    if [[ -n "$promotion_json" && "$promotion_json" != "null" ]]; then
+        promotion_name="$(printf '%s\n' "$promotion_json" | "$JQ_BIN" -r '.metadata.name // empty')"
+        phase="$(printf '%s\n' "$promotion_json" | "$JQ_BIN" -r '.status.phase // empty')"
+        thread_ts="$(
+            printf '%s\n' "$promotion_json" |
+                "$JQ_BIN" -r '
+                    .status.stepExecutionMetadata[0].output.slack.threadTS //
+                    .status.state["step-1"].slack.threadTS //
+                    empty
+                '
+        )"
+    fi
+    case "$phase" in
+        Succeeded)
+            break
+            ;;
+        Failed|Errored|Aborted)
+            log_fail "send-message smoke promotion reached terminal phase $phase"
+            "$KUBECTL_BIN" get promotion.kargo.akuity.io "$promotion_name" \
+                -n "$TEST_PROJECT" \
+                -o yaml
+            exit 1
+            ;;
+    esac
+    sleep 2
+done
+
+if [[ "$phase" != "Succeeded" ]]; then
+    log_fail "send-message smoke promotion did not succeed in time"
+    "$KUBECTL_BIN" get promotion.kargo.akuity.io -n "$TEST_PROJECT" -o yaml
+    exit 1
+fi
+
+if [[ -z "$thread_ts" ]]; then
+    log_fail "send-message smoke did not produce slack.threadTS output"
+    "$KUBECTL_BIN" get promotion.kargo.akuity.io "$promotion_name" \
+        -n "$TEST_PROJECT" \
+        -o yaml
+    exit 1
+fi
+
+log_pass "send-message StepPlugin smoke promotion finished successfully"
diff --git a/extended/tests/e2e_stepplugins.sh b/extended/tests/e2e_stepplugins.sh
index 075f9a67f3..459082898f 100644
--- a/extended/tests/e2e_stepplugins.sh
+++ b/extended/tests/e2e_stepplugins.sh
@@ -356,5 +356,71 @@ EOF
     fi
 
     wait_for_stepplugin_promotion "$promotion_name"
+    run_send_message_stepplugin_e2e_tests "$smoke_freight_name"
     stepplugin_e2e_end "success"
 }
+
+run_send_message_stepplugin_e2e_tests() {
+    local smoke_freight_name="$1"
+    local send_message_impl="${STEPPLUGIN_SEND_MESSAGE_IMPL:-go}"
+    local smoke_cmd=()
+
+    if [[ "${STEPPLUGIN_SEND_MESSAGE_SMOKE:-false}" != "true" ]]; then
+        log_info "Skipping send-message StepPlugin smoke path; set STEPPLUGIN_SEND_MESSAGE_SMOKE=true to enable it"
+        return 0
+    fi
+
+    if [[ -z "${STEPPLUGIN_SEND_MESSAGE_SLACK_API_KEY:-}" ]]; then
+        log_error "STEPPLUGIN_SEND_MESSAGE_SLACK_API_KEY is required for send-message StepPlugin smoke"
+        stepplugin_e2e_fail
+    fi
+
+    if [[ -z "${STEPPLUGIN_SEND_MESSAGE_CHANNEL_ID:-}" ]]; then
+        log_error "STEPPLUGIN_SEND_MESSAGE_CHANNEL_ID is required for send-message StepPlugin smoke"
+        stepplugin_e2e_fail
+    fi
+
+    case "$send_message_impl" in
+        go)
+            smoke_cmd=(
+                "$REPO_ROOT/extended/plugins/send-message-go/smoke/smoke-test.sh"
+            )
+            ;;
+        python)
+            smoke_cmd=(
+                python3
+                "$REPO_ROOT/extended/plugins/send-message-python/smoke/smoke_test.py"
+            )
+            ;;
+        ruby)
+            smoke_cmd=(
+                ruby
+                "$REPO_ROOT/extended/plugins/send-message-ruby/smoke/smoke_test.rb"
+            )
+            ;;
+        ts)
+            smoke_cmd=(
+                bash
+                -lc
+                "cd \"$REPO_ROOT/extended/plugins/send-message-ts\" && npm ci && npm run smoke"
+            )
+            ;;
+        *)
+            log_error "Unsupported STEPPLUGIN_SEND_MESSAGE_IMPL=$send_message_impl; expected one of: go, python, ruby, ts"
+            stepplugin_e2e_fail
+            ;;
+    esac
+
+    log_test "Run send-message StepPlugin smoke script"
+    if ! SEND_MESSAGE_SMOKE_PROJECT="$TEST_PROJECT" \
+        SEND_MESSAGE_SMOKE_WAREHOUSE="$TEST_WAREHOUSE" \
+        SEND_MESSAGE_SMOKE_FREIGHT_NAME="$smoke_freight_name" \
+        SEND_MESSAGE_SMOKE_SLACK_API_KEY="$STEPPLUGIN_SEND_MESSAGE_SLACK_API_KEY" \
+        SEND_MESSAGE_SMOKE_CHANNEL_ID="$STEPPLUGIN_SEND_MESSAGE_CHANNEL_ID" \
+        SEND_MESSAGE_SMOKE_SYSTEM_RESOURCES_NAMESPACE="$SYSTEM_RESOURCES_NS" \
+        KARGO_BIN="$KARGO_BIN" \
+        KARGO_FLAGS="$KARGO_FLAGS" \
+        "${smoke_cmd[@]}"; then
+        stepplugin_e2e_fail
+    fi
+}