fix(speculation): stop one dependency failing the whole snapshot

## Summary

### Why?

Speculation died for an entire queue whenever any batch depended on one that had reached `merging`:

```
speculator failed for queue demo-queue: score dependency "demo-queue/batch/1":
failed to resolve storage for queue "": queue name must not be empty
```

The root cause is a caller contract violation, not a generator bug. `speculator.Speculate` documents `batches` as "every in-flight batch of the queue, plus any finalized batch still referenced as a dependency by an in-flight one". `ask` was passing `snap.speculating` — the speculating heads alone. `read` already assembles the right set in `snap.batches`; it simply was not the slice handed over. A merging dependency was therefore absent from the generator's index, `batchByID[id]` returned a zero `entity.Batch`, and that zero batch reached the scorer with an empty `Queue`.

The blast radius was the whole queue rather than one batch, because `Generate` scores every unresolved dependency up front to seed its heap and returned on the first error. `standard.Speculate` then short-circuited before the allocator pulled a single candidate: no path failed, none was ever produced, and the fall-back-to-the-next-path machinery sits downstream of a stage that had already died. A measured run of 20 requests left 9 in `error` and 11 wedged in `speculating` with nothing recorded against them.


### What?

**Hand the Speculator the whole queue.** `ask` now flattens `snap.batches` — every batch the run read — and passes that. No ordering is imposed: the generator's heap comparator is a strict total order, so its candidate sequence is identical whatever order the input arrives in (checked over 200 shuffles), and a Speculator that read meaning into input order would be relying on something the contract never offered. `ask`'s doc comment argued for the narrow slice and is rewritten.

Widening is safe because of the commit below this one. A head this run has just decided still reads as `Speculating` in the snapshot — `finalize` records only terminal outcomes — and both the Speculator and `check` read head eligibility off that same map, so a stale entry would fool both filters at once. What stops it mattering is that a head now merges only once *every* dependency has settled: the generator pins them all, can therefore construct nothing but the path that already passed, and the allocator skips that as finished. Verified by driving the real `bestfirst`/`sticky` pair with a merged head and settled dependencies — zero actions proposed for it.

That ordering is load-bearing, which is why the merge gate is the parent rather than a follow-up. With the old gate a head could merge past a dependency that was still live; the generator would then see that dependency as an open question, offer a path ID the set had never held, and the allocator would fund a fresh build for a batch already handed to Runway.

**Make one unpriceable dependency cost only its own estimate.** `score` now substitutes `defaultProbability` when the scorer returns an error, and never calls the scorer at all for a dependency the snapshot did not carry — that batch is zero in every field, so scoring it would price some other batch entirely or fail on its empty queue name. Context cancellation is still fatal, including when it surfaces *as* the scorer's error: the loop checks `ctx` before each call, so a context that dies during the last one would otherwise be absorbed as an unpriceable dependency and hand back an iterator to a caller that has already gone. Scorer failures stay observable through the scorer's own metrics span, which already reports them via `op.Complete(retErr)`.

That is the whole containment fix. An earlier revision of this branch also made scoring lazy — heads seeded at an optimistic bound and priced on first pull — and it has been dropped. The only admissible bound for an unpriced head is `log 1`, identical for every head, so the first pull priced the entire queue anyway; the laziness bought one narrow case (a run that pulls nothing because the budget is saturated) in exchange for a placeholder, an admissibility argument, priced and unpriced items sharing a heap, and five reworked tests. Defaulting on failure fixes the bug on its own.

**`Merging` is left as an open question in the generator.** Tempting to pin it to *succeeds* — the batch looks committed to landing — but a merge can fail, so nothing is settled, and it would put a state-specific policy inside the search when whether a path betting against a merging batch is worth funding is a question of price that belongs to the scorer. The allocator already draws exactly this line — "no batch state enters this decision — `merging` and the rest are states of a batch, never of a path" — and the generator holds it too.


## Test Plan

- ✅ `make test` — 96/96 pass
- ✅ `make lint`, `make check-gazelle`, `make check-tidy`

New and reworked coverage, per defect:

