// Package quiesce implements the slice-8B app-consistent backup loop (doc 03 §6/§8): the
// in-guest controller polls the host agent's GET /backup/due, and when due it QUIESCES (stops its
// app stacks) → POST /backup → polls GET /backup/status to completion → UNQUIESCES (restarts
// exactly the stacks it stopped). An agent-initiated vzdump is crash-consistent only (an LXC has
// no fsfreeze); stopping the stacks first makes the captured state clean-shutdown-consistent.
//
// The correctness centerpiece is crash-safety: a stranded-down app is worse than a crash-consistent
// backup. So: a persisted marker is written BEFORE stopping anything; unquiesce is guaranteed (it
// runs even when the backup errors or times out); a max-quiesce bound restarts the app no matter
// what; and on controller startup Recover() restarts any stacks left stopped by a mid-quiesce crash.
package quiesce

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"os"
	"path/filepath"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-controller/internal/backupwindow"
)

// ErrBackupInProgress is returned by TriggerNow when a scheduled or manual quiesce cycle is already
// running (single-flight). The caller (the "Mentés most" handler) surfaces it as a benign 409.
var ErrBackupInProgress = errors.New("quiesce: a backup cycle is already in progress")

// Backend is the agent local-API surface the loop needs (satisfied by an adapter over
// *agentapi.Client). Kept minimal (bool/int/string) so the loop is testable with plain fakes.
// Due also returns the age of the newest successful backup in seconds (nil = none yet) — the
// window gate's safety valve reads it so a box powered on only outside its window never starves.
type Backend interface {
	Due(ctx context.Context) (due bool, ageSecs *int64, err error)
	StartBackup(ctx context.Context) (jobID string, err error)
	BackupStatus(ctx context.Context) (phase string, err error)
}

// Stacks is the stack-control surface (satisfied by *stacks.Manager). RunningAppStacks must return
// only deployed, non-protected, currently-up stacks (so unquiesce restarts exactly those).
type Stacks interface {
	RunningAppStacks() []string
	StopStack(name string) error
	StartStack(name string) error
}

// Backup status phases (mirror the agent's vocabulary).
const (
	phaseSnapshotted = "snapshotted" // 8B.2: storage snapshot taken → app may resume early
	phaseDone        = "done"
	phaseFailed      = "failed"
)

// Marker is the persisted quiesce state — the crash-safety + single-flight record. It is written
// (atomically, 0600) BEFORE any stack is stopped, so a controller crash mid-quiesce leaves a
// durable "these stacks were stopped, restart them" note that Recover honors at next startup.
type Marker struct {
	Active        bool      `json:"active"`
	StartedAt     time.Time `json:"started_at"`
	StoppedStacks []string  `json:"stopped_stacks"`
	JobID         string    `json:"job_id"`
}

// Options configures a Loop.
type Options struct {
	Backend    Backend
	Stacks     Stacks
	MarkerPath string        // persisted marker (e.g. <data_dir>/quiesce-state.json)
	Poll       time.Duration // how often to check /backup/due
	StatusPoll time.Duration // how often to poll /backup/status while quiesced
	MaxQuiesce time.Duration // hard bound on app downtime (unquiesce no matter what)
	Logger     *log.Logger
	// WindowStartFn returns the CURRENT effective backup-window start "HH:MM" (customer-configurable,
	// so it is read fresh each poll — a window change must take effect without restart). When nil the
	// window gate is disabled and a due cycle runs whenever the agent says due (pre-v0.168.0 behavior).
	WindowStartFn func() string
	// Cadence is the agent's backup cadence, used only by the gate's safety valve (run regardless of
	// the window once the last successful backup is older than Cadence+24h). Defaults to 24h.
	Cadence time.Duration
}

