feat(advisor): snapshot builder + deterministic detectors

- internal/snapshot: compact cluster state via schedulerclient (nodes,
  PEND top-N most-starved with parsed pending_reason, cluster sums, queue
  garnish, recent exits). Field whitelists + stall-threshold filter keep
  the state a few KiB. Module-local PEND-reason table mirrors
  mcp-server/internal/pendreason.go (lift into schedulerclient is a P1
  cleanup).
- internal/detect: pend_stall / worker_health / util_skew / exit_burst
  triggers — threshold, then per-scenario cooldown, then dedup on a hash
  of the situation (job set + reasons / node set) so an unchanged cluster
  is never re-evaluated.

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

src/frontend/advisor/internal/snapshot/pendreason.go

@@ -0,0 +1,95 @@
+package snapshot
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+// pendingReason mirrors the apiserver rawPendReason (convert_pure.go),
+// surfaced by schedulerclient as the JSON-encoded string
+// Extensions["pending_reason"].
+type pendingReason struct {
+	MainReasonName string `json:"main_reason_name"`
+}
+
+// reasonScenario maps a PREASON_* name to the human-readable scenario.
+// This mirrors mcp-server/internal/pendreason.go (module-local copy: the
+// advisor and mcp-server are separate modules; lifting the table into
+// schedulerclient is a deliberate P1 cleanup, not forgotten).
+func reasonScenario(name string) string {
+	switch name {
+	case "bad_resreq":
+		return "the resource request is invalid or cannot be satisfied"
+	case "no_resource":
+		return "insufficient resource capacity"
+	case "match_fail":
+		return "no host matches the job's host-selection condition"
+	case "limit_global":
+		return "a global quota is exhausted"
+	case "limit_host":
+		return "a host-level quota is exhausted"
+	case "queue_closed":
+		return "the queue is closed (not accepting new jobs)"
+	case "queue_draining":
+		return "the queue is draining (no new jobs until it empties)"
+	case "time_window":
+		return "outside the queue's dispatch window"
+	case "host_unavail":
+		return "the host is unavailable"
+	default:
+		return fmt.Sprintf("unknown reason (%s)", name)
+	}
+}
+
+// parsePendingReason extracts the main reason name + scenario from a job's
+// Extensions. Returns nil when absent or unparseable — a missing reason is
+// reported as such, never guessed.
+func parsePendingReason(ext map[string]any) *JobReason {
+	raw, ok := ext["pending_reason"]
+	if !ok {
+		return nil
+	}
+	var pr pendingReason
+	switch v := raw.(type) {
+	case string:
+		if v == "" {
+			return nil
+		}
+		if err := json.Unmarshal([]byte(v), &pr); err != nil {
+			return nil
+		}
+	case map[string]any:
+		b, _ := json.Marshal(v)
+		if err := json.Unmarshal(b, &pr); err != nil {
+			return nil
+		}
+	default:
+		return nil
+	}
+	if pr.MainReasonName == "" {
+		return nil
+	}
+	return &JobReason{
+		MainReasonName: pr.MainReasonName,
+		Scenario:       reasonScenario(pr.MainReasonName),
+	}
+}
+
+// round2 keeps snapshot numbers at two decimals so the JSON stays small.
+func round2(f float64) float64 {
+	return float64(int(f*100+0.5*sign(f))) / 100
+}
+
+func sign(f float64) float64 {
+	if f < 0 {
+		return -1
+	}
+	return 1
+}
+
+func frac(a, b int) float64 {
+	if b == 0 {
+		return 0
+	}
+	return float64(a) / float64(b)
+}

src/frontend/advisor/internal/snapshot/snapshot.go

