feat(observe): expose terminal tombstone lifecycle

config/example-runner-1.yaml

@@ -6,7 +6,7 @@ platform:
 
 control_plane:
   manager: garm
-  manager_version: v0.2.1-nddev.82
+  manager_version: v0.2.1-nddev.83
   scheduling_mode: scale-set
   provider: incus
   provider_version: v0.1.5-nddev.79

config/example-runner-2.yaml

@@ -6,7 +6,7 @@ platform:
 
 control_plane:
   manager: garm
-  manager_version: v0.2.1-nddev.82
+  manager_version: v0.2.1-nddev.83
   scheduling_mode: scale-set
   provider: incus
   provider_version: v0.1.5-nddev.79

config/example-runner-3.yaml

@@ -6,7 +6,7 @@ platform:
 
 control_plane:
   manager: garm
-  manager_version: v0.2.1-nddev.82
+  manager_version: v0.2.1-nddev.83
   scheduling_mode: scale-set
   provider: incus
   provider_version: v0.1.5-nddev.79

config/example-runner-4.yaml

@@ -6,7 +6,7 @@ platform:
 
 control_plane:
   manager: garm
-  manager_version: v0.2.1-nddev.82
+  manager_version: v0.2.1-nddev.83
   scheduling_mode: scale-set
   provider: incus
   provider_version: v0.1.5-nddev.79

config/example-services.yaml

@@ -24,7 +24,7 @@ platform:
 
 control_plane:
   manager: garm
-  manager_version: v0.2.1-nddev.82
+  manager_version: v0.2.1-nddev.83
   scheduling_mode: scale-set
   provider: incus
   provider_version: v0.1.5-nddev.79

config/garm-derivative.yaml

@@ -1,6 +1,6 @@
 schema_version: 1
 artifact: garm
-derivative_version: v0.2.1-nddev.82
+derivative_version: v0.2.1-nddev.83
 upstream:
   repository: https://github.com/cloudbase/garm
   release: v0.2.1
@@ -90,8 +90,8 @@ patches:
     purpose: Bound every upstream Linux and Gitea bootstrap curl path to two retries after the initial attempt so GARM and the provider enforce the same three-total-attempt contract.
 overlays:
   - path: third_party/garm/overlay/workers/scaleset/queue_intent.go
-    sha256: 180e60554d264358d5a71c3b16b90d897a82cdb51a7b3d70b587029f97e5e3fe
-    purpose: Fsync-backed central queue admission where bounded terminal tombstones prevent delayed JobAssigned redelivery from resurrecting completed work, while provisional capacity, repository binding, running rehydration and stable phase-entry semantics remain intact.
+    sha256: 31486d264f0a2b8dc63bb902ee49163c32e4c8774a6ad5b6d29efad92a228a03
+    purpose: Fsync-backed central queue admission where bounded terminal tombstones prevent delayed JobAssigned redelivery from resurrecting completed work and emit one structured suppression event, while provisional capacity, repository binding, running rehydration and stable phase-entry semantics remain intact.
   - path: third_party/garm/overlay/workers/scaleset/queue_intent_test.go
     sha256: 2f7d98f63033d73d972a9e91ca7cca0934e09abbc62fa971b6ed99e328f531b9
     purpose: Deterministic terminal-redelivery suppression, disabled-scale-set isolation, provisional bootstrap, authoritative repository enrichment, orphan-start rehydration, phase-entry and expiry, priority, fairness, width, idempotency, acknowledgement and concurrent-selection tests.
@@ -115,7 +115,7 @@ build:
     - sqlite_omit_load_extension
   reproducible_rebuilds: 2
   maximum_required_glibc: "2.34"
-  binary_sha256: 18089d33be778705d0a5b20a87f5b1ddea06b0a7bc9daab883a8014b0507d741
+  binary_sha256: f9594a5ff67643a3b1f9e3357db6a96b45e1ec83797ff9ad0189302a264bb238
 runtime_contract:
   queue_intent_schema_version: 5
   event_driven_scale_set_wake: true

internal/fleetobserve/metrics.go

