feat(orchestrator): re-plan the queue from the Speculator each run

## Summary

### Why?

The speculate controller had no speculation in it: it advanced one batch at a time along a single hard-coded chain — every dependency assumed to pass — and nothing ever called the Speculator, the Generator, or the Allocator that had been built for it. This wires that machinery into the pipeline: a queue's paths are chosen by a swappable extension within a build budget, and each is built against only the dependencies it assumes will succeed.

### What?

Every message is a dirty signal naming a batch; the controller re-plans that batch's whole queue from a single read: read the state, cancel paths whose assumptions a finished dependency has proven wrong, ask the Speculator, filter its proposals, dispatch what survives. Nothing carries over between runs, so duplicated or reordered signals are harmless and a later run repairs whatever an earlier one left half-done. `doc.go` explains the model in plain terms — no vocabulary section, every term is defined where it is used.

The path set keeps exactly one writer — this run. The build stages record what CI did on per-build records; the run folds those into the set and alone decides each path's status. Pending paths are re-dispatched every run until their build is seen running, dispatches partition by batch so heads proceed in parallel, and cancelling paths need no dispatch at all: the poll loop reads the stop off the set and enacts it.

Speculation lands inert. The wiring layer passes a placeholder Speculator that proposes nothing, so the run executes end to end but funds no paths; composing real per-queue speculators and turning the feature on is the wiring change at the top of this stack. Batch outcomes still come from the legacy per-batch finalizer, which waits on every dependency — strictly stricter than path-aware finalization, so the system stays correct until the next commit replaces it.

## Test Plan

✅ `bazel test //submitqueue/orchestrator/controller/speculate/...` — assumption checks and proposal filtering are table-driven; run tests cover funding a first path, re-dispatching pending paths, broken-path cancellation, build results recorded onto paths, lost CAS races skipped rather than failed, and Speculator errors abandoning the run.

✅ `make fmt`, `make gazelle`

# Conflicts:
#	service/submitqueue/orchestrator/server/BUILD.bazel
#	service/submitqueue/orchestrator/server/main.go
#	submitqueue/orchestrator/controller/speculate/BUILD.bazel
#	submitqueue/orchestrator/controller/speculate/speculate.go
#	submitqueue/orchestrator/controller/speculate/speculate_test.go

# Please enter the commit message for your changes. Lines starting
# with '#' will be kept; you may remove them yourself if you want to.
# An empty message aborts the commit.
#
# interactive rebase in progress; onto 452922a9
# Last command done (1 command done):
#    pick b9f222a # feat(orchestrator): re-plan the queue from the Speculator each run
# No commands remaining.
# You are currently rebasing branch 'preetam/speculation-run' on '452922a9'.
#
# Changes to be committed:
#	modified:   service/submitqueue/orchestrator/server/BUILD.bazel
#	modified:   service/submitqueue/orchestrator/server/main.go
#	modified:   submitqueue/orchestrator/BUILD.bazel
#	modified:   submitqueue/orchestrator/controller/speculate/BUILD.bazel
#	new file:   submitqueue/orchestrator/controller/speculate/check.go
#	new file:   submitqueue/orchestrator/controller/speculate/check_test.go
#	new file:   submitqueue/orchestrator/controller/speculate/dispatch.go
#	new file:   submitqueue/orchestrator/controller/speculate/doc.go
#	new file:   submitqueue/orchestrator/controller/speculate/run.go
#	new file:   submitqueue/orchestrator/controller/speculate/run_test.go
#	new file:   submitqueue/orchestrator/controller/speculate/snapshot.go
#	new file:   submitqueue/orchestrator/controller/speculate/snapshot_test.go
#	modified:   submitqueue/orchestrator/controller/speculate/speculate.go
#	modified:   submitqueue/orchestrator/controller/speculate/speculate_test.go
#	modified:   submitqueue/orchestrator/pipeline.go
#

# Conflicts:
#	service/submitqueue/orchestrator/server/main.go

