engine: resolve bisection from durable ordered evidence

.github/workflows/ci.yml

@@ -14,7 +14,7 @@ jobs:
       - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5
       - uses: actions/setup-go@40f1582b2485089dde7abd97c1529aa768e1baff
         with:
-          go-version: "1.25.12"
+          go-version: "1.25.13"
       - name: gofmt
         run: test -z "$(gofmt -l .)" || { echo 'gofmt needed:'; gofmt -l .; exit 1; }
       - name: vet

docs/design.md

@@ -78,22 +78,21 @@ State, per `(repo, base)`:
 - `lingerSince` — when the idle engine first saw ready PRs while the optional
   batch-accumulation window was active.
 
-The engine has a checkpoint boundary around that state: when configured with its
-consumer-side `CheckpointStore`, it loads one queue snapshot before the first
-reconcile tick, saves after each tick that leaves queue work in progress, and
-deletes the snapshot once the queue is idle. The production command can use
-bbolt with `SHUNT_STATE_PATH` or Postgres with `SHUNT_POSTGRES_DSN`; otherwise
-releases keep the historical process-local state.
-
-When Postgres is configured, each `(owner, repo, base)` must first acquire its
-durable queue lease before loading a checkpoint or calling the forge. The lease
-is renewed once per `Reconcile()` call, and that call receives a deadline at
-half the configured lease TTL so no holder keeps mutating after its lease can
-expire. A replica that cannot acquire it does nothing for that queue; one that
-takes it over drops process-local queue and comment caches, then reloads the
-durable checkpoint. Restored active batches are re-staged, as on process
-restart. bbolt and in-memory state are single-process options and do not
-coordinate replicas.
+The engine saves queue state through `CheckpointStore`. It loads one snapshot
+before the first reconcile. It saves work after each reconcile. It deletes the
+snapshot when the queue is idle.
+
+bbolt is the recommended local durable store. Set `SHUNT_STATE_PATH` to use
+bbolt. Set `SHUNT_POSTGRES_DSN` to use Postgres. Postgres is a supported store
+for replicas that need a shared queue lease. Without either setting, queue
+state exists only in the process.
+
+A Postgres replica acquires its queue lease before it loads state or calls the
+forge. Each `Reconcile()` call renews the lease. The call ends before half of
+the lease period expires. A replica that does not hold the lease does not act.
+A replica that acquires the lease clears its local queue and comment state. It
+then loads the durable checkpoint. bbolt and in-memory state support one
+process only.
 
 Each `Reconcile()` tick advances one step. Ticks are driven by relevant
 Forgejo/Gitea webhooks when available, with `SHUNT_POLL_INTERVAL` as the