// Loop is the quiesce background loop.
type Loop struct {
	backend    Backend
	stacks     Stacks
	markerPath string
	poll       time.Duration
	statusPoll time.Duration
	maxQuiesce time.Duration
	logger     *log.Logger
	now        func() time.Time
	// windowStartFn (nil = gate disabled) + cadence drive the scheduled-cycle window gate (Part 3).
	windowStartFn func() string
	cadence       time.Duration
	// mu single-flights the quiesce cycle across the scheduled loop AND the manual trigger, so the
	// two can never stop the same stacks concurrently (the persisted marker covers crash-safety across
	// restarts; this covers concurrency within the process — which a manual trigger introduces).
	mu sync.Mutex
	// degradeOnce reports the pre-R-82 agent fallback exactly once per process (see tiers.go).
	degradeOnce sync.Once
	// ageStateDegradeOnce reports a pre-v0.105.0 agent (no age_state) exactly once (R-88 Part 2).
	ageStateDegradeOnce sync.Once
	// breaker (R-88) defers the QUIESCE for a tier whose backups keep failing, so a broken target
	// cannot stop the customer's apps every 5 minutes forever. Scheduled path only — see breaker.go.
	breaker *failureBreaker
	// contention (F-A1) tracks per-tier HTTP 409 runs. A refusal is NOT a failure, so it must never
	// touch the breaker — but it still has to defer the quiesce (else the apps are stopped every
	// poll for the whole restore-test) and it must alarm if it never ends. See contention.go.
	contention *contentionTracker
	// tierNotify (R-97a) reports a tier's backup outcome to the hub. nil = not wired (pre-provisioning).
	// Init-only: set once at startup via SetTierNotifier, before Run.
	tierNotify TierNotifier
	// suppressed (R-97b) is stack name → grace expiry (zero = still quiesced). Read by
	// SuppressedStacks so an app WE stopped is not reported to the customer as broken.
	suppressMu sync.Mutex
	suppressed map[string]time.Time
	// restartFailed (F-CRIT-1) is the set of stacks this loop stopped and could NOT restart. Guarded
	// by suppressMu — same concern, same lock. See suppress.go.
	restartFailed map[string]struct{}
}

// SetTierNotifier wires the hub-event seam. INIT-ONLY — call once at startup, before Run.
//
// It is a setter rather than an Options field because main.go constructs the notifier AFTER the
// quiesce loop, and reordering that has a wider blast radius than a setter does. nil is legal and
// means "no hub" — a guest that is not provisioned yet still backs up, it just cannot report.
func (l *Loop) SetTierNotifier(n TierNotifier) { l.tierNotify = n }

// New builds a Loop with sane defaults for any unset duration.
func New(o Options) *Loop {
	if o.Poll <= 0 {
		o.Poll = 5 * time.Minute
	}
	if o.StatusPoll <= 0 {
		o.StatusPoll = 10 * time.Second
	}
	if o.MaxQuiesce <= 0 {
		o.MaxQuiesce = 30 * time.Minute
	}
	if o.Logger == nil {
		o.Logger = log.Default()
	}
	if o.Cadence <= 0 {
		o.Cadence = 24 * time.Hour
	}
	return &Loop{
		backend: o.Backend, stacks: o.Stacks, markerPath: o.MarkerPath,
		poll: o.Poll, statusPoll: o.StatusPoll, maxQuiesce: o.MaxQuiesce,
		logger: o.Logger, now: time.Now,
		windowStartFn: o.WindowStartFn, cadence: o.Cadence,
		breaker:    newFailureBreaker(),
		contention: newContentionTracker(),
	}
}

// Recover restarts any stacks left stopped by a controller crash mid-quiesce, then clears the
// marker. Call ONCE at startup, before Run. Idempotent — StartStack on an already-running stack is
// tolerated; an absent/inactive marker is a no-op.
func (l *Loop) Recover() {
	m, ok := l.readMarker()
	if !ok || !m.Active {
		return
	}
	l.logger.Printf("[WARN] [quiesce] crash recovery: a quiesce was in progress (job %q, %d stack(s) stopped) — restarting them",
		m.JobID, len(m.StoppedStacks))
	l.noteRestartOutcome(m.StoppedStacks, l.restartAll(m.StoppedStacks))
	if err := l.clearMarker(); err != nil {
		l.logger.Printf("[ERROR] [quiesce] crash recovery: clear marker: %v", err)
	}
}

// Run polls for a due backup and runs the quiesce cycle, until ctx is cancelled.
func (l *Loop) Run(ctx context.Context) {
	l.logger.Printf("[INFO] [quiesce] loop started (poll %s, max-quiesce %s)", l.poll, l.maxQuiesce)
	ticker := time.NewTicker(l.poll)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			l.logger.Printf("[INFO] [quiesce] loop stopping")
			return
		case <-ticker.C:
			if err := l.runOnce(ctx); err != nil && ctx.Err() == nil {
				l.logger.Printf("[ERROR] [quiesce] cycle error: %v", err)
			}
		}
	}
}

