feat(advisor): P1 outlets — findings push to SSE + MCP scheduler_advisor_findings

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 <noreply@anthropic.com>
This commit is contained in:
lei.chen 2026-10-19 10:06:32 +00:00
parent 3e3b6d7947
commit 1691e49a05
17 changed files with 677 additions and 9 deletions

View file

@ -21,7 +21,7 @@
apiserver REST ──────▶│ snapshot: 紧凑集群状态(字段白名单+top-N 截断) │ apiserver REST ──────▶│ snapshot: 紧凑集群状态(字段白名单+top-N 截断) │
(schedulerclient) └──────────────┬───────────────────────────────┘ (schedulerclient) └──────────────┬───────────────────────────────┘
▼ ▼
SSE /xce/v1/events ─▶ detect: 确定性触发器(阈值→冷却→去重) SSE /xce/v1/events ─▶ detect: 确定性触发器(去重;变化即触发)
(事件只提前扫描) │ 命中 (事件只提前扫描) │ 命中
▼ ▼
provider.Evaluate(Scenario, State, Runbook) provider.Evaluate(Scenario, State, Runbook)
@ -31,13 +31,22 @@
▼ ▼
幻觉防火墙 VerifyEvidence 幻觉防火墙 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`), - 代码位置:`src/frontend/advisor/`,独立 Go module(`replace schedulerclient`),
`CGO_ENABLED=0`,不进 meson(与 mcp-server 一致,`go build` 直出)。 `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 契约 ## 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 + 爆炸半径上限 + 审计 | | P2 | act 档(仅当引入带校准置信度的 provider——见附录 A):allowlist + 爆炸半径上限 + 审计 |
| — | 自洽性伪置信度(k 次采样一致率)作为 act 档的非 Jev 替代——5 倍成本,只在关键判断用 | | — | 自洽性伪置信度(k 次采样一致率)作为 act 档的非 Jev 替代——5 倍成本,只在关键判断用 |

View file

