feat(stovepipe): Announce validation results

doc/rfc/stovepipe/workflow.md

@@ -55,10 +55,10 @@ The ref is a *cache* of the last-green URI, not a second record of greenness. It
 |---|---|
 | **SourceControl** | Resolve a Queue name to its current head URI; answer ancestry/comparison questions between two URIs (is the new head a fast-forward descendant of the last green, or was history rewritten?); enumerate commits in a range; advance the Queue's **promotion ref** to a commit. The sole owner of URI semantics, including which refs a Queue name resolves to. |
 | **build-runner** | Build a scope at a URI (optionally relative to a baseline URI), returning pass/fail and the target graph. See [build-runner.md](../submitqueue/build-runner.md). |
-| **Hooks** | Deliver Stovepipe's greenness events to downstream systems — "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. Not implemented yet; it will be the shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.md). |
+| **Hooks** | Deliver Stovepipe's greenness events to downstream systems — "this URI / this project is now green (or not green)". Fire-and-forget notification, decoupled so Stovepipe does not know or care who consumes the event. The shared cross-domain hook seam rather than a Stovepipe-specific extension. See [hook-framework.md](../hook-framework.md). |
 | **Storage** | Persist Queues (incl. last-green URI), Requests, build records, and per-URI / per-project greenness. Key/value-shaped per the extension-design rules in [AGENTS.md](../../../AGENTS.md). |
 