// runOnce performs one due-check → (if due) quiesce → backup → poll → unquiesce cycle. Unquiesce
// is guaranteed via the deferred closure: a backup error, a status-poll error, the max-quiesce
// bound, or context cancellation all still restart the stacks and clear the marker.
func (l *Loop) runOnce(ctx context.Context) error {
	// Single-flight: skip the scheduled check if a cycle (scheduled or manual) is already running.
	if !l.mu.TryLock() {
		l.logger.Printf("[INFO] [quiesce] a backup cycle is already running — skipping this scheduled check")
		return nil
	}
	defer l.mu.Unlock()

	// Defensive single-flight: never quiesce on top of an active marker (Recover clears one left
	// by a crash; the mutex above serializes within the process).
	if m, ok := l.readMarker(); ok && m.Active {
		l.logger.Printf("[WARN] [quiesce] a marker is already active — skipping this cycle")
		return nil
	}

	// R-82: resolve EVERY due tier up front. This is the dedup rule (tiers.go): both tiers due on
	// the weekly night yields ONE window with two backups, never two stop/start cycles.
	dueTiers, _, err := l.resolveDueTiers(ctx)
	if err != nil {
		return fmt.Errorf("check due: %w", err)
	}
	if len(dueTiers) == 0 {
		return nil
	}

	// R-88 breaker — drop tiers whose backups keep failing, BEFORE anything is stopped. This is the
	// gate that actually ends the app thrash: the harm was never the failing backup, it was the
	// outage taken to attempt it. Per-tier, so a broken offsite tier leaves a healthy local one alone.
	// SCHEDULED path only — TriggerNow never reaches here.
	dueTiers = l.dropBackedOffTiers(dueTiers)
	if len(dueTiers) == 0 {
		return nil
	}

	// Window gate (Part 3) — SCHEDULED path only. TriggerNow calls quiesceAndPoll directly and is
	// never gated. Disabled when no window fn is wired (pre-v0.168.0 behavior).
	//
	// With several tiers due, the gate is evaluated against the OLDEST (most overdue) tier's age,
	// so the safety valve — "run regardless of the window once the last success is older than
	// cadence+24h" — cannot be suppressed by a fresher sibling tier.
	if l.windowStartFn != nil {
		window := l.windowStartFn()
		if !scheduledRunAllowed(l.now().In(budapestLocation()), window, oldestAge(dueTiers), valveLicensed(dueTiers), l.cadence) {
			from, to := gateBounds(window)
			l.logger.Printf("[DEBUG] [quiesce] scheduled backup due but outside the backup window [%s–%s) — deferring to the next poll inside it", from, to)
			return nil
		}
	}

	return l.quiesceAndPollTiers(ctx, dueTiers)
}

// noteTierFailure arms/extends the tier's backoff and announces the deferral exactly ONCE — here, at
// the moment it is armed. Called from BOTH the scheduled and the manual path: a manual run that
// fails is evidence about the tier too. Only the GATING is scheduler-only.
func (l *Loop) noteTierFailure(target, label, errMsg string) {
	n, d := l.breaker.recordFailure(target, l.now())
	l.logger.Printf("[WARN] [quiesce] tier %s has now failed %d time(s) in a row — deferring its next quiesce by %s (cap %s) so the apps are not stopped again for a backup that cannot succeed",
		label, n, d, breakerMaxDelay)
	// R-97a: report ONCE, when the breaker ARMS (n == 1), never on the retries behind it.
	//
	// This got MORE urgent when R-88 shipped, not less: before the breaker a failing backup retried
	// every 5 minutes — harmful, but loud enough to notice. Now it backs off to 4h and goes quiet,
	// leaving the hub's deadline monitor as the only signal at ~26h (local) / ~8 days (PBS) — a full
	// cycle of the weekly tier. This trades that delay for an immediate one.
	if n == 1 && l.tierNotify != nil {
		l.tierNotify.BackupFailed(label,
			fmt.Sprintf("Whole-guest backup FAILED on the %s tier — retrying with backoff (next attempt in %s)", label, d),
			errMsg)
	}
}

// noteTierContention records a 409 refusal: log it as contention, defer the tier, and — only if
// contention has run past contentionAlarmAfter — raise it as a fault.
//
// The headline deliberately says BLOCKED, not FAILED. The operator's action for "a concurrent
// operation has held the agent for three hours" is entirely different from "the backup errored",
// and the whole point of F-A1 is that the two stopped being distinguishable.
//
// It reuses the existing `whole_guest_backup_failed` event type rather than inventing one: a new
// type would need the hub's allowedEventTypes + customerMessages pair changed, i.e. a wire change,
// which this fix deliberately does not make.
func (l *Loop) noteTierContention(target, label string) {
	elapsed, alarmNow := l.contention.note(target, l.now())
	l.logger.Printf("[INFO] [quiesce] tier %s is BUSY — the agent refused the backup because a concurrent heavy operation holds it. This is contention, NOT a failure: the tier stays due and retries in %s (contended for %s)",
		label, contentionRetryAfter, elapsed.Round(time.Second))
	if !alarmNow {
		return
	}
	l.logger.Printf("[ERROR] [quiesce] tier %s has been BLOCKED by a concurrent operation for over %s — that exceeds the agent's own restore-test ceiling, so the gate is stuck, not busy",
		label, contentionAlarmAfter)
	if l.tierNotify != nil {
		l.tierNotify.BackupFailed(label,
			fmt.Sprintf("Whole-guest backup BLOCKED on the %s tier — a concurrent operation has held the agent for over %s and the backup has still not run", label, contentionAlarmAfter),
			fmt.Sprintf("contention (HTTP 409) unresolved for %s; exceeds the %s restore-test ceiling", elapsed.Round(time.Second), contentionAlarmAfter))
	}
}