@@ -139,6 +139,8 @@ func RenderPrometheus(snapshot Snapshot, now time.Time, maxStaleness time.Durati
 	gauge(&output, "gha_fleet_queue_intents_stored", "Queue intents stored in the current journal generation.", float64(snapshot.Queue.Stored))
 	gauge(&output, "gha_fleet_queue_intents_active", "Unexpired queue intents owned by central admission.", float64(snapshot.Queue.Active))
 	gauge(&output, "gha_fleet_queue_intents_expired", "Expired queue intents awaiting the next writer cleanup transaction.", float64(snapshot.Queue.Expired))
+	gauge(&output, "gha_fleet_queue_terminal_tombstones", "Completed job identities retained to suppress delayed lifecycle redelivery.", float64(snapshot.Queue.TerminalJobs))
+	gauge(&output, "gha_fleet_queue_terminal_tombstone_next_expiry_seconds", "Seconds until the next bounded terminal job tombstone expires.", float64(snapshot.Queue.TerminalNextExpirySeconds))
 	gauge(&output, "gha_fleet_queue_intents_in_flight", "Acquiring, acquired, assigned or running queue intents.", float64(snapshot.Queue.InFlight))
 	gauge(&output, "gha_fleet_queue_oldest_age_seconds", "Age of the oldest active GitHub queue intent.", float64(snapshot.Queue.OldestQueueAgeSeconds))
 	gauge(&output, "gha_fleet_queue_uncovered_running", "Running queue intents without a durable created or warm-claimed execution lease.", float64(snapshot.Queue.UncoveredRunning))

internal/fleetobserve/metrics_test.go

@@ -45,6 +45,8 @@ func TestRenderPrometheusIsDeterministicAndBounded(t *testing.T) {
 		"gha_fleet_provider_warm_preemptions_total 0\n",
 		"gha_fleet_queue_journal_generation 1\n",
 		"gha_fleet_queue_intents_active 0\n",
+		"gha_fleet_queue_terminal_tombstones 0\n",
+		"gha_fleet_queue_terminal_tombstone_next_expiry_seconds 0\n",
 		"gha_fleet_queue_intents_in_flight 0\n",
 		"gha_fleet_queue_uncovered_running 0\n",
 		"gha_fleet_queue_missing_runner_request_id 0\n",

internal/fleetobserve/observe.go

@@ -199,6 +199,8 @@ type QueueSummary struct {
 	Stored                          int              `json:"stored"`
 	Active                          int              `json:"active"`
 	Expired                         int              `json:"expired"`
+	TerminalJobs                    int              `json:"terminal_jobs"`
+	TerminalNextExpirySeconds       int64            `json:"terminal_next_expiry_seconds"`
 	InFlight                        int              `json:"in_flight"`
 	OldestQueueAgeSeconds           int64            `json:"oldest_queue_age_seconds"`
 	UncoveredRunning                int              `json:"uncovered_running"`
@@ -560,14 +562,16 @@ func summarizeQueue(snapshot queueintent.Snapshot, platform config.Config, now t
 		knownScaleSets[pool.ScaleSetName] = struct{}{}
 	}
 	summary := QueueSummary{
-		Generation:            snapshot.Generation,
-		Stored:                snapshot.Stored,
-		Active:                len(snapshot.Active),
-		Expired:               snapshot.Expired,
-		ByState:               make(map[string]int),
-		OldestStateAgeSeconds: make(map[string]int64),
-		ByPriority:            make(map[int]int),
-		ByScaleSet:            make(map[string]int),
+		Generation:                snapshot.Generation,
+		Stored:                    snapshot.Stored,
+		Active:                    len(snapshot.Active),
+		Expired:                   snapshot.Expired,
+		TerminalJobs:              snapshot.TerminalJobs,
+		TerminalNextExpirySeconds: snapshot.TerminalNextExpirySeconds,
+		ByState:                   make(map[string]int),
+		OldestStateAgeSeconds:     make(map[string]int64),
+		ByPriority:                make(map[int]int),
+		ByScaleSet:                make(map[string]int),
 	}
 	for scaleSet := range knownScaleSets {
 		summary.ByScaleSet[scaleSet] = 0

internal/queueintent/queueintent.go

@@ -88,10 +88,12 @@ type Journal struct {
 }
 
 type Snapshot struct {
-	Generation uint64
-	Stored     int
-	Expired    int
-	Active     []Intent
+	Generation                uint64
+	Stored                    int
+	Expired                   int
+	TerminalJobs              int
+	TerminalNextExpirySeconds int64
+	Active                    []Intent
 }
 
 func (r Reader) HasActive(ctx context.Context) (bool, error) {
@@ -118,6 +120,16 @@ func (r Reader) ReadActive(ctx context.Context) (Snapshot, error) {
 	now := r.now()
 	active := make([]Intent, 0, len(journal.Intents))
 	expired := 0
+	terminalNextExpirySeconds := int64(0)
+	for _, expiry := range journal.TerminalJobs {
+		remaining := int64(expiry.Sub(now).Seconds())
+		if remaining < 0 {
+			remaining = 0
+		}
+		if terminalNextExpirySeconds == 0 || remaining < terminalNextExpirySeconds {
+			terminalNextExpirySeconds = remaining
+		}
+	}
 	for _, intent := range journal.Intents {
 		if intent.ExpiresAt.After(now) {
 			active = append(active, intent)
@@ -134,7 +146,11 @@ func (r Reader) ReadActive(ctx context.Context) (Snapshot, error) {
 		}
 		return active[left].Key < active[right].Key
 	})
-	return Snapshot{Generation: journal.Generation, Stored: len(journal.Intents), Expired: expired, Active: active}, nil
+	return Snapshot{
+		Generation: journal.Generation, Stored: len(journal.Intents), Expired: expired,
+		TerminalJobs: len(journal.TerminalJobs), TerminalNextExpirySeconds: terminalNextExpirySeconds,
+		Active: active,
+	}, nil
 }
 
 // ActiveForScaleSet reports whether the named scale set of the named tenant

internal/queueintent/queueintent_test.go

@@ -73,6 +73,28 @@ func TestReaderReturnsOnlyActiveIntentsInDeterministicOrder(t *testing.T) {
 	}
 }
 
+func TestReaderReportsBoundedTerminalTombstones(t *testing.T) {
+	now := time.Date(2026, 8, 26, 14, 0, 0, 0, time.UTC)
+	path := writeFixture(t, `{
+  "schema_version": 5,
+  "generation": 9,
+  "updated_at": "2026-08-26T13:59:00Z",
+  "intents": {},
+  "repositories": {},
+  "terminal_jobs": {
+    "11111111-1111-4111-8111-111111111111": "2026-08-27T14:00:00Z",
+    "22222222-2222-4222-8222-222222222222": "2026-08-26T15:00:00Z"
+  }
+}`)
+	snapshot, err := (Reader{Path: path, Now: func() time.Time { return now }}).ReadActive(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if snapshot.TerminalJobs != 2 || snapshot.TerminalNextExpirySeconds != 3600 {
+		t.Fatalf("terminal snapshot = %#v", snapshot)
+	}
+}
+
 func TestReaderRejectsFinalComponentSymlink(t *testing.T) {
 	directory := t.TempDir()
 	target := filepath.Join(directory, "real-journal.json")

internal/schedulerrecovery/controller.go

@@ -41,6 +41,29 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error)
 	if controller.Observer == nil || controller.Heartbeat == nil || controller.Attempts == nil || controller.Executor == nil || controller.Events == nil || controller.Now == nil {
 		return Decision{}, Result{}, fmt.Errorf("scheduler recovery controller is incomplete")
 	}
+	active, err := controller.Attempts.Active(ctx)
+	if err != nil {
+		return Decision{}, Result{}, fmt.Errorf("read active recovery attempt: %w", err)
+	}
+	if len(active) > 1 {
+		return Decision{}, Result{}, fmt.Errorf("multiple active recovery attempts require deterministic repair")
+	}
+	if len(active) == 1 {
+		attempt := active[0]
+		if err := controller.emit(ctx, Event{At: controller.Now().UTC(), State: "recovering", Reason: "resume-interrupted-recovery", AttemptID: attempt.ID, Stuck: attempt.Stuck}); err != nil {
+			return Decision{}, Result{}, err
+		}
+		result, recoveryErr := resumeAcquired(ctx, attempt, controller.Attempts, controller.Executor, controller.Now)
+		terminal := Event{At: controller.Now().UTC(), State: "recovered", Reason: "resume-interrupted-recovery", AttemptID: attempt.ID, Stuck: result.Remaining}
+		if recoveryErr != nil {
+			terminal.State = "failed"
+			terminal.Error = recoveryErr.Error()
+		}
+		if err := controller.emit(ctx, terminal); err != nil {
+			return Decision{}, result, err
+		}
+		return Decision{Reason: "resume-interrupted-recovery", Stuck: attempt.Stuck}, result, recoveryErr
+	}
 	observation, err := controller.Observer.Observe(ctx)
 	if err != nil {
 		return Decision{}, Result{}, fmt.Errorf("observe scheduler: %w", err)

internal/schedulerrecovery/controller_test.go

@@ -73,3 +73,24 @@ func TestControllerReportsHealthyCurrentHeartbeat(t *testing.T) {
 	require.Equal(t, "dispatcher-heartbeat-current", decision.Reason)
 	require.Equal(t, "healthy", events.events[0].State)
 }
+
+func TestControllerFinishesInterruptedRecoveryAfterRestartProgressed(t *testing.T) {
+	t.Parallel()
+	at := time.Date(2026, 8, 26, 14, 13, 56, 0, time.UTC)
+	attempt := NewAttempt(at, []string{"retry-1"})
+	store := &memoryAttempts{active: map[string]Attempt{attempt.ID: attempt}}
+	executor := &faultExecutor{progressed: []string{"retry-1"}}
+	events := &eventRecorder{}
+	controller := Controller{
+		Policy:   Policy{MinimumStuckAge: time.Minute, MinimumUptime: time.Minute, Cooldown: time.Minute, HeartbeatStale: time.Minute},
+		Observer: staticObserver{}, Heartbeat: staticHeartbeat{}, Attempts: store, Executor: executor,
+		Events: events, Now: func() time.Time { return at.Add(time.Minute) },
+	}
+	decision, result, err := controller.Tick(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, "resume-interrupted-recovery", decision.Reason)
+	require.True(t, result.Recovered)
+	require.Zero(t, executor.restarts)
+	require.Empty(t, store.active)
+	require.Equal(t, []string{"recovering", "recovered"}, []string{events.events[0].State, events.events[1].State})
+}

internal/schedulerrecovery/recover.go

@@ -28,10 +28,32 @@ type Result struct {
 }
 
 type AttemptStore interface {
+	Active(context.Context) ([]Attempt, error)
 	Begin(context.Context, Attempt) (bool, error)
 	Finish(context.Context, Result) error
 }
 
+func resumeAcquired(ctx context.Context, attempt Attempt, store AttemptStore, executor Executor, now func() time.Time) (Result, error) {
+	progressed, remaining, progressErr := executor.AwaitProgress(ctx, attempt)
+	if progressErr == nil && len(remaining) == 0 {
+		result := Result{
+			AttemptID: attempt.ID, Progressed: slices.Clone(progressed), Recovered: true,
+			FinishedAt: now().UTC(),
+		}
+		if err := store.Finish(ctx, result); err != nil {
+			return result, fmt.Errorf("finish resumed recovery attempt: %w", err)
+		}
+		return result, nil
+	}
+	// The previous process may have died before or during the manager restart.
+	// Re-running the checkpoint-first sequence is idempotent; restricting it to
+	// the still-stuck identities avoids replaying work already proven progressed.
+	if len(remaining) > 0 {
+		attempt.Stuck = slices.Clone(remaining)
+	}
+	return recoverAcquired(ctx, attempt, store, executor, now)
+}
+
 type Executor interface {
 	Checkpoint(context.Context, Attempt) (string, error)
 	RestartDispatcher(context.Context, Attempt) error

internal/schedulerrecovery/recover_test.go

@@ -12,30 +12,46 @@ import (
 
 type memoryAttempts struct {
 	mu       sync.Mutex
-	active   map[string]bool
+	active   map[string]Attempt
 	finished []Result
 }
 
 func (store *memoryAttempts) Begin(_ context.Context, attempt Attempt) (bool, error) {
 	store.mu.Lock()
 	defer store.mu.Unlock()
 	if store.active == nil {
-		store.active = map[string]bool{}
+		store.active = map[string]Attempt{}
 	}
-	if store.active[attempt.ID] {
+	if _, exists := store.active[attempt.ID]; exists {
 		return false, nil
 	}
-	store.active[attempt.ID] = true
+	for _, result := range store.finished {
+		if result.AttemptID == attempt.ID {
+			return false, nil
+		}
+	}
+	store.active[attempt.ID] = attempt
 	return true, nil
 }
 
 func (store *memoryAttempts) Finish(_ context.Context, result Result) error {
 	store.mu.Lock()
 	defer store.mu.Unlock()
 	store.finished = append(store.finished, result)
+	delete(store.active, result.AttemptID)
 	return nil
 }
 
+func (store *memoryAttempts) Active(_ context.Context) ([]Attempt, error) {
+	store.mu.Lock()
+	defer store.mu.Unlock()
+	attempts := make([]Attempt, 0, len(store.active))
+	for _, attempt := range store.active {
+		attempts = append(attempts, attempt)
+	}
+	return attempts, nil
+}
+
 type faultExecutor struct {
 	mu          sync.Mutex
 	checkpoints int

internal/schedulerrecovery/store.go

@@ -74,6 +74,21 @@ func (store FileStore) Begin(_ context.Context, attempt Attempt) (bool, error) {
 	return acquired, err
 }
 
+func (store FileStore) Active(_ context.Context) ([]Attempt, error) {
+	var attempts []Attempt
+	err := store.locked(func(state *fileState) error {
+		attempts = make([]Attempt, 0, len(state.Active))
+		for _, attempt := range state.Active {
+			attempts = append(attempts, attempt)
+		}
+		slices.SortFunc(attempts, func(left, right Attempt) int {
+			return left.StartedAt.Compare(right.StartedAt)
+		})
+		return nil
+	})
+	return attempts, err
+}
+
 func (store FileStore) Finish(_ context.Context, result Result) error {
 	return store.locked(func(state *fileState) error {
 		if _, exists := state.Active[result.AttemptID]; !exists {

internal/schedulerrecovery/store_test.go

@@ -20,6 +20,9 @@ func TestFileStorePersistsAttemptAndSuppressesReplay(t *testing.T) {
 	require.True(t, acquired)
 
 	reopened := FileStore{Path: store.Path, LockPath: store.LockPath}
+	active, err := reopened.Active(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, []Attempt{attempt}, active)
 	acquired, err = reopened.Begin(context.Background(), attempt)
 	require.NoError(t, err)
 	require.False(t, acquired)

scripts/build-garm-nddev.sh

@@ -19,7 +19,7 @@ set -Eeuo pipefail
 # Every value below is the manifest's. Editing one here detaches the build
 # from the provenance it is reviewed against, which is why the region is
 # regenerated and compared rather than maintained.
-readonly derivative_version="v0.2.1-nddev.82"
+readonly derivative_version="v0.2.1-nddev.83"
 readonly upstream_repository="https://github.com/cloudbase/garm"
 readonly upstream_commit="154638445c3949c1958b01812f69d9a1e4d82684"
 readonly build_image="docker.io/library/golang@sha256:116d58cbd88c1297624acc6e967a060012422bacf9930927e23fb719189c6f36"
@@ -32,7 +32,7 @@ readonly build_module_mode="vendor"
 readonly build_tags="osusergo,netgo,sqlite_omit_load_extension"
 readonly build_reproducible_rebuilds="2"
 readonly build_maximum_required_glibc="2.34"
-readonly expected_binary_sha256="18089d33be778705d0a5b20a87f5b1ddea06b0a7bc9daab883a8014b0507d741"
+readonly expected_binary_sha256="f9594a5ff67643a3b1f9e3357db6a96b45e1ec83797ff9ad0189302a264bb238"
 readonly patch_paths=(
   "third_party/garm/patches/0001-event-driven-reconciliation.patch"
   "third_party/garm/patches/0002-central-queue-admission.patch"
@@ -98,7 +98,7 @@ readonly overlay_paths=(
   "third_party/garm/overlay/workers/provider/nddev_create_retry_test.go"
 )
 readonly overlay_sha256s=(
-  "180e60554d264358d5a71c3b16b90d897a82cdb51a7b3d70b587029f97e5e3fe"
+  "31486d264f0a2b8dc63bb902ee49163c32e4c8774a6ad5b6d29efad92a228a03"
   "2f7d98f63033d73d972a9e91ca7cca0934e09abbc62fa971b6ed99e328f531b9"
   "611d67a74ff626f04cc0240409a22848525a06c2ae4dc7180c4b75001fb4b599"
   "9379d69b8a40295e176fdfb0587b299fabda69de9708e63672c41546fd9a3508"