engine: safely recover serial frontier staging attempts

docs/design.md

@@ -289,9 +289,10 @@ bad PR, bounce it, and land the good PRs.
   the queue from open auto-merge PRs. It may repeat a staging run; a PR already
   released to the forge may finish while shunt restarts.
   With `SHUNT_STATE_PATH`, shunt persists the pending frontier, linger state,
-  bisection counters, and active batch metadata in bbolt. Active batches are
-  re-staged after restore so no additional PR is released from a pre-restart
-  result that may have been invalidated by a base change.
+  bisection counters, and active batch metadata in bbolt. On restore, shunt
+  removes each known staging branch and re-stages its PRs, so no additional PR
+  is released from a pre-restart result that may have been invalidated by a base
+  or source-head change.
 
 ## Observability
 

internal/checkpoint/bolt/bolt_test.go

@@ -32,6 +32,8 @@ func TestStoreSavesLoadsAndDeletesQueue(t *testing.T) {
 			StagingSHA:     "stage-4",
 			BaseGeneration: 2,
 			Outcome:        "failure",
+			Phase:          "bisecting",
+			PhaseSince:     time.Date(2026, 6, 24, 10, 1, 0, 0, time.UTC),
 		}},
 		LingerSince:     time.Date(2026, 6, 24, 10, 0, 0, 0, time.UTC),
 		BaseGeneration:  2,