@@ -0,0 +1,179 @@
+// Package snapshot builds the compact JSON cluster snapshot the providers
+// evaluate. It reads through schedulerclient (apiserver REST) and trims
+// hard: field whitelists and top-N truncation, because a cheap LLM's
+// attention is the scarce resource.
+package snapshot
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"time"
+
+	sclient "schedulerclient"
+)
+
+// Builder pulls cluster state via the schedulerclient.
+type Builder struct {
+	client sclient.Client
+	// PendTop bounds the pending-job list sent to the model (most-starved
+	// first). 20 keeps the state under a few KiB.
+	PendTop int
+	// StallThreshold is the PEND wait beyond which a job counts as stalled
+	// (default 5m, matching the pend_stall detector).
+	StallThreshold time.Duration
+}
+
+// NewBuilder returns a Builder with the defaults applied.
+func NewBuilder(client sclient.Client) *Builder {
+	return &Builder{client: client, PendTop: 20, StallThreshold: 5 * time.Minute}
+}
+
+// State is the snapshot shape (also the echo provider's parse target — the
+// two must evolve together; see provider/echo.go echoState).
+type State struct {
+	GeneratedAt string      `json:"generated_at"`
+	Cluster     ClusterSum  `json:"cluster"`
+	Nodes       []NodeState `json:"nodes"`
+	PendTop     []JobState  `json:"pend_top"`
+	// RecentExits carries jobs that exited since the previous scan
+	// (exit_burst detector input); empty when none.
+	RecentExits []ExitState `json:"recent_exits,omitempty"`
+	Queues      []QueueSum  `json:"queues,omitempty"`
+}
+
+type ClusterSum struct {
+	PendingJobs int     `json:"pending_jobs"`
+	RunningJobs int     `json:"running_jobs"`
+	Utilization float64 `json:"utilization"`
+}
+
+type NodeState struct {
+	Name        string  `json:"name"`
+	Status      string  `json:"status"`
+	CoresTotal  int     `json:"cores_total"`
+	CoresUsed   int     `json:"cores_used"`
+	Utilization float64 `json:"utilization"`
+}
+
+type JobState struct {
+	ID          string     `json:"id"`
+	Name        string     `json:"name"`
+	Owner       string     `json:"owner"`
+	Queue       string     `json:"queue"`
+	WaitSeconds float64    `json:"wait_seconds"`
+	Reason      *JobReason `json:"reason,omitempty"`
+	Cores       int        `json:"cores,omitempty"`
+}
+
+// JobReason is the parsed pending_reason (main_reason_name + the
+// human-readable scenario name from the translation table).
+type JobReason struct {
+	MainReasonName string `json:"main_reason_name,omitempty"`
+	Scenario       string `json:"scenario,omitempty"`
+}
+
+type ExitState struct {
+	ID       string `json:"id"`
+	Name     string `json:"name,omitempty"`
+	ExitCode int    `json:"exit_code"`
+}
+
+type QueueSum struct {
+	Name        string `json:"name"`
+	PendingJobs int    `json:"pending_jobs"`
+	RunningJobs int    `json:"running_jobs"`
+}
+
+// Build fetches jobs/nodes/queues and assembles the trimmed snapshot. The
+// previous-exits input lets exit_burst see what changed since the last
+// round; pass nil on the first round.
+func (b *Builder) Build(ctx context.Context, prevExits []ExitState) (*State, error) {
+	st := &State{GeneratedAt: time.Now().UTC().Format(time.RFC3339)}
+
+	nodes, err := b.client.ListNodes(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("list nodes: %w", err)
+	}
+	st.Nodes = make([]NodeState, 0, len(nodes))
+	var totalCores, usedCores int
+	for _, n := range nodes {
+		util := 0.0
+		if n.CoresTotal > 0 {
+			util = float64(n.CoresUsed) / float64(n.CoresTotal)
+		}
+		st.Nodes = append(st.Nodes, NodeState{
+			Name: n.Name, Status: n.Status,
+			CoresTotal: n.CoresTotal, CoresUsed: n.CoresUsed,
+			Utilization: round2(util),
+		})
+		totalCores += n.CoresTotal
+		usedCores += n.CoresUsed
+	}
+
+	// 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}})
+	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}})
+	if err != nil {
+		return nil, fmt.Errorf("list running jobs: %w", err)
+	}
+	st.Cluster = ClusterSum{
+		PendingJobs: len(pend),
+		RunningJobs: len(run),
+		Utilization: round2(frac(usedCores, totalCores)),
+	}
+
+	st.PendTop = topPending(pend, b.PendTop, b.StallThreshold)
+
+	if prevExits != nil {
+		st.RecentExits = prevExits
+	}
+
+	queues, err := b.client.ListQueues(ctx)
+	if err == nil { // queues are garnish; never fail the snapshot for them
+		for _, q := range queues {
+			st.Queues = append(st.Queues, QueueSum{Name: q.Name, PendingJobs: q.PendingJobs, RunningJobs: q.RunningJobs})
+		}
+	}
+
+	return st, nil
+}
+
+// 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).
+func topPending(pend []*sclient.URMJob, n int, stall time.Duration) []JobState {
+	type keyed struct {
+		js   JobState
+		wait time.Duration
+	}
+	var out []keyed
+	for _, j := range pend {
+		wait := time.Since(j.SubmissionTime)
+		if wait < stall {
+			continue
+		}
+		js := JobState{
+			ID: j.ID, Name: j.Name, Owner: j.Owner, Queue: j.Queue,
+			WaitSeconds: round2(wait.Seconds()),
+			Cores:       j.Cores,
+		}
+		if pr := parsePendingReason(j.Extensions); pr != nil {
+			js.Reason = pr
+		}
+		out = append(out, keyed{js, wait})
+	}
+	sort.Slice(out, func(i, k int) bool { return out[i].wait > out[k].wait })
+	if len(out) > n {
+		out = out[:n]
+	}
+	res := make([]JobState, 0, len(out))
+	for _, k := range out {
+		res = append(res, k.js)
+	}
+	return res
+}

