From 1ab96521b81b03276e8304655e432e2315658e6a Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Mon, 19 Oct 2026 03:31:36 +0000 Subject: [PATCH] =?UTF-8?q?feat(advisor):=20Go=20module=20skeleton=20?= =?UTF-8?q?=E2=80=94=20config,=20provider=20contract,=20echo=20+=20openai?= =?UTF-8?q?=20providers?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P0 of the scheduler advisor (observe-only LLM ops copilot). New standalone Go module src/frontend/advisor (CGO_ENABLED=0, mcp-server layout, replaces schedulerclient). No LLM executes anything: providers turn scenario snapshots into structured Findings only. - internal/config: env-driven config (XCE_API_URL/KEY, ADVISOR_LLM_*, ADVISOR_INTERVAL, ADVISOR_LOG); LLM unset is valid (echo-only). - internal/provider: Provider interface + Finding contract (kind/severity/ summary/evidence/confidence/suggestion). Confidence is display-only — never calibrated for LLM backends, never gated on. - echo: deterministic rules provider (fallback, test baseline, yardstick). - openai: OpenAI-compatible chat-completions (GLM/Qwen/DeepSeek/local vLLM), strict JSON response_format, enum-validated findings, bounded 429/5xx retries with backoff, 30s default timeout. - VerifyEvidence: hallucination firewall — evidence elements must appear verbatim in the state snapshot or the finding is demoted (unverified, confidence 0.2) while staying in the log. Jev (TypeSafe System One) deliberately not wired: recorded in the design doc (later commit) as a future provider behind the same interface. Co-Authored-By: Claude Code --- src/frontend/advisor/go.mod | 15 + .../advisor/internal/config/config.go | 104 +++++++ .../advisor/internal/provider/echo.go | 274 +++++++++++++++++ .../advisor/internal/provider/echo_test.go | 166 ++++++++++ .../advisor/internal/provider/openai.go | 286 ++++++++++++++++++ .../advisor/internal/provider/openai_test.go | 177 +++++++++++ .../advisor/internal/provider/provider.go | 102 +++++++ 7 files changed, 1124 insertions(+) create mode 100644 src/frontend/advisor/go.mod create mode 100644 src/frontend/advisor/internal/config/config.go create mode 100644 src/frontend/advisor/internal/provider/echo.go create mode 100644 src/frontend/advisor/internal/provider/echo_test.go create mode 100644 src/frontend/advisor/internal/provider/openai.go create mode 100644 src/frontend/advisor/internal/provider/openai_test.go create mode 100644 src/frontend/advisor/internal/provider/provider.go diff --git a/src/frontend/advisor/go.mod b/src/frontend/advisor/go.mod new file mode 100644 index 00000000..a4170a68 --- /dev/null +++ b/src/frontend/advisor/go.mod @@ -0,0 +1,15 @@ +module advisor + +go 1.25.5 + +replace schedulerclient => ../schedulerclient + +require ( + schedulerclient v0.0.0-00010101000000-000000000000 + gopkg.in/yaml.v3 v3.0.1 +) + +require ( + resty.dev/v3 v3.0.0-beta.2 // indirect + golang.org/x/net v0.33.0 // indirect +) diff --git a/src/frontend/advisor/internal/config/config.go b/src/frontend/advisor/internal/config/config.go new file mode 100644 index 00000000..55fbd0a5 --- /dev/null +++ b/src/frontend/advisor/internal/config/config.go @@ -0,0 +1,104 @@ +// Package config resolves the advisor's runtime configuration from +// environment variables, mirroring the mcp-server's XCE_API_URL/XCE_API_KEY +// convention. Every knob has a usable default except the two credentials. +package config + +import ( + "errors" + "fmt" + "os" + "time" +) + +// Config is the resolved advisor configuration. +type Config struct { + // APIURL is the apiserver base URL (XCE_API_URL). + APIURL string + // APIKey is the apiserver service-account key (XCE_API_KEY, xce_ak_...). + // P0 intends an mcp_readonly role key: job/queue/host/cluster:view + event:view. + APIKey string + + // LLMBaseURL is an OpenAI-compatible chat-completions endpoint + // (ADVISOR_LLM_BASE_URL). Empty disables the LLM provider — the advisor + // then runs echo-only (deterministic rules, no external calls). + LLMBaseURL string + // LLMAPIKey authenticates to the LLM endpoint (ADVISOR_LLM_API_KEY). + LLMAPIKey string + // LLMModel is the model name sent to the endpoint (ADVISOR_LLM_MODEL), + // e.g. "glm-4-flash" or a local vLLM model id. + LLMModel string + // LLMTimeout bounds a single LLM call (ADVISOR_LLM_TIMEOUT, default 30s). + LLMTimeout time.Duration + + // Interval is the periodic full-scan cadence (ADVISOR_INTERVAL, default + // 120s). SSE events only pull the next scan earlier; they never bypass + // the detectors. + Interval time.Duration + + // LogPath is the JSONL decision log (ADVISOR_LOG, default + // /tmp/advisor-decisions.jsonl). + LogPath string + // RunbookPath optionally overrides the embedded runbook + // (ADVISOR_RUNBOOK). + RunbookPath string +} + +// Defaults. +const ( + defaultAPIURL = "http://172.31.1.30:8080" + defaultLLMTimeout = 30 * time.Second + defaultInterval = 120 * time.Second + defaultLogPath = "/tmp/advisor-decisions.jsonl" +) + +// FromEnv resolves the configuration, returning an error only for genuinely +// missing required values (XCE_API_KEY). LLM settings being empty is valid — +// the advisor degrades to the echo provider. +func FromEnv() (*Config, error) { + c := &Config{ + APIURL: getenv("XCE_API_URL", defaultAPIURL), + APIKey: os.Getenv("XCE_API_KEY"), + LLMBaseURL: getenv("ADVISOR_LLM_BASE_URL", ""), + LLMAPIKey: getenv("ADVISOR_LLM_API_KEY", ""), + LLMModel: getenv("ADVISOR_LLM_MODEL", ""), + LogPath: getenv("ADVISOR_LOG", defaultLogPath), + RunbookPath: getenv("ADVISOR_RUNBOOK", ""), + Interval: defaultInterval, + LLMTimeout: defaultLLMTimeout, + } + + if c.APIKey == "" { + return nil, errors.New("XCE_API_KEY is required (an mcp_readonly service-account key)") + } + + if s := os.Getenv("ADVISOR_INTERVAL"); s != "" { + d, err := time.ParseDuration(s) + if err != nil || d <= 0 { + return nil, fmt.Errorf("invalid ADVISOR_INTERVAL %q: want e.g. 120s", s) + } + c.Interval = d + } + if s := os.Getenv("ADVISOR_LLM_TIMEOUT"); s != "" { + d, err := time.ParseDuration(s) + if err != nil || d <= 0 { + return nil, fmt.Errorf("invalid ADVISOR_LLM_TIMEOUT %q: want e.g. 30s", s) + } + c.LLMTimeout = d + } + if c.LLMBaseURL != "" { + if c.LLMModel == "" { + return nil, errors.New("ADVISOR_LLM_BASE_URL set without ADVISOR_LLM_MODEL") + } + } + return c, nil +} + +// LLMEnabled reports whether the OpenAI-compatible provider is configured. +func (c *Config) LLMEnabled() bool { return c.LLMBaseURL != "" } + +func getenv(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} diff --git a/src/frontend/advisor/internal/provider/echo.go b/src/frontend/advisor/internal/provider/echo.go new file mode 100644 index 00000000..f07d268a --- /dev/null +++ b/src/frontend/advisor/internal/provider/echo.go @@ -0,0 +1,274 @@ +package provider + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "time" +) + +// Echo is the deterministic rules provider. It never fails and never calls +// out: the same state always yields the same findings. It is the fallback +// when the LLM provider is unavailable, the unit-test baseline, and the +// yardstick to compare LLM output against. +type Echo struct{} + +// NewEcho returns the echo provider. +func NewEcho() *Echo { return &Echo{} } + +// Name implements Provider. +func (e *Echo) Name() string { return "echo" } + +// echoState is the subset of the snapshot shape the echo rules read. The +// snapshot builder (internal/snapshot) emits this shape; unknown fields are +// ignored so the two can evolve independently. +type echoState struct { + GeneratedAt string `json:"generated_at"` + Cluster echoCluster `json:"cluster"` + Nodes []echoNode `json:"nodes"` + PendTop []echoJob `json:"pend_top"` + Exits []echoExit `json:"recent_exits,omitempty"` + Queues []echoQueue `json:"queues,omitempty"` +} + +type echoCluster struct { + PendingJobs int `json:"pending_jobs"` + RunningJobs int `json:"running_jobs"` + Utilization float64 `json:"utilization"` +} + +type echoNode struct { + Name string `json:"name"` + Status string `json:"status"` + CoresTotal int `json:"cores_total"` + CoresUsed int `json:"cores_used"` + Utilization float64 `json:"utilization"` +} + +type echoJob struct { + ID string `json:"id"` + Name string `json:"name"` + Owner string `json:"owner"` + Queue string `json:"queue"` + WaitSeconds float64 `json:"wait_seconds"` + Reason *echoReason `json:"reason,omitempty"` + Cores int `json:"cores,omitempty"` +} + +type echoReason struct { + MainReasonName string `json:"main_reason_name,omitempty"` + Scenario string `json:"scenario,omitempty"` +} + +type echoExit struct { + ID string `json:"id"` + ExitCode int `json:"exit_code"` + Name string `json:"name,omitempty"` +} + +type echoQueue struct { + Name string `json:"name"` + PendingJobs int `json:"pending_jobs"` +} + +// Evaluate implements Provider with deterministic rules. Each rule mirrors +// one detector family so the echo output always explains the same trigger +// the detector fired on (req.Scenario). +func (e *Echo) Evaluate(ctx context.Context, req EvaluateRequest) (*EvaluateResponse, error) { + start := time.Now() + var st echoState + if err := json.Unmarshal(req.State, &st); err != nil { + return nil, fmt.Errorf("echo: state is not valid snapshot JSON: %w", err) + } + + var findings []Finding + switch req.Scenario { + case "pend_stall": + findings = echoPendStall(st) + case "worker_health": + findings = echoWorkerHealth(st) + case "util_skew": + findings = echoUtilSkew(st) + case "exit_burst": + findings = echoExitBurst(st) + default: + // Unknown scenario: state inspection only, never an error — an + // unknown trigger name must not break the observe loop. + findings = nil + } + + raw, _ := json.Marshal(findings) + return &EvaluateResponse{ + Findings: findings, + Model: "echo-rules", + Raw: string(raw), + Latency: time.Since(start), + }, nil +} + +// echoPendStall groups the stalled jobs by their PEND reason and emits one +// finding per group, worst severity first. This is the human summary of +// "why is nothing dispatching" without any LLM. +func echoPendStall(st echoState) []Finding { + if len(st.PendTop) == 0 { + return nil + } + byReason := map[string][]echoJob{} + for _, j := range st.PendTop { + name := "unknown" + if j.Reason != nil { + switch { + case j.Reason.Scenario != "": + name = j.Reason.Scenario + case j.Reason.MainReasonName != "": + name = j.Reason.MainReasonName + } + } + byReason[name] = append(byReason[name], j) + } + + names := make([]string, 0, len(byReason)) + for n := range byReason { + names = append(names, n) + } + sort.Strings(names) + + var out []Finding + for _, n := range names { + js := byReason[n] + oldest := js[0].WaitSeconds + for _, j := range js { + if j.WaitSeconds > oldest { + oldest = j.WaitSeconds + } + } + sev := SevDegraded + if oldest > 3600 { + sev = SevSevere + } + ids := make([]string, 0, len(js)) + for _, j := range js { + ids = append(ids, j.ID) + } + out = append(out, Finding{ + Kind: "diagnosis", + Severity: sev, + Summary: fmt.Sprintf("%d pending job(s) stalled on reason %q, oldest waiting %s", + len(js), n, time.Duration(oldest*float64(time.Second)).Round(time.Second)), + Evidence: append(ids, n), + Confidence: 1.0, + Suggestion: echoPendAdvice(n), + }) + } + return out +} + +// echoPendAdvice maps a PEND reason to the operator hint from our own +// debugging history (bug-1494 stale config, ABI skew dead-wait, etc.). +func echoPendAdvice(reason string) string { + switch { + case strings.Contains(reason, "no_resource") || strings.Contains(reason, "resource"): + return "Check requested resources against node capacity (gaps.md: job-level priority/limits config)." + case strings.Contains(reason, "match") || strings.Contains(reason, "host"): + return "Hosts may be unusable for the request (drain/down or resreq mismatch); compare resources.yaml against observed nodes." + case strings.Contains(reason, "limit"): + return "A user/queue/host limit is blocking dispatch; inspect the limit configuration." + default: + return "Correlate with scheduler/controller logs and the runbook before acting." + } +} + +// echoWorkerHealth reports unusable nodes. +func echoWorkerHealth(st echoState) []Finding { + var bad []echoNode + for _, n := range st.Nodes { + switch n.Status { + case "down", "unknown", "unavail", "closed": + bad = append(bad, n) + } + } + if len(bad) == 0 { + return nil + } + names := make([]string, 0, len(bad)) + for _, n := range bad { + names = append(names, n.Name) + } + return []Finding{{ + Kind: "anomaly", + Severity: SevSevere, + Summary: fmt.Sprintf("%d node(s) unusable: %s", + len(bad), strings.Join(names, ", ")), + Evidence: names, + Confidence: 1.0, + Suggestion: "Check xcecd on those nodes (recent upgrades must replace xcecd+xcecd_worker as a pair) and controller logs for join failures.", + }} +} + +// echoUtilSkew flags imbalance: one node near-saturated while another idles. +// This is the t1-affinity observation (first-fit × insertion-order slots) +// surfaced as data. +func echoUtilSkew(st echoState) []Finding { + var hot, idle []echoNode + for _, n := range st.Nodes { + if n.CoresTotal < 4 { + continue + } + switch { + case n.Utilization >= 0.8: + hot = append(hot, n) + case n.Utilization <= 0.2: + idle = append(idle, n) + } + } + if len(hot) == 0 || len(idle) == 0 { + return nil + } + ev := make([]string, 0, len(hot)+len(idle)) + var hotDesc, idleDesc []string + for _, n := range hot { + hotDesc = append(hotDesc, fmt.Sprintf("%s %.0f%%", n.Name, n.Utilization*100)) + ev = append(ev, n.Name) + } + for _, n := range idle { + idleDesc = append(idleDesc, fmt.Sprintf("%s %.0f%%", n.Name, n.Utilization*100)) + ev = append(ev, n.Name) + } + return []Finding{{ + Kind: "anomaly", + Severity: SevInfo, + Summary: fmt.Sprintf("Utilization skewed: hot [%s] vs idle [%s] — placement bias, not capacity shortage", + strings.Join(hotDesc, ", "), strings.Join(idleDesc, ", ")), + Evidence: ev, + Confidence: 1.0, + Suggestion: "Known first-fit insertion-order bias (see memory xce-firstfit-t1-affinity); policy change, not an incident.", + }} +} + +// echoExitBurst summarizes recent failures. +func echoExitBurst(st echoState) []Finding { + if len(st.Exits) == 0 { + return nil + } + nonZero := 0 + ids := make([]string, 0, len(st.Exits)) + for _, x := range st.Exits { + if x.ExitCode != 0 { + nonZero++ + ids = append(ids, x.ID) + } + } + if nonZero == 0 { + return nil + } + return []Finding{{ + Kind: "anomaly", + Severity: SevDegraded, + Summary: fmt.Sprintf("%d recent exit(s) with non-zero status", nonZero), + Evidence: ids, + Suggestion: "Group by exit code; a shared code (127, 139) points at environment breakage rather than job bugs.", + Confidence: 1.0, + }} +} diff --git a/src/frontend/advisor/internal/provider/echo_test.go b/src/frontend/advisor/internal/provider/echo_test.go new file mode 100644 index 00000000..a10e506f --- /dev/null +++ b/src/frontend/advisor/internal/provider/echo_test.go @@ -0,0 +1,166 @@ +package provider + +import ( + "context" + "encoding/json" + "strings" + "testing" +) + +func pendStallState(jobs ...echoJob) json.RawMessage { + st := echoState{ + Cluster: echoCluster{PendingJobs: len(jobs)}, + Nodes: []echoNode{ + {Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 8, Utilization: 1.0}, + {Name: "t2", Status: "idle", CoresTotal: 8, CoresUsed: 0, Utilization: 0.0}, + }, + } + st.PendTop = jobs + b, _ := json.Marshal(st) + return b +} + +func TestEcho_PendStall(t *testing.T) { + st := pendStallState( + echoJob{ID: "1-0-1-0", Name: "a", WaitSeconds: 600, Reason: &echoReason{MainReasonName: "no_resource"}}, + echoJob{ID: "1-0-2-0", Name: "b", WaitSeconds: 60, Reason: &echoReason{MainReasonName: "no_resource"}}, + echoJob{ID: "1-0-3-0", Name: "c", WaitSeconds: 7200, Reason: &echoReason{MainReasonName: "match_fail"}}, + ) + resp, err := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: st}) + if err != nil { + t.Fatalf("Evaluate: %v", err) + } + if resp.Model != "echo-rules" { + t.Fatalf("model = %q, want echo-rules", resp.Model) + } + if len(resp.Findings) != 2 { + t.Fatalf("got %d findings, want 2 (one per reason group): %+v", len(resp.Findings), resp.Findings) + } + // The >1h group must be severe, the other degraded. + sev := map[string]Severity{} + for _, f := range resp.Findings { + sev[f.Summary] = f.Severity + if f.Kind != "diagnosis" { + t.Errorf("kind = %q, want diagnosis", f.Kind) + } + if f.Confidence != 1.0 { + t.Errorf("echo confidence = %v, want 1.0", f.Confidence) + } + // Evidence must cite the reason and job ids verbatim. + if len(f.Evidence) < 2 { + t.Errorf("evidence too thin: %+v", f) + } + } + _ = sev +} + +func TestEcho_PendStallSeverityByAge(t *testing.T) { + old := pendStallState(echoJob{ID: "j1", WaitSeconds: 4000, Reason: &echoReason{MainReasonName: "no_resource"}}) + resp, _ := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: old}) + if len(resp.Findings) != 1 || resp.Findings[0].Severity != SevSevere { + t.Fatalf("want severe for >1h stall: %+v", resp.Findings) + } + + young := pendStallState(echoJob{ID: "j2", WaitSeconds: 400, Reason: &echoReason{MainReasonName: "no_resource"}}) + resp, _ = NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: young}) + if len(resp.Findings) != 1 || resp.Findings[0].Severity != SevDegraded { + t.Fatalf("want degraded for <1h stall: %+v", resp.Findings) + } +} + +func TestEcho_WorkerHealth(t *testing.T) { + st := echoState{ + Nodes: []echoNode{ + {Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4, Utilization: 0.5}, + {Name: "t3", Status: "down", CoresTotal: 8, CoresUsed: 0}, + }, + } + b, _ := json.Marshal(st) + resp, err := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "worker_health", State: b}) + if err != nil { + t.Fatalf("Evaluate: %v", err) + } + if len(resp.Findings) != 1 { + t.Fatalf("want 1 finding, got %+v", resp.Findings) + } + f := resp.Findings[0] + if f.Severity != SevSevere || f.Kind != "anomaly" { + t.Fatalf("unexpected finding: %+v", f) + } + if !strings.Contains(f.Summary, "t3") { + t.Errorf("summary should cite the node: %q", f.Summary) + } +} + +func TestEcho_WorkerHealthAllGood(t *testing.T) { + st := echoState{ + Nodes: []echoNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4, Utilization: 0.5}}, + } + b, _ := json.Marshal(st) + resp, _ := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "worker_health", State: b}) + if len(resp.Findings) != 0 { + t.Fatalf("healthy cluster should produce no findings: %+v", resp.Findings) + } +} + +func TestEcho_UtilSkew(t *testing.T) { + st := echoState{ + Nodes: []echoNode{ + {Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 8, Utilization: 1.0}, + {Name: "t2", Status: "idle", CoresTotal: 8, CoresUsed: 0, Utilization: 0.0}, + }, + } + b, _ := json.Marshal(st) + resp, _ := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "util_skew", State: b}) + if len(resp.Findings) != 1 { + t.Fatalf("100%%/0%% skew should fire: %+v", resp.Findings) + } + if resp.Findings[0].Severity != SevInfo { + t.Errorf("skew is informational: %+v", resp.Findings[0]) + } + + // Balanced nodes must not fire. + st.Nodes = []echoNode{ + {Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 5, Utilization: 0.6}, + {Name: "t2", Status: "busy", CoresTotal: 8, CoresUsed: 4, Utilization: 0.5}, + } + b, _ = json.Marshal(st) + resp, _ = NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "util_skew", State: b}) + if len(resp.Findings) != 0 { + t.Fatalf("balanced cluster should not fire: %+v", resp.Findings) + } +} + +func TestEcho_ExitBurst(t *testing.T) { + st := echoState{ + Exits: []echoExit{ + {ID: "j1", ExitCode: 127}, + {ID: "j2", ExitCode: 127}, + {ID: "j3", ExitCode: 0}, + }, + } + b, _ := json.Marshal(st) + resp, _ := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "exit_burst", State: b}) + if len(resp.Findings) != 1 || resp.Findings[0].Severity != SevDegraded { + t.Fatalf("want degraded exit-burst finding: %+v", resp.Findings) + } + // Zero exits only → nothing. + st.Exits = []echoExit{{ID: "j4", ExitCode: 0}} + b, _ = json.Marshal(st) + resp, _ = NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "exit_burst", State: b}) + if len(resp.Findings) != 0 { + t.Fatalf("all-zero exits should not fire: %+v", resp.Findings) + } +} + +func TestEcho_UnknownScenarioAndBadState(t *testing.T) { + // Unknown scenario: no findings, no error. + resp, err := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "martian", State: []byte(`{}`)}) + if err != nil || len(resp.Findings) != 0 { + t.Fatalf("unknown scenario must be a no-op: %v %+v", err, resp) + } + // Bad JSON: error (agent degrades/logging path). + if _, err := NewEcho().Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: []byte("not json")}); err == nil { + t.Fatal("bad state JSON should error") + } +} diff --git a/src/frontend/advisor/internal/provider/openai.go b/src/frontend/advisor/internal/provider/openai.go new file mode 100644 index 00000000..80b1cb98 --- /dev/null +++ b/src/frontend/advisor/internal/provider/openai.go @@ -0,0 +1,286 @@ +package provider + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +// OpenAI calls an OpenAI-compatible chat-completions endpoint (GLM, Qwen, +// DeepSeek, or a local vLLM) and asks it for strict JSON findings. +// +// Design constraints for a non-calibrated model: +// - findings are parsed strictly (response_format json_object); a parse +// failure is an error, never a guess; +// - Confidence is displayed only, never gated on; +// - the agent runs the evidence firewall (VerifyEvidence) after every +// call, because the model can fabricate identifiers. +type OpenAI struct { + baseURL string + apiKey string + model string + timeout time.Duration + client *http.Client +} + +// NewOpenAI builds the provider against e.g. https://open.bigmodel.cn/api/paas/v4. +func NewOpenAI(baseURL, apiKey, model string, timeout time.Duration) *OpenAI { + return &OpenAI{ + baseURL: strings.TrimRight(baseURL, "/"), + apiKey: apiKey, + model: model, + timeout: timeout, + client: &http.Client{Timeout: timeout}, + } +} + +// Name implements Provider. +func (o *OpenAI) Name() string { return "openai" } + +// chatRequest/response model the minimal OpenAI chat-completions wire shape +// every compatible endpoint speaks. +type chatRequest struct { + Model string `json:"model"` + Messages []chatMessage `json:"messages"` + // ResponseFormat forces strict-JSON mode on supporting backends. + // Sent unconditionally; incompatible backends get a 400 which surfaces + // as a provider error and the agent degrades to echo. + ResponseFormat *chatResponseFormat `json:"response_format,omitempty"` + Temperature float64 `json:"temperature"` +} + +type chatResponseFormat struct { + Type string `json:"type"` // "json_object" +} + +type chatMessage struct { + Role string `json:"role"` + Content string `json:"content"` +} + +type chatResponse struct { + Model string `json:"model"` + Choices []struct { + Message struct { + Content string `json:"content"` + } `json:"message"` + } `json:"choices"` + Usage struct { + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + } `json:"usage"` + Error *struct { + Message string `json:"message"` + } `json:"error,omitempty"` +} + +const findingsSystemPrompt = `You are the diagnostics provider for the XCE HPC scheduler advisor, running in OBSERVE mode: you report, you never act. + +You receive a JSON snapshot of cluster state (pending jobs, nodes, queues, recent exits) for one scenario. Answer with STRICT JSON only, matching exactly this schema: + +{"findings":[{"kind":"anomaly|diagnosis","severity":"info|degraded|severe|outage","summary":"one-line human statement","evidence":["concrete identifiers copied verbatim from the state: job ids, host names, reason names"],"confidence":0.0,"suggestion":"advisory text for a human operator"}]} + +Rules: +- Every evidence element MUST be a string that appears verbatim in the state snapshot. Never invent ids, hosts, or reason names. +- confidence is your self-assessment (0..1). It is displayed, not acted on. +- suggestion is advice for a human. The system will not execute it. +- If the state does not support a finding, return {"findings":[]}. Do not pad.` + +// Evaluate implements Provider. +func (o *OpenAI) Evaluate(ctx context.Context, req EvaluateRequest) (*EvaluateResponse, error) { + start := time.Now() + + userMsg := fmt.Sprintf("scenario: %s\n\nrunbook:\n%s\n\nstate:\n%s", + req.Scenario, req.Runbook, string(req.State)) + + body, err := json.Marshal(chatRequest{ + Model: o.model, + Messages: []chatMessage{ + {Role: "system", Content: findingsSystemPrompt}, + {Role: "user", Content: userMsg}, + }, + ResponseFormat: &chatResponseFormat{Type: "json_object"}, + Temperature: 0.2, + }) + if err != nil { + return nil, err + } + + content, respModel, usage, err := o.chatWithRetry(ctx, body) + if err != nil { + return nil, err + } + + findings, err := parseFindings(content) + if err != nil { + return nil, fmt.Errorf("openai: findings JSON: %w (raw kept)", err) + } + + return &EvaluateResponse{ + Findings: findings, + Model: respModel, + Raw: content, + InputTokens: usage.input, + OutputTokens: usage.output, + Latency: time.Since(start), + }, nil +} + +type tokenUsage struct{ input, output int } + +// chatWithRetry performs the POST with bounded retries on 429/5xx. +func (o *OpenAI) chatWithRetry(ctx context.Context, body []byte) (content, model string, usage tokenUsage, err error) { + const attempts = 3 + var lastErr error + for i := 0; i < attempts; i++ { + if i > 0 { + // Exponential backoff: 1s, 2s. Abort on ctx cancellation. + select { + case <-ctx.Done(): + return "", "", tokenUsage{}, ctx.Err() + case <-time.After(time.Duration(1<<(i-1)) * time.Second): + } + } + content, model, usage, lastErr = o.chatOnce(ctx, body) + if lastErr == nil { + return content, model, usage, nil + } + if !retryable(lastErr) { + return "", "", tokenUsage{}, lastErr + } + } + return "", "", tokenUsage{}, fmt.Errorf("openai: %d attempts failed: %w", attempts, lastErr) +} + +// retryableError wraps statuses worth a retry. +type retryableError struct{ status int } + +func (e *retryableError) Error() string { + return fmt.Sprintf("retryable HTTP %d", e.status) +} + +func retryable(err error) bool { + var re *retryableError + return errors.As(err, &re) +} + +func (o *OpenAI) chatOnce(ctx context.Context, body []byte) (string, string, tokenUsage, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + o.baseURL+"/chat/completions", bytes.NewReader(body)) + if err != nil { + return "", "", tokenUsage{}, 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 "", "", tokenUsage{}, err + } + defer resp.Body.Close() + + raw, err := io.ReadAll(io.LimitReader(resp.Body, 32<<20)) + if err != nil { + return "", "", tokenUsage{}, err + } + + if resp.StatusCode != http.StatusOK { + if resp.StatusCode == http.StatusTooManyRequests || resp.StatusCode >= 500 { + return "", "", tokenUsage{}, &retryableError{status: resp.StatusCode} + } + return "", "", tokenUsage{}, fmt.Errorf("openai: HTTP %d: %s", resp.StatusCode, truncate(string(raw), 500)) + } + + var cr chatResponse + if err := json.Unmarshal(raw, &cr); err != nil { + return "", "", tokenUsage{}, fmt.Errorf("openai: response not JSON: %w", err) + } + if cr.Error != nil { + return "", "", tokenUsage{}, fmt.Errorf("openai: API error: %s", cr.Error.Message) + } + if len(cr.Choices) == 0 { + return "", "", tokenUsage{}, errors.New("openai: empty choices") + } + + usage := tokenUsage{input: cr.Usage.PromptTokens, output: cr.Usage.CompletionTokens} + model := cr.Model + if model == "" { + model = o.model // some compat endpoints omit model; fall back to ours + } + return cr.Choices[0].Message.Content, model, usage, nil +} + +// findingsDoc is the strict parse target of the model's JSON answer. +type findingsDoc struct { + Findings []Finding `json:"findings"` +} + +// parseFindings decodes the model content strictly: JSON object with a +// findings array, kind/severity validated against the closed sets. +func parseFindings(content string) ([]Finding, error) { + // Tolerate fences from backends that ignore response_format. + content = strings.TrimSpace(content) + content = strings.TrimPrefix(content, "```json") + content = strings.TrimPrefix(content, "```") + content = strings.TrimSuffix(content, "```") + content = strings.TrimSpace(content) + + var doc findingsDoc + if err := json.Unmarshal([]byte(content), &doc); err != nil { + return nil, fmt.Errorf("content is not a findings object: %w", err) + } + for i := range doc.Findings { + f := &doc.Findings[i] + if f.Kind != "anomaly" && f.Kind != "diagnosis" { + return nil, fmt.Errorf("finding %d: kind %q not in {anomaly,diagnosis}", i, f.Kind) + } + switch f.Severity { + case SevInfo, SevDegraded, SevSevere, SevOutage: + default: + return nil, fmt.Errorf("finding %d: severity %q not in {info,degraded,severe,outage}", i, f.Severity) + } + if f.Confidence < 0 { + f.Confidence = 0 + } else if f.Confidence > 1 { + f.Confidence = 1 + } + } + return doc.Findings, nil +} + +// VerifyEvidence is the hallucination firewall: every evidence element must +// appear verbatim in the state snapshot the model saw. Findings citing +// absent identifiers get Unverified=true and drop to low confidence — they +// stay in the log (useful signal about the model) but are visibly demoted. +func VerifyEvidence(state json.RawMessage, findings []Finding) { + hay := string(state) + for i := range findings { + f := &findings[i] + bad := false + for _, ev := range f.Evidence { + if ev != "" && !strings.Contains(hay, ev) { + bad = true + break + } + } + if bad { + f.Unverified = true + f.Confidence = 0.2 + } + } +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "..." +} diff --git a/src/frontend/advisor/internal/provider/openai_test.go b/src/frontend/advisor/internal/provider/openai_test.go new file mode 100644 index 00000000..5a559ab4 --- /dev/null +++ b/src/frontend/advisor/internal/provider/openai_test.go @@ -0,0 +1,177 @@ +package provider + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +const goodContent = `{"findings":[{"kind":"diagnosis","severity":"degraded","summary":"2 jobs stalled on no_resource","evidence":["1-0-1-0","no_resource"],"confidence":0.8,"suggestion":"check capacity"}]}` + +func newTestSrv(t *testing.T, handler http.HandlerFunc) *OpenAI { + t.Helper() + srv := httptest.NewServer(handler) + t.Cleanup(srv.Close) + return NewOpenAI(srv.URL, "test-key", "test-model", 5*time.Second) +} + +func openAIState() json.RawMessage { + // Evidence "1-0-1-0" and "no_resource" must appear verbatim. + return json.RawMessage(`{"cluster":{"pending_jobs":2},"pend_top":[{"id":"1-0-1-0","reason":{"main_reason_name":"no_resource"}}]}`) +} + +func TestOpenAI_Success(t *testing.T) { + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + if got := r.Header.Get("Authorization"); got != "Bearer test-key" { + t.Errorf("auth header = %q", got) + } + var req chatRequest + _ = json.NewDecoder(r.Body).Decode(&req) + if req.ResponseFormat == nil || req.ResponseFormat.Type != "json_object" { + t.Errorf("response_format missing") + } + if req.Messages[0].Role != "system" { + t.Errorf("first message not system") + } + if !strings.Contains(req.Messages[1].Content, "scenario: pend_stall") { + t.Errorf("user content missing scenario: %q", req.Messages[1].Content) + } + if !strings.Contains(req.Messages[1].Content, `"id":"1-0-1-0"`) { + t.Errorf("user content missing state: %q", req.Messages[1].Content) + } + _, _ = w.Write([]byte(`{"model":"glm-test-1","choices":[{"message":{"content":` + + jsonQuote(goodContent) + `}}],"usage":{"prompt_tokens":100,"completion_tokens":50}}`)) + }) + + resp, err := p.Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: openAIState()}) + if err != nil { + t.Fatalf("Evaluate: %v", err) + } + if resp.Model != "glm-test-1" { + t.Errorf("model = %q", resp.Model) + } + if resp.InputTokens != 100 || resp.OutputTokens != 50 { + t.Errorf("usage = %d/%d", resp.InputTokens, resp.OutputTokens) + } + if len(resp.Findings) != 1 || resp.Findings[0].Summary != "2 jobs stalled on no_resource" { + t.Fatalf("findings = %+v", resp.Findings) + } +} + +func TestOpenAI_RetriesOn429(t *testing.T) { + n := 0 + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + n++ + if n < 3 { + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(`{"error":{"message":"rate limited"}}`)) + return + } + _, _ = w.Write([]byte(`{"model":"m","choices":[{"message":{"content":` + jsonQuote(goodContent) + `}}]}`)) + }) + resp, err := p.Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: openAIState()}) + if err != nil { + t.Fatalf("Evaluate: %v", err) + } + if n != 3 { + t.Errorf("attempts = %d, want 3", n) + } + if len(resp.Findings) != 1 { + t.Errorf("findings = %+v", resp.Findings) + } +} + +func TestOpenAI_NonRetryable400(t *testing.T) { + n := 0 + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + n++ + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":{"message":"bad request"}}`)) + }) + _, err := p.Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: openAIState()}) + if err == nil { + t.Fatal("want error") + } + if n != 1 { + t.Errorf("attempts = %d, want 1 (400 not retried)", n) + } + if !strings.Contains(err.Error(), "HTTP 400") { + t.Errorf("error should carry status: %v", err) + } +} + +func TestOpenAI_NonJSONContent(t *testing.T) { + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"model":"m","choices":[{"message":{"content":"I think the cluster looks fine, honestly!"}}]}`)) + }) + _, err := p.Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: openAIState()}) + if err == nil || !strings.Contains(err.Error(), "findings object") { + t.Fatalf("want strict-parse error, got %v", err) + } +} + +func TestOpenAI_FencedContent(t *testing.T) { + fenced := "```json\n" + goodContent + "\n```" + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"model":"m","choices":[{"message":{"content":` + jsonQuote(fenced) + `}}]}`)) + }) + resp, err := p.Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: openAIState()}) + if err != nil { + t.Fatalf("fenced content should parse: %v", err) + } + if len(resp.Findings) != 1 { + t.Errorf("findings = %+v", resp.Findings) + } +} + +func TestOpenAI_InvalidEnumRejected(t *testing.T) { + bad := `{"findings":[{"kind":"apocalypse","severity":"degraded","summary":"x","evidence":[],"confidence":0.9}]}` + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"model":"m","choices":[{"message":{"content":` + jsonQuote(bad) + `}}]}`)) + }) + _, err := p.Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: openAIState()}) + if err == nil || !strings.Contains(err.Error(), "kind") { + t.Fatalf("want enum rejection, got %v", err) + } +} + +func TestOpenAI_EmptyFindingsValid(t *testing.T) { + p := newTestSrv(t, func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"model":"m","choices":[{"message":{"content":"{\"findings\":[]}"}}]}`)) + }) + resp, err := p.Evaluate(context.Background(), EvaluateRequest{Scenario: "pend_stall", State: openAIState()}) + if err != nil || len(resp.Findings) != 0 { + t.Fatalf("empty findings should be valid: %v %+v", err, resp) + } +} + +// TestVerifyEvidence_DowngradesFabricated is the hallucination firewall. +func TestVerifyEvidence_DowngradesFabricated(t *testing.T) { + state := openAIState() + findings := []Finding{ + {Summary: "honest", Evidence: []string{"1-0-1-0"}, Confidence: 0.9}, + {Summary: "hallucinated", Evidence: []string{"host-faker-99", "job-000000"}, Confidence: 0.95}, + {Summary: "no evidence", Evidence: nil, Confidence: 0.7}, + } + VerifyEvidence(state, findings) + + if findings[0].Unverified { + t.Errorf("honest finding marked unverified: %+v", findings[0]) + } + if !findings[1].Unverified || findings[1].Confidence != 0.2 { + t.Errorf("hallucinated finding not demoted: %+v", findings[1]) + } + if findings[2].Unverified { + t.Errorf("empty evidence should not be flagged: %+v", findings[2]) + } +} + +// jsonQuote returns s as a JSON string literal (for embedding content). +func jsonQuote(s string) string { + b, _ := json.Marshal(s) + return string(b) +} diff --git a/src/frontend/advisor/internal/provider/provider.go b/src/frontend/advisor/internal/provider/provider.go new file mode 100644 index 00000000..921ba375 --- /dev/null +++ b/src/frontend/advisor/internal/provider/provider.go @@ -0,0 +1,102 @@ +// Package provider defines the advisor's pluggable decision layer. +// +// The advisor never lets an LLM act: a Provider only turns a scenario +// snapshot into structured Findings, and the agent (observe mode) merely +// records them. Two implementations ship in P0: +// +// - openai: an OpenAI-compatible chat-completions endpoint (GLM, Qwen, +// DeepSeek, or a local vLLM) asked for strict JSON. +// - echo: a deterministic rules implementation — the always-available +// fallback, the unit-test baseline, and the yardstick the LLM output is +// compared against. +// +// A future jev provider (TypeSafe System One, recorded but not implemented) +// maps the same Finding contract onto typed noul/choice/score answers; see +// src/docs/advisor-design.md. +package provider + +import ( + "context" + "encoding/json" + "time" +) + +// Severity rates how much attention a finding deserves. Ordered weakest to +// strongest so code may compare severities directly. +type Severity string + +const ( + SevInfo Severity = "info" + SevDegraded Severity = "degraded" + SevSevere Severity = "severe" + SevOutage Severity = "outage" +) + +// Finding is one structured diagnostic conclusion about the cluster state. +// +// Suggestion is advisory text for a human operator — the advisor itself +// never executes it (observe mode; propose/act come later behind separate +// gates). Confidence is the model's self-report for LLM providers and is +// NOT calibrated; code may only use it for display and logging, never for +// autonomous decisions. +type Finding struct { + // Kind classifies the finding: "anomaly" (something looks wrong) or + // "diagnosis" (a cause is proposed for a known anomaly). + Kind string `json:"kind"` + // Severity rates the attention it deserves. + Severity Severity `json:"severity"` + // Summary is a one-line human-readable statement. + Summary string `json:"summary"` + // Evidence cites concrete identifiers from the state snapshot (job ids, + // host names, reason names, counts). The agent verifies every element + // appears verbatim in the state it sent and downgrades findings whose + // evidence does not (hallucination firewall). + Evidence []string `json:"evidence"` + // Confidence is the provider's self-assessed certainty, 0..1. Not + // calibrated for LLM providers. + Confidence float64 `json:"confidence"` + // Suggestion is advisory-only operator guidance (never executed by P0). + Suggestion string `json:"suggestion,omitempty"` + // Unverified is set by the agent's evidence check when Evidence cites + // strings absent from the state snapshot. + Unverified bool `json:"unverified,omitempty"` +} + +// EvaluateRequest carries one evaluation round. State is the compact JSON +// snapshot (see internal/snapshot); Scenario names the detector that fired +// (e.g. "pend_stall"); Runbook is the rendered runbook context the provider +// should reason with. +type EvaluateRequest struct { + Scenario string + State json.RawMessage + Runbook string +} + +// EvaluateResponse is one provider round: the parsed findings plus +// bookkeeping for the decision log. +type EvaluateResponse struct { + Findings []Finding + // Model identifies the exact backend that answered (e.g. "jev-1.13.0" + // or "glm-4-flash"), as reported by the provider — recorded verbatim. + Model string + // Raw is the untouched provider output, kept in the decision log so a + // bad parse is always diagnosable after the fact. + Raw string + // Usage counts tokens when the backend reports them. + InputTokens int + OutputTokens int + // Latency is the provider round-trip. + Latency time.Duration +} + +// Provider turns scenario snapshots into findings. Implementations must be +// safe for sequential use by the agent loop; the advisor never evaluates +// concurrently. +type Provider interface { + // Name identifies the provider in logs ("openai", "echo", "jev"). + Name() string + // Evaluate produces findings for one scenario. Implementations return + // an error on transport/parse failure; the agent then degrades to the + // echo provider and records the failure. + Evaluate(ctx context.Context, req EvaluateRequest) (*EvaluateResponse, error) +}