felhom-agent/internal/localapi/opactions.go

package localapi

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

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

// The operator's actions for THIS agent (operator brief 2026-10-29, Part 1; felhom.eu `04` §6.1).
//
// THE DOOR IS THE HOST REPORT'S REPLY, the controller's shape: the operator presses a button on the hub's host page, the
// hub stores a row and pokes the box (the poke's out-of-cycle report gets the reply in seconds; without the tunnel the
// 15-minute cycle carries it), the reply lists `operator_actions: [{id, action, arg}]`, this agent acts ONCE per id, and
// the next report answers `operator_action_results: [{id, outcome, message}]` — the agent asks for that report at once.
// The result is re-sent until the reply stops listing the id.
//
// THE LIST IS CLOSED: one verb, restart_controller <vmid>. Anything else — a verb, a vmid that does not parse, a guest
// this agent did not provision — is refused and NOTHING is called. Pinned by TestOpActions_AgentClosedList.
//
// THE ACT is the supervisor's own (controllersupervisor.go): `systemctl restart felhom-controller-bootstrap.service` in
// the guest over the existing grants, with the supervisor's guards — not during a controller swap, not parked, not on a
// stopped, locked or vzdump-busy guest, not when the guest list or the vzdump state cannot be read — and the thrash brake
// counts these restarts too: 3 restarts (the supervisor's or the operator's) inside 15 minutes refuse the press, and
// presses stay refused for 30 minutes. It works with the controller stopped or stuck: the agent does it, not the
// controller. Every decision is one `audit:` line (the customer-visible audit form of reconcile.SlogAudit).

const (
	// OpRestartController is the agent's one operator verb.
	OpRestartController = "restart_controller"

	opOutcomeDone    = "done"
	opOutcomeRefused = "refused"
	opOutcomeFailed  = "failed"

	// opBrakeRefusal is how long presses stay refused after the brake refused one.
	opBrakeRefusal = 30 * time.Minute
	// opForgetAfter drops a started id the reply no longer lists.
	opForgetAfter = 24 * time.Hour

	defaultOpHealthPoll = 2 * time.Second
	// defaultOpHealthWait covers the bootstrap's pull-less rm/run (seconds), Docker's start period (10 s) and its first
	// probe (30 s interval) with room: the controller's own HEALTHCHECK is the answer.
	defaultOpHealthWait = 3 * time.Minute
	// opHealthTemplate reads running + health over the swap's `docker inspect -f` grant.
	opHealthTemplate = "{{.State.Running}}|{{if .State.Health}}{{.State.Health.Status}}{{else}}none{{end}}"
)

// operatorActionNames is THE agent's closed list. Adding an entry fails TestOpActions_AgentClosedList by design.
var operatorActionNames = []string{OpRestartController}

// OperatorActionNames returns the closed list (sorted copy).
func OperatorActionNames() []string {
	out := append([]string(nil), operatorActionNames...)
	sort.Strings(out)
	return out
}

type opActionState struct {
	mu      sync.Mutex
	started map[int64]time.Time
	results map[int64]hub.OperatorActionResult
}

// SetOperatorActionTrigger wires the report loop's out-of-band trigger (the result goes at once, not in 15 minutes).
func (s *Server) SetOperatorActionTrigger(f func()) { s.opTrigger = f }

// OnEnvelope is the report loop's hook (hub.EnvelopeObserver): it starts each newly listed action once and forgets
// the ids the reply no longer lists (the hub has their results).
func (s *Server) OnEnvelope(_ context.Context, env *hub.ControlEnvelope) {
	if env == nil {
		return
	}
	listed := map[int64]bool{}
	var start []hub.OperatorAction
	now := time.Now()
	s.opActs.mu.Lock()
	if s.opActs.started == nil {
		s.opActs.started = map[int64]time.Time{}
		s.opActs.results = map[int64]hub.OperatorActionResult{}
	}
	for _, a := range env.OperatorActions {
		listed[a.ID] = true
		if _, ok := s.opActs.started[a.ID]; ok {
			continue
		}
		s.opActs.started[a.ID] = now
		start = append(start, a)
	}
	for id, at := range s.opActs.started {
		if listed[id] {
			continue
		}
		if _, answered := s.opActs.results[id]; answered || now.Sub(at) > opForgetAfter {
			delete(s.opActs.started, id)
			delete(s.opActs.results, id)
		}
	}
	s.opActs.mu.Unlock()
	for _, a := range start {
		go s.runOperatorAction(a) // a restart waits for the health answer — never block the report loop
	}
}

