feat(speculate): write the speculation tree in shadow mode

## Summary

### Why?

This is the first of four slices reproducing the tree-driven speculate/build/buildsignal rework as an incremental, always-shippable stack instead of one large commit. Each slice keeps the orchestrator working end to end so the e2e suite stays green throughout, and reviewers can follow the pipeline shift (direct publish -> tree -> prioritize -> build) one deliberate step at a time.

This slice only adds the speculation tree as bookkeeping: nothing downstream reads it yet, so the batch's forward step (when it builds, when it merges) is untouched — with one deliberate exception called out below: a liveness fix to the dependent wake-up that review of this slice surfaced.

### What?

speculate's Controller now takes an enumerator, path scorer, selector, and dependency-limit factory, and every Created/Scored/Speculating pass loads or creates the batch's entity.SpeculationTree, applies the scorer's and selector's outputs, and persists it only if something changed (the apply steps report whether they mutated anything, gating a version+1 conditional Update). The controller stays pure mechanics: which paths exist, how they score, and which are promoted or cancelled are the seams' decisions alone — the controller validates and records them. Tree creation is gated by the queue's dependency limit; the controller wraps the enumerator's structure-only paths into persisted entries itself — stamping Candidate, minting each path's immutable ID as `{batchID}/path/{i}`, and skipping structural duplicates as a contract violation — and a concurrent create race re-reads the winner's tree instead of erroring. Seam outputs are consumed by path ID: scores merged by ID (unknown IDs skipped, out-of-[0,1] values clamped, both logged), selector decisions resolved through the ID-keyed PathIndex with duplicates logged and skipped.

The pre-existing forward step is otherwise unchanged: Created/Scored batches CAS to Speculating and publish straight to build, and Speculating batches run the original tryFinalize (merge once every dependency has landed, cascade-fail on a failed dependency). Cancelling is the unmodified pre-existing flow.

Liveness fix in the forward step: a batch waiting on dependencies (in tryFinalize, or now in the dependency gate) was only ever woken when a dependency was cancelled — a dependency reaching Succeeded or Failed never re-published its dependents, because mergesignal routes every terminal transition back through speculate under the batch's own ID and the terminal branch only re-published conclude. The terminal branch now fans out to dependents for every terminal state, and failOnDependency wakes its own dependents right after the terminal CAS so failures cascade downstream. The fan-out publishes a wake for every listed dependent; a dangling reverse-index entry (left by an abandoned batch creation) surfaces as a not-found speculate message that dead-letters and is skipped by the DLQ reconciler — eliminating that class at the source, by creating the batch before the reverse-index update, is tracked in #354.

Cancel decisions on tree paths are recorded as status only (Cancelling on an in-flight build) — nothing in this controller calls a build runner, which is why it takes no buildrunner.Factory.

main.go wires the new seams with parity defaults: chain enumerates a single path per batch, the probability path scorer, the all-selector promotes every candidate, and a static dependency limit that is effectively ungated.

## Test Plan

`bazel test //submitqueue/... //service/...` — 56/56 targets pass.

New unit coverage for the wake-up fix: dependent fan-out on every terminal state (with publish-order assertions), missing BatchDependent row surfaced as an invariant error, and dependent-publish failure after the terminal CAS nacking for redelivery convergence.

`make gazelle && make fmt` — no diffs beyond the intended BUILD file updates.

## Issue

Part of the speculation rework. Interim fan-out caveats tracked in #354.

