felhom-agent/internal/localapi/server.go

package localapi

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
	"gitea.dooplex.hu/admin/felhom-agent/internal/storage"
)

// GuestAPI is the narrow Proxmox surface the local API needs. Satisfied by *proxmox.Client.
// Every method here is invoked ONLY with the VMID resolved from the caller's token — never a
// caller-supplied id — so the proxmox op is structurally self-scoped.
type GuestAPI interface {
	GuestConfig(ctx context.Context, vmid int) (proxmox.GuestConfig, error)
	Snapshot(ctx context.Context, vmid int, snapname, description string) (string, error)
	Rollback(ctx context.Context, vmid int, snapname string) (string, error)
	WaitTask(ctx context.Context, upid string, opts proxmox.WaitOptions) (proxmox.TaskStatus, error)
}

// BackupService enqueues a vzdump/PBS backup of a guest. Satisfied by *backup.BackupRunner.
// BackupWithSnapshotHook (8B.2) invokes onSnapshot once mid-backup when the storage snapshot is
// taken (snapshot mode only) so the controller can resume its app early; in stop mode it is never
// called.
type BackupService interface {
	BackupWithSnapshotHook(ctx context.Context, vmid int, onSnapshot func()) (hub.Backup, error)
}

// BackupStore records + reads the latest backup/restore-test state. Satisfied by *backup.Store.
type BackupStore interface {
	RecordBackup(hub.Backup)
	Backups(ctx context.Context) []hub.Backup
	RestoreTests(ctx context.Context) []hub.RestoreTest
}

// StorageView yields the host's observed storage targets (for mapping a mount's storage id →
// fast/slow class). Satisfied by *storage.Observer.
type StorageView interface {
	Observe(ctx context.Context) ([]hub.StorageTarget, error)
}

// TokenAuthority resolves a presented bearer token to its guest VMID. Satisfied by *TokenStore.
type TokenAuthority interface {
	Lookup(token string) (int, bool)
}

// HostMetricsProvider does a FRESH host-metrics collect (cpu%/mem/load/uptime/cpu-temp) for
// GET /host/metrics (slice 9). Satisfied by *hub.Collector (which reuses the slice-4 collector —
// no duplicate collection). Optional: when nil, /host/metrics reports "not configured".
type HostMetricsProvider interface {
	HostMetricsNow(ctx context.Context) (hub.HostMetrics, error)
}