# Please enter the commit message for your changes. Lines starting
# with '#' will be kept; you may remove them yourself if you want to.
# An empty message aborts the commit.
#
# interactive rebase in progress; onto fc33bc5
# Last command done (1 command done):
#    pick d549bd7 # feat(orchestrator): re-plan the queue from the Speculator each run
# No commands remaining.
# You are currently rebasing branch 'preetam/speculation-run' on 'fc33bc54'.
#
# Changes to be committed:
#	modified:   service/submitqueue/orchestrator/server/BUILD.bazel
#	modified:   service/submitqueue/orchestrator/server/main.go
#	modified:   submitqueue/orchestrator/BUILD.bazel
#	modified:   submitqueue/orchestrator/controller/speculate/BUILD.bazel
#	new file:   submitqueue/orchestrator/controller/speculate/check.go
#	new file:   submitqueue/orchestrator/controller/speculate/check_test.go
#	new file:   submitqueue/orchestrator/controller/speculate/dispatch.go
#	new file:   submitqueue/orchestrator/controller/speculate/doc.go
#	new file:   submitqueue/orchestrator/controller/speculate/run.go
#	new file:   submitqueue/orchestrator/controller/speculate/run_test.go
#	new file:   submitqueue/orchestrator/controller/speculate/snapshot.go
#	new file:   submitqueue/orchestrator/controller/speculate/snapshot_test.go
#	modified:   submitqueue/orchestrator/controller/speculate/speculate.go
#	modified:   submitqueue/orchestrator/controller/speculate/speculate_test.go
#	modified:   submitqueue/orchestrator/pipeline.go
#

