feat(speculation): allocator contract and sticky impl

## Summary

### Why?

Ranked candidates need a budget policy: something must decide which paths get the queue's limited concurrent-build slots, without double-funding paths that are already running or finished.

### What?

Adds `submitqueue/extension/speculation/allocator` — the budget-spending contract the standard Speculator hands its candidate stream to — plus the `sticky` implementation and mocks.

The Allocator owns terminal-path suppression: it reconciles the stream against the stored path sets by path ID, so an in-flight candidate keeps the slot it already holds and a terminal one is skipped. Budget accounting reads path status and nothing else — pending, building, and cancelling paths hold a slot until their builds actually stop — and the unit is deliberately coarse: one slot per concurrent build.

`sticky` fills only free slots and never preempts a running build; its policy counterpart is a preempting allocator. Output is all-or-nothing (a partial list would fund an arbitrary prefix of the ranking), and a stored path with no status fails fast rather than guessing whether it holds a slot.

## Test Plan

✅ `bazel test //submitqueue/extension/speculation/allocator/...` — terminal candidates skipped per status, in-flight candidates kept, budget exhaustion, context cancellation, corrupt-status failure.

✅ `make fmt`, `make gazelle`, `make mocks`

Author: behinddwalls

submitqueue/extension/speculation/allocator/BUILD.bazel

@@ -0,0 +1,12 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["allocator.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/generator:go_default_library",
+    ],
+)

submitqueue/extension/speculation/allocator/README.md

@@ -0,0 +1,13 @@
+# allocator
+
+The `allocator` package defines a composition point used by the `standard` `Speculator`: an `Allocator` spends the queue's build budget over a candidate iterator. Like `generator`, it is **not** a controller-facing extension — an alternate `Speculator` need not use or expose this split — so there is no `Config` or `Factory` here; an `Allocator` is chosen when the `standard` `Speculator` is constructed.
+
+`Allocate` pulls candidates in the order the iterator yields them and matches them against the queue's current path sets by path ID. A pending or building path keeps the slot it already holds rather than starting a second attempt. A path whose build already finished comes back round as a candidate — the generator enumerates the whole space — and is skipped. Neither draws on free budget.
+
+Pending, building, and cancelling paths all charge the budget; a cancelling build holds its slot until it reaches a terminal state. Terminal paths charge nothing. `Allocate` returns the build and cancel actions that spend what is left.
+
+A cancelled or expired context aborts the run with its error and no actions. The output is all-or-nothing on purpose: a partial list would fund an arbitrary prefix of the ranking, leaving the budget half-spent on whatever was pulled first.
+
+Because cancellation is best-effort, an `Allocator` should not spend capacity it only expects a cancel to release. A build cancelled to make room keeps charging the budget until that cancel reaches a terminal state, so the queue converges over successive runs instead of oversubscribing the hard cap in one pass.
+
+How the budget is measured is the implementation's choice. The simplest unit is a build: every path costs one slot, whatever its build does. An `Allocator` that understands build size — target count, historical cost — could weight paths instead and pack the budget more tightly.

submitqueue/extension/speculation/allocator/allocator.go

@@ -0,0 +1,44 @@
+// 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 allocator defines the budget-spending composition point used by the
+// standard Speculator. An Allocator spends the queue's build budget over a
+// candidate iterator. It is not a controller-facing extension — an alternate
+// Speculator need not use or expose this split.
+package allocator
+
+//go:generate mockgen -source=allocator.go -destination=mock/allocator_mock.go -package=mock
+
+import (
+	"context"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator"
+)
+
+// Allocator decides how to spend the queue's build budget over the generator's
+// candidate stream, returning the build and cancel actions to take this run. How
+// it rations the budget across new candidates and paths already in flight — and
+// whether it preempts running builds — is the implementation's policy.
+type Allocator interface {
+	// Allocate returns the build and cancel actions to take. It reconciles the
+	// candidate stream against the queue's current path sets by path ID: a
+	// candidate matching an in-flight path stays funded rather than starting a
+	// new attempt, and one matching a path whose build already finished is
+	// skipped — the generator enumerates the whole space, so suppressing
+	// finished paths is the Allocator's job.
+	//
+	// A cancelled or expired ctx aborts with its error and no actions.
+	Allocate(ctx context.Context, pathSets []entity.SpeculationPathSet, iter generator.Iterator) ([]entity.Speculation, error)
+}

submitqueue/extension/speculation/allocator/mock/BUILD.bazel

@@ -0,0 +1,13 @@
+load("@rules_go//go:def.bzl", "go_library")
+
+go_library(
+    name = "go_default_library",
+    srcs = ["allocator_mock.go"],
+    importpath = "github.com/uber/submitqueue/submitqueue/extension/speculation/allocator/mock",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//submitqueue/entity:go_default_library",
+        "//submitqueue/extension/speculation/generator:go_default_library",
+        "@org_uber_go_mock//gomock:go_default_library",
+    ],
+)

submitqueue/extension/speculation/allocator/mock/allocator_mock.go

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

submitqueue/extension/speculation/allocator/sticky/BUILD.bazel

