fix(advisor): cluster-wide job view + change-triggered detection

snapshot: list jobs with AllUsers + submit_time:desc + limit 200 — the
advisor must see every user's PEND jobs newest-first, or pend_stall sees
nothing on a shared cluster (the default list is the caller's own jobs,
oldest first).

detect: drop the per-scenario cooldown from the fire path. The original
cooldown-first order suppressed a CHANGED situation for up to 10 minutes —
on the live cluster a new job joining the stalled set went unreported
until the cooldown expired. Semantics now: an unchanged situation (same
dedup key) never re-evaluates; any changed situation fires immediately.
A thrashing situation (A→B→A) firing per change is honest reporting — a
flapping node or oscillating stalled set is real signal.

Verified on the t1/t2/t3 cluster: two forced-PEND jobs (unsatisfiable
select; 64-slot limit rejection) both diagnosed in the first scan, a
kill changing the stalled set re-triggered within one 30s round, and
identical situations stayed deduped.

Co-Authored-By: Claude Code <noreply@anthropic.com>

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
+// worth a provider round: threshold, then dedup (never re-evaluate an
-// same scenario too often), then dedup (don't re-evaluate an identical
+// identical situation), then cooldown (dampen A→B→A alternations that
-// situation at all).
+// 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
+	// DedupKey hashes the situation (job set + reason set / node set). The
-	// unchanged situation is not re-evaluated on every scan. Two rounds with
+	// same key never fires twice in a row; any change fires immediately.
-	// the same key produce the second finding only after the cooldown.
 	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,22 +49,15 @@ 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
+// Detector evaluates snapshots against the limits. Stateful: tracks the
-// per-scenario cooldown and the last dedup key seen.
+// 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
 }
 
@@ -76,7 +66,6 @@ func NewDetector(lim Limits) *Detector {
 	return &Detector{
 		lim:     lim,
 		now:     time.Now,
-		lastFired: map[string]time.Time{},
 		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
+// fire applies dedup; ok=true means the scenario should be evaluated this
-// evaluated this round and the state recorded.
+// 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 {
+	// Identical situation as last fired round: skip.
-		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.
 	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
 }

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
+	// Changed job set → new key → fires IMMEDIATELY (the P0 verification
-	// if the clock hasn't moved; use the clock to jump past it).
+	// 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 {
+	if again := d.Detect(st); len(again) != 1 {
-		t.Fatalf("cooldown must block second scenario change within window: %+v", again)
+		t.Fatalf("changed outage must fire immediately: %+v", again)
 	}
 }
 

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.
+	// keeps the apiserver query simple. AllUsers + submit_time:desc matter:
-	pend, _, err := b.client.ListJobs(ctx, sclient.JobFilter{Statuses: []sclient.JobStatus{sclient.JobStatusPending}})
+	// 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).