feat(orch): Consume hook events

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -16,6 +16,7 @@ go_library(
     importpath = "github.com/uber/submitqueue/service/submitqueue/orchestrator/server",
     visibility = ["//visibility:private"],
     deps = [
+        "//api/base/hook:go_default_library",
         "//api/submitqueue/orchestrator/protopb:go_default_library",
         "//platform/buildkite:go_default_library",
         "//platform/errs/generic:go_default_library",
@@ -25,6 +26,8 @@ go_library(
         "//platform/extension/consumergate/noop:go_default_library",
         "//platform/extension/counter:go_default_library",
         "//platform/extension/counter/mysql:go_default_library",
+        "//platform/extension/hook:go_default_library",
+        "//platform/extension/hook/noop:go_default_library",
         "//platform/extension/messagequeue/mysql:go_default_library",
         "//platform/git/repo:go_default_library",
         "//platform/githubactions:go_default_library",

service/submitqueue/orchestrator/server/main.go

@@ -29,6 +29,7 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 
 	"github.com/uber-go/tally"
+	basehook "github.com/uber/submitqueue/api/base/hook"
 	pb "github.com/uber/submitqueue/api/submitqueue/orchestrator/protopb"
 	genericerrs "github.com/uber/submitqueue/platform/errs/generic"
 	mysqlerrs "github.com/uber/submitqueue/platform/errs/mysql"
@@ -37,6 +38,8 @@ import (
 	consumergatenoop "github.com/uber/submitqueue/platform/extension/consumergate/noop"
 	"github.com/uber/submitqueue/platform/extension/counter"
 	mysqlcounter "github.com/uber/submitqueue/platform/extension/counter/mysql"
+	hookext "github.com/uber/submitqueue/platform/extension/hook"
+	hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop"
 	queueMySQL "github.com/uber/submitqueue/platform/extension/messagequeue/mysql"
 	"github.com/uber/submitqueue/platform/pipeline"
 	"github.com/uber/submitqueue/submitqueue/core/changeset"
@@ -199,6 +202,7 @@ func run() error {
 		Analyzer:       profiles.AnalyzerFactory(),
 		Speculator:     profiles.SpeculatorFactory(),
 		Validator:      validatorFactory{},
+		Hooks:          hookResolver{},
 	}
 
 	// Assemble the pipeline: one call builds the topic registry, creates
@@ -439,6 +443,17 @@ func (f counterFactory) For(config counter.Config) (counter.Counter, error) {
 	return mysqlcounter.NewCounter(f.db, f.scope, config.QueueName), nil
 }
 
+// hookResolver sends every event to the no-op hook. Which hooks an event goes
+// to is host policy, so the resolver lives here rather than in the extension
+// package. A deployment with real integrations swaps this for one that selects
+// on the event's source and type.
+type hookResolver struct{}
+
+// For returns the hooks that run for event.
+func (hookResolver) For(*basehook.HookEvent) []hookext.Hook {
+	return []hookext.Hook{hooknoop.New()}
+}
+
 // validatorFactory routes every queue to the always-passing fake validator.
 // Choosing an impl per queue is host policy, so the adapter lives here rather
 // than in the extension package. A deployment with real validation swaps this

submitqueue/orchestrator/BUILD.bazel

@@ -1,14 +1,17 @@
-load("@rules_go//go:def.bzl", "go_library")
+load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
     srcs = ["pipeline.go"],
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator",
     visibility = ["//visibility:public"],
     deps = [
+        "//api/base/hook:go_default_library",
         "//api/runway/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/extension/counter:go_default_library",
+        "//platform/extension/hook:go_default_library",
+        "//platform/hook:go_default_library",
         "//platform/pipeline:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/extension/buildrunner:go_default_library",
@@ -35,3 +38,19 @@ go_library(
         "@org_uber_go_zap//:go_default_library",
     ],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["pipeline_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//api/base/hook:go_default_library",
+        "//platform/extension/hook:go_default_library",
+        "//platform/extension/hook/noop:go_default_library",
+        "//platform/pipeline: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",
+        "@org_uber_go_zap//zaptest:go_default_library",
+    ],
+)

submitqueue/orchestrator/pipeline.go

@@ -18,10 +18,15 @@
 package orchestrator
 
 import (
+	"fmt"
+
 	"github.com/uber-go/tally"
+	basehook "github.com/uber/submitqueue/api/base/hook"
 	runwaymq "github.com/uber/submitqueue/api/runway/messagequeue"
 	"github.com/uber/submitqueue/platform/consumer"
 	"github.com/uber/submitqueue/platform/extension/counter"
+	hookext "github.com/uber/submitqueue/platform/extension/hook"
+	platformhook "github.com/uber/submitqueue/platform/hook"
 	"github.com/uber/submitqueue/platform/pipeline"
 	"github.com/uber/submitqueue/submitqueue/core/topickey"
 	"github.com/uber/submitqueue/submitqueue/extension/buildrunner"
@@ -78,6 +83,11 @@ type Deps struct {
 
 	// Validator resolves the validator for each queue.
 	Validator validator.Factory
+
+	// Hooks resolves the hooks that receive each lifecycle event for
+	// fire-and-forget side effects. Wire a resolver returning noop when the
+	// deployment has no integrations.
+	Hooks hookext.Hooks
 }
 
 // Stages is the orchestrator's pipeline topology as a typed table.
@@ -223,6 +233,22 @@ var Stages = []pipeline.Stage[Deps]{
 			return dlq.NewDLQBatchController(d.Logger, d.Scope, d.Storage, sc.Registry, sc.TopicKey, sc.ConsumerGroup), nil
 		},
 	},
+	// Any stage can publish here and this one publishes nothing onward, so the
+	// row sits outside the flow the rest of the table is ordered by.
+	{
+		Key:           basehook.TopicKeyHook,
+		Name:          "submitqueue-hook",
+		ConsumerGroup: "orchestrator",
+		New: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
+			if d.Hooks == nil {
+				return nil, fmt.Errorf("hooks resolver is required; wire one returning noop when the deployment has no integrations")
+			}
+			return platformhook.NewController(d.Logger, d.Scope, d.Hooks, sc.TopicKey, sc.ConsumerGroup), nil
+		},
+		DLQ: func(d Deps, sc pipeline.StageContext) (consumer.Controller, error) {
+			return platformhook.NewDLQController(d.Logger, d.Scope, sc.TopicKey, sc.ConsumerGroup), nil
+		},
+	},
 }
 
 // PublishOnlyTopics declares topics the orchestrator publishes to but does

submitqueue/orchestrator/pipeline_test.go

@@ -0,0 +1,80 @@
+// 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 orchestrator
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber-go/tally"
+	basehook "github.com/uber/submitqueue/api/base/hook"
+	hookext "github.com/uber/submitqueue/platform/extension/hook"
+	hooknoop "github.com/uber/submitqueue/platform/extension/hook/noop"
+	"github.com/uber/submitqueue/platform/pipeline"
+	"go.uber.org/zap/zaptest"
+)
+
+type noopHooks struct{}
+
+func (noopHooks) For(*basehook.HookEvent) []hookext.Hook {
+	return []hookext.Hook{hooknoop.New()}
+}
+
+func hookStage(t *testing.T) pipeline.Stage[Deps] {
+	t.Helper()
+	for _, s := range Stages {
+		if s.Key == basehook.TopicKeyHook {
+			return s
+		}
+	}
+	require.FailNow(t, "no stage is registered for the hook topic key")
+	return pipeline.Stage[Deps]{}
+}
+
+func TestHookStage(t *testing.T) {
+	deps := func(t *testing.T) Deps {
+		return Deps{
+			Logger: zaptest.NewLogger(t).Sugar(),
+			Scope:  tally.NoopScope,
+			Hooks:  noopHooks{},
+		}
+	}
+	stageContext := func(key string) pipeline.StageContext {
+		return pipeline.StageContext{
+			TopicKey:      basehook.TopicKey(key),
+			ConsumerGroup: "orchestrator",
+		}
+	}
+
+	t.Run("the hook controller subscribes to the key the engine assigns", func(t *testing.T) {
+		controller, err := hookStage(t).New(deps(t), stageContext("hook"))
+		require.NoError(t, err)
+		assert.Equal(t, basehook.TopicKeyHook, controller.TopicKey())
+	})
+
+	t.Run("the dead-letter controller subscribes to the key the engine derives", func(t *testing.T) {
+		controller, err := hookStage(t).DLQ(deps(t), stageContext("hook_dlq"))
+		require.NoError(t, err)
+		assert.Equal(t, basehook.TopicKey("hook_dlq"), controller.TopicKey())
+	})
+
+	t.Run("a host that wires no hooks resolver fails to construct", func(t *testing.T) {
+		d := deps(t)
+		d.Hooks = nil
+		_, err := hookStage(t).New(d, stageContext("hook"))
+		require.Error(t, err)
+	})
+}