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

The previous wording told operators to leave the slot blank and then
warned that doing so disables prune, which is contradictory. Prune is
gated on the admin slot for every backend, on both paths:
scheduled (maintenance_dispatch.go:52) skips silently, manual
(repo_ops.go:39) returns 400 admin_creds_required.

Also reworded the opening line, which still said 'only needed for
rest-server repos' while the new text asks other backends to fill it in.

.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,83 @@ 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
+
+- Prevented agents from panicking when a WebSocket connection ends. The
+  WebSocket library transfers ownership of a successful upgrade stream to
+  the connection and leaves the HTTP response body nil; attempting to close
+  that body caused agents to restart and left forget jobs permanently stuck
+  in `running`. ([#37])
+- Manual forget jobs now receive the same per-source-group retention policies
+  as scheduled forget jobs. The API also supports a validated `--dry-run`
+  option through the complete server-to-agent-to-restic path, and rejects
+  hosts without configured retention before creating a job. ([#36])
+- Corrected the admin-credentials help text to reflect that forget uses normal
+  append-only credentials and blank admin credentials do not provide a
+  fallback for prune. ([#34])
+
 ## [1.1.0] - 2026-06-15
 
 ### Added
@@ -123,5 +200,22 @@ 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.0.0...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
+
+[#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

CONTRIBUTING.md

@@ -2,7 +2,8 @@
 
 Thanks for your interest in restic-manager. This document covers how
 to set up a development environment, the conventions the project
-follows, and how patches make it from your machine into `main`.
+follows, and how to contribute through issues as well as patches that
+make it from your machine into `main`.
 
 ## Project status and scope
 
@@ -108,6 +109,32 @@ admin user.
 
 ## Workflow
 
+### Opening an issue
+
+Issues are contributions too. Use them to report a bug, suggest a
+feature, improve the documentation, or start a design discussion even
+if you do not plan to submit a patch.
+
+Before opening one, search the existing issues and check `tasks.md` to
+see whether the topic is already tracked. Then choose the closest issue
+template:
+
+- [Bug report](./.gitea/issue_template/bug_report.md) for behaviour that
+  does not match the documentation or expected operation.
+- [Feature request](./.gitea/issue_template/feature_request.md) for a new
+  capability or a change to existing behaviour.
+
+Give the issue a specific title, keep it to one problem or proposal,
+and complete the relevant template fields. If no template is an exact
+fit, open a regular issue and explain the context, desired outcome, and
+any alternatives you have considered. Maintainers may ask follow-up
+questions or close requests that duplicate existing work or fall
+outside the project's scope.
+
+Security-sensitive reports are the exception: follow the
+[SECURITY.md](./SECURITY.md) disclosure process and do not open a public
+issue.
+
 ### Before opening a PR
 
 1. **Open an issue first** for non-trivial changes. The design is
@@ -136,25 +163,24 @@ The PR template asks for:
 
 ### Reporting bugs
 
-Open an issue with:
+Use the bug report issue template and include:
 
 - restic-manager version (`server --version`) and agent version.
 - restic version on the affected host.
 - Steps to reproduce.
 - Server and agent logs (sanitise any tokens before pasting).
 
-Security-sensitive bugs go through the [SECURITY.md](./SECURITY.md)
+For security-sensitive bugs, use the private disclosure process noted
-disclosure path instead — please don't open a public issue for
+above.
-them.
 
 ### Suggesting features
 
-Open an issue describing the use case (not just the proposed
+Use the feature request issue template and describe the use case (not
-solution). The roadmap in `tasks.md` shows where the project is
+just the proposed solution). The roadmap in `tasks.md` shows where the
-heading; if the suggestion fits a future phase we'll wire it in
+project is heading; if the suggestion fits a future phase we'll wire it
-there. If it falls outside the project's scope (multi-tenancy, SaaS,
+in there. If it falls outside the project's scope (multi-tenancy, SaaS,
-non-restic backends — see `spec.md` §2 non-goals) we'll say so
+non-restic backends — see `spec.md` §2 non-goals) we'll say so early to
-early to save your time.
+save your time.
 
 ## Code of conduct
 

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)
@@ -601,6 +632,11 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc
 			failJob(p, tx, "forget: command.run carried no forget_groups (server didn't populate them)")
 			return fmt.Errorf("forget: command.run carried no forget_groups (server didn't populate them)")
 		}
+		if len(p.Args) > 1 || (len(p.Args) == 1 && p.Args[0] != "--dry-run") {
+			failJob(p, tx, "forget: command.run carried unsupported arguments")
+			return fmt.Errorf("forget: command.run carried unsupported arguments")
+		}
+		dryRun := len(p.Args) == 1
 		groups := make([]restic.ForgetGroup, 0, len(p.ForgetGroups))
 		for _, g := range p.ForgetGroups {
 			groups = append(groups, restic.ForgetGroup{
@@ -615,9 +651,9 @@ func (d *dispatcher) runJob(ctx context.Context, p api.CommandRunPayload, tx wsc
 				},
 			})
 		}
-		slog.Info("agent: accepting forget job", "job_id", p.JobID, "groups", len(groups))
+		slog.Info("agent: accepting forget job", "job_id", p.JobID, "groups", len(groups), "dry_run", dryRun)
 		spawn("forget", func(jobCtx context.Context) error {
-			return r.RunForget(jobCtx, p.JobID, groups)
+			return r.RunForget(jobCtx, p.JobID, groups, dryRun)
 		})
 	case api.JobPrune:
 		// Prune may require admin creds (delete authority on rest-server).

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)
@@ -274,16 +276,14 @@ func (r *Runner) RunInit(ctx context.Context, jobID string) error {
 // snapshot projection (forget rewrites the snapshot index — the
 // host's snapshot list shrinks). Snapshot refresh runs once after
 // every group completes, not per-group.
-func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.ForgetGroup) error {
+func (r *Runner) RunForget(ctx context.Context, jobID string, groups []restic.ForgetGroup, dryRun bool) error {
 	startedAt := time.Now().UTC()
 	r.sendStarted(jobID, api.JobForget, startedAt)
 
 	env := r.resticEnv()
 	var seq atomic.Int64
-	err := env.RunForget(ctx, groups, r.streamHandler(jobID, &seq))
+	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) {
@@ -398,9 +409,13 @@ esac
 		Tag:    "documents",
 		Policy: restic.ForgetPolicy{KeepLast: &keepLast},
 	}}
-	if err := r.RunForget(context.Background(), "job-forget", groups); err != nil {
+	if err := r.RunForget(context.Background(), "job-forget", groups, false); err != nil {
 		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

@@ -103,15 +103,15 @@ func connectOnce(ctx context.Context, cfg Config, handle Handler) error {
 	}
 
 	dialCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
-	conn, res, err := websocket.Dial(dialCtx, wsURL, dialOpts)
+	conn, _, err := websocket.Dial(dialCtx, wsURL, dialOpts) //nolint:bodyclose // successful upgrades have a nil response body owned by conn
 	cancel()
 	if err != nil {
 		return fmt.Errorf("dial: %w", err)
 	}
-	// websocket.Dial returns the upgrade response separately from the
+	conn.SetReadLimit(api.MaxWebSocketMessageBytes)
-	// conn. Body is empty on a successful upgrade but Go's net/http
+	// On a successful upgrade coder/websocket transfers ownership of the
-	// still expects it closed to release the connection.
+	// response stream to conn and deliberately sets res.Body to nil. Closing
-	defer func() { _ = res.Body.Close() }()
+	// the connection below releases that stream.
 	defer conn.CloseNow() //nolint:errcheck
 
 	// Send hello.

internal/agent/wsclient/client_test.go

@@ -0,0 +1,103 @@
+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) {
+	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
+
+		// Wait for the agent hello so Dial and the first client write have both
+		// completed before ending the connection normally.
+		if _, _, err := conn.Read(r.Context()); err != nil {
+			serverErr <- err
+			return
+		}
+		serverErr <- conn.Close(websocket.StatusNormalClosure, "test complete")
+	}))
+	defer srv.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	err := connectOnce(ctx, Config{
+		ServerURL:       srv.URL,
+		AgentToken:      "test-token",
+		HeartbeatPeriod: time.Hour,
+	}, nil)
+	if err == nil {
+		t.Fatal("connectOnce returned nil after server disconnected")
+	}
+	if err := <-serverErr; err != nil {
+		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,7 +86,12 @@ func NewEngine(st *store.Store, hub *notification.Hub) *Engine {
 		hostUp:            make(chan string, 32),
 		agentOfflineFloor: 15 * time.Minute,
 		tickPeriod:        60 * time.Second,
-		done:              make(chan struct{}),
+		stuckThresholds: map[string]time.Duration{
+			"backup":  longStuckJobThreshold,
+			"restore": longStuckJobThreshold,
+			"check":   longStuckJobThreshold,
+		},
+		done: make(chan struct{}),
 	}
 }
 
@@ -125,6 +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"
@@ -29,12 +36,13 @@ const (
 
 // Server → agent message types.
 const (
-	MsgCommandRun    MessageType = "command.run"
+	MsgCommandRun       MessageType = "command.run"
-	MsgCommandCancel MessageType = "command.cancel"
+	MsgCommandCancel    MessageType = "command.cancel"
-	MsgScheduleSet   MessageType = "schedule.set"
+	MsgScheduleSet      MessageType = "schedule.set"
-	MsgConfigUpdate  MessageType = "config.update"
+	MsgConfigUpdate     MessageType = "config.update"
-	MsgCommandUpdate MessageType = "command.update"
+	MsgCommandUpdate    MessageType = "command.update"
-	MsgTreeList      MessageType = "tree.list" // sync RPC: list a snapshot's children
+	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/restic/runner.go

@@ -322,7 +322,7 @@ type ForgetGroup struct {
 // any keep-* would delete every snapshot in the tagged set).
 // Returns the first error encountered, or nil when every group runs
 // to a clean exit.
-func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, handle LineHandler) error {
+func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, dryRun bool, handle LineHandler) error {
 	if len(groups) == 0 {
 		return fmt.Errorf("restic forget: refusing to run with no groups (would be a no-op)")
 	}
@@ -332,6 +332,9 @@ func (e Env) RunForget(ctx context.Context, groups []ForgetGroup, handle LineHan
 		}
 		args := []string{"forget", "--json", "--tag", g.Tag}
 		args = append(args, g.Policy.args()...)
+		if dryRun {
+			args = append(args, "--dry-run")
+		}
 		cmd := e.resticCmd(ctx, args...)
 		if err := runWithPump(cmd, handle); err != nil {
 			return err

internal/restic/runner_test.go

@@ -60,6 +60,26 @@ func TestRunPruneInvokesPrune(t *testing.T) {
 	t.Fatalf("expected 'prune' in captured output; got: %v", *lines)
 }
 
+func TestRunForgetDryRunArgument(t *testing.T) {
+	bin := setupScriptBin(t, `echo "$@"`)
+	env := Env{Bin: bin}
+	lines, h := captureLines()
+	keepLast := 1
+	groups := []ForgetGroup{{
+		Tag:    "documents",
+		Policy: ForgetPolicy{KeepLast: &keepLast},
+	}}
+	if err := env.RunForget(context.Background(), groups, true, h); err != nil {
+		t.Fatalf("RunForget: %v", err)
+	}
+	for _, line := range *lines {
+		if strings.Contains(line, "forget --json --tag documents --keep-last 1 --dry-run") {
+			return
+		}
+	}
+	t.Fatalf("expected forget invocation with --dry-run; got: %v", *lines)
+}
+
 // --- B2: RunCheck ---
 
 func TestRunCheckLockSniff(t *testing.T) {

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

@@ -65,10 +65,36 @@ func (s *Server) handleRunNow(w stdhttp.ResponseWriter, r *stdhttp.Request) {
 func (s *Server) dispatchJob(ctx context.Context, user *store.User,
 	hostID string, kind api.JobKind, args []string,
 ) (res runNowResponse, status int, code, msg string) {
-	return s.dispatchJobWithPayload(ctx, user, hostID, kind, nil, api.CommandRunPayload{
+	payload := api.CommandRunPayload{
 		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",
+				"forget accepts no arguments other than --dry-run"
+		}
+		var ok bool
+		var err error
+		payload, ok, err = s.buildForgetPayloadForHost(ctx, hostID)
+		if err != nil {
+			return res, stdhttp.StatusInternalServerError, "internal", ""
+		}
+		if !ok {
+			return res, stdhttp.StatusUnprocessableEntity, "no_retention_policy",
+				"host has no source groups with a retention policy"
+		}
+		payload.Args = args
+	}
+	return s.dispatchJobWithPayload(ctx, user, hostID, kind, nil, payload)
+}
+
+func validForgetArgs(args []string) bool {
+	return len(args) == 0 || (len(args) == 1 && args[0] == "--dry-run")
 }
 
 // dispatchJobWithPayload is dispatchJob's variant that lets callers

internal/server/http/jobs_test.go

@@ -0,0 +1,134 @@
+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/api"
+	"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)
+	hostID, token := enrolHostForWS(t, srv, st, "manual-forget-host")
+	seedInitJob(t, st, hostID)
+
+	keepDaily := 7
+	if err := st.CreateSourceGroup(context.Background(), &store.SourceGroup{
+		ID:              ulid.Make().String(),
+		HostID:          hostID,
+		Name:            "documents",
+		Includes:        []string{"/home/documents"},
+		RetentionPolicy: store.RetentionPolicy{KeepDaily: &keepDaily},
+	}); err != nil {
+		t.Fatalf("create source group: %v", err)
+	}
+
+	c := agentDial(t, srv, ts, hostID, token)
+	sendHello(t, c, "manual-forget-host")
+	_ = drainUntil(t, c, api.MsgScheduleSet)
+
+	body, err := json.Marshal(runNowRequest{Kind: api.JobForget, Args: []string{"--dry-run"}})
+	if err != nil {
+		t.Fatalf("marshal request: %v", err)
+	}
+	req, err := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/jobs", bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.AddCookie(loginAsAdmin(t, st))
+	res, err := stdhttp.DefaultClient.Do(req)
+	if err != nil {
+		t.Fatalf("post run-now: %v", err)
+	}
+	defer res.Body.Close()
+	if res.StatusCode != stdhttp.StatusAccepted {
+		t.Fatalf("status: got %d, want %d", res.StatusCode, stdhttp.StatusAccepted)
+	}
+
+	got := readNextCommandRun(t, c, time.Now().Add(2*time.Second))
+	if got == nil {
+		t.Fatal("no command.run received")
+	}
+	if len(got.Args) != 1 || got.Args[0] != "--dry-run" {
+		t.Fatalf("Args: got %q, want [--dry-run]", got.Args)
+	}
+	if len(got.ForgetGroups) != 1 {
+		t.Fatalf("ForgetGroups: got %d, want 1", len(got.ForgetGroups))
+	}
+	group := got.ForgetGroups[0]
+	if group.Tag != "documents" || group.Policy.KeepDaily == nil || *group.Policy.KeepDaily != 7 {
+		t.Fatalf("ForgetGroups[0]: got %+v", group)
+	}
+}
+
+func TestRunNowForgetRejectsHostWithoutRetention(t *testing.T) {
+	t.Parallel()
+	srv, ts, st := rawTestServer(t)
+	hostID, token := enrolHostForWS(t, srv, st, "no-manual-retention-host")
+	seedInitJob(t, st, hostID)
+
+	c := agentDial(t, srv, ts, hostID, token)
+	sendHello(t, c, "no-manual-retention-host")
+	_ = drainUntil(t, c, api.MsgScheduleSet)
+
+	body := []byte(`{"kind":"forget","args":["--dry-run"]}`)
+	req, err := stdhttp.NewRequest(stdhttp.MethodPost, ts.URL+"/api/hosts/"+hostID+"/jobs", bytes.NewReader(body))
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.AddCookie(loginAsAdmin(t, st))
+	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 %d", res.StatusCode, stdhttp.StatusUnprocessableEntity)
+	}
+
+	var jobs int
+	if err := st.DB().QueryRow(`SELECT COUNT(*) FROM jobs WHERE host_id = ? AND kind = 'forget'`, hostID).Scan(&jobs); err != nil {
+		t.Fatalf("count forget jobs: %v", err)
+	}
+	if jobs != 0 {
+		t.Fatalf("forget jobs: got %d, want 0", jobs)
+	}
+}

internal/server/http/maintenance_dispatch.go

@@ -37,7 +37,12 @@ func (s *Server) DispatchMaintenance(ctx context.Context, decisions []maintenanc
 		}
 		switch d.Kind {
 		case "forget":
-			payload, ok := s.buildForgetPayloadForHost(ctx, d.HostID)
+			payload, ok, err := s.buildForgetPayloadForHost(ctx, d.HostID)
+			if err != nil {
+				slog.Warn("maintenance: list source groups failed",
+					"host_id", d.HostID, "err", err)
+				continue
+			}
 			if !ok {
 				slog.Info("maintenance: forget skipped — no source groups with retention",
 					"host_id", d.HostID)
@@ -88,11 +93,10 @@ func (s *Server) DispatchMaintenance(ctx context.Context, decisions []maintenanc
 // that has a non-empty retention policy and builds a CommandRunPayload
 // with ForgetGroups populated. Returns ok=false if the host has no
 // such groups (the dispatcher then skips this kind).
-func (s *Server) buildForgetPayloadForHost(ctx context.Context, hostID string) (api.CommandRunPayload, bool) {
+func (s *Server) buildForgetPayloadForHost(ctx context.Context, hostID string) (api.CommandRunPayload, bool, error) {
 	groups, err := s.deps.Store.ListSourceGroupsByHost(ctx, hostID)
 	if err != nil {
-		slog.Warn("maintenance: list source groups failed", "host_id", hostID, "err", err)
+		return api.CommandRunPayload{}, false, err
-		return api.CommandRunPayload{}, false
 	}
 	fg := make([]api.ForgetGroup, 0, len(groups))
 	for _, g := range groups {
@@ -105,9 +109,9 @@ func (s *Server) buildForgetPayloadForHost(ctx context.Context, hostID string) (
 		})
 	}
 	if len(fg) == 0 {
-		return api.CommandRunPayload{}, false
+		return api.CommandRunPayload{}, false, nil
 	}
-	return api.CommandRunPayload{ForgetGroups: fg}, true
+	return api.CommandRunPayload{ForgetGroups: fg}, true, nil
 }
 
 func isEmptyRetention(p store.RetentionPolicy) bool {

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/pending_drain_test.go

@@ -512,11 +512,27 @@ func TestDrainPendingSerializesPerHost(t *testing.T) {
 	// Connect the agent so DrainPending can dispatch.
 	c := agentDial(t, srv, ts, hostID, token)
 	sendHello(t, c, "serialise-host")
-	// Drain the on-hello goroutine's pass first (no pending rows yet),
+	// Wait for the on-hello push to settle.
-	// then wait for the schedule.set so the connection is fully settled.
 	_ = drainUntil(t, c, api.MsgScheduleSet)
 
-	// Insert 5 pending rows now that the on-hello drain has already run.
+	// A real agent is always in a read loop. Keep this test client
+	// reading in the background for the rest of the test: without an
+	// active reader the server-side conn can be dropped under parallel
+	// load, which unregisters it from the hub and makes DrainPending
+	// no-op (conn == nil) — the historical source of this test's
+	// flakiness (it would observe 0 or a partial drain). The reader also
+	// consumes the command.run envelopes our drains emit.
+	readerCtx, stopReader := context.WithCancel(context.Background())
+	defer stopReader()
+	go func() {
+		for {
+			if _, _, err := c.Read(readerCtx); err != nil {
+				return
+			}
+		}
+	}()
+
+	// Insert 5 due pending rows.
 	now := time.Now().UTC()
 	for i := range 5 {
 		pid := ulid.Make().String()
@@ -533,7 +549,8 @@ func TestDrainPendingSerializesPerHost(t *testing.T) {
 		}
 	}
 
-	// Spawn 10 goroutines all calling DrainPending concurrently.
+	// Fire 10 concurrent DrainPending calls. The per-host mutex must
+	// ensure each row is dispatched at most once (no double-dispatch).
 	var wg sync.WaitGroup
 	for range 10 {
 		wg.Add(1)
@@ -544,24 +561,26 @@ func TestDrainPendingSerializesPerHost(t *testing.T) {
 	}
 	wg.Wait()
 
-	// Drain any envelopes the agent received so we don't block below.
+	// Drain to completion. The fire-and-forget on-hello DrainPending
-	// We read with short timeouts and stop when the connection goes quiet.
+	// shares the same per-host mutex and can hold it during the burst,
-	drainDeadline := time.Now().Add(500 * time.Millisecond)
+	// leaving rows for a later pass — exactly how production drains
-	for time.Now().Before(drainDeadline) {
+	// (repeatedly, via the 30s tick / on reconnect). Re-drain until the
-		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	// queue is empty; because every drain is still serialised, each row
-		_, _, err := c.Read(ctx)
+	// is dispatched at most once, so the exactly-5 job count below proves
-		cancel()
+	// there was no double-dispatch.
-		if err != nil {
+	deadline := time.Now().Add(5 * time.Second)
-			break
+	for countPendingForHost(t, st, hostID) > 0 && time.Now().Before(deadline) {
-		}
+		srv.DrainPending(context.Background(), hostID)
+		time.Sleep(10 * time.Millisecond)
 	}
 
-	// All 5 pending rows must be gone.
+	// All 5 pending rows must be drained.
 	if n := countPendingForHost(t, st, hostID); n != 0 {
-		t.Errorf("pending rows after concurrent drain: got %d, want 0", n)
+		t.Errorf("pending rows after drain-to-completion: got %d, want 0", n)
 	}
 
-	// Exactly 5 backup job rows (one per pending row), not 10+ from a race.
+	// Exactly 5 backup job rows (one per pending row) — never more, which
+	// would mean the per-host mutex failed to prevent double-dispatch.
 	var n int
 	_ = st.DB().QueryRow(
 		`SELECT COUNT(*) FROM jobs WHERE host_id = ? AND kind = 'backup' AND actor_kind = 'schedule'`,

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

@@ -82,10 +82,14 @@
         <div class="text-[12px] text-ok mb-3 mono">✓ saved</div>
       {{end}}
       <p class="text-[12.5px] text-ink-mid leading-[1.6] mb-4 max-w-[640px]">
-        Only needed for rest-server repos that distinguish an append-only
+        Required for prune. On rest-server repos this is the
-        user (everyday backups) from a delete-capable user (prune /
+        delete-capable user, as distinct from the append-only user used
-        forget). For S3 / B2 / SFTP / local, leave this blank — the
+        for everyday backups. Note that <strong>forget</strong> always
-        everyday repo credentials handle prune too.
+        runs with the everyday repo credentials, so those must have
+        delete authority. For S3 / B2 / SFTP / local, enter the same
+        delete-capable repository credentials here if you want prune
+        enabled. <strong>Prune is skipped when admin credentials are
+        unset</strong>, on any backend.
       </p>
       <div class="grid grid-cols-2 gap-4">
         <div>

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>