Author: behinddwalls

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -26,6 +26,7 @@ go_library(
         "//platform/http:go_default_library",
         "//platform/pipeline:go_default_library",
         "//submitqueue/core/changeset:go_default_library",
+        "//submitqueue/entity:go_default_library",
         "//submitqueue/extension/buildrunner:go_default_library",
         "//submitqueue/extension/buildrunner/fake:go_default_library",
         "//submitqueue/extension/changeprovider:go_default_library",
@@ -38,6 +39,7 @@ go_library(
         "//submitqueue/extension/conflict/fake:go_default_library",
         "//submitqueue/extension/conflict/fileoverlap:go_default_library",
         "//submitqueue/extension/conflict/none:go_default_library",
+        "//submitqueue/extension/speculation/speculator:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mysql:go_default_library",
         "//submitqueue/extension/validator/fake:go_default_library",

service/submitqueue/orchestrator/server/main.go

@@ -43,11 +43,13 @@ import (
 	"github.com/uber/submitqueue/platform/http"
 	"github.com/uber/submitqueue/platform/pipeline"
 	"github.com/uber/submitqueue/submitqueue/core/changeset"
+	"github.com/uber/submitqueue/submitqueue/entity"
 	"github.com/uber/submitqueue/submitqueue/extension/changeprovider"
 	cpfake "github.com/uber/submitqueue/submitqueue/extension/changeprovider/fake"
 	githubprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/github"
 	phabprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/phabricator"
 	routingprovider "github.com/uber/submitqueue/submitqueue/extension/changeprovider/routing"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
 	"github.com/uber/submitqueue/submitqueue/extension/storage"
 	mysqlstorage "github.com/uber/submitqueue/submitqueue/extension/storage/mysql"
 	validatorfake "github.com/uber/submitqueue/submitqueue/extension/validator/fake"
@@ -199,7 +201,12 @@ func run() error {
 		BuildRunner:    profiles.BuildRunnerFactory(),
 		ChangeProvider: profiles.ChangeProviderFactory(),
 		Analyzer:       profiles.AnalyzerFactory(),
-		Validator:      validatorfake.NewFactory(),
+		// Speculation is wired but inert: the placeholder below proposes
+		// nothing, so no path is ever funded and no speculative build starts.
+		// The wiring change at the top of this stack replaces it with real
+		// per-queue speculators composed from each profile's scorer.
+		Speculator: noopSpeculators{},
+		Validator:  validatorfake.NewFactory(),
 	}
 
 	// Assemble the pipeline: one call builds the topic registry, creates
@@ -464,3 +471,20 @@ func (f counterFactory) For(config counter.Config) (counter.Counter, error) {
 	}
 	return mysqlcounter.NewCounter(f.db, f.scope, config.QueueName), nil
 }
+
+// noopSpeculators resolves every queue to a speculator that proposes nothing.
+// It keeps the speculate stage inert — no path funded, no build started —
+// until per-queue speculators are composed in the profiles.
+type noopSpeculators struct{}
+
+// For returns the propose-nothing speculator for any queue.
+func (noopSpeculators) For(speculator.Config) (speculator.Speculator, error) {
+	return noopSpeculator{}, nil
+}
+
+type noopSpeculator struct{}
+
+// Speculate proposes no actions, whatever the queue looks like.
+func (noopSpeculator) Speculate(context.Context, []entity.Batch, []entity.SpeculationPathSet) ([]entity.Speculation, error) {
+	return nil, nil
+}

submitqueue/orchestrator/BUILD.bazel

@@ -14,6 +14,7 @@ go_library(
         "//submitqueue/extension/buildrunner:go_default_library",
         "//submitqueue/extension/changeprovider:go_default_library",
         "//submitqueue/extension/conflict:go_default_library",
+        "//submitqueue/extension/speculation/speculator:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/validator:go_default_library",
         "//submitqueue/orchestrator/controller:go_default_library",

submitqueue/orchestrator/controller/speculate/BUILD.bazel

@@ -2,16 +2,24 @@ load("@rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["speculate.go"],
+    srcs = [
+        "check.go",
+        "dispatch.go",
+        "doc.go",
+        "run.go",
+        "snapshot.go",
+        "speculate.go",
+    ],
     importpath = "github.com/uber/submitqueue/submitqueue/orchestrator/controller/speculate",
     visibility = ["//visibility:public"],
     deps = [
-        "//platform/base/messagequeue:go_default_library",
         "//platform/consumer:go_default_library",
         "//platform/metrics:go_default_library",
         "//submitqueue/core/batch:go_default_library",
+        "//submitqueue/core/publish:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/speculator:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
@@ -20,7 +28,12 @@ go_library(
 
 go_test(
     name = "go_default_test",
-    srcs = ["speculate_test.go"],
+    srcs = [
+        "check_test.go",
+        "run_test.go",
+        "snapshot_test.go",
+        "speculate_test.go",
+    ],
     embed = [":go_default_library"],
     deps = [
         "//platform/base/messagequeue:go_default_library",
@@ -30,6 +43,7 @@ go_test(
         "//platform/extension/messagequeue/mock:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/speculator:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",

submitqueue/orchestrator/controller/speculate/check.go

@@ -0,0 +1,169 @@
+// 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 speculate
+
+import "github.com/uber/submitqueue/submitqueue/entity"
+
+// rejection is why one proposed action was dropped. The reasons are the metric
+// dimension for a misbehaving Speculator, so each names a distinct fault.
+type rejection string
+
+const (
+	// rejectUnknownAction is a zero-value or unrecognized action.
+	rejectUnknownAction rejection = "unknown_action"
+	// rejectUnknownHead names a batch this run never read.
+	rejectUnknownHead rejection = "unknown_head"
+	// rejectHeadNotSpeculating targets a batch that is not open to new work.
+	rejectHeadNotSpeculating rejection = "head_not_speculating"
+	// rejectMalformedPath is a path whose assumptions do not line up with its
+	// head's dependency list: one missing or extra, a duplicate, a wrong head,
+	// or a made-up assumption value.
+	rejectMalformedPath rejection = "malformed_path"
+	// rejectBrokenAssumption is a path with an assumption a finished
+	// dependency has already proven wrong.
+	rejectBrokenAssumption rejection = "broken_assumption"
+	// rejectPathTerminal would rebuild a path whose build already finished.
+	rejectPathTerminal rejection = "path_terminal"
+	// rejectCancelNotInFlight cancels a path that is not running.
+	rejectCancelNotInFlight rejection = "cancel_not_in_flight"
+	// rejectCancelPassed would throw away a build that already passed.
+	rejectCancelPassed rejection = "cancel_passed"
+)
+
+// filterProposals narrows a Speculator's proposals down to the ones the
+// controller is willing to enact, returning the survivors and a reason for
+// each drop.
+//
+// The Speculator is an extension, so its output is untrusted input: it decides
+// which paths run, never whether a batch merges or fails. Every rule here
+// protects an invariant the extension could otherwise break — acting on a batch
+// that is finalizing, resurrecting a path a resolved dependency has ruled out,
+// or discarding a passed build the queue is about to merge on. A proposal that
+// trips one of these is a bug in the Speculator, not a normal outcome, which is
+// why the caller counts them.
+func filterProposals(proposals []entity.Speculation, snap snapshot) ([]entity.Speculation, []rejection) {
+	var kept []entity.Speculation
+	var rejected []rejection
+
+	for _, proposal := range proposals {
+		if reason, ok := rejectionReason(proposal, snap); ok {
+			rejected = append(rejected, reason)
+			continue
+		}
+		kept = append(kept, proposal)
+	}
+
+	return kept, rejected
+}
+
+// rejectionReason reports why a proposal cannot be enacted, or ok=false if
+// it can.
+func rejectionReason(proposal entity.Speculation, snap snapshot) (rejection, bool) {
+	switch proposal.Action {
+	case entity.PathActionBuild, entity.PathActionCancel:
+	default:
+		return rejectUnknownAction, true
+	}
+
+	head, ok := snap.batches[proposal.Path.Head]
+	if !ok {
+		return rejectUnknownHead, true
+	}
+	// Only a speculating head is open to new work. Batches in every other
+	// state are facts the Speculator may reason from, never action targets.
+	if head.State != entity.BatchStateSpeculating {
+		return rejectHeadNotSpeculating, true
+	}
+	if !isWellFormed(proposal.Path, head) {
+		return rejectMalformedPath, true
+	}
+	if assumptionBroken(proposal.Path, snap) {
+		return rejectBrokenAssumption, true
+	}
+
+	entry, stored := findPath(snap.pathSets[head.ID], proposal.Path.ID())
+
+	if proposal.Action == entity.PathActionCancel {
+		// Cancelling is only meaningful for a path that is actually running,
+		// and never for one that passed: that build is the head's way out of
+		// the queue, and the budget it holds is already spent.
+		if !stored {
+			return rejectCancelNotInFlight, true
+		}
+		if entry.Status == entity.SpeculationPathStatusPassed {
+			return rejectCancelPassed, true
+		}
+		if entry.Status.IsTerminal() {
+			return rejectCancelNotInFlight, true
+		}
+		return "", false
+	}
+
+	// A build proposal for a path whose build already finished would discard a
+	// recorded result and start the same work again.
+	if stored && entry.Status.IsTerminal() {
+		return rejectPathTerminal, true
+	}
+
+	return "", false
+}
+
+// isWellFormed reports whether a path is a proper guess about its head:
+// exactly one assumption for each of the head's dependencies, no more and no
+// fewer, and every assumption a real value.
+//
+// A malformed path is not merely suboptimal, it is unmergeable — the merge
+// preconditions are read off the path's assumptions, so a path missing a
+// dependency would let its head merge without waiting for it.
+func isWellFormed(path entity.SpeculationPath, head entity.Batch) bool {
+	if path.Head != head.ID {
+		return false
+	}
+	if len(path.Dependencies) != len(head.Dependencies) {
+		return false
+	}
+
+	required := make(map[string]struct{}, len(head.Dependencies))
+	for _, dep := range head.Dependencies {
+		required[dep] = struct{}{}
+	}
+
+	for _, dep := range path.Dependencies {
+		if _, ok := required[dep.Batch]; !ok {
+			return false
+		}
+		delete(required, dep.Batch)
+
+		switch dep.Assumption {
+		case entity.DependencyAssumptionSucceeds,
+			entity.DependencyAssumptionFails,
+			entity.DependencyAssumptionIgnored:
+		default:
+			return false
+		}
+	}
+
+	return len(required) == 0
+}
+
+// findPath returns the entry for a path ID in the set.
+func findPath(set entity.SpeculationPathSet, pathID string) (entity.SpeculationPathEntry, bool) {
+	for _, entry := range set.Paths {
+		if entry.ID == pathID {
+			return entry, true
+		}
+	}
+	return entity.SpeculationPathEntry{}, false
+}