// Options configures a Server.
type Options struct {
	ListenAddr string // bridge IP:port
	Cert       tls.Certificate
	Guests     GuestAPI
	Backups    BackupService
	Store      BackupStore
	Storage    StorageView
	// DriveTargets (Impl-2a, optional) yields registry+units-sourced drives for the /disks view, so a
	// drive with NO PVE storage still appears. Unioned with Storage.Observe (deduped by mount path).
	DriveTargets storage.KnownTargets
	Tokens     TokenAuthority
	// BackupCadence is the per-guest backup interval driving GET /backup/due (slice 8B). A guest
	// is "due" when no successful backup is recorded OR the newest one is older than this. 0 → a
	// safe default (24h). The hub-served policy is slice 10; this is the agent-local cadence.
	BackupCadence time.Duration
	// Disk management (slice 8C) — OPTIONAL. When Disks + DiskGate are set, the /disks endpoints
	// are served; otherwise they report "not configured". DiskGate authorizes the destructive
	// (data-bearing) format path; Guests lists guests for the eject dependent-warning.
	Disks    DiskOps
	DiskGate StorageGate
	Guests2  GuestLister
	// GuestAttach binds an enrolled user-data drive's felhom-data namespace into the guest (slice 10
	// P2, Model A). OPTIONAL — when nil, POST /disks/guest-attach reports "not configured".
	GuestAttach GuestAttacher
	// NetStorage is the privileged network-mount (NAS) surface (Part A1). OPTIONAL — when nil, the
	// /netstorage endpoints report "not configured". Satisfied by *storage.SudoHostOps.
	NetStorage NetworkStorageOps
	// SmbCredsDir is where the agent writes the 0600 SMB credentials files (out-of-band). "" →
	// /var/lib/felhom-agent/smb-creds.
	SmbCredsDir string
	// ControllerSwap runs guest commands (pct exec) for the agentic controller-update swap (Phase 1).
	// OPTIONAL — when nil, POST /controller/swap reports "not configured". Satisfied by *GuestBinder.
	ControllerSwap GuestExecutor
	// StaleLock recovers a guest left with a stale vzdump lock by a reboot-during-backup (F2-b), run at
	// startup by RecoverStaleLockedGuests. OPTIONAL — when nil, the recovery is a no-op.
	StaleLock StaleLockController
	// ControllerSwapStateDir holds the per-guest swap state file (crash-safety). "" → /var/lib/felhom-agent.
	ControllerSwapStateDir string
	// Intent records drive enroll/eject intent for the self-heal watchdog (slice 10 P3). OPTIONAL —
	// when nil, no intent is recorded (self-heal runs ungated).
	Intent IntentRecorder
	// GuestBinds persists which user-data drives (by durable-id) are enrolled into each guest, so the
	// startup re-assert (ReassertGuestBinds) can restore a bind that a re-provision dropped (F9).
	// OPTIONAL — when nil, guest binds are not recorded and the startup re-assert is a no-op.
	GuestBinds *GuestBindStore
	// FormatJobs persists the in-flight/last disk-format job so mkfs runs detached from the request
	// (F20-BUG3: a request deadline can't kill it) and survives an agent restart (RecoverFormatJob).
	// OPTIONAL — when nil, formats still run detached but are not persisted/recovered.
	FormatJobs *FormatJobStore
	// HostReader is the root-free host topology reader used to classify a device/mount's protection
	// ROLE (it backs SystemDisks for the eject role-gate + the /disks role hints). OPTIONAL — when nil
	// it defaults to the production *storage.ProcHostReader. Injectable so the role-gate is testable.
	HostReader storage.HostReader
	// HostMetrics serves GET /host/metrics (slice 9) — host-wide health (cpu%/mem/load/uptime/
	// cpu-temp) + per-storage capacity, host-wide and token-authed (one-customer-per-host). When
	// nil the endpoint reports "not configured" (host still reports/reconciles).
	HostMetrics HostMetricsProvider
	// HostID is this agent's host id — surfaced in a data-bearing-format pending-op so the operator
	// signs an op bound to THIS host (slice 10B anti-retarget). Optional (only used for the hint).
	HostID string
	Logger *slog.Logger
}

// defaultBackupCadence is the fallback /backup/due window when none is configured.
const defaultBackupCadence = 24 * time.Hour

// Backup phase vocabulary reported by GET /backup/status (slice 8B). The 8B.2 fast-follow adds a
// `snapshotted` phase (vzdump --mode snapshot) so the controller can unquiesce at snapshot-taken.
const (
	PhaseIdle        = "idle"
	PhaseRunning     = "running"
	PhaseSnapshotted = "snapshotted" // 8B.2: storage snapshot taken — app may resume; backup continues
	PhaseDone        = "done"
	PhaseFailed      = "failed"
)

// backupJob is the in-flight/last backup job for one guest (drives /backup/status phases).
type backupJob struct {
	JobID      string
	Phase      string
	StartedAt  time.Time
	FinishedAt time.Time
	Archive    string
	Error      string
}

