From ab12f492564dd1110d36519ba093cf8c148572f3 Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Mon, 19 Oct 2026 03:42:01 +0000 Subject: [PATCH] feat(advisor): agent loop, decision log, runbook, run/once CLI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - internal/agent: observe-mode loop — periodic scan + SSE job-event pull (events only pull the next scan earlier, debounced 5s; job.finished feeds exit_burst). Provider chain: primary (LLM) → echo fallback on failure, recorded as degraded. VerifyEvidence firewall runs on every LLM answer. Observe-only by construction: no write path reachable. - internal/declog: JSONL decision log (scenario, trigger, provider, model, latency, tokens, findings, raw output, state snapshot + digest, degraded flag), size-rotated. - internal/runbook: incident-pattern seed (embedded YAML) from the cluster's real debugging history (bug-1494 stale config, IPC ABI skew, bug-1513 rejoin crashes, bug-1521 reason race, first-fit bias, dispatch-fail repend family), rendered into the provider prompt. - cmd: advisor run (signal-handled observe loop) and advisor once (single round, JSON report to stdout, no decision log). - internal/config tests; full suite CGO_ENABLED=0 go test ./... clean. Co-Authored-By: Claude Code --- src/frontend/advisor/cmd/main.go | 151 ++++++++++ src/frontend/advisor/go.mod | 5 +- src/frontend/advisor/go.sum | 4 + src/frontend/advisor/internal/agent/agent.go | 278 ++++++++++++++++++ .../advisor/internal/agent/agent_test.go | 274 +++++++++++++++++ .../advisor/internal/config/config_test.go | 70 +++++ .../advisor/internal/declog/declog.go | 111 +++++++ .../advisor/internal/declog/declog_test.go | 94 ++++++ .../advisor/internal/runbook/runbook.go | 72 +++++ .../advisor/internal/runbook/runbook.yaml | 102 +++++++ .../advisor/internal/runbook/runbook_test.go | 53 ++++ 11 files changed, 1213 insertions(+), 1 deletion(-) create mode 100644 src/frontend/advisor/cmd/main.go create mode 100644 src/frontend/advisor/internal/agent/agent.go create mode 100644 src/frontend/advisor/internal/agent/agent_test.go create mode 100644 src/frontend/advisor/internal/config/config_test.go create mode 100644 src/frontend/advisor/internal/declog/declog.go create mode 100644 src/frontend/advisor/internal/declog/declog_test.go create mode 100644 src/frontend/advisor/internal/runbook/runbook.go create mode 100644 src/frontend/advisor/internal/runbook/runbook.yaml create mode 100644 src/frontend/advisor/internal/runbook/runbook_test.go diff --git a/src/frontend/advisor/cmd/main.go b/src/frontend/advisor/cmd/main.go new file mode 100644 index 00000000..267056d6 --- /dev/null +++ b/src/frontend/advisor/cmd/main.go @@ -0,0 +1,151 @@ +// Command advisor is the XCE scheduler advisor — an observe-only LLM ops +// copilot. It watches the cluster (periodic scans + SSE job events), runs +// deterministic detectors, asks a provider (OpenAI-compatible LLM or the +// deterministic echo fallback) for structured findings, and records +// everything in a JSONL decision log. It never executes anything: findings +// are for humans. +// +// Usage: +// +// advisor run # long-running observe loop +// advisor once # one scan + evaluate round, report to stdout (no log) +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + "os/signal" + "syscall" + + "advisor/internal/agent" + "advisor/internal/config" + "advisor/internal/declog" + "advisor/internal/detect" + "advisor/internal/provider" + "advisor/internal/runbook" + + sclient "schedulerclient" +) + +func main() { + if len(os.Args) < 2 { + usage() + os.Exit(2) + } + switch os.Args[1] { + case "run": + cmdRun() + case "once": + cmdOnce() + case "-h", "--help", "help": + usage() + default: + fmt.Fprintf(os.Stderr, "unknown command %q\n", os.Args[1]) + usage() + os.Exit(2) + } +} + +func usage() { + fmt.Fprint(os.Stderr, `advisor — XCE scheduler advisor (observe mode) + +usage: + advisor run start the observe loop (periodic scan + SSE events) + advisor once run one scan/evaluate round and print the report + +configuration (env): + XCE_API_KEY required — mcp_readonly service-account key + XCE_API_URL apiserver base URL (default http://172.31.1.30:8080) + ADVISOR_LLM_BASE_URL OpenAI-compatible endpoint (unset = echo-only) + ADVISOR_LLM_API_KEY LLM credentials + ADVISOR_LLM_MODEL model name (required when BASE_URL set) + ADVISOR_LLM_TIMEOUT per-call timeout (default 30s) + ADVISOR_INTERVAL scan interval (default 120s) + ADVISOR_LOG decision log path (default /tmp/advisor-decisions.jsonl) + ADVISOR_RUNBOOK runbook YAML overriding the embedded one +`) +} + +func mustConfig() *config.Config { + cfg, err := config.FromEnv() + if err != nil { + log.Fatalf("advisor: %v", err) + } + return cfg +} + +func mustClient(cfg *config.Config) sclient.Client { + return sclient.NewClient(cfg.APIURL, cfg.APIKey) +} + +// pickProvider builds the primary provider per config: LLM when configured, +// otherwise echo-only. +func pickProvider(cfg *config.Config) provider.Provider { + if !cfg.LLMEnabled() { + log.Printf("advisor: no ADVISOR_LLM_BASE_URL — echo-only mode (deterministic rules)") + return nil + } + log.Printf("advisor: LLM provider %s model=%s", cfg.LLMBaseURL, cfg.LLMModel) + return provider.NewOpenAI(cfg.LLMBaseURL, cfg.LLMAPIKey, cfg.LLMModel, cfg.LLMTimeout) +} + +func loadRunbook(cfg *config.Config) *runbook.Runbook { + rb, err := runbook.Load(cfg.RunbookPath) + if err != nil { + log.Fatalf("advisor: runbook: %v", err) + } + return rb +} + +func cmdRun() { + cfg := mustConfig() + rb := loadRunbook(cfg) + dl, err := declog.Open(cfg.LogPath) + if err != nil { + log.Fatalf("advisor: decision log: %v", err) + } + defer dl.Close() + + a := agent.New(mustClient(cfg), detect.NewDetector(detect.DefaultLimits()), + pickProvider(cfg), dl, cfg.Interval) + a.RunbookText = rb.Render() + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + log.Printf("advisor: observe loop starting (interval %s, log %s)", cfg.Interval, cfg.LogPath) + if err := a.Run(ctx); err != nil && ctx.Err() == nil { + log.Fatalf("advisor: %v", err) + } + log.Printf("advisor: stopped") +} + +func cmdOnce() { + cfg := mustConfig() + rb := loadRunbook(cfg) + + // `once` prints the report to stdout and does not touch the decision + // log — it is the development/demo/diagnostic path. + a := agent.New(mustClient(cfg), detect.NewDetector(detect.DefaultLimits()), + pickProvider(cfg), discardLog(), cfg.Interval) + a.RunbookText = rb.Render() + + report, err := a.Once(context.Background()) + if err != nil { + log.Fatalf("advisor: %v", err) + } + out, _ := json.MarshalIndent(report, "", " ") + fmt.Println(string(out)) +} + +// discardLog swallows records for the `once` path (stdout is the channel). +func discardLog() *declog.Log { + l, err := declog.Open(os.DevNull) + if err != nil { + log.Fatalf("advisor: %v", err) + } + return l +} diff --git a/src/frontend/advisor/go.mod b/src/frontend/advisor/go.mod index aab476f8..212a49de 100644 --- a/src/frontend/advisor/go.mod +++ b/src/frontend/advisor/go.mod @@ -4,7 +4,10 @@ go 1.25.5 replace schedulerclient => ../schedulerclient -require schedulerclient v0.0.0-00010101000000-000000000000 +require ( + gopkg.in/yaml.v3 v3.0.1 + schedulerclient v0.0.0-00010101000000-000000000000 +) require ( golang.org/x/net v0.33.0 // indirect diff --git a/src/frontend/advisor/go.sum b/src/frontend/advisor/go.sum index a5fed1cd..eb052bfd 100644 --- a/src/frontend/advisor/go.sum +++ b/src/frontend/advisor/go.sum @@ -1,4 +1,8 @@ golang.org/x/net v0.33.0 h1:74SYHlV8BIgHIFC/LrYkOGIwL19eTYXQ5wc6TBuO36I= golang.org/x/net v0.33.0/go.mod h1:HXLR5J+9DxmrqMwG9qjGCxZ+zKXxBru04zlTvWlWuN4= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= resty.dev/v3 v3.0.0-beta.2 h1:xu4mGAdbCLuc3kbk7eddWfWm4JfhwDtdapwss5nCjnQ= resty.dev/v3 v3.0.0-beta.2/go.mod h1:OgkqiPvTDtOuV4MGZuUDhwOpkY8enjOsjjMzeOHefy4= diff --git a/src/frontend/advisor/internal/agent/agent.go b/src/frontend/advisor/internal/agent/agent.go new file mode 100644 index 00000000..82ed9dff --- /dev/null +++ b/src/frontend/advisor/internal/agent/agent.go @@ -0,0 +1,278 @@ +// Package agent wires the advisor loop: periodic scan + SSE event pull, +// deterministic detectors, provider evaluation (with echo degradation), +// and JSONL decision logging. Observe-only: this package has no path to a +// write operation — the schedulerclient write methods are unreachable from +// here, and provider output is recorded, never executed. +package agent + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "log" + "strconv" + "time" + + "advisor/internal/declog" + "advisor/internal/detect" + "advisor/internal/provider" + "advisor/internal/snapshot" + + sclient "schedulerclient" +) + +// Agent is the observe-mode advisor. +type Agent struct { + client sclient.Client + detector *detect.Detector + builder *snapshot.Builder + primary provider.Provider // may be nil: echo-only mode + echo *provider.Echo + log *declog.Log + // RunbookText is the rendered runbook context handed to providers. + RunbookText string + // ScanInterval mirrors cfg.Interval; SSE events only pull the next scan + // earlier (min-pull gap still applies). + ScanInterval time.Duration + + // lastExits carries the recent-exit list into the next snapshot; the + // snapshot consumes it (exit_burst) and it resets each round. + lastExits []snapshot.ExitState +} + +// New assembles an Agent. primary may be nil (echo-only). +func New(client sclient.Client, det *detect.Detector, primary provider.Provider, dl *declog.Log, scanInterval time.Duration) *Agent { + return &Agent{ + client: client, + detector: det, + builder: snapshot.NewBuilder(client), + primary: primary, + echo: provider.NewEcho(), + log: dl, + ScanInterval: scanInterval, + } +} + +// Run blocks until ctx is cancelled: ticker scans + SSE event pull. +// Event types only accelerate the next scan (debounced); detection and +// evaluation always run on the same periodic path. +func (a *Agent) Run(ctx context.Context) error { + // WatchJobs failures must not kill the loop: reconnect on next tick. + eventCh := a.watchBestEffort(ctx) + + ticker := time.NewTicker(a.ScanInterval) + defer ticker.Stop() + + // Scan immediately at startup so `run` produces a first round without + // waiting a full interval. + a.scanRound(ctx) + + var pendingEvents int + for { + select { + case <-ctx.Done(): + return ctx.Err() + case <-ticker.C: + pendingEvents = 0 + a.scanRound(ctx) + // Re-establish the stream if it died. + if eventCh == nil { + eventCh = a.watchBestEffort(ctx) + } + case ev, ok := <-eventCh: + if !ok { + eventCh = nil // stream ended; ticker will re-establish + continue + } + // Job lifecycle events: PEND transitions and finished jobs are + // the interesting ones. Finished jobs feed exit_burst on the + // next scan; every event pulls the scan earlier (debounced). + pendingEvents++ + a.noteEvent(ev) + // Debounce: let a burst of events land, then scan early. + if pendingEvents == 1 { + go func() { + select { + case <-ctx.Done(): + return + case <-time.After(5 * time.Second): + } + // A tiny timer race with the ticker is fine: the scan + // itself is idempotent and deduped. + a.scanRound(ctx) + }() + } + } + } +} + +// watchBestEffort starts the SSE stream; errors are logged, never fatal. +func (a *Agent) watchBestEffort(ctx context.Context) <-chan sclient.JobEvent { + ch, err := a.client.WatchJobs(ctx) + if err != nil { + log.Printf("advisor: WatchJobs unavailable (%v) — periodic scan only", err) + return nil + } + return ch +} + +// scanRound is one full observe round: snapshot → detect → evaluate → log. +func (a *Agent) scanRound(ctx context.Context) { + st, err := a.builder.Build(ctx, a.lastExits) + if err != nil { + log.Printf("advisor: snapshot failed: %v", err) + return + } + a.lastExits = nil // exits are consumed once + + 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)) + return + } + + stateJSON, err := json.Marshal(st) + if err != nil { + log.Printf("advisor: state marshal: %v", err) + return + } + + for _, tr := range triggers { + a.evaluateTrigger(ctx, tr, stateJSON) + } +} + +// OnceReport is the `advisor once` output: one round's findings by scenario. +type OnceReport struct { + GeneratedAt string `json:"generated_at"` + Scenarios map[string][]provider.Finding `json:"scenarios,omitempty"` + Providers map[string]string `json:"providers"` // scenario → provider name + Models map[string]string `json:"models"` // scenario → model + Degraded map[string]bool `json:"degraded,omitempty"` // scenario → fell back to echo +} + +// Once runs exactly one scan/evaluate round and returns the report without +// logging decisions (the caller prints them). +func (a *Agent) Once(ctx context.Context) (*OnceReport, error) { + st, err := a.builder.Build(ctx, nil) + if err != nil { + return nil, err + } + stateJSON, err := json.Marshal(st) + if err != nil { + return nil, err + } + rep := &OnceReport{ + GeneratedAt: st.GeneratedAt, + Scenarios: map[string][]provider.Finding{}, + Providers: map[string]string{}, + Models: map[string]string{}, + Degraded: map[string]bool{}, + } + for _, tr := range a.detector.Detect(st) { + req := provider.EvaluateRequest{Scenario: tr.Scenario, State: stateJSON, Runbook: a.RunbookText} + resp, err := a.evaluateWith(ctx, req, a.primary) + degraded := false + if err != nil { + degraded = true + resp, err = a.evaluateWith(ctx, req, a.echo) + if err != nil { + return nil, err + } + } + provider.VerifyEvidence(stateJSON, resp.Findings) + name := a.providerName() + rep.Scenarios[tr.Scenario] = resp.Findings + rep.Providers[tr.Scenario] = name + rep.Models[tr.Scenario] = resp.Model + rep.Degraded[tr.Scenario] = degraded + } + return rep, nil +} + +// noteEvent consumes one SSE event: finished jobs feed exit_burst on the +// next scan (the event's exit code mirrors the apiserver's: 0 → DONE). +func (a *Agent) noteEvent(ev sclient.JobEvent) { + if ev.Type == "job.finished" { + a.lastExits = append(a.lastExits, snapshot.ExitState{ + ID: strconv.FormatUint(ev.JobID, 10), + ExitCode: ev.ExitCode, + }) + } +} + +// mustJSON marshals v or returns "[]" on failure (findings always marshal). +func mustJSON(v any) json.RawMessage { + b, err := json.Marshal(v) + if err != nil { + return json.RawMessage("[]") + } + return b +} + +// evaluateTrigger runs the provider chain for one trigger and logs it. +func (a *Agent) evaluateTrigger(ctx context.Context, tr detect.Trigger, stateJSON []byte) { + req := provider.EvaluateRequest{ + Scenario: tr.Scenario, + State: stateJSON, + Runbook: a.RunbookText, + } + + resp, err := a.evaluateWith(ctx, req, a.primary) + degraded := false + if err != nil { + // Primary failed → echo answers, failure is recorded. + degraded = true + log.Printf("advisor: provider %s failed on %s (%v) — degrading to echo", a.providerName(), tr.Scenario, err) + resp, err = a.evaluateWith(ctx, req, a.echo) + if err != nil { + // Echo never fails on valid state; this is a wiring bug. + log.Printf("advisor: echo fallback failed: %v", err) + return + } + } + + // Hallucination firewall (LLM providers): demote findings citing + // identifiers absent from the state. + provider.VerifyEvidence(stateJSON, resp.Findings) + + stateDigest := sha256.Sum256(stateJSON) + rec := declog.Record{ + Scenario: tr.Scenario, + Trigger: tr.Fingerprint, + Provider: a.providerName(), + Model: resp.Model, + LatencyMS: resp.Latency.Milliseconds(), + InputTok: resp.InputTokens, + OutputTok: resp.OutputTokens, + Findings: mustJSON(resp.Findings), + Raw: resp.Raw, + StateDigest: hex.EncodeToString(stateDigest[:]), + State: stateJSON, + Degraded: degraded, + } + if err := a.log.Append(rec); err != nil { + log.Printf("advisor: decision log append failed: %v", err) + } + + log.Printf("advisor: %s [%s] -> %d finding(s) (model=%s, %s, degraded=%v)", + tr.Scenario, tr.Fingerprint, len(resp.Findings), resp.Model, resp.Latency.Round(time.Millisecond), degraded) +} + +func (a *Agent) evaluateWith(ctx context.Context, req provider.EvaluateRequest, p provider.Provider) (*provider.EvaluateResponse, error) { + if p == nil { + // Echo-only mode without a primary: skip straight to echo via the + // nil check at the call site; this branch is for clarity in tests. + return a.echo.Evaluate(ctx, req) + } + return p.Evaluate(ctx, req) +} + +func (a *Agent) providerName() string { + if a.primary == nil { + return a.echo.Name() + } + return a.primary.Name() +} diff --git a/src/frontend/advisor/internal/agent/agent_test.go b/src/frontend/advisor/internal/agent/agent_test.go new file mode 100644 index 00000000..40ecc4b3 --- /dev/null +++ b/src/frontend/advisor/internal/agent/agent_test.go @@ -0,0 +1,274 @@ +package agent + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "advisor/internal/declog" + "advisor/internal/detect" + "advisor/internal/provider" + + sclient "schedulerclient" +) + +// fakeProvider records Evaluate calls. +type fakeProvider struct { + name string + err error + calls int + mu sync.Mutex +} + +func (f *fakeProvider) Name() string { return f.name } +func (f *fakeProvider) Evaluate(_ context.Context, req provider.EvaluateRequest) (*provider.EvaluateResponse, error) { + f.mu.Lock() + f.calls++ + f.mu.Unlock() + if f.err != nil { + return nil, f.err + } + return &provider.EvaluateResponse{ + Findings: []provider.Finding{{ + Kind: "diagnosis", Severity: provider.SevDegraded, + Summary: "fake says stalled", Evidence: []string{"1-0-1-0"}, + Confidence: 0.9, + }}, + Model: f.name + "-model", + Raw: `{"raw":true}`, + }, nil +} + +// fakeClient: read-only fixture, panics on writes. +type fakeClient struct { + nodes []*sclient.URMNode + pend []*sclient.URMJob +} + +func (f *fakeClient) ListJobs(_ context.Context, filter sclient.JobFilter) ([]*sclient.URMJob, *sclient.Page, error) { + var out []*sclient.URMJob + for _, s := range filter.Statuses { + if s == sclient.JobStatusPending { + out = append(out, f.pend...) + } + } + return out, &sclient.Page{}, nil +} +func (f *fakeClient) ListNodes(context.Context) ([]*sclient.URMNode, error) { + return f.nodes, nil +} +func (f *fakeClient) ListQueues(context.Context) ([]*sclient.URMQueue, error) { + return nil, 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 violation: SubmitJob reached") +} +func (f *fakeClient) CancelJob(context.Context, string) error { panic("observe-only violation") } +func (f *fakeClient) HoldJob(context.Context, string) error { panic("observe-only violation") } +func (f *fakeClient) ReleaseJob(context.Context, string) error { + panic("observe-only violation") +} +func (f *fakeClient) CheckpointJob(context.Context, string) error { panic("observe-only violation") } +func (f *fakeClient) RestartJob(context.Context, string) (*sclient.RestartJobResult, error) { + panic("observe-only violation") +} +func (f *fakeClient) WatchJobs(context.Context) (<-chan sclient.JobEvent, error) { + ch := make(chan sclient.JobEvent) + return ch, nil // never sends: tests drive scanRound directly +} + +func stalledFixture() *fakeClient { + return &fakeClient{ + nodes: []*sclient.URMNode{ + {Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}, + }, + pend: []*sclient.URMJob{ + {ID: "1-0-1-0", SubmissionTime: time.Now().Add(-time.Hour), + Extensions: map[string]any{"pending_reason": `{"main_reason_name":"no_resource"}`}}, + }, + } +} + +func newTestAgent(t *testing.T, fc *fakeClient, primary provider.Provider) (*Agent, *fakeProvider, string) { + t.Helper() + path := filepath.Join(t.TempDir(), "dec.jsonl") + dl, err := declog.Open(path) + if err != nil { + t.Fatalf("declog: %v", err) + } + t.Cleanup(func() { dl.Close() }) + fp, _ := primary.(*fakeProvider) + a := New(fc, detect.NewDetector(detect.DefaultLimits()), primary, dl, time.Minute) + a.RunbookText = "test runbook" + return a, fp, path +} + +func TestScanRound_PrimaryProviderLogged(t *testing.T) { + fc := stalledFixture() + fp := &fakeProvider{name: "fake"} + a, _, path := newTestAgent(t, fc, fp) + + a.scanRound(context.Background()) + + if fp.calls != 1 { + t.Fatalf("provider calls = %d", fp.calls) + } + rec := readOneRecord(t, path) + if rec.Provider != "fake" || rec.Model != "fake-model" || rec.Scenario != "pend_stall" { + t.Errorf("record mismatch: %+v", rec) + } + if rec.Degraded { + t.Errorf("healthy provider marked degraded") + } + if len(rec.Findings) == 0 { + t.Errorf("findings empty") + } + // Evidence firewall ran: "1-0-1-0" IS in the state (pending job id). + var fs []provider.Finding + _ = json.Unmarshal(rec.Findings, &fs) + for _, f := range fs { + if f.Unverified { + t.Errorf("verified evidence marked unverified: %+v", f) + } + } +} + +func TestScanRound_DegradeToEchoOnPrimaryFailure(t *testing.T) { + fc := stalledFixture() + fp := &fakeProvider{name: "broken", err: context.DeadlineExceeded} + a, _, path := newTestAgent(t, fc, fp) + + a.scanRound(context.Background()) + + rec := readOneRecord(t, path) + if !rec.Degraded { + t.Errorf("degraded flag not set: %+v", rec) + } + if rec.Provider != "broken" { + t.Errorf("record should name the primary provider: %+v", rec) + } + if rec.Model != "echo-rules" { + t.Errorf("echo should have answered: model=%q", rec.Model) + } +} + +func TestScanRound_CleanClusterNoCalls(t *testing.T) { + fc := &fakeClient{ + nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}}, + } + fp := &fakeProvider{name: "fake"} + a, _, _ := newTestAgent(t, fc, fp) + + a.scanRound(context.Background()) + if fp.calls != 0 { + t.Fatalf("clean cluster must not call the provider: %d calls", fp.calls) + } +} + +func TestScanRound_DedupSuppressesRepeat(t *testing.T) { + fc := stalledFixture() + fp := &fakeProvider{name: "fake"} + a, _, _ := newTestAgent(t, fc, fp) + + a.scanRound(context.Background()) + a.scanRound(context.Background()) + if fp.calls != 1 { + t.Fatalf("identical situation must dedup: %d calls", fp.calls) + } +} + +func TestScanRound_HallucinatedEvidenceDemoted(t *testing.T) { + fc := stalledFixture() + + path := filepath.Join(t.TempDir(), "dec.jsonl") + dl, _ := declog.Open(path) + defer dl.Close() + p := &oneShotProvider{resp: &provider.EvaluateResponse{ + Findings: []provider.Finding{{ + Kind: "diagnosis", Severity: provider.SevSevere, + Summary: "fabricated", Evidence: []string{"host-never-existed"}, + Confidence: 0.99, + }}, + Model: "liar-model", + }, name: "liar"} + a := New(fc, detect.NewDetector(detect.DefaultLimits()), p, dl, time.Minute) + a.scanRound(context.Background()) + + rec := readOneRecord(t, path) + var fs []provider.Finding + if err := json.Unmarshal(rec.Findings, &fs); err != nil { + t.Fatalf("findings parse: %v", err) + } + if len(fs) != 1 || !fs[0].Unverified || fs[0].Confidence != 0.2 { + t.Fatalf("hallucinated finding not demoted: %+v", fs) + } +} + +type oneShotProvider struct { + resp *provider.EvaluateResponse + name string +} + +func (o *oneShotProvider) Name() string { return o.name } +func (o *oneShotProvider) Evaluate(context.Context, provider.EvaluateRequest) (*provider.EvaluateResponse, error) { + return o.resp, nil +} + +func TestNoteEvent_FinishedFeedsExitBurst(t *testing.T) { + fc := &fakeClient{ + nodes: []*sclient.URMNode{{Name: "t1", Status: "busy", CoresTotal: 8, CoresUsed: 4}}, + } + fp := &fakeProvider{name: "fake"} + a, _, path := newTestAgent(t, fc, fp) + + // Three finished jobs with non-zero exits → exit_burst fires next round. + for i := 0; i < 3; i++ { + a.noteEvent(sclient.JobEvent{Type: "job.finished", JobID: uint64(100 + i), ExitCode: 127}) + } + a.scanRound(context.Background()) + + rec := readOneRecord(t, path) + if rec.Scenario != "exit_burst" { + t.Fatalf("want exit_burst, got %s", rec.Scenario) + } +} + +func readOneRecord(t *testing.T, path string) declog.Record { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read log: %v", err) + } + // First line only (tests produce exactly one record per round). + line := data + if i := indexByte(data, '\n'); i >= 0 { + line = data[:i] + } + var rec declog.Record + if err := json.Unmarshal(line, &rec); err != nil { + t.Fatalf("record parse: %v (%s)", err, string(line)) + } + return rec +} + +func indexByte(b []byte, c byte) int { + for i, x := range b { + if x == c { + return i + } + } + return -1 +} diff --git a/src/frontend/advisor/internal/config/config_test.go b/src/frontend/advisor/internal/config/config_test.go new file mode 100644 index 00000000..1aedc8de --- /dev/null +++ b/src/frontend/advisor/internal/config/config_test.go @@ -0,0 +1,70 @@ +package config + +import ( + "testing" + "time" +) + +func setenvs(t *testing.T, kv map[string]string) { + t.Helper() + for _, k := range []string{"XCE_API_KEY", "XCE_API_URL", "ADVISOR_LLM_BASE_URL", "ADVISOR_LLM_MODEL", + "ADVISOR_INTERVAL", "ADVISOR_LLM_TIMEOUT", "ADVISOR_LOG", "ADVISOR_RUNBOOK"} { + t.Setenv(k, "") // clear first + } + for k, v := range kv { + t.Setenv(k, v) + } +} + +func TestFromEnv_Minimal(t *testing.T) { + setenvs(t, map[string]string{"XCE_API_KEY": "xce_ak_x"}) + c, err := FromEnv() + if err != nil { + t.Fatalf("FromEnv: %v", err) + } + if c.APIURL != defaultAPIURL { + t.Errorf("default APIURL = %q", c.APIURL) + } + if c.LLMEnabled() { + t.Errorf("LLM should be disabled by default") + } + if c.Interval != defaultInterval { + t.Errorf("default interval = %v", c.Interval) + } +} + +func TestFromEnv_MissingKey(t *testing.T) { + setenvs(t, nil) + if _, err := FromEnv(); err == nil { + t.Fatal("missing XCE_API_KEY must error") + } +} + +func TestFromEnv_LLMRequiresModel(t *testing.T) { + setenvs(t, map[string]string{"XCE_API_KEY": "k", "ADVISOR_LLM_BASE_URL": "http://llm"}) + if _, err := FromEnv(); err == nil { + t.Fatal("BASE_URL without MODEL must error") + } + setenvs(t, map[string]string{ + "XCE_API_KEY": "k", "ADVISOR_LLM_BASE_URL": "http://llm", "ADVISOR_LLM_MODEL": "glm-4-flash", + }) + c, err := FromEnv() + if err != nil || !c.LLMEnabled() || c.LLMModel != "glm-4-flash" { + t.Fatalf("LLM config failed: %v %+v", err, c) + } +} + +func TestFromEnv_Durations(t *testing.T) { + setenvs(t, map[string]string{"XCE_API_KEY": "k", "ADVISOR_INTERVAL": "45s", "ADVISOR_LLM_TIMEOUT": "10s"}) + c, err := FromEnv() + if err != nil { + t.Fatalf("FromEnv: %v", err) + } + if c.Interval != 45*time.Second || c.LLMTimeout != 10*time.Second { + t.Errorf("durations = %v %v", c.Interval, c.LLMTimeout) + } + setenvs(t, map[string]string{"XCE_API_KEY": "k", "ADVISOR_INTERVAL": "bogus"}) + if _, err := FromEnv(); err == nil { + t.Fatal("bad interval must error") + } +} diff --git a/src/frontend/advisor/internal/declog/declog.go b/src/frontend/advisor/internal/declog/declog.go new file mode 100644 index 00000000..1806c172 --- /dev/null +++ b/src/frontend/advisor/internal/declog/declog.go @@ -0,0 +1,111 @@ +// Package declog appends the advisor's decision log as JSON Lines: one +// record per evaluation with the full context (scenario, trigger, provider, +// model, findings, raw output, the state snapshot) so any conclusion is +// reconstructible after the fact. Size-rotated by rename. +package declog + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" + "time" +) + +// Record is one evaluation entry. +type Record struct { + TS string `json:"ts"` // RFC3339 + Scenario string `json:"scenario"` + Trigger string `json:"trigger"` // detector fingerprint + Provider string `json:"provider"` + Model string `json:"model"` + LatencyMS int64 `json:"latency_ms"` + InputTok int `json:"input_tokens,omitempty"` + OutputTok int `json:"output_tokens,omitempty"` + Findings json.RawMessage `json:"findings"` + Raw string `json:"raw,omitempty"` + StateDigest string `json:"state_digest"` + State json.RawMessage `json:"state"` + // Degraded is set when the primary provider failed and echo answered. + Degraded bool `json:"degraded,omitempty"` + Error string `json:"error,omitempty"` +} + +// Log is a JSONL appender. +type Log struct { + path string + mu sync.Mutex + f *os.File + // MaxBytes rotates at this size (default 50 MiB); 0 disables. + MaxBytes int64 +} + +// Open creates/opens the log file for appending. +func Open(path string) (*Log, error) { + if dir := filepath.Dir(path); dir != "" && dir != "." { + if err := os.MkdirAll(dir, 0o755); err != nil { + return nil, fmt.Errorf("declog: mkdir: %w", err) + } + } + f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return nil, fmt.Errorf("declog: open: %w", err) + } + return &Log{path: path, f: f, MaxBytes: 50 << 20}, nil +} + +// Append writes one record as a single JSON line. +func (l *Log) Append(rec Record) error { + if rec.TS == "" { + rec.TS = time.Now().UTC().Format(time.RFC3339) + } + line, err := json.Marshal(rec) + if err != nil { + return fmt.Errorf("declog: marshal: %w", err) + } + line = append(line, '\n') + + l.mu.Lock() + defer l.mu.Unlock() + if err := l.maybeRotate(int64(len(line))); err != nil { + return err + } + _, err = l.f.Write(line) + return err +} + +// maybeRotate renames the log to .1 when it has grown past MaxBytes. +// Callers hold l.mu. +func (l *Log) maybeRotate(incoming int64) error { + if l.MaxBytes <= 0 { + return nil + } + info, err := l.f.Stat() + if err != nil { + return nil // stat failure never blocks logging + } + if info.Size()+incoming < l.MaxBytes { + return nil + } + if err := l.f.Close(); err != nil { + return err + } + _ = os.Remove(l.path + ".1") + if err := os.Rename(l.path, l.path+".1"); err != nil { + return fmt.Errorf("declog: rotate: %w", err) + } + f, err := os.OpenFile(l.path, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o644) + if err != nil { + return fmt.Errorf("declog: reopen: %w", err) + } + l.f = f + return nil +} + +// Close flushes and closes the file. +func (l *Log) Close() error { + l.mu.Lock() + defer l.mu.Unlock() + return l.f.Close() +} diff --git a/src/frontend/advisor/internal/declog/declog_test.go b/src/frontend/advisor/internal/declog/declog_test.go new file mode 100644 index 00000000..58290e78 --- /dev/null +++ b/src/frontend/advisor/internal/declog/declog_test.go @@ -0,0 +1,94 @@ +package declog + +import ( + "bufio" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestAppendAndParse(t *testing.T) { + path := filepath.Join(t.TempDir(), "dec", "log.jsonl") + l, err := Open(path) + if err != nil { + t.Fatalf("Open: %v", err) + } + defer l.Close() + + rec := Record{ + Scenario: "pend_stall", Trigger: "pending=2", Provider: "echo", + Model: "echo-rules", Findings: json.RawMessage(`[{"kind":"diagnosis"}]`), + StateDigest: "abc", State: json.RawMessage(`{}`), + } + if err := l.Append(rec); err != nil { + t.Fatalf("Append: %v", err) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read: %v", err) + } + lines := strings.Split(strings.TrimRight(string(data), "\n"), "\n") + if len(lines) != 1 { + t.Fatalf("want 1 line, got %d", len(lines)) + } + var back Record + if err := json.Unmarshal([]byte(lines[0]), &back); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if back.Scenario != "pend_stall" || back.Provider != "echo" { + t.Errorf("roundtrip mismatch: %+v", back) + } + if back.TS == "" { + t.Errorf("TS auto-filled") + } +} + +func TestRotate(t *testing.T) { + path := filepath.Join(t.TempDir(), "log.jsonl") + l, err := Open(path) + if err != nil { + t.Fatalf("Open: %v", err) + } + l.MaxBytes = 300 // tiny to force rotation + + for i := 0; i < 5; i++ { + rec := Record{ + Scenario: "s", Trigger: "t", Provider: "echo", Model: "m", + Findings: json.RawMessage(`[]`), StateDigest: "d", State: json.RawMessage(`{"i":` + jsonInt(i) + `}`), + } + if err := l.Append(rec); err != nil { + t.Fatalf("Append %d: %v", i, err) + } + } + l.Close() + + if _, err := os.Stat(path + ".1"); err != nil { + t.Fatalf("rotated file missing: %v", err) + } + // Both files parse as JSONL. + for _, p := range []string{path, path + ".1"} { + f, err := os.Open(p) + if err != nil { + if os.IsNotExist(err) && p == path { + continue // everything rotated is fine + } + t.Fatalf("open %s: %v", p, err) + } + sc := bufio.NewScanner(f) + for sc.Scan() { + var r Record + if err := json.Unmarshal(sc.Bytes(), &r); err != nil { + t.Fatalf("%s: bad line: %v", p, err) + } + } + f.Close() + } +} + +func jsonInt(i int) string { + b, _ := json.Marshal(i) + return string(b) +} diff --git a/src/frontend/advisor/internal/runbook/runbook.go b/src/frontend/advisor/internal/runbook/runbook.go new file mode 100644 index 00000000..6e14f8a7 --- /dev/null +++ b/src/frontend/advisor/internal/runbook/runbook.go @@ -0,0 +1,72 @@ +// Package runbook loads the advisor's incident playbook: structured records +// of past failures (symptom fingerprint → root cause → fix) rendered as the +// context LLM providers reason with. This is the poor-man's RAG: a handful +// of curated entries inline in the prompt, no vector store. +package runbook + +import ( + "embed" + "fmt" + "os" + "strings" + + "gopkg.in/yaml.v3" +) + +//go:embed runbook.yaml +var embedded embed.FS + +// Entry is one incident record from history. +type Entry struct { + Name string `yaml:"name" json:"name"` + Symptoms string `yaml:"symptoms" json:"symptoms"` + RootCause string `yaml:"root_cause" json:"root_cause"` + Fix string `yaml:"fix" json:"fix"` + Signals string `yaml:"signals" json:"signals"` +} + +// Runbook is the loaded playbook. +type Runbook struct { + Entries []Entry `yaml:"entries" json:"entries"` +} + +// Load reads the runbook: the path override when given, otherwise the +// embedded seed. +func Load(path string) (*Runbook, error) { + var data []byte + if path != "" { + b, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read %s: %w", path, err) + } + data = b + } else { + b, err := embedded.ReadFile("runbook.yaml") + if err != nil { + return nil, fmt.Errorf("embedded runbook: %w", err) + } + data = b + } + var rb Runbook + if err := yaml.Unmarshal(data, &rb); err != nil { + return nil, fmt.Errorf("parse runbook: %w", err) + } + return &rb, nil +} + +// Render flattens the runbook into the text block handed to providers. +func (r *Runbook) Render() string { + if r == nil || len(r.Entries) == 0 { + return "(no runbook entries)" + } + var b strings.Builder + b.WriteString("Known incident patterns from this cluster's history (symptoms → root cause → fix):\n") + for _, e := range r.Entries { + fmt.Fprintf(&b, "- %s\n symptoms: %s\n root cause: %s\n fix: %s\n", + e.Name, e.Symptoms, e.RootCause, e.Fix) + if e.Signals != "" { + fmt.Fprintf(&b, " distinguishing signals: %s\n", e.Signals) + } + } + return b.String() +} diff --git a/src/frontend/advisor/internal/runbook/runbook.yaml b/src/frontend/advisor/internal/runbook/runbook.yaml new file mode 100644 index 00000000..92720309 --- /dev/null +++ b/src/frontend/advisor/internal/runbook/runbook.yaml @@ -0,0 +1,102 @@ +# Advisor runbook — incident patterns from this cluster's real debugging +# history. Each entry is seed knowledge for the LLM provider's prompt (and a +# checklist for humans). Keep entries short and specific; add new ones as +# incidents are resolved. +entries: + - name: stale-deployment-config + symptoms: > + Every job PENDs with no_resource or match_fail although nodes look + healthy; scheduler logs show no dispatch attempts or repeated dispatch + failures; began after a cluster reconfiguration or redeploy. + root_cause: > + Deployment config drift — the controller's cluster.bin was built from + stale resources.yaml/hosts.yaml (missing slots/mem, placeholder + hostnames), so the scheduler's view of the cluster does not match the + nodes actually running (bug-1494). + fix: > + Compare config-declared resources against observed nodes + (advisor snapshot nodes vs resources.yaml); rebuild cluster.bin from + the repo's src/configs, config push, restart controller. + signals: > + pend reason uniform across unrelated jobs; config hash or deploy date + changed recently; node slots/mem in snapshot differ from config. + + - name: ipc-abi-skew-dead-wait + symptoms: > + Jobs submitted but never start; worker logs show repeated + "Invalid worker PID: 0" from the master side; worker waits forever on + job assignment; began after a partial binary upgrade. + root_cause: > + Master/worker IPC ABI skew — xcecd and xcecd_worker from different + builds (ipc_message_t changed by bug-1936/1957: union grew, tail + fields shifted), so the master misreads the worker's MSG_GET_JOB and + discards it. + fix: > + Never upgrade xcecd without xcecd_worker: replace the pair on every + node in the same maintenance window; verify identical md5 across + nodes; restart workers. + signals: > + "Invalid worker PID: 0" in logs; recent node-local upgrade; md5 + mismatch between nodes' xcecd/xcecd_worker. + + - name: rejoin-state-resync-crash + symptoms: > + Scheduler crashes on worker rejoin (random, load-dependent); core + dumps in makeGroup or limit key paths; worker reconnect storms after + network blips. + root_cause: > + Rejoin state-resync bugs: shared string raw-freed in makeGroup, + EEXIST double-free on limit key_add, config-vs-job race (bug-1513 + family). + fix: > + Update to a build containing the bug-1513 fixes; check core dump + stacks to confirm which path; do not cycle workers while the old + build is live. + signals: > + core stack in rejoin paths; crash correlates with worker reconnect, + not with job mix. + + - name: pend-reason-race-heap-corruption + symptoms: > + Random scheduler crashes in fcfs picker or allocator; heap corruption + signatures; reason strings garbled in logs before the crash. + root_cause: > + Data race: fcfs picker thread writes job->schedobj->reason while the + allocator thread MOVE_PTRs it (bug-1521); #17 mitigates by not + attaching, fix pending. + fix: > + Avoid the mitigated-against path (do not attach reason), or run a + build with the race fixed; collect coredumps for confirmation. + signals: > + crash stacks split between picker and allocator; ASAN or glibc + "free(): invalid pointer" style signatures. + + - name: firstfit-placement-bias + symptoms: > + All jobs land on one host (historically t1 slot 0) while other hosts + idle; no errors; cluster "works" but is half utilized. + root_cause: > + First-fit placement with insertion-order slot iteration — not a bug, + a policy bias (see memory xce-firstfit-t1-affinity). + fix: > + Policy change, not an incident: adjust placement strategy (round-robin + / best-fit) at the scheduler level; do not restart anything. + signals: > + per-host utilization skew persistent across days; zero dispatch + failures; hot host is the first in the snapshot's node list. + + - name: dispatch-fail-repend-storm + symptoms: > + Jobs cycle PEND without progress; scheduler logs show dispatch + attempts failing; worker slots leak or multi-worker double-dispatch + confusion. + root_cause: > + Dispatch-failure repend family (bug-1510 no retry, bug-1511 worker + rejoin slots, bug-1512 multi-worker over-dispatch ack misjudged with + code=0). + fix: > + Fixed in current builds (e4aaef4 and friends); if seen, check build + version and worker slot counts after rejoin. + signals: > + dispatch failures in scheduler log with code=0 acks; slot counts on + workers diverge from controller view. diff --git a/src/frontend/advisor/internal/runbook/runbook_test.go b/src/frontend/advisor/internal/runbook/runbook_test.go new file mode 100644 index 00000000..9193669c --- /dev/null +++ b/src/frontend/advisor/internal/runbook/runbook_test.go @@ -0,0 +1,53 @@ +package runbook + +import ( + "os" + "path/filepath" + "strings" + "testing" +) + +func TestLoadEmbedded(t *testing.T) { + rb, err := Load("") + if err != nil { + t.Fatalf("Load embedded: %v", err) + } + if len(rb.Entries) < 5 { + t.Fatalf("seed runbook too thin: %d entries", len(rb.Entries)) + } + names := map[string]bool{} + for _, e := range rb.Entries { + names[e.Name] = true + if e.Symptoms == "" || e.RootCause == "" || e.Fix == "" { + t.Errorf("entry %q incomplete: %+v", e.Name, e) + } + } + for _, want := range []string{"stale-deployment-config", "ipc-abi-skew-dead-wait", "firstfit-placement-bias"} { + if !names[want] { + t.Errorf("missing entry %q", want) + } + } +} + +func TestLoadOverride(t *testing.T) { + path := filepath.Join(t.TempDir(), "rb.yaml") + _ = os.WriteFile(path, []byte("entries:\n - name: custom\n symptoms: s\n root_cause: r\n fix: f\n"), 0o644) + rb, err := Load(path) + if err != nil { + t.Fatalf("Load override: %v", err) + } + if len(rb.Entries) != 1 || rb.Entries[0].Name != "custom" { + t.Fatalf("override not used: %+v", rb.Entries) + } +} + +func TestRender(t *testing.T) { + rb, _ := Load("") + text := rb.Render() + if !strings.Contains(text, "ipc-abi-skew-dead-wait") { + t.Errorf("render missing entry name") + } + if !strings.Contains(text, "root cause:") { + t.Errorf("render missing field labels") + } +}