// noteTierSuccess clears any backoff. Quiet unless there was something to clear — a line per healthy
// backup would be noise, but a recovery is worth one.
func (l *Loop) noteTierSuccess(target, label string) {
	if l.contention.clear(target) {
		l.logger.Printf("[INFO] [quiesce] tier %s is no longer contended — the concurrent operation released the agent", label)
	}
	// recordSuccess's bool is the edge: true only when there WAS a backoff to clear. That is exactly
	// the recovery edge — an operator told a tier broke must also be told it healed, and a line (or
	// an event) per healthy backup would be noise.
	if l.breaker.recordSuccess(target) {
		l.logger.Printf("[INFO] [quiesce] tier %s succeeded — clearing its failure backoff; normal cadence resumes", label)
		if l.tierNotify != nil {
			l.tierNotify.BackupRecovered(label,
				fmt.Sprintf("Whole-guest backup RECOVERED on the %s tier — it succeeded after a run of failures", label))
		}
	}
}

// dropBackedOffTiers removes tiers currently inside an R-88 backoff window.
//
// Silent by design at INFO: the deferral is announced ONCE, when the backoff is armed in
// quiesceAndPollTiers. Logging here would fire every 5 minutes for hours and bury the one line that
// matters — the same "a repeating log is not a signal" problem the loop itself had.
func (l *Loop) dropBackedOffTiers(tiers []dueTier) []dueTier {
	now := l.now()
	kept := make([]dueTier, 0, len(tiers))
	for _, t := range tiers {
		if until, blocked := l.breaker.blocked(t.target, now); blocked {
			l.logger.Printf("[DEBUG] [quiesce] tier %s is in backoff after %d consecutive failure(s) — not quiescing until %s",
				tierLabel(t.target), l.breaker.failuresFor(t.target), until.Format(time.RFC3339))
			continue
		}
		// F-A1: a CONTENDED tier is dropped here too, for the same reason R-88 drops a failing one —
		// before any stack is stopped. Without this, removing the (wrong) failure treatment would
		// leave the loop re-quiescing every 5 minutes for the whole restore-test, stopping the
		// customer's apps each time: worse than the bug being fixed.
		if until, busy := l.contention.blocked(t.target, now); busy {
			l.logger.Printf("[DEBUG] [quiesce] tier %s is BUSY (a concurrent heavy operation holds the agent) — not quiescing until %s; the tier stays due",
				tierLabel(t.target), until.Format(time.RFC3339))
			continue
		}
		kept = append(kept, t)
	}
	return kept
}

// valveLicensed reports whether ANY due tier holds a POSITIVE claim of "never backed up" — the only
// thing that may fire the window-gate safety valve on a nil age (R-88 Part 2).
//
// Two states license it, and the second is the important one:
//   - AgeStateAbsent — the agent looked and there is genuinely nothing there;
//   - AgeStateLegacy — a pre-v0.105.0 agent that cannot tell us either way. Preserving the OLD
//     behaviour is correct here: reading its silence as "unknown" would stop the valve firing on
//     every un-upgraded box, so a genuinely new box would never take its first backup outside its
//     window and nobody would notice for weeks. The MinAgent floor drives the upgrade; the valve is
//     not the place to force it.
//
// AgeStateUnknown does NOT license it. That is the entire fix: an unreadable storage no longer
// masquerades as a first-ever backup.
func valveLicensed(tiers []dueTier) bool {
	for _, t := range tiers {
		if t.ageSecs != nil {
			continue // a real age needs no licence; the age comparison decides
		}
		if t.state == AgeStateAbsent || t.state == AgeStateLegacy {
			return true
		}
	}
	return false
}

// oldestAge returns the largest (most overdue) age among the due tiers; nil when any tier has never
// backed up (nil age = "never", which is maximally overdue and must win).
func oldestAge(tiers []dueTier) *int64 {
	var oldest *int64
	for _, t := range tiers {
		if t.ageSecs == nil {
			return nil // never backed up — the strongest claim on the safety valve
		}
		if oldest == nil || *t.ageSecs > *oldest {
			oldest = t.ageSecs
		}
	}
	return oldest
}

