package reconcile

import (
	"context"
	"errors"
	"fmt"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

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

// The unified bring-up reconcile job (doc 03 §9, slice 7 Phase 1). It is the shared FRONT
// HALF of provision and guest-loss DR: restore an archive → reset identity (scenario-specific)
// → size → attach mounts → start LINK-UP. It mirrors RunRestoreTest's crash-safety (journal the
// owning entry BEFORE any mutation; Recover reaps a leak via ListLXC; idempotent, fail-safe) but
// DIFFERS in two load-bearing ways: it KEEPS the guest on success (no teardown), and its
// identity policy is scenario-specific (provision = fresh identity; DR = preserve continuity).
//
// Grounded by documentation/tests/slice7-bringup-spike-findings.md (commit 3342993): F1 (restore
// preserves the archived MAC → provision MUST reset it), F3 (machine-id + SSH host keys
// regenerate guest-side on first boot from the clean golden + its baked unit — the agent does
// NOT touch guest internals here), F4 (a transient PVE config-lock 500 → bounded retry).
//
// Out of scope (deferred — see REPORT): the provisioning BACK HALF (controller deploy, bootstrap,
// per-guest token mint → slice 8); host-loss DR + escrow consumption → slice 10; the SOURCE of a
// BringUpSpec (hub desired-state) → slice 10 (this job takes the spec as input).

// BringUpMode selects the identity policy + archive semantics.
type BringUpMode string

const (
	// ModeProvision: from the golden base — a NEW guest, so reset identity fully.
	ModeProvision BringUpMode = "provision"
	// ModeDRGuestLoss: from a customer backup — CONTINUE the customer's world, so preserve
	// continuity identity (hostname, host keys, and by default MAC), reset only what collides.
	ModeDRGuestLoss BringUpMode = "dr_guest_loss"
)

const bringUpKind = "bring_up"

// DefaultDataVolMount is the mpN slot the golden bakes the Docker-data volume (/var/lib/docker) at.
const DefaultDataVolMount = "mp0"

// DefaultPool is the PVE pool every Felhom-managed guest is restored INTO. Under the pool-scoped agent
// token, restore-into-pool is how a fresh vmid gets allocated (VM.Allocate + Pool.Allocate at
// /pool/felhom) and how the guest becomes reachable by the scoped token (SPIKE-pool-scoped-acl-2026-07-01).
// Single source of truth for both restore sites (provision bring-up + restore-test).
const DefaultPool = "felhom"

// DefaultSysDataMount is RETIRED (agent v0.120.0, R-165 / decision D-a). The golden no longer bakes a
// second volume: since build-golden.sh v3.0.0 there is ONE data volume at /var/lib/felhom (mp0) and
// both /var/lib/docker and /mnt/sys_drive are binds of subdirectories of it, so there is no mp1 to
// resize. The constant is kept, and deliberately points at nothing, so that a stale caller fails
// loudly at review rather than silently resizing a slot that does not exist.
//
// SysDataGrowGB itself is NOT removed — see its field comment: the host installer still passes
// `-sysdata-grow`, and its GiB are FOLDED INTO the single volume's grow rather than dropped.
const DefaultSysDataMount = "" 

// Structural host-bind mountpoints every provisioned guest carries (GL-5; verdict of
// SPIKE-dr-bindmount-source-2026-07-07): the permanent drives parent bind (mp8,
// /mnt/felhom-drives, same in-guest path) and the bootstrap config bind (mp9,
// <stateDir>/guests/<vmid>/bootstrap -> /etc/felhom-bootstrap, read-only). These are PLATFORM
// CONSTANTS - backhalf.go names mp8 "the single permanent parent bind" and mp9's host path is
// templated only by vmid - which is exactly why DR can synthesize them without reading the lost
// guest's config. Mirrors provision/backhalf.go's stableParentDir/parentBindSlot/DefaultGuestPath/
// DefaultMountIndex as literals (the same avoid-the-import-edge rationale backhalf itself uses for
// its localapi mirror). A customer archive CARRIES these mpN entries, and a pct restore of a
// bind mount is root@pam-only ("restoring 'mpN' to bind mount is only possible for root") - so a
// DR restore under the privsep token MUST override them (throwaway volumes) and swap the real
// binds back post-restore (step 4d).
const (
	structuralParentSlot    = "mp8"
	structuralParentDir     = "/mnt/felhom-drives"
	structuralBootSlot      = "mp9"
	structuralBootGuestPath = "/etc/felhom-bootstrap"
)

// structuralBootHostDir is the mp9 bind's host source for vmid (mirrors the back-half's
// <stateDir>/guests/<vmid>/bootstrap layout). Joined with "/" explicitly: this is a HOST (Linux)
// path that flows into pct arguments - filepath.Join would mangle it on a non-Linux test runner.
func structuralBootHostDir(stateDir string, vmid int) string {
	return strings.TrimRight(stateDir, "/") + "/guests/" + strconv.Itoa(vmid) + "/bootstrap"
}

// archiveCurrentConfig parses the CURRENT-config key/value lines out of an extracted archive
// config (raw pct-conf text). Snapshot sections ("[name]") follow the current config; parsing
// stops at the first one so a snapshot's values can never shadow the live ones.
func archiveCurrentConfig(raw string) map[string]string {
	out := map[string]string{}
	for _, line := range strings.Split(raw, "\n") {
		if strings.HasPrefix(line, "[") {
			break
		}
		k, v, ok := strings.Cut(line, ":")
		if !ok {
			continue
		}
		out[k] = strings.TrimSpace(v)
	}
	return out
}

// drRestoreOverrides builds the COMPLETE restore param set a DR bring-up must pass, from the
// archive's extracted config (GL-5; both PVE constraints discovered live — see the call site):
// explicit rootfs (sized from the archive), every storage-backed mpN passed through (recreated at
// its archived size + in-guest path + backup flag so vzrestore extracts its content), and the two
// structural bind slots replaced with throwaway volumes for 4d to swap. Unknown bind mpN or an
// unparseable size → error (never silently restore a guest missing a mount).
func drRestoreOverrides(rawArchiveCfg, restoreStorage string) (map[string]string, error) {
	cfg := archiveCurrentConfig(rawArchiveCfg)
	sz := rootfsSizeGB(cfg["rootfs"])
	if sz <= 0 {
		return nil, fmt.Errorf("archive config carries no parseable rootfs size — refusing a mount-override restore without an explicit rootfs")
	}
	out := map[string]string{"rootfs": fmt.Sprintf("%s:%d", restoreStorage, sz)}
	for key, val := range cfg {
		if len(key) <= 2 || key[:2] != "mp" || key[2] < '0' || key[2] > '9' {
			continue
		}
		volPart, rest, _ := strings.Cut(val, ",")
		if strings.HasPrefix(volPart, "/") {
			// a host bind — must be one of the two structural slots (replaced below)
			if key != structuralParentSlot && key != structuralBootSlot {
				return nil, fmt.Errorf("archive carries an unknown bind mountpoint %s=%q — refusing (only the structural %s/%s binds are known)", key, val, structuralParentSlot, structuralBootSlot)
			}
			continue
		}
		msz := rootfsSizeGB(val)
		if msz <= 0 {
			return nil, fmt.Errorf("archive mountpoint %s=%q carries no parseable size — cannot pass it through the explicit-params restore", key, val)
		}
		mp := mountPathOf(rest)
		if mp == "" {
			return nil, fmt.Errorf("archive mountpoint %s=%q carries no mp= path", key, val)
		}
		v := fmt.Sprintf("%s:%d,mp=%s", restoreStorage, msz, mp)
		if strings.Contains(","+rest+",", ",backup=1,") {
			v += ",backup=1"
		}
		out[key] = v
	}
	out[structuralParentSlot] = throwawayVolumeOverride(restoreStorage, structuralParentDir)
	out[structuralBootSlot] = throwawayVolumeOverride(restoreStorage, structuralBootGuestPath)
	return out, nil
}

// configLockMaxAttempts bounds the F4 config-lock retry. configLockBackoff is a package var so
// tests can shrink it (the production value gives PVE time to release its async config lock).
const configLockMaxAttempts = 5

var configLockBackoff = 1500 * time.Millisecond

// GuestMount is one additive mountpoint to attach as mpN. Defined minimally now; slice 10 wires
// the hub storage manifest into this (do NOT couple to a not-yet-existing hub desired-state type).
type GuestMount struct {
	Storage    string // PVE storage id (e.g. "local-lvm")
	SizeGB     int    // new-volume size in GiB
	MountPoint string // in-guest path (e.g. "/mnt/data")
	// Backup includes this mountpoint in vzdump/PBS. MANDATORY for any data-bearing mount (DB
	// volumes), because extra LXC mountpoints default to backup=0 = EXCLUDED from the snapshot
	// (storage-split finding B3). The Docker-data volume normally rides in from the golden archive
	// (already backup=1) and is grown via DataVolGrowGB rather than attached here, but any data
	// mount attached through spec.Mounts MUST set this or its contents silently fall out of PBS.
	Backup bool
}

// BringUpSpec is the input to one bring-up. The caller resolves it (the selftest, or slice-10
// hub desired-state); this job does not decide WHAT to provision.
type BringUpSpec struct {
	Mode           BringUpMode // provision | dr_guest_loss
	Archive        string      // source volid (golden for provision; customer backup for DR)
	VMID           int         // caller-provided target VMID (NOT the restore-test band / 9999)
	RestoreStorage string      // rootfs target storage
	Hostname       string      // hostname to set (provision); ignored for DR (continuity)
	Pool           string      // restore the guest INTO this PVE pool ("" = none); required under a pool-scoped token
	Cores          int         // 0 = leave as restored
	MemoryMB       int         // 0 = leave as restored
	RootfsGrowGB   int         // optional grow-only rootfs resize (0 = skip)
	// DataVolGrowGB grows the golden-carried Docker-data volume (DataVolMount, default mp0) to the
	// per-customer target. The golden ships a small data volume with the baked images; provision
	// grows it online (grow-only, storage-split B4) rather than attaching a fresh empty volume that
	// would shadow the baked images. 0 = skip (keep the golden's size).
	DataVolGrowGB int
	// DataVolMount is the mpN slot of the golden's Docker-data volume to grow; "" → DefaultDataVolMount ("mp0").
	DataVolMount string
	// SysDataGrowGB is a COMPATIBILITY INPUT since agent v0.120.0 (R-165). There is no longer a second
	// volume to grow — but `felhom.eu/scripts/felhom-host-install.sh` computes and passes
	// `-sysdata-grow` (its step_grows derives both numbers from the thin pool's free space), and an
	// installer and an agent do not upgrade in the same instant.
	//
	// SO ITS GiB ARE FOLDED INTO THE SINGLE VOLUME'S GROW RATHER THAN DROPPED. Dropping them would
	// silently shrink every appliance by the user-data share — on the ≥300 GiB branch that is 42 of
	// 250 GiB — which is exactly the "a knob that silently does nothing" outcome R-165 was told to
	// avoid. Folding keeps total capacity identical whichever installer version runs.
	SysDataGrowGB int
	// SysDataMount is RETIRED and ignored (see DefaultSysDataMount). Kept so an older caller still
	// compiles; it selects nothing.
	SysDataMount string
	Mounts       []GuestMount  // additive mpN mounts (slice 7 may pass empty/test)
	KeepMAC      bool          // DR knob: keep the archived MAC (true) unless a source may be live
	BootTimeout  time.Duration // 0 → DefaultBootTimeout; bounds the link-up liveness wait
	// IslandBridge + IslandGuestAddr (R-50): when BOTH are set, the guest gets a static net1 on the
	// host-internal island bridge, so the controller reaches the agent over a fixed private address
	// that survives any LAN/DHCP/site move (the F1 fix). Empty (default) = no net1, byte-for-byte the
	// pre-R-50 config. Set from cfg.LocalAPI (island_bridge/island_guest_addr) at both call sites.
	IslandBridge    string // e.g. "vmbr9"
	IslandGuestAddr string // guest net1 CIDR, e.g. "169.254.253.2/30"
}

// BringUpResult is the outcome. It reuses the restore-test's WARNINGS surface
// (StartWarnings/WarningsRecognized) — the start step here is the same liveness-anchored boot.
type BringUpResult struct {
	VMID               int
	AssignedMAC        string // the guest's net0 MAC after identity reset (fresh for provision)
	Hostname           string
	Pass               bool
	Verified           string // "boot+running"
	Err                error
	StartedAt          time.Time
	Duration           time.Duration
	StartWarnings      []string
	WarningsRecognized bool
}

// IntentForRollbackDestroy builds the benign compensating-rollback teardown intent for a guest
// the agent created earlier in THIS journaled bring-up transaction: ClassGuestDestroy made benign
// by SameTxnCreated provenance (classify.go) — a rollback, not data loss. Gate-authorized unsigned
// but genuinely in-path (wrong provenance → pending_signature). Distinct from the restore-test's
// IntentForScratchDestroy (AgentTaggedScratch) — different audit label, same destroy machinery.
func IntentForRollbackDestroy(hostID string, vmid int) Intent {
	return Intent{
		Class:      ClassGuestDestroy,
		HostID:     hostID,
		GuestID:    strconv.Itoa(vmid),
		VMID:       vmid,
		Provenance: Provenance{SameTxnCreated: true},
		Source:     SourceOneShotJob,
	}
}

// RunBringUp runs one bring-up on the target VMID's queue lane (inherits §10 per-guest
// serialization). On success the guest is KEPT; on any mid-flight failure it is
// compensating-rolled-back (destroyed). The returned Err is the job verdict's error.
func (e *Engine) RunBringUp(ctx context.Context, spec BringUpSpec) BringUpResult {
	now := time.Now().UTC()
	res := BringUpResult{VMID: spec.VMID, Hostname: spec.Hostname, StartedAt: now}

	if spec.Archive == "" || spec.RestoreStorage == "" {
		res.Err = fmt.Errorf("reconcile: bring-up needs an archive and a restore storage")
		return res
	}
	if spec.VMID <= 0 {
		res.Err = fmt.Errorf("reconcile: bring-up needs a target VMID")
		return res
	}
	// Fence the reserved bands: never provision over the standing scratch (9999) or the
	// restore-test scratch band (990000–990009).
	if spec.VMID == 9999 || (spec.VMID >= 990000 && spec.VMID <= 990009) {
		res.Err = fmt.Errorf("reconcile: bring-up VMID %d is reserved (9999 / restore-test scratch band)", spec.VMID)
		return res
	}
	switch spec.Mode {
	case ModeProvision, ModeDRGuestLoss:
	default:
		res.Err = fmt.Errorf("reconcile: bring-up unknown mode %q", spec.Mode)
		return res
	}
	// DR needs the host runner: the structural-bind swap (4d) is a root pct op the API token
	// cannot perform. Refuse up front rather than fail after a restore (GL-5).
	if spec.Mode == ModeDRGuestLoss && e.hostRun == nil {
		res.Err = fmt.Errorf("reconcile: dr bring-up needs a host runner (the mp8/mp9 structural-bind swap is a root pct op) — wire EngineOptions.HostRunner")
		return res
	}

	ch := e.queue.Submit(spec.VMID, func() error {
		e.runBringUp(ctx, spec, &res)
		return res.Err
	})
	<-ch
	res.Duration = time.Since(now)
	return res
}

// runBringUp is the journaled body (runs on the target VMID's queue lane).
func (e *Engine) runBringUp(ctx context.Context, spec BringUpSpec, res *BringUpResult) {
	// PROVISION is to a NEW VMID. Restoring OVER an existing guest is ClassRestoreOverwrite
	// (destructive, slice 10) — never this benign path. Refuse before journaling.
	lxc, err := e.api.ListLXC(ctx)
	if err != nil {
		res.Err = fmt.Errorf("reconcile: bring-up list guests: %w", err)
		return
	}
	for _, g := range lxc {
		if g.VMID == spec.VMID {
			res.Err = fmt.Errorf("reconcile: bring-up VMID %d already exists (restore-over-existing is a signed op)", spec.VMID)
			return
		}
	}

	base := JournalEntry{OpID: e.bringUpOpID(spec.VMID), VMID: spec.VMID, Kind: bringUpKind, Rollback: true}

	// OWN the rollback BEFORE any mutation. From here a crash leaves an in-flight Rollback
	// entry; Recover destroys the vmid ONLY when the entry carries a restore UPID
	// (proof-of-launch — see Recover's no-UPID abandon path).
	e.append(withState(base, OpStarted))

	// Compensating rollback on every non-committed exit AFTER the restore launched (defer):
	// destroy the just-created guest. On success we set committed and KEEP it (the key
	// difference from the restore-test). `launched` is the proof-of-launch gate (campaign
	// pool-effects F1a): a restore that failed synchronously (no UPID — e.g. PVE refusing a
	// vmid that already holds a guest the pool-blind duplicate guard can't see) created
	// NOTHING, so the rollback must NEVER destroy the vmid — a pre-existing guest, possibly
	// another customer's, may sit there. This must hold WITHOUT the pool ACL (that 403 is
	// defense-in-depth, not the guard). The owning entry is then closed terminal-failed
	// in-process (nothing exists to recover).
	committed := false
	launched := false
	defer func() {
		if committed {
			return
		}
		if !launched {
			e.append(withState(base, OpFailed))
			return
		}
		e.rollbackBringUp(ctx, base)
	}()

	// 1. Restore archive → VMID (token-covered ClassCreate; keyctl preserved — phase3 + spike).
	//    DR (GL-5): the customer archive carries the two STRUCTURAL host-bind mountpoints (mp8
	//    parent bind + mp9 bootstrap bind) that a restore under the privsep token cannot recreate
	//    — without overrides the whole restore FAILS ("restoring 'mp8' to bind mount is only
	//    possible for root"). Synthesize throwaway-volume overrides for the two known-constant mpN
	//    via the shared format helper (bindMountOverrides' is-a-bind filter is for reading real
	//    configs, which DR by definition cannot do — the guest is gone); step 4d swaps the real
	//    binds back post-restore. An archive that LACKS one of them (older backup) is fine: the
	//    override simply creates that mpN at restore and 4d normalizes it — the end state is
	//    identical (C3). Provision stays override-free (nil): the golden has no mp8/mp9 (the
	//    back-half adds them post-bring-up) — that asymmetry is the whole GL-5 bug.
	var overrides map[string]string
	if spec.Mode == ModeDRGuestLoss {
		// PVE's explicit-params restore is ALL-OR-NOTHING (both halves hit live in the GL-5
		// validation): (a) any mpN param without an explicit `rootfs` → 500 "mount points
		// configured, but 'rootfs' not set" (the constraint restoretest.go:211 documents);
		// (b) mountpoints NOT named in the params are silently DROPPED from the restore — a DR
		// guest restored with only the bind overrides came up WITHOUT its mp0/mp1 data volumes.
		// So the FULL layout must be specified, from the archive's own embedded config: rootfs
		// sized from it, every storage-backed mpN passed through (size+path+backup preserved →
		// vzrestore extracts its content), and the two structural binds replaced by throwaways
		// that 4d swaps for the real binds. A bind mpN outside the two structural slots means an
		// unknown topology — refuse loudly rather than restore a guest missing a mount.
		raw, err := e.api.ExtractArchiveConfig(ctx, spec.Archive)
		if err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up dr: extract archive config (the restore params derive from it): %w", err)
			return
		}
		overrides, err = drRestoreOverrides(raw, spec.RestoreStorage)
		if err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up dr: %w", err)
			return
		}
	}
	upid, err := e.api.RestoreLXC(ctx, proxmox.RestoreLXCOptions{
		VMID: spec.VMID, Archive: spec.Archive, Storage: spec.RestoreStorage, Pool: spec.Pool,
		MountOverrides: overrides,
	})
	if err != nil {
		// No UPID ⇒ nothing was created ⇒ the defer closes the entry WITHOUT a destroy.
		res.Err = fmt.Errorf("reconcile: bring-up restore: %w", err)
		return
	}
	// Proof-of-launch: the POST was accepted — from here a failure means a half-built guest the
	// compensating rollback (or Recover, via the journaled UPID) must destroy. Accepted residual:
	// a crash in the one-statement window before the UPID is journaled leaks a half-built guest
	// that Recover won't destroy — cleanable, and preferable to destroying an innocent guest.
	launched = true
	e.append(withUPID(base, upid, OpTaskRunning))
	if _, err := e.waitTask(ctx, upid, proxmox.WaitOptions{}); err != nil {
		res.Err = fmt.Errorf("reconcile: bring-up restore task: %w", err)
		return
	}

	// 2. Read as-restored config (the net0 the MAC handling keys off).
	cfg, err := e.api.GuestConfig(ctx, spec.VMID)
	if err != nil {
		res.Err = fmt.Errorf("reconcile: bring-up read config: %w", err)
		return
	}

	// 3+5 coalesced (F4): identity reset + cores/mem sizing + additive mounts in ONE config PUT,
	// with the bounded retry that fires ONLY on the transient PVE config-lock 500.
	if params := buildBringUpConfig(spec, cfg); len(params) > 0 {
		if err := e.setConfigWithLockRetry(ctx, spec.VMID, params); err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up config: %w", err)
			return
		}
	}

	// 4. rootfs grow-only resize as its OWN call (F4: kept separate from the config PUT).
	if spec.RootfsGrowGB > 0 {
		rupid, err := e.api.ResizeLXC(ctx, spec.VMID, "rootfs", fmt.Sprintf("+%dG", spec.RootfsGrowGB))
		if err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up resize: %w", err)
			return
		}
		if _, err := e.waitTask(ctx, rupid, proxmox.WaitOptions{}); err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up resize task: %w", err)
			return
		}
	}

	// 4b. Grow the golden-carried Docker-data volume (mp0) to the per-customer target. Grow-only,
	//     online (storage-split B4); its OWN call like the rootfs resize. The volume + baked images
	//     came in with the restore, so we grow it rather than attach a fresh one that would shadow
	//     the baked images.
	//
	//     R-165: ONE volume, therefore ONE grow. `SysDataGrowGB` is FOLDED IN here rather than driving
	//     a second resize — see its field comment. This is the only arithmetic the merge added.
	growGB := spec.DataVolGrowGB + spec.SysDataGrowGB
	if growGB > 0 {
		mount := spec.DataVolMount
		if mount == "" {
			mount = DefaultDataVolMount
		}
		if spec.SysDataGrowGB > 0 {
			e.logger.Info("bring-up: folding the retired sys-data grow into the single data volume (R-165)",
				"data_grow_gb", spec.DataVolGrowGB, "sysdata_grow_gb", spec.SysDataGrowGB, "total_gb", growGB, "mount", mount)
		}
		dupid, err := e.api.ResizeLXC(ctx, spec.VMID, mount, fmt.Sprintf("+%dG", growGB))
		if err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up data-volume resize (%s): %w", mount, err)
			return
		}
		if _, err := e.waitTask(ctx, dupid, proxmox.WaitOptions{}); err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up data-volume resize task (%s): %w", mount, err)
			return
		}
	}

	// 4c. RETIRED (R-165). There is no second volume: the golden ships ONE, and the sys-data grow is
	//     folded into 4b above. Deliberately left as a comment rather than silently vanishing, so a
	//     reader of a v0.119.0 archive's provision log can see where the second resize went.

	// 4d. DR structural-bind swap (GL-5): replace the two restore-time throwaway volumes (see the
	//     restore call) with the REAL host binds, then delete the displaced volumes so a KEPT DR
	//     guest carries no unusedN residue. Bind-mount pct sets are root@pam-only — the swap goes
	//     through the host runner, never the API. Runs BEFORE start so the first boot already sees
	//     the real binds (the golden's baked bootstrap unit + the drives parent). A failure here is
	//     surfaced with the exact mpN state and rolls back per the envelope (committed is still
	//     false) — never a silent half-wired success (C2).
	if spec.Mode == ModeDRGuestLoss {
		if err := e.swapStructuralBinds(ctx, spec, res); err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up structural-bind swap: %w", err)
			return
		}
	}

	// Capture the post-reset MAC for the result (fresh for provision; archived for DR keep).
	if cfg2, err := e.api.GuestConfig(ctx, spec.VMID); err == nil {
		res.AssignedMAC = net0MAC(cfg2)
	}

	// 6. Start LINK-UP (ClassStart). The VERDICT is liveness (waitRunning), NEVER the start
	//    exitstatus — same liveness-anchoring as the restore-test fix; AllowWarnings so a benign
	//    start advisory (systemd-nesting) is surfaced, not failed.
	startUPID, err := e.api.Start(ctx, spec.VMID)
	if err != nil {
		res.Err = fmt.Errorf("reconcile: bring-up start: %w", err)
		return
	}
	if startUPID != "" {
		st, err := e.api.WaitTask(ctx, startUPID, proxmox.WaitOptions{AllowWarnings: true})
		if err != nil {
			res.Err = fmt.Errorf("reconcile: bring-up start task: %w", err)
			return
		}
		if strings.HasPrefix(st.ExitStatus, "WARNINGS") {
			tail, logErr := e.api.TaskLogTail(ctx, startUPID, 50)
			if logErr != nil {
				e.logger.Warn("bring-up: could not read start-task log for warnings", "vmid", spec.VMID, "err", logErr)
			}
			res.StartWarnings = extractWarningLines(tail)
			res.WarningsRecognized = warningsRecognized(res.StartWarnings)
		}
	}
	bootTO := spec.BootTimeout
	if bootTO <= 0 {
		bootTO = DefaultBootTimeout
	}
	if err := e.waitRunning(ctx, spec.VMID, bootTO); err != nil {
		res.Err = err
		return
	}

	// 6b. Re-assert pool membership (campaign-2 R2). `pct restore --pool` sets membership only at
	//     CREATE; a restore OVER AN EXISTING VMID (host-loss finale) does not re-apply it, silently
	//     dropping the guest from the pool and 403-ing the NEXT restore-test/DR. Idempotent on a
	//     fresh-VMID restore that already got membership. This runs AFTER liveness is proven: a
	//     pool-add hiccup is surfaced LOUD as a warning but must NOT flip a healthy, running guest's
	//     verdict to fail (membership matters for the next op, not this guest's boot).
	if spec.Pool != "" {
		if err := e.api.PoolAddVMID(ctx, spec.Pool, spec.VMID); err != nil {
			e.logger.Error("bring-up: pool membership re-assert FAILED (next restore-test/DR may 403); guest is healthy",
				"vmid", spec.VMID, "pool", spec.Pool, "err", err)
			res.StartWarnings = append(res.StartWarnings, fmt.Sprintf("pool re-assert failed (pool=%s): %v", spec.Pool, err))
		} else {
			e.logger.Info("bring-up: pool membership re-asserted", "vmid", spec.VMID, "pool", spec.Pool)
		}
	}

	// 7. Success — KEEP the guest; mark the owning entry terminal so Recover ignores it.
	res.Pass = true
	res.Verified = "boot+running"
	committed = true
	e.append(withState(base, OpSucceeded))
}

