feat(buildrunner): noop impl, poll-driven buildsignal, pipeline wiring

### Why?

The `BuildRunner` interface needs an in-tree noop implementation to
unblock wiring tests, and the orchestrator's build stage needs to drive
the new contract end-to-end: trigger the runner, persist the result,
and poll `Status` until terminal so the batch state machine can react.

### What?

Stacks on top of the BuildRunner interface and `PublishAfter` PRs.
Implements the contract and wires it into the orchestrator pipeline.

The build poll loop runs as queue traffic inside the existing
`buildsignal` consumer (no separate stage). On each delivery it calls
`BuildRunner.Status`, persists the result via `BuildStore.UpdateStatus`,
publishes the batch ID to `speculate` so the state machine re-evaluates,
and re-publishes itself via `Publisher.PublishAfter` until the build
reaches a terminal state. A webhook-capable backend can publish into
the same topic — the consumer cannot tell a poll-driven message from a
push.

Pieces:

- `extension/buildrunner/noop`: a `BuildRunner` stub that returns
  `BuildStatusSucceeded` immediately. Useful as a wiring backstop and a
  best-case baseline.
- `orchestrator/controller/build`: assembles `base` from
  `batch.Dependencies` and `head` from `batch.Contains`, calls
  `Trigger`, persists the initial `Build{Accepted}` via
  `BuildStore.Create` (`ErrAlreadyExists` is swallowed for redelivery),
  publishes to `buildsignal`.
- `orchestrator/controller/buildsignal`: the polling consumer described
  above. `PollDelayAcceptedMs=5000`, `PollDelayRunningMs=2000` by
  default (vars so tests can override).
- `example/server/orchestrator/main.go`: passes the `BuildRunner` to
  both `build.NewController` and `buildsignal.NewController`; pipeline
  diagram updated.

Author: behinddwalls

core/consumer/registry.go