@ -25,6 +25,7 @@ import (
"advisor/internal/declog" "advisor/internal/declog"
"advisor/internal/detect" "advisor/internal/detect"
"advisor/internal/provider" "advisor/internal/provider"
"advisor/internal/push"
"advisor/internal/runbook" "advisor/internal/runbook"
sclient "schedulerclient" sclient "schedulerclient"
@ -66,6 +67,7 @@ configuration (env):
ADVISOR_INTERVAL scan interval (default 120s) ADVISOR_INTERVAL scan interval (default 120s)
ADVISOR_LOG decision log path (default /tmp/advisor-decisions.jsonl) ADVISOR_LOG decision log path (default /tmp/advisor-decisions.jsonl)
ADVISOR_RUNBOOK runbook YAML overriding the embedded one 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) 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 { func loadRunbook(cfg *config.Config) *runbook.Runbook {
rb, err := runbook.Load(cfg.RunbookPath) rb, err := runbook.Load(cfg.RunbookPath)
if err != nil { if err != nil {
@ -112,6 +126,7 @@ func cmdRun() {
a := agent.New(mustClient(cfg), detect.NewDetector(detect.DefaultLimits()), a := agent.New(mustClient(cfg), detect.NewDetector(detect.DefaultLimits()),
pickProvider(cfg), dl, cfg.Interval) pickProvider(cfg), dl, cfg.Interval)
a.RunbookText = rb.Render() a.RunbookText = rb.Render()
a.SetPusher(maybePusher(cfg))
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop() defer stop()

View file

@ -17,6 +17,7 @@ import (
"advisor/internal/declog" "advisor/internal/declog"
"advisor/internal/detect" "advisor/internal/detect"
"advisor/internal/provider" "advisor/internal/provider"
"advisor/internal/push"
"advisor/internal/snapshot" "advisor/internal/snapshot"
sclient "schedulerclient" sclient "schedulerclient"
@ -30,6 +31,9 @@ type Agent struct {
primary provider.Provider // may be nil: echo-only mode primary provider.Provider // may be nil: echo-only mode
echo *provider.Echo echo *provider.Echo
log *declog.Log 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 is the rendered runbook context handed to providers.
RunbookText string RunbookText string
// ScanInterval mirrors cfg.Interval; SSE events only pull the next scan // ScanInterval mirrors cfg.Interval; SSE events only pull the next scan
@ -41,7 +45,8 @@ type Agent struct {
lastExits []snapshot.ExitState 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 { func New(client sclient.Client, det *detect.Detector, primary provider.Provider, dl *declog.Log, scanInterval time.Duration) *Agent {
return &Agent{ return &Agent{
client: client, 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. // Run blocks until ctx is cancelled: ticker scans + SSE event pull.
// Event types only accelerate the next scan (debounced); detection and // Event types only accelerate the next scan (debounced); detection and
// evaluation always run on the same periodic path. // evaluation always run on the same periodic path.
@ -212,6 +220,18 @@ func mustJSON(v any) json.RawMessage {
return b 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. // evaluateTrigger runs the provider chain for one trigger and logs it.
func (a *Agent) evaluateTrigger(ctx context.Context, tr detect.Trigger, stateJSON []byte) { func (a *Agent) evaluateTrigger(ctx context.Context, tr detect.Trigger, stateJSON []byte) {
req := provider.EvaluateRequest{ 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) 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)", 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) tr.Scenario, tr.Fingerprint, len(resp.Findings), resp.Model, resp.Latency.Round(time.Millisecond), degraded)
} }

View file

@ -41,6 +41,11 @@ type Config struct {
// RunbookPath optionally overrides the embedded runbook // RunbookPath optionally overrides the embedded runbook
// (ADVISOR_RUNBOOK). // (ADVISOR_RUNBOOK).
RunbookPath string 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. // Defaults.
@ -55,6 +60,12 @@ const (
// missing required values (XCE_API_KEY). LLM settings being empty is valid — // missing required values (XCE_API_KEY). LLM settings being empty is valid —
// the advisor degrades to the echo provider. // the advisor degrades to the echo provider.
func FromEnv() (*Config, error) { func FromEnv() (*Config, error) {
pushEnabled := true
switch os.Getenv("ADVISOR_PUSH") {
case "off", "false", "0":
pushEnabled = false
}
c := &Config{ c := &Config{
APIURL: getenv("XCE_API_URL", defaultAPIURL), APIURL: getenv("XCE_API_URL", defaultAPIURL),
APIKey: os.Getenv("XCE_API_KEY"), APIKey: os.Getenv("XCE_API_KEY"),
@ -65,6 +76,7 @@ func FromEnv() (*Config, error) {
RunbookPath: getenv("ADVISOR_RUNBOOK", ""), RunbookPath: getenv("ADVISOR_RUNBOOK", ""),
Interval: defaultInterval, Interval: defaultInterval,
LLMTimeout: defaultLLMTimeout, LLMTimeout: defaultLLMTimeout,
PushEnabled: pushEnabled,
} }
if c.APIKey == "" { if c.APIKey == "" {

View file

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

View file

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

View file

@ -12,16 +12,38 @@ const (
EventJobUpdated = "job.updated" EventJobUpdated = "job.updated"
EventJobDeleted = "job.deleted" EventJobDeleted = "job.deleted"
EventJobFinished = "job.finished" 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) // AdvisorFinding is the advisor-facing shape of one diagnosis. Deliberately
// is populated depending on Type. // 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 Event struct {
Type string `json:"type"` Type string `json:"type"`
JobID uint64 `json:"job_id,omitempty"` JobID uint64 `json:"job_id,omitempty"`
ExitCode int `json:"exit_code,omitempty"` ExitCode int `json:"exit_code,omitempty"`
Status string `json:"job_status,omitempty"` // JobStateStr of the job (PEND/RUN/DONE/EXIT/...) Status string `json:"job_status,omitempty"` // JobStateStr of the job (PEND/RUN/DONE/EXIT/...)
Job *models.JobInfo `json:"job,omitempty"` 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. // Broker is a thread-safe fan-out of lifecycle events to subscribed channels.

View file

@ -18,6 +18,7 @@ permissions:
- usergroup:view # view usergroup info - usergroup:view # view usergroup info
- resource:view # view resource info - resource:view # view resource info
- event:view # subscribe to the live job-event SSE stream (scheduler-agent Phase 2) - 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:view # view advanced reservation info
- ar:submit # submit advanced reservation - ar:submit # submit advanced reservation
- ar:control # delete, modify, create - ar:control # delete, modify, create
@ -211,3 +212,33 @@ roles:
- action: event:view - action: event:view
domain: domain:
- type: all - 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

View file

@ -8,6 +8,7 @@ import (
"github.com/gin-gonic/gin" "github.com/gin-gonic/gin"
"apiserver/backend" "apiserver/backend"
"apiserver/common"
logger "apiserver/utils" logger "apiserver/utils"
"go.uber.org/zap" "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),
})
}

View file

@ -50,3 +50,46 @@ func hasViewSuffix(action string) bool {
} }
return action[len(action)-len(suffix):] == suffix 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)
}
}

View file

@ -48,6 +48,7 @@ var WhitelistPOSTURL = []string{
"/xce/v1/roles/create", // create role "/xce/v1/roles/create", // create role
"/xce/v1/roles/*", // add permissions for role "/xce/v1/roles/*", // add permissions for role
"/xce/v1/users/*", // add roles for user "/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{ var WhitelistDELETEURL = []string{

View file

@ -63,6 +63,11 @@ func TestPathMatchesPattern(t *testing.T) {
{"/xce/v1/roles/*", "/xce/v1/roles/cluster_admin", true}, {"/xce/v1/roles/*", "/xce/v1/roles/cluster_admin", true},
{"/xce/v1/users/*", "/xce/v1/users/current", true}, {"/xce/v1/users/*", "/xce/v1/users/current", true},
{"/xce/v1/users/*", "/xce/v1/users/admin/basic_user", 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 // exact patterns
{"/xce/v1/hosts", "/xce/v1/hosts", true}, {"/xce/v1/hosts", "/xce/v1/hosts", true},
{"/xce/v1/hosts", "/xce/v1/host", false}, {"/xce/v1/hosts", "/xce/v1/host", false},

View file

@ -167,6 +167,17 @@ func Router() *gin.Engine {
eventsApi.GET("", controllers.RequirePermission("event:view"), eventsHandler.SubscribeJobs) 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 // ################ case: resources route
resApi := r.Group("/xce/v1/resources") resApi := r.Group("/xce/v1/resources")
{ {

View file

@ -170,6 +170,14 @@ func main() {
mcp.WithDestructiveHintAnnotation(false), mcp.WithDestructiveHintAnnotation(false),
), app.HandlePendReasons) ), 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 (Phase 1) ──
// Write tools are annotated destructive so the MCP client (e.g. Claude // Write tools are annotated destructive so the MCP client (e.g. Claude
// Code) prompts the user for confirmation before executing them — the // Code) prompts the user for confirmation before executing them — the

View file

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

View file

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

View file

@ -5,6 +5,7 @@ import (
"context" "context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"os"
"strconv" "strconv"
"time" "time"
@ -16,12 +17,15 @@ import (
type App struct { type App struct {
client sclient.Client client sclient.Client
templates *TemplateRegistry 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. // NewApp creates a new App with a schedulerclient connected to the given base URL.
func NewApp(baseURL, token string) (*App, error) { func NewApp(baseURL, token string) (*App, error) {
hc := sclient.NewClient(baseURL, token) 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 ───────────────────────────────────────────────────────────────── // ── 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.", "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).",
})
}