fix(scheduler): recover overdue provider retries

internal/schedulerrecovery/evaluate.go

@@ -15,10 +15,20 @@ type PendingCreate struct {
 	CreateAttempt int           `json:"create_attempt"`
 }
 
+// ProviderRetry identifies a non-terminal provider-create retry that has not
+// advanced or cleared after its own next_allowed_at deadline. Unlike the
+// dispatcher heartbeat, this is scoped to one exact create path: sibling
+// scale sets may continue making progress while this retry is parked.
+type ProviderRetry struct {
+	ID         string        `json:"id"`
+	OverdueAge time.Duration `json:"overdue_age_nanoseconds"`
+}
+
 type Observation struct {
 	ObservedAt      time.Time
 	ActiveIntents   int
 	PendingCreates  []PendingCreate
+	OverdueRetries  []ProviderRetry
 	ManagerUptime   time.Duration
 	LastRecoveryAt  time.Time
 	HeartbeatAt     time.Time
@@ -38,16 +48,26 @@ func Evaluate(policy Policy, observation Observation) Decision {
 	if observation.ActiveIntents == 0 {
 		return Decision{Reason: "no-admitted-demand"}
 	}
-	stuck := make([]string, 0, len(observation.PendingCreates))
+	stuck := make([]string, 0, len(observation.PendingCreates)+len(observation.OverdueRetries))
+	overdueRetry := false
 	for _, pending := range observation.PendingCreates {
 		if pending.CreateAttempt == 0 && pending.Age >= policy.MinimumStuckAge {
 			stuck = append(stuck, pending.ID)
 		}
 	}
+	for _, retry := range observation.OverdueRetries {
+		if retry.OverdueAge >= policy.MinimumStuckAge {
+			stuck = append(stuck, retry.ID)
+			overdueRetry = true
+		}
+	}
 	if len(stuck) == 0 {
 		return Decision{Reason: "no-stale-undispatched-instance"}
 	}
-	if !observation.HeartbeatAt.IsZero() && observation.ObservedAt.Sub(observation.HeartbeatAt) < policy.HeartbeatStale {
+	// A process-wide heartbeat proves only that some dispatcher work advanced.
+	// It cannot clear an exact retry that is already overdue: production has
+	// shown one scale set parked while sibling classes kept the heartbeat fresh.
+	if !overdueRetry && !observation.HeartbeatAt.IsZero() && observation.ObservedAt.Sub(observation.HeartbeatAt) < policy.HeartbeatStale {
 		return Decision{Reason: "dispatcher-heartbeat-current", Stuck: stuck}
 	}
 	if observation.ManagerUptime < policy.MinimumUptime {
@@ -56,5 +76,9 @@ func Evaluate(policy Policy, observation Observation) Decision {
 	if !observation.LastRecoveryAt.IsZero() && observation.ObservedAt.Sub(observation.LastRecoveryAt) < policy.Cooldown {
 		return Decision{Reason: "recovery-cooldown", Stuck: stuck}
 	}
-	return Decision{Recover: true, Reason: "stale-pending-create-attempt-zero", Stuck: stuck}
+	reason := "stale-pending-create-attempt-zero"
+	if overdueRetry {
+		reason = "stale-provider-retry-past-next-allowed"
+	}
+	return Decision{Recover: true, Reason: reason, Stuck: stuck}
 }

internal/schedulerrecovery/evaluate_test.go

@@ -44,3 +44,18 @@ func TestEvaluatePreventsDuplicateRecovery(t *testing.T) {
 	observation.RecoveryRunning = true
 	require.Equal(t, "recovery-already-running", Evaluate(policy, observation).Reason)
 }
+
+func TestEvaluateRecoversOverdueProviderRetryDespiteCurrentSiblingHeartbeat(t *testing.T) {
+	t.Parallel()
+	now := time.Date(2026, 8, 26, 12, 0, 0, 0, time.UTC)
+	policy := Policy{MinimumStuckAge: 90 * time.Second, MinimumUptime: 2 * time.Minute, Cooldown: 10 * time.Minute, HeartbeatStale: time.Minute}
+	observation := Observation{
+		ObservedAt: now, ActiveIntents: 9, ManagerUptime: time.Hour,
+		HeartbeatAt:    now.Add(-10 * time.Second),
+		OverdueRetries: []ProviderRetry{{ID: "scale-set:example:2:job:stuck", OverdueAge: 2 * time.Minute}},
+	}
+	require.Equal(t, Decision{
+		Recover: true, Reason: "stale-provider-retry-past-next-allowed",
+		Stuck: []string{"scale-set:example:2:job:stuck"},
+	}, Evaluate(policy, observation))
+}

internal/schedulerrecovery/observe.go

@@ -20,6 +20,7 @@ type observationOutput struct {
 	ObservedAt           time.Time       `json:"observed_at"`
 	ActiveIntents        int             `json:"active_intents"`
 	PendingCreates       []PendingCreate `json:"pending_creates"`
+	OverdueRetries       []ProviderRetry `json:"overdue_provider_retries"`
 	ManagerUptimeSeconds int64           `json:"manager_uptime_seconds"`
 	LastRecoveryAt       time.Time       `json:"last_recovery_at"`
 	RecoveryRunning      bool            `json:"recovery_running"`
@@ -68,9 +69,15 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error
 			return Observation{}, fmt.Errorf("scheduler observation contains an invalid pending create")
 		}
 	}
+	for _, retry := range decoded.OverdueRetries {
+		if retry.ID == "" || retry.OverdueAge < 0 {
+			return Observation{}, fmt.Errorf("scheduler observation contains an invalid overdue provider retry")
+		}
+	}
 	return Observation{
 		ObservedAt: decoded.ObservedAt, ActiveIntents: decoded.ActiveIntents,
-		PendingCreates: decoded.PendingCreates, ManagerUptime: time.Duration(decoded.ManagerUptimeSeconds) * time.Second,
+		PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries,
+		ManagerUptime:  time.Duration(decoded.ManagerUptimeSeconds) * time.Second,
 		LastRecoveryAt: decoded.LastRecoveryAt, RecoveryRunning: decoded.RecoveryRunning,
 	}, nil
 }

internal/schedulerrecovery/observe_test.go

@@ -16,6 +16,7 @@ func TestCommandObserverDecodesStrictSnapshot(t *testing.T) {
 	require.NoError(t, err)
 	observation, err := (CommandObserver{Argv: []string{command, "-test.run=TestSchedulerRecoveryObserverHelper", "--", "valid"}, Timeout: 10 * time.Second}).Observe(context.Background())
 	require.NoError(t, err)
+	require.Equal(t, []ProviderRetry{{ID: "retry-1", OverdueAge: 2 * time.Minute}}, observation.OverdueRetries)
 	require.Equal(t, 2, observation.ActiveIntents)
 	require.Equal(t, 10*time.Minute, observation.ManagerUptime)
 	require.Equal(t, 2*time.Minute, observation.PendingCreates[0].Age)
@@ -44,7 +45,7 @@ func TestSchedulerRecoveryObserverHelper(t *testing.T) {
 	}
 	switch os.Args[separator+1] {
 	case "valid":
-		fmt.Print(`{"observed_at":"2026-08-24T10:00:00Z","active_intents":2,"pending_creates":[{"id":"instance-1","age_nanoseconds":120000000000,"create_attempt":0}],"manager_uptime_seconds":600,"last_recovery_at":"0001-01-01T00:00:00Z","recovery_running":false}`)
+		fmt.Print(`{"observed_at":"2026-08-24T10:00:00Z","active_intents":2,"pending_creates":[{"id":"instance-1","age_nanoseconds":120000000000,"create_attempt":0}],"overdue_provider_retries":[{"id":"retry-1","overdue_age_nanoseconds":120000000000}],"manager_uptime_seconds":600,"last_recovery_at":"0001-01-01T00:00:00Z","recovery_running":false}`)
 	case "unknown":
 		fmt.Print(`{"observed_at":"2026-08-24T10:00:00Z","active_intents":1,"manager_uptime_seconds":1,"unexpected":true}`)
 	case "invalid":