// Server is the per-guest local API (doc 03 §6). It serves the agent's pinned self-signed leaf
// and authorizes every request against the token's guest only.
type Server struct {
	addr    string
	cert    tls.Certificate
	guests  GuestAPI
	backups BackupService
	store   BackupStore
	storage      StorageView
	driveTargets storage.KnownTargets // Impl-2a: registry+units drives for /disks (optional)
	tokens  TokenAuthority
	cadence time.Duration
	logger  *slog.Logger
	now     func() time.Time

	disks       DiskOps            // slice 8C (optional)
	diskGate    StorageGate        // slice 8C (optional)
	guestList   GuestLister        // slice 8C (optional)
	guestAttach GuestAttacher      // slice 10 P2 (optional)
	netStorage  NetworkStorageOps  // Part A1: NAS network mounts (optional)
	netMountRoot string            // the user-data namespace root for the network-mount role gate
	smbCredsDir  string            // where SMB creds files are written (out-of-band, 0600)
	intent      IntentRecorder     // slice 10 P3 (optional)
	guestBinds  *GuestBindStore    // F9 startup bind re-assert record (optional)
	formatJobs  *FormatJobStore    // F20-BUG3 detached-format job record (optional)
	staleLock   StaleLockController // F2-b startup stale-lock recovery (optional)
	host        storage.HostReader // role classification source (optional; defaults to ProcHostReader)

	hostMetrics HostMetricsProvider // slice 9 (optional)
	hostID      string              // slice 10B: for the data-bearing-format pending-op hint

	// reresolveWipe performs the [AGENT-001] anti-retarget re-resolution before an
	// inline customer-confirmed wipe (durable id → current device, re-derive+match,
	// re-inspect). Defaults to s.reresolveDurableForWipe (real storage funcs); tests
	// override it to avoid touching real /dev.
	reresolveWipe func(ctx context.Context, durableID string) (string, error)

	// reresolveBlank is the BLANK-format sibling (audit D3): same anti-retarget
	// sequence, but requires the re-inspected device to STILL be blank. Defaults
	// to s.reresolveDurableForBlankFormat; tests override it.
	reresolveBlank func(ctx context.Context, durableID string) (string, error)

	// deviceDurableID derives the WIPE-binding durable id of a block device (the byid:/byuuid: scheme
	// the wipe gate resolves against). F20-BUG2: BOTH the /disks list (DiskInfo.WipeDurableID) and the
	// format gate use this single seam, so the id the customer copies from the list is exactly the id
	// the gate accepts (no more uuid: vs byid: binding_mismatch). Defaults to storage.DeviceDurableID;
	// tests override it. (DiskInfo.DurableID stays the uuid: storage id — that one feeds /disks/assign.)
	deviceDurableID func(device string) (string, error)

	// boundCheck reports whether a stable guest path has felhom-data bound under it (intermediary model).
	// Optional — nil defaults to the real host mount-table read (isHostMountpoint); tests inject a fake.
	boundCheck func(string) bool

	jobsMu sync.Mutex
	jobs   map[int]*backupJob // per-guest backup job state (slice 8B)

	// agentic controller update (Phase 1): the swapper + per-guest single-flight gate.
	swap         *ControllerSwapper
	swapMu       sync.Mutex
	swapInFlight map[int]bool

	baseCtx context.Context // for fire-and-forget backups; set in Run
}

// NewServer builds a Server. It does not bind a socket until Run.
func NewServer(o Options) (*Server, error) {
	if o.ListenAddr == "" {
		return nil, fmt.Errorf("localapi: listen addr required")
	}
	if o.Guests == nil || o.Backups == nil || o.Store == nil || o.Storage == nil || o.Tokens == nil {
		return nil, fmt.Errorf("localapi: all dependencies (guests, backups, store, storage, tokens) are required")
	}
	if o.Logger == nil {
		o.Logger = slog.Default()
	}
	cadence := o.BackupCadence
	if cadence <= 0 {
		cadence = defaultBackupCadence
	}
	s := &Server{
		addr:         o.ListenAddr,
		cert:         o.Cert,
		guests:       o.Guests,
		backups:      o.Backups,
		store:        o.Store,
		storage:      o.Storage,
		driveTargets: o.DriveTargets,
		tokens:       o.Tokens,
		cadence:      cadence,
		logger:       o.Logger,
		now:          func() time.Time { return time.Now().UTC() },
		disks:        o.Disks,
		diskGate:     o.DiskGate,
		guestList:    o.Guests2,
		guestAttach:  o.GuestAttach,
		netStorage:   o.NetStorage,
		netMountRoot: storage.NetworkMountRoot,
		smbCredsDir:  o.SmbCredsDir,
		intent:       o.Intent,
		guestBinds:   o.GuestBinds,
		formatJobs:   o.FormatJobs,
		staleLock:    o.StaleLock,
		host:         o.HostReader,
		hostMetrics:  o.HostMetrics,
		hostID:       o.HostID,
		jobs:         map[int]*backupJob{},
		swapInFlight: map[int]bool{},
	}
	s.reresolveWipe = s.reresolveDurableForWipe
	s.reresolveBlank = s.reresolveDurableForBlankFormat
	s.deviceDurableID = storage.DeviceDurableID
	if o.ControllerSwap != nil {
		s.swap = NewControllerSwapper(o.ControllerSwap, o.ControllerSwapStateDir, o.Logger)
	}
	return s, nil
}