// swapStructuralBinds is bring-up step 4d (DR only): set mp8/mp9 to the real host binds via the
// root runner (one slot per call so an error names the exact mpN that failed — C2), then delete
// the displaced throwaway volumes PVE parked as unusedN. The mp9 host dir is created first (pct
// validates the bind source; agent-owned path, plain MkdirAll like the back-half's own bootstrap
// dir) — idempotent: on a same-host guest-loss the dir usually still exists WITH bootstrap.json,
// which the swap must not touch. The unusedN delete goes through the API config PUT
// (VM.Config.Disk + Datastore.Allocate cover it); if the scoped token refuses, the residue is
// logged LOUDLY + surfaced as a result warning and the bring-up continues — a correctly-wired
// guest with a stray volume beats a rollback, and privileges are never widened silently.
func (e *Engine) swapStructuralBinds(ctx context.Context, spec BringUpSpec, res *BringUpResult) error {
	bootDir := structuralBootHostDir(e.stateDir, spec.VMID)
	if err := os.MkdirAll(bootDir, 0o700); err != nil {
		return fmt.Errorf("mp9 bootstrap host dir %s: %w", bootDir, err)
	}
	if _, stderr, err := e.hostRun.Run(ctx, "mkdir", "-p", structuralParentDir); err != nil {
		return fmt.Errorf("parent dir %s: %w: %s", structuralParentDir, err, stderr)
	}
	parentSpec := structuralParentDir + ",mp=" + structuralParentDir
	if _, stderr, err := e.hostRun.Run(ctx, "pct", "set", strconv.Itoa(spec.VMID), "-"+structuralParentSlot, parentSpec); err != nil {
		return fmt.Errorf("set %s (parent bind; neither bind landed): %w: %s", structuralParentSlot, err, stderr)
	}
	bootSpec := bootDir + ",mp=" + structuralBootGuestPath + ",ro=1"
	if _, stderr, err := e.hostRun.Run(ctx, "pct", "set", strconv.Itoa(spec.VMID), "-"+structuralBootSlot, bootSpec); err != nil {
		return fmt.Errorf("set %s (bootstrap bind; %s already landed): %w: %s", structuralBootSlot, structuralParentSlot, err, stderr)
	}
	e.logger.Info("bring-up: structural binds swapped in", "vmid", spec.VMID,
		structuralParentSlot, parentSpec, structuralBootSlot, bootSpec)

	// The displaced throwaway volumes now sit as unusedN — read the config and delete them.
	cfg, err := e.api.GuestConfig(ctx, spec.VMID)
	if err != nil {
		return fmt.Errorf("read config after bind swap: %w", err)
	}
	var unused []string
	for k := range cfg.Unused() {
		unused = append(unused, k)
	}
	if len(unused) == 0 {
		return nil
	}
	sort.Strings(unused)
	if err := e.setConfigWithLockRetry(ctx, spec.VMID, map[string]string{"delete": strings.Join(unused, ",")}); err != nil {
		e.logger.Error("bring-up: could not delete displaced throwaway volumes (guest is correctly wired; residue remains)",
			"vmid", spec.VMID, "unused", unused, "err", err)
		res.StartWarnings = append(res.StartWarnings,
			fmt.Sprintf("structural-bind swap: displaced volumes not deleted (%s): %v", strings.Join(unused, ","), err))
		return nil
	}
	e.logger.Info("bring-up: displaced throwaway volumes deleted", "vmid", spec.VMID, "unused", unused)
	return nil
}

