fix(recovery): require exact progress and preserve blocked stalled work

docs/runbooks/recovery-evidence-contract.md

@@ -0,0 +1,53 @@
+# Scheduler recovery evidence contract
+
+## Implemented boundary
+
+A stalled identity remains observable even while restarting the dispatcher is
+unsafe. The observation command may return `recovery_blockers`, an array of
+bounded reason codes. `Evaluate` preserves `Decision.Stuck` and refuses recovery;
+the controller emits `unhealthy`, not `healthy`, for this condition. The collector
+must evaluate actual worker/claim occupancy independently from stalled-job
+observation. Do not remove worker protection to make the queue appear responsive.
+
+Progress commands must return exactly one JSON object whose `progressed` and
+`remaining` arrays form a complete disjoint partition of the attempt's expected
+identities. Missing identities, unrelated work, empty objects, duplicate IDs and
+concatenated JSON values are rejected. The recovery engine checks this contract
+for every executor, not only the command adapter. Invalid attempts are refused
+before acquiring their durable state.
+
+Interrupted attempts may verify progress but must not replay a manager restart
+from old authorization. Incomplete or unknown progress finishes the attempt as
+failed; a subsequent current observation must pass the normal policy and
+cooldown. A syntactically valid partition still needs real per-identity evidence
+from the deployment adapter. A vanished row is not a forward transition.
+
+## Coordinated rollout
+
+Pause only the recovery timer during the binary/adapter/config replacement;
+do not stop running build workers. Back up the installed recovery artifacts and
+state using the deployment's normal maintenance path. Install the new binary
+before the adapter that emits `recovery_blockers`: older binaries reject the
+unknown field rather than interpreting it as permission to restart. Verify the
+exact installed source/build/config hashes and exercise observation without an
+actual restart before re-enabling the timer.
+
+The adapter must bind checkpoints and progress to `GHA_SCHEDULER_RECOVERY_ATTEMPT`
+and the exact `GHA_SCHEDULER_RECOVERY_STUCK` set; preserve original snapshots and
+recheck eligibility immediately before a restart. Do not hold provider-journal
+locks across systemctl shutdown. An atomic admission fence and scale-set-local
+repair are separate work; a last-moment read alone does not eliminate that race.
+
+## Verification
+
+Run `go test -race ./internal/schedulerrecovery` and the repository's full checks
+on its pinned Go toolchain. The focused offline review exercised the changed
+production files with Go 1.23.2, an unchanged type-only extraction of `Heartbeat`,
+and the new standard-library regression tests. It did not execute FileStore,
+the full dependency graph, systemd, Incus, or real GitHub job delivery.
+
+The regressions cover exact partitions, unknown resume progress, fresh-decision
+requirements, blocked-but-visible incidents, and strict single-value JSON.
+Runtime acceptance additionally requires an affected real job to advance, no
+unrelated running job lost, and matched pre-start/total-workflow measurements.
+Neither a green PR nor fewer notifications is fleet throughput acceptance.

internal/schedulerrecovery/command.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"os/exec"
 	"path/filepath"
 	"strings"
@@ -78,6 +79,12 @@ func (executor CommandExecutor) AwaitProgress(ctx context.Context, attempt Attem
 	if err := decoder.Decode(&progress); err != nil {
 		return nil, nil, fmt.Errorf("decode progress output: %w", err)
 	}
+	if err := decoder.Decode(new(any)); err != io.EOF {
+		return nil, nil, fmt.Errorf("progress output must contain exactly one JSON value")
+	}
+	if err := validateProgress(attempt.Stuck, progress.Progressed, progress.Remaining); err != nil {
+		return nil, nil, err
+	}
 	return progress.Progressed, progress.Remaining, nil
 }
 

internal/schedulerrecovery/controller.go

@@ -3,6 +3,7 @@ package schedulerrecovery
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 )
 
@@ -75,7 +76,7 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error)
 	observation.HeartbeatAt = heartbeat.At
 	decision := Evaluate(controller.Policy, observation)
 	state := "healthy"
