fix(recovery): detect undispatched assigned intents

internal/schedulerrecovery/evaluate.go

@@ -24,11 +24,21 @@ type ProviderRetry struct {
 	OverdueAge time.Duration `json:"overdue_age_nanoseconds"`
 }
 
+// AssignedIntent identifies central queue ownership that never produced a
+// corresponding GARM instance. Sibling scale sets can keep both the process
+// heartbeat and provider retry journal moving while this exact dispatch path
+// is no longer scheduled.
+type AssignedIntent struct {
+	ID  string        `json:"id"`
+	Age time.Duration `json:"age_nanoseconds"`
+}
+
 type Observation struct {
 	ObservedAt      time.Time
 	ActiveIntents   int
 	PendingCreates  []PendingCreate
 	OverdueRetries  []ProviderRetry
+	StaleAssigned   []AssignedIntent
 	ManagerUptime   time.Duration
 	LastRecoveryAt  time.Time
 	HeartbeatAt     time.Time
@@ -48,8 +58,9 @@ func Evaluate(policy Policy, observation Observation) Decision {
 	if observation.ActiveIntents == 0 {
 		return Decision{Reason: "no-admitted-demand"}
 	}
-	stuck := make([]string, 0, len(observation.PendingCreates)+len(observation.OverdueRetries))
+	stuck := make([]string, 0, len(observation.PendingCreates)+len(observation.OverdueRetries)+len(observation.StaleAssigned))
 	overdueRetry := false
+	staleAssigned := false
 	for _, pending := range observation.PendingCreates {
 		if pending.CreateAttempt == 0 && pending.Age >= policy.MinimumStuckAge {
 			stuck = append(stuck, pending.ID)
@@ -61,13 +72,19 @@ func Evaluate(policy Policy, observation Observation) Decision {
 			overdueRetry = true
 		}
 	}
+	for _, assigned := range observation.StaleAssigned {
+		if assigned.Age >= policy.MinimumStuckAge {
+			stuck = append(stuck, assigned.ID)
+			staleAssigned = true
+		}
+	}
 	if len(stuck) == 0 {
 		return Decision{Reason: "no-stale-undispatched-instance"}
 	}
 	// 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 {
+	if !overdueRetry && !staleAssigned && !observation.HeartbeatAt.IsZero() && observation.ObservedAt.Sub(observation.HeartbeatAt) < policy.HeartbeatStale {
 		return Decision{Reason: "dispatcher-heartbeat-current", Stuck: stuck}
 	}
 	if observation.ManagerUptime < policy.MinimumUptime {
@@ -80,5 +97,8 @@ func Evaluate(policy Policy, observation Observation) Decision {
 	if overdueRetry {
 		reason = "stale-provider-retry-past-next-allowed"
 	}
+	if staleAssigned {
+		reason = "stale-assigned-intent-without-instance"
+	}
 	return Decision{Recover: true, Reason: reason, Stuck: stuck}
 }

internal/schedulerrecovery/evaluate_test.go

@@ -59,3 +59,17 @@ func TestEvaluateRecoversOverdueProviderRetryDespiteCurrentSiblingHeartbeat(t *t
 		Stuck: []string{"scale-set:example:2:job:stuck"},
 	}, Evaluate(policy, observation))
 }
+
+func TestEvaluateRecoversStaleAssignedIntentDespiteCurrentSiblingHeartbeat(t *testing.T) {
+	t.Parallel()
+	now := time.Date(2026, 8, 26, 13, 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: 8, ManagerUptime: time.Hour,
+		HeartbeatAt:   now.Add(-10 * time.Second),
+		StaleAssigned: []AssignedIntent{{ID: "intent-skipped", Age: 2 * time.Minute}},
+	}
+	require.Equal(t, Decision{
+		Recover: true, Reason: "stale-assigned-intent-without-instance", Stuck: []string{"intent-skipped"},
+	}, Evaluate(policy, observation))
+}

internal/schedulerrecovery/observe.go

@@ -17,13 +17,14 @@ type CommandObserver struct {
 }
 
 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"`
+	ObservedAt           time.Time        `json:"observed_at"`
+	ActiveIntents        int              `json:"active_intents"`
+	PendingCreates       []PendingCreate  `json:"pending_creates"`
+	OverdueRetries       []ProviderRetry  `json:"overdue_provider_retries"`
+	StaleAssigned        []AssignedIntent `json:"stale_assigned_intents"`
+	ManagerUptimeSeconds int64            `json:"manager_uptime_seconds"`
+	LastRecoveryAt       time.Time        `json:"last_recovery_at"`
+	RecoveryRunning      bool             `json:"recovery_running"`
 }
 
 func (observer CommandObserver) Validate() error {
@@ -74,9 +75,14 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error
 			return Observation{}, fmt.Errorf("scheduler observation contains an invalid overdue provider retry")
 		}
 	}