// Handler builds the routed mux (exposed for tests via httptest).
func (s *Server) Handler() http.Handler {
	mux := http.NewServeMux()
	mux.HandleFunc("GET /storage", s.withGuest(s.handleStorage))
	mux.HandleFunc("POST /snapshot", s.withGuest(s.handleSnapshot))
	mux.HandleFunc("POST /rollback", s.withGuest(s.handleRollback))
	mux.HandleFunc("POST /backup", s.withGuest(s.handleBackup))
	mux.HandleFunc("GET /backup/due", s.withGuest(s.handleBackupDue))
	mux.HandleFunc("GET /backup/status", s.withGuest(s.handleBackupStatus))
	mux.HandleFunc("GET /restore-test/status", s.withGuest(s.handleRestoreTestStatus))
	// Host metrics (slice 9): host-wide health + per-storage capacity for the customer's monitoring
	// view. Host-wide, token-authed, fresh (a live collect, not the 15-min hub snapshot).
	mux.HandleFunc("GET /host/metrics", s.withGuest(s.handleHostMetrics))
	// Disk management (slice 8C) — self-scoped; format routes through the data-bearing classifier+gate.
	mux.HandleFunc("GET /disks", s.withGuest(s.handleDisks))
	mux.HandleFunc("GET /disks/candidates", s.withGuest(s.handleDiskCandidates))
	mux.HandleFunc("POST /disks/assign", s.withGuest(s.handleDiskAssign))
	mux.HandleFunc("POST /disks/eject", s.withGuest(s.handleDiskEject))
	mux.HandleFunc("POST /disks/decommission", s.withGuest(s.handleDiskDecommission))
	mux.HandleFunc("POST /disks/format", s.withGuest(s.handleDiskFormat))
	mux.HandleFunc("GET /disks/format/status", s.withGuest(s.handleDiskFormatStatus))
	// Guest data-drive passthrough (slice 10 P2): bind an enrolled drive's felhom-data namespace in.
	mux.HandleFunc("POST /disks/guest-attach", s.withGuest(s.handleDiskGuestAttach))
	// Guest reboot (slice 10 P2 activation): user-triggered restart to activate pending drive binds.
	mux.HandleFunc("POST /guest/reboot", s.withGuest(s.handleGuestReboot))

	// Network storage (NAS) — Part A1: mount/list/remove a bulk-media NAS share host-side (automount
	// idle-unmount; +100000 uid recipe). A distinct class from a drive — no enroll/eject/wipe.
	mux.HandleFunc("POST /netstorage/add", s.withGuest(s.handleNetStorageAdd))
	mux.HandleFunc("GET /netstorage", s.withGuest(s.handleNetStorageList))
	mux.HandleFunc("POST /netstorage/remove", s.withGuest(s.handleNetStorageRemove))

	// agentic controller update (Phase 1): in-guest image swap + rollback, owned by the agent.
	mux.HandleFunc("POST /controller/swap", s.withGuest(s.handleControllerSwap))
	mux.HandleFunc("GET /controller/swap/status", s.withGuest(s.handleControllerSwapStatus))
	return mux
}

// Run binds the bridge socket, serves TLS, and shuts down gracefully on ctx cancellation. It
// returns nil on a clean shutdown (mirrors the other daemon loops' ctx-cancel contract).
func (s *Server) Run(ctx context.Context) error {
	s.baseCtx = ctx
	srv := &http.Server{
		Addr:    s.addr,
		Handler: s.Handler(),
		TLSConfig: &tls.Config{
			Certificates: []tls.Certificate{s.cert},
			MinVersion:   tls.VersionTLS12,
		},
		ReadHeaderTimeout: 10 * time.Second,
	}
	ln, err := net.Listen("tcp", s.addr)
	if err != nil {
		return fmt.Errorf("localapi: bind %s: %w", s.addr, err)
	}
	s.logger.Info("local-api server listening", "addr", s.addr)

	errc := make(chan error, 1)
	go func() { errc <- srv.ServeTLS(ln, "", "") }()

	select {
	case <-ctx.Done():
		shutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		_ = srv.Shutdown(shutCtx)
		return nil
	case err := <-errc:
		if errors.Is(err, http.ErrServerClosed) {
			return nil
		}
		return err
	}
}

// ---- auth + self-scoping ----------------------------------------------------------------