Author: behinddwalls

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -41,9 +41,17 @@ go_library(
         "//submitqueue/extension/scorer/composite:go_default_library",
         "//submitqueue/extension/scorer/fake:go_default_library",
         "//submitqueue/extension/scorer/heuristic:go_default_library",
+        "//submitqueue/extension/speculation/dependencylimit:go_default_library",
+        "//submitqueue/extension/speculation/dependencylimit/static:go_default_library",
+        "//submitqueue/extension/speculation/enumerator:go_default_library",
+        "//submitqueue/extension/speculation/enumerator/chain:go_default_library",
+        "//submitqueue/extension/speculation/pathscorer:go_default_library",
+        "//submitqueue/extension/speculation/pathscorer/probability:go_default_library",
         "//submitqueue/extension/speculation/prioritizationlimit/static:go_default_library",
         "//submitqueue/extension/speculation/prioritizer:go_default_library",
         "//submitqueue/extension/speculation/prioritizer/sticky:go_default_library",
+        "//submitqueue/extension/speculation/selector:go_default_library",
+        "//submitqueue/extension/speculation/selector/all: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

@@ -61,9 +61,17 @@ import (
 	"github.com/uber/submitqueue/submitqueue/extension/scorer/composite"
 	scorerfake "github.com/uber/submitqueue/submitqueue/extension/scorer/fake"
 	"github.com/uber/submitqueue/submitqueue/extension/scorer/heuristic"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/dependencylimit"
+	dependencylimitstatic "github.com/uber/submitqueue/submitqueue/extension/speculation/dependencylimit/static"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/enumerator"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/enumerator/chain"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/pathscorer"
+	pathscorerprobability "github.com/uber/submitqueue/submitqueue/extension/speculation/pathscorer/probability"
 	prioritizationlimitstatic "github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizationlimit/static"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizer"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/prioritizer/sticky"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/selector"
+	selectorall "github.com/uber/submitqueue/submitqueue/extension/speculation/selector/all"
 	"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"
@@ -238,7 +246,7 @@ func run() error {
 	// back to a baseline profile for queues without an explicit entry. This is
 	// the single place queue topology is known; the extension packages stay
 	// queue-agnostic.
-	queues, err := newQueueRegistry(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore()))
+	queues, err := newQueueRegistry(logger, scope, changeset.New(store.GetRequestStore(), store.GetChangeStore()), store.GetBatchStore())
 	if err != nil {
 		return fmt.Errorf("failed to build queue registry: %w", err)
 	}
@@ -249,9 +257,13 @@ func run() error {
 	scf := scorerFactory{queues}
 	cof := analyzerFactory{queues}
 	prf := prioritizerFactory{queues}
+	enf := enumeratorFactory{queues}
+	ssf := speculationScorerFactory{queues}
+	slf := speculationSelectorFactory{queues}
+	dlf := dependencyLimitFactory{queues}
 
 	// Register controllers
-	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, registry, cpf, brf, scf, cof, prf, cnt, store)
+	primaryCount, err := registerPrimaryControllers(primaryConsumer, logger.Sugar(), scope, registry, cpf, brf, scf, cof, prf, enf, ssf, slf, dlf, cnt, store)
 	if err != nil {
 		return err
 	}
@@ -502,11 +514,15 @@ func newTopicRegistry(q extqueue.Queue, subscriberName string) (consumer.TopicRe
 // read as "for this queue, here are its scorer, analyzer, change provider, …", and lets
 // a queue profile start from a baseline and override only what differs.
 type queueExtensions struct {
-	changeProvider changeprovider.ChangeProvider
-	buildRunner    buildrunner.BuildRunner
-	scorer         scorer.Scorer
-	analyzer       conflict.Analyzer
-	prioritizer    prioritizer.Prioritizer
+	changeProvider      changeprovider.ChangeProvider
+	buildRunner         buildrunner.BuildRunner
+	scorer              scorer.Scorer
+	analyzer            conflict.Analyzer
+	prioritizer         prioritizer.Prioritizer
+	enumerator          enumerator.Enumerator
+	speculationScorer   pathscorer.Scorer
+	speculationSelector selector.Selector
+	dependencyLimit     dependencylimit.DependencyLimit
 }
 
 // queueRegistry maps a queue name to its extensions, falling back to a default
@@ -558,7 +574,31 @@ func (f prioritizerFactory) For(cfg prioritizer.Config) (prioritizer.Prioritizer
 	return f.reg.get(cfg.QueueName).prioritizer, nil
 }
 
-func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, cpf changeprovider.Factory, brf buildrunner.Factory, scf scorer.Factory, cof conflict.Factory, prf prioritizer.Factory, cnt counter.Counter, store storage.Storage) (int, error) {
+type enumeratorFactory struct{ reg queueRegistry }
+
+func (f enumeratorFactory) For(cfg enumerator.Config) (enumerator.Enumerator, error) {
+	return f.reg.get(cfg.QueueName).enumerator, nil
+}
+
+type speculationScorerFactory struct{ reg queueRegistry }
+
+func (f speculationScorerFactory) For(cfg pathscorer.Config) (pathscorer.Scorer, error) {
+	return f.reg.get(cfg.QueueName).speculationScorer, nil
+}
+
+type speculationSelectorFactory struct{ reg queueRegistry }
+
+func (f speculationSelectorFactory) For(cfg selector.Config) (selector.Selector, error) {
+	return f.reg.get(cfg.QueueName).speculationSelector, nil
+}
+
+type dependencyLimitFactory struct{ reg queueRegistry }
+
+func (f dependencyLimitFactory) For(cfg dependencylimit.Config) (dependencylimit.DependencyLimit, error) {
+	return f.reg.get(cfg.QueueName).dependencyLimit, nil
+}
+
+func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger, scope tally.Scope, registry consumer.TopicRegistry, cpf changeprovider.Factory, brf buildrunner.Factory, scf scorer.Factory, cof conflict.Factory, prf prioritizer.Factory, enf enumerator.Factory, ssf pathscorer.Factory, slf selector.Factory, dlf dependencylimit.Factory, cnt counter.Counter, store storage.Storage) (int, error) {
 	var count int
 	requestController := start.NewController(
 		logger,
@@ -648,6 +688,10 @@ func registerPrimaryControllers(c consumer.Consumer, logger *zap.SugaredLogger,
 		logger,
 		scope,
 		store,
+		enf,
+		ssf,
+		slf,
+		dlf,
 		registry,
 		topickey.TopicKeySpeculate,
 		"orchestrator-speculate",
@@ -899,14 +943,19 @@ func newPhabChangeProvider(logger *zap.Logger, scope tally.Scope) (changeprovide
 // effectively admit-all — until per-queue budgets are configured.
 const defaultPrioritizationLimit = 1000
 
+// defaultDependencyLimit is the baseline cap on active dependencies a batch
+// may speculate over. It is a parity default — effectively ungated — until
+// per-queue limits are configured.
+const defaultDependencyLimit = 1000
+
 // newQueueRegistry builds the per-queue extension profiles for the example.
 // Edge integrations (change provider) and the build
 // runner form a shared baseline; each per-queue profile starts from that
 // baseline and overrides only the extensions that differ — here the scorer and
 // conflict analyzer. Queues without an explicit profile fall back to the
 // baseline. This is the one place queue topology lives; extension packages stay
 // queue-agnostic.
-func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver) (queueRegistry, error) {
+func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.Resolver, batchStore storage.BatchStore) (queueRegistry, error) {
 	cp, err := newChangeProvider(logger, scope)
 	if err != nil {
 		return queueRegistry{}, fmt.Errorf("failed to create change provider: %w", err)
@@ -933,6 +982,12 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.
 	// below does. The prioritizer is sticky over a static budget: it never
 	// preempts a running build and admits Selected candidates by score until
 	// defaultPrioritizationLimit concurrent builds are in flight.
+	//
+	// The speculation seams default to the single-chain parity policies:
+	// chain enumerates one path per batch (built on the full ordered
+	// dependency chain), probability scores paths from the batches'
+	// predicted-success probabilities and resolved outcomes, all promotes
+	// every candidate, and the dependency limit is effectively ungated.
 	base := queueExtensions{
 		changeProvider: cp,
 		buildRunner:    buildfake.New(resolver),
@@ -943,8 +998,12 @@ func newQueueRegistry(logger *zap.Logger, scope tally.Scope, resolver changeset.
 		)),
 		// TODO: replace the delegate with a real analyzer (e.g. Tango target
 		// analysis). "all" serializes the queue conservatively.
-		analyzer:    conflictfake.New(all.New(), nil),
-		prioritizer: sticky.New(prioritizationlimitstatic.New(defaultPrioritizationLimit)),
+		analyzer:            conflictfake.New(all.New(), nil),
+		prioritizer:         sticky.New(prioritizationlimitstatic.New(defaultPrioritizationLimit)),
+		enumerator:          chain.New(),
+		speculationScorer:   pathscorerprobability.New(batchStore),
+		speculationSelector: selectorall.New(),
+		dependencyLimit:     dependencylimitstatic.New(defaultDependencyLimit),
 	}
 
 	// test-queue: bucketed heuristic scorer; conservative (serialized) conflicts

submitqueue/orchestrator/controller/speculate/BUILD.bazel

@@ -11,6 +11,10 @@ go_library(
         "//platform/metrics:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/dependencylimit:go_default_library",
+        "//submitqueue/extension/speculation/enumerator:go_default_library",
+        "//submitqueue/extension/speculation/pathscorer:go_default_library",
+        "//submitqueue/extension/speculation/selector:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "@com_github_uber_go_tally//:go_default_library",
         "@org_uber_go_zap//:go_default_library",
@@ -28,6 +32,14 @@ go_test(
         "//platform/extension/messagequeue/mock:go_default_library",
         "//submitqueue/core/topickey:go_default_library",
         "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/dependencylimit/fake:go_default_library",
+        "//submitqueue/extension/speculation/dependencylimit/mock:go_default_library",
+        "//submitqueue/extension/speculation/enumerator/fake:go_default_library",
+        "//submitqueue/extension/speculation/enumerator/mock:go_default_library",
+        "//submitqueue/extension/speculation/pathscorer/fake:go_default_library",
+        "//submitqueue/extension/speculation/pathscorer/mock:go_default_library",
+        "//submitqueue/extension/speculation/selector/fake:go_default_library",
+        "//submitqueue/extension/speculation/selector/mock:go_default_library",
         "//submitqueue/extension/storage:go_default_library",
         "//submitqueue/extension/storage/mock:go_default_library",
         "@com_github_stretchr_testify//assert:go_default_library",