// TriggerNow forces an app-consistent backup NOW (the manual "Mentés most" action), bypassing the
// /backup/due check. It runs the SAME quiesce flow the scheduled loop uses (stop stacks → POST
// /backup → poll → resume), so it is app-consistent and crash-safe (marker-protected). Single-flight
// via the same mutex: it returns ErrBackupInProgress if a scheduled or manual cycle is already
// running. The cycle runs ASYNCHRONOUSLY (it can take minutes) on a background context bounded by
// maxQuiesce; the caller polls /backup/status for progress. The controller — not the agent — owns
// quiescing (the agent's vzdump is crash-consistent only), so this MUST go through the loop.
func (l *Loop) TriggerNow() error {
	if !l.mu.TryLock() {
		return ErrBackupInProgress
	}
	if m, ok := l.readMarker(); ok && m.Active {
		l.mu.Unlock()
		return ErrBackupInProgress
	}
	go func() {
		defer l.mu.Unlock()
		// Detached from any request context; bounded so a hung backup still unquiesces.
		ctx, cancel := context.WithTimeout(context.Background(), l.maxQuiesce+5*time.Minute)
		defer cancel()
		l.logger.Printf("[INFO] [quiesce] manual backup requested — quiescing now")
		// Manual runs bypass due-ness (that is the point) but must still cover EVERY tier, in one
		// window. A manual "Mentés most" that silently skipped the DR tier would be the same
		// applied-and-empty fault in a different costume.
		if err := l.quiesceAndPollTiers(ctx, l.allTiersForManualRun(ctx)); err != nil {
			l.logger.Printf("[ERROR] [quiesce] manual backup cycle error: %v", err)
		}
	}()
	return nil
}

// quiesceAndPoll performs the marked, guaranteed-unquiesce cycle: write marker → stop running app
// stacks → POST /backup → poll /backup/status → restart exactly the stacks it stopped. The caller
// MUST hold l.mu. Unquiesce is guaranteed via the deferred closure (backup error, status-poll error,
// the max-quiesce bound, or context cancellation all still restart the stacks and clear the marker).
func (l *Loop) quiesceAndPoll(ctx context.Context) error {
	return l.quiesceAndPollTiers(ctx, []dueTier{{target: ""}})
}

// allTiersForManualRun lists every tier a manual run should cover: all advertised tiers on an R-82
// agent, or the single untargeted tier otherwise. Due-ness is deliberately NOT consulted.
func (l *Loop) allTiersForManualRun(ctx context.Context) []dueTier {
	tb, ok := l.backend.(TieredBackend)
	if !ok {
		return []dueTier{{target: ""}}
	}
	tiers, err := tb.Tiers(ctx)
	if err != nil || len(tiers) == 0 {
		if errors.Is(err, ErrTiersUnsupported) {
			l.logTierDegradeOnce()
		} else if err != nil {
			l.logger.Printf("[WARN] [quiesce] manual run: tier list unavailable (%v) — using the untargeted tier", err)
		}
		return []dueTier{{target: ""}}
	}
	out := make([]dueTier, 0, len(tiers))
	for _, t := range tiers {
		out = append(out, dueTier{target: t.Target})
	}
	return out
}