// withGuest wraps a handler with bearer auth and self-scoping: it resolves the token → VMID
// (401 on absent/unknown), and refuses any explicit `vmid` that disagrees with the token's
// guest (403, cross-guest). The wrapped handler is invoked ONLY with the token's VMID, so a
// proxmox op is never issued for another guest. Self-scoping is by the token→guest map; a
// caller-supplied id is only ever a consistency check, never the authority.
func (s *Server) withGuest(fn func(w http.ResponseWriter, r *http.Request, vmid int)) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		tok := bearer(r)
		if tok == "" {
			writeErr(w, http.StatusUnauthorized, "missing bearer token")
			return
		}
		vmid, ok := s.tokens.Lookup(tok)
		if !ok {
			writeErr(w, http.StatusUnauthorized, "unknown token")
			return
		}
		// Cross-guest probe via an explicit query vmid → 403 (the op is not run).
		if q := strings.TrimSpace(r.URL.Query().Get("vmid")); q != "" {
			if want, err := strconv.Atoi(q); err != nil || want != vmid {
				s.logger.Warn("local-api: cross-guest request refused",
					"token_guest", vmid, "requested", q, "path", r.URL.Path)
				writeErr(w, http.StatusForbidden, "token is not scoped to that guest")
				return
			}
		}
		fn(w, r, vmid)
	}
}

// bearer extracts the Authorization: Bearer <token> value.
func bearer(r *http.Request) string {
	h := r.Header.Get("Authorization")
	const p = "Bearer "
	if len(h) > len(p) && strings.EqualFold(h[:len(p)], p) {
		return strings.TrimSpace(h[len(p):])
	}
	return ""
}

// scopedFromBody enforces self-scoping for a POST body that may carry an explicit vmid: a
// non-zero body vmid that disagrees with the token's guest → 403 (false return; caller stops).
func (s *Server) scopedFromBody(w http.ResponseWriter, bodyVMID, tokenVMID int, path string) bool {
	if bodyVMID != 0 && bodyVMID != tokenVMID {
		s.logger.Warn("local-api: cross-guest request refused (body)",
			"token_guest", tokenVMID, "requested", bodyVMID, "path", path)
		writeErr(w, http.StatusForbidden, "token is not scoped to that guest")
		return false
	}
	return true
}

// ---- handlers ---------------------------------------------------------------------------

// MountInfo is one of the guest's attached mountpoints with its placement class.
type MountInfo struct {
	Key        string `json:"key"`         // mp0, mp1, …
	Storage    string `json:"storage"`     // PVE storage id
	MountPoint string `json:"mount_point"` // in-guest path
	Class      string `json:"class"`       // fast | slow | "" (unknown) — from the host storage view
	Backup     bool   `json:"backup"`      // included in vzdump (backup=1)
}

// StorageResponse is GET /storage — this guest's mounts + class so the controller can place
// hot vs bulk volumes per .felhom.yml.
type StorageResponse struct {
	VMID   int         `json:"vmid"`
	Mounts []MountInfo `json:"mounts"`
}

func (s *Server) handleStorage(w http.ResponseWriter, r *http.Request, vmid int) {
	cfg, err := s.guests.GuestConfig(r.Context(), vmid)
	if err != nil {
		s.logger.Error("local-api: /storage guest config", "vmid", vmid, "err", err)
		writeErr(w, http.StatusBadGateway, "could not read guest config")
		return
	}
	classByStore := s.classByStorage(r.Context())
	mounts := make([]MountInfo, 0)
	for key, val := range cfg.MountPoints() {
		store, mp, backup := parseMount(val)
		mounts = append(mounts, MountInfo{
			Key: key, Storage: store, MountPoint: mp,
			Class: classByStore[store], Backup: backup,
		})
	}
	writeOK(w, StorageResponse{VMID: vmid, Mounts: mounts})
}

type snapshotRequest struct {
	VMID        int    `json:"vmid"` // optional; must match the token's guest if set
	Snapname    string `json:"snapname"`
	Description string `json:"description"`
}

func (s *Server) handleSnapshot(w http.ResponseWriter, r *http.Request, vmid int) {
	var req snapshotRequest
	if !decodeBody(w, r, &req) {
		return
	}
	if !s.scopedFromBody(w, req.VMID, vmid, r.URL.Path) {
		return
	}
	snap := strings.TrimSpace(req.Snapname)
	if snap == "" {
		snap = "felhom-" + strconv.FormatInt(time.Now().UTC().Unix(), 10)
	}
	if !validSnapname(snap) {
		writeErr(w, http.StatusBadRequest, "invalid snapname (allowed: letters, digits, '_', '-')")
		return
	}
	upid, err := s.guests.Snapshot(r.Context(), vmid, snap, req.Description)
	if err != nil {
		s.logger.Error("local-api: snapshot", "vmid", vmid, "err", err)
		writeErr(w, http.StatusBadGateway, "snapshot failed")
		return
	}
	if _, err := s.guests.WaitTask(r.Context(), upid, proxmox.WaitOptions{}); err != nil {
		s.logger.Error("local-api: snapshot task", "vmid", vmid, "err", err)
		writeErr(w, http.StatusBadGateway, "snapshot task failed")
		return
	}
	writeOK(w, map[string]any{"vmid": vmid, "snapname": snap})
}