// rollbackBringUp destroys the just-created guest (benign ClassGuestDestroy via SameTxnCreated
// provenance) and records the owning entry terminal. Called ONLY launch-proven (the restore POST
// was accepted — campaign pool-effects F1a): the SameTxnCreated provenance is then real, not
// assumed. On any teardown failure it leaves the entry in-flight so Recover reaps the guest later
// (via the journaled UPID) — never force-destroys.
func (e *Engine) rollbackBringUp(ctx context.Context, base JournalEntry) {
	tctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute)
	defer cancel()

	dec := e.gate.Authorize(IntentForRollbackDestroy(e.hostID, base.VMID), nil)
	if !dec.Allowed {
		e.logger.Error("bring-up: rollback destroy refused by gate (unexpected); left for Recover",
			"vmid", base.VMID, "reason", dec.Reason)
		return
	}
	upid, err := e.api.DestroyLXC(tctx, base.VMID)
	if err != nil {
		e.logger.Error("bring-up: rollback destroy failed; left for Recover", "vmid", base.VMID, "err", err)
		return
	}
	if _, err := e.waitTask(tctx, upid, proxmox.WaitOptions{}); err != nil {
		e.logger.Error("bring-up: rollback destroy task failed; left for Recover", "vmid", base.VMID, "err", err)
		return
	}
	e.append(withState(base, OpSucceeded))
	e.logger.Warn("bring-up: rolled back (destroyed half-built guest)", "vmid", base.VMID)
}