- `TestRun_PassesSnapshotToSpeculator` — the Speculator receives every batch the run read, in a stable order
- `TestBestFirst_AbsorbsScorerError` — a scorer error costs that dependency its estimate and nothing else
- `TestBestFirst_NeverScoresAnAbsentDependency` — an absent dependency never reaches the scorer, even when the caller hands over a malformed snapshot
- `TestBestFirst_MergingDependencyStaysOpen` — a merging dependency is priced like any other and keeps both sides
- `TestBestFirst_HonorsCancelledContext/a scorer that fails on a dead context ends the run`

The last was added for a defect a review of this branch turned up, and was confirmed to fail against the code as it stood before the fix.

Not verified end to end: `make demo-pr` lives on the `sq/demo-pr` branch, so reproducing the original 20-request run needs that target ported across worktrees plus a Docker stack.

## Issue

Fixes https://linear.app/uber/issue/CODEM-424

That issue proposed lazy scoring as its primary fix; this lands the containment it was after without the algorithm change, for the reasons above.

Follow-up filed as https://linear.app/uber/issue/CODEM-428 — this stops queues wedging this way, but a queue already wedged still has no event that will wake it, because speculation is edge-triggered only.

The merge-gate defect found while investigating this one — a head could merge on a *fails* assumption that had not come true — is the parent commit, since the widening here relies on the invariant it restores.

Author: behinddwalls

doc/rfc/submitqueue/speculation-generator-best-first.md