type rollbackRequest struct {
	VMID     int    `json:"vmid"`
	Snapname string `json:"snapname"`
}

func (s *Server) handleRollback(w http.ResponseWriter, r *http.Request, vmid int) {
	var req rollbackRequest
	if !decodeBody(w, r, &req) {
		return
	}
	if !s.scopedFromBody(w, req.VMID, vmid, r.URL.Path) {
		return
	}
	snap := strings.TrimSpace(req.Snapname)
	if snap == "" || !validSnapname(snap) {
		writeErr(w, http.StatusBadRequest, "snapname is required (letters, digits, '_', '-')")
		return
	}
	upid, err := s.guests.Rollback(r.Context(), vmid, snap)
	if err != nil {
		s.logger.Error("local-api: rollback", "vmid", vmid, "snap", snap, "err", err)
		writeErr(w, http.StatusBadGateway, "rollback failed")
		return
	}
	if _, err := s.guests.WaitTask(r.Context(), upid, proxmox.WaitOptions{}); err != nil {
		s.logger.Error("local-api: rollback task", "vmid", vmid, "err", err)
		writeErr(w, http.StatusBadGateway, "rollback task failed")
		return
	}
	writeOK(w, map[string]any{"vmid": vmid, "rolled_back_to": snap})
}

type backupRequest struct {
	VMID int `json:"vmid"`
}

// BackupResponse is POST /backup. The controller polls GET /backup/status on job_id to completion.
type BackupResponse struct {
	VMID  int    `json:"vmid"`
	JobID string `json:"job_id"`
	Phase string `json:"phase"`
}

func (s *Server) handleBackup(w http.ResponseWriter, r *http.Request, vmid int) {
	// Body is optional; if present its vmid must match the token's guest.
	if r.ContentLength != 0 {
		var req backupRequest
		if !decodeBody(w, r, &req) {
			return
		}
		if !s.scopedFromBody(w, req.VMID, vmid, r.URL.Path) {
			return
		}
	}
	// Single-flight per guest: if a backup is already running for this guest, return that job
	// (don't start a second concurrent vzdump). The controller polls /backup/status on it.
	s.jobsMu.Lock()
	if cur := s.jobs[vmid]; cur != nil && cur.Phase == PhaseRunning {
		job := *cur
		s.jobsMu.Unlock()
		writeStatus(w, http.StatusAccepted, true, BackupResponse{VMID: vmid, JobID: job.JobID, Phase: job.Phase}, "")
		return
	}
	jobID := "backup-" + strconv.Itoa(vmid) + "-" + strconv.FormatInt(s.now().UnixNano(), 10)
	s.jobs[vmid] = &backupJob{JobID: jobID, Phase: PhaseRunning, StartedAt: s.now()}
	s.jobsMu.Unlock()

	// Enqueue: a vzdump runs for minutes, so we do not block the request. The backup runs on the
	// server's base context (cancelled on daemon shutdown), updates the job phase, and records
	// into the store; the controller polls GET /backup/status. This is the host-side half of the
	// 8B app-consistent path — the controller quiesces (stops its stacks) BEFORE calling this, so
	// the vzdump captures a clean-shutdown-consistent state.
	base := s.baseCtx
	if base == nil {
		base = context.Background()
	}
	go func() {
		bctx, cancel := context.WithTimeout(base, 2*time.Hour)
		defer cancel()
		// 8B.2: flip the job to `snapshotted` when the storage snapshot is taken, so the
		// controller resumes its app early (snapshot mode only; in stop mode this never fires).
		b, err := s.backups.BackupWithSnapshotHook(bctx, vmid, func() { s.markSnapshotted(vmid, jobID) })
		if err != nil {
			b.VMID = vmid
			b.Success = false
			if b.Error == "" {
				b.Error = err.Error()
			}
			s.logger.Error("local-api: backup job failed", "vmid", vmid, "job", jobID, "err", err)
		} else {
			s.logger.Info("local-api: backup job complete", "vmid", vmid, "job", jobID, "archive", b.Archive)
		}
		s.store.RecordBackup(b)
		s.finishJob(vmid, jobID, b)
	}()
	writeStatus(w, http.StatusAccepted, true, BackupResponse{VMID: vmid, JobID: jobID, Phase: PhaseRunning}, "")
}