// OperatorActionResults is the report's answer list, oldest id first.
func (s *Server) OperatorActionResults() []hub.OperatorActionResult {
	s.opActs.mu.Lock()
	defer s.opActs.mu.Unlock()
	out := make([]hub.OperatorActionResult, 0, len(s.opActs.results))
	for _, r := range s.opActs.results {
		out = append(out, r)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
	return out
}

func (s *Server) runOperatorAction(a hub.OperatorAction) {
	ctx, cancel := context.WithTimeout(context.Background(), s.opWait()+time.Minute)
	defer cancel()
	outcome, msg := opOutcomeRefused, fmt.Sprintf("unknown action %q — the agent's list holds restart_controller only", a.Action)
	if a.Action == OpRestartController {
		outcome, msg = s.operatorRestartController(ctx, a.ID, a.Arg)
	}
	s.logger.Warn("audit: operator action", "id", a.ID, "action", a.Action, "arg", a.Arg, "by", "the operator via the hub",
		"outcome", outcome, "message", msg)
	s.opActs.mu.Lock()
	if _, still := s.opActs.started[a.ID]; still {
		s.opActs.results[a.ID] = hub.OperatorActionResult{ID: a.ID, Outcome: outcome, Message: msg}
	}
	s.opActs.mu.Unlock()
	if s.opTrigger != nil {
		s.opTrigger()
	}
}

func (s *Server) opPoll() time.Duration {
	if s.opHealthPoll > 0 {
		return s.opHealthPoll
	}
	return defaultOpHealthPoll
}

func (s *Server) opWait() time.Duration {
	if s.opHealthWait > 0 {
		return s.opHealthWait
	}
	return defaultOpHealthWait
}

// operatorRestartController is the press: the supervisor's guards, the brake, the restart, then the wait for the
// controller's healthy answer. Returns the outcome and the sentence the host page shows.
func (s *Server) operatorRestartController(ctx context.Context, id int64, arg string) (string, string) {
	vmid, err := strconv.Atoi(arg)
	if err != nil || strconv.Itoa(vmid) != arg {
		return opOutcomeRefused, fmt.Sprintf("refused: %q is not a guest id", arg)
	}
	if s.staleLock == nil || s.guestExec == nil {
		return opOutcomeRefused, "refused: the controller supervisor is not wired on this agent"
	}
	if !s.provisionedGuest(vmid) {
		return opOutcomeRefused, fmt.Sprintf("refused: guest %d is not a guest this agent provisioned", vmid)
	}
	guests, err := s.staleLock.Guests(ctx)
	if err != nil {
		return opOutcomeRefused, "refused: the guest list cannot be read — ownership unproven, nothing touched"
	}
	status := ""
	for _, g := range guests {
		if g.VMID == vmid {
			status = g.Status
		}
	}
	if status != "running" {
		return opOutcomeRefused, fmt.Sprintf("refused: guest %d is not running (%s) — the guest-power watchdog owns it", vmid, orDash(status))
	}
	if s.controllerParked(vmid) {
		return opOutcomeRefused, fmt.Sprintf("refused: guest %d's controller is parked (remove the controller-parked marker first)", vmid)
	}
	s.swapMu.Lock()
	swapping := s.swapInFlight[vmid]
	s.swapMu.Unlock()
	if swapping {
		return opOutcomeRefused, "refused: a controller swap (update) is in flight — it owns the controller"
	}
	if lock, _, err := s.staleLock.Lock(ctx, vmid); err != nil {
		return opOutcomeRefused, "refused: the guest lock cannot be read — nothing touched"
	} else if lock != "" {
		return opOutcomeRefused, fmt.Sprintf("refused: guest %d is locked (%s) — another operation owns it", vmid, lock)
	}
	if busy, berr := s.staleLock.BackupRunning(ctx, vmid); berr != nil || busy {
		return opOutcomeRefused, "refused: a whole-guest backup (vzdump) may be in flight for the guest"
	}

	// The brake: every restart counts — the supervisor's and the operator's.
	now := s.clock()
	s.ctrlSup.mu.Lock()
	st := s.supState(vmid)
	if now.Before(st.opBrakeUntil) {
		until := st.opBrakeUntil
		s.ctrlSup.mu.Unlock()
		return opOutcomeRefused, fmt.Sprintf("refused: 3 restarts in 15 minutes — presses are refused until %s UTC", until.UTC().Format("15:04"))
	}
	st.restarts = pruneBefore(st.restarts, now.Add(-controllerCrashloopWindow))
	if len(st.restarts) >= controllerCrashloopMax {
		st.opBrakeUntil = now.Add(opBrakeRefusal)
		until := st.opBrakeUntil
		s.ctrlSup.mu.Unlock()
		return opOutcomeRefused, fmt.Sprintf("refused: 3 restarts in 15 minutes — presses are refused until %s UTC", until.UTC().Format("15:04"))
	}
	reason := fmt.Sprintf("the operator via the hub (action #%d)", id)
	st.restarts = append(st.restarts, now)
	st.restartsTotal++
	st.lastRestartAt = now
	st.lastReason = reason
	st.notRunningSeen = 0
	s.ctrlSup.mu.Unlock()

	s.logger.Warn("controller-supervisor: RESTARTING the controller on the operator's press", "vmid", vmid, "action_id", id, "unit", bootstrapUnit)
	began := time.Now()
	if _, err := s.guestExec.GuestExec(ctx, vmid, "systemctl", "restart", bootstrapUnit); err != nil {
		return opOutcomeFailed, fmt.Sprintf("the restart of guest %d's controller failed: %v", vmid, err)
	}
	for time.Since(began) < s.opWait() {
		out, err := s.guestExec.GuestExec(ctx, vmid, "docker", "inspect", "-f", opHealthTemplate, controllerContainer)
		if err == nil {
			f := strings.SplitN(strings.TrimSpace(out), "|", 2)
			if len(f) == 2 && f[0] == "true" && (f[1] == "healthy" || f[1] == "none") {
				secs := int(time.Since(began).Round(time.Second) / time.Second)
				return opOutcomeDone, fmt.Sprintf("restarted; the controller answered after %d s (Docker's health check: healthy)", secs)
			}
		}
		select {
		case <-ctx.Done():
			return opOutcomeFailed, "restarted, but the wait for its answer was cut"
		case <-time.After(s.opPoll()):
		}
	}
	return opOutcomeFailed, fmt.Sprintf("restarted, but no healthy answer within %s — look at the box", s.opWait().Round(time.Second))
}

func orDash(s string) string {
	if s == "" {
		return "not listed"
	}
	return s
}