@@ -40,7 +40,11 @@ const (
 	TopicKeySpeculate TopicKey = "speculate"
 	// TopicKeyBuild is the pipeline stage where speculated batches are published for builds.
 	TopicKeyBuild TopicKey = "build"
-	// TopicKeyBuildSignal is the pipeline stage where builds are published for build signal processing.
+	// TopicKeyBuildSignal is the polling stage for triggered builds. Each
+	// message carries a Build; the consumer calls BuildRunner.Status,
+	// persists the latest status, publishes the batch ID to TopicKeySpeculate
+	// so the state machine re-evaluates, and re-publishes itself via
+	// PublishAfter when the build has not yet reached a terminal state.
 	TopicKeyBuildSignal TopicKey = "buildsignal"
 	// TopicKeyMerge is the pipeline stage where speculated batches are published for merging.
 	TopicKeyMerge TopicKey = "merge"

example/server/orchestrator/BUILD.bazel

@@ -14,6 +14,8 @@ go_library(
         "//core/consumer",
         "//core/httpclient",
         "//entity",
+        "//extension/buildrunner",
+        "//extension/buildrunner/noop",
         "//extension/changeprovider",
         "//extension/changeprovider/github",
         "//extension/changestore",

example/server/orchestrator/main.go

@@ -33,6 +33,8 @@ import (
 	"github.com/uber/submitqueue/core/consumer"
 	"github.com/uber/submitqueue/core/httpclient"
 	"github.com/uber/submitqueue/entity"
+	"github.com/uber/submitqueue/extension/buildrunner"
+	buildnoop "github.com/uber/submitqueue/extension/buildrunner/noop"
 	"github.com/uber/submitqueue/extension/changeprovider"
 	githubprovider "github.com/uber/submitqueue/extension/changeprovider/github"
 	"github.com/uber/submitqueue/extension/changestore"
@@ -216,8 +218,12 @@ func run() error {
 		return fmt.Errorf("failed to create pusher: %w", err)
 	}
 
+	// Create build runner. The noop runner is the pass-through default
+	// (every build immediately succeeds) until a real backend is wired in.
+	br := buildnoop.New()
+
 	// Register controllers
-	if err := registerControllers(c, logger.Sugar(), scope, registry, mc, cp, psh, cnt, store, changeStore); err != nil {
+	if err := registerControllers(c, logger.Sugar(), scope, registry, mc, cp, psh, br, cnt, store, changeStore); err != nil {
 		return err
 	}
 
@@ -397,12 +403,12 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 // Pipeline:
 //
 //   request → validate → batch → score → speculate → build → buildsignal ─┐
-//                                        ↑     ↘                           │
-//                                        │      merge → conclude           │
-//                                        │        │                        │
-//                                        └────────┴────────────────────────┘
+//                                        ↑     ↘             ↻ poll       │
+//                                        │      merge → conclude          │
+//                                        │        │                       │
+//                                        └────────┴───────────────────────┘
 
-func registerControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, mc mergechecker.MergeChecker, cp changeprovider.ChangeProvider, psh pusher.Pusher, cnt counter.Counter, store storage.Storage, changeStore changestore.ChangeStore) error {
+func registerControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, mc mergechecker.MergeChecker, cp changeprovider.ChangeProvider, psh pusher.Pusher, br buildrunner.BuildRunner, cnt counter.Counter, store storage.Storage, changeStore changestore.ChangeStore) error {
 	requestController := start.NewController(
 		logger,
 		scope,
@@ -488,6 +494,7 @@ func registerControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope t
 		logger,
 		scope,
 		store,
+		br,
 		registry,
 		consumer.TopicKeyBuild,
 		"orchestrator-build",
@@ -500,6 +507,7 @@ func registerControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope t
 		logger,
 		scope,
 		store,
+		br,
 		registry,
 		consumer.TopicKeyBuildSignal,
 		"orchestrator-buildsignal",

extension/buildrunner/noop/BUILD.bazel

@@ -0,0 +1,24 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "noop",
+    srcs = ["noop.go"],
+    importpath = "github.com/uber/submitqueue/extension/buildrunner/noop",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//entity",
+        "//extension/buildrunner",
+    ],
+)
+
+go_test(
+    name = "noop_test",
+    srcs = ["noop_test.go"],
+    embed = [":noop"],
+    deps = [
+        "//entity",
+        "//extension/buildrunner",
+        "@com_github_stretchr_testify//assert",
+        "@com_github_stretchr_testify//require",
+    ],
+)

extension/buildrunner/noop/noop.go

@@ -0,0 +1,56 @@
+// Copyright (c) 2025 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 noop provides a buildrunner.BuildRunner that performs no real
+// work: every triggered build immediately succeeds. It is intended as a
+// stub for wiring tests and as a best-case baseline where every build
+// passes.
+package noop
+
+import (
+	"context"
+	"fmt"
+	"sync/atomic"
+
+	"github.com/uber/submitqueue/entity"
+	"github.com/uber/submitqueue/extension/buildrunner"
+)
+
+// runner is a buildrunner.BuildRunner that does no real work and reports
+// every build as immediately succeeded. The atomic counter hands out
+// unique build IDs and makes the type safe for concurrent use.
+type runner struct {
+	counter atomic.Uint64
+}
+
+// New returns a buildrunner.BuildRunner that performs no real work.
+func New() buildrunner.BuildRunner {
+	return &runner{}
+}
+
+// Trigger returns a unique build ID without contacting any runner.
+// Inputs are ignored.
+func (r *runner) Trigger(_ context.Context, _ string, _ []entity.Change, _ []entity.Change, _ entity.BuildMetadata) (string, error) {
+	return fmt.Sprintf("noop-%d", r.counter.Add(1)), nil
+}
+
+// Status always reports BuildStatusSucceeded with no metadata.
+func (r *runner) Status(_ context.Context, _ string) (entity.BuildStatus, entity.BuildMetadata, error) {
+	return entity.BuildStatusSucceeded, nil, nil
+}
+
+// Cancel is a no-op.
+func (r *runner) Cancel(_ context.Context, _ string) error {
+	return nil
+}

extension/buildrunner/noop/noop_test.go

@@ -0,0 +1,61 @@
+// Copyright (c) 2025 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 noop
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/uber/submitqueue/entity"
+	"github.com/uber/submitqueue/extension/buildrunner"
+)
+
+func TestNew_ImplementsInterface(t *testing.T) {
+	var _ buildrunner.BuildRunner = New()
+}
+
+func TestRunner_Trigger(t *testing.T) {
+	r := New()
+	ctx := context.Background()
+
+	id1, err := r.Trigger(ctx, "queueA",
+		[]entity.Change{{URIs: []string{"github://owner/repo/pull/1"}}},
+		[]entity.Change{{URIs: []string{"github://owner/repo/pull/2"}}},
+		entity.BuildMetadata{"requester": "alice"},
+	)
+	require.NoError(t, err)
+	assert.NotEmpty(t, id1)
+
+	// IDs are unique across calls, even with empty inputs.
+	id2, err := r.Trigger(ctx, "queueA", nil, nil, nil)
+	require.NoError(t, err)
+	assert.NotEqual(t, id1, id2)
+}
+
+func TestRunner_Status(t *testing.T) {
+	r := New()
+
+	status, meta, err := r.Status(context.Background(), "any-id")
+	require.NoError(t, err)
+	assert.Equal(t, entity.BuildStatusSucceeded, status)
+	assert.Empty(t, meta)
+}
+
+func TestRunner_Cancel(t *testing.T) {
+	r := New()
+	assert.NoError(t, r.Cancel(context.Background(), "any-id"))
+}

orchestrator/controller/build/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "//core/metrics",
         "//entity",
         "//entity/queue",
+        "//extension/buildrunner",
         "//extension/storage",
         "@com_github_uber_go_tally_v4//:tally",
         "@org_uber_go_zap//:zap",
@@ -25,7 +26,11 @@ go_test(
         "//core/errs",
         "//entity",
         "//entity/queue",
+        "//extension/buildrunner",
+        "//extension/buildrunner/mock",
+        "//extension/buildrunner/noop",
         "//extension/queue/mock",
+        "//extension/storage",
         "//extension/storage/mock",
         "@com_github_stretchr_testify//assert",
         "@com_github_stretchr_testify//require",

orchestrator/controller/build/build.go

@@ -16,13 +16,15 @@ package build
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/uber-go/tally/v4"
 	"github.com/uber/submitqueue/core/consumer"
 	"github.com/uber/submitqueue/core/metrics"
 	"github.com/uber/submitqueue/entity"
 	entityqueue "github.com/uber/submitqueue/entity/queue"
+	"github.com/uber/submitqueue/extension/buildrunner"
 	"github.com/uber/submitqueue/extension/storage"
 	"go.uber.org/zap"
 )
@@ -34,6 +36,7 @@ type Controller struct {
 	logger        *zap.SugaredLogger
 	metricsScope  tally.Scope
 	store         storage.Storage
+	buildRunner   buildrunner.BuildRunner
 	registry      consumer.TopicRegistry
 	topicKey      consumer.TopicKey
 	consumerGroup string
@@ -47,6 +50,7 @@ func NewController(
 	logger *zap.SugaredLogger,
 	scope tally.Scope,
 	store storage.Storage,
+	buildRunner buildrunner.BuildRunner,
 	registry consumer.TopicRegistry,
 	topicKey consumer.TopicKey,
 	consumerGroup string,
@@ -55,6 +59,7 @@ func NewController(
 		logger:        logger.Named("build_controller"),
 		metricsScope:  scope.SubScope("build_controller"),
 		store:         store,
+		buildRunner:   buildRunner,
 		registry:      registry,
 		topicKey:      topicKey,
 		consumerGroup: consumerGroup,
@@ -95,17 +100,45 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 		"partition_key", msg.PartitionKey,
 	)
 
-	// TODO: Add build logic
-	// - Trigger CI build
-	// - Track build status
+	// Assemble base (dependency batches in order) and head (this batch).
+	base, err := c.collectChanges(ctx, batch.Dependencies)
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
+		return fmt.Errorf("failed to assemble base changes for batch %s: %w", batch.ID, err)
+	}
+	head, err := c.collectChanges(ctx, []string{batch.ID})
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
+		return fmt.Errorf("failed to assemble head changes for batch %s: %w", batch.ID, err)
+	}
+
+	// Trigger the build with the configured build manager. metadata is nil
+	// until a caller-supplied source materializes (e.g. requester / ticket
+	// pulled off the originating LandRequest).
+	buildID, err := c.buildRunner.Trigger(ctx, batch.Queue, base, head, nil)
+	if err != nil {
+		metrics.NamedCounter(c.metricsScope, opName, "trigger_errors", 1)
+		return fmt.Errorf("failed to trigger build for batch %s: %w", batch.ID, err)
+	}
 
 	build := entity.Build{
-		ID:      batch.ID,
-		BatchID: batch.ID,
-		Status:  entity.BuildStatusAccepted,
+		ID:              buildID,
+		BatchID:         batch.ID,
+		SpeculationPath: entity.SpeculationPathInfo{Base: append([]string{}, batch.Dependencies...)},
+		Status:          entity.BuildStatusAccepted,
+	}
+
+	// Persist the initial Build snapshot so the buildsignal poll loop has a
+	// row to UpdateStatus against. ErrAlreadyExists is benign — a redelivery
+	// of this message after a previous successful Create.
+	if err := c.store.GetBuildStore().Create(ctx, build); err != nil && !errors.Is(err, storage.ErrAlreadyExists) {
+		metrics.NamedCounter(c.metricsScope, opName, "storage_errors", 1)
+		return fmt.Errorf("failed to persist build %s: %w", build.ID, err)
 	}
 
-	// Publish build to build signal topic
+	// Hand off to the buildsignal poll loop; it calls Status, updates the
+	// persisted Build, publishes to speculate, and re-publishes itself via
+	// PublishAfter until terminal.
 	if err := c.publish(ctx, consumer.TopicKeyBuildSignal, build); err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "publish_errors", 1)
 		return fmt.Errorf("failed to publish to buildsignal: %w", err)
@@ -114,12 +147,37 @@ func (c *Controller) Process(ctx context.Context, delivery consumer.Delivery) (r
 	c.logger.Infow("published build to buildsignal",
 		"batch_id", batch.ID,
 		"build_id", build.ID,
+		"status", string(build.Status),
 		"topic_key", consumer.TopicKeyBuildSignal,
 	)
 
 	return nil // Success - message will be acked
 }
 
+// collectChanges loads each batch by ID and concatenates the Change values
+// from its contained requests in batch order. Used to build the base
+// (dependency batches) and head (this batch) inputs to BuildRunner.Trigger.
+func (c *Controller) collectChanges(ctx context.Context, batchIDs []string) ([]entity.Change, error) {
+	if len(batchIDs) == 0 {
+		return nil, nil
+	}
+	var changes []entity.Change
+	for _, bID := range batchIDs {
+		b, err := c.store.GetBatchStore().Get(ctx, bID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to get batch %s: %w", bID, err)
+		}
+		for _, reqID := range b.Contains {
+			req, err := c.store.GetRequestStore().Get(ctx, reqID)
+			if err != nil {
+				return nil, fmt.Errorf("failed to get request %s for batch %s: %w", reqID, bID, err)
+			}
+			changes = append(changes, req.Change)
+		}
+	}
+	return changes, nil
+}
+
 // publish publishes a build to the specified topic key.
 func (c *Controller) publish(ctx context.Context, key consumer.TopicKey, build entity.Build) error {
 	payload, err := build.ToBytes()