felhom-agent/internal/reconcile/engine.go

package reconcile

import (
	"context"
	"fmt"
	"log/slog"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

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

// Engine converges actual Proxmox state toward the desired state. One Reconcile pass:
// read desired (from the provider), read actual (from Proxmox), Plan the minimal
// benign action set, and dispatch each action onto the per-guest Queue — journaling
// each op for crash-safety. At slice 4 the provider is EmptyProvider, so the action
// set is empty and the pass performs zero mutations (correct and expected).
//
// Concurrency: actions for different guests run in parallel (separate Queue lanes);
// actions for the same guest run serially in plan order. Every Proxmox mutation is
// async-or-sync per the mutate.go contract: a non-empty UPID is WaitTask'd and its
// exitstatus asserted; an empty UPID is a clean synchronous success.
type Engine struct {
	api      GuestAPI
	queue    *Queue
	journal  *Journal
	provider DesiredProvider
	norm     FieldNormalizers
	gate     *Gate
	hostID   string
	logger   *slog.Logger

	// hostRun executes host-root commands for the DR structural-bind swap (bring-up 4d): a
	// bind-mount `pct set` is root@pam-only, so it cannot go through the API token (the same
	// constraint the provision back-half documents). nil on API-only engines — RunBringUp
	// refuses ModeDRGuestLoss then (GL-5).
	hostRun proxmox.Runner
	// stateDir is the agent state dir the mp9 bootstrap host dir lives under
	// (<stateDir>/guests/<vmid>/bootstrap); "" → /var/lib/felhom-agent (mirrors provision.NewBackHalf).
	stateDir string

	opSeq uint64 // atomic; makes each op id unique per attempt

	// lastRes records the most recent successful Reconcile Result (v0.90.0, R-28 fast-tick source).
	// The fast-tick reads it to decide convergence: actionable drift is Planned − Pending > 0 (a
	// destructive pending_signature refusal is EXPECTED state, not drift to hammer on). lastOK is
	// false until the first successful pass.
	lastMu  sync.Mutex
	lastRes Result
	lastOK  bool
}

// LastResult returns the most recent successful Reconcile Result and whether one has been recorded
// (false until the first successful pass). Read by the fast-tick convergence source; safe for
// concurrent use.
func (e *Engine) LastResult() (Result, bool) {
	e.lastMu.Lock()
	defer e.lastMu.Unlock()
	return e.lastRes, e.lastOK
}

// recordResult stores the latest successful pass Result (called from reconcileOnce).
func (e *Engine) recordResult(res Result) {
	e.lastMu.Lock()
	e.lastRes = res
	e.lastOK = true
	e.lastMu.Unlock()
}

// EngineOptions configures a new Engine. Norm defaults to DefaultNormalizers, Logger
// to a discard logger, Gate to a no-verifier gate (benign-allow, destructive-pending).
type EngineOptions struct {
	API      GuestAPI
	Queue    *Queue
	Journal  *Journal
	Provider DesiredProvider
	Norm     FieldNormalizers
	Gate     *Gate
	HostID   string
	Logger   *slog.Logger
	// HostRunner enables the DR structural-bind swap (bring-up 4d, root pct ops). Optional —
	// engines that never run ModeDRGuestLoss may leave it nil.
	HostRunner proxmox.Runner
	// StateDir is the agent state dir ("" → /var/lib/felhom-agent); only the 4d swap reads it.
	StateDir string
}

// NewEngine builds an Engine. The Queue is shared (the single §10 choke point); the
// caller owns its lifecycle (Close on shutdown).
func NewEngine(opts EngineOptions) *Engine {
	norm := opts.Norm
	if norm == nil {
		norm = DefaultNormalizers()
	}
	logger := opts.Logger
	if logger == nil {
		logger = slog.New(slog.NewTextHandler(discard{}, nil))
	}
	provider := opts.Provider
	if provider == nil {
		provider = EmptyProvider{}
	}
	gate := opts.Gate
	if gate == nil {
		// No verifier configured: benign actions pass, destructive are pending. This is
		// the common slice-4 daemon state (no signers pinned, no desired state).
		gate = NewGate(nil, opts.HostID, nil, logger)
	}
	stateDir := opts.StateDir
	if stateDir == "" {
		stateDir = "/var/lib/felhom-agent"
	}
	return &Engine{
		api:      opts.API,
		queue:    opts.Queue,
		journal:  opts.Journal,
		provider: provider,
		norm:     norm,
		gate:     gate,
		hostID:   opts.HostID,
		logger:   logger,
		hostRun:  opts.HostRunner,
		stateDir: stateDir,
	}
}

// Result summarizes one Reconcile pass.
type Result struct {
	Planned  int
	Executed int     // succeeded
	Failed   int     // errored
	Pending  int     // destructive actions gated pending_signature (slice 10A — expected, not failed)
	Errors   []error // one per failed action
}

// Reconcile runs one convergence pass. It returns an error only on a pass-level
// failure (can't read desired/actual); per-action failures are counted in Result and
// do not abort the pass (other guests still converge).
func (e *Engine) Reconcile(ctx context.Context) (Result, error) {
	desired, err := e.provider.Desired(ctx)
	if err != nil {
		return Result{}, fmt.Errorf("reconcile: desired state: %w", err)
	}
	actual, err := e.readActual(ctx)
	if err != nil {
		return Result{}, fmt.Errorf("reconcile: actual state: %w", err)
	}

	actions := Plan(desired, actual, e.norm)
	res := Result{Planned: len(actions)}
	if len(actions) == 0 {
		e.logger.Debug("reconcile: no drift, no actions",
			"desired_guests", len(desired.Guests), "actual_guests", len(actual.Guests))
		return res, nil
	}

	// Every mutation passes the reversibility gate before the queue (doc 03 §4). Benign actions
	// are allowed unsigned; a DESTRUCTIVE delta (slice 10A: an explicit decommission) is refused
	// `pending_signature` when no operator signature is present — that is EXPECTED, not a failure:
	// 10A serves destructive intent but never executes it (the signed-op execution is 10B). So a
	// pending_signature refusal is counted as Pending and logged at INFO; any OTHER refusal (a
	// benign action denied, or a destructive one rejected for a different reason) is a real failure.
	type dispatched struct {
		act Action
		ch  <-chan error
	}
	var sent []dispatched
	for i := range actions {
		act := actions[i]
		dec := e.gate.Authorize(intentForAction(e.hostID, act), nil)
		if !dec.Allowed {
			if dec.Disposition == Destructive && dec.Reason == ReasonPendingSignature {
				res.Pending++
				e.logger.Info("reconcile: destructive action gated pending operator signature (slice 10B)",
					"vmid", act.VMID, "kind", act.Kind, "reason", dec.Reason)
				continue
			}
			res.Failed++
			res.Errors = append(res.Errors, fmt.Errorf("reconcile: gate refused %s vmid %d: %s",
				act.Kind, act.VMID, dec.Reason))
			e.logger.Error("reconcile: gate refused an action unexpectedly",
				"vmid", act.VMID, "kind", act.Kind, "reason", dec.Reason)
			continue
		}
		sent = append(sent, dispatched{act: act, ch: e.queue.Submit(act.VMID, func() error { return e.execute(ctx, act) })})
	}
	for _, d := range sent {
		if err := <-d.ch; err != nil {
			res.Failed++
			res.Errors = append(res.Errors, err)
			e.logger.Error("reconcile: action failed",
				"vmid", d.act.VMID, "kind", d.act.Kind, "err", err)
		} else {
			res.Executed++
			e.logger.Info("reconcile: action applied",
				"vmid", d.act.VMID, "kind", d.act.Kind, "reason", d.act.Reason)
		}
	}
	return res, nil
}

// execute dispatches one benign action against Proxmox and journals its lifecycle.
// Reconcile actions carry NO idempotency key (convergent — safe to re-run on drift);
// crash-safety comes from the in-flight journal records, not idempotency suppression.
func (e *Engine) execute(ctx context.Context, act Action) error {
	opID := e.nextOpID(act)
	e.append(JournalEntry{OpID: opID, VMID: act.VMID, Kind: string(act.Kind),
		Params: act.Params, State: OpStarted, At: time.Now().UTC()})

	var upid string
	var err error
	switch act.Kind {
	case ActionStart:
		upid, err = e.api.Start(ctx, act.VMID)
	case ActionStop:
		upid, err = e.api.Stop(ctx, act.VMID)
	case ActionSetConfig:
		upid, err = e.api.SetConfig(ctx, act.VMID, act.Params)
	case ActionResize:
		// Defensive grow-only guard at the executor: a resize size MUST be a "+<n>" grow.
		// The planner only ever emits grows, but never let a shrink reach Proxmox here.
		disk, size := act.Params["disk"], act.Params["size"]
		if !strings.HasPrefix(size, "+") {
			err = fmt.Errorf("reconcile: refusing non-grow resize size %q (data-losing shrink is a signed op)", size)
		} else {
			upid, err = e.api.ResizeLXC(ctx, act.VMID, disk, size)
		}
	case ActionDecommission:
		// Reaching here means a destructive decommission passed the gate (a verified signature) —
		// which only happens once 10B wires the signed-op executor. In 10A there is no signer, so
		// the gate refuses it before dispatch and this branch is unreachable. Fail safe loudly
		// rather than silently no-op, so a future signed path can't accidentally execute here.
		err = fmt.Errorf("reconcile: decommission executor is slice 10B (refusing to execute vmid %d)", act.VMID)
	default:
		err = fmt.Errorf("reconcile: unknown action kind %q", act.Kind)
	}
	if err != nil {
		e.append(JournalEntry{OpID: opID, VMID: act.VMID, Kind: string(act.Kind),
			State: OpFailed, At: time.Now().UTC()})
		return fmt.Errorf("reconcile: %s vmid %d: %w", act.Kind, act.VMID, err)
	}

	// Record the task id (if any) before awaiting it, so a crash mid-wait is
	// detectable on restart and the task status can be re-checked.
	e.append(JournalEntry{OpID: opID, VMID: act.VMID, Kind: string(act.Kind),
		UPID: upid, State: OpTaskRunning, At: time.Now().UTC()})

	if upid != "" {
		st, err := e.api.WaitTask(ctx, upid, proxmox.WaitOptions{})
		if err != nil { // WaitTask already errors on a non-OK exitstatus
			e.append(JournalEntry{OpID: opID, VMID: act.VMID, Kind: string(act.Kind),
				UPID: upid, State: OpFailed, At: time.Now().UTC()})
			return fmt.Errorf("reconcile: %s vmid %d: %w", act.Kind, act.VMID, err)
		}
		if st.ExitStatus != "OK" { // defensive — WaitTask should have errored
			e.append(JournalEntry{OpID: opID, VMID: act.VMID, Kind: string(act.Kind),
				UPID: upid, State: OpFailed, At: time.Now().UTC()})
			return fmt.Errorf("reconcile: %s vmid %d: exitstatus=%s", act.Kind, act.VMID, st.ExitStatus)
		}
	}
	// upid == "" is the synchronous path (slice-4 proven for SetConfig description).

	e.append(JournalEntry{OpID: opID, VMID: act.VMID, Kind: string(act.Kind),
		UPID: upid, State: OpSucceeded, At: time.Now().UTC()})
	return nil
}

// readActual reads observed state from Proxmox: run-state from the list, sizing +
// description from per-guest config. A GuestConfig read failure keeps the run-state
// (SpecKnown=false) rather than dropping the guest — matching the collector.
func (e *Engine) readActual(ctx context.Context) (ActualState, error) {
	lxc, err := e.api.ListLXC(ctx)
	if err != nil {
		return ActualState{}, err
	}
	guests := make(map[int]ActualGuest, len(lxc))
	for _, g := range lxc {
		// MaxDisk (bytes) comes from the list entry and is reliable independent of the
		// per-guest config read — it is the actual side of the grow comparison.
		a := ActualGuest{VMID: g.VMID, Run: normRun(g.Status), DiskBytes: g.MaxDisk}
		cfg, err := e.api.GuestConfig(ctx, g.VMID)
		if err != nil {
			e.logger.Warn("reconcile: GuestConfig failed; spec unknown (run-state kept)",
				"vmid", g.VMID, "err", err)
		} else {
			a.SpecKnown = true
			a.Cores = cfg.Cores
			a.MemoryMiB = cfg.Memory
			a.Description = guestDescription(cfg)
		}
		guests[g.VMID] = a
	}
	return ActualState{Guests: guests}, nil
}

// Run reconciles once immediately, then on every interval tick until ctx is done. A
// per-pass failure is logged and the loop continues (drift is corrected next tick).
// At slice 4 (EmptyProvider) every pass is a logged no-op.
func (e *Engine) Run(ctx context.Context, interval time.Duration) error {
	e.reconcileOnce(ctx)
	t := time.NewTicker(interval)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-t.C:
			e.reconcileOnce(ctx)
		}
	}
}