@@ -200,29 +199,25 @@ else:               mid = len/2
                     pending.push_front(nums[:mid], nums[mid:])   # test first half next
 ```
 
-Because candidates are just lists of PR numbers and staging is always rebuilt
-from the *current* base tip, a successful sub-batch that advances the base is
-handled safely: any later speculative staging run from the old base generation is
-abandoned and re-queued, then re-staged before it can land.
-
-The same preflight protects active batches from PR updates. While a gate is
-running, shunt rechecks every open PR head before accepting the result. During
-landing, it rechecks each PR immediately before release. A changed head is
-re-queued for fresh staging. A pull-request webhook wakes this path promptly;
-polling remains the backstop for missed webhook deliveries.
-
-The neutral `checkpoint` package defines the snapshot DTOs. The engine owns the
-consumer-side store interface, and the concrete bbolt implementation lives in
-the more specific `checkpoint/bolt` package. Restored active batches are
-conservatively re-queued for fresh staging instead of resuming an old staging
-branch/run, so shunt does not release additional PRs from a result that may now
-be stale. A PR released before the restart may still finish through the forge;
-it was released only after a passing batch, and the remaining PRs are re-staged
-on the resulting base. The production command wires the default bbolt
-implementation when `SHUNT_STATE_PATH` is set. That store persists one snapshot
-per `(owner, repo, base)` and keeps the binary static/CGO-free;
-operators should place the database on persistent storage if they want queue
-state to survive pod replacement or host reboots.
+Each bisection root uses one fixed base commit. Shunt builds each root child
+on this base commit. The exact gate key includes the base commit, merge style,
+and ordered PR heads. Shunt discards a speculative result when its exact key no
+longer matches the ordered frontier.
+
+Shunt checks each active PR head before it accepts a gate result. Shunt checks
+again before it releases a PR. A changed head starts fresh staging. A pull
+request webhook starts this check early. Polling starts the check if a webhook
+is missed.
+
+The `checkpoint` package defines the snapshot types. The engine owns the store
+interface. The `checkpoint/bolt` package provides the bbolt store. On restart,
+Shunt resumes an active staging attempt only when its stored evidence is valid.
+Otherwise, Shunt removes the staging branch and derives fresh queue work. A PR
+that Shunt released before restart can still merge through the forge.
+
+The bbolt store saves one snapshot for each `(owner, repo, base)` queue. The
+store keeps the binary static and free of CGO. Use persistent storage when queue
+state must survive host or pod replacement.
 
 ### Worked example
 

internal/checkpoint/bolt/bolt_test.go

@@ -19,23 +19,48 @@ func TestStoreSavesLoadsAndDeletesQueue(t *testing.T) {
 	defer store.Close()
 
 	snapshot := checkpoint.QueueSnapshot{
-		Key: checkpoint.QueueKey{Owner: "o", Repo: "r", Base: "main"},
+		FormatVersion: checkpoint.CurrentFormatVersion,
+		Key:           checkpoint.QueueKey{Owner: "o", Repo: "r", Base: "main"},
 		Pending: [][]int{
 			{1, 2},
 			{3},
 		},
+		PendingNodes: []checkpoint.PendingNodeSnapshot{{PRs: []int{3}, RunID: "run-1", Path: "r1"}},
 		Active: []checkpoint.ActiveBatchSnapshot{{
 			PRs: []checkpoint.PullRequestSnapshot{
 				{Number: 4, HeadSHA: "head-4"},
 			},
-			StagingBranch:  "mq/main/staging",
+			StagingBranch:  "mq/main/staging-4",
 			StagingSHA:     "stage-4",
 			BaseGeneration: 2,
 			Outcome:        "failure",
+			PhaseSince:     time.Date(2026, 6, 24, 10, 1, 0, 0, time.UTC),
+			RunID:          "run-1",
+			LineagePath:    "r0",
+			ExactKey:       "base-1|head-4",
+			BaseAnchor:     "base-1",
+			DebugURL:       "https://forge.example/runs/4",
+			Speculative:    true,
 		}},
 		LingerSince:     time.Date(2026, 6, 24, 10, 0, 0, 0, time.UTC),
 		BaseGeneration:  2,
 		StagingSequence: 7,
+		Trees: []checkpoint.BisectionTreeSnapshot{{
+			RunID:    "run-1",
+			Anchor:   "base-1",
+			Open:     1,
+			Accepted: []checkpoint.PullRequestSnapshot{{Number: 1, HeadSHA: "head-1"}},
+			Results:  map[string]string{"base-1|head-1": "success"},
+			Held: []checkpoint.HeldLeafSnapshot{{
+				Batch: checkpoint.ActiveBatchSnapshot{
+					PRs:           []checkpoint.PullRequestSnapshot{{Number: 2, HeadSHA: "head-2"}},
+					StagingBranch: "mq/main/staging-2",
+					StagingSHA:    "stage-2",
+				},
+				Outcome: "failure",
+			}},
+		}},
+		TransitionOutbox: []checkpoint.OutboxTransitionSnapshot{{Kind: "replaced", PRs: []int{4}, EventID: "event-4"}},
 	}
 
 	if err := store.SaveQueue(context.Background(), snapshot); err != nil {

internal/checkpoint/checkpoint.go

@@ -8,10 +8,16 @@ import (
 	pub "github.com/rbtr/shunt/mq/checkpoint"
 )
 
+const CurrentFormatVersion = pub.CurrentFormatVersion
+
 type QueueKey = pub.QueueKey
 type QueueSnapshot = pub.QueueSnapshot
 type ActiveBatchSnapshot = pub.ActiveBatchSnapshot
 type PullRequestSnapshot = pub.PullRequestSnapshot
+type PendingNodeSnapshot = pub.PendingNodeSnapshot
+type BisectionTreeSnapshot = pub.BisectionTreeSnapshot
+type HeldLeafSnapshot = pub.HeldLeafSnapshot
+type OutboxTransitionSnapshot = pub.OutboxTransitionSnapshot
 
 // Store is the public mq/checkpoint.Store (kept here for internal imports).
 type Store = pub.Store

internal/checkpoint/postgres/migrations/003_format_version.sql

@@ -0,0 +1,10 @@
+-- Add the checkpoint format version to the Postgres queue-state store.
+--
+-- The store persists the snapshot as individual columns, not a JSON blob, so
+-- QueueSnapshot.FormatVersion was silently dropped on save and always read
+-- back as 0 — unlike the bolt store, which json.Marshals the whole struct.
+-- Existing rows default to 0; the engine treats a below-current version with
+-- in-flight work as a legacy checkpoint it cannot resume exactly, discards it,
+-- and re-derives the queue from the forge.
+ALTER TABLE shunt_queue_state
+    ADD COLUMN IF NOT EXISTS format_version integer NOT NULL DEFAULT 0;

internal/checkpoint/postgres/migrations/004_snapshot_blob.sql

@@ -0,0 +1,17 @@
+-- Store the whole QueueSnapshot as one JSON document.
+--
+-- The store was written as a fixed set of columns (pending, active,
+-- linger_since, base_generation, staging_sequence, format_version). Every
+-- field added to QueueSnapshot since — the bisection Trees, PendingNodes,
+-- the TransitionOutbox, and the per-batch RunID / BaseAnchor / ExactKey /
+-- LineagePath — was silently dropped on save and read back as zero, because
+-- nothing here knew about it. The bolt and in-memory stores json.Marshal the
+-- whole struct and were unaffected; production runs on Postgres.
+--
+-- From now on SaveQueue writes the full snapshot into `snapshot` and LoadQueue
+-- reads from it. The individual columns are still written (external queries and
+-- the updated_at index depend on them) but are no longer the source of truth.
+-- Rows written before this migration have snapshot IS NULL; LoadQueue falls
+-- back to reconstructing what it can from the columns for those.
+ALTER TABLE shunt_queue_state
+    ADD COLUMN IF NOT EXISTS snapshot jsonb;

internal/checkpoint/postgres/postgres.go

@@ -24,6 +24,17 @@ var PostgresMigrationV1 string
 //go:embed migrations/002_queue_leases.sql
 var PostgresMigrationV2 string
 
+// PostgresMigrationV3 adds the checkpoint format version column.
+//
+//go:embed migrations/003_format_version.sql
+var PostgresMigrationV3 string
+
+// PostgresMigrationV4 adds the full-snapshot JSON column so no QueueSnapshot
+// field can be silently dropped by a store that predates it.
+//
+//go:embed migrations/004_snapshot_blob.sql
+var PostgresMigrationV4 string
+
 type rowScanner interface {
 	Scan(dest ...any) error
 }
@@ -102,7 +113,7 @@ func (p *Store) ApplyMigrations(ctx context.Context) error {
 	if err := p.ready(); err != nil {
 		return err
 	}
-	for _, migration := range []string{PostgresMigrationV1, PostgresMigrationV2} {
+	for _, migration := range []string{PostgresMigrationV1, PostgresMigrationV2, PostgresMigrationV3, PostgresMigrationV4} {
 		m := strings.ReplaceAll(migration, "shunt_queue_state", p.stateTbl)
 		m = strings.ReplaceAll(m, "shunt_queue_leases", p.leaseTbl)
 		if _, err := p.db.ExecContext(ctx, m); err != nil {
@@ -166,22 +177,31 @@ func (p *Store) SaveQueue(ctx context.Context, snapshot checkpoint.QueueSnapshot
 	if err != nil {
 		return fmt.Errorf("state: marshal active batches: %w", err)
 	}
+	// The whole snapshot, verbatim. This is the source of truth on load; the
+	// individual columns above are kept for external queries and the
+	// updated_at index but never lose a field the way hand-picked columns do.
+	full, err := json.Marshal(snapshot)
+	if err != nil {
+		return fmt.Errorf("state: marshal snapshot: %w", err)
+	}
 	var linger any
 	if !snapshot.LingerSince.IsZero() {
 		linger = snapshot.LingerSince.UTC()
 	}
 	_, err = p.db.ExecContext(ctx, fmt.Sprintf(`
 INSERT INTO %s (
-    owner, repo, base, pending, active, linger_since, base_generation, staging_sequence, updated_at
-) VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6, $7, $8, now())
+    owner, repo, base, pending, active, linger_since, base_generation, staging_sequence, format_version, snapshot, updated_at
+) VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6, $7, $8, $9, $10::jsonb, now())
 ON CONFLICT (owner, repo, base) DO UPDATE SET
     pending = EXCLUDED.pending,
     active = EXCLUDED.active,
     linger_since = EXCLUDED.linger_since,
     base_generation = EXCLUDED.base_generation,
     staging_sequence = EXCLUDED.staging_sequence,
