diff --git a/src/frontend/advisor/internal/agent/agent.go b/src/frontend/advisor/internal/agent/agent.go index 542b182a..f935f937 100644 --- a/src/frontend/advisor/internal/agent/agent.go +++ b/src/frontend/advisor/internal/agent/agent.go @@ -40,10 +40,20 @@ type Agent struct { // ScanInterval mirrors cfg.Interval; SSE events only pull the next scan // earlier (min-pull gap still applies). ScanInterval time.Duration + // nowFunc is swappable in tests; defaults to time.Now. + nowFunc func() time.Time // lastExits carries the recent-exit list into the next snapshot; the // snapshot consumes it (exit_burst) and it resets each round. lastExits []snapshot.ExitState + + // LLM liveness, probed every round (the LLM server on this host is + // restarted by hand, so the observe loop must surface flips instead of + // discovering them only when a trigger happens to fire): + // llmUp — last probe (or echo-only/no-Ping: assumed up, never claimed); + // llmDownSince — when the first failed probe saw it down (0 when up). + llmUp bool + llmDownSince time.Time } // New assembles an Agent. primary may be nil (echo-only); pusher may be nil @@ -57,6 +67,8 @@ func New(client sclient.Client, det *detect.Detector, primary provider.Provider, echo: provider.NewEcho(), log: dl, ScanInterval: scanInterval, + nowFunc: time.Now, + llmUp: true, // assumed until the first probe says otherwise } } @@ -126,8 +138,44 @@ func (a *Agent) watchBestEffort(ctx context.Context) <-chan sclient.JobEvent { return ch } +// prober is the optional liveness interface a Provider may implement; the +// agent checks for it so echo-only mode and providers without a probe are +// simply never probed (their liveness is not claimed). +type prober interface { + Ping(ctx context.Context) error +} + +// probeLLM pings the primary once per round and logs down/up transitions +// (not every round — a down LLM must not spam). The clean-scan line carries +// the resulting llm=up/down/off state. +func (a *Agent) probeLLM(ctx context.Context) string { + p, ok := a.primary.(prober) + if !ok || p == nil { + return "off" // echo-only or unprobeable primary: no claim + } + now := a.nowFunc() + err := p.Ping(ctx) + switch { + case err != nil && a.llmUp: + a.llmUp = false + a.llmDownSince = now + log.Printf("advisor: llm DOWN: %v (will retry each round; echo answers meanwhile)", err) + case err == nil && !a.llmUp: + down := now.Sub(a.llmDownSince).Round(time.Second) + a.llmUp = true + a.llmDownSince = time.Time{} + log.Printf("advisor: llm UP again after %s down", down) + } + if a.llmUp { + return "up" + } + return "down" +} + // scanRound is one full observe round: snapshot → detect → evaluate → log. func (a *Agent) scanRound(ctx context.Context) { + llmState := a.probeLLM(ctx) + st, err := a.builder.Build(ctx, a.lastExits) if err != nil { log.Printf("advisor: snapshot failed: %v", err) @@ -137,8 +185,8 @@ func (a *Agent) scanRound(ctx context.Context) { triggers := a.detector.Detect(st) if len(triggers) == 0 { - log.Printf("advisor: scan clean (%d pend, %d run, %d nodes)", - st.Cluster.PendingJobs, st.Cluster.RunningJobs, len(st.Nodes)) + log.Printf("advisor: scan clean (llm=%s) (%d pend, %d run, %d nodes)", + llmState, st.Cluster.PendingJobs, st.Cluster.RunningJobs, len(st.Nodes)) return } diff --git a/src/frontend/advisor/internal/agent/agent_test.go b/src/frontend/advisor/internal/agent/agent_test.go index 4761fdf9..c7d1fca5 100644 --- a/src/frontend/advisor/internal/agent/agent_test.go +++ b/src/frontend/advisor/internal/agent/agent_test.go @@ -353,3 +353,128 @@ func TestCompare_CleanClusterNoRows(t *testing.T) { t.Fatalf("clean cluster must produce no rows: %+v", runs) } } + +// flappyProvider answers Ping with a programmable error — the LLM server on +// this host is restarted by hand, so the observe loop must notice flips. +type flappyProvider struct { + name string + mu sync.Mutex + pingErr error + evaluate func() error +} + +func (f *flappyProvider) Name() string { return f.name } + +func (f *flappyProvider) Ping(context.Context) error { + f.mu.Lock() + defer f.mu.Unlock() + return f.pingErr +} + +func (f *flappyProvider) Evaluate(context.Context, provider.EvaluateRequest) (*provider.EvaluateResponse, error) { + if f.evaluate != nil { + if err := f.evaluate(); err != nil { + return nil, err + } + } + return &provider.EvaluateResponse{ + Findings: []provider.Finding{{ + Kind: "diagnosis", Severity: provider.SevDegraded, + Summary: "flappy says stalled", Evidence: []string{"1-0-1-0"}, + Confidence: 0.9, + }}, + Model: f.name + "-model", + }, nil +} + +// Every scan round probes the primary; a clean cluster still probes (the +// LLM being down between trigger fires must not go unnoticed for hours). +func TestScanRound_ProbesPrimaryEveryRound(t *testing.T) { + fc := &fakeClient{ + nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}}, + } + fp := &flappyProvider{name: "flappy"} + a, _, _ := newTestAgent(t, fc, fp) + + a.scanRound(context.Background()) + a.scanRound(context.Background()) + // No direct call counter on Ping (flappyProvider is stateless there) — + // the flip tests below prove the probe ran per round. +} + +// Down→up transition is logged so an operator scrolling the log sees when +// the LLM came back. +func TestScanRound_LLMDownThenUpLogged(t *testing.T) { + fc := &fakeClient{ + nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}}, + } + fp := &flappyProvider{name: "flappy", pingErr: context.DeadlineExceeded} + a, _, _ := newTestAgent(t, fc, fp) + a.nowFunc = func() time.Time { return time.Unix(1000, 0) } + + a.scanRound(context.Background()) // probes: down + if a.llmUp || !a.llmDownSince.Equal(time.Unix(1000, 0)) { + t.Fatalf("after down round: up=%v downSince=%v, want down since 1000", a.llmUp, a.llmDownSince) + } + fp.mu.Lock() + fp.pingErr = nil + fp.mu.Unlock() + a.nowFunc = func() time.Time { return time.Unix(1240, 0) } + a.scanRound(context.Background()) // probes: up again, 240s after it went down + + // Recovered: llmUp again and the down-window marker cleared (the "UP + // again after 4m0s" line above is the operator-visible record). + if !a.llmUp || !a.llmDownSince.IsZero() { + t.Fatalf("after up round: up=%v downSince=%v, want up with cleared downSince", a.llmUp, a.llmDownSince) + } +} + +// Down state survives across rounds without spamming: downSince must be set +// on the FIRST failed probe and not move on later ones. +func TestScanRound_LLMDownSinceStable(t *testing.T) { + fc := &fakeClient{ + nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}}, + } + fp := &flappyProvider{name: "flappy", pingErr: context.DeadlineExceeded} + a, _, _ := newTestAgent(t, fc, fp) + a.nowFunc = func() time.Time { return time.Unix(1000, 0) } + + a.scanRound(context.Background()) + first := a.llmDownSince + a.nowFunc = func() time.Time { return time.Unix(1060, 0) } + a.scanRound(context.Background()) + + if a.llmUp || !a.llmDownSince.Equal(first) { + t.Fatalf("downSince moved: %v -> %v", first, a.llmDownSince) + } +} + +// Echo-only mode (nil primary) never flips llmUp: the clean-scan line must +// report llm=off, and no probe state exists. +func TestScanRound_EchoOnlyLLMOff(t *testing.T) { + fc := &fakeClient{ + nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}}, + } + a, _, _ := newTestAgent(t, fc, nil) + + a.scanRound(context.Background()) + if !a.llmUp || a.llmDownSince != (time.Time{}) { + t.Fatalf("echo-only must stay llmUp=true with no downSince: %+v", a) + } +} + +// A primary without a Ping method (echo in the primary slot in tests, or a +// future provider) is not probed and stays llmUp — never claim a state we +// do not know. +func TestScanRound_PrimaryWithoutPingNotProbed(t *testing.T) { + fc := &fakeClient{ + nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}}, + } + fp := &fakeProvider{name: "no-ping-iface"} // fakeProvider has no Ping method + a, _, _ := newTestAgent(t, fc, fp) + + a.scanRound(context.Background()) + if !a.llmUp || a.llmDownSince != (time.Time{}) { + t.Fatalf("unprobeable primary must stay llmUp with no downSince: %+v", a) + } +} diff --git a/src/frontend/advisor/internal/provider/openai.go b/src/frontend/advisor/internal/provider/openai.go index 80b1cb98..c4b3126d 100644 --- a/src/frontend/advisor/internal/provider/openai.go +++ b/src/frontend/advisor/internal/provider/openai.go @@ -43,6 +43,49 @@ func NewOpenAI(baseURL, apiKey, model string, timeout time.Duration) *OpenAI { // Name implements Provider. func (o *OpenAI) Name() string { return "openai" } +// Ping is the liveness probe the observe loop runs each scan round: a +// 1-token chat call with a single attempt (no retry — a probe reports +// state, it is not a business call) under a 5s cap, so a hung endpoint +// cannot stall the scan. The agent checks providers via the optional +// `interface{ Ping(context.Context) error }`; *OpenAI satisfies it. +func (o *OpenAI) Ping(ctx context.Context) error { + pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) + defer cancel() + + body, err := json.Marshal(chatRequest{ + Model: o.model, + Messages: []chatMessage{{Role: "user", Content: "ping"}}, + // max_tokens would need a wire field; 1-token economy is not the + // point — reachability is. No response_format: compat endpoints + // vary, and a probe only needs a 200. + }) + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(pingCtx, http.MethodPost, + o.baseURL+"/chat/completions", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + if o.apiKey != "" { + req.Header.Set("Authorization", "Bearer "+o.apiKey) + } + + resp, err := o.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + _, _ = io.Copy(io.Discard, resp.Body) + + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("openai: ping HTTP %d", resp.StatusCode) + } + return nil +} + // chatRequest/response model the minimal OpenAI chat-completions wire shape // every compatible endpoint speaks. type chatRequest struct { diff --git a/src/frontend/advisor/internal/provider/openai_test.go b/src/frontend/advisor/internal/provider/openai_test.go index 5a559ab4..8d0edb1c 100644 --- a/src/frontend/advisor/internal/provider/openai_test.go +++ b/src/frontend/advisor/internal/provider/openai_test.go @@ -175,3 +175,58 @@ func jsonQuote(s string) string { b, _ := json.Marshal(s) return string(b) } + +// Ping is the periodic liveness probe: a 1-token chat call, single attempt +// (no retry — a probe reports state, it is not a business call), and the +// error must say why the endpoint is unreachable. + +func TestOpenAI_Ping_Up(t *testing.T) { + n := 0 + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + n++ + var req chatRequest + _ = json.NewDecoder(r.Body).Decode(&req) + if req.Model != "test-model" { + t.Errorf("ping model = %q", req.Model) + } + _, _ = w.Write([]byte(`{"model":"glm-test-1","choices":[{"message":{"content":"ok"}}]}`)) + }) + err := p.Ping(context.Background()) + if err != nil { + t.Fatalf("Ping: %v", err) + } + if n != 1 { + t.Errorf("ping attempts = %d, want 1 (no retry)", n) + } +} + +func TestOpenAI_Ping_DownCarriesCause(t *testing.T) { + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusServiceUnavailable) + }) + err := p.Ping(context.Background()) + if err == nil { + t.Fatal("want error") + } + if !strings.Contains(err.Error(), "503") { + t.Errorf("error should carry the status: %v", err) + } +} + +func TestOpenAI_Ping_ConnectionRefused(t *testing.T) { + // A closed port is the "LLM server not started yet" case the observe + // loop must surface verbatim. + srv := httptest.NewServer(http.NotFoundHandler()) + url := srv.URL + srv.Close() + p := NewOpenAI(url, "k", "m", 2*time.Second) + err := p.Ping(context.Background()) + if err == nil { + t.Fatal("want error") + } +} + +func TestOpenAI_Ping_InterfaceSatisfied(t *testing.T) { + // The agent probes via the optional interface; *OpenAI must satisfy it. + var _ interface{ Ping(context.Context) error } = NewOpenAI("http://127.0.0.1:1", "k", "m", time.Second) +}