From 1691e49a05ddedd2e042265c3f83d748dca5556d Mon Sep 17 00:00:00 2001 From: "lei.chen" Date: Mon, 19 Oct 2026 10:06:32 +0000 Subject: [PATCH] =?UTF-8?q?feat(advisor):=20P1=20outlets=20=E2=80=94=20fin?= =?UTF-8?q?dings=20push=20to=20SSE=20+=20MCP=20scheduler=5Fadvisor=5Ffindi?= =?UTF-8?q?ngs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The advisor's findings previously lived only in its JSONL decision log on its own host. P1 adds two observe-only outlets so diagnoses reach humans and other agents live: apiserver side: - backend/broker.go: EventAdvisorFinding ("advisor.finding") event type + AdvisorFinding struct + Event.AdvisorFindings field. - controllers/events.go: ReportFindings — POST /xce/v1/advisor/findings (generated_at/scenario/findings), publishes one broker event. Broadcast only: the decision log remains the store of record. - router/routers.go: route behind RequirePermission("advisor:report"). - middleware/whitelist.go: admit /xce/v1/advisor/findings in the POST URL whitelist (the URL middleware 403s before the router's RBAC otherwise — same shape as bug-1489). - config/rbac.yaml: advisor_agent role (advisor:report + the six *:view grants the snapshot needs) and advisor:report in the permission whitelist. mcp_readonly stays strictly *:view. New TestRBACYAML_AdvisorAgentRole_PinsReportSurface pins the role surface. advisor side: - internal/push: Pusher — best-effort POST of each round's findings, one retry, nil/unconfigured/empty findings are no-ops; failures log and never affect the observe loop. - internal/agent: SetPusher wired after the decision-log append (the log is authoritative; push is a fan-out). - internal/config + cmd: push default-on to the same XCE_API_URL with the same XCE_API_KEY (the key's role needs advisor:report); ADVISOR_PUSH= off|false|0 disables. mcp-server side: - internal/advisor.go: lenient JSONL reader for the advisor's decision log (ADVISOR_LOG env), tail-bounded, corrupt lines skipped. - scheduler_advisor_findings tool: recent records with parsed findings; friendly errors when the log is missing/unconfigured. Passive reader — never talks to the advisor process. Cluster verification (t1/t2/t3): redeployed apiserver closure, deployed rbac.yaml, issued advisor_agent key. Forced-PEND job → pend_stall fired on the real LLM (glm-5-3-flash, 19.4s, degraded=false) → 2 findings, all evidence verbatim in the snapshot, unverified=false → SSE broadcast captured on /xce/v1/events → MCP tool returned both records. Second PEND job joining the stalled set fired a new decision within one round (change-triggered semantics). Verification jobs killed; sanity job submitted→ran→DONE; all daemons and 3 hosts healthy. Design doc: architecture diagram gains the push branch; roadmap P1 marked done (CLI outlet not built). Co-Authored-By: Claude Code --- src/docs/advisor-design.md | 19 +++- src/frontend/advisor/cmd/main.go | 15 +++ src/frontend/advisor/internal/agent/agent.go | 29 ++++- .../advisor/internal/config/config.go | 12 ++ src/frontend/advisor/internal/push/push.go | 99 +++++++++++++++++ .../advisor/internal/push/push_test.go | 99 +++++++++++++++++ src/frontend/apiserver/backend/broker.go | 26 ++++- src/frontend/apiserver/config/rbac.yaml | 31 ++++++ src/frontend/apiserver/controllers/events.go | 37 ++++++ .../controllers/rbac_mcp_readonly_test.go | 43 +++++++ .../apiserver/middleware/whitelist.go | 1 + .../apiserver/middleware/whitelist_test.go | 5 + src/frontend/apiserver/router/routers.go | 11 ++ src/frontend/mcp-server/cmd/main.go | 8 ++ src/frontend/mcp-server/internal/advisor.go | 86 ++++++++++++++ .../mcp-server/internal/advisor_test.go | 105 ++++++++++++++++++ src/frontend/mcp-server/internal/app.go | 60 +++++++++- 17 files changed, 677 insertions(+), 9 deletions(-) create mode 100644 src/frontend/advisor/internal/push/push.go create mode 100644 src/frontend/advisor/internal/push/push_test.go create mode 100644 src/frontend/mcp-server/internal/advisor.go create mode 100644 src/frontend/mcp-server/internal/advisor_test.go diff --git a/src/docs/advisor-design.md b/src/docs/advisor-design.md index 6d0514b3..a5ffd354 100644 --- a/src/docs/advisor-design.md +++ b/src/docs/advisor-design.md @@ -21,7 +21,7 @@ apiserver REST ──────▶│ snapshot: 紧凑集群状态（字段白名单+top-N 截断） │ (schedulerclient) └──────────────┬───────────────────────────────┘ ▼ - SSE /xce/v1/events ─▶ detect: 确定性触发器（阈值→冷却→去重） + SSE /xce/v1/events ─▶ detect: 确定性触发器（去重；变化即触发） (事件只提前扫描) │ 命中 ▼ provider.Evaluate(Scenario, State, Runbook) @@ -31,13 +31,22 @@ ▼ 幻觉防火墙 VerifyEvidence │ - ▼ - declog: JSONL 决策日志（全上下文） + ┌─────────────┼──────────────────┐ + ▼ ▼ ▼ + declog JSONL push（P1） （未来：CLI 直出） + 决策日志 ─▶ POST /xce/v1/advisor/findings + （存档/审计） └─ apiserver broker ─▶ SSE advisor.finding 事件 ``` - 代码位置：`src/frontend/advisor/`，独立 Go module（`replace schedulerclient`）， `CGO_ENABLED=0`，不进 meson（与 mcp-server 一致，`go build` 直出）。 -- 部署：t1 上以 `XCE_API_KEY`（mcp_readonly 角色）运行 `advisor run`。 +- 部署：t1 上以 `XCE_API_KEY`（advisor_agent 角色）运行 `advisor run`。 +- 出口（P1）：决策日志是存档底账；push 是尽力而为的广播（失败只记日志， + 永不影响观察循环），目标 = 同一 apiserver 的 + `POST /xce/v1/advisor/findings`（`advisor:report` 权限），服务端原样 + 转成 SSE `advisor.finding` 事件分发给订阅者。`ADVISOR_PUSH=off` 关闭。 + MCP 侧 `scheduler_advisor_findings` 工具直接读决策日志 JSONL + （`ADVISOR_LOG`），是被动读者、不经过 advisor 进程。 ## 3. Provider 契约 @@ -108,7 +117,7 @@ tokens / findings / raw / state(+digest) / degraded`。JSONL、按大小轮转 | 阶段 | 内容 | |---|---| -| P1 | propose 档：findings 推送（SSE/MCP 工具/CLI）；runbook 上提 schedulerclient；评估 echo vs LLM 的对照数据 | +| P1 ✅ | propose 档：findings 推送（SSE 广播 + MCP 工具，已实现；CLI 直出未做）；评估 echo vs LLM 的对照数据 | | P2 | act 档（仅当引入带校准置信度的 provider——见附录 A）：allowlist + 爆炸半径上限 + 审计 | | — | 自洽性伪置信度（k 次采样一致率）作为 act 档的非 Jev 替代——5 倍成本，只在关键判断用 | diff --git a/src/frontend/advisor/cmd/main.go b/src/frontend/advisor/cmd/main.go index 267056d6..b10be69c 100644 --- a/src/frontend/advisor/cmd/main.go +++ b/src/frontend/advisor/cmd/main.go @@ -25,6 +25,7 @@ import ( "advisor/internal/declog" "advisor/internal/detect" "advisor/internal/provider" + "advisor/internal/push" "advisor/internal/runbook" sclient "schedulerclient" @@ -66,6 +67,7 @@ configuration (env): ADVISOR_INTERVAL scan interval (default 120s) ADVISOR_LOG decision log path (default /tmp/advisor-decisions.jsonl) ADVISOR_RUNBOOK runbook YAML overriding the embedded one + ADVISOR_PUSH findings push to the apiserver (default on; off/false/0 to disable) `) } @@ -92,6 +94,18 @@ func pickProvider(cfg *config.Config) provider.Provider { return provider.NewOpenAI(cfg.LLMBaseURL, cfg.LLMAPIKey, cfg.LLMModel, cfg.LLMTimeout) } +// maybePusher wires the findings push (P1) when enabled. Push is best-effort +// by design — a 403/timeout never affects the observe loop; the decision log +// remains the store of record. +func maybePusher(cfg *config.Config) *push.Pusher { + if !cfg.PushEnabled { + log.Printf("advisor: findings push disabled (ADVISOR_PUSH=off)") + return nil + } + log.Printf("advisor: findings push -> %s (same credential as read; role needs advisor:report)", cfg.APIURL) + return push.New(cfg.APIURL, cfg.APIKey) +} + func loadRunbook(cfg *config.Config) *runbook.Runbook { rb, err := runbook.Load(cfg.RunbookPath) if err != nil { @@ -112,6 +126,7 @@ func cmdRun() { a := agent.New(mustClient(cfg), detect.NewDetector(detect.DefaultLimits()), pickProvider(cfg), dl, cfg.Interval) a.RunbookText = rb.Render() + a.SetPusher(maybePusher(cfg)) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() diff --git a/src/frontend/advisor/internal/agent/agent.go b/src/frontend/advisor/internal/agent/agent.go index 82ed9dff..561c1b46 100644 --- a/src/frontend/advisor/internal/agent/agent.go +++ b/src/frontend/advisor/internal/agent/agent.go @@ -17,6 +17,7 @@ import ( "advisor/internal/declog" "advisor/internal/detect" "advisor/internal/provider" + "advisor/internal/push" "advisor/internal/snapshot" sclient "schedulerclient" @@ -30,6 +31,9 @@ type Agent struct { primary provider.Provider // may be nil: echo-only mode echo *provider.Echo log *declog.Log + // pusher reports findings to the apiserver's advisor endpoint (P1); + // may be nil (push disabled). + pusher *push.Pusher // RunbookText is the rendered runbook context handed to providers. RunbookText string // ScanInterval mirrors cfg.Interval; SSE events only pull the next scan @@ -41,7 +45,8 @@ type Agent struct { lastExits []snapshot.ExitState } -// New assembles an Agent. primary may be nil (echo-only). +// New assembles an Agent. primary may be nil (echo-only); pusher may be nil +// (findings stay in the decision log only). func New(client sclient.Client, det *detect.Detector, primary provider.Provider, dl *declog.Log, scanInterval time.Duration) *Agent { return &Agent{ client: client, @@ -54,6 +59,9 @@ func New(client sclient.Client, det *detect.Detector, primary provider.Provider, } } +// SetPusher enables reporting each round's findings to the apiserver. +func (a *Agent) SetPusher(p *push.Pusher) { a.pusher = p } + // 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. @@ -212,6 +220,18 @@ func mustJSON(v any) json.RawMessage { return b } +// snapshotTime extracts the snapshot's generated_at from the state JSON for +// the findings push; "unknown" on a malformed state never blocks the report. +func snapshotTime(stateJSON []byte) string { + var st struct { + GeneratedAt string `json:"generated_at"` + } + if err := json.Unmarshal(stateJSON, &st); err != nil || st.GeneratedAt == "" { + return "unknown" + } + return st.GeneratedAt +} + // 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{ @@ -257,6 +277,13 @@ func (a *Agent) evaluateTrigger(ctx context.Context, tr detect.Trigger, stateJSO log.Printf("advisor: decision log append failed: %v", err) } + // P1 propose tier: report the round to the apiserver's SSE fan-out so + // dashboards/MCP clients see diagnoses live. Best-effort; the decision + // log above already holds the record. + if a.pusher != nil { + a.pusher.BestEffort(ctx, tr.Scenario, snapshotTime(stateJSON), resp.Findings) + } + 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) } diff --git a/src/frontend/advisor/internal/config/config.go b/src/frontend/advisor/internal/config/config.go index 55fbd0a5..80c5bb42 100644 --- a/src/frontend/advisor/internal/config/config.go +++ b/src/frontend/advisor/internal/config/config.go @@ -41,6 +41,11 @@ type Config struct { // RunbookPath optionally overrides the embedded runbook // (ADVISOR_RUNBOOK). RunbookPath string + // PushEnabled reports each round's findings to the apiserver's advisor + // endpoint (SSE fan-out, P1). On by default; ADVISOR_PUSH=off disables. + // The push target is APIURL with the same XCE_API_KEY — the key's role + // needs advisor:report. + PushEnabled bool } // Defaults. @@ -55,6 +60,12 @@ const ( // missing required values (XCE_API_KEY). LLM settings being empty is valid — // the advisor degrades to the echo provider. func FromEnv() (*Config, error) { + pushEnabled := true + switch os.Getenv("ADVISOR_PUSH") { + case "off", "false", "0": + pushEnabled = false + } + c := &Config{ APIURL: getenv("XCE_API_URL", defaultAPIURL), APIKey: os.Getenv("XCE_API_KEY"), @@ -65,6 +76,7 @@ func FromEnv() (*Config, error) { RunbookPath: getenv("ADVISOR_RUNBOOK", ""), Interval: defaultInterval, LLMTimeout: defaultLLMTimeout, + PushEnabled: pushEnabled, } if c.APIKey == "" { diff --git a/src/frontend/advisor/internal/push/push.go b/src/frontend/advisor/internal/push/push.go new file mode 100644 index 00000000..36074c53 --- /dev/null +++ b/src/frontend/advisor/internal/push/push.go @@ -0,0 +1,99 @@ +// Package push reports evaluation rounds' findings to the apiserver's +// advisor endpoint, which fans them out to SSE subscribers as +// advisor.finding events (P1 propose tier). Best-effort by design: a push +// failure is logged and otherwise swallowed — the JSONL decision log on +// this host is the store of record, and a monitoring loop must never die +// because its telemetry is unreachable. +package push + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "time" + + "advisor/internal/provider" +) + +// Pusher posts one round's findings to the apiserver. +type Pusher struct { + baseURL string + token string + hc *http.Client +} + +// New returns a Pusher for the apiserver at baseURL (e.g. +// http://t1:8080) authenticated with a token (JWT or xce_ak_ key) whose +// role grants advisor:report. +func New(baseURL, token string) *Pusher { + return &Pusher{ + baseURL: baseURL, + token: token, + hc: &http.Client{Timeout: 10 * time.Second}, + } +} + +// Report uploads one evaluation round. scenario names the trigger's +// scenario; generatedAt is the snapshot time the findings evaluate against; +// degraded marks the round as answered by the echo fallback. +func (p *Pusher) Report(ctx context.Context, scenario, generatedAt string, findings []provider.Finding) error { + if p == nil || p.baseURL == "" { + return nil // push not configured: a no-op, not an error + } + if len(findings) == 0 { + return nil + } + body, err := json.Marshal(map[string]any{ + "generated_at": generatedAt, + "scenario": scenario, + "findings": findings, + }) + if err != nil { + return fmt.Errorf("marshal findings: %w", err) + } + + // One retry: the apiserver may be restarting (or momentarily slow). A + // second failure is final for this round — the decision log still has + // the record, and the next diagnosis re-reports. + var lastErr error + for attempt := 0; attempt < 2; attempt++ { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + p.baseURL+"/xce/v1/advisor/findings", bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("build request: %w", err) + } + req.Header.Set("Authorization", p.token) + req.Header.Set("Content-Type", "application/json") + + resp, err := p.hc.Do(req) + if err != nil { + lastErr = err + } else if resp.StatusCode == http.StatusOK { + resp.Body.Close() + return nil + } else { + lastErr = fmt.Errorf("HTTP %d", resp.StatusCode) + resp.Body.Close() + } + if attempt == 0 { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(time.Second): + } + } + } + return fmt.Errorf("report findings: %w", lastErr) +} + +// BestEffort wraps Report, logging failures instead of returning them. The +// agent loop calls this — an unreportable finding is a degraded monitoring +// setup, never a reason to stop observing. +func (p *Pusher) BestEffort(ctx context.Context, scenario, generatedAt string, findings []provider.Finding) { + if err := p.Report(ctx, scenario, generatedAt, findings); err != nil { + log.Printf("advisor: findings push failed (non-fatal): %v", err) + } +} diff --git a/src/frontend/advisor/internal/push/push_test.go b/src/frontend/advisor/internal/push/push_test.go new file mode 100644 index 00000000..3cf9aaca --- /dev/null +++ b/src/frontend/advisor/internal/push/push_test.go @@ -0,0 +1,99 @@ +package push + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + + "advisor/internal/provider" +) + +func sampleFindings() []provider.Finding { + return []provider.Finding{ + {Kind: "diagnosis", Severity: provider.SevDegraded, Summary: "1 job stalled", + Evidence: []string{"73937053183639552", "unknown(0)"}, Confidence: 1.0}, + } +} + +func TestReport_PostsShapeAndAuth(t *testing.T) { + var gotAuth, gotPath string + var body map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + gotPath = r.URL.Path + _ = json.NewDecoder(r.Body).Decode(&body) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"code":200,"msg":"Success","data":{"published":1}}`)) + })) + defer srv.Close() + + p := New(srv.URL, "xce_ak_test") + if err := p.Report(context.Background(), "pend_stall", "2026-10-19T08:00:00Z", sampleFindings()); err != nil { + t.Fatalf("Report: %v", err) + } + if gotPath != "/xce/v1/advisor/findings" { + t.Errorf("path = %q", gotPath) + } + if gotAuth != "xce_ak_test" { + t.Errorf("auth = %q", gotAuth) + } + if body["scenario"] != "pend_stall" { + t.Errorf("scenario = %v", body["scenario"]) + } + fs, ok := body["findings"].([]any) + if !ok || len(fs) != 1 { + t.Fatalf("findings = %v", body["findings"]) + } +} + +func TestReport_RetriesOnceThenSucceeds(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if calls.Add(1) == 1 { + w.WriteHeader(http.StatusBadGateway) + return + } + _, _ = w.Write([]byte(`{"code":200,"msg":"Success","data":{}}`)) + })) + defer srv.Close() + + p := New(srv.URL, "tok") + if err := p.Report(context.Background(), "pend_stall", "now", sampleFindings()); err != nil { + t.Fatalf("retry after 502 should succeed: %v", err) + } + if got := calls.Load(); got != 2 { + t.Fatalf("calls = %d, want 2", got) + } +} + +func TestReport_GivesUpAfterTwoFailures(t *testing.T) { + var calls atomic.Int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.WriteHeader(http.StatusServiceUnavailable) + })) + defer srv.Close() + + p := New(srv.URL, "tok") + if err := p.Report(context.Background(), "pend_stall", "now", sampleFindings()); err == nil { + t.Fatal("expected error after both attempts failed") + } + if got := calls.Load(); got != 2 { + t.Fatalf("calls = %d, want exactly 2 (one retry, no loop)", got) + } +} + +func TestReport_EmptyFindingsAndUnconfiguredAreNoOps(t *testing.T) { + // nil pusher and empty findings must not touch the network. + p := New("", "tok") + if err := p.Report(context.Background(), "s", "now", nil); err != nil { + t.Fatalf("empty findings: %v", err) + } + var q *Pusher + if err := q.Report(context.Background(), "s", "now", sampleFindings()); err != nil { + t.Fatalf("nil pusher: %v", err) + } +} diff --git a/src/frontend/apiserver/backend/broker.go b/src/frontend/apiserver/backend/broker.go index f77cbbaf..03a010b7 100644 --- a/src/frontend/apiserver/backend/broker.go +++ b/src/frontend/apiserver/backend/broker.go @@ -12,16 +12,38 @@ const ( EventJobUpdated = "job.updated" EventJobDeleted = "job.deleted" EventJobFinished = "job.finished" + // EventAdvisorFinding is one scheduler-advisor diagnosis (P1 propose + // tier). Carried in AdvisorFindings; the advisor's JSONL decision log on + // its own host remains the store of record — the broker only fans the + // finding out live. + EventAdvisorFinding = "advisor.finding" ) -// Event is a single lifecycle event. Exactly one of Job / (JobID, ExitCode) -// is populated depending on Type. +// AdvisorFinding is the advisor-facing shape of one diagnosis. Deliberately +// decoupled from the advisor module (separate go modules); the advisor's +// provider.Finding marshals to exactly this shape (kind/severity/summary/ +// evidence/confidence/suggestion/unverified). +type AdvisorFinding struct { + Kind string `json:"kind"` + Severity string `json:"severity"` + Summary string `json:"summary"` + Evidence []string `json:"evidence,omitempty"` + Confidence float64 `json:"confidence,omitempty"` + Suggestion string `json:"suggestion,omitempty"` + Unverified bool `json:"unverified,omitempty"` +} + +// Event is a single lifecycle event. Exactly one of Job / (JobID, ExitCode) / +// AdvisorFindings is populated depending on Type. type Event struct { Type string `json:"type"` JobID uint64 `json:"job_id,omitempty"` ExitCode int `json:"exit_code,omitempty"` Status string `json:"job_status,omitempty"` // JobStateStr of the job (PEND/RUN/DONE/EXIT/...) Job *models.JobInfo `json:"job,omitempty"` + // AdvisorFindings carries the diagnoses of one advisor evaluation round + // (Type EventAdvisorFinding). + AdvisorFindings []AdvisorFinding `json:"advisor_findings,omitempty"` } // Broker is a thread-safe fan-out of lifecycle events to subscribed channels. diff --git a/src/frontend/apiserver/config/rbac.yaml b/src/frontend/apiserver/config/rbac.yaml index 59ad0fc7..07ba3a3c 100644 --- a/src/frontend/apiserver/config/rbac.yaml +++ b/src/frontend/apiserver/config/rbac.yaml @@ -18,6 +18,7 @@ permissions: - usergroup:view # view usergroup info - resource:view # view resource info - event:view # subscribe to the live job-event SSE stream (scheduler-agent Phase 2) + - advisor:report # advisor service account reports findings (broadcast-only; scheduler-advisor P1) - ar:view # view advanced reservation info - ar:submit # submit advanced reservation - ar:control # delete, modify, create @@ -211,3 +212,33 @@ roles: - action: event:view domain: - type: all + + # advisor_agent: service-account role for the observe-only scheduler-advisor + # (P1 propose tier). Cluster visibility equal to mcp_readonly (the snapshot + # needs job/queue/host/event views) plus ONE write action, advisor:report — + # POST /xce/v1/advisor/findings, which only broadcasts diagnoses to SSE + # subscribers. No job:control/queue:control/cluster:control: the advisor is + # architecturally incapable of mutating cluster state. + advisor_agent: + permissions: + - action: advisor:report + domain: + - type: all + - action: job:view + domain: + - type: all + - action: queue:view + domain: + - type: all + - action: host:view + domain: + - type: all + - action: resource:view + domain: + - type: all + - action: cluster:view + domain: + - type: all + - action: event:view + domain: + - type: all diff --git a/src/frontend/apiserver/controllers/events.go b/src/frontend/apiserver/controllers/events.go index 46283289..c5eac59f 100644 --- a/src/frontend/apiserver/controllers/events.go +++ b/src/frontend/apiserver/controllers/events.go @@ -8,6 +8,7 @@ import ( "github.com/gin-gonic/gin" "apiserver/backend" + "apiserver/common" logger "apiserver/utils" "go.uber.org/zap" @@ -106,3 +107,39 @@ func (e *EventsController) SubscribeJobs(ctx *gin.Context) { } } } + +// ReportFindings accepts one advisor evaluation round's diagnoses and fans +// them out to SSE subscribers as advisor.finding events (P1 propose tier). +// The advisor's JSONL decision log on its own host remains the store of +// record; this endpoint only broadcasts. Auth: advisor:report (granted to +// the advisor_agent service role; see rbac.yaml). +func (e *EventsController) ReportFindings(ctx *gin.Context) { + var req struct { + // GeneratedAt is the snapshot time the findings were evaluated + // against (RFC3339); echoed to subscribers unchanged. + GeneratedAt string `json:"generated_at"` + Scenario string `json:"scenario"` + Findings []backend.AdvisorFinding `json:"findings"` + } + if err := ctx.ShouldBindJSON(&req); err != nil { + ReturnError(ctx, common.INVALID_PARAMS_CODE, "Invalid request format: "+err.Error(), nil) + return + } + if len(req.Findings) == 0 { + ReturnError(ctx, common.INVALID_PARAMS_CODE, "findings must be non-empty", nil) + return + } + if req.Scenario == "" { + ReturnError(ctx, common.INVALID_PARAMS_CODE, "scenario is required", nil) + return + } + + backend.DefaultBroker.Publish(backend.Event{ + Type: backend.EventAdvisorFinding, + Status: req.Scenario, + AdvisorFindings: req.Findings, + }) + ReturnSuccess(ctx, common.SUCCESS_CODE, common.SUCCESS_MSG, map[string]interface{}{ + "published": len(req.Findings), + }) +} diff --git a/src/frontend/apiserver/controllers/rbac_mcp_readonly_test.go b/src/frontend/apiserver/controllers/rbac_mcp_readonly_test.go index 18074acf..d62773cb 100644 --- a/src/frontend/apiserver/controllers/rbac_mcp_readonly_test.go +++ b/src/frontend/apiserver/controllers/rbac_mcp_readonly_test.go @@ -50,3 +50,46 @@ func hasViewSuffix(action string) bool { } return action[len(action)-len(suffix):] == suffix } + +// TestRBACYAML_AdvisorAgentRole_PinsReportSurface verifies the advisor_agent +// role (P1 propose tier): it exists, grants advisor:report (its ONE write +// action — POST /xce/v1/advisor/findings), and everything else it grants is +// a plain *:view. The advisor must never hold a control/submit permission. +func TestRBACYAML_AdvisorAgentRole_PinsReportSurface(t *testing.T) { + resetRBACState() + + path, err := filepath.Abs("../config/rbac.yaml") + if err != nil { + t.Fatalf("resolve rbac.yaml path: %v", err) + } + if _, err := os.Stat(path); err != nil { + t.Skipf("config/rbac.yaml not found at %s: %v", path, err) + } + if err := LoadRBACConfig(path); err != nil { + t.Fatalf("LoadRBACConfig: %v", err) + } + + if !store.Roles.Has("advisor_agent") { + t.Fatal("expected advisor_agent role to be loaded from config/rbac.yaml") + } + + perms := store.GetRolePermissions("advisor_agent") + if len(perms) == 0 { + t.Fatal("advisor_agent should have at least one permission") + } + + reports := 0 + for _, p := range perms { + switch { + case p.Action == "advisor:report": + reports++ + case hasViewSuffix(p.Action): + // fine — read-only visibility + default: + t.Errorf("advisor_agent grants forbidden permission %q — the advisor is observe-only; only advisor:report and *:view are allowed", p.Action) + } + } + if reports != 1 { + t.Errorf("advisor_agent should grant exactly one advisor:report action, got %d", reports) + } +} diff --git a/src/frontend/apiserver/middleware/whitelist.go b/src/frontend/apiserver/middleware/whitelist.go index dde86090..adfd9566 100644 --- a/src/frontend/apiserver/middleware/whitelist.go +++ b/src/frontend/apiserver/middleware/whitelist.go @@ -48,6 +48,7 @@ var WhitelistPOSTURL = []string{ "/xce/v1/roles/create", // create role "/xce/v1/roles/*", // add permissions for role "/xce/v1/users/*", // add roles for user + "/xce/v1/advisor/findings", // advisor reports findings (scheduler-advisor P1; RBAC advisor:report applies downstream) } var WhitelistDELETEURL = []string{ diff --git a/src/frontend/apiserver/middleware/whitelist_test.go b/src/frontend/apiserver/middleware/whitelist_test.go index 35cf2ea1..65ba3bea 100644 --- a/src/frontend/apiserver/middleware/whitelist_test.go +++ b/src/frontend/apiserver/middleware/whitelist_test.go @@ -63,6 +63,11 @@ func TestPathMatchesPattern(t *testing.T) { {"/xce/v1/roles/*", "/xce/v1/roles/cluster_admin", true}, {"/xce/v1/users/*", "/xce/v1/users/current", true}, {"/xce/v1/users/*", "/xce/v1/users/admin/basic_user", true}, + // advisor findings (scheduler-advisor P1): the POST whitelist must + // admit the advisor's report path or the URL middleware 403s before + // the router (and its advisor:report check) is ever reached. + {"/xce/v1/advisor/findings", "/xce/v1/advisor/findings", true}, + {"/xce/v1/advisor/findings", "/xce/v1/advisor/findings/extra", false}, // exact patterns {"/xce/v1/hosts", "/xce/v1/hosts", true}, {"/xce/v1/hosts", "/xce/v1/host", false}, diff --git a/src/frontend/apiserver/router/routers.go b/src/frontend/apiserver/router/routers.go index 57185056..c76c049c 100644 --- a/src/frontend/apiserver/router/routers.go +++ b/src/frontend/apiserver/router/routers.go @@ -167,6 +167,17 @@ func Router() *gin.Engine { eventsApi.GET("", controllers.RequirePermission("event:view"), eventsHandler.SubscribeJobs) } + // ################ case: advisor route (P1 propose tier) + // The scheduler-advisor posts its evaluation round's findings here; the + // apiserver fans them out to SSE subscribers as advisor.finding events. + // Write-scoped to advisor:report (mcp_readonly) — the advisor's key can + // only report diagnoses, not read or mutate anything else. + advisorApi := r.Group("/xce/v1/advisor") + { + eventsHandler := &controllers.EventsController{} + advisorApi.POST("/findings", controllers.RequirePermission("advisor:report"), eventsHandler.ReportFindings) + } + // ################ case: resources route resApi := r.Group("/xce/v1/resources") { diff --git a/src/frontend/mcp-server/cmd/main.go b/src/frontend/mcp-server/cmd/main.go index 1637f089..768cbb07 100644 --- a/src/frontend/mcp-server/cmd/main.go +++ b/src/frontend/mcp-server/cmd/main.go @@ -170,6 +170,14 @@ func main() { mcp.WithDestructiveHintAnnotation(false), ), app.HandlePendReasons) + // ── scheduler_advisor_findings (observe-mode advisor outlet) ── + s.AddTool(mcp.NewTool("scheduler_advisor_findings", + mcp.WithDescription("Recent findings from the XCE scheduler-advisor — an observe-only LLM ops copilot that periodically diagnoses cluster state (pend stalls, worker health). Read from the advisor's decision log; each record has ts, scenario, provider, and structured findings (kind/severity/summary/evidence/confidence/suggestion). Findings are advisory for a human — never executed. Requires the ADVISOR_LOG env var pointing at the advisor's decision log."), + mcp.WithNumber("tail", mcp.Description("Max recent decision records to return (default 10, max 100)")), + mcp.WithReadOnlyHintAnnotation(true), + mcp.WithDestructiveHintAnnotation(false), + ), app.HandleAdvisorFindings) + // ── write tools (Phase 1) ── // Write tools are annotated destructive so the MCP client (e.g. Claude // Code) prompts the user for confirmation before executing them — the diff --git a/src/frontend/mcp-server/internal/advisor.go b/src/frontend/mcp-server/internal/advisor.go new file mode 100644 index 00000000..80c97474 --- /dev/null +++ b/src/frontend/mcp-server/internal/advisor.go @@ -0,0 +1,86 @@ +// advisor.go reads the scheduler-advisor's JSONL decision log for the +// scheduler_advisor_findings tool. The log is the advisor's store of record +// (see src/frontend/advisor/internal/declog); the MCP tool is a passive +// reader — it never talks to the advisor process. +package internal + +import ( + "bufio" + "encoding/json" + "fmt" + "os" + "strings" +) + +// AdvisorRecord mirrors one declog.Record line. Parsed leniently: unknown +// fields are ignored, missing fields render as empty. +type AdvisorRecord struct { + TS string `json:"ts"` + Scenario string `json:"scenario"` + Trigger string `json:"trigger"` + Provider string `json:"provider"` + Model string `json:"model"` + LatencyMS int64 `json:"latency_ms"` + Findings json.RawMessage `json:"findings"` + Degraded bool `json:"degraded,omitempty"` + Error string `json:"error,omitempty"` +} + +// AdvisorFinding mirrors one provider.Finding inside the findings array. +type AdvisorFinding struct { + Kind string `json:"kind"` + Severity string `json:"severity"` + Summary string `json:"summary"` + Evidence []string `json:"evidence"` + Confidence float64 `json:"confidence"` + Suggestion string `json:"suggestion,omitempty"` + Unverified bool `json:"unverified,omitempty"` +} + +// AdvisorFindingsOptions bounds the read. +type AdvisorFindingsOptions struct { + // Path is the decision log path (ADVISOR_LOG). + Path string + // Tail is the max number of records returned (most recent last). + Tail int +} + +// ReadAdvisorFindings reads up to Tail records from the advisor decision +// log, returning them in file order (oldest first, newest last). It returns +// a missing-file error sentinel the handler translates into a friendly +// message, and skips corrupt lines rather than failing the whole read. +func ReadAdvisorFindings(opts AdvisorFindingsOptions) ([]AdvisorRecord, error) { + if opts.Path == "" { + return nil, fmt.Errorf("advisor log path not configured (set ADVISOR_LOG for the mcp-server)") + } + f, err := os.Open(opts.Path) + if err != nil { + if os.IsNotExist(err) { + return nil, fmt.Errorf("advisor decision log not found at %s — the advisor may not be running (ADVISOR_LOG)", opts.Path) + } + return nil, fmt.Errorf("open advisor log: %w", err) + } + defer f.Close() + + var recs []AdvisorRecord + sc := bufio.NewScanner(f) + sc.Buffer(make([]byte, 0, 64*1024), 8<<20) // decision records embed the full state snapshot; can be large + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + var r AdvisorRecord + if err := json.Unmarshal([]byte(line), &r); err != nil { + continue // corrupt line — skip, keep reading + } + recs = append(recs, r) + } + if err := sc.Err(); err != nil { + return nil, fmt.Errorf("read advisor log: %w", err) + } + if opts.Tail > 0 && len(recs) > opts.Tail { + recs = recs[len(recs)-opts.Tail:] + } + return recs, nil +} diff --git a/src/frontend/mcp-server/internal/advisor_test.go b/src/frontend/mcp-server/internal/advisor_test.go new file mode 100644 index 00000000..3d65e348 --- /dev/null +++ b/src/frontend/mcp-server/internal/advisor_test.go @@ -0,0 +1,105 @@ +package internal + +import ( + "os" + "path/filepath" + "strings" + "testing" + + "github.com/mark3labs/mcp-go/mcp" +) + +// textOf extracts the text of the first content element. +func textOf(res *mcp.CallToolResult) string { + if res == nil || len(res.Content) == 0 { + return "" + } + tc, ok := res.Content[0].(mcp.TextContent) + if !ok { + return "" + } + return tc.Text +} + +// writeAdvisorLog writes a small JSONL decision log and returns its path. +func writeAdvisorLog(t *testing.T, lines ...string) string { + t.Helper() + p := filepath.Join(t.TempDir(), "advisor-decisions.jsonl") + if err := os.WriteFile(p, []byte(strings.Join(lines, "\n")+"\n"), 0o644); err != nil { + t.Fatal(err) + } + return p +} + +func TestReadAdvisorFindings_ParsesTailSkipsCorrupt(t *testing.T) { + p := writeAdvisorLog(t, + `{"ts":"2026-01-01T00:00:00Z","scenario":"pend_stall","trigger":"fp-1","provider":"echo","latency_ms":1,"findings":[{"kind":"anomaly","severity":"severe","summary":"2 pending job(s) stalled","evidence":["job 101","no_resource"],"confidence":0.9,"suggestion":"check queue limits"}]}`, + `{corrupt line`, + `{"ts":"2026-01-01T00:01:00Z","scenario":"worker_health","trigger":"fp-2","provider":"openai","model":"glm-5-3-flash","latency_ms":24000,"findings":[{"kind":"diagnosis","severity":"degraded","summary":"worker w2 unreachable","evidence":["w2"],"confidence":0.7,"unverified":true}]}`, + ) + recs, err := ReadAdvisorFindings(AdvisorFindingsOptions{Path: p, Tail: 10}) + if err != nil { + t.Fatalf("read: %v", err) + } + if len(recs) != 2 { + t.Fatalf("want 2 records (corrupt line skipped), got %d", len(recs)) + } + if recs[0].Scenario != "pend_stall" || recs[1].Scenario != "worker_health" { + t.Fatalf("scenario order wrong: %s, %s", recs[0].Scenario, recs[1].Scenario) + } + if recs[0].Provider != "echo" || recs[1].Model != "glm-5-3-flash" { + t.Fatalf("provider/model parse wrong: %+v", recs) + } + if recs[0].LatencyMS != 1 || recs[1].LatencyMS != 24000 { + t.Fatalf("latency parse wrong: %+v", recs) + } + + // Tail bounds the result to the most recent records. + recs, err = ReadAdvisorFindings(AdvisorFindingsOptions{Path: p, Tail: 1}) + if err != nil { + t.Fatalf("read tail=1: %v", err) + } + if len(recs) != 1 || recs[0].Scenario != "worker_health" { + t.Fatalf("tail=1 want only newest worker_health, got %+v", recs) + } +} + +func TestReadAdvisorFindings_MissingLog(t *testing.T) { + _, err := ReadAdvisorFindings(AdvisorFindingsOptions{Path: "/nonexistent/log.jsonl"}) + if err == nil || !strings.Contains(err.Error(), "not found") { + t.Fatalf("want friendly not-found error, got %v", err) + } + _, err = ReadAdvisorFindings(AdvisorFindingsOptions{Path: ""}) + if err == nil || !strings.Contains(err.Error(), "not configured") { + t.Fatalf("want not-configured error, got %v", err) + } +} + +func TestHandleAdvisorFindings_EmptyLogMeansQuietCluster(t *testing.T) { + p := writeAdvisorLog(t) // exists, zero records + app := &App{advisorLogPath: p} + res, err := app.HandleAdvisorFindings(t.Context(), mcp.CallToolRequest{}) + if err != nil { + t.Fatalf("handler: %v", err) + } + if res == nil || len(res.Content) == 0 { + t.Fatal("want text content") + } + if txt := textOf(res); !strings.Contains(txt, "no evaluation records") { + t.Fatalf("want quiet-cluster message, got %q", txt) + } +} + +func TestHandleAdvisorFindings_LogMissingReportsFriendlyError(t *testing.T) { + app := &App{advisorLogPath: "/nonexistent/log.jsonl"} + res, err := app.HandleAdvisorFindings(t.Context(), mcp.CallToolRequest{}) + if err != nil { + t.Fatalf("handler: %v", err) + } + if res == nil || !res.IsError { + t.Fatal("want tool error result") + } + if txt := textOf(res); !strings.Contains(txt, "not found") { + t.Fatalf("want not-found message, got %q", txt) + } +} diff --git a/src/frontend/mcp-server/internal/app.go b/src/frontend/mcp-server/internal/app.go index 907bcb88..867e6dc8 100644 --- a/src/frontend/mcp-server/internal/app.go +++ b/src/frontend/mcp-server/internal/app.go @@ -5,6 +5,7 @@ import ( "context" "encoding/json" "fmt" + "os" "strconv" "time" @@ -16,12 +17,15 @@ import ( type App struct { client sclient.Client templates *TemplateRegistry + // advisorLogPath is the scheduler-advisor's decision log (ADVISOR_LOG, + // empty disables the scheduler_advisor_findings tool's data source). + advisorLogPath string } // NewApp creates a new App with a schedulerclient connected to the given base URL. func NewApp(baseURL, token string) (*App, error) { hc := sclient.NewClient(baseURL, token) - return &App{client: hc, templates: NewTemplateRegistry()}, nil + return &App{client: hc, templates: NewTemplateRegistry(), advisorLogPath: os.Getenv("ADVISOR_LOG")}, nil } // ── helpers ───────────────────────────────────────────────────────────────── @@ -552,3 +556,57 @@ func (a *App) HandleEstimateWaitTime(ctx context.Context, req mcp.CallToolReques "estimate_basis": "Rough heuristic from current pending count and free cores; not a guarantee.", }) } + +// HandleAdvisorFindings serves the scheduler_advisor_findings tool: the +// scheduler-advisor's recent findings, read from its JSONL decision log +// (ADVISOR_LOG). Read-only and passive — the log is the store of record. +func (a *App) HandleAdvisorFindings(ctx context.Context, req mcp.CallToolRequest) (*mcp.CallToolResult, error) { + tail := int(getNumber(req, "tail")) + if tail <= 0 { + tail = 10 + } + if tail > 100 { + tail = 100 + } + recs, err := ReadAdvisorFindings(AdvisorFindingsOptions{Path: a.advisorLogPath, Tail: tail}) + if err != nil { + return mcp.NewToolResultError(err.Error()), nil + } + if len(recs) == 0 { + return textResultStr("The advisor decision log exists but has no evaluation records yet — the advisor has not fired any detectors since it started (a healthy cluster produces no decisions)."), nil + } + + // Compact projection: one summary block per record; the raw decision + // log stays on disk for full detail. + type recordOut struct { + TS string `json:"ts"` + Scenario string `json:"scenario"` + Provider string `json:"provider"` + Model string `json:"model,omitempty"` + LatencyMS int64 `json:"latency_ms,omitempty"` + Degraded bool `json:"degraded,omitempty"` + Error string `json:"error,omitempty"` + Findings []AdvisorFinding `json:"findings"` + } + out := make([]recordOut, 0, len(recs)) + for _, r := range recs { + o := recordOut{ + TS: r.TS, + Scenario: r.Scenario, + Provider: r.Provider, + Model: r.Model, + LatencyMS: r.LatencyMS, + Degraded: r.Degraded, + Error: r.Error, + } + if len(r.Findings) > 0 { + _ = json.Unmarshal(r.Findings, &o.Findings) // findings are advisory text; a malformed array renders as null + } + out = append(out, o) + } + return textResult(map[string]any{ + "log_path": a.advisorLogPath, + "records": out, + "note": "Findings are observe-only diagnostics from the scheduler-advisor; suggestions are advisory for a human and never executed. unverified=true means the finding cited evidence not present in the cluster state (downgrade trust accordingly).", + }) +}