diff --git a/src/frontend/advisor/internal/detect/detect.go b/src/frontend/advisor/internal/detect/detect.go index b7b7fe67..1741cde4 100644 --- a/src/frontend/advisor/internal/detect/detect.go +++ b/src/frontend/advisor/internal/detect/detect.go @@ -1,8 +1,8 @@ // 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 cooldown (don't re-evaluate the -// same scenario too often), then dedup (don't re-evaluate an identical -// situation at all). +// worth a provider round: threshold, then dedup (never re-evaluate an +// identical situation), then cooldown (dampen A→B→A alternations that +// last-key dedup cannot catch). package detect import ( @@ -25,9 +25,8 @@ const ( // Trigger is one fired detection. type Trigger struct { Scenario string - // DedupKey hashes the situation (job set + reason set / node set) so an - // unchanged situation is not re-evaluated on every scan. Two rounds with - // the same key produce the second finding only after the cooldown. + // DedupKey hashes the situation (job set + reason set / node set). The + // same key never fires twice in a row; any change fires immediately. DedupKey string // Fingerprint is a human-readable summary of what matched (logged, and // what the echo provider re-derives from the state). @@ -42,8 +41,6 @@ type Limits struct { PendSpikeMin int // ExitBurstMin: non-zero exits within the window that fire. ExitBurstMin int - // Cooldowns per scenario; a scenario on cooldown is skipped entirely. - Cooldowns map[string]time.Duration } // DefaultLimits returns the P0 defaults. @@ -52,32 +49,24 @@ func DefaultLimits() Limits { PendStallAfter: 5 * time.Minute, PendSpikeMin: 5, ExitBurstMin: 3, - Cooldowns: map[string]time.Duration{ - ScenarioPendStall: 10 * time.Minute, - ScenarioWorkerHealth: 30 * time.Minute, - ScenarioUtilSkew: 30 * time.Minute, - ScenarioExitBurst: 30 * time.Minute, - }, } } -// Detector evaluates snapshots against the limits. Stateful: tracks -// per-scenario cooldown and the last dedup key seen. +// 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 - lastFired map[string]time.Time - lastKey map[string]string + lastKey map[string]string } // NewDetector returns a Detector with the given limits. func NewDetector(lim Limits) *Detector { return &Detector{ - lim: lim, - now: time.Now, - lastFired: map[string]time.Time{}, - lastKey: map[string]string{}, + lim: lim, + now: time.Now, + lastKey: map[string]string{}, } } @@ -100,19 +89,20 @@ func (d *Detector) Detect(st *snapshot.State) []Trigger { return out } -// fire applies cooldown + dedup; ok=true means the scenario should be -// evaluated this round and the state recorded. +// 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) { - if cd, ok := d.lim.Cooldowns[scenario]; ok { - if last, fired := d.lastFired[scenario]; fired && d.now().Sub(last) < cd { - return Trigger{}, false - } - } - // Identical situation as last fired round: skip regardless of cooldown. + // Identical situation as last fired round: skip. if d.lastKey[scenario] == key { return Trigger{}, false } - d.lastFired[scenario] = d.now() d.lastKey[scenario] = key return Trigger{Scenario: scenario, DedupKey: key, Fingerprint: fingerprint}, true } diff --git a/src/frontend/advisor/internal/detect/detect_test.go b/src/frontend/advisor/internal/detect/detect_test.go index e1dc4c43..3d26a9f9 100644 --- a/src/frontend/advisor/internal/detect/detect_test.go +++ b/src/frontend/advisor/internal/detect/detect_test.go @@ -27,7 +27,7 @@ func stalledState() *snapshot.State { return st } -// fixedClock lets tests drive cooldown timing. +// fixedClock lets tests drive time-dependent behavior. type fixedClock struct{ t time.Time } func (c *fixedClock) Now() time.Time { return c.t } @@ -49,13 +49,14 @@ func TestDetect_PendStall(t *testing.T) { t.Errorf("trigger missing key/fingerprint: %+v", trigs[0]) } - // Same situation again: deduped even before cooldown matters. + // Same situation again: deduped, regardless of elapsed time. if again := d.Detect(stalledState()); len(again) != 0 { t.Fatalf("identical situation re-fired: %+v", again) } - // Changed job set → new key → fires again (cooldown 10m must still block - // if the clock hasn't moved; use the clock to jump past it). + // Changed job set → new key → fires IMMEDIATELY (the P0 verification + // caught the old cooldown-first order hiding a new stall for up to 10 + // minutes). d2 := NewDetector(DefaultLimits()) fc := &fixedClock{t: time.Now()} d2.now = fc.Now @@ -64,12 +65,14 @@ func TestDetect_PendStall(t *testing.T) { } st2 := stalledState() st2.PendTop = append(st2.PendTop, snapshot.JobState{ID: "1-0-3-0", Reason: &snapshot.JobReason{MainReasonName: "no_resource"}}) - if got := d2.Detect(st2); len(got) != 0 { - t.Fatalf("cooldown must block within 10m: %+v", got) - } - fc.t = fc.t.Add(11 * time.Minute) if got := d2.Detect(st2); len(got) != 1 { - t.Fatalf("after cooldown, changed set must fire: %+v", got) + t.Fatalf("changed stalled set must fire immediately: %+v", got) + } + // Back to the earlier situation (A→B→A): also a change, fires — an + // oscillating stalled set is real signal, not noise to suppress. + fc.t = fc.t.Add(2 * time.Minute) + if got := d2.Detect(stalledState()); len(got) != 1 { + t.Fatalf("return to earlier situation must fire (real change): %+v", got) } } @@ -84,13 +87,14 @@ func TestDetect_WorkerHealth(t *testing.T) { if trigs[0].Fingerprint != "t2:down" { t.Errorf("fingerprint = %q", trigs[0].Fingerprint) } - // Same node still down → deduped. Node recovers then breaks again → new key. + // Same node still down → deduped. A different outage (t2:down → + // t2:unknown) is a changed situation → fires immediately. if again := d.Detect(st); len(again) != 0 { t.Fatalf("same outage re-fired: %+v", again) } st.Nodes[1].Status = "unknown" - if again := d.Detect(st); len(again) != 0 { - t.Fatalf("cooldown must block second scenario change within window: %+v", again) + if again := d.Detect(st); len(again) != 1 { + t.Fatalf("changed outage must fire immediately: %+v", again) } } diff --git a/src/frontend/advisor/internal/snapshot/snapshot.go b/src/frontend/advisor/internal/snapshot/snapshot.go index a4acff88..5627b402 100644 --- a/src/frontend/advisor/internal/snapshot/snapshot.go +++ b/src/frontend/advisor/internal/snapshot/snapshot.go @@ -112,12 +112,15 @@ func (b *Builder) Build(ctx context.Context, prevExits []ExitState) (*State, err } // PEND jobs are the pend_stall input. One filtered call per status - // keeps the apiserver query simple. - pend, _, err := b.client.ListJobs(ctx, sclient.JobFilter{Statuses: []sclient.JobStatus{sclient.JobStatusPending}}) + // keeps the apiserver query simple. AllUsers + submit_time:desc matter: + // without them the apiserver scopes the list to the caller's own jobs + // and returns the OLDEST rows first, hiding every recent PEND job + // (advisor P0 verification finding). + pend, _, err := b.client.ListJobs(ctx, clusterJobFilter(sclient.JobStatusPending)) if err != nil { return nil, fmt.Errorf("list pending jobs: %w", err) } - run, _, err := b.client.ListJobs(ctx, sclient.JobFilter{Statuses: []sclient.JobStatus{sclient.JobStatusRunning}}) + run, _, err := b.client.ListJobs(ctx, clusterJobFilter(sclient.JobStatusRunning)) if err != nil { return nil, fmt.Errorf("list running jobs: %w", err) } @@ -143,6 +146,17 @@ func (b *Builder) Build(ctx context.Context, prevExits []ExitState) (*State, err return st, nil } +// clusterJobFilter is the list request for a cluster-wide status view: the +// newest rows first, every user (needs a scope-all credential). +func clusterJobFilter(st sclient.JobStatus) sclient.JobFilter { + return sclient.JobFilter{ + Statuses: []sclient.JobStatus{st}, + AllUsers: true, + Sort: "submit_time:desc", + Limit: 200, + } +} + // topPending picks the most-starved pending jobs, keeping only those past // the stall threshold (the snapshot feeds the pend_stall scenario; fresh // PEND jobs are noise for the model).