seon
91cdd9ada5
TypeSafe System One（Jev）への最小クライアントと、診断ループに並走する shadow ゲート 2 つを追加した。triage はアラートと事前取得サンプルからラベル・ 回復中か・層を判定し、review は診断の根拠・ランブック・target を検証して best_action を選ぶ。JEV_MODE=shadow のときだけ動き、答えは ai.jev_shadow と 診断結果の jev フィールドに記録するだけで、処理の分岐は一切変えない。
199 lines
11 KiB
Go
199 lines
11 KiB
Go
package agent
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/jev"
|
|
"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/rag"
|
|
)
|
|
|
|
// Shadow gates (2026-10-08): two System One (Jev) judgments run next to the LLM diagnosis and are only recorded.
|
|
// Triage runs before the LLM sees the alert (same pre-fetched samples the LLM gets), Review runs on the finished
|
|
// diagnosis (evidence + runbook of the proposed action). Neither changes the disposition; the campaign report compares
|
|
// the answers with the LLM output and the scenario ground truth to decide whether an enforce mode is worth it.
|
|
|
|
// ShadowRecord is one gate call, saved to ai.jev_shadow and echoed in the Result for the harness event payload.
|
|
type ShadowRecord struct {
|
|
Gate string `json:"gate"`
|
|
Model string `json:"model,omitempty"`
|
|
Answers map[string]any `json:"answers,omitempty"`
|
|
LatencyMs int64 `json:"latency_ms"`
|
|
InputTokens int `json:"input_tokens,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
state any
|
|
questions map[string]jev.Question
|
|
}
|
|
|
|
// Label descriptions (system.md label hints) used as Choice criteria; the model reads these, not the enum names alone.
|
|
var labelHints = map[string]string{
|
|
"api_pod_killed": "one api pod was killed; the ReplicaSet recreates it (platform recovers)",
|
|
"api_pod_failure": "api pods crash-looping or failing readiness; restart counts rising",
|
|
"memory_pressure_api": "api pods under memory stress; latency up, OOMKilled possible",
|
|
"network_latency_api_to_db": "the api to database path is slow (p95 up, no restarts, the database itself is healthy)",
|
|
"gitops_drift": "an Argo CD application is OutOfSync; selfHeal restores it (platform recovers)",
|
|
"legacy_process_killed": "the legacy-api container was killed on a MIG instance; compose restarts it (platform recovers)",
|
|
"legacy_instance_lost": "a legacy MIG instance was deleted or stopped; autohealing recreates it (platform recovers)",
|
|
"network_latency_legacy": "the ILB to legacy path is slow or lossy",
|
|
"replication_paused": "db-replica WAL replay is paused (pg_is_wal_replay_paused = 1, lag rising)",
|
|
"db_primary_down": "db-primary is unreachable on 5432 (probe db-primary = 0)",
|
|
"schema_mismatch": "application errors from a schema or data change on the database (5xx with SQL errors in logs)",
|
|
"transient_partition": "a short network partition that already cleared",
|
|
"transient_loss_legacy": "short packet loss on the legacy path that already cleared",
|
|
"unknown": "the evidence does not support any of the other labels",
|
|
}
|
|
|
|
var actionHints = map[string]string{
|
|
"none": "no action: the platform already recovers or the symptom cleared",
|
|
"escalate": "hand to a human: irreversible action needed, no catalogue action fits, or evidence is insufficient",
|
|
"k8s.rollout_restart": "restart the api or front pods one by one (degraded pods without a configuration change)",
|
|
"k8s.rollout_undo": "roll a deployment back to the previous revision (the newest revision fails, the previous one was healthy)",
|
|
"k8s.scale": "add api replicas (capacity problem: pods Running, CPU or memory saturated, no crashes)",
|
|
"k8s.node_drain": "drain one unhealthy worker node (NotReady or kubelet errors on that node only)",
|
|
"k8s.node_uncordon": "uncordon a worker node after a drain",
|
|
"gce.instance_reset": "reset a hung pet VM (status RUNNING but no SSH, metrics or docker); never db-primary",
|
|
"mig.recreate_instance": "recreate one legacy-api MIG instance that serves errors and was not replaced by autohealing",
|
|
"mig.resize": "set the legacy MIG back to 2 instances",
|
|
"db.replay_resume": "resume WAL replay on db-replica when pg_is_wal_replay_paused = 1",
|
|
"net.route_change": "operator changes a route or firewall through Terraform (BGP sessions down, routes missing on one side; not transient)",
|
|
}
|
|
|
|
func (a *Agent) shadowOn() bool { return a.jev != nil }
|
|
|
|
// ShadowTriage asks Jev about the raw alert plus the pre-fetched samples (the same material the LLM user message
|
|
// carries), before the LLM diagnosis starts. Returns nil when shadow mode is off.
|
|
func (a *Agent) ShadowTriage(ctx context.Context, req Request, prefetched string) *ShadowRecord {
|
|
if !a.shadowOn() {
|
|
return nil
|
|
}
|
|
labels := map[string]string{}
|
|
for _, l := range a.enums.Labels {
|
|
labels[l] = labelHints[l]
|
|
}
|
|
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%",
|
|
}
|
|
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)?",
|
|
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)",
|
|
}},
|
|
}
|
|
a.ask(ctx, rec)
|
|
return rec
|
|
}
|
|
|
|
// 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 {
|
|
if !a.shadowOn() || res == nil {
|
|
return nil
|
|
}
|
|
d := res.Diagnosis
|
|
runbook := ""
|
|
if q := strings.TrimSpace(d.ProposedAction); q != "" && q != "none" {
|
|
if hits, err := a.tools.SearchRunbooks(ctx, q, 2); err == nil {
|
|
if hs, ok := hits.([]rag.Hit); ok {
|
|
var parts []string
|
|
for _, h := range hs {
|
|
parts = append(parts, "["+h.Source+"] "+h.Chunk)
|
|
}
|
|
runbook = strings.Join(parts, "\n\n")
|
|
}
|
|
}
|
|
}
|
|
actions := map[string]string{"none": actionHints["none"], "escalate": actionHints["escalate"]}
|
|
for _, id := range a.enums.Actions {
|
|
actions[id] = actionHints[id]
|
|
}
|
|
rec := &ShadowRecord{Gate: "review"}
|
|
rec.state = map[string]any{
|
|
"alert": compactAlert(req.Alert),
|
|
"diagnosis": map[string]any{"root_cause_label": d.RootCauseLabel, "confidence": d.Confidence, "evidence": d.Evidence,
|
|
"rationale": d.Rationale, "proposed_action": d.ProposedAction, "target": d.Target, "harness_target": targetFor(d.ProposedAction, d.Target)},
|
|
"runbook": runbook,
|
|
}
|
|
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},
|
|
}
|
|
a.ask(ctx, rec)
|
|
return rec
|
|
}
|
|
|
|
func (a *Agent) ask(ctx context.Context, rec *ShadowRecord) {
|
|
c, cancel := context.WithTimeout(ctx, 12*time.Second)
|
|
defer cancel()
|
|
t0 := time.Now()
|
|
resp, err := a.jev.Ask(c, rec.state, rec.questions)
|
|
rec.LatencyMs = time.Since(t0).Milliseconds()
|
|
if err != nil {
|
|
rec.Error = err.Error()
|
|
if err != jev.ErrNoKey {
|
|
log.Printf("jev %s: %v", rec.Gate, err)
|
|
}
|
|
return
|
|
}
|
|
rec.Model = resp.Model
|
|
rec.InputTokens = resp.Usage.InputTokens
|
|
rec.Answers = map[string]any{}
|
|
for k, v := range resp.Answers {
|
|
rec.Answers[k] = v
|
|
}
|
|
}
|
|
|
|
// SaveShadow stores the gate records with the session id and the LLM's own decision for the later comparison.
|
|
func (a *Agent) SaveShadow(ctx context.Context, req Request, sessionID string, res *Result, recs ...*ShadowRecord) {
|
|
if a.store == nil {
|
|
return
|
|
}
|
|
var llm map[string]any
|
|
if res != nil {
|
|
llm = map[string]any{"root_cause_label": res.RootCauseLabel, "confidence": res.Confidence, "proposed_action": res.ProposedAction, "target": res.Target, "disposition": res.Disposition}
|
|
}
|
|
for _, r := range recs {
|
|
if r == nil {
|
|
continue
|
|
}
|
|
row := rag.JevShadow{SessionID: sessionID, RunID: req.RunID, ScenarioID: req.ScenarioID, Gate: r.Gate, Model: r.Model,
|
|
State: r.state, Questions: r.questions, Answers: r.Answers, LLM: llm, LatencyMs: r.LatencyMs, InputTokens: r.InputTokens, Error: r.Error}
|
|
if err := a.store.SaveJev(ctx, row); err != nil {
|
|
log.Printf("jev shadow save (%s): %v", r.Gate, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// compactAlert keeps the fields a judgment needs (name, labels, annotations, status, times) and drops Grafana noise.
|
|
func compactAlert(alert map[string]any) map[string]any {
|
|
out := map[string]any{}
|
|
for _, k := range []string{"alertname", "status", "labels", "annotations", "startsAt", "endsAt", "valueString"} {
|
|
if v, ok := alert[k]; ok {
|
|
out[k] = v
|
|
}
|
|
}
|
|
if len(out) == 0 {
|
|
b, _ := json.Marshal(alert)
|
|
out["raw"] = fmt.Sprintf("%.1500s", string(b))
|
|
}
|
|
return out
|
|
}
|