feat(advisor): Go module skeleton — config, provider contract, echo + openai providers

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 <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-10-19 03:31:36 +00:00
parent f94cc08680
commit 1ab96521b8
7 changed files with 1124 additions and 0 deletions

View file

@ -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
)

View file

@ -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
}

View file

@ -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,
}}
}

View file

@ -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")
}
}

View file

@ -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] + "..."
}

View file

@ -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)
}

View file

@ -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)
}