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

CHANGELOG.md

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

cmd/agent/cancel_test.go

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

cmd/agent/main.go

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

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/alert/engine.go

@@ -61,9 +61,20 @@ type Engine struct {
 	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 {

internal/alert/rules.go

@@ -60,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,
@@ -85,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
 }
 
@@ -99,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
 }
 
@@ -164,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

@@ -25,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",

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

internal/server/http/job_abandon.go

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

internal/server/http/job_abandon_test.go

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

internal/server/http/jobs.go

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

internal/server/http/jobs_test.go

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

internal/server/http/p2r01_test.go

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

internal/server/http/server.go

@@ -261,6 +261,8 @@ 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)
 
@@ -298,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/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/ws/handler.go

@@ -357,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)

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

@@ -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/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")}}
 
@@ -172,20 +172,9 @@
     </div>
     <form class="mt-3 flex items-center gap-3" onsubmit="return fleetStart(event)">
 	  <label class="text-[11.5px] text-ink-mute"><input id="fleet-canary" type="checkbox" checked> Pause after first host for canary review</label>
-      <label class="text-[11.5px] text-ink-mute">Type selected count to confirm:</label>
+      <button id="fleet-update-start-btn" class="btn btn-amber">Start agent update</button>
-      <input id="fleet-update-confirm" class="field mono text-[12.5px]" style="width:80px;padding:5px 8px" oninput="fleetReview()" autocomplete="off">
-      <button id="fleet-update-start-btn" class="btn btn-amber" disabled>Start agent update</button>
       <span id="fleet-start-error" class="text-bad text-[12px]"></span>
     </form>
-    <script>
-      function fleetBoxes(){return Array.from(document.querySelectorAll('.fleet-host:not(:disabled)'))}
-      function fleetReview(){const n=fleetBoxes().filter(x=>x.checked).length;document.getElementById('fleet-selected-count').textContent=n;document.getElementById('fleet-timeout').textContent=n+' × 95s';document.getElementById('fleet-update-start-btn').disabled=n===0||document.getElementById('fleet-update-confirm').value!==String(n)}
-      function fleetFilter(){const q=document.getElementById('fleet-filter-name').value.toLowerCase(),v=document.getElementById('fleet-filter-version').value.toLowerCase(),s=document.getElementById('fleet-filter-state').value;document.querySelectorAll('.fleet-candidate').forEach(r=>r.hidden=!(r.dataset.name.toLowerCase().includes(q)&&r.dataset.version.toLowerCase().includes(v)&&(s==='all'||r.dataset.state===s)))}
-      function fleetSelectVisible(on){document.querySelectorAll('.fleet-candidate:not([hidden]) .fleet-host:not(:disabled)').forEach(x=>x.checked=on);fleetReview()}
-	  function fleetSelectOnly(id){fleetBoxes().forEach(x=>x.checked=x.value===id);fleetReview()}
-	  async function fleetStart(e){e.preventDefault();const ids=fleetBoxes().filter(x=>x.checked).map(x=>x.value),out=document.getElementById('fleet-start-error');out.textContent='';const res=await fetch('/api/fleet/update',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({host_ids:ids,canary_first:document.getElementById('fleet-canary').checked})});if(res.ok){location.reload();return false}const body=await res.json();out.textContent=body.message||body.code||'Unable to start';return false}
-      fleetReview()
-    </script>
   {{end}}
 </div>
 {{end}}