+	for _, assigned := range decoded.StaleAssigned {
+		if assigned.ID == "" || assigned.Age < 0 {
+			return Observation{}, fmt.Errorf("scheduler observation contains an invalid stale assigned intent")
+		}
+	}
 	return Observation{
 		ObservedAt: decoded.ObservedAt, ActiveIntents: decoded.ActiveIntents,
-		PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries,
+		PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries, StaleAssigned: decoded.StaleAssigned,
 		ManagerUptime:  time.Duration(decoded.ManagerUptimeSeconds) * time.Second,
 		LastRecoveryAt: decoded.LastRecoveryAt, RecoveryRunning: decoded.RecoveryRunning,
 	}, nil

internal/schedulerrecovery/observe_test.go

@@ -17,6 +17,7 @@ func TestCommandObserverDecodesStrictSnapshot(t *testing.T) {
 	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, []AssignedIntent{{ID: "assigned-1", Age: 2 * time.Minute}}, observation.StaleAssigned)
 	require.Equal(t, 2, observation.ActiveIntents)
 	require.Equal(t, 10*time.Minute, observation.ManagerUptime)
 	require.Equal(t, 2*time.Minute, observation.PendingCreates[0].Age)
@@ -45,7 +46,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}],"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}`)
+		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}],"stale_assigned_intents":[{"id":"assigned-1","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":

internal/schedulerrecovery/recover.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"slices"
 	"strings"
@@ -27,6 +28,62 @@ type Result struct {
 	Error      string
 }
 
+type resultWire struct {
+	AttemptID  string    `json:"attempt_id"`
+	Suppressed bool      `json:"suppressed"`
+	Checkpoint string    `json:"checkpoint,omitempty"`
+	Progressed []string  `json:"progressed,omitempty"`
+	Remaining  []string  `json:"remaining,omitempty"`
+	Recovered  bool      `json:"recovered"`
+	FinishedAt time.Time `json:"finished_at"`
+	Error      string    `json:"error,omitempty"`
+}
+
+func (result Result) MarshalJSON() ([]byte, error) {
+	return json.Marshal(resultWire{
+		AttemptID: result.AttemptID, Suppressed: result.Suppressed, Checkpoint: result.Checkpoint,
+		Progressed: result.Progressed, Remaining: result.Remaining, Recovered: result.Recovered,
+		FinishedAt: result.FinishedAt, Error: result.Error,
+	})
+}
+
+func (result *Result) UnmarshalJSON(data []byte) error {
+	var normalized resultWire
+	if err := json.Unmarshal(data, &normalized); err != nil {
+		return err
+	}
+	if normalized.AttemptID != "" || !normalized.FinishedAt.IsZero() {
+		*result = Result{
+			AttemptID: normalized.AttemptID, Suppressed: normalized.Suppressed, Checkpoint: normalized.Checkpoint,
+			Progressed: normalized.Progressed, Remaining: normalized.Remaining, Recovered: normalized.Recovered,
+			FinishedAt: normalized.FinishedAt, Error: normalized.Error,
+		}
+		return nil
+	}
+	// v1 was deployed before an explicit JSON contract and therefore used Go
+	// field names. Decode it once; the next atomic state write normalizes every
+	// retained result without losing cooldown or recovery history.
+	var legacy struct {
+		AttemptID  string
+		Suppressed bool
+		Checkpoint string
+		Progressed []string
+		Remaining  []string
+		Recovered  bool
+		FinishedAt time.Time
+		Error      string
+	}
+	if err := json.Unmarshal(data, &legacy); err != nil {
+		return err
+	}
+	*result = Result{
+		AttemptID: legacy.AttemptID, Suppressed: legacy.Suppressed, Checkpoint: legacy.Checkpoint,
+		Progressed: legacy.Progressed, Remaining: legacy.Remaining, Recovered: legacy.Recovered,
+		FinishedAt: legacy.FinishedAt, Error: legacy.Error,
+	}
+	return nil
+}
+
 type AttemptStore interface {
 	Active(context.Context) ([]Attempt, error)
 	Begin(context.Context, Attempt) (bool, error)

internal/schedulerrecovery/store_test.go

@@ -68,3 +68,18 @@ func TestFileStorePersistsMonotonicHeartbeat(t *testing.T) {
 	require.Equal(t, Heartbeat{At: at.Add(time.Minute), Progress: "job-2"}, heartbeat)
 	require.ErrorContains(t, reopened.RecordHeartbeat(context.Background(), Heartbeat{At: at.Add(-time.Second), Progress: "job-0"}), "moved backwards")
 }
+
+func TestFileStoreReadsLegacyResultAndNormalizesNextWrite(t *testing.T) {
+	t.Parallel()
+	directory := t.TempDir()
+	path := filepath.Join(directory, "state.json")
+	legacy := `{"schema_version":1,"heartbeat":{},"active":{},"finished":[{"AttemptID":"attempt-1","Recovered":false,"FinishedAt":"2026-08-26T15:38:36Z","Error":"incomplete"}]}`
+	require.NoError(t, os.WriteFile(path, []byte(legacy), 0o600))
+	store := FileStore{Path: path, LockPath: filepath.Join(directory, "state.lock")}
+	_, err := store.ReadHeartbeat(context.Background())
+	require.NoError(t, err)
+	data, err := os.ReadFile(path)
+	require.NoError(t, err)
+	require.Contains(t, string(data), `"finished_at": "2026-08-26T15:38:36Z"`)
+	require.NotContains(t, string(data), `"FinishedAt"`)
+}