feat(speculation): standard composed speculator

submitqueue/extension/speculation/speculator/README.md

@@ -8,6 +8,8 @@ The `speculator` package defines the one speculation extension the speculate con
 
 Like the other extensions, a `Speculator` is selected **per queue** by the wiring layer through the `Config` (queue name) and `Factory` interface. Budget, clock, and any extra data are injected at construction by the integrator, not carried on the contract.
 
+[`standard`](standard/README.md) is the default implementation: it funds the most promising paths first until the build budget is spent.
+
 ## Adding a backend
 
 Create a package under `speculator/<backend>/` whose `New(...)` returns a `speculator.Speculator`, injecting whatever it needs at construction. Resolve any content it requires internally; do not add a `Config` or `Factory` implementation here — per-queue routing and the factory adapter live in the wiring layer.

submitqueue/extension/speculation/speculator/standard/BUILD.bazel

@@ -0,0 +1,30 @@
+load("@rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["standard.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/speculator/standard",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/allocator:go_default_library",
+        "//submitqueue/extension/speculation/generator:go_default_library",
+        "//submitqueue/extension/speculation/speculator:go_default_library",
+    ],
+)
+
+go_test(
+    name = "go_default_test",
+    srcs = ["standard_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/allocator/mock:go_default_library",
+        "//submitqueue/extension/speculation/allocator/sticky:go_default_library",
+        "//submitqueue/extension/speculation/generator/bestfirst:go_default_library",
+        "//submitqueue/extension/speculation/generator/mock:go_default_library",
+        "@com_github_stretchr_testify//assert:go_default_library",
+        "@com_github_stretchr_testify//require:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

submitqueue/extension/speculation/speculator/standard/README.md

@@ -0,0 +1,13 @@
+# standard
+
+The `standard` `Speculator` funds the queue's most promising speculation paths first, until the build budget is spent.
+
+Each run it considers candidate paths in descending order of their probability of being the future that actually happens, and proposes builds down that ranking. Paths already pending or building keep the slot they hold rather than restarting; paths whose builds already finished are skipped for as long as their records remain in the supplied path sets, so a finished path can be proposed again — for a retry, say — once retention drops it; new builds fill whatever budget remains.
+
+When the budget runs out, everything below the cut waits for a later run. That is safe because a batch's verdict never depends on what was funded — only on how its dependencies resolve and which builds pass.
+
+Both halves are swappable. The ranking is the `Generator`'s: the default `bestfirst` scores each path by the probability that all its assumptions hold. The budget policy is the `Allocator`'s: the default `sticky` fills only free slots and never preempts, where a preempting allocator would cancel a low-value in-flight path to fund a better one.
+
+`standard` itself decides nothing — it connects the `Generator`'s stream to the `Allocator` — so changing prioritization or budget behavior means swapping a part, not writing a new `Speculator`.
+
+The `Generator`/`Allocator` split is internal to `standard`, not part of the `Speculator` contract: an alternate `Speculator` could compute build and cancel decisions directly.

submitqueue/extension/speculation/speculator/standard/standard.go

@@ -0,0 +1,53 @@
+// 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 standard provides the standard speculator.Speculator: it composes a
+// Generator and an Allocator so a queue's candidate scoring and its budget
+// rationing policy can vary independently without changing the Speculator. The
+// Speculator itself decides nothing — it opens the Generator's candidate stream
+// and hands it to the Allocator; all of its behavior comes from those two parts.
+package standard
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/speculator"
+)
+
+// spec is a speculator.Speculator that opens the generator's candidate stream
+// and hands it to the allocator to spend the build budget.
+type spec struct {
+	gen   generator.Generator
+	alloc allocator.Allocator
+}
+
+// New returns the standard speculator.Speculator, composing a Generator and an
+// Allocator.
+func New(gen generator.Generator, alloc allocator.Allocator) speculator.Speculator {
+	return spec{gen: gen, alloc: alloc}
+}
+
+// Speculate opens the generator over the queue's batches, then lets the
+// allocator spend the budget over the resulting candidate iterator, reconciling
+// it against the path sets.
+func (s spec) Speculate(ctx context.Context, batches []entity.Batch, pathSets []entity.SpeculationPathSet) ([]entity.Speculation, error) {
+	iter, err := s.gen.Open(ctx, batches)
+	if err != nil {
+		return nil, err
+	}
+	return s.alloc.Allocate(ctx, pathSets, iter)
+}

submitqueue/extension/speculation/speculator/standard/standard_test.go

@@ -0,0 +1,125 @@
+// 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 standard
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	allocatormock "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/mock"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/sticky"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator/bestfirst"
+	generatormock "github.com/uber/submitqueue/submitqueue/extension/speculation/generator/mock"
+)
+
+// assumptionFor returns what a path assumes about a given dependency batch.
+func assumptionFor(p entity.SpeculationPath, dep string) entity.DependencyAssumption {
+	for _, d := range p.Dependencies {
+		if d.Batch == dep {
+			return d.Assumption
+		}
+	}
+	return entity.DependencyAssumptionUnknown
+}
+
+// constScorer is a minimal scorer.Scorer that scores every batch identically.
+type constScorer struct{ v float64 }
+
+func (c constScorer) Score(context.Context, entity.Batch) (float64, error) { return c.v, nil }
+
+func TestComposed_EndToEnd_NaivePair(t *testing.T) {
+	batches := []entity.Batch{
+		{ID: "q/1", State: entity.BatchStateSpeculating},
+		{ID: "q/2", State: entity.BatchStateSpeculating, Dependencies: []string{"q/1"}},
+	}
+
+	// bestfirst generator + sticky allocator with a 2-build budget.
+	spec := New(bestfirst.New(constScorer{0.9}), sticky.New(2))
+	got, err := spec.Speculate(context.Background(), batches, nil)
+	require.NoError(t, err)
+
+	// Two free budget slots -> two builds. Everything proposed is a build.
+	require.Len(t, got, 2)
+	var q2 entity.Speculation
+	var found bool
+	for _, s := range got {
+		assert.Equal(t, entity.PathActionBuild, s.Action)
+		if s.Path.Head == "q/2" {
+			q2, found = s, true
+		}
+	}
+	// q/2's funded path is the optimistic one: it assumes q/1 succeeds.
+	require.True(t, found, "q/2 should have been funded")
+	assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(q2.Path, "q/1"))
+}
+
+func TestComposed_WiresGeneratorIntoAllocator(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	batches := []entity.Batch{{ID: "q/1", State: entity.BatchStateSpeculating}}
+	pathSets := []entity.SpeculationPathSet{{Head: "q/1"}}
+	want := []entity.Speculation{{
+		Path:   entity.SpeculationPath{Head: "q/1"},
+		Action: entity.PathActionBuild,
+	}}
+
+	iter := generatormock.NewMockPathIterator(ctrl)
+	gen := generatormock.NewMockGenerator(ctrl)
+	alloc := allocatormock.NewMockAllocator(ctrl)
+
+	gen.EXPECT().Open(gomock.Any(), batches).Return(iter, nil)
+	alloc.EXPECT().Allocate(gomock.Any(), pathSets, iter).Return(want, nil)
+
+	got, err := New(gen, alloc).Speculate(context.Background(), batches, pathSets)
+	require.NoError(t, err)
+	assert.Equal(t, want, got)
+}
+
+func TestComposed_PropagatesGeneratorError(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	errOpen := errors.New("open boom")
+	gen := generatormock.NewMockGenerator(ctrl)
+	alloc := allocatormock.NewMockAllocator(ctrl)
+
+	gen.EXPECT().Open(gomock.Any(), gomock.Any()).Return(nil, errOpen)
+	// Allocate must not be called when Open fails (no alloc.EXPECT()).
+
+	_, err := New(gen, alloc).Speculate(context.Background(), nil, nil)
+	require.ErrorIs(t, err, errOpen)
+}
+
+func TestComposed_PropagatesContextCancellation(t *testing.T) {
+	// standard adds no cancellation handling of its own — it inherits it from
+	// the two parts. This pins that the seam actually propagates: a cancelled
+	// run yields the context error and no actions.
+	batches := []entity.Batch{
+		{ID: "q/1", State: entity.BatchStateSpeculating},
+		{ID: "q/2", State: entity.BatchStateSpeculating, Dependencies: []string{"q/1"}},
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	spec := New(bestfirst.New(constScorer{0.9}), sticky.New(2))
+	got, err := spec.Speculate(ctx, batches, nil)
+	require.ErrorIs(t, err, context.Canceled)
+	assert.Nil(t, got)
+}