Merge pull request 'Add selectable canary-first agent updates' (#48) from fix-issue-43-agent-updates-workflow into main

cmd/agent/main.go

@@ -296,6 +296,9 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S
 		}
 		go d.handleTreeList(ctx, env.ID, p, tx)
 
+	case api.MsgSnapshotsRefresh:
+		go d.refreshSnapshots(ctx, tx)
+
 	case api.MsgScheduleSet:
 		var p api.ScheduleSetPayload
 		if err := env.UnmarshalPayload(&p); err != nil {
@@ -405,6 +408,22 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S
 	return nil
 }
 
+func (d *dispatcher) refreshSnapshots(ctx context.Context, tx wsclient.Sender) {
+	creds, err := d.secrets.Load()
+	if err != nil || creds.Empty() {
+		slog.Warn("ws agent: snapshots.refresh unavailable", "err", err)
+		return
+	}
+	r := runner.New(runner.Config{
+		ResticBin: d.resticBin, ResticVersion: d.resticVer,
+		RepoURL: creds.URL, RepoUsername: creds.Username, RepoPassword: creds.Password,
+		SupportsRestoreNoOwnership: d.resticSupportsNoOwnership,
+	}, tx, time.Second)
+	if err := r.RefreshSnapshots(ctx); err != nil {
+		slog.Warn("ws agent: snapshots.refresh failed", "err", err)
+	}
+}
+
 // handleTreeList runs `restic ls --json <snapshot> <path>` and ships
 // the matching tree.list.result envelope back, correlated by the
 // request envelope's ID. Errors (missing creds, restic failure)

internal/agent/runner/runner.go

@@ -77,6 +77,12 @@ func (r *Runner) resticEnv() restic.Env {
 	}
 }
 