// buildBringUpConfig assembles the coalesced config PUT params per the scenario-specific identity
// policy (doc 03 §9). Provision: fresh MAC (strip hwaddr → PVE regenerates, F1) + hostname.
// DR: preserve continuity — keep MAC (unless KeepMAC=false: a source may be live) and keep
// hostname (no force-reset). Both: cores/mem sizing + additive mpN mounts. machine-id and SSH
// host keys are NOT touched here — they regenerate guest-side on first boot (golden bake + the
// baked first-boot unit), keeping the agent's front half host-side-only.
func buildBringUpConfig(spec BringUpSpec, cfg proxmox.GuestConfig) map[string]string {
	params := map[string]string{}

	resetMAC := spec.Mode == ModeProvision || (spec.Mode == ModeDRGuestLoss && !spec.KeepMAC)
	if resetMAC {
		if net0, ok := cfg.Nets()["net0"]; ok && net0 != "" {
			params["net0"] = withoutHwaddr(net0) // omit hwaddr → PVE generates a fresh MAC (F1)
		}
	}
	// R-50 island NIC: attach a static net1 on the host-internal bridge so the control plane
	// (controller→agent local API) rides a fixed private address, immune to any LAN/DHCP/site move.
	// Both modes: a provisioned guest AND a DR-restored guest need to reach the island-bound agent on
	// the target host. No hwaddr → PVE mints a fresh per-guest MAC (the /30 is one guest per host, so
	// a MAC would not collide either way, but a fresh one keeps net1 symmetric with net0). Additive:
	// omitted entirely when the island is not configured, keeping non-island hosts unchanged.
	if strings.TrimSpace(spec.IslandBridge) != "" && strings.TrimSpace(spec.IslandGuestAddr) != "" {
		params["net1"] = fmt.Sprintf("name=eth1,bridge=%s,ip=%s", spec.IslandBridge, spec.IslandGuestAddr)
	}
	if spec.Mode == ModeProvision && spec.Hostname != "" {
		params["hostname"] = spec.Hostname
	}
	if spec.Cores > 0 {
		params["cores"] = strconv.Itoa(spec.Cores)
	}
	if spec.MemoryMB > 0 {
		params["memory"] = strconv.Itoa(spec.MemoryMB)
	}
	for i, m := range spec.Mounts {
		// backup=1 for data-bearing mounts: extra LXC mountpoints default to backup=0 = EXCLUDED
		// from vzdump/PBS (storage-split B3), which would silently drop their DBs from the snapshot.
		spec := fmt.Sprintf("%s:%d,mp=%s", m.Storage, m.SizeGB, m.MountPoint)
		if m.Backup {
			spec += ",backup=1"
		}
		params[fmt.Sprintf("mp%d", i)] = spec
	}
	return params
}