@@ -49,7 +49,7 @@ The batch being built is written before its assumptions. For example, `C [A succ
 
 `Generate` receives the queue's live batches as a snapshot and takes it as given. A well-formed snapshot carries unique, non-empty batch IDs, includes every batch a head's direct dependencies reference, and gives no head an empty, duplicate, or self dependency. Those are preconditions the caller owns, established where the snapshot is assembled. The generator does not re-check them: it is on the hot path of every run, the checks it could make are the ones an assembled-correctly snapshot can never fail, and paying for them here only spreads the same contract across two places. A malformed snapshot yields undefined candidates rather than an error.
 
-A score that is not a probability is the one bad input the generator absorbs, because it arrives from the injected scorer rather than from the caller and there is no earlier point that could catch it. A score outside `[0, 1]`, or `NaN`, is replaced with a default of 0.95 — optimistic on purpose, so a dependency nobody could estimate keeps its head's preferred path near the front instead of burying it or failing the whole run on one number. Any deliberate defaulting still belongs to the scorer implementation, which knows what information it does and does not have; this is only the floor under it.
+A dependency the generator cannot price is the one bad input it absorbs, because it arrives from the injected scorer rather than from the caller and there is no earlier point that could catch it. Three cases take the same 0.95 default: a score outside `[0, 1]` or `NaN`, a scorer call that returned an error, and a dependency the snapshot never carried. The default is optimistic on purpose, so a dependency nobody could estimate keeps its head's preferred path near the front instead of burying it or failing the whole run on one number — and failing the whole run is the real hazard, because `Generate` seeds the heap for every head at once, so one unpriceable dependency would otherwise cost the queue every candidate it had. A batch absent from the snapshot is never passed to the scorer at all: it would resolve to a zero-valued batch belonging to no queue, so scoring it would price some other batch entirely or fail on the empty queue name. Any deliberate defaulting still belongs to the scorer implementation, which knows what information it does and does not have; this is only the floor under it.
 
 ## Step 1: `Generate` prepares each head
 
@@ -447,6 +447,7 @@ The ordering stays the same. `CandidatePath.RankingScore` contains this logarith
 - `Succeeded` fixes an assumption to succeeds.
 - `Failed` or `Cancelled` fixes an assumption to fails.
 - `Cancelling` remains undecided because cancellation may lose a race with completion.
+- `Merging` also remains undecided, because a merge can fail. It is tempting to treat it as committed to landing and skip the scorer call, but that puts a state-specific policy inside the search: whether a path betting against a merging batch is worth funding is a question of price, and price belongs to the scorer. The allocator draws the same line — "no batch state enters this decision" — and the generator holds it too. Nothing is lost by staying open, because a head can never merge ahead of a dependency it took a position on (see [speculation.md](speculation.md)); the cost of an unlikely path is budget, which is the allocator's to ration.
 - A fixed assumption stays in the returned path but contributes probability 1 and has no flip.
 - A shared dependency is scored once per run.
 

submitqueue/extension/speculation/generator/README.md

@@ -2,7 +2,7 @@
 
 The `generator` package is a piece the `standard` `Speculator` is built from: a `Generator` produces the queue's candidate paths as one ordered stream across all heads. It is **not** controller-facing — the speculate controller only knows the `Speculator` contract, and a different `Speculator` need not split its work this way. So there is no `Config` or `Factory` here; a `Generator` is chosen when the `standard` `Speculator` is constructed.
 
-`Generate` starts the stream over the queue's live batches and returns an `Iterator`. Beyond any ranking work required up front, the generator computes only the candidates the caller pulls. A cancelled or expired context ends the stream with its error. The snapshot must include every batch a head's direct dependencies reference; a snapshot that does not — or that carries empty or duplicate IDs, or a head with an empty, duplicate, or self dependency — is malformed input and errors instead of opening a stream.
+`Generate` starts the stream over the queue's live batches and returns an `Iterator`. Beyond any ranking work required up front, the generator computes only the candidates the caller pulls. A cancelled or expired context ends the stream with its error. The snapshot must include every batch a head's direct dependencies reference, carry unique non-empty IDs, and give no head an empty, duplicate, or self dependency. Those are the caller's preconditions: a generator may assume them and is not required to detect a breach, so a malformed snapshot yields undefined candidates rather than an error.
 
 Candidates never repeat and never contradict a known fact. Beyond that, the order is the `Generator`'s own: it yields candidates in whatever ranking it implements, and each carries the score it ranked by — higher first, on a scale the generator defines. Consumers take the iterator in the order given and do not interpret the score. Scores mean something only within the run and are never stored.
 

submitqueue/extension/speculation/generator/bestfirst/README.md

@@ -10,6 +10,8 @@ The [best-first speculation path generation RFC](../../../../../doc/rfc/submitqu
 - `Next` removes the highest-ranked candidate, advances only that head's stream, constructs that candidate's complete path, and returns it. Pulling long enough returns every path exactly once in non-increasing score order.
 - Ranking scores are sums of log probabilities, avoiding underflow while preserving probability order. They are meaningful only within the run that produced them.
 - Exact ties prefer fewer flips; head ID then decides between heads (the cross-head heap holds one candidate per head), and taken flip indexes decide within a head.
-- The snapshot must contain every batch a head's direct dependencies reference; a snapshot missing one — or carrying empty or duplicate batch IDs, or a head with an empty, duplicate, or self dependency — is malformed input and errors instead of opening a stream. Any defaulting for a batch that is hard to score belongs to the scorer, not the generator.
+- A dependency counts as resolved only once it is terminal. Merging and cancelling are both still in progress and either can end the other way, so both stay open questions here. Whether a path betting against a merging dependency is worth funding is a matter of price, and price is the scorer's to say.
+- A dependency that cannot be priced — the scorer call failed, the score was not a probability, or the snapshot never carried the batch — is treated as very likely to succeed rather than ending the run. One unusable number costs its own estimate, never the queue's whole set of candidates. A batch missing from the snapshot is never handed to the scorer at all: it would resolve to a zero batch belonging to no queue.
+- The snapshot must contain every batch a head's direct dependencies reference, carry unique non-empty batch IDs, and give no head an empty, duplicate, or self dependency. That is the caller's precondition, not something checked here: a malformed snapshot yields undefined candidates rather than an error.
 
 The behavior is covered by `bestfirst_test.go`.

submitqueue/extension/speculation/generator/bestfirst/bestfirst.go

@@ -25,7 +25,6 @@ import (
 	"cmp"
 	"container/heap"
 	"context"
-	"fmt"
 	"maps"
 	"math"
 	"slices"
@@ -104,26 +103,47 @@ func speculatingHeads(batches []entity.Batch, batchByID map[string]entity.Batch)
 
 // score asks the scorer for each unresolved dependency exactly once, however
 // many heads wait on it.
+//
+// A dependency that cannot be priced takes defaultProbability rather than
+// ending the run — one unusable number must not cost the queue every candidate
+// it had. Only cancellation is an error.
 func (g *bestFirst) score(ctx context.Context, ids []string, batchByID map[string]entity.Batch) (map[string]float64, error) {
 	probabilityByID := make(map[string]float64, len(ids))
 	for _, id := range ids {
 		if err := ctx.Err(); err != nil {
 			return nil, err
 		}
-		probability, err := g.scorer.Score(ctx, batchByID[id])
+		batch, known := batchByID[id]
+		if !known {
+			// A batch the snapshot never carried is zero in every field, not
+			// just missing — scoring it would price some other batch entirely,
+			// or fail on its empty queue. It is unpriceable, not cheap.
+			probabilityByID[id] = defaultProbability
+			continue
+		}
+		probability, err := g.scorer.Score(ctx, batch)
 		if err != nil {
-			return nil, fmt.Errorf("score dependency %q: %w", id, err)
+			// A scorer that failed because the caller went away has not found
+			// an unpriceable dependency — it has found a dead ctx, which ends
+			// the run. The loop's own check would not catch it on the last
+			// dependency, and a cancelled Generate must never hand back an
+			// iterator.
+			if ctxErr := ctx.Err(); ctxErr != nil {
+				return nil, ctxErr
+			}
+			probability = defaultProbability
 		}
 		probabilityByID[id] = asProbability(probability)
 	}
 	return probabilityByID, nil
 }
 
-// defaultProbability stands in for a score that is not a probability. It is
-// optimistic on purpose: a dependency nobody could estimate is treated as very
-// likely to succeed, which keeps its head's preferred path near the front
-// rather than burying it or dropping the queue's whole snapshot on one bad
-// number.
+// defaultProbability stands in for a score that is not a probability, one the
+// scorer could not produce at all, and one for a dependency the snapshot never
+// carried. It is optimistic on purpose: a dependency nobody could estimate is
+// treated as very likely to succeed, which keeps its head's preferred path near
+// the front rather than burying it or dropping the queue's whole snapshot on
+// one bad number.
 const defaultProbability = 0.95
 
 // asProbability keeps a usable score and substitutes the default for anything

submitqueue/extension/speculation/generator/bestfirst/bestfirst_test.go

@@ -381,15 +381,61 @@ func TestBestFirst_HeadWithNoDependencies(t *testing.T) {
 	assert.InDelta(t, math.Log(1.0), cands[0].RankingScore, 1e-9)
 }
 
-func TestBestFirst_PropagatesScorerError(t *testing.T) {
+// A scorer that cannot price a dependency costs that dependency its estimate,
+// nothing more. The queue keeps every candidate it had, ranked as if the
+// dependency were very likely to succeed.
+func TestBestFirst_AbsorbsScorerError(t *testing.T) {
 	batches := []entity.Batch{
 		{ID: "q/A", State: entity.BatchStateSpeculating},
 		{ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/A"}},
 	}
 
 	iter, err := New(errScorer{}).Generate(context.Background(), batches)
-	assert.Error(t, err)
-	assert.Nil(t, iter)
+	require.NoError(t, err)
+
+	cands := forHead(drainAll(t, iter), "q/H")
+	require.Len(t, cands, 2)
+	assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[0].Path, "q/A"))
+	assert.InDelta(t, math.Log(defaultProbability), cands[0].RankingScore, 1e-9)
+}
+
+// A dependency the snapshot never carried is unpriceable, not cheap: the zero
+// batch it would resolve to belongs to no queue, so it must never reach the
+// scorer.
+func TestBestFirst_NeverScoresAnAbsentDependency(t *testing.T) {
+	batches := []entity.Batch{
+		{ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/missing"}},
+	}
+	sc := newCountingScorer(map[string]float64{})
+
+	iter, err := New(sc).Generate(context.Background(), batches)
+	require.NoError(t, err)
+
+	cands := drainAll(t, iter)
+	require.Len(t, cands, 2, "the absent dependency is still an open question with two sides")
+	assert.Zero(t, sc.total, "the scorer is never handed a batch the snapshot did not carry")
+	assert.InDelta(t, math.Log(defaultProbability), cands[0].RankingScore, 1e-9)
+}
+
+// A merging dependency is still in progress — the merge can fail — so it stays
+// an open question here like any other. Whether a path betting against it is
+// worth funding is a matter of price, which is the scorer's to say, not a
+// state the search hard-codes.
+func TestBestFirst_MergingDependencyStaysOpen(t *testing.T) {
+	batches := []entity.Batch{
+		{ID: "q/landing", State: entity.BatchStateMerging},
+		{ID: "q/H", State: entity.BatchStateSpeculating, Dependencies: []string{"q/landing"}},
+	}
+	sc := newCountingScorer(map[string]float64{"q/landing": 0.9})
+
+	iter, err := New(sc).Generate(context.Background(), batches)
+	require.NoError(t, err)
+	cands := drainAll(t, iter)
+
+	assert.Equal(t, 1, sc.calls["q/landing"], "a merging dependency is priced like any other")
+	require.Len(t, cands, 2, "both sides of a merge that has not landed yet")
+	assert.Equal(t, entity.DependencyAssumptionSucceeds, assumptionFor(cands[0].Path, "q/landing"))
+	assert.Equal(t, entity.DependencyAssumptionFails, assumptionFor(cands[1].Path, "q/landing"))
 }
 
 func TestBestFirst_GeneratesOnlyWhatIsPulled(t *testing.T) {
@@ -774,6 +820,28 @@ func TestBestFirst_HonorsCancelledContext(t *testing.T) {
 		assert.False(t, ok)
 		assert.Equal(t, entity.CandidatePath{}, c)
 	})
+
+	t.Run("a scorer that fails on a dead context ends the run", func(t *testing.T) {
+		// The loop checks ctx before each call, so a context that dies during
+		// the LAST call is the one it cannot catch — and absorbing that as an
+		// unpriceable dependency would hand an iterator back to a caller that
+		// has already gone.
+		ctx, cancel := context.WithCancel(context.Background())
+		defer cancel()
+
+		iter, err := New(cancellingScorer{cancel: cancel}).Generate(ctx, batches)
+		require.ErrorIs(t, err, context.Canceled)
+		assert.Nil(t, iter)
+	})
+}
+
+// cancellingScorer kills the context and then fails, the way a scorer whose
+// own call was cancelled would.
+type cancellingScorer struct{ cancel context.CancelFunc }
+
+func (s cancellingScorer) Score(context.Context, entity.Batch) (float64, error) {
+	s.cancel()
+	return 0, context.Canceled
 }
 
 func TestBestFirst_DefaultsScoreOutsideUnitInterval(t *testing.T) {

submitqueue/orchestrator/controller/speculate/run.go

@@ -18,6 +18,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"maps"
+	"slices"
 
 	"github.com/uber/submitqueue/platform/metrics"
 	corebatch "github.com/uber/submitqueue/submitqueue/core/batch"
@@ -240,17 +242,20 @@ func terminalPathStatus(status entity.BuildStatus) entity.SpeculationPathStatus
 // ask hands the snapshot to the queue's Speculator. Its answer is a proposal,
 // not an instruction: check decides what is actually enacted.
 //
-// The two arguments are deliberately different slices of the queue. Only
-// speculating heads are offered as action targets, because only they are open
-// to new work. Every in-flight path set is handed over, though, whatever
-// state its head is in: a path holds its CI slot until its build actually
-// stops, so a merging head's superseded siblings and a cancelling head's live
-// builds spend the budget just like a speculating head's do. Hiding them
-// would let the allocator count occupied slots as free and oversubscribe CI.
+// Both arguments carry the whole queue, whatever state each batch is in. A
+// head's dependencies are the facts its paths are built from, so a dependency
+// withheld is one the Speculator has to plan around blind. Every in-flight
+// path set goes over for the same reason: a path holds its CI slot until its
+// build actually stops, so a merging head's superseded siblings and a
+// cancelling head's live builds spend the budget just like a speculating
+// head's do. Hiding either would let the allocator count occupied slots as
+// free and oversubscribe CI.
 //
-// Passing foreign sets cannot widen what gets proposed: a path ID hashes its
+// Passing the full queue cannot widen what gets proposed: a path ID hashes its
 // head, and check rejects any proposal aimed at a head that is not
-// speculating.
+// speculating. A head this run has just decided still reads as Speculating
+// here, but it can only rebuild the path that already passed — see
+// mergeablePath — which the allocator skips as finished.
 func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]entity.Speculation, error) {
 	spec, err := c.speculators.For(speculator.Config{QueueName: queue})
 	if err != nil {
@@ -265,7 +270,7 @@ func (c *Controller) ask(ctx context.Context, queue string, snap snapshot) ([]en
 		}
 	}
 
-	proposals, err := spec.Speculate(ctx, snap.speculating, sets)
+	proposals, err := spec.Speculate(ctx, slices.Collect(maps.Values(snap.batches)), sets)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "speculator_errors", 1)
 		return nil, fmt.Errorf("speculator failed for queue %s: %w", queue, err)

submitqueue/orchestrator/controller/speculate/run_test.go

@@ -94,7 +94,7 @@ func (h *runHarness) failPublishTo(topic string) {
 	h.failTopic = topic
 }
 
-// speculatedOver returns the IDs of the heads the Speculator was offered.
+// speculatedOver returns the IDs of the batches the Speculator was offered.
 func (h *runHarness) speculatedOver() []string {
 	ids := make([]string, 0, len(h.spec.gotBatches))
 	for _, b := range h.spec.gotBatches {
@@ -280,8 +280,8 @@ func TestRun_PassesSnapshotToSpeculator(t *testing.T) {
 	require.NoError(t, h.run(head))
 
 	require.Equal(t, 1, spec.calls)
-	require.Len(t, spec.gotBatches, 1)
-	assert.Equal(t, head, spec.gotBatches[0].ID, "only speculating heads are action targets")
+	assert.ElementsMatch(t, []string{dep1, dep2, head, merging.ID}, h.speculatedOver(),
+		"every batch the run read, in no particular order")
 	require.Len(t, spec.gotSets, 1)
 	assert.Equal(t, int32(3), spec.gotSets[0].Version)
 }
@@ -962,8 +962,8 @@ func TestRun_SpeculatorSeesPathSetsOfNonOpenHeads(t *testing.T) {
 
 	require.NoError(t, h.run(head))
 
-	assert.Equal(t, []entity.Batch{open}, spec.gotBatches,
-		"only an open head may be an action target")
+	assert.ElementsMatch(t, []entity.Batch{open, merging}, spec.gotBatches,
+		"a closed head is still a fact the open ones are planned against")
 	require.Len(t, spec.gotSets, 2, "every in-flight path set counts against the budget")
 	assert.Equal(t, merging.ID, spec.gotSets[1].Head)
 }

submitqueue/orchestrator/controller/speculate/speculate_test.go

@@ -35,14 +35,14 @@ import (
 )
 
 // quietSpeculator proposes nothing, which is what tests of the message-level
-// branches want: the run happens but changes no paths. It records the heads it
-// was offered, so a test can assert that a run reached them at all.
+// branches want: the run happens but changes no paths. It records the batches
+// it was offered, so a test can assert that a run reached them at all.
 type quietSpeculator struct {
-	heads []entity.Batch
+	saw []entity.Batch
 }
 
 func (s *quietSpeculator) Speculate(_ context.Context, batches []entity.Batch, _ []entity.SpeculationPathSet) ([]entity.Speculation, error) {
-	s.heads = append(s.heads, batches...)
+	s.saw = append(s.saw, batches...)
 	return nil, nil
 }
 
@@ -239,8 +239,8 @@ func TestProcess_TerminalReplansQueue(t *testing.T) {
 		Return(entity.SpeculationPathSet{}, storage.ErrNotFound)
 
 	require.NoError(t, h.process(t, ctrl, batch.ID))
-	assert.Equal(t, []entity.Batch{dependent}, h.spec.heads,
-		"the dependent must be re-planned against the terminal outcome")
+	assert.ElementsMatch(t, []entity.Batch{batch, dependent}, h.spec.saw,
+		"the dependent must be re-planned against the terminal outcome, which it can only be weighed against if the terminal batch comes too")
 }
 
 // A Merging batch is the merge stage's to finish; the run still happens for the