feat(speculation): wire predictor into speculation pipeline

## Summary

### Why?

The predictor implementation is additive until the speculation pipeline supplies each dependency's run-local path evidence and uses the revised probability for ranking.

### What?

Thread path sets through the Generator contract and standard Speculator, replace `bestfirst`'s scorer dependency with the predictor, and compose the evidence predictor from per-queue YAML configuration in orchestrator profiles. Neutral default factors preserve scorer-only ranking when no factors are configured.

## Test Plan

- `bazel test //submitqueue/extension/speculation/generator/... //submitqueue/extension/speculation/speculator/... //service/submitqueue/orchestrator/server:go_default_test`
- `make check-gazelle`

Author: behinddwalls

service/submitqueue/orchestrator/server/BUILD.bazel

@@ -52,6 +52,8 @@ go_library(
         "//submitqueue/extension/conflict/pathoverlap:go_default_library",
         "//submitqueue/extension/speculation/allocator/sticky:go_default_library",
         "//submitqueue/extension/speculation/generator/bestfirst:go_default_library",
+        "//submitqueue/extension/speculation/predictor:go_default_library",
+        "//submitqueue/extension/speculation/predictor/evidence:go_default_library",
         "//submitqueue/extension/speculation/scorer:go_default_library",
         "//submitqueue/extension/speculation/scorer/composite:go_default_library",
         "//submitqueue/extension/speculation/scorer/fake:go_default_library",
@@ -121,6 +123,7 @@ go_test(
         "//submitqueue/extension/buildrunner:go_default_library",
         "//submitqueue/extension/changeprovider:go_default_library",
         "//submitqueue/extension/conflict:go_default_library",
+        "//submitqueue/extension/speculation/predictor:go_default_library",
         "//submitqueue/extension/speculation/scorer:go_default_library",
         "//submitqueue/extension/speculation/speculator:go_default_library",
         "//submitqueue/extension/storage:go_default_library",

service/submitqueue/orchestrator/server/config.go

@@ -67,6 +67,22 @@ const (
 // Ways a composite scorer combines its components.
 const combineAvg = "avg"
 
+// Predictor types selectable from configuration.
+const predictorTypeEvidence = "evidence"
+
+// Evidence an evidence predictor prices, as named in configuration. The set is
+// closed: a factor under any other name would be applied to nothing and never
+// noticed.
+const (
+	factorPathPassed = "pathPassed"
+	factorPathFailed = "pathFailed"
+	factorMerging    = "merging"
+	factorCancelling = "cancelling"
+)
+
+// neutralFactor leaves the scorer's price untouched: odds multiplied by one.
+const neutralFactor = 1.0
+
 // defaultBuildBudget is how many builds a queue may have occupying CI at once
 // when it states no budget of its own. Four is enough for speculation to be
 // visible — a queue that can only build one path never speculates — while
@@ -110,6 +126,7 @@ type namedQueueProfileConfig struct {
 	Analyzer       *analyzerConfig       `yaml:"analyzer"`
 	Scorer         *scorerConfig         `yaml:"scorer"`
 	Speculator     *speculatorConfig     `yaml:"speculator"`
+	Predictor      *predictorConfig      `yaml:"predictor"`
 }
 
 // queueProfileConfig is the full set of extensions a queue resolves to.
@@ -119,6 +136,7 @@ type queueProfileConfig struct {
 	Analyzer       analyzerConfig       `yaml:"analyzer"`
 	Scorer         scorerConfig         `yaml:"scorer"`
 	Speculator     speculatorConfig     `yaml:"speculator"`
+	Predictor      predictorConfig      `yaml:"predictor"`
 }
 
 // changeProviderConfig selects how change metadata is fetched. The github and
@@ -239,6 +257,19 @@ type speculatorConfig struct {
 	BuildBudget int `yaml:"buildBudget"`
 }
 
+// predictorConfig tunes how a queue turns its scorer's price into the
+// probability the generator ranks on. The scorer being revised is the queue's
+// own, so it is not named again here.
+type predictorConfig struct {
+	Type string `yaml:"type"`
+	// Factors multiply the odds of the scorer's price, one per piece of
+	// evidence, keyed by evidence name. An omitted factor is neutral, so an
+	// omitted block ranks on the scorer's price alone. Values are hand-set
+	// placeholders, not measured: they are uncalibrated until the fitting work
+	// in doc/rfc/submitqueue/outcome-predictor.md lands.
+	Factors map[string]float64 `yaml:"factors"`
+}
+
 // loadProfilesConfig reads and validates the profiles configuration at path.
 func loadProfilesConfig(path string) (profilesConfig, error) {
 	data, err := os.ReadFile(path)
@@ -300,6 +331,11 @@ func (c *profilesConfig) normalizeAndValidate() error {
 				return err
 			}
 		}
+		if q.Predictor != nil {
+			if err := q.Predictor.normalizeAndValidate(where); err != nil {
+				return err
+			}
+		}
 	}
 	return c.validateGitRepoPaths()
 }
@@ -358,6 +394,9 @@ func (c profilesConfig) resolve(q namedQueueProfileConfig) queueProfileConfig {
 	if q.Speculator != nil {
 		profile.Speculator = *q.Speculator
 	}
+	if q.Predictor != nil {
+		profile.Predictor = *q.Predictor
+	}
 	return profile
 }
 
@@ -374,7 +413,10 @@ func (p *queueProfileConfig) normalizeAndValidate(where string) error {
 	if err := p.Scorer.normalizeAndValidate(where); err != nil {
 		return err
 	}
-	return p.Speculator.normalizeAndValidate(where)
+	if err := p.Speculator.normalizeAndValidate(where); err != nil {
+		return err
+	}
+	return p.Predictor.normalizeAndValidate(where)
 }
 
 func (c *changeProviderConfig) normalizeAndValidate(where string) error {
@@ -556,6 +598,31 @@ func (s *scorerConfig) normalizeAndValidate(where string) error {
 	return nil
 }
 
+// normalizeAndValidate applies defaults and rejects a predictor that could not
+// be built. An empty block is an evidence predictor with every factor neutral,
+// which prices a batch at exactly its scorer's price.
+func (p *predictorConfig) normalizeAndValidate(where string) error {
+	if p.Type == "" {
+		p.Type = predictorTypeEvidence
+	}
+	if p.Type != predictorTypeEvidence {
+		return fmt.Errorf("%s: unknown predictor type %q", where, p.Type)
+	}
+	for name, factor := range p.Factors {
+		switch name {
+		case factorPathPassed, factorPathFailed, factorMerging, factorCancelling:
+		default:
+			return fmt.Errorf("%s: unknown predictor factor %q", where, name)
+		}
+		// Zero would pin every batch carrying the evidence to a probability of
+		// zero, and a negative multiplier on odds means nothing at all.
+		if factor <= 0 {
+			return fmt.Errorf("%s: predictor factor %q is %v, must be positive", where, name, factor)
+		}
+	}
+	return nil
+}
+
 func (s *speculatorConfig) normalizeAndValidate(where string) error {
 	// A negative budget is rejected rather than clamped: sticky would compute no
 	// free slots from it, so the queue would batch and then never build anything,

service/submitqueue/orchestrator/server/config_test.go

@@ -666,3 +666,49 @@ func TestLoadProfilesConfig_RejectsBadScorers(t *testing.T) {
 		})
 	}
 }