src/frontend/advisor/internal/snapshot/snapshot_test.go

@@ -0,0 +1,215 @@
+package snapshot
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	sclient "schedulerclient"
+)
+
+// fakeClient implements the read side of sclient.Client for tests.
+type fakeClient struct {
+	jobs   map[sclient.JobStatus][]*sclient.URMJob
+	nodes  []*sclient.URMNode
+	queues []*sclient.URMQueue
+}
+
+func (f *fakeClient) ListJobs(_ context.Context, filter sclient.JobFilter) ([]*sclient.URMJob, *sclient.Page, error) {
+	var out []*sclient.URMJob
+	for _, s := range filter.Statuses {
+		out = append(out, f.jobs[s]...)
+	}
+	return out, &sclient.Page{Total: len(out)}, nil
+}
+
+func (f *fakeClient) ListNodes(_ context.Context) ([]*sclient.URMNode, error) { return f.nodes, nil }
+func (f *fakeClient) ListQueues(_ context.Context) ([]*sclient.URMQueue, error) {
+	return f.queues, nil
+}
+
+func (f *fakeClient) GetJob(context.Context, string) (*sclient.URMJob, error) {
+	panic("not needed")
+}
+func (f *fakeClient) GetJobLogs(context.Context, string, sclient.LogOpts) ([]string, error) {
+	panic("not needed")
+}
+func (f *fakeClient) GetClusterMetrics(context.Context) (*sclient.Metrics, error) {
+	panic("not needed")
+}
+func (f *fakeClient) SubmitJob(context.Context, sclient.JobSpec) (*sclient.SubmitJobResult, error) {
+	panic("observe-only: writes must be unreachable")
+}
+func (f *fakeClient) CancelJob(context.Context, string) error { panic("observe-only") }
+func (f *fakeClient) HoldJob(context.Context, string) error   { panic("observe-only") }
+func (f *fakeClient) ReleaseJob(context.Context, string) error {
+	panic("observe-only")
+}
+func (f *fakeClient) CheckpointJob(context.Context, string) error { panic("observe-only") }
+func (f *fakeClient) RestartJob(context.Context, string) (*sclient.RestartJobResult, error) {
+	panic("observe-only")
+}
+func (f *fakeClient) WatchJobs(context.Context) (<-chan sclient.JobEvent, error) {
+	panic("not needed in snapshot tests")
+}
+
+func testClient() *fakeClient {
+	old := time.Now().Add(-2 * time.Hour)
+	recent := time.Now().Add(-30 * time.Second)
+	return &fakeClient{
+		jobs: map[sclient.JobStatus][]*sclient.URMJob{
+			sclient.JobStatusPending: {
+				{ID: "1-0-1-0", Name: "stalled", Owner: "u1", Queue: "normal", SubmissionTime: old,
+					Cores: 4, Extensions: map[string]any{
+						"pending_reason": `{"main_reason":1,"main_reason_name":"no_resource"}`,
+					}},
+				{ID: "1-0-2-0", Name: "fresh", Owner: "u1", Queue: "normal", SubmissionTime: recent},
+				{ID: "1-0-3-0", Name: "matchfail", Owner: "u2", Queue: "gpu", SubmissionTime: old,
+					Extensions: map[string]any{
+						"pending_reason": `{"main_reason":2,"main_reason_name":"match_fail"}`,
+					}},
+			},
+			sclient.JobStatusRunning: {
+				{ID: "1-0-9-0", Name: "runner", SubmissionTime: old},
+			},
+		},
+		nodes: []*sclient.URMNode{
+			{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 6},
+			{Name: "t2", Status: "idle", CoresTotal: 8, CoresUsed: 0},
+		},
+		queues: []*sclient.URMQueue{
+			{Name: "normal", PendingJobs: 2, RunningJobs: 1},
+		},
+	}
+}
+
+func TestBuild_TrimsAndSorts(t *testing.T) {
+	b := NewBuilder(testClient())
+	st, err := b.Build(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("Build: %v", err)
+	}
+
+	if st.Cluster.PendingJobs != 3 || st.Cluster.RunningJobs != 1 {
+		t.Errorf("cluster counts = %+v", st.Cluster)
+	}
+	// 6/16 cores used.
+	if st.Cluster.Utilization != 0.38 {
+		t.Errorf("utilization = %v, want 0.38", st.Cluster.Utilization)
+	}
+
+	// Fresh PEND job (30s) filtered by stall threshold; two stalled remain,
+	// most-starved first. Both stalled are 2h old so order is by (equal)
+	// wait — assert set membership instead.
+	if len(st.PendTop) != 2 {
+		t.Fatalf("pend_top = %+v, want 2 stalled jobs", st.PendTop)
+	}
+	got := map[string]bool{}
+	for _, j := range st.PendTop {
+		got[j.ID] = true
+		if j.WaitSeconds < 300 {
+			t.Errorf("%s below stall threshold: %v", j.ID, j.WaitSeconds)
+		}
+	}
+	if !got["1-0-1-0"] || !got["1-0-3-0"] {
+		t.Errorf("stalled jobs missing: %+v", st.PendTop)
+	}
+
+	// Reason parsing.
+	for _, j := range st.PendTop {
+		if j.ID == "1-0-1-0" && (j.Reason == nil || j.Reason.MainReasonName != "no_resource") {
+			t.Errorf("reason parse failed: %+v", j.Reason)
+		}
+		if j.ID == "1-0-3-0" && (j.Reason == nil || j.Reason.Scenario != "no host matches the job's host-selection condition") {
+			t.Errorf("scenario translation failed: %+v", j.Reason)
+		}
+		if j.ID == "1-0-2-0" {
+			t.Errorf("fresh job leaked into pend_top")
+		}
+	}
+
+	// Nodes.
+	if len(st.Nodes) != 2 || st.Nodes[0].Utilization != 0.75 {
+		t.Errorf("nodes = %+v", st.Nodes)
+	}
+
+	// Queues garnish.
+	if len(st.Queues) != 1 || st.Queues[0].Name != "normal" {
+		t.Errorf("queues = %+v", st.Queues)
+	}
+}
+
+func TestBuild_PendTopCap(t *testing.T) {
+	fc := testClient()
+	var pend []*sclient.URMJob
+	for i := 0; i < 30; i++ {
+		pend = append(pend, &sclient.URMJob{
+			ID: "1-0-" + jsonInt(i) + "-0", SubmissionTime: time.Now().Add(-time.Duration(i+1) * time.Hour),
+		})
+	}
+	fc.jobs[sclient.JobStatusPending] = pend
+	b := NewBuilder(fc)
+	b.PendTop = 5
+	st, err := b.Build(context.Background(), nil)
+	if err != nil {
+		t.Fatalf("Build: %v", err)
+	}
+	if len(st.PendTop) != 5 {
+		t.Fatalf("cap failed: %d entries", len(st.PendTop))
+	}
+	// Most-starved first: the oldest job (i=29 → 30h) is top.
+	if st.PendTop[0].ID != "1-0-29-0" {
+		t.Errorf("top = %s, want oldest", st.PendTop[0].ID)
+	}
+}
+
+func TestBuild_StateJSONMatchesEchoShape(t *testing.T) {
+	// The snapshot must decode cleanly into the echo provider's echoState —
+	// the contract between the two packages.
+	st, err := NewBuilder(testClient()).Build(context.Background(), []ExitState{{ID: "1-0-7-0", ExitCode: 127}})
+	if err != nil {
+		t.Fatalf("Build: %v", err)
+	}
+	raw, err := json.Marshal(st)
+	if err != nil {
+		t.Fatalf("marshal: %v", err)
+	}
+	var probe struct {
+		GeneratedAt string `json:"generated_at"`
+		Cluster     struct {
+			PendingJobs int `json:"pending_jobs"`
+		} `json:"cluster"`
+		Nodes []struct {
+			Name string `json:"name"`
+		} `json:"nodes"`
+		PendTop []struct {
+			ID     string `json:"id"`
+			Reason *struct {
+				MainReasonName string `json:"main_reason_name"`
+				Scenario       string `json:"scenario"`
+			} `json:"reason"`
+		} `json:"pend_top"`
+		RecentExits []struct {
+			ID       string `json:"id"`
+			ExitCode int    `json:"exit_code"`
+		} `json:"recent_exits"`
+	}
+	if err := json.Unmarshal(raw, &probe); err != nil {
+		t.Fatalf("state JSON: %v", err)
+	}
+	if probe.Cluster.PendingJobs != 3 || len(probe.Nodes) != 2 {
+		t.Errorf("roundtrip mismatch: %+v", probe)
+	}
+	if len(probe.RecentExits) != 1 || probe.RecentExits[0].ID != "1-0-7-0" {
+		t.Errorf("recent_exits: %+v", probe.RecentExits)
+	}
+	if len(probe.PendTop) == 0 || probe.PendTop[0].Reason == nil {
+		t.Errorf("reason missing in JSON: %+v", probe.PendTop)
+	}
+}
+
+func jsonInt(i int) string {
+	b, _ := json.Marshal(i)
+	return string(b)
+}