engine: record replaced staging attempts

internal/engine/engine.go

@@ -763,6 +763,7 @@ func (e *Engine) checkActive(ctx context.Context) (bool, error) {
 				}
 
 				prs := a.prs
+				e.markBatchSuperseded(a, "staging gate produced no result")
 				e.cleanupBatch(ctx, a)
 				refs := make([]gitops.MergedRef, len(prs))
 				for i, pr := range prs {
@@ -943,6 +944,7 @@ func (e *Engine) discardIneligibleActive(ctx context.Context, a *activeBatch, pr
 		return
 	}
 	remaining := removeNum(numbersOf(a.prs), pr)
+	e.markBatchSuperseded(a, "a candidate became ineligible")
 	e.cleanupBatch(ctx, a)
 	e.enqueueWithState("requeued after another PR became ineligible", remaining)
 	e.observeQueueExit(pr, "ineligible")
@@ -996,6 +998,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge
 		if current.State != "open" {
 			e.skipLand(ctx, staged, current, fmt.Sprintf("state changed to %q", current.State), len(a.prs) > 1, a.debugURL, false)
 			e.requeueActiveRemainder("requeued after an earlier PR left the batch", a.prs[1:])
+			e.markBatchSuperseded(a, "a PR left the batch before landing")
 			e.cleanupBatch(ctx, a)
 			return true, merged, nil
 		}
@@ -1013,6 +1016,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge
 				true,
 			)
 			e.requeueActiveRemainder("requeued after PR head changed", a.prs)
+			e.markBatchSuperseded(a, "a PR head changed before landing")
 			e.cleanupBatch(ctx, a)
 			return true, merged, nil
 		}
@@ -1031,6 +1035,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge
 			}
 			e.skipLand(ctx, staged, current, "auto-merge is no longer scheduled", len(a.prs) > 1, a.debugURL, true)
 			e.requeueActiveRemainder("requeued after auto-merge was cancelled", a.prs)
+			e.markBatchSuperseded(a, "auto-merge was cancelled before landing")
 			e.cleanupBatch(ctx, a)
 			return true, merged, nil
 		}
@@ -1067,6 +1072,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge
 				}
 				e.skipLand(ctx, staged, current, "auto-merge is no longer scheduled", len(a.prs) > 1, a.debugURL, true)
 				e.requeueActiveRemainder("requeued after auto-merge was cancelled", a.prs)
+				e.markBatchSuperseded(a, "auto-merge was cancelled during recovery")
 				e.cleanupBatch(ctx, a)
 				return true, merged, nil
 			}
@@ -1125,6 +1131,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge
 				return true, merged, nil
 			}
 			e.requeueActiveRemainder("retrying after forge merge recovery", a.prs)
+			e.markBatchSuperseded(a, "native auto-merge timed out")
 			e.cleanupBatch(ctx, a)
 			if err := e.scheduleAutomerge(ctx, current); err != nil {
 				return false, merged, fmt.Errorf("restore auto-merge for PR #%d: %w", staged.Number, err)
@@ -1179,6 +1186,7 @@ func (e *Engine) land(ctx context.Context, a *activeBatch) (resolved bool, merge
 			}
 			e.skipLand(ctx, staged, current, "auto-merge is no longer scheduled", len(a.prs) > 1, a.debugURL, true)
 			e.requeueActiveRemainder("requeued after auto-merge was cancelled", a.prs)
+			e.markBatchSuperseded(a, "auto-merge was cancelled before release")
 			e.cleanupBatch(ctx, a)
 			return true, merged, nil
 		}
@@ -1791,6 +1799,7 @@ func (e *Engine) freeSlotForEarlierPending(ctx context.Context) {
 		return
 	}
 	a := e.active[idx]
+	e.markBatchSuperseded(a, "an earlier queue entry took the staging slot")
 	e.cleanupBatch(ctx, a)
 	e.enqueueWithState("requeued; waiting for an earlier queue entry", numbersOf(a.prs))
 	e.logger.Info("speculative batch requeued for earlier candidate", "prs", numbersOf(a.prs), "earlier", e.pending[0])
