Merge pull request 'Release v1.2.2 — API and orphaned-job recovery' (#58) from release-v1.2.2 into main

.gitea/workflows/ci.yml

@@ -3,7 +3,7 @@
 # Notes for anyone editing this file:
 #
 # Custom runner image
-#   Every job runs inside `gitea.dcglab.co.uk/steve/ci-runner-go`
+#   Every job runs inside `docker.dcglab.co.uk/public/ci-runner-go:latest`
 #   (recipe: https://gitea.dcglab.co.uk/steve/ci/src/branch/main/images/ci-runner-go).
 #   That image already ships:
 #     * Go on PATH at /usr/local/go/bin (so `actions/setup-go` is
@@ -11,8 +11,8 @@
 #       otherwise re-download Go on every job)
 #     * Node.js + npm (used by docs / e2e workflows)
 #     * Docker CLI, Buildx, Compose v2 (used by docker-build steps)
-#   When bumping the Go floor, push a new ci-runner-go image with
+#   The runner image is intentionally tracked via `latest`: it is shared CI
-#   the matching Go version and bump the date pin in IMAGE below.
+#   infrastructure, and updates should propagate to consuming repositories.
 #
 # Self-hosted runner expectations
 #   Each runner host bind-mounts persistent volumes for
@@ -71,10 +71,7 @@ jobs:
     name: Test (${{ matrix.name }})
     runs-on: ubuntu-latest
     container:
-      image: docker.dcglab.co.uk/ci-runner-go:2026-05-15
+      image: docker.dcglab.co.uk/public/ci-runner-go:latest
-      credentials:
-        username: ${{ secrets.ZOT_USERNAME }}
-        password: ${{ secrets.ZOT_PASSWORD }}
     strategy:
       fail-fast: false
       matrix:
@@ -110,10 +107,7 @@ jobs:
     name: Lint
     runs-on: ubuntu-latest
     container:
-      image: docker.dcglab.co.uk/ci-runner-go:2026-05-15
+      image: docker.dcglab.co.uk/public/ci-runner-go:latest
-      credentials:
-        username: ${{ secrets.ZOT_USERNAME }}
-        password: ${{ secrets.ZOT_PASSWORD }}
     steps:
       - uses: actions/checkout@v4
       - uses: golangci/golangci-lint-action@v7
@@ -130,10 +124,7 @@ jobs:
     name: Build (${{ matrix.goos }}/${{ matrix.goarch }})
     runs-on: ubuntu-latest
     container:
-      image: docker.dcglab.co.uk/ci-runner-go:2026-05-15
+      image: docker.dcglab.co.uk/public/ci-runner-go:latest
-      credentials:
-        username: ${{ secrets.ZOT_USERNAME }}
-        password: ${{ secrets.ZOT_PASSWORD }}
     strategy:
       fail-fast: false
       matrix:

.gitea/workflows/e2e.yml

@@ -31,7 +31,7 @@ jobs:
   e2e:
     name: Playwright vs docker-compose
     runs-on: ubuntu-latest
-    container: gitea.dcglab.co.uk/steve/ci-runner-go:2026-05-08
+    container: docker.dcglab.co.uk/public/ci-runner-go:latest
     timeout-minutes: 15
     steps:
       - uses: actions/checkout@v4

.gitea/workflows/release.yml

@@ -12,7 +12,9 @@
 #     plus install.sh / install.ps1 / the systemd unit baked in under
 #     /opt/restic-manager/dist (the read-only fallback path the server
 #     handlers use when <DataDir>/... is empty).
-#   * Pushes to zot OCI registry (docker.dcglab.co.uk).
+#   * Pushes to the public namespace in the zot OCI registry
+#     (docker.dcglab.co.uk/public/restic-manager). Pulls are anonymous;
+#     pushes remain authenticated.
 #
 # Tag fan-out
 #   * tag push: :vX.Y.Z, :X.Y, :X
@@ -29,7 +31,7 @@ on:
 
 env:
   REGISTRY: docker.dcglab.co.uk
-  IMAGE_NAME: restic-manager
+  IMAGE_NAME: public/restic-manager
 
 # Force bash as the default shell — see ci.yml header.
 defaults:
@@ -41,10 +43,7 @@ jobs:
     name: Build + push image
     runs-on: ubuntu-latest
     container:
-      image: docker.dcglab.co.uk/ci-runner-go:2026-05-15
+      image: docker.dcglab.co.uk/public/ci-runner-go:latest
-      credentials:
-        username: ${{ secrets.ZOT_USERNAME }}
-        password: ${{ secrets.ZOT_PASSWORD }}
     steps:
       - uses: actions/checkout@v4
 

CHANGELOG.md

@@ -6,6 +6,66 @@ and the project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+## [1.2.2] - 2026-08-22
+
+### Added
+
+- Added audited single-job and bulk recovery controls for jobs that are no
+  longer running on their agents but remain stuck server-side. Agents now
+  acknowledge unknown cancellation targets so orphaned jobs become terminal,
+  while host ownership checks prevent cross-host updates. ([#56])
+
+### Fixed
+
+- Source-group JSON create and update requests now accept encrypted pre- and
+  post-hooks, preserve hooks when their fields are omitted, allow explicit
+  clearing, and expose only boolean presence indicators. ([#54])
+- Generic backup-job requests without paths are rejected before a broken job
+  is created. Configured run-now backups use the source-group endpoint. ([#55])
+
+## [1.2.1] - 2026-08-22
+
+### Fixed
+
+- Made the Agent updates host-selection screen stable and responsive: removed
+  the redundant typed-count confirmation, corrected name/version/eligibility
+  filtering, and stopped replacing idle or terminal controls through polling.
+  Progress polling now runs only while a rollout is actively running. ([#50])
+
+## [1.2.0] - 2026-08-22
+
+### Added
+
+- Fleet-wide agent updates are now discoverable from Settings and the
+  dashboard, with arbitrary host subsets, name/tag/version/eligibility
+  filtering, explicit exclusion reasons, server-side membership validation,
+  and a canary-first pause after the first verified reconnect. Halted rolls
+  expose retry and resume actions. ([#43])
+- Running jobs with no recent persisted activity are detected server-side and
+  surfaced through deduplicated alerts. Kind-aware thresholds protect normal
+  long-running work, terminal jobs self-resolve, and Prometheus exports stuck
+  job count and oldest inactivity age. ([#41])
+- Snapshot projections now retain host-level refresh timestamps—including
+  authoritative empty reports—expose a derived stale flag, refresh after
+  backup, forget, and prune, and support an explicit operator reconciliation
+  endpoint. ([#40])
+
+### Fixed
+
+- Raised the bounded WebSocket read limit on both peers so ordinary large
+  snapshot reports and restic events no longer disconnect otherwise healthy
+  agents. Regression coverage exercises payloads beyond the library's former
+  32 KiB default in both directions. ([#44])
+- Fleet-update timeout verification now performs a final authoritative read of
+  the agent version delivered by the reconnect `hello`, avoiding a poll/deadline
+  race and reporting the last observed version when verification fails. ([#43])
+
+### Changed
+
+- CI runner images and release images use the anonymous-pull `public`
+  namespace at `docker.dcglab.co.uk`; registry authentication remains required
+  only for publishing.
+
 ## [1.1.1] - 2026-08-22
 
 ### Fixed
@@ -140,7 +200,10 @@ with a web UI, JSON API, and self-updating agent fleet.
   go vet, golangci-lint).
 - Threat model published (`docs/threat-model.md`).
 
-[Unreleased]: https://gitea.dcglab.co.uk/steve/restic-manager/compare/v1.1.1...HEAD
+[Unreleased]: https://gitea.dcglab.co.uk/steve/restic-manager/compare/v1.2.2...HEAD
+[1.2.2]: https://gitea.dcglab.co.uk/steve/restic-manager/compare/v1.2.1...v1.2.2
+[1.2.1]: https://gitea.dcglab.co.uk/steve/restic-manager/compare/v1.2.0...v1.2.1
+[1.2.0]: https://gitea.dcglab.co.uk/steve/restic-manager/compare/v1.1.1...v1.2.0
 [1.1.1]: https://gitea.dcglab.co.uk/steve/restic-manager/compare/v1.1.0...v1.1.1
 [1.1.0]: https://gitea.dcglab.co.uk/steve/restic-manager/releases/tag/v1.1.0
 [1.0.0]: https://gitea.dcglab.co.uk/steve/restic-manager/releases/tag/v1.0.0
@@ -148,3 +211,11 @@ with a web UI, JSON API, and self-updating agent fleet.
 [#37]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/37
 [#36]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/36
 [#34]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/34
+[#40]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/40
+[#41]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/41
+[#43]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/43
+[#44]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/44
+[#50]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/50
+[#54]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/54
+[#55]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/55
+[#56]: https://gitea.dcglab.co.uk/steve/restic-manager/issues/56

cmd/agent/cancel_test.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"context"
+	"testing"
+
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
+)
+
+type recordingSender struct {
+	envelopes []api.Envelope
+}
+
+func (s *recordingSender) Send(env api.Envelope) error {
+	s.envelopes = append(s.envelopes, env)
+	return nil
+}
+
+func TestCancelCommandReportsUnknownJob(t *testing.T) {
+	d := &dispatcher{}
+	tx := &recordingSender{}
+	env, err := api.Marshal(api.MsgCommandCancel, "request-1", api.CommandCancelPayload{JobID: "missing-job"})
+	if err != nil {
+		t.Fatalf("marshal command: %v", err)
+	}
+
+	if err := d.handle(context.Background(), env, tx); err != nil {
+		t.Fatalf("handle cancel: %v", err)
+	}
+	if len(tx.envelopes) != 1 {
+		t.Fatalf("sent %d envelopes, want 1", len(tx.envelopes))
+	}
+	if tx.envelopes[0].Type != api.MsgCommandResult || tx.envelopes[0].ID != env.ID {
+		t.Fatalf("unexpected result envelope: %+v", tx.envelopes[0])
+	}
+	var result api.CommandResultPayload
+	if err := tx.envelopes[0].UnmarshalPayload(&result); err != nil {
+		t.Fatalf("unmarshal result: %v", err)
+	}
+	if result.JobID != "missing-job" || result.Accepted || result.Error != "job_not_found" {
+		t.Fatalf("unexpected result: %+v", result)
+	}
+}
+
+func TestCancelCommandCancelsTrackedJob(t *testing.T) {
+	d := &dispatcher{}
+	cancelled := false
+	d.trackJob("running-job", func() { cancelled = true })
+	tx := &recordingSender{}
+	env, err := api.Marshal(api.MsgCommandCancel, "request-2", api.CommandCancelPayload{JobID: "running-job"})
+	if err != nil {
+		t.Fatalf("marshal command: %v", err)
+	}
+
+	if err := d.handle(context.Background(), env, tx); err != nil {
+		t.Fatalf("handle cancel: %v", err)
+	}
+	if !cancelled {
+		t.Fatal("tracked job was not cancelled")
+	}
+	var result api.CommandResultPayload
+	if err := tx.envelopes[0].UnmarshalPayload(&result); err != nil {
+		t.Fatalf("unmarshal result: %v", err)
+	}
+	if !result.Accepted || result.Error != "" {
+		t.Fatalf("unexpected result: %+v", result)
+	}
+}

cmd/agent/main.go

@@ -276,7 +276,8 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S
 		if err := env.UnmarshalPayload(&p); err != nil {
 			return fmt.Errorf("command.cancel: %w", err)
 		}
-		if d.cancelJob(p.JobID) {
+		known := d.cancelJob(p.JobID)
+		if known {
 			slog.Info("ws agent: command.cancel applied", "job_id", p.JobID)
 		} else {
 			// Job already finished or was never seen on this agent.
@@ -284,6 +285,17 @@ func (d *dispatcher) handle(ctx context.Context, env api.Envelope, tx wsclient.S
 			// natural completion. Server-side state is authoritative.
 			slog.Info("ws agent: command.cancel for unknown job (already finished?)", "job_id", p.JobID)
 		}
+		result := api.CommandResultPayload{JobID: p.JobID, Accepted: known}
+		if !known {
+			result.Error = "job_not_found"
+		}
+		ack, err := api.Marshal(api.MsgCommandResult, env.ID, result)
+		if err != nil {
+			return fmt.Errorf("command.cancel result: %w", err)
+		}
+		if err := tx.Send(ack); err != nil {
+			return fmt.Errorf("command.cancel result send: %w", err)
+		}
 
 	case api.MsgTreeList:
 		// Synchronous RPC for the restore wizard's tree browser. The
@@ -296,6 +308,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 +420,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)

docs/book/src/operations/backups-and-restores.md

@@ -6,11 +6,12 @@ Three ways to trigger one:
 
 1. **Scheduled** — the agent's local cron fires at the time set
    on the schedule.
-2. **Run-now** — operator clicks **Run now** on the host detail
+2. **Run-now** — operator clicks **Run now** for a specific source
-   right rail. Posts to `/hosts/{id}/run-backup` (defaults to all
+   group. This uses `POST /hosts/{id}/source-groups/{gid}/run`.
-   source groups) or to a per-group form for finer control.
+3. **API** — use `POST /api/hosts/{id}/source-groups/{gid}/run` for a
-3. **API** — `POST /api/hosts/{id}/jobs` with the appropriate
+   configured source group. The lower-level `POST /api/hosts/{id}/jobs`
-   payload. Same audit + dispatch path.
+   backup form requires explicit paths in `args`; an empty backup is
+   rejected instead of dispatching a job that restic cannot run.
 
 In every case the server creates a `jobs` row, broadcasts a
 `command.run` to the host, and lands the operator on the live

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;
+	// Do this before job.finished so a failure in terminal reporting cannot
-	// the snapshot list is a follow-up that the host detail page polls
+	// prevent the independently useful projection refresh.
-	// 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.
 	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,11 +58,23 @@ type Engine struct {
 	// we raise. Configurable for tests; default 15m.
 	agentOfflineFloor time.Duration
 	tickPeriod        time.Duration
+	stuckThresholds   map[string]time.Duration
 
 	closeOnce sync.Once
+	notifyWG  sync.WaitGroup
 	done      chan struct{}
 }
 
+func (e *Engine) dispatchNotification(ctx context.Context, payload notification.Payload) {
+	e.notifyWG.Add(1)
+	go func() {
+		defer e.notifyWG.Done()
+		e.hub.Dispatch(ctx, payload)
+	}()
+}
+
+func (e *Engine) waitNotifications() { e.notifyWG.Wait() }
+
 // NewEngine builds the engine. agentOfflineFloor + tickPeriod default
 // to 15min and 60s respectively when zero.
 func NewEngine(st *store.Store, hub *notification.Hub) *Engine {
@@ -69,6 +86,11 @@ func NewEngine(st *store.Store, hub *notification.Hub) *Engine {
 		hostUp:            make(chan string, 32),
 		agentOfflineFloor: 15 * time.Minute,
 		tickPeriod:        60 * time.Second,
+		stuckThresholds: map[string]time.Duration{
+			"backup":  longStuckJobThreshold,
+			"restore": longStuckJobThreshold,
+			"check":   longStuckJobThreshold,
+		},
 		done: make(chan struct{}),
 	}
 }
@@ -125,6 +147,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 +236,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 +284,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
@@ -56,7 +60,7 @@ func (e *Engine) raiseAndNotify(ctx context.Context, hostID, kind, dedupKey, sev
 	if err == nil {
 		hostName = host.Name
 	}
-	go e.hub.Dispatch(ctx, notification.Payload{
+	e.dispatchNotification(ctx, notification.Payload{
 		Event:    notification.EventRaised,
 		AlertID:  id,
 		Severity: severity,
@@ -81,7 +85,7 @@ func (e *Engine) Acknowledge(ctx context.Context, alertID, userID string, when t
 		return nil //nolint:nilerr
 	}
 	p := alertPayload(ctx, e.store, notification.EventAcknowledged, a)
-	go e.hub.Dispatch(context.WithoutCancel(ctx), p)
+	e.dispatchNotification(context.WithoutCancel(ctx), p)
 	return nil
 }
 
@@ -95,7 +99,7 @@ func (e *Engine) Resolve(ctx context.Context, alertID string, when time.Time) er
 		return nil
 	}
 	p := alertPayload(ctx, e.store, notification.EventResolved, a)
-	go e.hub.Dispatch(context.WithoutCancel(ctx), p)
+	e.dispatchNotification(context.WithoutCancel(ctx), p)
 	return nil
 }
 
@@ -160,7 +164,7 @@ func (e *Engine) resolveAndNotify(ctx context.Context, hostID, kind, dedupKey st
 		if a.Kind != kind || a.DedupKey != dedupKey {
 			continue
 		}
-		go e.hub.Dispatch(ctx, notification.Payload{
+		e.dispatchNotification(ctx, notification.Payload{
 			Event:    notification.EventResolved,
 			AlertID:  a.ID,
 			Severity: a.Severity,

internal/alert/rules_test.go

@@ -3,6 +3,7 @@ package alert
 import (
 	"context"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -24,6 +25,7 @@ func setupEngine(t *testing.T) (*Engine, *store.Store, string) {
 	aead, _ := crypto.NewAEAD(key)
 	hub := notification.NewHub(st, aead, "https://rm.example")
 	eng := NewEngine(st, hub)
+	t.Cleanup(eng.waitNotifications)
 	hostID := ulid.Make().String()
 	if err := st.CreateHost(context.Background(), store.Host{
 		ID: hostID, Name: "alfa-01", OS: "linux", Arch: "amd64",
@@ -69,6 +71,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"
@@ -35,6 +42,7 @@ const (
 	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

@@ -6,7 +6,9 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"io"
 	stdhttp "net/http"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -19,6 +21,67 @@ import (
 	"gitea.dcglab.co.uk/steve/restic-manager/internal/version"
 )
 
+func TestFleetUpdateIdlePageDoesNotPollOrRequireTypedConfirmation(t *testing.T) {
+	t.Parallel()
+	_, baseURL, st := newTestServerWithUI(t)
+	cookie := loginAsAdmin(t, st)
+	_ = makeHost(t, st, "fleet-filter-host")
+
+	req, _ := stdhttp.NewRequest("GET", baseURL+"/settings/fleet-update", nil)
+	req.AddCookie(cookie)
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("get fleet update page: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusOK {
+		t.Fatalf("status: got %d, want 200", res.StatusCode)
+	}
+	raw, _ := io.ReadAll(res.Body)
+	body := string(raw)
+
+	for _, want := range []string{
+		`id="fleet-filter-name"`,
+		`id="fleet-filter-version"`,
+		`id="fleet-filter-state"`,
+		"r.style.display=",
+	} {
+		if !strings.Contains(body, want) {
+			t.Errorf("page missing %q", want)
+		}
+	}
+	if strings.Contains(body, "Type selected count to confirm") || strings.Contains(body, `id="fleet-update-confirm"`) {
+		t.Error("page still renders the typed-count confirmation")
+	}
+	if strings.Contains(body, `hx-trigger="every 3s`) {
+		t.Error("idle selection page must not poll and replace operator input")
+	}
+}
+
+func TestFleetUpdateRunningPagePollsForProgress(t *testing.T) {
+	t.Parallel()
+	_, baseURL, st := newTestServerWithUI(t)
+	cookie, userID := loginAsAdminWithID(t, st)
+	hostID := makeHost(t, st, "fleet-running-host")
+	if err := st.CreateFleetUpdate(context.Background(), store.FleetUpdate{
+		ID: ulid.Make().String(), StartedByUserID: userID, TargetVersion: version.Version,
+	}, []string{hostID}); err != nil {
+		t.Fatalf("create fleet update: %v", err)
+	}
+
+	req, _ := stdhttp.NewRequest("GET", baseURL+"/settings/fleet-update", nil)
+	req.AddCookie(cookie)
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("get fleet update page: %v", err)
+	}
+	defer res.Body.Close()
+	raw, _ := io.ReadAll(res.Body)
+	if body := string(raw); !strings.Contains(body, `hx-trigger="every 3s`) {
+		t.Error("running rollout page must poll for progress")
+	}
+}
+
 // fakeFleetWorker stands in for *fleetupdate.Worker in HTTP tests.
 // It records what was passed to Start/Cancel and lets tests inject
 // canned errors. Satisfies the FleetWorker interface in
@@ -50,6 +113,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 +257,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/job_abandon.go

@@ -0,0 +1,143 @@
+package http
+
+import (
+	"context"
+	"errors"
+	stdhttp "net/http"
+	"time"
+
+	"github.com/go-chi/chi/v5"
+	"github.com/oklog/ulid/v2"
+
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/alert"
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/api"
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
+)
+
+func (s *Server) abandonJob(ctx context.Context, userID *string, jobID, reason string) error {
+	job, err := s.deps.Store.GetJob(ctx, jobID)
+	if err != nil {
+		return err
+	}
+	switch api.JobStatus(job.Status) {
+	case api.JobSucceeded, api.JobFailed, api.JobCancelled:
+		return errJobTerminal
+	}
+	when := time.Now().UTC()
+	if err := s.deps.Store.MarkJobFinished(ctx, jobID, string(api.JobCancelled), -1, nil, reason, when); err != nil {
+		return err
+	}
+	_ = s.deps.Store.AppendAudit(ctx, store.AuditEntry{
+		ID: ulid.Make().String(), UserID: userID, Actor: "user",
+		Action: "job.abandon", TargetKind: ptr("job"), TargetID: &jobID,
+		TS: when,
+	})
+	return nil
+}
+
+var errJobTerminal = errors.New("job already terminal")
+
+func (s *Server) handleAbandonJob(w stdhttp.ResponseWriter, r *stdhttp.Request) {
+	user, ok := s.requireUser(r)
+	if !ok {
+		writeJSONError(w, stdhttp.StatusUnauthorized, "unauthorised", "")
+		return
+	}
+	jobID := chi.URLParam(r, "id")
+	err := s.abandonJob(r.Context(), &user.ID, jobID, "abandoned by operator")
+	switch {
+	case errors.Is(err, store.ErrNotFound):
+		writeJSONError(w, stdhttp.StatusNotFound, "job_not_found", "")
+	case errors.Is(err, errJobTerminal):
+		writeJSONError(w, stdhttp.StatusConflict, "job_terminal", "job is already terminal")
+	case err != nil:
+		writeJSONError(w, stdhttp.StatusInternalServerError, "internal", "")
+	default:
+		writeJSON(w, stdhttp.StatusOK, map[string]string{"job_id": jobID, "status": string(api.JobCancelled)})
+	}
+}
+
+func (s *Server) abandonStuckAlerts(ctx context.Context, userID *string) (int, error) {
+	open, err := s.deps.Store.ListAlerts(ctx, store.AlertFilter{Status: "open"})
+	if err != nil {
+		return 0, err
+	}
+	acked, err := s.deps.Store.ListAlerts(ctx, store.AlertFilter{Status: "acknowledged"})
+	if err != nil {
+		return 0, err
+	}
+	count := 0
+	seen := map[string]bool{}
+	for _, a := range append(open, acked...) {
+		if a.Kind != alert.KindJobStuck || seen[a.DedupKey] {
+			continue
+		}
+		seen[a.DedupKey] = true
+		err := s.abandonJob(ctx, userID, a.DedupKey, "abandoned by operator after stuck-job alert")
+		if err != nil {
+			if errors.Is(err, errJobTerminal) || errors.Is(err, store.ErrNotFound) {
+				s.resolveAlert(ctx, a.ID)
+				continue
+			}
+			return count, err
+		}
+		count++
+		s.resolveAlert(ctx, a.ID)
+	}
+	return count, nil
+}
+
+func (s *Server) resolveAlert(ctx context.Context, alertID string) {
+	when := time.Now().UTC()
+	if s.deps.AlertEngine != nil {
+		_ = s.deps.AlertEngine.Resolve(ctx, alertID, when)
+		return
+	}
+	_ = s.deps.Store.Resolve(ctx, alertID, when)
+}
+
+func (s *Server) handleAbandonStuckJobs(w stdhttp.ResponseWriter, r *stdhttp.Request) {
+	user, ok := s.requireUser(r)
+	if !ok {
+		writeJSONError(w, stdhttp.StatusUnauthorized, "unauthorised", "")
+		return
+	}
+	count, err := s.abandonStuckAlerts(r.Context(), &user.ID)
+	if err != nil {
+		writeJSONError(w, stdhttp.StatusInternalServerError, "internal", "")
+		return
+	}
+	writeJSON(w, stdhttp.StatusOK, map[string]int{"abandoned": count})
+}
+
+func (s *Server) handleUIAbandonAlertJob(w stdhttp.ResponseWriter, r *stdhttp.Request) {
+	user := s.requireUIUser(w, r)
+	if user == nil {
+		return
+	}
+	alertID := chi.URLParam(r, "id")
+	a, err := s.deps.Store.GetAlert(r.Context(), alertID)
+	if err != nil || a == nil || a.Kind != alert.KindJobStuck {
+		stdhttp.Error(w, "stuck-job alert not found", stdhttp.StatusNotFound)
+		return
+	}
+	if err := s.abandonJob(r.Context(), &user.ID, a.DedupKey, "abandoned by operator after stuck-job alert"); err != nil && !errors.Is(err, errJobTerminal) {
+		stdhttp.Error(w, "unable to abandon job", stdhttp.StatusConflict)
+		return
+	}
+	s.resolveAlert(r.Context(), alertID)
+	w.Header().Set("HX-Redirect", "/alerts?"+r.URL.RawQuery)
+	w.WriteHeader(stdhttp.StatusNoContent)
+}
+
+func (s *Server) handleUIAbandonStuckJobs(w stdhttp.ResponseWriter, r *stdhttp.Request) {
+	user := s.requireUIUser(w, r)
+	if user == nil {
+		return
+	}
+	if _, err := s.abandonStuckAlerts(r.Context(), &user.ID); err != nil {
+		stdhttp.Error(w, "unable to abandon stuck jobs", stdhttp.StatusInternalServerError)
+		return
+	}
+	stdhttp.Redirect(w, r, "/alerts", stdhttp.StatusSeeOther)
+}

internal/server/http/job_abandon_test.go

@@ -0,0 +1,93 @@
+package http
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	stdhttp "net/http"
+	"testing"
+	"time"
+
+	"github.com/oklog/ulid/v2"
+
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/alert"
+	"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
+)
+
+func seedRunningJob(t *testing.T, st *store.Store, hostID string) string {
+	t.Helper()
+	id := ulid.Make().String()
+	when := time.Now().UTC().Add(-24 * time.Hour)
+	if err := st.CreateJob(context.Background(), store.Job{
+		ID: id, HostID: hostID, Kind: "backup", ActorKind: "user", CreatedAt: when,
+	}); err != nil {
+		t.Fatalf("create job: %v", err)
+	}
+	if err := st.MarkJobStarted(context.Background(), id, when); err != nil {
+		t.Fatalf("start job: %v", err)
+	}
+	return id
+}
+
+func TestAbandonJobMakesOrphanTerminalAndAudits(t *testing.T) {
+	t.Parallel()
+	_, ts, st := rawTestServer(t)
+	hostID := makeHost(t, st, "abandon-host")
+	jobID := seedRunningJob(t, st, hostID)
+	req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/jobs/"+jobID+"/abandon", nil)
+	req.AddCookie(loginAsAdmin(t, st))
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("abandon: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusOK {
+		t.Fatalf("status: got %d", res.StatusCode)
+	}
+	job, _ := st.GetJob(context.Background(), jobID)
+	if job.Status != "cancelled" || job.FinishedAt == nil || job.Error == nil {
+		t.Fatalf("job not terminalised: %+v", job)
+	}
+	var audits int
+	_ = st.DB().QueryRow(`SELECT COUNT(*) FROM audit_log WHERE action = 'job.abandon' AND target_id = ?`, jobID).Scan(&audits)
+	if audits != 1 {
+		t.Fatalf("audit rows: got %d", audits)
+	}
+}
+
+func TestBulkAbandonTargetsOnlyAlertedStuckJobs(t *testing.T) {
+	t.Parallel()
+	_, ts, st := rawTestServer(t)
+	hostID := makeHost(t, st, "bulk-abandon-host")
+	stuckID := seedRunningJob(t, st, hostID)
+	unalertedID := seedRunningJob(t, st, hostID)
+	_, _, err := st.RaiseOrTouch(context.Background(), hostID, alert.KindJobStuck, stuckID,
+		"warning", "stuck", time.Now().UTC())
+	if err != nil {
+		t.Fatalf("raise alert: %v", err)
+	}
+	req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/jobs/abandon-stuck", bytes.NewReader(nil))
+	req.AddCookie(loginAsAdmin(t, st))
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("bulk abandon: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusOK {
+		t.Fatalf("status: got %d", res.StatusCode)
+	}
+	var out map[string]int
+	_ = json.NewDecoder(res.Body).Decode(&out)
+	if out["abandoned"] != 1 {
+		t.Fatalf("abandoned: %+v", out)
+	}
+	stuck, _ := st.GetJob(context.Background(), stuckID)
+	unalerted, _ := st.GetJob(context.Background(), unalertedID)
+	if stuck.Status != "cancelled" || unalerted.Status != "running" {
+		t.Fatalf("statuses: stuck=%s unalerted=%s", stuck.Status, unalerted.Status)
+	}
+	open, _ := st.ListAlerts(context.Background(), store.AlertFilter{Status: "open"})
+	if len(open) != 0 {
+		t.Fatalf("stuck alerts remained open: %+v", open)
+	}
+}

internal/server/http/jobs.go

@@ -69,6 +69,10 @@ func (s *Server) dispatchJob(ctx context.Context, user *store.User,
 		Kind: kind,
 		Args: args,
 	}
+	if kind == api.JobBackup && len(args) == 0 {
+		return res, stdhttp.StatusUnprocessableEntity, "backup_paths_required",
+			"backup requires paths in args; configured backups must use POST /api/hosts/{id}/source-groups/{gid}/run"
+	}
 	if kind == api.JobForget {
 		if !validForgetArgs(args) {
 			return res, stdhttp.StatusBadRequest, "invalid_args",

internal/server/http/jobs_test.go

@@ -14,6 +14,34 @@ import (
 	"gitea.dcglab.co.uk/steve/restic-manager/internal/store"
 )
 
+func TestRunNowBackupRejectsEmptyPaths(t *testing.T) {
+	t.Parallel()
+	_, ts, st := rawTestServer(t)
+	cookie := loginAsAdmin(t, st)
+	hostID := makeHost(t, st, "empty-backup-host")
+	req, _ := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/jobs",
+		bytes.NewReader([]byte(`{"kind":"backup"}`)))
+	req.Header.Set("Content-Type", "application/json")
+	req.AddCookie(cookie)
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("post run-now: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusUnprocessableEntity {
+		t.Fatalf("status: got %d, want 422", res.StatusCode)
+	}
+	body := readJSONError(t, res.Body)
+	if body.Code != "backup_paths_required" {
+		t.Fatalf("code: got %q", body.Code)
+	}
+	var jobs int
+	_ = st.DB().QueryRow(`SELECT COUNT(*) FROM jobs WHERE host_id = ?`, hostID).Scan(&jobs)
+	if jobs != 0 {
+		t.Fatalf("invalid backup created %d jobs", jobs)
+	}
+}
+
 func TestRunNowForgetShipsRetentionGroupsAndDryRun(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/p2r01_test.go

@@ -153,6 +153,8 @@ func TestSourceGroupsCRUD(t *testing.T) {
 			},
 			"retry_max":             3,
 			"retry_backoff_seconds": 60,
+			"pre_hook":              "prepare-db",
+			"post_hook":             "resume-db",
 		}, cookie)
 	if status != 201 {
 		t.Fatalf("create status: %d, body: %+v", status, body)
@@ -161,6 +163,16 @@ func TestSourceGroupsCRUD(t *testing.T) {
 	if gid == "" {
 		t.Fatalf("create: no id returned: %+v", body)
 	}
+	if body["has_pre_hook"] != true || body["has_post_hook"] != true {
+		t.Fatalf("create hook indicators: %+v", body)
+	}
+	stored, err := st.GetSourceGroup(context.Background(), hostID, gid)
+	if err != nil {
+		t.Fatalf("get stored group: %v", err)
+	}
+	if stored.PreHook == "prepare-db" || stored.PostHook == "resume-db" || stored.PreHook == "" || stored.PostHook == "" {
+		t.Fatalf("hooks not encrypted at rest: pre=%q post=%q", stored.PreHook, stored.PostHook)
+	}
 
 	// Duplicate name → 409.
 	status, _ = doJSON(t, url, "POST", "/api/hosts/"+hostID+"/source-groups",
@@ -185,6 +197,20 @@ func TestSourceGroupsCRUD(t *testing.T) {
 	if got := body["name"]; got != "system" {
 		t.Errorf("rename: got %v want system", got)
 	}
+	stored, _ = st.GetSourceGroup(context.Background(), hostID, gid)
+	if stored.PreHook == "" || stored.PostHook == "" {
+		t.Fatal("PUT without hook fields cleared existing hooks")
+	}
+
+	// Explicit empty hook clears only that hook; plaintext is never returned.
+	status, body = doJSON(t, url, "PUT", "/api/hosts/"+hostID+"/source-groups/"+gid,
+		map[string]any{"name": "system", "includes": []string{"/etc"}, "pre_hook": ""}, cookie)
+	if status != 200 || body["has_pre_hook"] != false || body["has_post_hook"] != true {
+		t.Fatalf("clear hook status=%d body=%+v", status, body)
+	}
+	if _, exposed := body["post_hook"]; exposed {
+		t.Fatal("source-group response exposed hook plaintext field")
+	}
 
 	// Delete.
 	status, _ = doJSON(t, url, "DELETE", "/api/hosts/"+hostID+"/source-groups/"+gid, nil, cookie)

internal/server/http/server.go

@@ -261,7 +261,10 @@ func (s *Server) routes(r chi.Router) {
 		r.Post("/api/hosts/{id}/repo/check", s.handleRunRepoCheck)
 		r.Post("/api/hosts/{id}/repo/unlock", s.handleRunRepoUnlock)
 		r.Post("/api/jobs/{id}/cancel", s.handleCancelJob)
+		r.Post("/api/jobs/{id}/abandon", s.handleAbandonJob)
+		r.Post("/api/jobs/abandon-stuck", s.handleAbandonStuckJobs)
 		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)
@@ -297,6 +300,8 @@ func (s *Server) routes(r chi.Router) {
 			r.Post("/hosts/{id}/restore", s.handleUIRestorePost)
 			r.Post("/alerts/{id}/acknowledge", s.handleUIAlertAcknowledge)
 			r.Post("/alerts/{id}/resolve", s.handleUIAlertResolve)
+			r.Post("/alerts/{id}/abandon-job", s.handleUIAbandonAlertJob)
+			r.Post("/alerts/abandon-stuck", s.handleUIAbandonStuckJobs)
 		}
 	})
 

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 {
+	out.RefreshedAt = host.SnapshotRefreshedAt
-		t := snaps[0].RefreshedAt
+	mutationAt, err := s.deps.Store.LatestSuccessfulRepoMutation(r.Context(), hostID)
-		out.RefreshedAt = &t
+	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/http/source_groups.go

@@ -29,6 +29,8 @@ type sourceGroupView struct {
 	RetryMax            int                   `json:"retry_max"`
 	RetryBackoffSeconds int                   `json:"retry_backoff_seconds"`
 	ConflictDimension   string                `json:"conflict_dimension,omitempty"`
+	HasPreHook          bool                  `json:"has_pre_hook"`
+	HasPostHook         bool                  `json:"has_post_hook"`
 	CreatedAt           time.Time             `json:"created_at"`
 	UpdatedAt           time.Time             `json:"updated_at"`
 }
@@ -49,6 +51,8 @@ func toSourceGroupView(g store.SourceGroup) sourceGroupView {
 		RetryMax:            g.RetryMax,
 		RetryBackoffSeconds: g.RetryBackoffSeconds,
 		ConflictDimension:   g.ConflictDimension,
+		HasPreHook:          g.PreHook != "",
+		HasPostHook:         g.PostHook != "",
 		CreatedAt:           g.CreatedAt,
 		UpdatedAt:           g.UpdatedAt,
 	}
@@ -62,6 +66,29 @@ type sourceGroupWriteRequest struct {
 	RetentionPolicy     store.RetentionPolicy `json:"retention_policy"`
 	RetryMax            int                   `json:"retry_max"`
 	RetryBackoffSeconds int                   `json:"retry_backoff_seconds"`
+	// Pointer fields distinguish omission (preserve on PUT) from an explicit
+	// empty string (clear). Hook plaintext is accepted on write but never
+	// returned by the JSON API.
+	PreHook  *string `json:"pre_hook,omitempty"`
+	PostHook *string `json:"post_hook,omitempty"`
+}
+
+func (s *Server) applySourceGroupHooks(hostID string, req sourceGroupWriteRequest, g *store.SourceGroup) error {
+	if req.PreHook != nil {
+		enc, err := s.EncryptHookForGroup(hostID, "pre", *req.PreHook)
+		if err != nil {
+			return err
+		}
+		g.PreHook = enc
+	}
+	if req.PostHook != nil {
+		enc, err := s.EncryptHookForGroup(hostID, "post", *req.PostHook)
+		if err != nil {
+			return err
+		}
+		g.PostHook = enc
+	}
+	return nil
 }
 
 func (s *Server) handleListSourceGroups(w stdhttp.ResponseWriter, r *stdhttp.Request) {
@@ -142,6 +169,10 @@ func (s *Server) handleCreateSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Re
 		RetryMax:            req.RetryMax,
 		RetryBackoffSeconds: req.RetryBackoffSeconds,
 	}
+	if err := s.applySourceGroupHooks(hostID, req, &g); err != nil {
+		writeJSONError(w, stdhttp.StatusInternalServerError, "hook_encryption_failed", "")
+		return
+	}
 	if err := s.deps.Store.CreateSourceGroup(r.Context(), &g); err != nil {
 		writeJSONError(w, stdhttp.StatusInternalServerError, "internal", err.Error())
 		return
@@ -157,7 +188,8 @@ func (s *Server) handleUpdateSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Re
 	}
 	hostID := chi.URLParam(r, "id")
 	groupID := chi.URLParam(r, "gid")
-	if _, err := s.deps.Store.GetSourceGroup(r.Context(), hostID, groupID); err != nil {
+	existingGroup, err := s.deps.Store.GetSourceGroup(r.Context(), hostID, groupID)
+	if err != nil {
 		if errors.Is(err, store.ErrNotFound) {
 			writeJSONError(w, stdhttp.StatusNotFound, "group_not_found", "")
 			return
@@ -188,6 +220,12 @@ func (s *Server) handleUpdateSourceGroup(w stdhttp.ResponseWriter, r *stdhttp.Re
 		RetentionPolicy:     req.RetentionPolicy,
 		RetryMax:            req.RetryMax,
 		RetryBackoffSeconds: req.RetryBackoffSeconds,
+		PreHook:             existingGroup.PreHook,
+		PostHook:            existingGroup.PostHook,
+	}
+	if err := s.applySourceGroupHooks(hostID, req, &g); err != nil {
+		writeJSONError(w, stdhttp.StatusInternalServerError, "hook_encryption_failed", "")
+		return
 	}
 	if err := s.deps.Store.UpdateSourceGroup(r.Context(), &g); err != nil {
 		writeJSONError(w, stdhttp.StatusInternalServerError, "internal", err.Error())

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
@@ -356,10 +357,26 @@ func dispatchAgentMessage(ctx context.Context, c *Conn, hostID string, env api.E
 		}
 
 	case api.MsgCommandResult:
-		// TODO(P2): persist command.result acks for "did the agent
+		var p api.CommandResultPayload
-		// accept the dispatch?" forensics. Currently the job lifecycle
+		if err := env.UnmarshalPayload(&p); err != nil {
-		// (job.started → job.finished) is sufficient signal.
+			slog.Warn("ws: decode command result", "host_id", hostID, "err", err)
-		slog.Debug("ws msg not yet handled", "type", env.Type, "host_id", hostID)
+			break
+		}
+		// A cancel for a job unknown to the current agent process can never
+		// produce job.finished. Terminalise the orphan server-side.
+		if !p.Accepted && p.Error == "job_not_found" && p.JobID != "" {
+			job, err := deps.Store.GetJob(ctx, p.JobID)
+			if err != nil || job.HostID != hostID {
+				slog.Warn("ws: reject orphan result for foreign or missing job", "job_id", p.JobID, "host_id", hostID)
+				break
+			}
+			if job.Status == string(api.JobQueued) || job.Status == string(api.JobRunning) {
+				if err := deps.Store.MarkJobFinished(ctx, p.JobID, string(api.JobCancelled), -1, nil,
+					"agent reported job not found during cancellation", time.Now().UTC()); err != nil {
+					slog.Warn("ws: terminalise orphaned job", "job_id", p.JobID, "err", err)
+				}
+			}
+		}
 
 	case api.MsgTreeListResult:
 		// Reply to a synchronous tree.list RPC. Route to the waiter

internal/server/ws/handler_test.go

@@ -36,6 +36,54 @@ func seedHostWS(t *testing.T, s *store.Store, hostID string) {
 func int64ptrWS(v int64) *int64 { return &v }
 func boolptrWS(v bool) *bool    { return &v }
 
+func TestUnknownCancelResultTerminalisesOrphanedJob(t *testing.T) {
+	t.Parallel()
+	s := openWSTestStore(t)
+	ctx := context.Background()
+	const hostID = "h-orphan-ws"
+	seedHostWS(t, s, hostID)
+	jobID := "j-orphan-ws"
+	started := time.Now().UTC().Add(-time.Hour)
+	if err := s.CreateJob(ctx, store.Job{ID: jobID, HostID: hostID, Kind: "backup", ActorKind: "user", CreatedAt: started}); err != nil {
+		t.Fatalf("create job: %v", err)
+	}
+	if err := s.MarkJobStarted(ctx, jobID, started); err != nil {
+		t.Fatalf("start job: %v", err)
+	}
+	env, _ := api.Marshal(api.MsgCommandResult, jobID, api.CommandResultPayload{
+		JobID: jobID, Accepted: false, Error: "job_not_found",
+	})
+	dispatchAgentMessage(ctx, nil, hostID, env, HandlerDeps{Store: s})
+	job, _ := s.GetJob(ctx, jobID)
+	if job.Status != "cancelled" || job.FinishedAt == nil {
+		t.Fatalf("orphan not terminalised: %+v", job)
+	}
+}
+
+func TestUnknownCancelResultCannotTerminaliseAnotherHostsJob(t *testing.T) {
+	t.Parallel()
+	s := openWSTestStore(t)
+	ctx := context.Background()
+	seedHostWS(t, s, "reporting-host")
+	seedHostWS(t, s, "job-host")
+	const jobID = "j-foreign-ws"
+	started := time.Now().UTC().Add(-time.Hour)
+	if err := s.CreateJob(ctx, store.Job{ID: jobID, HostID: "job-host", Kind: "backup", ActorKind: "user", CreatedAt: started}); err != nil {
+		t.Fatalf("create job: %v", err)
+	}
+	if err := s.MarkJobStarted(ctx, jobID, started); err != nil {
+		t.Fatalf("start job: %v", err)
+	}
+	env, _ := api.Marshal(api.MsgCommandResult, jobID, api.CommandResultPayload{
+		JobID: jobID, Accepted: false, Error: "job_not_found",
+	})
+	dispatchAgentMessage(ctx, nil, "reporting-host", env, HandlerDeps{Store: s})
+	job, _ := s.GetJob(ctx, jobID)
+	if job.Status != "running" || job.FinishedAt != nil {
+		t.Fatalf("foreign job was modified: %+v", job)
+	}
+}
+
 func TestRepoStatsReportPersisted(t *testing.T) {
 	t.Parallel()
 	s := openWSTestStore(t)

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)
+		`INSERT INTO fleet_updates (id, started_at, started_by_user_id, target_version, status, pause_after_first)
-		 VALUES (?, ?, ?, ?, ?)`,
+		 VALUES (?, ?, ?, ?, ?, ?)`,
-		fu.ID, fu.StartedAt.UTC().Format(time.RFC3339Nano), fu.StartedByUserID, fu.TargetVersion, fu.Status,
+		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 = ?`,
+		`UPDATE hosts SET snapshot_count = ?, snapshot_refreshed_at = ? WHERE id = ?`,
-		len(snaps), hostID); err != nil {
+		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/alerts.html

@@ -24,6 +24,10 @@
       </h1>
     </div>
     <div class="flex gap-2">
+      <form method="post" action="/alerts/abandon-stuck"
+            onsubmit="return confirm('Mark every currently alerted stuck job as cancelled? Use this only for jobs no longer running on their agents.');">
+        <button type="submit" class="btn btn-danger">Abandon all stuck jobs</button>
+      </form>
       <a href="/settings/notifications" class="btn">Channel settings →</a>
     </div>
   </div>

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,14 +8,14 @@
   <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]">
@@ -28,5 +28,15 @@
 
   {{template "fleet_update_inner" .}}
 
+  <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}
+    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.style.display=r.dataset.name.toLowerCase().includes(q)&&r.dataset.version.toLowerCase().includes(v)&&(s==='all'||r.dataset.state===s)?'':'none')}
+    function fleetSelectVisible(on){document.querySelectorAll('.fleet-candidate').forEach(r=>{const x=r.querySelector('.fleet-host:not(:disabled)');if(x&&r.style.display!=='none')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}
+    if(document.getElementById('fleet-selected-count'))fleetReview()
+  </script>
+
 </div>
 {{end}}

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/alert_row.html

@@ -66,6 +66,14 @@
   {{/* actions */}}
   <div style="display: flex; gap: 6px; justify-content: flex-end; align-items: center;">
     {{if eq $status "open"}}
+      {{if eq $a.Kind "job_stuck"}}
+      <form method="post" action="/alerts/{{$a.ID}}/abandon-job">
+        <button type="submit" class="btn btn-danger"
+                hx-post="/alerts/{{$a.ID}}/abandon-job{{if $qs}}?{{$qs}}{{end}}"
+                hx-swap="none"
+                hx-confirm="Mark this job cancelled on the server? Use this only when it is no longer running on the agent.">Abandon job</button>
+      </form>
+      {{end}}
       <form method="post" action="/alerts/{{$a.ID}}/acknowledge">
         {{if $qs}}<input type="hidden" name="qs" value="{{$qs}}">{{end}}
         <button type="submit" class="btn"
@@ -89,6 +97,14 @@
         {{end}}
         ack'd{{if $ackedBy}} by {{$ackedBy}}{{end}} · {{relTime $a.AcknowledgedAt}}
       </span>
+      {{if eq $a.Kind "job_stuck"}}
+      <form method="post" action="/alerts/{{$a.ID}}/abandon-job">
+        <button type="submit" class="btn btn-danger"
+                hx-post="/alerts/{{$a.ID}}/abandon-job{{if $qs}}?{{$qs}}{{end}}"
+                hx-swap="none"
+                hx-confirm="Mark this job cancelled on the server? Use this only when it is no longer running on the agent.">Abandon job</button>
+      </form>
+      {{end}}
       <form method="post" action="/alerts/{{$a.ID}}/resolve">
         {{if $qs}}<input type="hidden" name="qs" value="{{$qs}}">{{end}}
         <button type="submit" class="btn"

web/templates/partials/fleet_update_inner.html

@@ -7,11 +7,11 @@
 */}}
 {{define "fleet_update_inner"}}
 {{$page := .Page}}
-<div id="fleet-update-panel" class="mt-5"
+<div id="fleet-update-panel" class="mt-5"{{if and $page.Active (eq $page.Active.Status "running")}}
      hx-get="{{$page.PollURL}}"
      hx-trigger="every 3s [document.visibilityState==='visible']"
      hx-select="#fleet-update-panel"
-     hx-swap="outerHTML">
+     hx-swap="outerHTML"{{end}}>
 
 {{if and $page.Active (eq $page.Active.Status "running")}}
 
@@ -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,34 +144,36 @@
       <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]">
+    <div class="mt-3 flex flex-wrap gap-2">
-      {{range $page.OutOfDateHosts}}
+      <input id="fleet-filter-name" class="field text-[12px]" placeholder="Filter name or tag" oninput="fleetFilter()">
-        <li class="flex items-center gap-3">
+      <input id="fleet-filter-version" class="field text-[12px] mono" placeholder="Version" oninput="fleetFilter()">
-          <span class="dot dot-online"></span>
+      <select id="fleet-filter-state" class="field text-[12px]" onchange="fleetFilter()">
-          <span class="mono text-ink">{{.Name}}</span>
+        <option value="all">All states</option><option value="eligible">Eligible</option><option value="excluded">Excluded</option>
-          <span class="mono text-ink-mute">{{if .AgentVersion}}{{.AgentVersion}}{{else}}—{{end}} → {{$page.TargetVersion}}</span>
+      </select>
-        </li>
+      <button type="button" class="btn" onclick="fleetSelectVisible(true)">Select visible eligible</button>
-      {{end}}
+      <button type="button" class="btn" onclick="fleetSelectVisible(false)">Clear selection</button>
-    </ul>
+    </div>
-
+    <div class="panel mt-3 rounded-[7px] overflow-hidden">
-    <form id="fleet-update-start-form" class="mt-4 flex items-center gap-3"
+      {{range $page.Candidates}}
-          hx-post="/api/fleet/update"
+      <label class="fleet-candidate grid items-center px-3 py-2 hairline text-[12px]"
-          hx-headers='{"Content-Type":"application/json"}'
+             data-name="{{.Host.Name}} {{range .Host.Tags}}{{.}} {{end}}" data-version="{{.Host.AgentVersion}}" data-state="{{if .Eligible}}eligible{{else}}excluded{{end}}"
-          hx-vals='{}'
+             style="grid-template-columns: 28px 1.4fr .8fr .8fr 1.2fr;gap:12px">
-          hx-swap="none"
+        <input class="fleet-host" type="checkbox" value="{{.Host.ID}}" {{if .Eligible}}checked onchange="fleetReview()"{{else}}disabled{{end}}>
-          hx-on::after-request="if(event.detail.successful) location.reload()">
+        <span class="mono">{{.Host.Name}}</span>
-      <label class="text-[11.5px] text-ink-mute">
+        <span class="mono text-ink-mute">{{if .Host.AgentVersion}}{{.Host.AgentVersion}}{{else}}unknown{{end}}</span>
-        Type the count
+        <span class="mono text-ink-mute">{{$page.TargetVersion}}</span>
-        <span class="mono text-ink-mid">({{len $page.OutOfDateHosts}})</span>
+        <span class="text-ink-mute">{{if .Eligible}}eligible{{else}}{{.Reason}}{{end}}</span>
-        to enable Start:
       </label>
-      <input type="text" id="fleet-update-confirm" class="field mono text-[12.5px]"
+      {{end}}
-             style="width: 80px; padding: 5px 8px;"
+    </div>
-             oninput="document.getElementById('fleet-update-start-btn').disabled = (this.value !== '{{len $page.OutOfDateHosts}}');"
+    <div class="mt-4 text-[12px] text-ink-mute">
-             autocomplete="off" />
+      <span id="fleet-selected-count">{{len $page.OutOfDateHosts}}</span> selected · sequential, halts on first failure · worst-case
-      <button type="submit" id="fleet-update-start-btn" class="btn btn-amber" disabled>
+      <span id="fleet-timeout" class="mono">{{len $page.OutOfDateHosts}} × 95s</span>
-        Start fleet update
+    </div>
-      </button>
+    <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>
+      <button id="fleet-update-start-btn" class="btn btn-amber">Start agent update</button>
+      <span id="fleet-start-error" class="text-bad text-[12px]"></span>
     </form>
   {{end}}
 </div>