+
+func TestLoadProfilesConfig_RejectsBadPredictors(t *testing.T) {
+	tests := []struct {
+		name     string
+		contents string
+	}{
+		{name: "unknown predictor type", contents: "defaults:\n  predictor: {type: vibes}\n"},
+		{name: "unknown factor", contents: "defaults:\n  predictor:\n    factors: {pathPased: 2}\n"},
+		{name: "zero factor", contents: "defaults:\n  predictor:\n    factors: {merging: 0}\n"},
+		{name: "negative factor", contents: "defaults:\n  predictor:\n    factors: {pathFailed: -1}\n"},
+		{name: "bad factor on a queue override", contents: "defaults: {}\nqueues:\n  - name: q\n    predictor:\n      factors: {merging: 0}\n"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := loadProfilesConfig(writeProfiles(t, tt.contents))
+			require.Error(t, err)
+		})
+	}
+}
+
+// An omitted predictor block leaves the queue ranking on its scorer's price
+// alone, which is what every queue does until someone states a factor.
+func TestLoadProfilesConfig_DefaultsThePredictorToNeutral(t *testing.T) {
+	cfg, err := loadProfilesConfig(writeProfiles(t, "defaults: {}\nqueues:\n  - name: q\n"))
+	require.NoError(t, err)
+
+	assert.Equal(t, predictorTypeEvidence, cfg.Defaults.Predictor.Type)
+
+	factors := factorsFrom(cfg.resolve(cfg.Queues[0]).Predictor)
+	assert.Equal(t, neutralFactor, factors.PathPassed)
+	assert.Equal(t, neutralFactor, factors.PathFailed)
+	assert.Equal(t, neutralFactor, factors.Merging)
+	assert.Equal(t, neutralFactor, factors.Cancelling)
+}
+
+func TestLoadProfilesConfig_ReadsPredictorFactors(t *testing.T) {
+	cfg, err := loadProfilesConfig(writeProfiles(t,
+		"defaults:\n  predictor:\n    factors: {pathPassed: 10, pathFailed: 0.3, merging: 12, cancelling: 0.1}\n"))
+	require.NoError(t, err)
+
+	factors := factorsFrom(cfg.Defaults.Predictor)
+	assert.Equal(t, 10.0, factors.PathPassed)
+	assert.Equal(t, 0.3, factors.PathFailed)
+	assert.Equal(t, 12.0, factors.Merging)
+	assert.Equal(t, 0.1, factors.Cancelling)
+}