// markSnapshotted flips the guest's running job to the `snapshotted` phase (8B.2) — only if it is
// still the current job and still running (don't regress done/failed, and don't touch a newer job).
func (s *Server) markSnapshotted(vmid int, jobID string) {
	s.jobsMu.Lock()
	defer s.jobsMu.Unlock()
	cur := s.jobs[vmid]
	if cur == nil || cur.JobID != jobID || cur.Phase != PhaseRunning {
		return
	}
	cur.Phase = PhaseSnapshotted
	s.logger.Info("local-api: backup reached snapshotted (app may resume)", "vmid", vmid, "job", jobID)
}

// finishJob transitions the guest's job to done/failed (only if it is still the current job — a
// later job started after a single-flight gap must not be overwritten by an older one's result).
func (s *Server) finishJob(vmid int, jobID string, b hub.Backup) {
	s.jobsMu.Lock()
	defer s.jobsMu.Unlock()
	cur := s.jobs[vmid]
	if cur == nil || cur.JobID != jobID {
		return
	}
	cur.FinishedAt = s.now()
	if b.Success {
		cur.Phase = PhaseDone
		cur.Archive = b.Archive
	} else {
		cur.Phase = PhaseFailed
		cur.Error = b.Error
	}
}

// jobSnapshot returns a copy of the guest's current job (ok=false if none).
func (s *Server) jobSnapshot(vmid int) (backupJob, bool) {
	s.jobsMu.Lock()
	defer s.jobsMu.Unlock()
	if j := s.jobs[vmid]; j != nil {
		return *j, true
	}
	return backupJob{}, false
}

// BackupDueResponse is GET /backup/due (slice 8B). A guest is due when no successful backup is
// recorded OR the newest successful one is older than the agent-local cadence. A successful
// POST /backup flips this to false for the window, so the controller won't re-quiesce in a loop.
// The hub-served policy is slice 10.
type BackupDueResponse struct {
	VMID    int    `json:"vmid"`
	Due     bool   `json:"due"`
	Reason  string `json:"reason"`
	AgeSecs *int64 `json:"age_seconds,omitempty"` // age of the newest successful backup; null if none
}

func (s *Server) handleBackupDue(w http.ResponseWriter, r *http.Request, vmid int) {
	latest := s.latestSuccessfulBackupFor(r.Context(), vmid)
	if latest == nil {
		writeOK(w, BackupDueResponse{VMID: vmid, Due: true, Reason: "no successful backup recorded yet"})
		return
	}
	age, ok := backupAge(latest.StartedAt, s.now())
	if !ok {
		// Unparseable timestamp: fail safe toward "due" so a backup still happens.
		writeOK(w, BackupDueResponse{VMID: vmid, Due: true, Reason: "last backup time unparseable — treating as due"})
		return
	}
	ageSecs := int64(age.Seconds())
	if age >= s.cadence {
		writeOK(w, BackupDueResponse{VMID: vmid, Due: true, Reason: "older than cadence", AgeSecs: &ageSecs})
		return
	}
	writeOK(w, BackupDueResponse{VMID: vmid, Due: false, Reason: "within cadence window", AgeSecs: &ageSecs})
}

// BackupStatusResponse is GET /backup/status (slice 8B): the current/last job phase + the latest
// recorded backup. Phase is idle when no job has run this process lifetime.
type BackupStatusResponse struct {
	VMID   int         `json:"vmid"`
	Phase  string      `json:"phase"` // idle | running | done | failed
	JobID  string      `json:"job_id,omitempty"`
	Error  string      `json:"error,omitempty"`
	Backup *hub.Backup `json:"backup,omitempty"` // latest recorded backup for this guest
}

func (s *Server) handleBackupStatus(w http.ResponseWriter, r *http.Request, vmid int) {
	resp := BackupStatusResponse{VMID: vmid, Phase: PhaseIdle, Backup: s.latestBackupFor(r.Context(), vmid)}
	if job, ok := s.jobSnapshot(vmid); ok {
		resp.Phase = job.Phase
		resp.JobID = job.JobID
		resp.Error = job.Error
	}
	writeOK(w, resp)
}