+    format_version = EXCLUDED.format_version,
+    snapshot = EXCLUDED.snapshot,
     updated_at = now()
-`, p.stateTbl), snapshot.Key.Owner, snapshot.Key.Repo, snapshot.Key.Base, string(pending), string(active), linger, snapshot.BaseGeneration, snapshot.StagingSequence)
+`, p.stateTbl), snapshot.Key.Owner, snapshot.Key.Repo, snapshot.Key.Base, string(pending), string(active), linger, snapshot.BaseGeneration, snapshot.StagingSequence, snapshot.FormatVersion, string(full))
 	if err != nil {
 		return fmt.Errorf("state: save queue %s/%s@%s: %w", snapshot.Key.Owner, snapshot.Key.Repo, snapshot.Key.Base, err)
 	}
@@ -197,20 +217,33 @@ func (p *Store) LoadQueue(ctx context.Context, key checkpoint.QueueKey) (checkpo
 	if err := key.Validate(); err != nil {
 		return checkpoint.QueueSnapshot{}, false, err
 	}
-	var pendingRaw, activeRaw []byte
+	var pendingRaw, activeRaw, snapshotRaw []byte
 	var linger sql.NullTime
-	var baseGeneration, stagingSequence int
+	var baseGeneration, stagingSequence, formatVersion int
 	err := p.db.QueryRowContext(ctx, fmt.Sprintf(`
-SELECT pending, active, linger_since, base_generation, staging_sequence
+SELECT pending, active, linger_since, base_generation, staging_sequence, format_version, snapshot
 FROM %s
 WHERE owner = $1 AND repo = $2 AND base = $3
-`, p.stateTbl), key.Owner, key.Repo, key.Base).Scan(&pendingRaw, &activeRaw, &linger, &baseGeneration, &stagingSequence)
+`, p.stateTbl), key.Owner, key.Repo, key.Base).Scan(&pendingRaw, &activeRaw, &linger, &baseGeneration, &stagingSequence, &formatVersion, &snapshotRaw)
 	if errors.Is(err, sql.ErrNoRows) {
 		return checkpoint.QueueSnapshot{}, false, nil
 	}
 	if err != nil {
 		return checkpoint.QueueSnapshot{}, false, fmt.Errorf("state: load queue %s/%s@%s: %w", key.Owner, key.Repo, key.Base, err)
 	}
+	// Preferred path: the row carries the whole snapshot. Rows written before
+	// migration 004 have snapshot IS NULL and fall through to the columns.
+	if len(snapshotRaw) > 0 {
+		var snapshot checkpoint.QueueSnapshot
+		if err := json.Unmarshal(snapshotRaw, &snapshot); err != nil {
+			return checkpoint.QueueSnapshot{}, false, fmt.Errorf("state: decode snapshot: %w", err)
+		}
+		snapshot.Key = key
+		if err := snapshot.Validate(); err != nil {
+			return checkpoint.QueueSnapshot{}, false, err
+		}
+		return snapshot.Clone(), true, nil
+	}
 	var pending [][]int
 	if err := json.Unmarshal(pendingRaw, &pending); err != nil {
 		return checkpoint.QueueSnapshot{}, false, fmt.Errorf("state: decode pending queue: %w", err)
@@ -220,6 +253,7 @@ WHERE owner = $1 AND repo = $2 AND base = $3
 		return checkpoint.QueueSnapshot{}, false, fmt.Errorf("state: decode active batches: %w", err)
 	}
 	snapshot := checkpoint.QueueSnapshot{
+		FormatVersion:   formatVersion,
 		Key:             key,
 		Pending:         pending,
 		Active:          activeFromJSON(activeJSON),