service/submitqueue/orchestrator/server/profiles.go

@@ -47,6 +47,8 @@ import (
 	"github.com/uber/submitqueue/submitqueue/extension/conflict/pathoverlap"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/predictor"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/predictor/evidence"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/scorer"
 	"github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/composite"
 	scorerfake "github.com/uber/submitqueue/submitqueue/extension/speculation/scorer/fake"
@@ -80,6 +82,10 @@ type Profile struct {
 	// likely their assumptions are to hold.
 	Scorer scorer.Factory
 
+	// Predictor turns this queue's scorer price into the probability the
+	// generator ranks on, revising it with the batch's observed progress.
+	Predictor predictor.Factory
+
 	// Speculator decides which of this queue's speculation paths to build and
 	// which running ones to preempt, within the build budget.
 	Speculator speculator.Factory
@@ -142,6 +148,14 @@ func (p Profiles) ScorerFactory() scorer.Factory {
 	})
 }
 
+// PredictorFactory returns a predictor.Factory that resolves the
+// Predictor for each queue from the profile registry.
+func (p Profiles) PredictorFactory() predictor.Factory {
+	return predictorFunc(func(c predictor.Config) (predictor.Predictor, error) {
+		return p.For(c.QueueName).Predictor.For(c)
+	})
+}
+
 // StorageFactory returns a storage.Factory that routes each queue to its
 // profile's storage backend before binding the queue-scoped store aggregate.
 func (p Profiles) StorageFactory() storage.Factory {
@@ -176,6 +190,10 @@ type scorerFunc func(scorer.Config) (scorer.Scorer, error)
 
 func (f scorerFunc) For(c scorer.Config) (scorer.Scorer, error) { return f(c) }
 
+type predictorFunc func(predictor.Config) (predictor.Predictor, error)
+
+func (f predictorFunc) For(c predictor.Config) (predictor.Predictor, error) { return f(c) }
+
 type speculatorFunc func(speculator.Config) (speculator.Speculator, error)
 
 func (f speculatorFunc) For(c speculator.Config) (speculator.Speculator, error) { return f(c) }
@@ -265,33 +283,71 @@ func (b *profileBuilder) build(cfg queueProfileConfig, where string) (Profile, e
 	if err != nil {
 		return Profile{}, err
 	}
-	// The speculator is composed last, because it is built from whatever scorer
-	// the profile ended up with.
-	return withSpeculator(Profile{
+	// The predictor and the speculator are composed last, because each is built
+	// from what the profile ended up with one level below it.
+	return withSpeculator(withPredictor(Profile{
 		ChangeProvider: provider,
 		BuildRunner:    runner,
 		Analyzer:       analyzer,
 		Storage:        b.stores,
 		Scorer:         sc,
-	}, cfg.Speculator.BuildBudget), nil
+	}, cfg.Predictor, b.scope), cfg.Speculator.BuildBudget), nil
+}
+
+// withPredictor returns the profile with its predictor composed over its own
+// scorer: the scorer prices the batch's change, and the predictor revises that
+// price with what the batch's builds have done.
+//
+// The scorer is resolved lazily, at the queue the predictor itself was asked
+// for, so the queue's identity reaches one level down into the scorer too.
+func withPredictor(p Profile, cfg predictorConfig, scope tally.Scope) Profile {
+	p.Predictor = predictorFunc(func(c predictor.Config) (predictor.Predictor, error) {
+		sc, err := p.Scorer.For(scorer.Config{QueueName: c.QueueName})
+		if err != nil {
+			return nil, fmt.Errorf("failed to resolve scorer for queue %q: %w", c.QueueName, err)
+		}
+		return evidence.New(c, sc, factorsFrom(cfg), scope.SubScope("predictor"))
+	})
+	return p
+}
+
+// factorsFrom reads the configured factors onto the named fields the predictor
+// takes, leaving an unstated one neutral. Names are validated when the config
+// is loaded.
+func factorsFrom(cfg predictorConfig) evidence.Factors {
+	factors := evidence.AllOnes()
+	for name, factor := range cfg.Factors {
+		switch name {
+		case factorPathPassed:
+			factors.PathPassed = factor
+		case factorPathFailed:
+			factors.PathFailed = factor
+		case factorMerging:
+			factors.Merging = factor
+		case factorCancelling:
+			factors.Cancelling = factor
+		}
+	}
+	return factors
 }
 
 // withSpeculator returns the profile with its speculator composed from its own
-// scorer: bestfirst ranks a queue's candidate paths by how likely all their
+// predictor: bestfirst ranks a queue's candidate paths by how likely all their
 // assumptions are to hold, and sticky spends buildBudget down that ranking
 // without preempting builds already running. Swapping either part changes the
 // policy without touching the speculate controller, which depends only on the
 // Speculator contract.
 //
-// The scorer is resolved lazily, at the queue the speculator itself was asked
-// for, so the queue's identity reaches one level down into the scorer too.
+// The predictor is resolved lazily, at the queue the speculator itself was
+// asked for, so the queue's identity reaches down through the predictor to the
+// scorer under it.
 func withSpeculator(p Profile, buildBudget int) Profile {
 	p.Speculator = speculatorFunc(func(c speculator.Config) (speculator.Speculator, error) {
-		sc, err := p.Scorer.For(scorer.Config{QueueName: c.QueueName})
+		pred, err := p.Predictor.For(predictor.Config{QueueName: c.QueueName})
 		if err != nil {
-			return nil, fmt.Errorf("failed to resolve scorer for queue %q: %w", c.QueueName, err)
+			return nil, fmt.Errorf("failed to resolve predictor for queue %q: %w", c.QueueName, err)
 		}
-		return specstandard.New(c, bestfirst.New(sc), sticky.New(buildBudget)), nil
+		return specstandard.New(c, bestfirst.New(pred), sticky.New(buildBudget)), nil
 	})
 	return p
 }