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.** `snapshot.batchesForSpeculator()` returns every batch the run read, sorted by ID so a plan never varies with map iteration order, and `ask` passes that. `ask`'s doc comment argued for the narrow slice and is rewritten.

Widening only stays safe if the snapshot is honest about who is still open, because both the Speculator and `check` read head eligibility off that same map — a stale entry fools both filters at once. Two things had to change for that to hold:

- `recordOutcome` now records `Merging` alongside the terminal states. It previously recorded terminal outcomes only, reasoning that merging resolves nothing for dependents — true, and every terminal-state check still treats a merging batch as unresolved. But a head this run had just handed to the merge stage went on reading as `Speculating`, so the Speculator would offer it and `check` would wave it through, funding a fresh build for a batch already on its way out of the queue.
- A head whose outcome lost the compare-and-swap is withheld from the Speculator entirely. Its recorded state still says `Speculating` while another writer has moved it on, and this run cannot know where to — so it says nothing about it rather than something wrong. `finalize` already dropped such heads from the open list for exactly this reason; now the batch list agrees.

**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
- `TestRun_MergedHeadIsNoLongerOfferedAsOpen`, `TestRun_HeadThatLostTheOutcomeRaceIsWithheld` — the snapshot handed over never describes a head the run has closed as still open
- `TestBestFirst_HonorsCancelledContext/a scorer that fails on a dead context ends the run`

The last two were added for defects a review of this branch turned up; each was confirmed to fail against the code as it stood before its 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.

A separate correctness defect in the merge gate — a head could merge on a *fails* assumption that had not come true — was found while investigating this one. It is stacked on top of this PR rather than folded in, since nothing here depends on it.

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/finalize.go

@@ -206,18 +206,22 @@ func (c *Controller) commitOutcome(ctx context.Context, snap *snapshot, batch en
 	return c.applyOutcome(ctx, snap.store, batch, decision, snap.isTrigger(batch.ID))
 }
 
-// recordOutcome writes an outcome's terminal state back into the snapshot so
-// the rest of the run reasons from it, exactly as it would from an outcome
-// recorded before the run started. It is called only once that state is
+// recordOutcome writes the state an outcome moved the batch to back into the
+// snapshot so the rest of the run reasons from it, exactly as it would from an
+// outcome recorded before the run started. It is called only once that state is
 // durable — see commitOutcome — because everything concluded about the
 // batches stacked on this one is derived from it.
 //
-// Only a terminal outcome is recorded. Merging is not terminal — a head
-// stacked on this one assumed it would *succeed*, and it has not yet — so a
-// merge outcome resolves nothing for anybody else.
+// Merging is recorded like the terminal outcomes even though it resolves
+// nothing for anybody else: a head stacked on this one assumed it would
+// *succeed*, and it has not yet, so every terminal-state check still treats it
+// as unresolved. Recording it buys something else — the batch stops reading as
+// a head open to new work. Head eligibility is read off this map, both by the
+// Speculator and by check, so a head left saying Speculating after the run has
+// handed it to the merge stage is a head the run will happily fund again.
 func (c *Controller) recordOutcome(snap *snapshot, batchID string, decision outcome) {
-	state, terminal := decision.terminalState()
-	if !terminal {
+	state, written := decision.writtenState()
+	if !written {
 		return
 	}
 	batch := snap.batches[batchID]
@@ -254,7 +258,8 @@ func (c *Controller) applyOutcome(ctx context.Context, store storage.Storage, ba
 		}
 
 	case outcomeFail, outcomeCancel:
-		state, terminal = decision.terminalState()
+		state, _ = decision.writtenState()
+		terminal = state.IsTerminal()
 		// A batch decided by a cascade is not the one on the message, so no
 		// retry or dead letter would ever come back to it — give it a recovery
 		// message of its own before it turns terminal.

submitqueue/orchestrator/controller/speculate/outcome.go

@@ -35,11 +35,14 @@ const (
 	outcomeCancel outcome = "cancel"
 )
 
-// terminalState returns the batch state an outcome writes, and whether the
-// outcome leaves the batch terminal. Merge is the odd one out: it hands the
+// writtenState returns the batch state an outcome moves the batch to.
+// outcomeWait moves it nowhere, so it reports false. Merge is the odd one out
+// among the states it can return: it is not terminal, because it hands the
 // batch to the merge stage, which owns the terminal write that follows.
-func (v outcome) terminalState() (entity.BatchState, bool) {
+func (v outcome) writtenState() (entity.BatchState, bool) {
 	switch v {
+	case outcomeMerge:
+		return entity.BatchStateMerging, true
 	case outcomeFail:
 		return entity.BatchStateFailed, true
 	case outcomeCancel:

submitqueue/orchestrator/controller/speculate/run.go

@@ -240,17 +240,21 @@ 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. The
+// Speculator is handed every batch the run read, because a head's dependencies
+// are the facts its paths are built from: a dependency that has moved on to
+// merging is no longer a candidate for work, but it is still the reason a path
+// assumes what it does, and withholding it leaves the Speculator guessing about
+// a batch it cannot see. Every in-flight path set is handed 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
-// head, and check rejects any proposal aimed at a head that is not
-// speculating.
+// Passing the full queue cannot widen what gets proposed. A path ID hashes its
+// head, and both this function and check read head eligibility off the same
+// snapshot — which finalize keeps current, so a head this run has already
+// decided no longer reads as open to work.
 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 +269,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, snap.batchesForSpeculator(), sets)
 	if err != nil {
 		metrics.NamedCounter(c.metricsScope, opName, "speculator_errors", 1)
 		return nil, fmt.Errorf("speculator failed for queue %s: %w", queue, err)