-	if decision.Recover {
+	if decision.Recover || strings.HasPrefix(decision.Reason, "recovery-blocked:") {
 		state = "unhealthy"
 	}
 	if err := controller.emit(ctx, Event{At: observation.ObservedAt, State: state, Reason: decision.Reason, Stuck: decision.Stuck}); err != nil {
@@ -84,6 +85,9 @@ func (controller Controller) Tick(ctx context.Context) (Decision, Result, error)
 	if !decision.Recover {
 		return decision, Result{}, nil
 	}
+	if err := validateProgress(decision.Stuck, nil, decision.Stuck); err != nil {
+		return decision, Result{}, fmt.Errorf("invalid recovery identities: %w", err)
+	}
 	attempt := NewAttempt(observation.ObservedAt, decision.Stuck)
 	acquired, err := controller.Attempts.Begin(ctx, attempt)
 	if err != nil {

internal/schedulerrecovery/evaluate.go

@@ -1,6 +1,10 @@
 package schedulerrecovery
 
-import "time"
+import (
+	"slices"
+	"strings"
+	"time"
+)
 
 type Policy struct {
 	MinimumStuckAge time.Duration
@@ -40,10 +44,12 @@ type Observation struct {
 	OverdueRetries       []ProviderRetry
 	StaleAssigned        []AssignedIntent
 	CapacityBackpressure bool
-	ManagerUptime        time.Duration
-	LastRecoveryAt       time.Time
-	HeartbeatAt          time.Time
-	RecoveryRunning      bool
+	// RecoveryBlockers preserve stalled identities while a manager-wide restart is unsafe.
+	RecoveryBlockers []string
+	ManagerUptime    time.Duration
+	LastRecoveryAt   time.Time
+	HeartbeatAt      time.Time
+	RecoveryRunning  bool
 }
 
 type Decision struct {
@@ -82,6 +88,11 @@ func Evaluate(policy Policy, observation Observation) Decision {
 	if len(stuck) == 0 {
 		return Decision{Reason: "no-stale-undispatched-instance"}
 	}
+	if len(observation.RecoveryBlockers) != 0 {
+		blockers := slices.Clone(observation.RecoveryBlockers)
+		slices.Sort(blockers)
+		return Decision{Reason: "recovery-blocked:" + strings.Join(slices.Compact(blockers), ","), Stuck: stuck}
+	}
 	// 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.

internal/schedulerrecovery/observe.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os/exec"
 	"path/filepath"
 	"strings"
@@ -26,6 +27,7 @@ type observationOutput struct {
 	ManagerUptimeSeconds int64            `json:"manager_uptime_seconds"`
 	LastRecoveryAt       time.Time        `json:"last_recovery_at"`
 	RecoveryRunning      bool             `json:"recovery_running"`
+	RecoveryBlockers     []string         `json:"recovery_blockers"`
 }
 
 func (observer CommandObserver) Validate() error {
@@ -63,6 +65,14 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error
 	if err := decoder.Decode(&decoded); err != nil {
 		return Observation{}, fmt.Errorf("decode scheduler observation: %w", err)
 	}
+	if err := decoder.Decode(new(any)); err != io.EOF {
+		return Observation{}, fmt.Errorf("scheduler observation must contain exactly one JSON value")
+	}
+	for _, blocker := range decoded.RecoveryBlockers {
+		if strings.TrimSpace(blocker) == "" || len(blocker) > 128 || strings.ContainsAny(blocker, "\r\n\x00,") {
+			return Observation{}, fmt.Errorf("scheduler observation contains an invalid recovery blocker")
+		}
+	}
 	if decoded.ObservedAt.IsZero() || decoded.ActiveIntents < 0 || decoded.ManagerUptimeSeconds < 0 {
 		return Observation{}, fmt.Errorf("scheduler observation contains invalid values")
 	}
@@ -85,6 +95,7 @@ func (observer CommandObserver) Observe(ctx context.Context) (Observation, error
 		ObservedAt: decoded.ObservedAt, ActiveIntents: decoded.ActiveIntents,
 		PendingCreates: decoded.PendingCreates, OverdueRetries: decoded.OverdueRetries, StaleAssigned: decoded.StaleAssigned,
 		CapacityBackpressure: decoded.CapacityBackpressure,
+		RecoveryBlockers:     decoded.RecoveryBlockers,
 		ManagerUptime:        time.Duration(decoded.ManagerUptimeSeconds) * time.Second,
 		LastRecoveryAt:       decoded.LastRecoveryAt, RecoveryRunning: decoded.RecoveryRunning,
 	}, nil

internal/schedulerrecovery/progress_contract.go

@@ -0,0 +1,40 @@
+package schedulerrecovery
+
+import (
+	"fmt"
+	"strings"
+)
+
+// validateProgress requires a complete, disjoint partition of the original
+// identities. Empty output, omitted identities, duplicates and unrelated jobs
+// are not proof of recovery. Adapters must separately prove each transition.
+func validateProgress(expected, progressed, remaining []string) error {
+	if len(expected) == 0 {
+		return fmt.Errorf("recovery progress requires expected identities")
+	}
+	want := make(map[string]bool, len(expected))
+	for _, id := range expected {
+		if strings.TrimSpace(id) == "" || strings.ContainsAny(id, ",\x00\r\n") {
+			return fmt.Errorf("invalid recovery identity")
+		}
+		if _, duplicate := want[id]; duplicate {
+			return fmt.Errorf("duplicate expected recovery identity")
+		}
+		want[id] = false
+	}
+	for _, group := range [][]string{progressed, remaining} {
+		for _, id := range group {
+			seen, exists := want[id]
+			if !exists || seen {
+				return fmt.Errorf("progress contains an unexpected or repeated identity")
+			}
+			want[id] = true
+		}
+	}
+	for _, seen := range want {
+		if !seen {
+			return fmt.Errorf("progress omits an expected recovery identity")
+		}
+	}
+	return nil
+}