diff --git a/ai/cmd/agent/main.go b/ai/cmd/agent/main.go index bc4a755..e69d46d 100644 --- a/ai/cmd/agent/main.go +++ b/ai/cmd/agent/main.go @@ -95,11 +95,11 @@ func main() { HitlURL: env("HITL_URL", "http://hitl:8081"), } ag := agent.New(acfg, oai, tr, hc, store, enums, pv) - jevMode := env("JEV_MODE", "off") // off | shadow (shadow: Jev answers are recorded next to the LLM diagnosis, never acted on) + jevMode := env("JEV_MODE", "off") // off | shadow | enforce (shadow records Jev answers next to the diagnosis; enforce also applies the two review rules) var jc *jev.Client - if jevMode == "shadow" { + if jevMode == "shadow" || jevMode == "enforce" { jc = jev.New(env("JEV_BASE_URL", ""), env("JEV_MODEL", "jev-1.13.0"), keyReader(env("JEV_KEY_FILE", "/run/secrets/typesafe-api-key"))) - ag.EnableShadow(jc) + ag.EnableShadow(jc, jevMode == "enforce") } log.Printf("agent up: model=%s prompt=%s labels=%d actions=%d jev=%s", acfg.Model, pv, len(enums.Labels), len(enums.Actions), jevMode) if jc != nil && !jc.KeyPresent() { @@ -173,10 +173,11 @@ func main() { return } rev := ag.ShadowReview(c, req, res) - ag.Dispose(c, req, res) if tri != nil || rev != nil { - res.Jev = map[string]any{"mode": "shadow", "triage": tri, "review": rev} + res.Jev = map[string]any{"mode": jevMode, "triage": tri, "review": rev} } + ag.Enforce(res, rev) // no-op unless JEV_MODE=enforce + ag.Dispose(c, req, res) ag.SaveShadow(context.Background(), req, res.SessionID, res, tri, rev) writeJSON(w, 200, res) }) diff --git a/ai/internal/agent/agent.go b/ai/internal/agent/agent.go index 4ebf5b7..6cb1cf8 100644 --- a/ai/internal/agent/agent.go +++ b/ai/internal/agent/agent.go @@ -36,15 +36,16 @@ type Config struct { } type Agent struct { - cfg Config - oai *openai.Client - tools *tools.Runner - harness *harness.Client - store *rag.Store - enums diag.Enums - promptV string - http *http.Client - jev *jev.Client // nil = shadow gates off + cfg Config + oai *openai.Client + tools *tools.Runner + harness *harness.Client + store *rag.Store + enums diag.Enums + promptV string + http *http.Client + jev *jev.Client // nil = shadow gates off + jevEnforce bool // JEV_MODE=enforce: review rules may downgrade a proposal to escalate } func New(cfg Config, oai *openai.Client, tr *tools.Runner, hc *harness.Client, store *rag.Store, enums diag.Enums, promptVersion string) *Agent { @@ -63,8 +64,8 @@ func New(cfg Config, oai *openai.Client, tr *tools.Runner, hc *harness.Client, s return &Agent{cfg: cfg, oai: oai, tools: tr, harness: hc, store: store, enums: enums, promptV: promptVersion, http: &http.Client{Timeout: 30 * time.Second}} } -// EnableShadow turns the Jev shadow gates on (JEV_MODE=shadow). Answers are recorded only. -func (a *Agent) EnableShadow(c *jev.Client) { a.jev = c } +// EnableShadow turns the Jev gates on (JEV_MODE=shadow records only; enforce also applies the review rules). +func (a *Agent) EnableShadow(c *jev.Client, enforce bool) { a.jev, a.jevEnforce = c, enforce } // Request is the harness -> AI VM contract (harness alerts.go diagnose()). type Request struct { diff --git a/ai/internal/agent/enforce_test.go b/ai/internal/agent/enforce_test.go new file mode 100644 index 0000000..37f8055 --- /dev/null +++ b/ai/internal/agent/enforce_test.go @@ -0,0 +1,50 @@ +package agent + +import ( + "testing" + + "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/jev" +) + +func f(v float64) *float64 { return &v } + +func TestEnforceRules(t *testing.T) { + a := &Agent{jevEnforce: true} + // rule A: best_action disagrees with high confidence -> escalate + res := &Result{} + res.ProposedAction, res.Target = "net.route_change", "" + rev := &ShadowRecord{Gate: "review", Answers: map[string]any{"best_action": jev.Answer{Type: "choice", Choice: "escalate", Confidence: f(0.95)}}} + a.Enforce(res, rev) + if res.ProposedAction != "escalate" || res.Jev["enforced"].(map[string]any)["rule"] != "best_action_disagrees" { + t.Fatalf("rule A: %+v %v", res.Diagnosis, res.Jev) + } + // rule B: target invalid -> escalate + res = &Result{} + res.ProposedAction, res.Target = "k8s.rollout_restart", "" + rev = &ShadowRecord{Gate: "review", Answers: map[string]any{"best_action": jev.Answer{Type: "choice", Choice: "k8s.rollout_restart", Confidence: f(0.9)}, "target_valid": jev.Answer{Type: "noul", Noul: f(0.2)}}} + a.Enforce(res, rev) + if res.ProposedAction != "escalate" || res.Jev["enforced"].(map[string]any)["rule"] != "target_invalid" { + t.Fatalf("rule B: %+v %v", res.Diagnosis, res.Jev) + } + // agreement with valid target -> unchanged; none/escalate -> untouched; shadow mode -> untouched + res = &Result{} + res.ProposedAction, res.Target = "k8s.rollout_restart", "api" + rev = &ShadowRecord{Gate: "review", Answers: map[string]any{"best_action": jev.Answer{Choice: "k8s.rollout_restart", Confidence: f(0.99)}, "target_valid": jev.Answer{Noul: f(0.9)}}} + a.Enforce(res, rev) + if res.ProposedAction != "k8s.rollout_restart" || res.Jev != nil { + t.Fatalf("agreement: %+v", res.Diagnosis) + } + res = &Result{} + res.ProposedAction = "none" + a.Enforce(res, &ShadowRecord{Answers: map[string]any{"best_action": jev.Answer{Choice: "escalate", Confidence: f(0.99)}}}) + if res.ProposedAction != "none" { + t.Fatalf("none must not change") + } + s := &Agent{jevEnforce: false} + res = &Result{} + res.ProposedAction = "k8s.scale" + s.Enforce(res, &ShadowRecord{Answers: map[string]any{"best_action": jev.Answer{Choice: "escalate", Confidence: f(0.99)}}}) + if res.ProposedAction != "k8s.scale" { + t.Fatalf("shadow must not change") + } +} diff --git a/ai/internal/agent/shadow.go b/ai/internal/agent/shadow.go index d23e40e..e539753 100644 --- a/ai/internal/agent/shadow.go +++ b/ai/internal/agent/shadow.go @@ -77,26 +77,69 @@ func (a *Agent) ShadowTriage(ctx context.Context, req Request, prefetched string labels["unknown"] = labelHints["unknown"] rec := &ShadowRecord{Gate: "triage"} rec.state = map[string]any{ - "alert": compactAlert(req.Alert), - "samples_last_10m": prefetched, - "slo": "availability: probe_success = 1; latency: api p95 < 0.5 s; errors: 5xx ratio < 1%", + "alert": compactAlert(req.Alert), + "now": a.nowSummary(ctx), + "samples_last_10m": prefetched, + "recent_error_logs": a.recentErrorLogs(ctx), + "slo": "availability: probe_success = 1; latency: api p95 < 0.5 s; errors: 5xx ratio < 1%", } rec.questions = map[string]jev.Question{ - "label": {Type: "choice", Instructions: "Which root cause label best explains `alert` together with `samples_last_10m`?", Criteria: labels}, - "recovering": {Type: "noul", Instructions: "Looking only at the newest values in `samples_last_10m`, is the symptom already gone: probe_success back to 1 and api_p95_latency_s and api_error_ratio inside `slo`?", - Criteria: map[string]string{"true": "the newest samples are inside the SLO again", "false": "the newest samples still show the failure, or the samples are missing"}}, - "needs_investigation": {Type: "noul", Instructions: "To choose a safe remediation action, would an operator need evidence that is not in this state (pod logs, Kubernetes objects, VM status, database replication status)?", + "label": {Type: "choice", Instructions: "Which root cause label best explains `alert`, `now`, `samples_last_10m` and `recent_error_logs`?", Criteria: labels}, + "recovering": {Type: "noul", Instructions: "Look only at `now`: is `now.slo_now_ok` true, meaning the newest probe, the 1-minute p95 and the 1-minute error ratio are all inside `slo` right now?", + Criteria: map[string]string{"true": "now.slo_now_ok is true: the symptom is gone in the newest samples", "false": "now.slo_now_ok is false, or the values are missing"}}, + "needs_investigation": {Type: "noul", Instructions: "To choose a safe remediation action, would an operator need evidence that is not in this state (more logs, Kubernetes objects, VM status, database replication status)?", Criteria: map[string]string{"true": "more evidence is needed before any action", "false": "this state alone is enough to decide the action"}}, - "layer": {Type: "choice", Instructions: "Which recovery layer handles this alert?", Criteria: map[string]string{ - "layer1": "the platform recovers by itself (ReplicaSet, Argo CD selfHeal, MIG autohealing, compose restart); no action needed", - "layer2": "a reversible catalogue action with human approval is needed", - "layer3": "an irreversible or human-only action is needed (database promotion, data fix, instance lifecycle)", - }}, + "irreversible_needed": {Type: "noul", Instructions: "Would fixing the root cause shown in this state require an irreversible or human-only operation: promoting the database replica, repairing a database schema or data, or starting or recreating a stopped VM?", + Criteria: map[string]string{"true": "the fix is irreversible or human-only (database promotion, schema or data repair, VM lifecycle)", "false": "the platform recovers by itself or a reversible catalogue action is enough"}}, } a.ask(ctx, rec) return rec } +// nowSummary computes the newest values and the SLO verdict in code (Jev reads text literally and should not compare +// numbers or read the tail of a series: jev-1.13 jaggedness notes). +func (a *Agent) nowSummary(ctx context.Context) map[string]any { + q := func(promql string) any { + v, err := a.tools.Instant(ctx, promql) + if err != nil || v != v { // NaN + return nil + } + return v + } + probeALB := q(`probe_success{target="alb"}`) + probeDB := q(`probe_success{target="db-primary"}`) + probeMin1m := q(`min_over_time(probe_success{target="alb"}[1m])`) + probeMin3m := q(`min_over_time(probe_success{target="alb"}[3m])`) + p95 := q(`histogram_quantile(0.95, sum by (le) (rate(demo_http_request_duration_seconds_bucket[1m])))`) + errRatio := q(`sum(rate(demo_http_requests_total{status=~"5.."}[1m])) / sum(rate(demo_http_requests_total[1m]))`) + restarts := q(`sum(increase(kube_pod_container_status_restarts_total{namespace="demo"}[10m]))`) + legacyUp := q(`count(up{job="legacy-api"} == 1)`) + f := func(v any) (float64, bool) { + x, ok := v.(float64) + return x, ok + } + pa, okA := f(probeALB) + pv, okP := f(p95) + ev, okE := f(errRatio) + probeOK := okA && pa >= 1 + latOK := !okP || pv < 0.5 // no traffic = inside SLO (same rule as the harness judge) + errOK := !okE || ev < 0.01 + return map[string]any{ + "probe_alb_now": probeALB, "probe_db_primary_now": probeDB, "probe_alb_min_1m": probeMin1m, "probe_alb_min_3m": probeMin3m, + "api_p95_1m_s": p95, "api_error_ratio_1m": errRatio, "api_restarts_10m": restarts, "legacy_instances_up": legacyUp, + "probe_ok_now": probeOK, "latency_ok_now": latOK, "errors_ok_now": errOK, "slo_now_ok": probeOK && latOK && errOK, + } +} + +// recentErrorLogs returns a short, newest-first excerpt of api error lines (5 minutes, 8 lines) for the triage state. +func (a *Agent) recentErrorLogs(ctx context.Context) string { + out := a.tools.Call(ctx, "query_logs", `{"logql":"{namespace=\"demo\", app=\"api\"} |~ \"(?i)error|timeout|refused|SQLSTATE\"","range_minutes":5,"limit":8}`) + if len(out) > 1800 { + out = out[:1800] + "..." + } + return out +} + // ShadowReview asks Jev to check the finished diagnosis against its evidence and the runbook of the proposed action. // It runs for every diagnosis (also none / escalate) so the comparison covers the whole campaign. func (a *Agent) ShadowReview(ctx context.Context, req Request, res *Result) *ShadowRecord { @@ -127,19 +170,53 @@ func (a *Agent) ShadowReview(ctx context.Context, req Request, res *Result) *Sha "rationale": d.Rationale, "proposed_action": d.ProposedAction, "target": d.Target, "harness_target": targetFor(d.ProposedAction, d.Target)}, "runbook": runbook, } + catalogue := d.ProposedAction != "" && d.ProposedAction != "none" && d.ProposedAction != "escalate" rec.questions = map[string]jev.Question{ "label_supported": {Type: "noul", Instructions: "Do the items in `diagnosis.evidence` support `diagnosis.root_cause_label`?", Criteria: map[string]string{"true": "the evidence items show the condition that the label describes", "false": "the evidence is about something else, or too thin to support the label"}}, "runbook_conditions_met": {Type: "noul", Instructions: "Does `diagnosis.evidence` satisfy the conditions written after 'When' or '使う場面' in `runbook` for `diagnosis.proposed_action`? If `runbook` is empty, answer false.", Criteria: map[string]string{"true": "each condition in the runbook is backed by a quoted evidence item", "false": "a condition is not backed by evidence, the evidence contradicts it, or the runbook is empty"}}, - "target_valid": {Type: "noul", Instructions: "Is `diagnosis.target` a concrete, non-empty object name that the harness can act on for `diagnosis.proposed_action` (a deployment such as api or demo/api, a node name such as ip-10-10-10-11, a VM name such as selfheal-db-replica, or a legacy instance name such as legacy-xxxx)? For proposed_action none or escalate, answer true.", - Criteria: map[string]string{"true": "the target names one existing object of the kind the action needs", "false": "the target is empty, generic, or of the wrong kind"}}, "best_action": {Type: "choice", Instructions: "Given `diagnosis.evidence`, `diagnosis.rationale` and `runbook`, which catalogue action should be proposed? Prefer the smallest action; choose escalate for irreversible changes or weak evidence; choose none when the platform already recovers.", Criteria: actions}, } + if catalogue { + rec.questions["target_valid"] = jev.Question{Type: "noul", Instructions: "Is `diagnosis.target` a concrete, non-empty object name that the harness can act on for `diagnosis.proposed_action` (a deployment such as api or demo/api, a node name such as ip-10-10-10-11, a VM name such as selfheal-db-replica, or a legacy instance name such as legacy-xxxx)?", + Criteria: map[string]string{"true": "the target names one existing object of the kind the action needs", "false": "the target is empty, generic, or of the wrong kind"}} + } a.ask(ctx, rec) return rec } +// Enforce applies the two review rules when JEV_MODE=enforce (2026-10-08 campaign-2 report, 3.3). It only ever downgrades +// a catalogue proposal to escalate; none and escalate are never changed. The decision is recorded in res.Jev["enforced"]. +func (a *Agent) Enforce(res *Result, rev *ShadowRecord) { + if !a.jevEnforce || res == nil || rev == nil || rev.Answers == nil { + return + } + d := &res.Diagnosis + if d.ProposedAction == "" || d.ProposedAction == "none" || d.ProposedAction == "escalate" { + return + } + var rule, reason string + if ans, ok := rev.Answers["best_action"].(jev.Answer); ok && ans.Choice != "" && ans.Choice != d.ProposedAction && ans.Confidence != nil && *ans.Confidence >= 0.9 { + rule, reason = "best_action_disagrees", fmt.Sprintf("jev best_action=%s (confidence %.2f) != proposed %s", ans.Choice, *ans.Confidence, d.ProposedAction) + } + if rule == "" { + if ans, ok := rev.Answers["target_valid"].(jev.Answer); ok && ans.Noul != nil && *ans.Noul < 0.5 { + rule, reason = "target_invalid", fmt.Sprintf("jev target_valid=%.2f for target %q", *ans.Noul, d.Target) + } + } + if rule == "" { + return + } + log.Printf("jev enforce: %s: %s -> escalate", rule, reason) + if res.Jev == nil { + res.Jev = map[string]any{} + } + res.Jev["enforced"] = map[string]any{"rule": rule, "reason": reason, "from": d.ProposedAction, "to": "escalate"} + d.Rationale = fmt.Sprintf("%s [jev %s: %s]", d.Rationale, rule, reason) + d.ProposedAction = "escalate" +} + func (a *Agent) ask(ctx context.Context, rec *ShadowRecord) { c, cancel := context.WithTimeout(ctx, 12*time.Second) defer cancel() diff --git a/ai/internal/tools/tools.go b/ai/internal/tools/tools.go index 27c1e47..4b1b1a9 100644 --- a/ai/internal/tools/tools.go +++ b/ai/internal/tools/tools.go @@ -7,6 +7,7 @@ import ( "encoding/json" "fmt" "io" + "math" "net/http" "net/url" "os" @@ -206,6 +207,24 @@ func (r *Runner) queryMetrics(ctx context.Context, promql string, minutes int64) return map[string]any{"series_count": len(pr.Data.Result), "series": out, "range_minutes": minutes}, nil } +// Instant runs an instant PromQL query and returns the first sample as a float (NaN when the series is absent). +func (r *Runner) Instant(ctx context.Context, promql string) (float64, error) { + q := url.Values{"query": {promql}} + var pr promResp + if err := r.getJSON(ctx, strings.TrimRight(r.cfg.MimirURL, "/")+"/api/v1/query?"+q.Encode(), &pr); err != nil { + return math.NaN(), err + } + if len(pr.Data.Result) == 0 || len(pr.Data.Result[0].Value) != 2 { + return math.NaN(), nil + } + sv, _ := pr.Data.Result[0].Value[1].(string) + f, err := strconv.ParseFloat(sv, 64) + if err != nil { + return math.NaN(), err + } + return f, nil +} + func (r *Runner) probeStatus(ctx context.Context) (any, error) { q := url.Values{"query": {"probe_success"}} var pr promResp