diff --git a/ai/cmd/hitl/main.go b/ai/cmd/hitl/main.go index efe59ca..947a430 100644 --- a/ai/cmd/hitl/main.go +++ b/ai/cmd/hitl/main.go @@ -59,6 +59,7 @@ func main() { } connected = true bot.RearmPending(ctx) + go bot.SyncLoop(ctx, 60*time.Second) } mux := http.NewServeMux() diff --git a/ai/internal/agent/shadow.go b/ai/internal/agent/shadow.go index e539753..0cc28c5 100644 --- a/ai/internal/agent/shadow.go +++ b/ai/internal/agent/shadow.go @@ -78,15 +78,15 @@ func (a *Agent) ShadowTriage(ctx context.Context, req Request, prefetched string rec := &ShadowRecord{Gate: "triage"} rec.state = map[string]any{ "alert": compactAlert(req.Alert), - "now": a.nowSummary(ctx), + "now": a.nowSummary(ctx, req.Alert), "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`, `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"}}, + "recovering": {Type: "noul", Instructions: "Look only at `now`: is `now.recovered_now` true, meaning the alert's own condition no longer holds (`now.alert_condition_clear`) and 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.recovered_now is true: the alert condition is clear and the SLO values are inside", "false": "now.recovered_now 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"}}, "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?", @@ -98,7 +98,7 @@ func (a *Agent) ShadowTriage(ctx context.Context, req Request, prefetched string // 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 { +func (a *Agent) nowSummary(ctx context.Context, alert map[string]any) map[string]any { q := func(promql string) any { v, err := a.tools.Instant(ctx, promql) if err != nil || v != v { // NaN @@ -110,8 +110,18 @@ func (a *Agent) nowSummary(ctx context.Context) map[string]any { 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]))`) + p95 := q(`histogram_quantile(0.95, sum by (le) (rate(demo_http_request_duration_seconds_bucket{route=~"orders|price"}[1m])))`) + errRatio := q(`sum(rate(demo_http_requests_total{route=~"orders|price",status=~"5.."}[1m])) / sum(rate(demo_http_requests_total{route=~"orders|price"}[1m]))`) + // the alert's own condition (Grafana annotation `condition`, the rule expression in bool form). A fault that does + // not touch the SLO (replay paused, memory pressure) is "recovering" only when this clears (campaign 3/4: 0.98). + var condNow any + condClear := true + if cond := alertCondition(alert); cond != "" { + condNow = q("max(" + cond + ")") + if v, ok := condNow.(float64); ok { + condClear = v == 0 + } + } 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) { @@ -127,10 +137,22 @@ func (a *Agent) nowSummary(ctx context.Context) map[string]any { 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, + "alert_condition_now": condNow, "alert_condition_clear": condClear, "probe_ok_now": probeOK, "latency_ok_now": latOK, "errors_ok_now": errOK, "slo_now_ok": probeOK && latOK && errOK, + "recovered_now": probeOK && latOK && errOK && condClear, } } +// alertCondition returns the rule expression carried in the alert annotations (empty when absent). +func alertCondition(alert map[string]any) string { + ann, _ := alert["annotations"].(map[string]any) + if ann == nil { + return "" + } + c, _ := ann["condition"].(string) + return c +} + // 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}`) diff --git a/ai/internal/harness/client.go b/ai/internal/harness/client.go index 130755b..d394b3b 100644 --- a/ai/internal/harness/client.go +++ b/ai/internal/harness/client.go @@ -57,6 +57,27 @@ type Proposal struct { Approved bool `json:"approved"` } +// ProposalState is GET /proposals/{token}: the decision as the harness knows it (approval may come through the API). +type ProposalState struct { + Token string `json:"token"` + ActionID string `json:"action_id"` + Approved bool `json:"approved"` + ApprovedAt *string `json:"approved_at"` + ApprovedBy *string `json:"approved_by"` + Executed bool `json:"executed"` + ExecutedAt *string `json:"executed_at"` + ExpiresAt string `json:"expires_at"` +} + +// GetProposal reads the harness-side state of a proposal. +func (c *Client) GetProposal(ctx context.Context, token string) (*ProposalState, error) { + var out ProposalState + if _, err := c.call(ctx, "GET", "/proposals/"+token, nil, &out); err != nil { + return nil, err + } + return &out, nil +} + // Propose registers a layer-2 proposal (POST /proposals); the harness records the proposal event and returns the token. func (c *Client) Propose(ctx context.Context, runID, scenarioID *string, actionID string, target, diagnosis map[string]any, ttl time.Duration) (*Proposal, error) { body := map[string]any{"run_id": runID, "scenario_id": scenarioID, "action_id": actionID, "target": target, "diagnosis": diagnosis} diff --git a/ai/internal/hitl/bot.go b/ai/internal/hitl/bot.go index c1727bc..c41a62c 100644 --- a/ai/internal/hitl/bot.go +++ b/ai/internal/hitl/bot.go @@ -230,12 +230,57 @@ func (b *Bot) edit(m *rag.HitlMessage, status string, color int, result map[stri return b.dc.Edit(m.ChannelID, m.MessageID, em, b.buttons(m.Token, true)) } +// syncOne mirrors an approval recorded by the harness (API path, not the button) into the message. Returns true when +// the message was decided here. +func (b *Bot) syncOne(ctx context.Context, m *rag.HitlMessage) bool { + st, err := b.harness.GetProposal(ctx, m.Token) + if err != nil || !st.Approved { + return false + } + by := "unknown" + if st.ApprovedBy != nil && *st.ApprovedBy != "" { + by = *st.ApprovedBy + } + first, err := b.store.Decide(ctx, m.Token, "approved", by, map[string]any{"via": "harness-api", "executed": st.Executed}) + if err != nil || !first { + return false + } + b.disarm(m.Token) + _ = b.edit(m, b.t.StatusApprovedAPI(by, st.Executed), colorApproved, nil) + log.Printf("proposal %s approved outside Discord by %s (executed=%v)", m.Token[len(m.Token)-6:], by, st.Executed) + return true +} + +// SyncLoop polls the harness for pending proposals decided outside Discord (campaign 3/4 2026-10-08: approvals through +// POST /proposals/{token}/approve left the message to expire by timer). +func (b *Bot) SyncLoop(ctx context.Context, every time.Duration) { + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + list, err := b.store.Pending(ctx) + if err != nil { + continue + } + for i := range list { + b.syncOne(ctx, &list[i]) + } + } + } +} + func (b *Bot) expire(token string) { ctx := context.Background() m, err := b.store.GetHitl(ctx, token) if err != nil { return } + if b.syncOne(ctx, m) { + return + } first, err := b.store.Decide(ctx, token, "expired", "timer", nil) if err != nil || !first { return diff --git a/ai/internal/hitl/i18n.go b/ai/internal/hitl/i18n.go index f8c4a30..b933bdc 100644 --- a/ai/internal/hitl/i18n.go +++ b/ai/internal/hitl/i18n.go @@ -25,6 +25,7 @@ type Strings struct { StatusExecFailed func(by, err string) string StatusExpiredBefore string StatusNoDecision string + StatusApprovedAPI func(by string, executed bool) string ExpiryMessage func(actionID, target, runID, scenarioID string) string EscalationTitle func(label string) string EscalationStatus string @@ -48,7 +49,13 @@ var langs = map[string]Strings{ StatusApproveFailed: func(e string) string { return "approval failed: " + e }, StatusExecFailed: func(by, e string) string { return "approved by " + by + "; execution failed: " + e }, StatusExpiredBefore: "expired before approval", - StatusNoDecision: "no decision within the window; moved to layer 3", + StatusApprovedAPI: func(by string, executed bool) string { + if executed { + return "approved through the harness API by " + by + "; executed" + } + return "approved through the harness API by " + by + }, + StatusNoDecision: "no decision within the window; moved to layer 3", ExpiryMessage: func(a, t, r, s string) string { return fmt.Sprintf("Proposal `%s` for `%s` (run %s, scenario %s) received no decision within the approval window. Layer 3: follow the runbook or call the harness `/actions/execute` with the operator token.", a, t, r, s) }, @@ -72,7 +79,13 @@ var langs = map[string]Strings{ StatusApproveFailed: func(e string) string { return "承認の記録に失敗: " + e }, StatusExecFailed: func(by, e string) string { return by + " が承認したが実行に失敗: " + e }, StatusExpiredBefore: "承認前に期限切れ", - StatusNoDecision: "期限内に判断がなかったため第3層へ引き継ぎ", + StatusApprovedAPI: func(by string, executed bool) string { + if executed { + return by + " がハーネス API で承認、実行済み" + } + return by + " がハーネス API で承認" + }, + StatusNoDecision: "期限内に判断がなかったため第3層へ引き継ぎ", ExpiryMessage: func(a, t, r, s string) string { return fmt.Sprintf("提案 `%s`（対象 `%s`、run %s、シナリオ %s）は承認期限内に判断されませんでした。第3層の対応です。ランブックに従って手動で対処するか、オペレーター用トークンでハーネスの `/actions/execute` を呼び出してください。", a, t, r, s) }, diff --git a/harness/alerts.go b/harness/alerts.go index 4fbe117..e4b2750 100644 --- a/harness/alerts.go +++ b/harness/alerts.go @@ -24,10 +24,14 @@ type AlertRouter struct { mu sync.Mutex timers map[string]*time.Timer // fingerprint -> escalation timer firing map[string]string // fingerprint -> alertname while Grafana reports it firing (judge: no recovery with an open alert) + // alertname+scenario -> when the last layer-2 diagnosis was dispatched. Grafana emits one alert per label set + // (ApiMemoryPressure: one per pod) and resends the group 30 s later; campaign 4 (2026-10-08) turned one fault into + // four diagnoses and four proposals. Inside cfg.AlertDedupeS the later alerts are recorded and skipped. + lastDiag map[string]time.Time } func NewAlertRouter(a *App) *AlertRouter { - return &AlertRouter{app: a, timers: map[string]*time.Timer{}, firing: map[string]string{}} + return &AlertRouter{app: a, timers: map[string]*time.Timer{}, firing: map[string]string{}, lastDiag: map[string]time.Time{}} } // Firing lists the alert names currently firing (deduplicated, sorted). @@ -113,6 +117,7 @@ func (r *AlertRouter) handleWebhook(w http.ResponseWriter, req *http.Request) { r.setFiring(al.Fingerprint, name, al.Status == "firing" && sig == "") if al.Status != "firing" { r.cancelTimer(al.Fingerprint) + r.clearDedupe(name, sid) continue } if sig != "" { @@ -129,11 +134,53 @@ func (r *AlertRouter) handleWebhook(w http.ResponseWriter, req *http.Request) { r.startEscalation(al.Fingerprint, payload, runID, sid) continue } + if since, dup := r.dedupe(name, sid); dup { + r.app.store.Event(bg, runID, sid, "note", "runner", map[string]any{"phase": "dedupe", "alertname": name, "fingerprint": al.Fingerprint, + "since_last_diagnosis_s": since.Seconds(), "note": "same alert name already under diagnosis; no second /diagnose"}) + continue + } go r.diagnose(payload, runID, sid) } writeJSON(w, 200, map[string]any{"received": handled, "run_id": runID, "scenario_id": sid}) } +// dedupe reports whether a layer-2 diagnosis for this alert name (within the scenario) was dispatched less than +// cfg.AlertDedupeS ago; when not, it records now as the dispatch time. +func (r *AlertRouter) dedupe(name string, sid *string) (time.Duration, bool) { + if r.app.cfg.AlertDedupeS <= 0 { + return 0, false + } + key := name + if sid != nil { + key += "|" + *sid + } + r.mu.Lock() + defer r.mu.Unlock() + now := time.Now() + if t, ok := r.lastDiag[key]; ok && now.Sub(t) < time.Duration(r.app.cfg.AlertDedupeS)*time.Second { + return now.Sub(t), true + } + r.lastDiag[key] = now + return 0, false +} + +// clearDedupe forgets the dispatch time once no alert of that name is firing any more, so the next firing starts a +// new diagnosis even inside the window. +func (r *AlertRouter) clearDedupe(name string, sid *string) { + key := name + if sid != nil { + key += "|" + *sid + } + r.mu.Lock() + defer r.mu.Unlock() + for _, n := range r.firing { + if n == name { + return + } + } + delete(r.lastDiag, key) +} + func targetFromLabels(l map[string]string) map[string]any { t := map[string]any{} for _, k := range []string{"namespace", "pod", "deployment", "node", "instance"} { diff --git a/harness/alerts_test.go b/harness/alerts_test.go new file mode 100644 index 0000000..f799b16 --- /dev/null +++ b/harness/alerts_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "testing" + "time" +) + +// one fault, several alerts with the same name (per-pod label sets, group resend): one diagnosis inside the window +func TestDedupeSameAlertName(t *testing.T) { + r := NewAlertRouter(&App{cfg: Config{AlertDedupeS: 300}}) + sid := "C-G2" + if _, dup := r.dedupe("ApiMemoryPressure", &sid); dup { + t.Fatal("first alert must be diagnosed") + } + if _, dup := r.dedupe("ApiMemoryPressure", &sid); !dup { + t.Fatal("second alert inside the window must be skipped") + } + other := "V-G1" + if _, dup := r.dedupe("ApiMemoryPressure", &other); dup { + t.Fatal("another scenario is a new fault") + } + // all alerts of that name resolved -> the next firing starts a new diagnosis + r.setFiring("fp1", "ApiMemoryPressure", true) + r.clearDedupe("ApiMemoryPressure", &sid) // still firing: keep + if _, dup := r.dedupe("ApiMemoryPressure", &sid); !dup { + t.Fatal("still firing: keep the window") + } + r.setFiring("fp1", "ApiMemoryPressure", false) + r.clearDedupe("ApiMemoryPressure", &sid) + if _, dup := r.dedupe("ApiMemoryPressure", &sid); dup { + t.Fatal("after resolve the next firing is a new fault") + } + // window expiry + r.lastDiag["ApiMemoryPressure|"+sid] = time.Now().Add(-301 * time.Second) + if _, dup := r.dedupe("ApiMemoryPressure", &sid); dup { + t.Fatal("outside the window a new diagnosis is allowed") + } + // off + r2 := NewAlertRouter(&App{cfg: Config{AlertDedupeS: 0}}) + if _, dup := r2.dedupe("X", nil); dup { + t.Fatal("dedupe off") + } + if _, dup := r2.dedupe("X", nil); dup { + t.Fatal("dedupe off") + } +} diff --git a/harness/main.go b/harness/main.go index 918223f..868cbe0 100644 --- a/harness/main.go +++ b/harness/main.go @@ -9,6 +9,7 @@ import ( "log" "net/http" "os" + "strconv" "strings" "sync" "time" @@ -33,6 +34,7 @@ type Config struct { DBPassFile string // harness role password on db-primary / db-replica HumanTokFile string AIDiagnoseURL string + AlertDedupeS int // layer-2 alerts with the same name inside this window share one diagnosis (0 = off) ScenarioDir string JudgeEvery time.Duration ObserveMax time.Duration @@ -62,6 +64,7 @@ func loadConfig() Config { DBPassFile: env("DB_PASSWORD_FILE", "/run/secrets/db-password"), HumanTokFile: env("HUMAN_TOKEN_FILE", "/run/secrets/harness-token"), AIDiagnoseURL: env("AI_DIAGNOSE_URL", "http://10.20.0.12:8080/diagnose"), + AlertDedupeS: envInt("ALERT_DEDUPE_S", 300), ScenarioDir: env("SCENARIO_DIR", "/etc/harness/scenarios"), JudgeEvery: 30 * time.Second, ObserveMax: 30 * time.Minute, @@ -115,6 +118,7 @@ func main() { mux.HandleFunc("POST /runs/{id}/abort", app.handleRunAbort) mux.HandleFunc("POST /hooks/alert", app.hooks.handleWebhook) mux.HandleFunc("POST /proposals", app.handleProposal) + mux.HandleFunc("GET /proposals/{token}", app.handleProposalGet) mux.HandleFunc("POST /proposals/{token}/approve", app.handleProposalApprove) mux.HandleFunc("POST /proposals/{token}/reject", app.handleProposalReject) mux.HandleFunc("POST /actions/execute", app.handleActionExecute) @@ -208,3 +212,12 @@ func (a *App) handleScenarios(w http.ResponseWriter, r *http.Request) { sortScenarios(list) writeJSON(w, 200, list) } + +func envInt(k string, d int) int { + if v := os.Getenv(k); v != "" { + if n, err := strconv.Atoi(v); err == nil { + return n + } + } + return d +} diff --git a/harness/proposals.go b/harness/proposals.go index f638fdb..16ae96a 100644 --- a/harness/proposals.go +++ b/harness/proposals.go @@ -62,6 +62,18 @@ func (a *App) handleProposal(w http.ResponseWriter, r *http.Request) { writeJSON(w, 201, map[string]any{"token": token, "expires_in_s": ttl.Seconds(), "approved": req.Approved}) } +// GET /proposals/{token}: the HITL bot reads the decision state (approval may arrive through this API instead of the +// Discord button; the bot mirrors it into its message instead of letting the message expire, campaign 3/4 2026-10-08). +func (a *App) handleProposalGet(w http.ResponseWriter, r *http.Request) { + p, err := a.store.GetProposal(r.Context(), r.PathValue("token")) + if err != nil { + writeErr(w, 404, errors.New("unknown proposal")) + return + } + writeJSON(w, 200, map[string]any{"token": p.Token, "run_id": p.RunID, "scenario_id": p.ScenarioID, "action_id": p.ActionID, + "approved": p.ApprovedAt != nil, "approved_at": p.ApprovedAt, "approved_by": p.ApprovedBy, "executed": p.ExecutedAt != nil, "executed_at": p.ExecutedAt, "expires_at": p.ExpiresAt}) +} + // POST /proposals/{token}/approve: the HITL bot records the human decision (design 6.5 step 4). The proposal was // registered at diagnosis time, so MTTA = approval - proposal (design 5.4) measures the real waiting time. func (a *App) handleProposalApprove(w http.ResponseWriter, r *http.Request) {