@@ -1800,6 +1809,7 @@ func (e *Engine) requeueStaleActive(ctx context.Context, a *activeBatch) {
 	if e.requeuedTreeNode(ctx, a, "base branch advanced", "re-queued: base branch advanced mid-test") {
 		return
 	}
+	e.markBatchSuperseded(a, "base branch advanced")
 	e.cleanupBatch(ctx, a)
 	e.enqueueWithState("requeued after base branch advanced", numbersOf(a.prs))
 	e.logger.Info("stale speculative batch requeued after base advanced", "prs", numbersOf(a.prs))
@@ -1809,11 +1819,26 @@ func (e *Engine) requeueChangedActive(ctx context.Context, a *activeBatch) {
 	if e.requeuedTreeNode(ctx, a, "a pinned candidate changed", "re-queued: a pinned candidate changed mid-test") {
 		return
 	}
+	e.markBatchSuperseded(a, "a PR head changed")
 	e.cleanupBatch(ctx, a)
 	e.enqueueWithState("requeued after PR head changed", numbersOf(a.prs))
 	e.logger.Info("active batch requeued after PR head changed", "prs", numbersOf(a.prs))
 }
 
+// markBatchSuperseded records that a staging attempt ended without a source
+// decision because its candidates will be tested again. The transition lets
+// callers retire the old staging row instead of leaving it as running.
+// `superseded` means "this attempt was replaced", not "a PR failed".
+func (e *Engine) markBatchSuperseded(a *activeBatch, reason string) {
+	if a.stagingBranch == "" {
+		return
+	}
+	e.recordTransition(Transition{
+		Kind: "node_superseded", PRs: numbersOf(a.prs), StagingBranch: a.stagingBranch,
+		RunID: a.runID, LineagePath: a.lineagePath, Reason: reason,
+	})
+}
+
 // supersedeSpeculative discards a fanout-staged bisection node whose gate ran
 // against an accumulator the resolved frontier no longer matches, and re-queues
 // the same node — keeping its run id and lineage path — so it re-stages on the

internal/engine/engine_test.go

@@ -1796,6 +1796,15 @@ func TestMissingGateRetriesWithBackoffThenBounces(t *testing.T) {
 		if got := e.active[0].missingGateRetries; got != retry {
 			t.Fatalf("missing-gate retries = %d, want %d", got, retry)
 		}
+		found := false
+		for _, tr := range e.Transitions() {
+			if tr.Kind == "node_superseded" && tr.StagingBranch == m.stagingBranches[retry-1] {
+				found = true
+			}
+		}
+		if !found {
+			t.Fatalf("retry %d did not report the replaced staging attempt", retry)
+		}
 	}
 	if got := fmt.Sprint(m.calls); !strings.Contains(got, "delete:mq/main/staging") {
 		t.Fatalf("calls = %s, want stale staging branch deleted", got)
@@ -2191,6 +2200,16 @@ func TestNativeMergeTimeoutBlocksRestoresAndRequeues(t *testing.T) {
 	if got := fmt.Sprint(e.pending); got != "[[1]]" {
 		t.Fatalf("pending = %s, want timed-out PR requeued", got)
 	}
+	var superseded bool
+	for _, transition := range e.Transitions() {
+		if transition.Kind == "node_superseded" && transition.StagingBranch == m.stagingBranches[0] {
+			superseded = true
+			break
+		}
+	}
+	if !superseded {
+		t.Fatalf("transitions = %v, want timed-out staging attempt marked superseded", e.Transitions())
+	}
 	if got := strings.Join(m.comments[1], "\n"); !strings.Contains(got, "Merge did not complete") {
 		t.Fatalf("outcome comment = %q, want timeout outcome", got)
 	}

internal/engine/transitions.go

@@ -8,8 +8,9 @@ import (
 
 // Transition is a structured record of one merge-queue lifecycle event that
 // happened during a single Reconcile call: a batch was staged, its gate
-// passed, it was bisected, a PR bounced, or a PR landed. It exists so a
-// caller can persist what actually happened without parsing log text.
+// passed, it was bisected, a staging attempt was superseded, a PR bounced, or
+// a PR landed. It exists so a caller can persist what actually happened without
+// parsing log text.
 //
 // Additive to the engine's existing logger.Info/Warn calls at the same
 // sites — recording a Transition never changes what gets logged.
@@ -21,13 +22,12 @@ type Transition struct {
 	// whole bisection root was torn down after its base branch advanced or a
 	// pinned candidate changed mid-test; its candidates are re-queued as a
 	// fresh root and no source decision is published), or "node_superseded" (a
-	// speculatively-staged bisection node whose gate ran against an accumulator
-	// the resolved frontier no longer matches; re-staged on the correct
-	// baseline, no source decision).
+	// staging attempt was replaced without a source decision, including a
+	// speculative node whose accumulator no longer matches).
 	Kind string `json:"kind"`
-	// PRs is the batch's PR set for "staged"/"gate_success"/"bisected"/"held",
-	// or the single terminated/landed PR (as a length-1 slice) for
-	// "bounced"/"landed".
+	// PRs is the batch's PR set for "staged"/"gate_success"/"bisected"/
+	// "held"/"node_superseded", or the single terminated/landed PR (as a
+	// length-1 slice) for "bounced"/"landed".
 	PRs           []int  `json:"prs"`
 	StagingBranch string `json:"staging_branch"`
 	// RunID and LineagePath identify this batch's position in a bisection
@@ -37,8 +37,9 @@ type Transition struct {
 	// name strings from scratch.
 	RunID       string `json:"run_id"`
 	LineagePath string `json:"lineage_path"`
-	// Reason is set for "bounced" (why the PR was rejected) and for "held"
-	// (the held gate outcome: "success", "failure", or "error").
+	// Reason is set for "bounced" (why the PR was rejected), "held" (the held
+	// gate outcome: "success", "failure", or "error"), and
+	// "node_superseded" (why the staging attempt was replaced).
 	Reason string `json:"reason,omitempty"`
 	// EventID is a deterministic key for transitions that drive an
 	// irreversible side effect (a merge counter, a bounce notification): the

mq/mq.go

@@ -197,8 +197,9 @@ func (eng *Engine) Reconcile(ctx context.Context) error {
 }
 
 // Transition is a structured record of one merge-queue lifecycle event
-// (a batch staged, its gate passed, it bisected, a PR bounced, or a PR
-// landed) recorded during a Reconcile call. See LastTransitions.
+// (a batch staged, its gate passed, it bisected, a staging attempt was
+// superseded, a PR bounced, or a PR landed) recorded during a Reconcile call.
+// See LastTransitions.
 type Transition = engine.Transition
 
 // LastTransitions returns the transitions recorded during the most