myxce/src/frontend/advisor/internal/detect/detect.go

// Package detect holds the advisor's deterministic triggers — cheap, no
// LLM. A detector looks at one snapshot and decides whether a scenario is
// worth a provider round: threshold, then dedup. The dedup key hashes the
// situation: an unchanged situation never re-fires, any change fires
// immediately, and a situation that resolves fully disarms its key so a
// later recurrence reports again (fire/clear pair).
package detect

import (
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"time"

	"advisor/internal/snapshot"
)

// Scenario names, shared by detector, provider, and log.
const (
	ScenarioPendStall    = "pend_stall"
	ScenarioWorkerHealth = "worker_health"
	ScenarioUtilSkew     = "util_skew"
	ScenarioExitBurst    = "exit_burst"
)

// Trigger is one fired detection.
type Trigger struct {
	Scenario string
	// DedupKey hashes the situation (job set + reason set / node set). The
	// same key never fires twice in a row; any change fires immediately;
	// a full resolution disarms the key (clear) so a recurrence fires again.
	DedupKey string
	// Fingerprint is a human-readable summary of what matched (logged, and
	// what the echo provider re-derives from the state).
	Fingerprint string
}

// Limits configures the detectors.
type Limits struct {
	// PendStallAfter: PEND wait counted as stalled (default 5m).
	PendStallAfter time.Duration
	// PendSpikeMin: absolute PEND count growth that fires without waits.
	PendSpikeMin int
	// ExitBurstMin: non-zero exits within the window that fire.
	ExitBurstMin int
}

// DefaultLimits returns the P0 defaults.
func DefaultLimits() Limits {
	return Limits{
		PendStallAfter: 5 * time.Minute,
		PendSpikeMin:   5,
		ExitBurstMin:   3,
	}
}

// Detector evaluates snapshots against the limits. Stateful: tracks the
// last dedup key per scenario.
type Detector struct {
	lim Limits
	now func() time.Time // injectable clock for tests

	lastKey map[string]string
}

// NewDetector returns a Detector with the given limits.
func NewDetector(lim Limits) *Detector {
	return &Detector{
		lim:     lim,
		now:     time.Now,
		lastKey: map[string]string{},
	}
}

// Detect runs all detectors over the snapshot and returns the scenarios to
// evaluate this round, in stable order. Each detector arms its dedup key
// when it fires and disarms it (clear) when its condition is absent, so a
// resolved-then-recurred situation reports again.
func (d *Detector) Detect(st *snapshot.State) []Trigger {
	var out []Trigger
	if t, ok := d.pendStall(st); ok {
		out = append(out, t)
	}
	if t, ok := d.workerHealth(st); ok {
		out = append(out, t)
	}
	if t, ok := d.utilSkew(st); ok {
		out = append(out, t)
	}
	if t, ok := d.exitBurst(st); ok {
		out = append(out, t)
	}
	return out
}

// fire applies dedup; ok=true means the scenario should be evaluated this
// round and the state recorded.
//
// Semantics (P0 verification finding): an unchanged situation never
// re-evaluates (the dedup key); a CHANGED situation always fires immediately.
// The original cooldown-first order hid a genuinely new stall for up to 10
// minutes on the live cluster — unacceptable for "things are getting worse".
// Situations that thrash (A→B→A) therefore also fire per change, which is
// honest: a flapping node or an oscillating stalled set is real signal.
func (d *Detector) fire(scenario, key, fingerprint string) (Trigger, bool) {
	// Identical situation as last fired round: skip.
	if d.lastKey[scenario] == key {
		return Trigger{}, false
	}
	d.lastKey[scenario] = key
	return Trigger{Scenario: scenario, DedupKey: key, Fingerprint: fingerprint}, true
}

// clear forgets the scenario's dedup key when its condition is absent this
// round. Without this, a situation that resolves and then recurs identically
// (same node goes down twice, the same stall re-forms) is swallowed as
// "unchanged" forever — observed live (2026-10-20): t3 went down, recovered,
// went down again, and the second outage never reported. The key is armed by
// fire() and disarmed by clear(); each detector's healthy/empty early-return
// is the disarm point.
func (d *Detector) clear(scenario string) {
	delete(d.lastKey, scenario)
}

func (d *Detector) pendStall(st *snapshot.State) (Trigger, bool) {
	if len(st.PendTop) == 0 && st.Cluster.PendingJobs < d.lim.PendSpikeMin {
		d.clear(ScenarioPendStall)
		return Trigger{}, false
	}
	// Key: the stalled job set and their reasons.
	key := hashOf(reasonKey(st.PendTop))
	fp := "pending=" + itoa(st.Cluster.PendingJobs) + " stalled=" + itoa(len(st.PendTop))
	return d.fire(ScenarioPendStall, key, fp)
}

func (d *Detector) workerHealth(st *snapshot.State) (Trigger, bool) {
	var bad []string
	for _, n := range st.Nodes {
		switch n.Status {
		case "down", "unknown", "unavail", "closed":
			bad = append(bad, n.Name+":"+n.Status)
		}
	}
	if len(bad) == 0 {
		d.clear(ScenarioWorkerHealth)
		return Trigger{}, false
	}
	key := hashOf(join(bad, ","))
	return d.fire(ScenarioWorkerHealth, key, join(bad, ","))
}

func (d *Detector) utilSkew(st *snapshot.State) (Trigger, bool) {
	var hot, idle []string
	for _, n := range st.Nodes {
		if n.CoresTotal < 4 {
			continue
		}
		switch {
		case n.Utilization >= 0.8:
			hot = append(hot, n.Name)
		case n.Utilization <= 0.2:
			idle = append(idle, n.Name)
		}
	}
	if len(hot) == 0 || len(idle) == 0 {
		d.clear(ScenarioUtilSkew)
		return Trigger{}, false
	}
	key := hashOf("hot=" + join(hot, ",") + ";idle=" + join(idle, ","))
	fp := "hot=[" + join(hot, ",") + "] idle=[" + join(idle, ",") + "]"
	return d.fire(ScenarioUtilSkew, key, fp)
}

func (d *Detector) exitBurst(st *snapshot.State) (Trigger, bool) {
	var ids []string
	for _, x := range st.RecentExits {
		if x.ExitCode != 0 {
			ids = append(ids, x.ID+":"+itoa(x.ExitCode))
		}
	}
	if len(ids) < d.lim.ExitBurstMin {
		// Note: NOT cleared. RecentExits is a per-scan window (exits are
		// consumed once, see snapshot Build); an empty window is the steady
		// state, not "the burst resolved" — arming/disarming here would
		// re-fire a burst whose situation never actually changed.
		return Trigger{}, false
	}
	key := hashOf(join(ids, ","))
	return d.fire(ScenarioExitBurst, key, join(ids, ","))
}

// reasonKey builds a stable string from the pending set: job id + reason.
func reasonKey(pend []snapshot.JobState) string {
	parts := make([]string, 0, len(pend))
	for _, j := range pend {
		r := "-"
		if j.Reason != nil {
			r = j.Reason.MainReasonName
		}
		parts = append(parts, j.ID+"="+r)
	}
	return join(parts, "|")
}

func hashOf(s string) string {
	sum := sha256.Sum256([]byte(s))
	return hex.EncodeToString(sum[:8])
}

func join(parts []string, sep string) string {
	out := ""
	for i, p := range parts {
		if i > 0 {
			out += sep
		}
		out += p
	}
	return out
}

func itoa(i int) string {
	b, _ := json.Marshal(i)
	return string(b)
}