feat(advisor): agent loop, decision log, runbook, run/once CLI

- 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 <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-10-19 03:42:01 +00:00
parent fbc81d61d1
commit ab12f49256
11 changed files with 1213 additions and 1 deletions

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -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 <path>.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()
}

View file

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

View file

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

View file

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

View file

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