+// RefreshSnapshots reconciles the server's cached projection without running
+// a mutating repository job.
+func (r *Runner) RefreshSnapshots(ctx context.Context) error {
+	return r.reportSnapshots(ctx, r.resticEnv())
+}
+
 // sendStarted ships a job.started envelope.
 func (r *Runner) sendStarted(jobID string, kind api.JobKind, startedAt time.Time) {
 	env, _ := api.Marshal(api.MsgJobStarted, jobID, api.JobStartedPayload{
@@ -226,14 +232,9 @@ func (r *Runner) RunBackup(ctx context.Context, jobID string, paths, excludes, t
 		}
 	}
 
-	r.sendFinished(ctx, jobID, finishedAt, err, statsBlob)
-
 	// On a successful backup, refresh the server's snapshot projection.
-	// We do this *after* job.finished so the UI sees the job land first;
-	// the snapshot list is a follow-up that the host detail page polls
-	// or the dashboard sees on its next refresh. A failure here is
-	// logged but doesn't fail the job — the next successful backup will
-	// catch the projection up.
+	// Do this before job.finished so a failure in terminal reporting cannot
+	// prevent the independently useful projection refresh.
 	if err == nil {
 		if rerr := r.reportSnapshots(ctx, env); rerr != nil {
 			slog.Warn("runner: snapshots.report failed", "job_id", jobID, "err", rerr)
@@ -242,6 +243,7 @@ func (r *Runner) RunBackup(ctx context.Context, jobID string, paths, excludes, t
 			slog.Warn("runner: stats.report after backup failed", "job_id", jobID, "err", rerr)
 		}
 	}
+	r.sendFinished(ctx, jobID, finishedAt, err, statsBlob)
 
 	if err != nil {
 		return fmt.Errorf("runner backup: %w", err)
@@ -282,8 +284,6 @@ func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.Fo
 	var seq atomic.Int64
 	err := env.RunForget(ctx, groups, dryRun, r.streamHandler(jobID, &seq))
 	finishedAt := time.Now().UTC()
-	r.sendFinished(ctx, jobID, finishedAt, err, nil)
-
 	// Refresh the server's snapshot projection — forget rewrites the
 	// index so the host's snapshot list almost certainly shrunk.
 	if err == nil {
@@ -292,6 +292,7 @@ func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.Fo
 				"job_id", jobID, "err", rerr)
 		}
 	}
+	r.sendFinished(ctx, jobID, finishedAt, err, nil)
 
 	if err != nil {
 		return fmt.Errorf("runner forget: %w", err)
@@ -318,6 +319,9 @@ func (r *Runner) RunPrune(ctx context.Context, jobID string) error {
 		if rerr := r.reportStats(ctx, env, api.RepoStatsPayload{LastPruneAt: &pruneAt}); rerr != nil {
 			slog.Warn("runner: stats.report after prune failed", "job_id", jobID, "err", rerr)
 		}
+		if rerr := r.reportSnapshots(ctx, env); rerr != nil {
+			slog.Warn("runner: snapshots.report after prune failed", "job_id", jobID, "err", rerr)
+		}
 	}
 
 	r.sendFinished(ctx, jobID, finishedAt, err, nil)

internal/agent/runner/runner_test.go

@@ -116,7 +116,8 @@ func envelopeOrder(envs []api.Envelope) []api.MessageType {
 // TestRunPruneShipsExpectedEnvelopes drives RunPrune with a fake
 // binary that prints "prune" on stdout (for the log.stream envelope)
 // and emits valid stats JSON so reportStats can populate size fields.
-// Expected sequence: job.started → log.stream → repo.stats → job.finished.
+// Expected sequence: job.started → log.stream → repo.stats → snapshots.report
+// → job.finished.
 func TestRunPruneShipsExpectedEnvelopes(t *testing.T) {
 	t.Parallel()
 
@@ -126,6 +127,7 @@ func TestRunPruneShipsExpectedEnvelopes(t *testing.T) {
 case "$1" in
   prune)  echo "prune" ;;
   stats)  echo '`+statsJSON+`' ;;
+	  snapshots) echo "[]" ;;
   *)      echo "unknown: $*" ;;
 esac
 `)
@@ -138,7 +140,7 @@ esac
 
 	order := envelopeOrder(tx.envs)
 	// Confirm landmark envelope types appear in the required order.
-	wantTypes := []api.MessageType{api.MsgJobStarted, api.MsgLogStream, api.MsgRepoStats, api.MsgJobFinished}
+	wantTypes := []api.MessageType{api.MsgJobStarted, api.MsgLogStream, api.MsgRepoStats, api.MsgSnapshotsRpt, api.MsgJobFinished}
 	positions := map[api.MessageType]int{}
 	for i, mt := range order {
 		if _, seen := positions[mt]; !seen {
@@ -379,6 +381,15 @@ func TestRunInitShipsStartedAndFinished(t *testing.T) {
 	_ = firstEnvOfType(t, tx.envs, api.MsgJobFinished)
 }
 
+func firstIndexOfType(envs []api.Envelope, typ api.MessageType) int {
+	for i, env := range envs {
+		if env.Type == typ {
+			return i
+		}
+	}
+	return -1
+}
+
 // TestRunForgetShipsStartedAndFinished confirms the refactored
 // RunForget still produces job.started and job.finished envelopes.
 func TestRunForgetShipsStartedAndFinished(t *testing.T) {
@@ -402,5 +413,9 @@ esac
 		t.Fatalf("RunForget: %v", err)
 	}
 	_ = firstEnvOfType(t, tx.envs, api.MsgJobStarted)
-	_ = firstEnvOfType(t, tx.envs, api.MsgJobFinished)
+	finished := firstIndexOfType(tx.envs, api.MsgJobFinished)
+	refreshed := firstIndexOfType(tx.envs, api.MsgSnapshotsRpt)
+	if refreshed < 0 || finished < 0 || refreshed >= finished {
+		t.Fatalf("snapshot refresh must precede terminal report: %v", envelopeOrder(tx.envs))
+	}
 }

internal/agent/wsclient/client.go

@@ -108,6 +108,7 @@ func connectOnce(ctx context.Context, cfg Config, handle Handler) error {
 	if err != nil {
 		return fmt.Errorf("dial: %w", err)
 	}
+	conn.SetReadLimit(api.MaxWebSocketMessageBytes)
 	// On a successful upgrade coder/websocket transfers ownership of the
 	// response stream to conn and deliberately sets res.Body to nil. Closing
 	// the connection below releases that stream.

internal/agent/wsclient/client_test.go

@@ -2,12 +2,16 @@ package wsclient
 
 import (
 	"context"
+	"encoding/json"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/coder/websocket"
+
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
 )
 
 func TestConnectOnceCleanDisconnectDoesNotPanic(t *testing.T) {
@@ -44,3 +48,56 @@ func TestConnectOnceCleanDisconnectDoesNotPanic(t *testing.T) {
 		t.Fatalf("server websocket: %v", err)
 	}
 }
+
+func TestConnectOnceAcceptsMessageLargerThanDefaultReadLimit(t *testing.T) {
+	received := make(chan struct{}, 1)
+	serverErr := make(chan error, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := websocket.Accept(w, r, nil)
+		if err != nil {
+			serverErr <- err
+			return
+		}
+		defer conn.CloseNow() //nolint:errcheck
+		if _, _, err := conn.Read(r.Context()); err != nil {
+			serverErr <- err
+			return
+		}
+		env := api.Envelope{
+			Type:    api.MsgConfigUpdate,
+			Payload: json.RawMessage(`{"padding":"` + strings.Repeat("x", 40*1024) + `"}`),
+		}
+		raw, _ := json.Marshal(env)
+		serverErr <- conn.Write(r.Context(), websocket.MessageText, raw)
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() {
+		done <- connectOnce(ctx, Config{
+			ServerURL:       srv.URL,
+			AgentToken:      "test-token",
+			HeartbeatPeriod: time.Hour,
+		}, func(_ context.Context, env api.Envelope, _ Sender) error {
+			if env.Type == api.MsgConfigUpdate {
+				received <- struct{}{}
+				cancel()
+			}
+			return nil
+		})
+	}()
+
+	select {
+	case <-received:
+	case <-time.After(5 * time.Second):
+		t.Fatal("agent did not receive oversized server message")
+	}
+	if err := <-serverErr; err != nil {
+		t.Fatalf("server websocket: %v", err)
+	}
+	if err := <-done; err == nil {
+		t.Fatal("connectOnce returned nil")
+	}
+}

internal/alert/engine.go

@@ -28,6 +28,11 @@ import (
 // are evaluated — always-on hosts' stale_schedule stays a no-op.
 const staleBackupThreshold = 7 * 24 * time.Hour
 
+const (
+	defaultStuckJobThreshold = 6 * time.Hour
+	longStuckJobThreshold    = 24 * time.Hour
+)
+
 // JobFinishedEvent carries everything the engine needs to evaluate
 // the failed-X rules. Pushed via Engine.NotifyJobFinished from the
 // MarkJobFinished site.
@@ -53,6 +58,7 @@ type Engine struct {
 	// we raise. Configurable for tests; default 15m.
 	agentOfflineFloor time.Duration
 	tickPeriod        time.Duration
+	stuckThresholds   map[string]time.Duration
 
 	closeOnce sync.Once
 	done      chan struct{}
@@ -69,7 +75,12 @@ func NewEngine(st *store.Store, hub *notification.Hub) *Engine {
 		hostUp:            make(chan string, 32),
 		agentOfflineFloor: 15 * time.Minute,
 		tickPeriod:        60 * time.Second,
-		done:              make(chan struct{}),
+		stuckThresholds: map[string]time.Duration{
+			"backup":  longStuckJobThreshold,
+			"restore": longStuckJobThreshold,
+			"check":   longStuckJobThreshold,
+		},
+		done: make(chan struct{}),
 	}
 }
 
@@ -125,6 +136,10 @@ func (e *Engine) NotifyHostOnline(hostID string) {
 }
 
 func (e *Engine) handleJobFinished(ctx context.Context, ev JobFinishedEvent) {
+	// A late terminal message is authoritative and clears any stuck alert for
+	// this exact job regardless of its outcome or kind.
+	e.resolveAndNotify(ctx, ev.HostID, KindJobStuck, ev.JobID, ev.When)
+
 	// Determine which kind/severity pair this job maps to. Jobs not
 	// listed here (init, unlock, restore, diff) produce no alerts in v1.
 	var kind, severity string
@@ -210,6 +225,7 @@ func (e *Engine) tick(ctx context.Context, now time.Time) {
 	if _, err := e.store.CleanupExpiredOIDCState(ctx, now.Add(-5*time.Minute)); err != nil {
 		slog.Warn("alert: cleanup expired oidc state", "err", err)
 	}
+	e.evaluateStuckJobs(ctx, now)
 
 	hosts, err := e.store.ListHosts(ctx)
 	if err != nil {
@@ -257,6 +273,46 @@ func (e *Engine) tick(ctx context.Context, now time.Time) {
 	}
 }
 
+func (e *Engine) evaluateStuckJobs(ctx context.Context, now time.Time) {
+	running, err := e.store.ListRunningJobActivity(ctx)
+	if err != nil {
+		slog.Warn("alert: tick list running jobs", "err", err)
+		return
+	}
+	active := make(map[string]store.RunningJobActivity, len(running))
+	for _, job := range running {
+		active[job.JobID] = job
+		threshold := defaultStuckJobThreshold
+		if configured, ok := e.stuckThresholds[job.Kind]; ok {
+			threshold = configured
+		}
+		age := now.Sub(job.LastActivity)
+		if age < threshold {
+			continue
+		}
+		e.raiseAndNotify(ctx, job.HostID, KindJobStuck, job.JobID, "warning",
+			fmt.Sprintf("%s job %s is stuck: started %s, last activity %s (%s ago; threshold %s)",
+				job.Kind, job.JobID, job.StartedAt.Format(time.RFC3339),
+				job.LastActivity.Format(time.RFC3339), roundDur(age), threshold), now)
+	}
+
+	// Self-heal alerts when another server path made a job terminal without
+	// emitting JobFinishedEvent, or after a restart missed the event.
+	alerts, err := e.store.ListAlerts(ctx, store.AlertFilter{Status: "open"})
+	if err != nil {
+		return
+	}
+	acked, _ := e.store.ListAlerts(ctx, store.AlertFilter{Status: "acknowledged"})
+	for _, item := range append(alerts, acked...) {
+		if item.Kind != KindJobStuck || item.HostID == nil {
+			continue
+		}
+		if _, ok := active[item.DedupKey]; !ok {
+			e.resolveAndNotify(ctx, *item.HostID, KindJobStuck, item.DedupKey, now)
+		}
+	}
+}
+
 // roundDur returns a human-readable duration string, rounding to the
 // nearest minute. Durations under a minute are reported as "less than
 // a minute".

internal/alert/rules.go

@@ -36,6 +36,10 @@ const (
 	// KindAgentOffline is raised when a host's last_seen_at is older
 	// than the 15-minute floor and resolved when the host reconnects.
 	KindAgentOffline = "agent_offline"
+
+	// KindJobStuck is raised per job when a running job has no persisted
+	// activity beyond its kind-specific threshold. The job ID is the dedup key.
+	KindJobStuck = "job_stuck"
 )
 
 // raiseAndNotify is the standard raise pattern: store.RaiseOrTouch

internal/alert/rules_test.go

@@ -3,6 +3,7 @@ package alert
 import (
 	"context"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -69,6 +70,65 @@ func TestEngineBackupFailedRaisesThenResolves(t *testing.T) {
 	}
 }
 
+func TestEngineStuckJobRaisesDeduplicatesAndResolves(t *testing.T) {
+	t.Parallel()
+	eng, st, hostID := setupEngine(t)
+	ctx := context.Background()
+	now := time.Now().UTC().Truncate(time.Second)
+	started := now.Add(-7 * time.Hour)
+	if err := st.CreateJob(ctx, store.Job{ID: "stuck-job", HostID: hostID, Kind: "prune", ActorKind: "user", CreatedAt: started}); err != nil {
+		t.Fatalf("create job: %v", err)
+	}
+	if err := st.MarkJobStarted(ctx, "stuck-job", started); err != nil {
+		t.Fatalf("start job: %v", err)
+	}
+
+	eng.evaluateStuckJobs(ctx, now)
+	eng.evaluateStuckJobs(ctx, now.Add(time.Minute))
+	open, err := st.ListAlerts(ctx, store.AlertFilter{Status: "open", HostID: hostID})
+	if err != nil {
+		t.Fatalf("list alerts: %v", err)
+	}
+	if len(open) != 1 || open[0].Kind != KindJobStuck || open[0].DedupKey != "stuck-job" {
+		t.Fatalf("expected one deduplicated stuck alert, got %+v", open)
+	}
+	if !strings.Contains(open[0].Message, started.Format(time.RFC3339)) {
+		t.Errorf("alert does not identify start/activity time: %q", open[0].Message)
+	}
+
+	if err := st.MarkJobFinished(ctx, "stuck-job", "succeeded", 0, nil, "", now); err != nil {
+		t.Fatalf("finish job: %v", err)
+	}
+	eng.evaluateStuckJobs(ctx, now.Add(2*time.Minute))
+	open, _ = st.ListAlerts(ctx, store.AlertFilter{Status: "open", HostID: hostID})
+	if len(open) != 0 {
+		t.Fatalf("expected terminal job alert resolved, got %+v", open)
+	}
+}
+
+func TestEngineStuckJobUsesLatestLogActivity(t *testing.T) {
+	t.Parallel()
+	eng, st, hostID := setupEngine(t)
+	ctx := context.Background()
+	now := time.Now().UTC().Truncate(time.Second)
+	started := now.Add(-7 * time.Hour)
+	if err := st.CreateJob(ctx, store.Job{ID: "active-job", HostID: hostID, Kind: "forget", ActorKind: "user", CreatedAt: started}); err != nil {
+		t.Fatalf("create job: %v", err)
+	}
+	if err := st.MarkJobStarted(ctx, "active-job", started); err != nil {
+		t.Fatalf("start job: %v", err)
+	}
+	if err := st.AppendJobLog(ctx, "active-job", 1, now.Add(-time.Hour), "stdout", "active"); err != nil {
+		t.Fatalf("append log: %v", err)
+	}
+
+	eng.evaluateStuckJobs(ctx, now)
+	open, _ := st.ListAlerts(ctx, store.AlertFilter{Status: "open", HostID: hostID})
+	if len(open) != 0 {
+		t.Fatalf("recent activity should suppress stuck alert, got %+v", open)
+	}
+}
+
 func TestEngineCheckFailedSeverityCritical(t *testing.T) {
 	t.Parallel()
 	eng, st, hostID := setupEngine(t)

internal/api/wire.go

@@ -10,6 +10,13 @@ import (
 // (not iota ints) makes traffic readable in logs and packet captures.
 type MessageType string
 
+// MaxWebSocketMessageBytes is the protocol-wide upper bound for one agent ↔
+// server envelope. Snapshot projections and restic JSON log events can exceed
+// coder/websocket's 32 KiB default on ordinary repositories, so both peers set
+// this limit explicitly. It remains bounded to protect either process from an
+// untrusted or malfunctioning peer allocating without limit.
+const MaxWebSocketMessageBytes int64 = 8 << 20
+
 // Agent → server message types.
 const (
 	MsgHello          MessageType = "hello"
@@ -29,12 +36,13 @@ const (
 
 // Server → agent message types.
 const (
-	MsgCommandRun    MessageType = "command.run"
-	MsgCommandCancel MessageType = "command.cancel"
-	MsgScheduleSet   MessageType = "schedule.set"
-	MsgConfigUpdate  MessageType = "config.update"
-	MsgCommandUpdate MessageType = "command.update"
-	MsgTreeList      MessageType = "tree.list" // sync RPC: list a snapshot's children
+	MsgCommandRun       MessageType = "command.run"
+	MsgCommandCancel    MessageType = "command.cancel"
+	MsgScheduleSet      MessageType = "schedule.set"
+	MsgConfigUpdate     MessageType = "config.update"
+	MsgCommandUpdate    MessageType = "command.update"
+	MsgTreeList         MessageType = "tree.list" // sync RPC: list a snapshot's children
+	MsgSnapshotsRefresh MessageType = "snapshots.refresh"
 )
 
 // Envelope is the framing for every WS message in either direction.

internal/server/fleetupdate/worker.go

@@ -71,6 +71,16 @@ func NewWorker(st *store.Store, hub Hub, disp Dispatcher, alerts AlertRaiser) *W
 // worker goroutine. Returns the new fleet_update_id on success.
 // store.ErrFleetUpdateRunning bubbles up unchanged.
 func (w *Worker) Start(ctx context.Context, userID, targetVersion string, hostIDs []string) (string, error) {
+	return w.start(ctx, userID, targetVersion, hostIDs, false)
+}
+
+// StartCanary starts the same sequential rollout but pauses after the first
+// verified host so an operator can inspect it before starting the remainder.
+func (w *Worker) StartCanary(ctx context.Context, userID, targetVersion string, hostIDs []string) (string, error) {
+	return w.start(ctx, userID, targetVersion, hostIDs, true)
+}
+
+func (w *Worker) start(ctx context.Context, userID, targetVersion string, hostIDs []string, pauseAfterFirst bool) (string, error) {
 	if userID == "" || targetVersion == "" {
 		return "", errors.New("fleetupdate: userID and targetVersion required")
 	}
@@ -85,6 +95,7 @@ func (w *Worker) Start(ctx context.Context, userID, targetVersion string, hostID
 		StartedByUserID: userID,
 		TargetVersion:   targetVersion,
 		Status:          "running",
+		PauseAfterFirst: pauseAfterFirst,
 	}, hostIDs); err != nil {
 		return "", err
 	}
@@ -137,6 +148,13 @@ func (w *Worker) run(ctx context.Context, fuID, userID, targetVersion string) {
 
 		next := pending[0]
 		w.processHost(ctx, fuID, userID, next)
+		if fu.PauseAfterFirst && next.Position == 0 {
+			updated, _, gerr := w.store.GetFleetUpdate(ctx, fuID)
+			if gerr == nil && updated.Status == "running" {
+				_ = w.store.HaltFleetUpdate(ctx, fuID, "canary succeeded; review the host, then resume remaining agents", time.Now().UTC())
+			}
+			return
+		}
 	}
 }
 
@@ -195,7 +213,19 @@ func (w *Worker) processHost(ctx context.Context, fuID, userID string, slot stor
 			return
 		}
 	}
-	reason := fmt.Sprintf("timeout waiting for %s to reach %s", hostID, w.targetVersion)
+	// One authoritative final read closes the poll/deadline race. AgentVersion
+	// is written directly from the reconnect hello handshake.
+	lastVersion := "unknown"
+	if h, err := w.store.GetHost(ctx, hostID); err == nil && h != nil {
+		if h.AgentVersion == w.targetVersion {
+			_ = w.store.SetFleetUpdateHostStatus(ctx, fuID, hostID, "succeeded", "", jobID)
+			return
+		}
+		if h.AgentVersion != "" {
+			lastVersion = h.AgentVersion
+		}
+	}
+	reason := fmt.Sprintf("timeout waiting for %s reconnect hello at %s (last reported %s)", hostID, w.targetVersion, lastVersion)
 	_ = w.store.SetFleetUpdateHostStatus(ctx, fuID, hostID, "failed", reason, jobID)
 	w.halt(ctx, fuID, reason)
 }

internal/server/fleetupdate/worker_test.go

@@ -160,6 +160,35 @@ func TestWorkerTwoHostsBothSucceed(t *testing.T) {
 	}
 }
 
+func TestWorkerCanaryPausesAfterFirstVerifiedHost(t *testing.T) {
+	st := openStore(t)
+	uid := mustCreateAdmin(t, st)
+	h1 := mustCreateHost(t, st, "canary", "v0")
+	h2 := mustCreateHost(t, st, "remainder", "v0")
+	hub := &fakeHub{online: map[string]bool{h1: true, h2: true}}
+	disp := &fakeDispatcher{st: st, target: "v2", delayMS: 20}
+	alerts := &recAlert{}
+	w := NewWorker(st, hub, disp, alerts)
+	w.pollPeriod = 10 * time.Millisecond
+	w.hostTimeout = time.Second
+
+	fuID, err := w.StartCanary(context.Background(), uid, "v2", []string{h1, h2})
+	if err != nil {
+		t.Fatalf("start canary: %v", err)
+	}
+	fu := waitForStatus(t, st, fuID, "halted", 2*time.Second)
+	if fu.HaltedReason != "canary succeeded; review the host, then resume remaining agents" {
+		t.Fatalf("halt reason: %q", fu.HaltedReason)
+	}
+	_, hosts, _ := st.GetFleetUpdate(context.Background(), fuID)
+	if hosts[0].Status != "succeeded" || hosts[1].Status != "pending" {
+		t.Fatalf("canary statuses: %+v", hosts)
+	}
+	if len(alerts.reasons) != 0 {
+		t.Fatalf("successful canary pause must not alert: %v", alerts.reasons)
+	}
+}
+
 func TestWorkerSecondHostTimesOutHalts(t *testing.T) {
 	st := openStore(t)
 	uid := mustCreateAdmin(t, st)

internal/server/http/fleet_update.go

@@ -32,6 +32,13 @@ import (
 type fleetUpdateStartReq struct {
 	TargetVersion string   `json:"target_version,omitempty"`
 	HostIDs       []string `json:"host_ids,omitempty"`
+	CanaryFirst   bool     `json:"canary_first,omitempty"`
+}
+
+type fleetUpdateCandidate struct {
+	Host     store.Host
+	Eligible bool
+	Reason   string
 }
 
 // fleetUpdateHostView is one row in the JSON response for GET
@@ -56,6 +63,7 @@ type fleetUpdateView struct {
 	CurrentHostID   string                `json:"current_host_id,omitempty"`
 	HaltedReason    string                `json:"halted_reason,omitempty"`
 	CompletedAt     *string               `json:"completed_at,omitempty"`
+	CanaryFirst     bool                  `json:"canary_first"`
 	Hosts           []fleetUpdateHostView `json:"hosts"`
 }
 
@@ -65,6 +73,7 @@ type fleetUpdateView struct {
 type fleetUpdatePage struct {
 	// Idle-state fields.
 	OutOfDateHosts []store.Host // online hosts whose version != target
+	Candidates     []fleetUpdateCandidate
 	TargetVersion  string
 
 	// Active-state fields. Nil when no fleet update has ever run.
@@ -100,6 +109,11 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp.
 	if target == "" {
 		target = version.Version
 	}
+	if target != version.Version {
+		writeJSONError(w, stdhttp.StatusUnprocessableEntity, "unsupported_target_version",
+			"fleet updates can only target the running server version")
+		return
+	}
 	hostIDs := body.HostIDs
 	if len(hostIDs) == 0 {
 		derived, err := s.deriveOutOfDateOnlineHostIDs(r.Context(), target)
@@ -108,6 +122,19 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp.
 			return
 		}
 		hostIDs = derived
+	} else {
+		validated, reasons, err := s.validateFleetUpdateHostIDs(r.Context(), target, hostIDs)
+		if err != nil {
+			writeJSONError(w, stdhttp.StatusInternalServerError, "internal", err.Error())
+			return
+		}
+		if len(reasons) > 0 {
+			writeJSON(w, stdhttp.StatusUnprocessableEntity, map[string]any{
+				"code": "ineligible_hosts", "message": "one or more selected hosts are ineligible", "reasons": reasons,
+			})
+			return
+		}
+		hostIDs = validated
 	}
 	if len(hostIDs) == 0 {
 		writeJSONError(w, stdhttp.StatusConflict, "no_hosts_eligible",
@@ -115,7 +142,20 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp.
 		return
 	}
 
-	fuID, err := s.deps.FleetWorker.Start(r.Context(), user.ID, target, hostIDs)
+	var fuID string
+	var err error
+	if body.CanaryFirst && len(hostIDs) > 1 {
+		worker, supported := s.deps.FleetWorker.(interface {
+			StartCanary(context.Context, string, string, []string) (string, error)
+		})
+		if !supported {
+			writeJSONError(w, stdhttp.StatusServiceUnavailable, "canary_unavailable", "")
+			return
+		}
+		fuID, err = worker.StartCanary(r.Context(), user.ID, target, hostIDs)
+	} else {
+		fuID, err = s.deps.FleetWorker.Start(r.Context(), user.ID, target, hostIDs)
+	}
 	if err != nil {
 		if errors.Is(err, store.ErrFleetUpdateRunning) {
 			writeJSONError(w, stdhttp.StatusConflict, "fleet_update_in_progress", err.Error())
@@ -129,6 +169,7 @@ func (s *Server) handleAPIFleetUpdateStart(w stdhttp.ResponseWriter, r *stdhttp.
 		"fleet_update_id": fuID,
 		"target_version":  target,
 		"host_count":      len(hostIDs),
+		"canary_first":    body.CanaryFirst,
 	})
 	_ = s.deps.Store.AppendAudit(r.Context(), store.AuditEntry{
 		ID: ulid.Make().String(), UserID: &user.ID, Actor: "user",
@@ -209,6 +250,7 @@ func (s *Server) handleAPIFleetUpdateGet(w stdhttp.ResponseWriter, r *stdhttp.Re
 		Status:          fu.Status,
 		CurrentHostID:   fu.CurrentHostID,
 		HaltedReason:    fu.HaltedReason,
+		CanaryFirst:     fu.PauseAfterFirst,
 		Hosts:           make([]fleetUpdateHostView, 0, len(hosts)),
 	}
 	if fu.CompletedAt != nil {
@@ -286,6 +328,27 @@ func (s *Server) buildFleetUpdatePage(r *stdhttp.Request) (fleetUpdatePage, erro
 	}
 	for _, h := range hosts {
 		page.HostNames[h.ID] = h.Name
+		candidate := fleetUpdateCandidate{Host: h}
+		switch {
+		case h.AgentVersion == "":
+			candidate.Reason = "version unknown"
+		case h.AgentVersion == page.TargetVersion:
+			candidate.Reason = "already current"
+		case s.deps.Hub == nil || !s.deps.Hub.Connected(h.ID):
+			candidate.Reason = "offline"
+		default:
+			updating, uerr := s.deps.Store.RunningUpdateJobForHost(r.Context(), h.ID)
+			if uerr != nil {
+				return page, uerr
+			}
+			if updating != "" {
+				candidate.Reason = "update already running"
+			} else {
+				candidate.Eligible = true
+				page.OutOfDateHosts = append(page.OutOfDateHosts, h)
+			}
+		}
+		page.Candidates = append(page.Candidates, candidate)
 	}
 
 	active, err := s.deps.Store.ActiveFleetUpdate(r.Context())
@@ -328,20 +391,54 @@ func (s *Server) buildFleetUpdatePage(r *stdhttp.Request) (fleetUpdatePage, erro
 		}
 	}
 
-	// Idle list (or "still out of date" reference even when an active
-	// roll is running — cheap to compute, harmless to attach).
-	for _, h := range hosts {
-		if h.Status != "online" {
-			continue
-		}
-		if h.AgentVersion == "" || h.AgentVersion == page.TargetVersion {
-			continue
-		}
-		page.OutOfDateHosts = append(page.OutOfDateHosts, h)
-	}
 	return page, nil
 }
 
+// validateFleetUpdateHostIDs deduplicates an explicit selection while
+// preserving review order and rejects the entire request if membership has
+// changed or any selected host is not dispatchable.
+func (s *Server) validateFleetUpdateHostIDs(ctx context.Context, target string, requested []string) ([]string, map[string]string, error) {
+	hosts, err := s.deps.Store.ListHosts(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+	byID := make(map[string]store.Host, len(hosts))
+	for _, h := range hosts {
+		byID[h.ID] = h
+	}
+	seen := map[string]bool{}
+	validated := make([]string, 0, len(requested))
+	reasons := map[string]string{}
+	for _, id := range requested {
+		if seen[id] {
+			continue
+		}
+		seen[id] = true
+		h, ok := byID[id]
+		switch {
+		case !ok:
+			reasons[id] = "host not found"
+		case h.AgentVersion == "":
+			reasons[id] = "agent version unknown"
+		case h.AgentVersion == target:
+			reasons[id] = "already at target version"
+		case s.deps.Hub == nil || !s.deps.Hub.Connected(id):
+			reasons[id] = "host offline"
+		default:
+			jobID, jerr := s.deps.Store.RunningUpdateJobForHost(ctx, id)
+			if jerr != nil {
+				return nil, nil, jerr
+			}
+			if jobID != "" {
+				reasons[id] = "update already in progress"
+			} else {
+				validated = append(validated, id)
+			}
+		}
+	}
+	return validated, reasons, nil
+}
+
 // deriveOutOfDateOnlineHostIDs returns the list of host IDs that
 // (a) are online (Hub.Connected) and (b) have an agent_version that's
 // non-empty AND != target. Used by the start endpoint when the caller

internal/server/http/fleet_update_test.go

@@ -50,6 +50,10 @@ func (f *fakeFleetWorker) Start(_ context.Context, userID, target string, hostID
 	return f.startID, nil
 }
 
+func (f *fakeFleetWorker) StartCanary(ctx context.Context, userID, target string, hostIDs []string) (string, error) {
+	return f.Start(ctx, userID, target, hostIDs)
+}
+
 func (f *fakeFleetWorker) Cancel(_ context.Context, id string) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -190,6 +194,67 @@ func TestFleetUpdateStartDerivesHostIDsWhenEmpty(t *testing.T) {
 	}
 }
 
+func TestFleetUpdateStartDeduplicatesExplicitSelection(t *testing.T) {
+	t.Parallel()
+	srv, ts, st := rawTestServer(t)
+	worker := &fakeFleetWorker{startID: ulid.Make().String()}
+	srv.deps.FleetWorker = worker
+	cookie := loginAsAdmin(t, st)
+	hostID := helloOnlineHost(t, srv, st, "duplicate-host", "v0")
+	raw, _ := json.Marshal(map[string]any{"host_ids": []string{hostID, hostID}})
+	req, _ := stdhttp.NewRequest("POST", ts.URL+"/api/fleet/update", bytes.NewReader(raw))
+	req.AddCookie(cookie)
+	req.Header.Set("Content-Type", "application/json")
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("do: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusAccepted {
+		t.Fatalf("status: %d", res.StatusCode)
+	}
+	if got := worker.startCalls[0].HostIDs; len(got) != 1 || got[0] != hostID {
+		t.Fatalf("deduplicated ids: %v", got)
+	}
+}
+
+func TestFleetUpdateStartRejectsUnknownAndIneligibleHosts(t *testing.T) {
+	t.Parallel()
+	srv, ts, st := rawTestServer(t)
+	worker := &fakeFleetWorker{startID: ulid.Make().String()}
+	srv.deps.FleetWorker = worker
+	cookie := loginAsAdmin(t, st)
+	offline := makeHost(t, st, "offline-host")
+	if err := st.MarkHostHello(context.Background(), offline, "v0", "0.17", api.CurrentProtocolVersion, time.Now().UTC()); err != nil {
+		t.Fatalf("mark offline host: %v", err)
+	}
+	raw, _ := json.Marshal(map[string]any{"host_ids": []string{offline, "does-not-exist"}})
+	req, _ := stdhttp.NewRequest("POST", ts.URL+"/api/fleet/update", bytes.NewReader(raw))
+	req.AddCookie(cookie)
+	req.Header.Set("Content-Type", "application/json")
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("do: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusUnprocessableEntity {
+		t.Fatalf("status: got %d, want 422", res.StatusCode)
+	}
+	var body struct {
+		Code    string            `json:"code"`
+		Reasons map[string]string `json:"reasons"`
+	}
+	if err := json.NewDecoder(res.Body).Decode(&body); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if body.Code != "ineligible_hosts" || len(body.Reasons) != 2 {
+		t.Fatalf("structured reasons: %+v", body)
+	}
+	if len(worker.startCalls) != 0 {
+		t.Fatal("worker must not start with invalid membership")
+	}
+}
+
 func TestFleetUpdateCancelHappyPath(t *testing.T) {
 	t.Parallel()
 	srv, ts, st := rawTestServer(t)

internal/server/http/metrics.go

@@ -8,7 +8,9 @@ import (
 	"net/netip"
 	"runtime"
 	"strings"
+	"time"
 
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/alert"
 	"gitea.dcglab.co.uk/steve/restic-manager/internal/server/config"
 	"gitea.dcglab.co.uk/steve/restic-manager/internal/server/metrics"
 	"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
@@ -173,13 +175,37 @@ func (s *Server) gatherMetricsSnapshot(ctx context.Context) (metrics.Snapshot, e
 		return metrics.Snapshot{}, err
 	}
 	bySeverity := map[string]int{"info": 0, "warning": 0, "critical": 0}
+	stuckJobs := 0
+	var oldestStuckAge time.Duration
+	now := time.Now().UTC()
+	running, err := s.deps.Store.ListRunningJobActivity(ctx)
+	if err != nil {
+		return metrics.Snapshot{}, err
+	}
+	activityByJob := make(map[string]time.Time, len(running))
+	for _, job := range running {
+		activityByJob[job.JobID] = job.LastActivity
+	}
 	for _, a := range open {
 		bySeverity[a.Severity]++
+		if a.Kind == alert.KindJobStuck {
+			stuckJobs++
+			lastActivity, ok := activityByJob[a.DedupKey]
+			if !ok {
+				continue
+			}
+			if age := now.Sub(lastActivity); age > oldestStuckAge {
+				oldestStuckAge = age
+			}
+		}
 	}
 
 	reg := s.deps.Metrics
 	if reg == nil {
 		reg = metrics.NewRegistry() // empty histogram block
 	}
-	return reg.SnapshotWith(hostRows, bySeverity, version.Version, version.Commit, runtime.Version()), nil
+	snap := reg.SnapshotWith(hostRows, bySeverity, version.Version, version.Commit, runtime.Version())
+	snap.StuckJobs = stuckJobs
+	snap.OldestStuckAge = oldestStuckAge
+	return snap, nil
 }

internal/server/http/server.go

@@ -262,6 +262,7 @@ func (s *Server) routes(r chi.Router) {
 		r.Post("/api/hosts/{id}/repo/unlock", s.handleRunRepoUnlock)
 		r.Post("/api/jobs/{id}/cancel", s.handleCancelJob)
 		r.Post("/api/hosts/{id}/snapshots/diff", s.handleSnapshotDiff)
+		r.Post("/api/hosts/{id}/snapshots/refresh", s.handleRefreshHostSnapshots)
 
 		// HTMX form variants outside /api.
 		r.Post("/hosts/{id}/snapshots/diff", s.handleSnapshotDiff)

internal/server/http/snapshots.go

@@ -5,6 +5,10 @@ import (
 	"time"
 
 	"github.com/go-chi/chi/v5"
+	"github.com/oklog/ulid/v2"
+
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
 )
 
 // snapshotView is the public JSON shape for a snapshot. Matches the
@@ -26,6 +30,7 @@ type listSnapshotsResponse struct {
 	HostID      string         `json:"host_id"`
 	Count       int            `json:"count"`
 	RefreshedAt *time.Time     `json:"refreshed_at,omitempty"`
+	Stale       bool           `json:"stale"`
 	Snapshots   []snapshotView `json:"snapshots"`
 }
 
@@ -45,7 +50,8 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re
 		return
 	}
 
-	if _, err := s.deps.Store.GetHost(r.Context(), hostID); err != nil {
+	host, err := s.deps.Store.GetHost(r.Context(), hostID)
+	if err != nil {
 		writeJSONError(w, stdhttp.StatusNotFound, "host_not_found", "")
 		return
 	}
@@ -61,10 +67,13 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re
 		Count:     len(snaps),
 		Snapshots: make([]snapshotView, len(snaps)),
 	}
-	if len(snaps) > 0 {
-		t := snaps[0].RefreshedAt
-		out.RefreshedAt = &t
+	out.RefreshedAt = host.SnapshotRefreshedAt
+	mutationAt, err := s.deps.Store.LatestSuccessfulRepoMutation(r.Context(), hostID)
+	if err != nil {
+		writeJSONError(w, stdhttp.StatusInternalServerError, "internal", "")
+		return
 	}
+	out.Stale = mutationAt != nil && (out.RefreshedAt == nil || out.RefreshedAt.Before(*mutationAt))
 	for i, sn := range snaps {
 		out.Snapshots[i] = snapshotView{
 			ID:        sn.ID,
@@ -80,3 +89,31 @@ func (s *Server) handleListHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Re
 
 	writeJSON(w, stdhttp.StatusOK, out)
 }
+
+func (s *Server) handleRefreshHostSnapshots(w stdhttp.ResponseWriter, r *stdhttp.Request) {
+	user, ok := s.requireUser(r)
+	if !ok {
+		writeJSONError(w, stdhttp.StatusUnauthorized, "unauthorised", "")
+		return
+	}
+	hostID := chi.URLParam(r, "id")
+	if _, err := s.deps.Store.GetHost(r.Context(), hostID); err != nil {
+		writeJSONError(w, stdhttp.StatusNotFound, "host_not_found", "")
+		return
+	}
+	if s.deps.Hub == nil || !s.deps.Hub.Connected(hostID) {
+		writeJSONError(w, stdhttp.StatusConflict, "host_offline", "agent is not currently connected")
+		return
+	}
+	env, _ := api.Marshal(api.MsgSnapshotsRefresh, ulid.Make().String(), nil)
+	if err := s.deps.Hub.Send(r.Context(), hostID, env); err != nil {
+		writeJSONError(w, stdhttp.StatusConflict, "host_offline", err.Error())
+		return
+	}
+	now := time.Now().UTC()
+	_ = s.deps.Store.AppendAudit(r.Context(), store.AuditEntry{
+		ID: ulid.Make().String(), UserID: &user.ID, Actor: "user",
+		Action: "host.snapshots_refresh", TargetKind: ptr("host"), TargetID: &hostID, TS: now,
+	})
+	writeJSON(w, stdhttp.StatusAccepted, map[string]string{"status": "refresh_requested"})
+}

internal/server/http/snapshots_test.go

@@ -0,0 +1,83 @@
+package http
+
+import (
+	"context"
+	"encoding/json"
+	stdhttp "net/http"
+	"testing"
+	"time"
+
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
+)
+
+func TestSnapshotsFreshnessAndExplicitRefresh(t *testing.T) {
+	t.Parallel()
+	srv, ts, st := rawTestServerWithUI(t)
+	hostID, token := enrolHostForUI(t, srv, st, "snapshot-refresh-host")
+	c := agentDial(t, srv, ts, hostID, token)
+	sendHello(t, c, "snapshot-refresh-host")
+	_ = drainUntil(t, c, api.MsgScheduleSet)
+	cookie := loginAsAdmin(t, st)
+
+	mutationAt := time.Now().UTC().Add(-time.Minute).Truncate(time.Millisecond)
+	if err := st.CreateJob(context.Background(), store.Job{
+		ID: "mutation-job", HostID: hostID, Kind: "forget", ActorKind: "user", CreatedAt: mutationAt.Add(-time.Minute),
+	}); err != nil {
+		t.Fatalf("create mutation: %v", err)
+	}
+	if err := st.MarkJobFinished(context.Background(), "mutation-job", "succeeded", 0, nil, "", mutationAt); err != nil {
+		t.Fatalf("finish mutation: %v", err)
+	}
+
+	get := func() listSnapshotsResponse {
+		req, _ := stdhttp.NewRequest(stdhttp.MethodGet, ts.URL+"/api/hosts/"+hostID+"/snapshots", nil)
+		req.AddCookie(cookie)
+		res, err := stdhttp.DefaultClient.Do(req)
+		if err != nil {
+			t.Fatalf("get snapshots: %v", err)
+		}
+		defer res.Body.Close()
+		var body listSnapshotsResponse
+		if err := json.NewDecoder(res.Body).Decode(&body); err != nil {
+			t.Fatalf("decode snapshots: %v", err)
+		}
+		return body
+	}
+	if body := get(); !body.Stale || body.RefreshedAt != nil {
+		t.Fatalf("unrefreshed projection should be stale: %+v", body)
+	}
+
+	refreshedAt := mutationAt.Add(time.Second)
+	if err := st.ReplaceHostSnapshots(context.Background(), hostID, nil, refreshedAt); err != nil {
+		t.Fatalf("replace empty: %v", err)
+	}
+	if body := get(); body.Stale || body.RefreshedAt == nil || !body.RefreshedAt.Equal(refreshedAt) {
+		t.Fatalf("fresh empty projection reported incorrectly: %+v", body)
+	}
+
+	req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/snapshots/refresh", nil)
+	req.AddCookie(cookie)
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("request refresh: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusAccepted {
+		t.Fatalf("refresh status = %d, want 202", res.StatusCode)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	_, raw, err := c.Read(ctx)
+	if err != nil {
+		t.Fatalf("read refresh command: %v", err)
+	}
+	var env api.Envelope
+	if err := json.Unmarshal(raw, &env); err != nil {
+		t.Fatalf("decode envelope: %v", err)
+	}
+	if env.Type != api.MsgSnapshotsRefresh {
+		t.Fatalf("message type = %q, want %q", env.Type, api.MsgSnapshotsRefresh)
+	}
+}

internal/server/metrics/metrics.go

@@ -155,6 +155,8 @@ type Snapshot struct {
 	BuildCommit      string
 	GoVersion        string
 	JobDurationRows  []HistogramRow
+	StuckJobs        int
+	OldestStuckAge   time.Duration
 }
 
 // SnapshotWith builds a Snapshot from raw inputs and the registry's
@@ -205,6 +207,13 @@ func Render(w io.Writer, s Snapshot) error {
 	fmt.Fprintf(&b, "rm_build_info{version=%q,commit=%q,go_version=%q} 1\n",
 		s.BuildVersion, s.BuildCommit, s.GoVersion)
 
+	b.WriteString("# HELP rm_stuck_jobs Number of open job_stuck alerts.\n")
+	b.WriteString("# TYPE rm_stuck_jobs gauge\n")
+	fmt.Fprintf(&b, "rm_stuck_jobs %d\n", s.StuckJobs)
+	b.WriteString("# HELP rm_oldest_stuck_job_age_seconds Time since last activity for the oldest open stuck job.\n")
+	b.WriteString("# TYPE rm_oldest_stuck_job_age_seconds gauge\n")
+	fmt.Fprintf(&b, "rm_oldest_stuck_job_age_seconds %.0f\n", s.OldestStuckAge.Seconds())
+
 	// --- Per-host gauges -------------------------------------------------
 	// Stable order: by host id.
 	hosts := append([]HostRow(nil), s.Hosts...)

internal/server/metrics/metrics_test.go

@@ -114,6 +114,8 @@ func TestRenderGolden(t *testing.T) {
 	snap := r.SnapshotWith(hosts,
 		map[string]int{"info": 0, "warning": 1, "critical": 0},
 		"v1.2.3", "deadbeef", "go1.25.0")
+	snap.StuckJobs = 2
+	snap.OldestStuckAge = 90 * time.Minute
 
 	var buf bytes.Buffer
 	if err := Render(&buf, snap); err != nil {
@@ -129,6 +131,8 @@ func TestRenderGolden(t *testing.T) {
 		`rm_active_alerts{severity="info"} 0`,
 		`rm_active_alerts{severity="critical"} 0`,
 		`rm_build_info{version="v1.2.3",commit="deadbeef",go_version="go1.25.0"} 1`,
+		`rm_stuck_jobs 2`,
+		`rm_oldest_stuck_job_age_seconds 5400`,
 		`rm_host_agent_online{host_id="01H0001",host="alpha"} 1`,
 		`rm_host_agent_online{host_id="01H0002",host="bravo"} 0`,
 		`rm_host_last_backup_timestamp_seconds{host_id="01H0001",host="alpha"} 1700000000`,

internal/server/ws/handler.go

@@ -77,6 +77,7 @@ func AgentHandler(deps HandlerDeps) stdhttp.Handler {
 			slog.Warn("ws accept failed", "err", err, "host_id", host.ID)
 			return
 		}
+		conn.SetReadLimit(api.MaxWebSocketMessageBytes)
 
 		c := NewConn(host.ID, conn)
 		// Keep agents alive across NAT boxes; coder/websocket

internal/server/ws/hub_test.go

@@ -123,6 +123,71 @@ func TestWSHelloAndHeartbeat(t *testing.T) {
 	t.Error("heartbeat did not update last_seen_at")
 }
 
+func TestWSAcceptsSnapshotReportLargerThanDefaultReadLimit(t *testing.T) {
+	t.Parallel()
+	url, token, hostID, st, hub := setupTestHub(t)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	c, resp, err := websocket.Dial(ctx, url, &websocket.DialOptions{
+		HTTPHeader: stdhttp.Header{"Authorization": []string{"Bearer " + token}},
+	})
+	if resp != nil && resp.Body != nil {
+		defer resp.Body.Close() //nolint:errcheck
+	}
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer c.CloseNow() //nolint:errcheck
+
+	hello, _ := api.Marshal(api.MsgHello, "", api.HelloPayload{
+		ProtocolVersion: api.CurrentProtocolVersion,
+		AgentVersion:    "0.1.0",
+		ResticVersion:   "0.17.1",
+		Hostname:        "h1",
+		OS:              api.OSLinux,
+		Arch:            api.ArchAmd64,
+	})
+	helloRaw, _ := json.Marshal(hello)
+	if err := c.Write(ctx, websocket.MessageText, helloRaw); err != nil {
+		t.Fatalf("write hello: %v", err)
+	}
+	deadline := time.Now().Add(time.Second)
+	for !hub.Connected(hostID) && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	report, _ := api.Marshal(api.MsgSnapshotsRpt, "", api.SnapshotsReportPayload{
+		Snapshots: []api.Snapshot{{
+			ID:       strings.Repeat("a", 64),
+			ShortID:  "aaaaaaaa",
+			Time:     time.Now().UTC(),
+			Hostname: "h1",
+			Paths:    []string{"/" + strings.Repeat("long-path/", 5000)},
+		}},
+	})
+	reportRaw, _ := json.Marshal(report)
+	if len(reportRaw) <= 32*1024 {
+		t.Fatalf("test payload is only %d bytes; must exceed old limit", len(reportRaw))
+	}
+	if int64(len(reportRaw)) >= api.MaxWebSocketMessageBytes {
+		t.Fatalf("test payload %d exceeds protocol limit", len(reportRaw))
+	}
+	if err := c.Write(ctx, websocket.MessageText, reportRaw); err != nil {
+		t.Fatalf("write snapshots.report: %v", err)
+	}
+
+	deadline = time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		host, err := st.GetHost(context.Background(), hostID)
+		if err == nil && host.SnapshotCount == 1 {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("oversized snapshots.report was not projected")
+}
+
 func TestWSRejectsOldProtocol(t *testing.T) {
 	t.Parallel()
 	url, token, _, _, _ := setupTestHub(t)

internal/store/fleet_updates.go

@@ -42,9 +42,9 @@ func (st *Store) CreateFleetUpdate(ctx context.Context, fu FleetUpdate, hostIDs
 	}
 
 	if _, err := tx.ExecContext(ctx,
-		`INSERT INTO fleet_updates (id, started_at, started_by_user_id, target_version, status)
-		 VALUES (?, ?, ?, ?, ?)`,
-		fu.ID, fu.StartedAt.UTC().Format(time.RFC3339Nano), fu.StartedByUserID, fu.TargetVersion, fu.Status,
+		`INSERT INTO fleet_updates (id, started_at, started_by_user_id, target_version, status, pause_after_first)
+		 VALUES (?, ?, ?, ?, ?, ?)`,
+		fu.ID, fu.StartedAt.UTC().Format(time.RFC3339Nano), fu.StartedByUserID, fu.TargetVersion, fu.Status, fu.PauseAfterFirst,
 	); err != nil {
 		return fmt.Errorf("store: insert fleet_updates: %w", err)
 	}
@@ -67,12 +67,13 @@ func (st *Store) ActiveFleetUpdate(ctx context.Context) (*FleetUpdate, error) {
 	var current sql.NullString
 	var halted sql.NullString
 	var completedAt sql.NullString
+	var pauseAfterFirst int
 	err := st.db.QueryRowContext(ctx,
 		`SELECT id, started_at, started_by_user_id, target_version, status,
-		   current_host_id, halted_reason, completed_at
+		   current_host_id, halted_reason, completed_at, pause_after_first
 		 FROM fleet_updates WHERE status = 'running' LIMIT 1`).
 		Scan(&fu.ID, &startedAt, &fu.StartedByUserID, &fu.TargetVersion, &fu.Status,
-			&current, &halted, &completedAt)
+			&current, &halted, &completedAt, &pauseAfterFirst)
 	if errors.Is(err, sql.ErrNoRows) {
 		return nil, nil
 	}
@@ -82,6 +83,7 @@ func (st *Store) ActiveFleetUpdate(ctx context.Context) (*FleetUpdate, error) {
 	fu.StartedAt, _ = time.Parse(time.RFC3339Nano, startedAt)
 	fu.CurrentHostID = current.String
 	fu.HaltedReason = halted.String
+	fu.PauseAfterFirst = pauseAfterFirst != 0
 	if completedAt.Valid {
 		t, _ := time.Parse(time.RFC3339Nano, completedAt.String)
 		fu.CompletedAt = &t
@@ -97,12 +99,13 @@ func (st *Store) GetFleetUpdate(ctx context.Context, id string) (*FleetUpdate, [
 	var current sql.NullString
 	var halted sql.NullString
 	var completedAt sql.NullString
+	var pauseAfterFirst int
 	err := st.db.QueryRowContext(ctx,
 		`SELECT id, started_at, started_by_user_id, target_version, status,
-		   current_host_id, halted_reason, completed_at
+		   current_host_id, halted_reason, completed_at, pause_after_first
 		 FROM fleet_updates WHERE id = ?`, id).
 		Scan(&fu.ID, &startedAt, &fu.StartedByUserID, &fu.TargetVersion, &fu.Status,
-			&current, &halted, &completedAt)
+			&current, &halted, &completedAt, &pauseAfterFirst)
 	if errors.Is(err, sql.ErrNoRows) {
 		return nil, nil, ErrNotFound
 	}
@@ -112,6 +115,7 @@ func (st *Store) GetFleetUpdate(ctx context.Context, id string) (*FleetUpdate, [
 	fu.StartedAt, _ = time.Parse(time.RFC3339Nano, startedAt)
 	fu.CurrentHostID = current.String
 	fu.HaltedReason = halted.String
+	fu.PauseAfterFirst = pauseAfterFirst != 0
 	if completedAt.Valid {
 		t, _ := time.Parse(time.RFC3339Nano, completedAt.String)
 		fu.CompletedAt = &t

internal/store/hosts.go

@@ -44,7 +44,7 @@ func (s *Store) LookupHostByAgentToken(ctx context.Context, tokenHash string) (*
 			repo_size_bytes, snapshot_count, open_alert_count,
 			applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps,
 			pre_hook_default, post_hook_default,
-			repo_status, repo_status_error, always_on
+			repo_status, repo_status_error, always_on, snapshot_refreshed_at
 		 FROM hosts WHERE agent_token_hash = ?`,
 		tokenHash)
 	return scanHost(row)
@@ -59,7 +59,7 @@ func (s *Store) GetHost(ctx context.Context, id string) (*Host, error) {
 			repo_size_bytes, snapshot_count, open_alert_count,
 			applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps,
 			pre_hook_default, post_hook_default,
-			repo_status, repo_status_error, always_on
+			repo_status, repo_status_error, always_on, snapshot_refreshed_at
 		 FROM hosts WHERE id = ?`, id)
 	return scanHost(row)
 }
@@ -227,7 +227,7 @@ func (s *Store) ListHosts(ctx context.Context) ([]Host, error) {
 			repo_size_bytes, snapshot_count, open_alert_count,
 			applied_schedule_version, bandwidth_up_kbps, bandwidth_down_kbps,
 			pre_hook_default, post_hook_default,
-			repo_status, repo_status_error, always_on
+			repo_status, repo_status_error, always_on, snapshot_refreshed_at
 		 FROM hosts ORDER BY name`)
 	if err != nil {
 		return nil, fmt.Errorf("store: list hosts: %w", err)
@@ -268,6 +268,7 @@ func scanHostRow(s hostScanner) (*Host, error) {
 		bwUp, bwDown                 sql.NullInt64
 		preHook, postHook            sql.NullString
 		alwaysOn                     int
+		snapshotRefreshedAt          sql.NullString
 	)
 	err := s.Scan(&h.ID, &h.Name, &h.OS, &h.Arch,
 		&h.AgentVersion, &h.ResticVersion, &h.ProtocolVersion,
@@ -276,7 +277,7 @@ func scanHostRow(s hostScanner) (*Host, error) {
 		&h.RepoSizeBytes, &h.SnapshotCount, &h.OpenAlertCount,
 		&h.AppliedScheduleVersion, &bwUp, &bwDown,
 		&preHook, &postHook,
-		&h.RepoStatus, &h.RepoStatusError, &alwaysOn)
+		&h.RepoStatus, &h.RepoStatusError, &alwaysOn, &snapshotRefreshedAt)
 	if err != nil {
 		if errors.Is(err, sql.ErrNoRows) {
 			return nil, ErrNotFound
@@ -332,6 +333,13 @@ func scanHostRow(s hostScanner) (*Host, error) {
 		h.PostHookDefault = postHook.String
 	}
 	h.AlwaysOn = alwaysOn != 0
+	if snapshotRefreshedAt.Valid {
+		t, err := time.Parse(time.RFC3339Nano, snapshotRefreshedAt.String)
+		if err != nil {
+			return nil, fmt.Errorf("store: parse snapshot_refreshed_at: %w", err)
+		}
+		h.SnapshotRefreshedAt = &t
+	}
 	return &h, nil
 }
 

internal/store/jobs.go

@@ -27,6 +27,73 @@ type Job struct {
 	CreatedAt     time.Time
 }
 
+// RunningJobActivity is the persisted activity summary used to detect jobs
+// whose terminal message was lost. LastActivity is the newer of started_at and
+// the most recent persisted log line; ephemeral progress events deliberately
+// do not extend it.
+type RunningJobActivity struct {
+	JobID        string
+	HostID       string
+	Kind         string
+	StartedAt    time.Time
+	LastActivity time.Time
+}
+
+// LatestSuccessfulRepoMutation returns the newest completion time for a job
+// that can change the repository's snapshot projection.
+func (s *Store) LatestSuccessfulRepoMutation(ctx context.Context, hostID string) (*time.Time, error) {
+	var raw sql.NullString
+	err := s.db.QueryRowContext(ctx, `
+		SELECT MAX(finished_at) FROM jobs
+		WHERE host_id = ? AND status = 'succeeded'
+		  AND kind IN ('backup', 'forget', 'prune')`, hostID).Scan(&raw)
+	if err != nil {
+		return nil, fmt.Errorf("store: latest successful repo mutation: %w", err)
+	}
+	if !raw.Valid {
+		return nil, nil
+	}
+	t, err := time.Parse(time.RFC3339Nano, raw.String)
+	if err != nil {
+		return nil, fmt.Errorf("store: parse latest repo mutation: %w", err)
+	}
+	return &t, nil
+}
+
+// ListRunningJobActivity returns every running job with its latest persisted
+// activity. Jobs without started_at are excluded because they have not actually
+// entered the running state coherently and cannot be aged safely here.
+func (s *Store) ListRunningJobActivity(ctx context.Context) ([]RunningJobActivity, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT j.id, j.host_id, j.kind, j.started_at,
+		       COALESCE(MAX(l.ts), j.started_at) AS last_activity
+		FROM jobs j
+		LEFT JOIN job_logs l ON l.job_id = j.id
+		WHERE j.status = 'running' AND j.started_at IS NOT NULL
+		GROUP BY j.id, j.host_id, j.kind, j.started_at
+		ORDER BY j.started_at`)
+	if err != nil {
+		return nil, fmt.Errorf("store: list running job activity: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var out []RunningJobActivity
+	for rows.Next() {
+		var item RunningJobActivity
+		var started, activity string
+		if err := rows.Scan(&item.JobID, &item.HostID, &item.Kind, &started, &activity); err != nil {
+			return nil, fmt.Errorf("store: scan running job activity: %w", err)
+		}
+		item.StartedAt, _ = time.Parse(time.RFC3339Nano, started)
+		item.LastActivity, _ = time.Parse(time.RFC3339Nano, activity)
+		out = append(out, item)
+	}
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("store: iterate running job activity: %w", err)
+	}
+	return out, nil
+}
+
 // CreateJob inserts a queued job. The agent will mark it running
 // when it actually starts work. ScheduledID is set when the job
 // originates from a cron fire (actor_kind="schedule"); nil for

internal/store/jobs_test.go

@@ -7,6 +7,48 @@ import (
 	"time"
 )
 
+func TestListRunningJobActivity(t *testing.T) {
+	t.Parallel()
+	s := openTestStore(t)
+	ctx := context.Background()
+	hostID := makeSchedHost(t, s)
+	started := time.Now().UTC().Add(-8 * time.Hour).Truncate(time.Second)
+
+	for _, id := range []string{"running-idle", "running-active", "finished"} {
+		if err := s.CreateJob(ctx, Job{ID: id, HostID: hostID, Kind: "backup", ActorKind: "user", CreatedAt: started}); err != nil {
+			t.Fatalf("create %s: %v", id, err)
+		}
+		if err := s.MarkJobStarted(ctx, id, started); err != nil {
+			t.Fatalf("start %s: %v", id, err)
+		}
+	}
+	activity := started.Add(7 * time.Hour)
+	if err := s.AppendJobLog(ctx, "running-active", 1, activity, "stdout", "still working"); err != nil {
+		t.Fatalf("append log: %v", err)
+	}
+	if err := s.MarkJobFinished(ctx, "finished", "succeeded", 0, nil, "", activity); err != nil {
+		t.Fatalf("finish: %v", err)
+	}
+
+	got, err := s.ListRunningJobActivity(ctx)
+	if err != nil {
+		t.Fatalf("list activity: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("got %d rows, want 2: %+v", len(got), got)
+	}
+	byID := make(map[string]RunningJobActivity, len(got))
+	for _, row := range got {
+		byID[row.JobID] = row
+	}
+	if !byID["running-idle"].LastActivity.Equal(started) {
+		t.Errorf("idle last activity = %s, want %s", byID["running-idle"].LastActivity, started)
+	}
+	if !byID["running-active"].LastActivity.Equal(activity) {
+		t.Errorf("active last activity = %s, want %s", byID["running-active"].LastActivity, activity)
+	}
+}
+
 func TestLatestJobByKind(t *testing.T) {
 	t.Parallel()
 	s := openTestStore(t)

internal/store/migrations/0025_hosts_snapshot_refreshed_at.sql

@@ -0,0 +1 @@
+ALTER TABLE hosts ADD COLUMN snapshot_refreshed_at TEXT;

internal/store/migrations/0026_fleet_update_canary.sql

@@ -0,0 +1 @@
+ALTER TABLE fleet_updates ADD COLUMN pause_after_first INTEGER NOT NULL DEFAULT 0;

internal/store/snapshots.go

@@ -69,8 +69,8 @@ func (s *Store) ReplaceHostSnapshots(ctx context.Context, hostID string, snaps [
 	}
 
 	if _, err := tx.ExecContext(ctx,
-		`UPDATE hosts SET snapshot_count = ? WHERE id = ?`,
-		len(snaps), hostID); err != nil {
+		`UPDATE hosts SET snapshot_count = ?, snapshot_refreshed_at = ? WHERE id = ?`,
+		len(snaps), when.UTC().Format(time.RFC3339Nano), hostID); err != nil {
 		return fmt.Errorf("store: update host snapshot_count: %w", err)
 	}
 

internal/store/snapshots_test.go

@@ -138,7 +138,8 @@ func TestReplaceHostSnapshotsEmpty(t *testing.T) {
 		t.Fatalf("replace 1: %v", err)
 	}
 	// Then empty — host has been wiped.
-	if err := s.ReplaceHostSnapshots(ctx, hostID, nil, time.Now().UTC()); err != nil {
+	refreshedAt := time.Now().UTC().Truncate(time.Millisecond)
+	if err := s.ReplaceHostSnapshots(ctx, hostID, nil, refreshedAt); err != nil {
 		t.Fatalf("replace empty: %v", err)
 	}
 	out, err := s.ListSnapshotsByHost(ctx, hostID)
@@ -152,4 +153,7 @@ func TestReplaceHostSnapshotsEmpty(t *testing.T) {
 	if h.SnapshotCount != 0 {
 		t.Errorf("snapshot_count should reset to 0, got %d", h.SnapshotCount)
 	}
+	if h.SnapshotRefreshedAt == nil || !h.SnapshotRefreshedAt.Equal(refreshedAt) {
+		t.Errorf("empty projection refresh time = %v, want %s", h.SnapshotRefreshedAt, refreshedAt)
+	}
 }

internal/store/types.go

@@ -77,6 +77,7 @@ type Host struct {
 	LastBackupStatus       *string
 	RepoSizeBytes          int64
 	SnapshotCount          int
+	SnapshotRefreshedAt    *time.Time
 	OpenAlertCount         int
 	AppliedScheduleVersion int64
 	// Host-wide bandwidth caps applied to every restic invocation
@@ -230,6 +231,7 @@ type FleetUpdate struct {
 	CurrentHostID   string
 	HaltedReason    string
 	CompletedAt     *time.Time
+	PauseAfterFirst bool
 }
 
 // FleetUpdateHost is one host's slot in a fleet update. Position is

web/templates/pages/dashboard.html

@@ -69,7 +69,7 @@
     {{/* ---------- Hosts-behind hero tile (P6-18) ---------- */}}
     {{if gt $page.UpdatesBehind 0}}
       <div class="pt-4">
-        <a href="?updates=behind" class="hero-tile hero-tile--amber" style="display:inline-flex;">
+		<a href="/settings/fleet-update" class="hero-tile hero-tile--amber" style="display:inline-flex;">
           <span class="hero-num">{{$page.UpdatesBehind}}</span>
           <span class="hero-label">{{if eq $page.UpdatesBehind 1}}host behind{{else}}hosts behind{{end}} · review →</span>
         </a>

web/templates/pages/fleet_update.html

@@ -8,18 +8,18 @@
   <div class="crumbs pt-6">
     <a href="/">Dashboard</a><span class="sep">/</span>
     <a href="/settings">Settings</a><span class="sep">/</span>
-    <span class="text-ink-mid">fleet update</span>
+	<span class="text-ink-mid">agent updates</span>
   </div>
 
   {{/* page header */}}
   <div class="flex items-baseline justify-between mt-3.5">
     <div>
       <h1 class="text-[22px] font-medium tracking-[-0.005em]">
-        Fleet update
+		Agent updates
         <span class="text-ink-fade font-normal text-[14px] ml-2 mono">target {{$page.TargetVersion}}</span>
       </h1>
       <p class="text-ink-mute text-[12px] mt-1 max-w-[760px] leading-[1.55]">
-        Rolling, sequential agent self-update. One host at a time, halts on first failure,
+		Rolling, sequential agent self-update. One host at a time, halts on first failure,
         cancellable mid-roll. Only online hosts whose <span class="mono">agent_version</span>
         differs from the server are eligible.
       </p>

web/templates/pages/settings.html

@@ -40,6 +40,7 @@
       {{if not $page.Form}}<span class="mono text-ink-fade text-[11px] ml-1">{{len $page.Channels}}</span>{{end}}
     </a>
     <a href="/settings/users" class="sub-tab {{if eq $page.ActiveTab "users"}}active{{end}}">Users</a>
+	<a href="/settings/fleet-update" class="sub-tab">Agent updates</a>
     <span class="sub-tab text-ink-fade cursor-default" title="lands later">Authentication</span>
   </div>
 

web/templates/partials/fleet_update_inner.html

@@ -32,6 +32,7 @@
     </div>
     <div class="text-[11.5px] text-ink-mute mt-1">
       target <span class="mono text-ink-mid">{{$page.Active.TargetVersion}}</span>
+	  {{if $page.Active.PauseAfterFirst}} · canary-first{{end}}
       · started <span class="mono text-ink-mid">{{relTime $page.Active.StartedAt}}</span>
       {{if $page.Active.CurrentHostID}}
         · waiting on <span class="mono text-ink-mid">{{index $page.HostNames $page.Active.CurrentHostID}}</span>
@@ -68,6 +69,12 @@
     {{if $page.Active.HaltedReason}}
       <div class="text-[12px] text-bad mt-2">{{$page.Active.HaltedReason}}</div>
     {{end}}
+	{{if eq $page.Active.Status "halted"}}
+	  <div class="mt-3 flex gap-2 text-[12px]">
+		<a class="btn" href="#fleet-update-start-form">Resume remaining as new rollout</a>
+		{{range $page.ActiveRows}}{{if eq .Status "failed"}}<button class="btn" type="button" onclick="fleetSelectOnly('{{.HostID}}');document.getElementById('fleet-update-start-form').scrollIntoView()">Retry {{.HostName}}</button>{{end}}{{end}}
+	  </div>
+	{{end}}
   </div>
 
   {{template "fleet_update_rows" $page}}
@@ -122,11 +129,11 @@
 {{define "fleet_update_idle_panel"}}
 {{$page := .}}
 <div class="panel rounded-[7px] px-5 py-4">
-  {{if eq (len $page.OutOfDateHosts) 0}}
+  {{if eq (len $page.Candidates) 0}}
     <div class="flex items-center gap-3">
       <span class="dot dot-online"></span>
       <div>
-        <div class="text-ink text-[14px] font-medium">All hosts are up to date.</div>
+        <div class="text-ink text-[14px] font-medium">No hosts enrolled.</div>
         <div class="text-ink-mute text-[12px] mt-0.5">
           Every online agent matches server version <span class="mono">{{$page.TargetVersion}}</span>.
         </div>
@@ -137,35 +144,48 @@
       <h2 class="text-[14px] font-medium">{{len $page.OutOfDateHosts}} host{{if ne (len $page.OutOfDateHosts) 1}}s{{end}} out of date</h2>
       <span class="mono text-[11px] text-ink-fade">target {{$page.TargetVersion}}</span>
     </div>
-    <ul class="mt-3 space-y-1 text-[12px]">
-      {{range $page.OutOfDateHosts}}
-        <li class="flex items-center gap-3">
-          <span class="dot dot-online"></span>
-          <span class="mono text-ink">{{.Name}}</span>
-          <span class="mono text-ink-mute">{{if .AgentVersion}}{{.AgentVersion}}{{else}}—{{end}} → {{$page.TargetVersion}}</span>
-        </li>
-      {{end}}
-    </ul>
-
-    <form id="fleet-update-start-form" class="mt-4 flex items-center gap-3"
-          hx-post="/api/fleet/update"
-          hx-headers='{"Content-Type":"application/json"}'
-          hx-vals='{}'
-          hx-swap="none"
-          hx-on::after-request="if(event.detail.successful) location.reload()">
-      <label class="text-[11.5px] text-ink-mute">
-        Type the count
-        <span class="mono text-ink-mid">({{len $page.OutOfDateHosts}})</span>
-        to enable Start:
+    <div class="mt-3 flex flex-wrap gap-2">
+      <input id="fleet-filter-name" class="field text-[12px]" placeholder="Filter name or tag" oninput="fleetFilter()">
+      <input id="fleet-filter-version" class="field text-[12px] mono" placeholder="Version" oninput="fleetFilter()">
+      <select id="fleet-filter-state" class="field text-[12px]" onchange="fleetFilter()">
+        <option value="all">All states</option><option value="eligible">Eligible</option><option value="excluded">Excluded</option>
+      </select>
+      <button type="button" class="btn" onclick="fleetSelectVisible(true)">Select visible eligible</button>
+      <button type="button" class="btn" onclick="fleetSelectVisible(false)">Clear selection</button>
+    </div>
+    <div class="panel mt-3 rounded-[7px] overflow-hidden">
+      {{range $page.Candidates}}
+      <label class="fleet-candidate grid items-center px-3 py-2 hairline text-[12px]"
+             data-name="{{.Host.Name}} {{range .Host.Tags}}{{.}} {{end}}" data-version="{{.Host.AgentVersion}}" data-state="{{if .Eligible}}eligible{{else}}excluded{{end}}"
+             style="grid-template-columns: 28px 1.4fr .8fr .8fr 1.2fr;gap:12px">
+        <input class="fleet-host" type="checkbox" value="{{.Host.ID}}" {{if .Eligible}}checked onchange="fleetReview()"{{else}}disabled{{end}}>
+        <span class="mono">{{.Host.Name}}</span>
+        <span class="mono text-ink-mute">{{if .Host.AgentVersion}}{{.Host.AgentVersion}}{{else}}unknown{{end}}</span>
+        <span class="mono text-ink-mute">{{$page.TargetVersion}}</span>
+        <span class="text-ink-mute">{{if .Eligible}}eligible{{else}}{{.Reason}}{{end}}</span>
       </label>
-      <input type="text" id="fleet-update-confirm" class="field mono text-[12.5px]"
-             style="width: 80px; padding: 5px 8px;"
-             oninput="document.getElementById('fleet-update-start-btn').disabled = (this.value !== '{{len $page.OutOfDateHosts}}');"
-             autocomplete="off" />
-      <button type="submit" id="fleet-update-start-btn" class="btn btn-amber" disabled>
-        Start fleet update
-      </button>
+      {{end}}
+    </div>
+    <div class="mt-4 text-[12px] text-ink-mute">
+      <span id="fleet-selected-count">{{len $page.OutOfDateHosts}}</span> selected · sequential, halts on first failure · worst-case
+      <span id="fleet-timeout" class="mono">{{len $page.OutOfDateHosts}} × 95s</span>
+    </div>
+    <form class="mt-3 flex items-center gap-3" onsubmit="return fleetStart(event)">
+	  <label class="text-[11.5px] text-ink-mute"><input id="fleet-canary" type="checkbox" checked> Pause after first host for canary review</label>
+      <label class="text-[11.5px] text-ink-mute">Type selected count to confirm:</label>
+      <input id="fleet-update-confirm" class="field mono text-[12.5px]" style="width:80px;padding:5px 8px" oninput="fleetReview()" autocomplete="off">
+      <button id="fleet-update-start-btn" class="btn btn-amber" disabled>Start agent update</button>
+      <span id="fleet-start-error" class="text-bad text-[12px]"></span>
     </form>
+    <script>
+      function fleetBoxes(){return Array.from(document.querySelectorAll('.fleet-host:not(:disabled)'))}
+      function fleetReview(){const n=fleetBoxes().filter(x=>x.checked).length;document.getElementById('fleet-selected-count').textContent=n;document.getElementById('fleet-timeout').textContent=n+' × 95s';document.getElementById('fleet-update-start-btn').disabled=n===0||document.getElementById('fleet-update-confirm').value!==String(n)}
+      function fleetFilter(){const q=document.getElementById('fleet-filter-name').value.toLowerCase(),v=document.getElementById('fleet-filter-version').value.toLowerCase(),s=document.getElementById('fleet-filter-state').value;document.querySelectorAll('.fleet-candidate').forEach(r=>r.hidden=!(r.dataset.name.toLowerCase().includes(q)&&r.dataset.version.toLowerCase().includes(v)&&(s==='all'||r.dataset.state===s)))}
+      function fleetSelectVisible(on){document.querySelectorAll('.fleet-candidate:not([hidden]) .fleet-host:not(:disabled)').forEach(x=>x.checked=on);fleetReview()}
+	  function fleetSelectOnly(id){fleetBoxes().forEach(x=>x.checked=x.value===id);fleetReview()}
+	  async function fleetStart(e){e.preventDefault();const ids=fleetBoxes().filter(x=>x.checked).map(x=>x.value),out=document.getElementById('fleet-start-error');out.textContent='';const res=await fetch('/api/fleet/update',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({host_ids:ids,canary_first:document.getElementById('fleet-canary').checked})});if(res.ok){location.reload();return false}const body=await res.json();out.textContent=body.message||body.code||'Unable to start';return false}
+      fleetReview()
+    </script>
   {{end}}
 </div>
 {{end}}