func (s *Server) handleRestoreTestStatus(w http.ResponseWriter, r *http.Request, vmid int) {
	// The self-restore-test is host-level (it tests the newest backup in a throwaway scratch),
	// not per-guest — the controller surfaces it in its UI. Return the latest (0 or 1).
	tests := s.store.RestoreTests(r.Context())
	var latest *hub.RestoreTest
	if len(tests) > 0 {
		latest = &tests[0]
	}
	writeOK(w, map[string]any{"restore_test": latest})
}

// ---- helpers ----------------------------------------------------------------------------

// latestBackupFor returns this guest's most recent backup from the store (nil if none).
func (s *Server) latestBackupFor(ctx context.Context, vmid int) *hub.Backup {
	return s.pickLatestBackup(ctx, vmid, false)
}

// latestSuccessfulBackupFor returns this guest's most recent SUCCESSFUL backup (nil if none) —
// the basis for /backup/due (a failed backup must not satisfy the cadence).
func (s *Server) latestSuccessfulBackupFor(ctx context.Context, vmid int) *hub.Backup {
	return s.pickLatestBackup(ctx, vmid, true)
}

func (s *Server) pickLatestBackup(ctx context.Context, vmid int, successOnly bool) *hub.Backup {
	var latest *hub.Backup
	for _, b := range s.store.Backups(ctx) {
		if b.VMID != vmid || (successOnly && !b.Success) {
			continue
		}
		bb := b
		if latest == nil || bb.StartedAt > latest.StartedAt { // RFC3339 sorts lexically
			latest = &bb
		}
	}
	return latest
}

// backupAge parses an RFC3339 backup start time and returns its age relative to now.
func backupAge(startedAt string, now time.Time) (time.Duration, bool) {
	t, err := time.Parse(time.RFC3339, startedAt)
	if err != nil {
		return 0, false
	}
	return now.Sub(t), true
}

// classByStorage builds a storage-id → class(fast|slow|"") map from the host storage view. A
// view error is logged and yields an empty map (class falls back to "" — a hint, not load-bearing).
func (s *Server) classByStorage(ctx context.Context) map[string]string {
	out := map[string]string{}
	targets, err := s.storage.Observe(ctx)
	if err != nil {
		s.logger.Warn("local-api: storage view unavailable for class hints", "err", err)
		return out
	}
	for _, t := range targets {
		out[t.Name] = t.ClassHint
	}
	return out
}

// parseMount splits a PVE mpN value like "local-lvm:1,mp=/mnt/bulk,backup=0" into its storage
// id, in-guest mount path, and backup flag.
func parseMount(val string) (storage, mountPoint string, backup bool) {
	parts := strings.Split(val, ",")
	if len(parts) > 0 {
		if i := strings.IndexByte(parts[0], ':'); i >= 0 {
			storage = parts[0][:i]
		} else {
			storage = parts[0]
		}
	}
	for _, p := range parts[1:] {
		switch {
		case strings.HasPrefix(p, "mp="):
			mountPoint = strings.TrimPrefix(p, "mp=")
		case strings.HasPrefix(p, "backup="):
			backup = strings.TrimPrefix(p, "backup=") == "1"
		}
	}
	return storage, mountPoint, backup
}

// validSnapname allows PVE-safe snapshot names (letters, digits, '_' and '-').
func validSnapname(s string) bool {
	if s == "" || len(s) > 64 {
		return false
	}
	for _, c := range s {
		if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '_' || c == '-' {
			continue
		}
		return false
	}
	return true
}

func decodeBody(w http.ResponseWriter, r *http.Request, v any) bool {
	if r.Body == nil || r.ContentLength == 0 {
		return true // empty body is allowed; fields stay zero
	}
	dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<16))
	dec.DisallowUnknownFields()
	if err := dec.Decode(v); err != nil {
		writeErr(w, http.StatusBadRequest, "invalid JSON body")
		return false
	}
	return true
}

// ---- response envelope (matches the controller's {ok,data,error} style) -----------------

type apiResponse struct {
	OK    bool   `json:"ok"`
	Data  any    `json:"data,omitempty"`
	Error string `json:"error,omitempty"`
}

func writeOK(w http.ResponseWriter, data any) {
	writeStatus(w, http.StatusOK, true, data, "")
}

func writeErr(w http.ResponseWriter, code int, msg string) {
	writeStatus(w, code, false, nil, msg)
}

func writeStatus(w http.ResponseWriter, code int, ok bool, data any, errMsg string) {
	w.Header().Set("Content-Type", "application/json")
	w.WriteHeader(code)
	_ = json.NewEncoder(w).Encode(apiResponse{OK: ok, Data: data, Error: errMsg})
}