// quiesceAndPollTiers is the R-82 multi-tier cycle: ONE marker, ONE stop, N backups run
// SEQUENTIALLY inside the window, ONE resume, then the tail polled to completion.
//
// Why sequential: vzdump takes a guest lock, so a second backup cannot start until the first
// finishes. Why the app stays down until the LAST tier snapshots: the whole point of quiescing is
// app-consistency, and resuming after tier 1's snapshot would leave tier 2 capturing a RUNNING app.
// Consequence, stated plainly because it is user-visible: on the both-due night downtime is
// (first tier's full backup) + (last tier's snapshot), not one snapshot. Tier ORDER therefore
// matters — see resolveDueTiers.
//
// Crash-safety is unchanged and non-negotiable: the marker is written BEFORE anything stops,
// unquiesce fires exactly once no matter which tier fails, and the GUARANTEE is the crash MARKER plus
// Recover() — not the defer. Campaign 8 fault 10 established that: a SIGKILL mid-quiesce runs no
// deferred function, and what brought the stacks back was Recover() reading the marker 1 s after
// restart. The defer covers the graceful exits (backup error, poll error, max-quiesce bound, context
// cancellation); the marker covers the hard crash. Do not describe this as "guaranteed by defer". A crash
// between two backups leaves the marker on disk and Recover() restarts the stacks at startup.
func (l *Loop) quiesceAndPollTiers(ctx context.Context, tiers []dueTier) error {
	if len(tiers) == 0 {
		return nil
	}
	running := l.stacks.RunningAppStacks()
	marker := Marker{Active: true, StartedAt: l.now(), StoppedStacks: running}
	if err := l.writeMarker(marker); err != nil {
		return fmt.Errorf("write quiesce marker (refusing to stop stacks unprotected): %w", err)
	}

	// GUARANTEED unquiesce + marker clear — runs on every exit path below.
	unquiesced := false
	unquiesce := func(reason string) {
		if unquiesced {
			return
		}
		unquiesced = true
		l.logger.Printf("[INFO] [quiesce] unquiescing (%s): restarting %d stack(s)", reason, len(running))
		// F-CRIT-1: record which stacks came back and which did not, so the app-down classifier can
		// tell "the user stopped this" from "we stopped it and could not restart it".
		l.noteRestartOutcome(running, l.restartAll(running))
		// R-97b: start the grace clock AFTER the restart call, so the window measures time the app
		// has actually had to come up rather than time it spent stopped.
		l.markUnquiesced(running)
		if err := l.clearMarker(); err != nil {
			l.logger.Printf("[ERROR] [quiesce] clear marker: %v", err)
		}
	}
	defer unquiesce("deferred")

	l.logger.Printf("[INFO] [quiesce] backup due on %d tier(s) — quiescing %d stack(s): %v",
		len(tiers), len(running), running)
	// R-97b: exempt these from app-down alarms BEFORE stopping them, or a health scan landing between
	// the stop and the mark would alarm on an app we are about to take down deliberately.
	l.markQuiesced(running)
	for _, st := range running {
		if err := l.stacks.StopStack(st); err != nil {
			l.logger.Printf("[ERROR] [quiesce] stop %s: %v (continuing)", st, err)
		}
	}

	deadline := l.now().Add(l.maxQuiesce)
	var firstErr error

	// ONE window, N tiers, sequential. The app resumes only after the LAST tier snapshots.
	for i, t := range tiers {
		label := tierLabel(t.target)
		last := i == len(tiers)-1

		jobID, err := l.startBackupOn(ctx, t.target)
		if errors.Is(err, ErrTierBusy) {
			// F-A1: the agent's single-flight gate refused (HTTP 409). Nothing is broken — a
			// restore-test or another backup holds it. This is CONTENTION, not failure: no breaker,
			// no whole_guest_backup_failed, no operator email. The tier stays DUE and retries.
			l.noteTierContention(t.target, label)
			if last {
				unquiesce("last tier is busy — deferring to a later cycle")
			}
			continue
		}
		if err != nil {
			l.logger.Printf("[ERROR] [quiesce] start backup on tier %s: %v", label, err)
			// A REAL error ends any contention run: whatever the gate was doing, this is a fault now.
			l.contention.clear(t.target)
			l.noteTierFailure(t.target, label, err.Error())
			if firstErr == nil {
				firstErr = fmt.Errorf("start backup on %s: %w", label, err)
			}
			// A tier that will not start must not hold the app down for the others.
			if last {
				unquiesce("last tier failed to start")
			}
			continue
		}
		// The tier started, so it is not contended. Clear before the poll so a long backup cannot be
		// mistaken for a stuck gate.
		l.contention.clear(t.target)
		marker.JobID = jobID
		_ = l.writeMarker(marker) // best-effort: record the CURRENT tier's job id for diagnosis
		l.logger.Printf("[INFO] [quiesce] tier %s: backup job %s started — polling", label, jobID)

		phase, stillRunning, perr := l.pollTier(ctx, t.target, jobID, label, deadline, last, &unquiesced, unquiesce)
		if perr != nil && firstErr == nil {
			firstErr = perr
		}
		switch {
		case phase == phaseFailed:
			l.logger.Printf("[WARN] [quiesce] tier %s: backup job %s failed", label, jobID)
			l.noteTierFailure(t.target, label, "backup job "+jobID+" reported phase=failed")
		case stillRunning:
			// Neither outcome yet — a first full offsite snapshot legitimately runs for hours. It
			// must NOT count as a failure, or a slow-but-healthy tier would back itself off.
		default:
			l.noteTierSuccess(t.target, label)
		}
		if stillRunning {
			// The max-quiesce guard fired while THIS tier's backup is still going (a first full
			// offsite snapshot legitimately runs for hours). The app is already back up. We must
			// NOT start the next tier: ONE BACKUP AT A TIME PER GUEST (operator ruling
			// 2026-07-26) — vzdump still holds the guest lock, so a second start would be refused
			// by the agent (409) or fail on the lock. The remaining tiers simply run on a later
			// poll, once this one has finished.
			//
			// CORRECTED v0.179.0 (F-A1): this used to say the 409 would "record a spurious
			// failure". Until v0.179.0 that was not a hypothetical the comment was guarding
			// against — it was what the start path ACTUALLY did, on every 409, on both boxes.
			// A 409 is now handled as CONTENTION (see contention.go): no breaker, no operator
			// email, the tier stays due. So the outcome this comment feared no longer exists.
			l.logger.Printf("[INFO] [quiesce] tier %s still running past the quiesce bound — deferring %d remaining tier(s) to a later cycle",
				label, len(tiers)-i-1)
			break
		}
	}

	// Belt: if every tier failed to start, nothing above unquiesced. The deferred call covers it,
	// but doing it here keeps the "resume as soon as possible" property explicit.
	unquiesce("cycle complete")
	return firstErr
}