-Hooks are the notification boundary. When a validation fact is recorded — whole-repo green/not-green, or later a project green/not-green — the event reaches deployment systems, dashboards, and developer tooling without any of them polling Stovepipe's store, and each environment can route it to its own downstream (a deploy gate, a Slack notifier, an event bus) without changing the pipeline. The mechanism is the cross-domain hook framework rather than a call out of the recording stage: `record` publishes a `HookEvent` to Stovepipe's `hook` topic, and a dispatcher stage consumes it and invokes the wired hooks, so a slow or failing downstream cannot add latency to the pipeline. Neither half exists yet; see [record.md](steps/record.md#hooks) for the fact-to-event mapping and its open questions.
+Hooks are the notification boundary. When a validation fact is recorded — whole-repo green/not-green, or later a project green/not-green — the event reaches deployment systems, dashboards, and developer tooling without any of them polling Stovepipe's store, and each environment can route it to its own downstream (a deploy gate, a Slack notifier, an event bus) without changing the pipeline. The mechanism is the cross-domain hook framework rather than a call out of the recording stage: `record` publishes a `HookEvent` to Stovepipe's `hook` topic, and a dispatcher stage consumes it and invokes the wired hooks, so a slow or failing downstream cannot add latency to the pipeline. Both halves exist; what a deployment supplies is the hooks themselves, since the example server resolves every event to `noop`. See [record.md](steps/record.md#hooks) for the fact-to-event mapping.
 
 ## Workflow
 
@@ -163,7 +163,7 @@ Per-stage design detail lives under `steps/` so this doc stays a pipeline overvi
 - [process.md](steps/process.md) — build-strategy decision, concurrency gate, backlog coalescing, [concurrency lifecycle](steps/process.md#concurrency-lifecycle), entity changes, [waiting for a slot](steps/process.md#waiting-for-a-slot)
 - [build.md](steps/build.md) — trigger-only stage: reads the decided scope off the Request, triggers the build-runner, hands off to buildsignal; the stovepipe `BuildRunner` contract and why it differs from SubmitQueue's
 - [buildsignal.md](steps/buildsignal.md) — the poll loop: hold-based re-poll cadence, target-graph return, per-build partitioning, and the fail-closed handoff to record
-- [record.md](steps/record.md) — turning a terminal build outcome into an immutable validation fact, monotonic last-green advancement and ref promotion, and the deferred hook and analyze handoffs
+- [record.md](steps/record.md) — turning a terminal build outcome into an immutable validation fact, monotonic last-green advancement and ref promotion, the hook event announcing the outcome, and the deferred analyze handoff
 
 ## Dedup, idempotency, and history rewrites
 

platform/hook/BUILD.bazel

@@ -5,6 +5,7 @@ go_library(
     srcs = [
         "controller.go",
         "dlq.go",
+        "publisher.go",
     ],
     importpath = "github.com/uber/submitqueue/platform/hook",
     visibility = ["//visibility:public"],
@@ -14,6 +15,7 @@ go_library(
         "//platform/errs:go_default_library",
         "//platform/extension/hook:go_default_library",
         "//platform/metrics:go_default_library",
+        "//platform/publish:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
     ],
@@ -24,14 +26,17 @@ go_test(
     srcs = [
         "controller_test.go",
         "dlq_test.go",
+        "publisher_test.go",
     ],
     embed = [":go_default_library"],
     deps = [
         "//api/base/hook:go_default_library",
         "//platform/base/failure:go_default_library",
         "//platform/base/messagequeue:go_default_library",
+        "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
         "//platform/extension/hook:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",
         "@com_github_stretchr_testify//require:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",

platform/hook/controller.go

@@ -12,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-// Package hook holds the consumer side of the hooks framework: the controller
-// that turns hook events on a queue into hook.Hook calls, and the reconciler for
-// the events that never made it.
+// Package hook holds the domain-neutral mechanics of the hooks framework: the
+// controller that turns hook events on a queue into hook.Hook calls, the
+// reconciler for the events that never made it, and the helper a producer
+// publishes an event through.
 //
 // The controller is domain-neutral. Each domain runs its own hook topic and its
 // own instance of this stage — "per-domain" is about the topic and the wiring,

platform/hook/publisher.go

@@ -0,0 +1,58 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hook
+
+import (
+	"context"
+	"fmt"
+
+	basehook "github.com/uber/submitqueue/api/base/hook"
+	"github.com/uber/submitqueue/platform/consumer"
+	"github.com/uber/submitqueue/platform/publish"
+)
+
+// Publish sends one hook event to the domain's hook topic, partitioned by
+// partitionKey. The topic key is not a parameter: a domain runs a single hook
+// topic, and the caller's registry is what binds that key to a wire topic.
+//
+// The event id is the message id, so a redelivery republishing the same event
+// dedups into the original message instead of enqueuing a second one. Callers
+// pass the partition key their own topic partitions on, carrying that ordering
+// across the seam.
+//
+// Errors are returned unclassified, leaving retryability to the caller's
+// classifier: a malformed event is the caller's bug, not a transient fault.
+func Publish(
+	ctx context.Context,
+	registry consumer.TopicRegistry,
+	event *basehook.HookEvent,
+	partitionKey string,
+) error {
+	if err := basehook.Validate(event); err != nil {
+		return fmt.Errorf("refusing to publish a malformed hook event: %w", err)
+	}
+
+	body, err := basehook.Marshal(event)
+	if err != nil {
+		return fmt.Errorf("failed to serialize hook event %s: %w", event.GetId(), err)
+	}
+
+	if err := publish.Message(
+		ctx, registry, basehook.TopicKeyHook, publish.IntentID(event.GetId()), body, partitionKey,
+	); err != nil {
+		return fmt.Errorf("failed to publish hook event %s: %w", event.GetId(), err)
+	}
+	return nil
+}

platform/hook/publisher_test.go

@@ -0,0 +1,125 @@
+// Copyright (c) 2026 Uber Technologies, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package hook
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	basehook "github.com/uber/submitqueue/api/base/hook"
+	entityqueue "github.com/uber/submitqueue/platform/base/messagequeue"
+	"github.com/uber/submitqueue/platform/consumer"
+	mqmock "github.com/uber/submitqueue/platform/extension/messagequeue/mock"
+	"go.uber.org/mock/gomock"
+)
+
+const (
+	testEventID      = "stovepipe/validation.repository.recorded/request/7/0"
+	testPartitionKey = "request/7"
+)
+
+func testEvent() *basehook.HookEvent {
+	return &basehook.HookEvent{
+		Id:          testEventID,
+		Source:      "stovepipe",
+		Type:        "validation.repository.recorded",
+		TimestampMs: 1756327200000,
+	}
+}
+
+// registryWithHookTopic returns a registry whose hook topic captures whatever is
+// published to it, and the slot the captured message lands in.
+func registryWithHookTopic(t *testing.T, ctrl *gomock.Controller, publishErr error) (consumer.TopicRegistry, *entityqueue.Message) {
+	t.Helper()
+
+	var published entityqueue.Message
+	publisher := mqmock.NewMockPublisher(ctrl)
+	publisher.EXPECT().Publish(gomock.Any(), "domain-hook", gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ string, msg entityqueue.Message) error {
+			published = msg
+			return publishErr
+		}).AnyTimes()
+
+	queue := mqmock.NewMockQueue(ctrl)
+	queue.EXPECT().Publisher().Return(publisher).AnyTimes()
+
+	registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+		{Key: basehook.TopicKeyHook, Name: "domain-hook", Queue: queue},
+	})
+	require.NoError(t, err)
+	return registry, &published
+}
+
+func TestPublish(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	registry, published := registryWithHookTopic(t, ctrl, nil)
+
+	require.NoError(t, Publish(context.Background(), registry, testEvent(), testPartitionKey))
+
+	// The event id is the message id, so a redelivery republishing the same
+	// event dedups into the original message.
+	assert.Equal(t, testEventID, published.ID)
+	assert.Equal(t, testPartitionKey, published.PartitionKey)
+
+	decoded := &basehook.HookEvent{}
+	require.NoError(t, basehook.Unmarshal(published.Payload, decoded))
+	assert.Equal(t, testEventID, decoded.GetId())
+	assert.Equal(t, "validation.repository.recorded", decoded.GetType())
+}
+
+func TestPublish_RejectsMalformedEvent(t *testing.T) {
+	tests := []struct {
+		name  string
+		event *basehook.HookEvent
+	}{
+		{name: "nil", event: nil},
+		{name: "no id", event: &basehook.HookEvent{Source: "stovepipe", Type: "t"}},
+		{name: "no source", event: &basehook.HookEvent{Id: testEventID, Type: "t"}},
+		{name: "no type", event: &basehook.HookEvent{Id: testEventID, Source: "stovepipe"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			publisher := mqmock.NewMockPublisher(ctrl)
+			queue := mqmock.NewMockQueue(ctrl)
+			queue.EXPECT().Publisher().Return(publisher).AnyTimes()
+			registry, err := consumer.NewTopicRegistry([]consumer.TopicConfig{
+				{Key: basehook.TopicKeyHook, Name: "domain-hook", Queue: queue},
+			})
+			require.NoError(t, err)
+
+			// No Publish expectation: a malformed event must not reach the queue.
+			require.Error(t, Publish(context.Background(), registry, tt.event, testPartitionKey))
+		})
+	}
+}
+
+func TestPublish_PropagatesPublishFailure(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	registry, _ := registryWithHookTopic(t, ctrl, errors.New("boom"))
+
+	require.Error(t, Publish(context.Background(), registry, testEvent(), testPartitionKey))
+}
+
+func TestPublish_FailsWhenHookTopicIsUnregistered(t *testing.T) {
+	registry, err := consumer.NewTopicRegistry(nil)
+	require.NoError(t, err)
+
+	require.Error(t, Publish(context.Background(), registry, testEvent(), testPartitionKey))
+}

service/stovepipe/server/main.go

@@ -447,7 +447,7 @@ func registerPrimaryControllers(
 	}
 	count++
 
-	recordController := record.NewController(logger, scope, store, sourceControl, stovepipemq.TopicKeyRecord, "stovepipe-record")
+	recordController := record.NewController(logger, scope, store, sourceControl, registry, stovepipemq.TopicKeyRecord, "stovepipe-record")
 	if err := c.Register(recordController); err != nil {
 		return count, fmt.Errorf("failed to register record controller: %w", err)
 	}
@@ -492,7 +492,7 @@ func registerDLQControllers(
 	}
 	count++
 
-	recordDLQController := record.NewController(logger, scope, store, sourceControl, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
+	recordDLQController := record.NewController(logger, scope, store, sourceControl, registry, dlq.TopicKey(stovepipemq.TopicKeyRecord), "stovepipe-record-dlq")
 	if err := c.Register(recordDLQController); err != nil {
 		return count, fmt.Errorf("failed to register record dlq controller: %w", err)
 	}

stovepipe/controller/record/BUILD.bazel

@@ -6,8 +6,11 @@ go_library(
     importpath = "github.com/uber/submitqueue/stovepipe/controller/record",
     visibility = ["//visibility:public"],
     deps = [
+        "//api/base/hook:go_default_library",
         "//platform/consumer:go_default_library",
+        "//platform/hook:go_default_library",
         "//platform/metrics:go_default_library",
+        "//stovepipe/core/hookevent:go_default_library",
         "//stovepipe/core/loader:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
@@ -23,10 +26,13 @@ go_test(
     srcs = ["record_test.go"],
     embed = [":go_default_library"],
     deps = [
+        "//api/base/hook:go_default_library",
         "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/consumer/mock:go_default_library",
+        "//platform/extension/messagequeue/mock:go_default_library",
         "//platform/metrics:go_default_library",
+        "//stovepipe/core/hookevent:go_default_library",
         "//stovepipe/core/messagequeue:go_default_library",
         "//stovepipe/entity:go_default_library",
         "//stovepipe/extension/sourcecontrol:go_default_library",