// setConfigWithLockRetry issues the coalesced config PUT, retrying ONLY the transient PVE
// config-lock 500 (F4) with bounded backoff. A non-lock error (any other 500 included) fails
// immediately — never retry a real error. The slice-4 per-guest serializer prevents cross-op
// contention; this covers PVE releasing its own async config lock within one job.
func (e *Engine) setConfigWithLockRetry(ctx context.Context, vmid int, params map[string]string) error {
	var lastErr error
	for attempt := 1; attempt <= configLockMaxAttempts; attempt++ {
		if _, err := e.api.SetConfig(ctx, vmid, params); err == nil {
			return nil
		} else if !pveConfigLock(err) {
			return err // real error — never retry
		} else {
			lastErr = err
			e.logger.Warn("bring-up: transient PVE config-lock; retrying",
				"vmid", vmid, "attempt", attempt, "max", configLockMaxAttempts, "err", err)
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(configLockBackoff):
		}
	}
	return fmt.Errorf("reconcile: bring-up config still lock-contended after %d attempts: %w", configLockMaxAttempts, lastErr)
}

// pveConfigLock reports whether err is the transient PVE config-lock 500 (F4) — and ONLY that.
// The lock surfaces as a proxmox.APIError 500 whose body carries the lock signature.
func pveConfigLock(err error) bool {
	var ae *proxmox.APIError
	if !errors.As(err, &ae) || ae.StatusCode != 500 {
		return false
	}
	b := strings.ToLower(ae.Body)
	return strings.Contains(b, "can't lock file") || strings.Contains(b, "got timeout")
}