// pollTier polls ONE tier's job. It returns the terminal (or snapshotted-at-deadline) phase.
//
// The app is resumed ONLY when this is the LAST tier — that is what keeps every tier
// app-consistent while still costing exactly one stop/start pair. For a non-last tier the loop
// waits for a TERMINAL phase (done/failed), because vzdump holds the guest lock and the next tier
// cannot start until this one truly finishes.
// Returns (phase, stillRunning, err). stillRunning=true means the quiesce bound elapsed while the
// backup is STILL going — the caller must not start another tier (one backup at a time per guest).
func (l *Loop) pollTier(ctx context.Context, target, jobID, label string, deadline time.Time,
	last bool, unquiesced *bool, unquiesce func(string)) (string, bool, error) {
	for {
		if !l.now().Before(deadline) {
			l.logger.Printf("[WARN] [quiesce] max-quiesce-duration (%s) exceeded on tier %s (job %s) — unquiescing while the backup continues on the agent",
				l.maxQuiesce, label, jobID)
			unquiesce("max-quiesce guard")
			return "", true, nil
		}
		phase, err := l.backupStatusOn(ctx, target)
		if err != nil {
			unquiesce("status poll failed")
			return "", false, fmt.Errorf("poll backup status on %s: %w", label, err)
		}
		switch phase {
		case phaseSnapshotted:
			// 8B.2 early resume — but ONLY on the last tier. Resuming here on a non-last tier would
			// leave the following tier capturing a running app, losing app-consistency for exactly
			// the DR tier we most want it on.
			if last && !*unquiesced {
				l.logger.Printf("[INFO] [quiesce] tier %s: job %s snapshotted — resuming app early (8B.2)", label, jobID)
				unquiesce("snapshotted (early resume, last tier)")
			}
		case phaseDone:
			if last {
				l.logger.Printf("[INFO] [quiesce] tier %s: backup job %s done", label, jobID)
				unquiesce("backup done")
			} else {
				l.logger.Printf("[INFO] [quiesce] tier %s: backup job %s done — next tier may start (app still quiesced)", label, jobID)
			}
			return phaseDone, false, nil
		case phaseFailed:
			if last {
				unquiesce("backup failed")
			}
			return phaseFailed, false, nil
		}
		select {
		case <-ctx.Done():
			unquiesce("controller shutting down")
			return "", false, ctx.Err()
		case <-time.After(l.statusPoll):
		}
	}
}

// ---- window gate (Part 3, v0.168.0) -----------------------------------------------------

var (
	quiesceBudapest     *time.Location
	quiesceBudapestOnce sync.Once
)

func budapestLocation() *time.Location {
	quiesceBudapestOnce.Do(func() {
		loc, err := time.LoadLocation("Europe/Budapest")
		if err != nil {
			quiesceBudapest = time.UTC
			return
		}
		quiesceBudapest = loc
	})
	return quiesceBudapest
}

const (
	gateOpenOffsetMin = 120 // gate opens at W+2h
	gateSpanMin       = 240 // 4h span → [W+2h, W+6h)
)