@@ -102,7 +104,7 @@ func TestQueueSnapshotJSONReadsLegacyFieldNames(t *testing.T) {
 	data := []byte(`{
 		"Key":{"Owner":"o","Repo":"r","Base":"main"},
 		"Pending":[[1]],
-		"Active":[{"PRs":[{"Number":1,"HeadSHA":"head-1"}],"StagingBranch":"mq/main/staging","StagingSHA":"stage-1","BaseGeneration":2,"Outcome":"failure"}],
+		"Active":[{"PRs":[{"Number":1,"HeadSHA":"head-1"}],"StagingBranch":"mq/main/staging","StagingSHA":"stage-1","BaseGeneration":2,"Outcome":"failure","Phase":"bisecting","PhaseSince":"2026-06-24T10:01:00Z"}],
 		"LingerSince":"0001-01-01T00:00:00Z",
 		"BaseGeneration":2,
 		"StagingSequence":7
@@ -120,4 +122,7 @@ func TestQueueSnapshotJSONReadsLegacyFieldNames(t *testing.T) {
 	if got := snapshot.StagingSequence; got != 7 {
 		t.Fatalf("staging sequence = %d, want 7", got)
 	}
+	if got := snapshot.Active[0].Phase; got != "bisecting" {
+		t.Fatalf("phase = %q, want bisecting", got)
+	}
 }

internal/engine/checkpoint.go

@@ -3,6 +3,8 @@ package engine
 import (
 	"context"
 	"fmt"
+	"sort"
+	"strings"
 
 	"github.com/rbtr/shunt/internal/checkpoint"
 )
@@ -33,7 +35,9 @@ func (e *Engine) loadCheckpoint(ctx context.Context) error {
 	if snapshot.Key != e.queueKey() {
 		return fmt.Errorf("queue checkpoint key mismatch: got %s/%s@%s", snapshot.Key.Owner, snapshot.Key.Repo, snapshot.Key.Base)
 	}
-	e.applySnapshot(snapshot)
+	if err := e.applySnapshot(ctx, snapshot); err != nil {
+		return err
+	}
 	e.checkpointLoaded = true
 	e.checkpointExists = true
 	return nil
@@ -85,6 +89,8 @@ func (e *Engine) snapshot() checkpoint.QueueSnapshot {
 			StagingSHA:     a.stagingSHA,
 			BaseGeneration: a.baseGen,
 			Outcome:        a.outcome,
+			Phase:          a.phase,
+			PhaseSince:     a.phaseSince,
 		}
 	}
 	return checkpoint.QueueSnapshot{
@@ -97,19 +103,38 @@ func (e *Engine) snapshot() checkpoint.QueueSnapshot {
 	}
 }
 
-func (e *Engine) applySnapshot(snapshot checkpoint.QueueSnapshot) {
+// applySnapshot restores unresolved candidates. In-flight staging attempts are
+// deliberately not resumed: after a restart, Shunt cannot prove their gate
+// evidence still includes the current base and source heads. It removes each
+// Shunt-owned staging branch and re-queues its PRs for fresh staging instead.
+func (e *Engine) applySnapshot(ctx context.Context, snapshot checkpoint.QueueSnapshot) error {
 	e.pending = clonePending(snapshot.Pending)
 	for _, active := range snapshot.Active {
+		if e.ownsStagingBranch(active.StagingBranch) {
+			if err := e.fc.DeleteBranch(ctx, e.cfg.Owner, e.cfg.Repo, active.StagingBranch); err != nil {
+				e.logger.Warn("failed to delete restored staging branch", "branch", active.StagingBranch, "error", err)
+			}
+		} else {
+			e.logger.Warn("not deleting restored branch outside staging namespace", "branch", active.StagingBranch)
+		}
 		nums := make([]int, len(active.PRs))
 		for i, pr := range active.PRs {
 			nums[i] = pr.Number
 		}
 		e.pending = append(e.pending, nums)
 	}
+	sort.SliceStable(e.pending, func(i, j int) bool {
+		return e.pending[i][0] < e.pending[j][0]
+	})
 	e.active = nil
 	e.lingerSince = snapshot.LingerSince
 	e.baseGen = snapshot.BaseGeneration
 	e.stagingSeq = snapshot.StagingSequence
+	return nil
+}
+
+func (e *Engine) ownsStagingBranch(branch string) bool {
+	return e.cfg.StagingBranch != "" && strings.HasPrefix(branch, e.cfg.StagingBranch+"-")
 }
 
 func clonePending(in [][]int) [][]int {

internal/engine/engine_test.go

@@ -840,6 +840,13 @@ func TestCheckpointRestoresActiveBatchByRestaging(t *testing.T) {
 	if store.saved == nil || len(store.saved.Active) != 1 {
 		t.Fatalf("checkpoint active batches = %v, want 1 active batch", store.saved)
 	}
+	if got := store.saved.Active[0].Phase; got != "waiting_gate" {
+		t.Fatalf("checkpoint phase = %q, want waiting_gate", got)
+	}
+	if store.saved.Active[0].PhaseSince.IsZero() {
+		t.Fatal("checkpoint phase timestamp is zero")
+	}
+	staleBranch := m.stagingBranches[0]
 
 	restarted := New(cfg, m, m)
 	if err := restarted.Reconcile(context.Background()); err != nil {
@@ -848,6 +855,9 @@ func TestCheckpointRestoresActiveBatchByRestaging(t *testing.T) {
 	if got := len(m.staged); got != 2 {
 		t.Fatalf("restored active batch should be restaged; staged = %d, want 2", got)
 	}
+	if got := fmt.Sprint(m.calls); !strings.Contains(got, "delete:"+staleBranch) {
+		t.Fatalf("calls = %s, want stale staging branch deleted", got)
+	}
 	if got := fmt.Sprint(m.merged); got != "[]" {
 		t.Fatalf("merged after restage = %s, want []", got)
 	}
@@ -860,6 +870,56 @@ func TestCheckpointRestoresActiveBatchByRestaging(t *testing.T) {
 	}
 }
 
+func TestCheckpointRestoreKeepsActiveCandidateBeforeLaterPendingWork(t *testing.T) {
+	m := newMock(-1, 1, 2, 3)
+	store := &memoryCheckpointStore{saved: &checkpoint.QueueSnapshot{
+		Key:     checkpoint.QueueKey{Owner: "o", Repo: "r", Base: "main"},
+		Pending: [][]int{{3}},
+		Active: []checkpoint.ActiveBatchSnapshot{{
+			PRs: []checkpoint.PullRequestSnapshot{
+				{Number: 1, HeadSHA: "head-1"},
+				{Number: 2, HeadSHA: "head-2"},
+			},
+			StagingBranch: "mq/main/staging-old",
+			StagingSHA:    "stage-old",
+		}},
+	}}
+	cfg := Config{Owner: "o", Repo: "r", Base: "main", StatusCtx: "merge-queue", StagingBranch: "mq/main/staging", Checkpoint: store}
+
+	if err := New(cfg, m, m).Reconcile(context.Background()); err != nil {
+		t.Fatalf("restore queue: %v", err)
+	}
+	if got := fmt.Sprint(m.staged); got != "[[1 2]]" {
+		t.Fatalf("staged = %s, want the restored active candidate before later work", got)
+	}
+	if got := fmt.Sprint(m.calls); !strings.Contains(got, "delete:mq/main/staging-old") {
+		t.Fatalf("calls = %s, want restored staging branch deleted", got)
+	}
+}
+
+func TestCheckpointRestoreDoesNotDeleteBranchOutsideStagingNamespace(t *testing.T) {
+	m := newMock(-1, 1)
+	store := &memoryCheckpointStore{saved: &checkpoint.QueueSnapshot{
+		Key: checkpoint.QueueKey{Owner: "o", Repo: "r", Base: "main"},
+		Active: []checkpoint.ActiveBatchSnapshot{{
+			PRs:           []checkpoint.PullRequestSnapshot{{Number: 1, HeadSHA: "head-1"}},
+			StagingBranch: "main",
+			StagingSHA:    "stage-old",
+		}},
+	}}
+	cfg := Config{Owner: "o", Repo: "r", Base: "main", StatusCtx: "merge-queue", StagingBranch: "mq/main/staging", Checkpoint: store}
+
+	if err := New(cfg, m, m).Reconcile(context.Background()); err != nil {
+		t.Fatalf("restore queue: %v", err)
+	}
+	if got := fmt.Sprint(m.calls); strings.Contains(got, "delete:main") {
+		t.Fatalf("calls = %s, must not delete a branch outside the staging namespace", got)
+	}
+	if got := fmt.Sprint(m.staged); got != "[[1]]" {
+		t.Fatalf("staged = %s, want PR requeued for fresh staging", got)
+	}
+}
+
 func TestCheckpointRestartRestagesRemainderAfterReleasedPRMerges(t *testing.T) {
 	m := newMock(-1, 1, 2)
 	store := &memoryCheckpointStore{}

mq/checkpoint/checkpoint.go

@@ -29,15 +29,17 @@ type QueueSnapshot struct {
 	StagingSequence int                   `json:"StagingSequence"`
 }
 
-// ActiveBatchSnapshot records a staging branch currently waiting on its gate.
-// On restore the engine re-queues these PRs for fresh staging (applySnapshot
-// sets active = nil) — it does not resume the staged branch.
+// ActiveBatchSnapshot records a staging attempt that was in flight. On
+// restore the engine removes its staging branch and re-queues its PRs for fresh
+// staging, rather than trusting evidence from before the restart.
 type ActiveBatchSnapshot struct {
 	PRs            []PullRequestSnapshot `json:"PRs"`
 	StagingBranch  string                `json:"StagingBranch"`
 	StagingSHA     string                `json:"StagingSHA"`
 	BaseGeneration int                   `json:"BaseGeneration"`
 	Outcome        string                `json:"Outcome"`
+	Phase          string                `json:"Phase"`
+	PhaseSince     time.Time             `json:"PhaseSince"`
 }
 
 // PullRequestSnapshot is the PR identity needed to re-queue a batch.
@@ -92,6 +94,9 @@ func (s QueueSnapshot) Validate() error {
 		if active.Outcome != "" && active.Outcome != "success" && active.Outcome != "failure" && active.Outcome != "cancelled" && active.Outcome != "error" {
 			return fmt.Errorf("queue checkpoint active batch %d has invalid outcome %q", i, active.Outcome)
 		}
+		if active.Phase != "" && active.Phase != "waiting_gate" && active.Phase != "waiting_merge" && active.Phase != "bisecting" {
+			return fmt.Errorf("queue checkpoint active batch %d has invalid phase %q", i, active.Phase)
+		}
 		if len(active.PRs) == 0 {
 			return fmt.Errorf("queue checkpoint active batch %d has no PRs", i)
 		}