// pveAlreadyExists reports whether err is PVE's synchronous refusal to create over an existing
// vmid ("CT <vmid> already exists on node '<node>'" — an APIError 500, observed live in the
// pool-effects campaign). By construction such a refusal returned no UPID: nothing was created.
// Used by the restore-test band-advance (F2) to distinguish "band vmid occupied by a guest the
// pool-blind list can't see" from a real restore failure — never misclassify the latter.
func pveAlreadyExists(err error) bool {
	var ae *proxmox.APIError
	if !errors.As(err, &ae) || ae.StatusCode != 500 {
		return false
	}
	return strings.Contains(strings.ToLower(ae.Body), "already exists")
}

// waitTask waits a (possibly empty) UPID — "" is the clean synchronous path.
func (e *Engine) waitTask(ctx context.Context, upid string, opts proxmox.WaitOptions) (proxmox.TaskStatus, error) {
	if upid == "" {
		return proxmox.TaskStatus{}, nil
	}
	return e.api.WaitTask(ctx, upid, opts)
}

func (e *Engine) bringUpOpID(vmid int) string {
	return "bring-up-" + strconv.Itoa(vmid) + "-" + nextSeq(&e.opSeq)
}

// withoutHwaddr strips the hwaddr token from a netN config string. PUTting a netN with NO hwaddr
// makes PVE generate a fresh MAC (slice-7 spike F1: a restore preserves the archived MAC, so a
// provision MUST strip it to avoid a fleet-wide MAC collision). Mirrors withLinkDown's approach.
func withoutHwaddr(netN string) string {
	parts := strings.Split(netN, ",")
	out := parts[:0]
	for _, p := range parts {
		if p == "" || strings.HasPrefix(p, "hwaddr=") {
			continue
		}
		out = append(out, p)
	}
	return strings.Join(out, ",")
}

// net0MAC extracts the hwaddr from a guest's net0 config ("" if absent).
func net0MAC(cfg proxmox.GuestConfig) string {
	net0, ok := cfg.Nets()["net0"]
	if !ok {
		return ""
	}
	for _, p := range strings.Split(net0, ",") {
		if strings.HasPrefix(p, "hwaddr=") {
			return strings.TrimPrefix(p, "hwaddr=")
		}
	}
	return ""
}