@@ -0,0 +1,11 @@
+# sticky
+
+The `sticky` `Allocator` fills only free budget slots and leaves every in-flight build running. Against the budget it counts the paths that still hold a slot: `pending`, `building`, and `cancelling` — a cancel is only a request, and the build keeps its slot until it actually stops. Only path status enters this — no batch state does, since the budget tracks builds occupying CI.
+
+It then proposes a build for each new candidate, in order, until the budget fills. Two kinds of candidate are skipped, for different reasons. A candidate matching an already-funded path is holding a slot already — it is counted above — so it needs no additional slot and no second build action. A candidate whose path is terminal in the path sets holds no slot and needs none. Neither consumes free budget. It never proposes a cancel.
+
+The budget is measured in builds: one slot per path, no matter what that path's build does. That is deliberately coarse — a twelve-hour build and a two-minute build cost the same slot. An allocator that understands build size (target count, historical duration) could pack the same budget more effectively; `sticky` trades that for simplicity.
+
+A stored path with no status at all is a data defect, and `sticky` fails fast: `Allocate` returns an error rather than guessing what the record means.
+
+The trade-off: a better candidate arriving while the budget is full waits for a running build to finish rather than displacing it. `sticky` never discards work already started, and it cannot react to a newly attractive path mid-flight. Its counterpart is a preempting `Allocator` that cancels a low-value in-flight path to fund a better one, spending a cancel now to converge faster on the next run.

submitqueue/extension/speculation/allocator/sticky/README.md

@@ -0,0 +1,133 @@
+// 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 sticky provides an allocator.Allocator that fills only free budget
+// slots and leaves every in-flight build running — it never preempts. Its policy
+// counterpart is a preempting allocator, which would cancel a low-value in-flight
+// path to fund a better candidate; sticky trades that responsiveness for never
+// discarding a build it has already started.
+package sticky
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/uber/submitqueue/submitqueue/entity"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/allocator"
+	"github.com/uber/submitqueue/submitqueue/extension/speculation/generator"
+)
+
+// alloc is a sticky allocator.Allocator.
+type alloc struct {
+	// budget is the queue's cap on concurrently held build slots, measured in
+	// builds: one slot per path whose attempt is pending, building, or
+	// cancelling, whatever that build's size. The unit is deliberately coarse —
+	// an allocator that understands build size could weight paths instead.
+	budget int
+}
+
+// New returns a sticky allocator.Allocator with the given budget, measured in
+// concurrent builds.
+func New(budget int) allocator.Allocator {
+	return alloc{budget: budget}
+}
+
+// Allocate keeps every in-flight path funded and pulls candidates in order into
+// the remaining free budget, proposing a build for each new path until the
+// budget fills. It never proposes a cancel.
+//
+// A cancelled or expired ctx aborts with its error and no actions. The run's
+// output is all-or-nothing: a partial action list would fund an arbitrary
+// prefix of the ranking, so a caller that gave up mid-run gets nothing rather
+// than a half-spent budget.
+func (a alloc) Allocate(ctx context.Context, pathSets []entity.SpeculationPathSet, iter generator.Iterator) ([]entity.Speculation, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+	// Paths already holding a build slot. This set does double duty: it is both
+	// the count of budget already spent and the guard against proposing a second
+	// build for something already running, so a status belongs here if either
+	// reason applies.
+	funded := make(map[string]bool)
+	// Paths whose build already finished. They hold no slot, but they must not
+	// be built again either.
+	terminal := make(map[string]bool)
+	for _, set := range pathSets {
+		for _, entry := range set.Paths {
+			if entry.Status == entity.SpeculationPathStatusUnknown {
+				// A stored path always carries a real status, so this record is
+				// corrupt. Fail fast rather than guess: any reading of it — slot
+				// held or free, buildable or not — could be wrong in a way that
+				// either pins a slot forever or double-builds a path.
+				return nil, fmt.Errorf("speculation path %q of head %q has no status", entry.ID, set.Head)
+			}
+			if entry.Status.IsTerminal() {
+				terminal[entry.ID] = true
+				continue
+			}
+			// Anything else that has not finished is still holding a slot:
+			// pending and building obviously, and cancelling too, since the build
+			// keeps its slot until it actually stops.
+			//
+			// Only the path's status matters. No batch state enters this
+			// decision — "merging" and the rest are states of a batch, never of
+			// a path — so the rule is simply that CI is still busy with it.
+			funded[entry.ID] = true
+		}
+	}
+
+	free := a.budget - len(funded)
+
+	var out []entity.Speculation
+	proposed := make(map[string]bool)
+	for free > 0 {
+		// Next reports cancellation too, but check here as well: a generator
+		// that ignores ctx must not be able to spin this loop forever.
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		c, ok, err := iter.Next(ctx)
+		if err != nil {
+			return nil, err
+		}
+		if !ok {
+			// ok=false means the generator has nothing left to offer: every path
+			// the queue could speculate on has been seen. Stopping here with
+			// budget still free is normal, not a failure — there is simply
+			// nothing else worth building right now.
+			break
+		}
+		id := c.Path.ID()
+		if terminal[id] {
+			// This path's build already ran. The generator enumerates the whole
+			// space with no knowledge of the path sets, so finished paths come
+			// back round as candidates; they need no slot and must not be rebuilt.
+			continue
+		}
+		if funded[id] || proposed[id] {
+			// funded: this path is already building. It needs no new action and
+			// is already charging the budget, so skip it without spending
+			// anything.
+			//
+			// proposed: a generator is not supposed to repeat a candidate, so
+			// this should never fire. It is a cheap guard against one that does,
+			// since a duplicate would otherwise be paid for twice.
+			continue
+		}
+		out = append(out, entity.Speculation{Path: c.Path, Action: entity.PathActionBuild})
+		proposed[id] = true
+		free--
+	}
+	return out, nil
+}