func (e *Engine) reconcileOnce(ctx context.Context) {
	res, err := e.Reconcile(ctx)
	if err != nil {
		e.logger.Error("reconcile: pass failed", "err", err)
		return
	}
	e.recordResult(res) // v0.90.0: fast-tick convergence source
	if res.Planned > 0 {
		e.logger.Info("reconcile: pass complete",
			"planned", res.Planned, "executed", res.Executed, "failed", res.Failed, "pending", res.Pending)
	}
}

// nextOpID builds a per-attempt unique op id (kind-vmid-seq) for journal correlation.
func (e *Engine) nextOpID(act Action) string {
	return string(act.Kind) + "-" + strconv.Itoa(act.VMID) + "-" + nextSeq(&e.opSeq)
}

// nextSeq atomically increments a counter and returns it as a string — the unique
// suffix that distinguishes journal op ids across attempts.
func nextSeq(p *uint64) string {
	return strconv.FormatUint(atomic.AddUint64(p, 1), 10)
}

// append journals a lifecycle record, logging (never failing the op on) a journal I/O
// error — the Proxmox op already happened; a missing journal line is a crash-recovery
// degradation, not a reason to abort.
func (e *Engine) append(rec JournalEntry) {
	if e.journal == nil {
		return
	}
	if err := e.journal.Append(rec); err != nil {
		e.logger.Error("reconcile: journal append failed", "op_id", rec.OpID, "state", rec.State, "err", err)
	}
}

// discard is an io.Writer sink for the default no-op logger.
type discard struct{}

func (discard) Write(p []byte) (int, error) { return len(p), nil }