// scheduledRunAllowed decides whether a DUE, scheduled whole-guest backup may run at `now` (passed by
// the caller as Budapest wall-clock — only its hour/minute are read). True when now is inside the gate
// window [W+2h, W+6h); otherwise true ONLY if the safety valve holds — the newest successful backup is
// missing (nil) or older than cadence+24h — so a box powered on only outside its window never starves.
// An unparseable window fails OPEN (allow) rather than block backups forever.
//
// ── THE INVARIANT (R-88). READ THIS BEFORE TOUCHING THE nil BRANCH. ──────────────────────────
//
// A missing value means UNKNOWN. It does not mean zero, and it does not mean "never backed up".
// Only a POSITIVE determination of "never backed up" may fire the safety valve.
//
// This project has now made the opposite mistake four times, in four different packages:
// hub v0.12.0, hub v0.73.0, R-81 (hub `assessBackupFreshness`), and R-88 (here). Each time, the
// absence of a signal was read as a specific value, and each time the fix was the same shape:
// give "unknown" its own representation instead of letting it collapse into a real answer.
//
// ── CLOSED BY R-88 PART 2 (agent v0.105.0 + controller v0.178.0) ─────────────────────────────
//
// The value is now THREE-STATE, not merely "nil or not". The agent reports `age_state` on
// /backup/due — `known` / `absent` / `unknown` — and a nil age fires the valve only when
// `valveLicensed` finds a tier holding a POSITIVE claim of "never backed up".
//
// It used to be that the agent returned BYTE-IDENTICAL responses for "the storage read errored" and
// "there has genuinely never been a backup" (same Due, same Reason, same nil AgeSecs), because
// `newestArchiveOn`'s `(time.Time, bool)` signature could not represent "unknown" — while its own
// doc comment promised exactly that. An unreadable storage therefore masqueraded as a first-ever
// backup and quiesced customer apps outside the window. Fixed at the source.
//
// STILL LICENSED, DELIBERATELY: `AgeStateLegacy` — a pre-v0.105.0 agent that omits the field. Its
// silence must NOT be read as "unknown", or the valve stops firing on every un-upgraded box and a
// genuinely new box never takes its first backup. The MinAgent floor drives the upgrade instead.
//
// DO NOT "fix" this by deleting the nil branch, or by dropping the legacy case from valveLicensed.
// Scenario D — a genuinely never-backed-up box that is only ever powered on outside its window —
// depends on BOTH, and TestContract_NeverBackedUp_RunsOutsideTheWindow will fail if you do.
// Silencing the valve trades a loud bug for a silent one: a box that never backs up at all, with
// nobody noticing for weeks.
func scheduledRunAllowed(now time.Time, windowStart string, lastAgeSecs *int64, valveOK bool, cadence time.Duration) bool {
	startMin, err := backupwindow.ParseHHMM(windowStart)
	if err != nil {
		return true
	}
	nowMin := now.Hour()*60 + now.Minute()
	if within(nowMin, mod1440(startMin+gateOpenOffsetMin), gateSpanMin) {
		return true
	}
	// Outside the window: only the safety valve may run it.
	if lastAgeSecs == nil {
		// R-88 Part 2: a nil age is no longer self-licensing. It fires the valve ONLY on a positive
		// "never backed up" (or a legacy agent that cannot say). An UNKNOWN age — an unreadable
		// storage — now defers, which is the whole point of this arc.
		return valveOK
	}
	return time.Duration(*lastAgeSecs)*time.Second > cadence+24*time.Hour
}

// gateBounds returns the gate window [W+2h, W+6h) as HH:MM for the deferral log line.
func gateBounds(windowStart string) (from, to string) {
	return backupwindow.GateWindow(windowStart)
}

func mod1440(m int) int { return ((m % 1440) + 1440) % 1440 }

// within reports whether minute-of-day p falls in [start, start+span) modulo 24h (wrap-safe).
func within(p, start, span int) bool {
	return mod1440(p-start) < span
}

// restartAll restarts the given stacks and RETURNS the ones that failed.
//
// F-CRIT-1 cause 1: this used to return nothing. The error was logged and dropped on the spot, so
// no caller could learn that a customer's app had not come back — and the classifier downstream
// therefore had nothing to key on. A restart failure is the single most important fact this loop
// produces; it must leave the function.
func (l *Loop) restartAll(stacks []string) []string {
	var failed []string
	for _, s := range stacks {
		if err := l.stacks.StartStack(s); err != nil {
			l.logger.Printf("[ERROR] [quiesce] restart %s: %v", s, err)
			failed = append(failed, s)
		}
	}
	return failed
}

// ---- marker persistence (atomic, 0600) --------------------------------------------------

func (l *Loop) writeMarker(m Marker) error {
	m.Active = true
	data, err := json.MarshalIndent(m, "", "  ")
	if err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(l.markerPath), 0o755); err != nil {
		return err
	}
	tmp := l.markerPath + ".tmp"
	if err := os.WriteFile(tmp, data, 0o600); err != nil {
		os.Remove(tmp)
		return err
	}
	return os.Rename(tmp, l.markerPath)
}

func (l *Loop) readMarker() (Marker, bool) {
	data, err := os.ReadFile(l.markerPath)
	if err != nil {
		return Marker{}, false
	}
	var m Marker
	if err := json.Unmarshal(data, &m); err != nil {
		// S3: a corrupt marker is NOT silently dropped — log it LOUD and quarantine the bad file (a real
		// corrupted-mid-quiesce marker would otherwise skip stack-recovery with no trace). Still return
		// false: "no usable marker" ⇒ no recovery is the correct contract.
		l.logger.Printf("[WARN] [quiesce] marker at %s is corrupt (%v) — quarantining; stacks not auto-recovered from it", l.markerPath, err)
		_ = os.Rename(l.markerPath, fmt.Sprintf("%s.corrupt-%d", l.markerPath, l.now().Unix()))
		return Marker{}, false
	}
	return m, true
}

func (l *Loop) clearMarker() error {
	err := os.Remove(l.markerPath)
	if os.IsNotExist(err) {
		return nil
	}
	return err
}