From 91cdd9ada5a2e07c9f4dd0ca5b9d1fc87917d302 Mon Sep 17 00:00:00 2001 From: seonNoh Date: Thu, 8 Oct 2026 15:10:07 +0900 Subject: [PATCH] feat(ai): add Jev shadow gates (triage before, review after the LLM diagnosis) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TypeSafe System One（Jev）への最小クライアントと、診断ループに並走する shadow ゲート 2 つを追加した。triage はアラートと事前取得サンプルからラベル・ 回復中か・層を判定し、review は診断の根拠・ランブック・target を検証して best_action を選ぶ。JEV_MODE=shadow のときだけ動き、答えは ai.jev_shadow と 診断結果の jev フィールドに記録するだけで、処理の分岐は一切変えない。 --- ai/cmd/agent/main.go | 27 ++++- ai/internal/agent/agent.go | 55 ++++++--- ai/internal/agent/shadow.go | 198 +++++++++++++++++++++++++++++++++ ai/internal/jev/client.go | 124 +++++++++++++++++++++ ai/internal/jev/client_test.go | 43 +++++++ ai/internal/rag/store.go | 51 +++++++++ 6 files changed, 479 insertions(+), 19 deletions(-) create mode 100644 ai/internal/agent/shadow.go create mode 100644 ai/internal/jev/client.go create mode 100644 ai/internal/jev/client_test.go diff --git a/ai/cmd/agent/main.go b/ai/cmd/agent/main.go index 16dcfca..bc4a755 100644 --- a/ai/cmd/agent/main.go +++ b/ai/cmd/agent/main.go @@ -13,6 +13,7 @@ import ( "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/agent" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/diag" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/harness" + "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/jev" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/rag" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/tools" @@ -94,7 +95,16 @@ func main() { HitlURL: env("HITL_URL", "http://hitl:8081"), } ag := agent.New(acfg, oai, tr, hc, store, enums, pv) - log.Printf("agent up: model=%s prompt=%s labels=%d actions=%d", acfg.Model, pv, len(enums.Labels), len(enums.Actions)) + jevMode := env("JEV_MODE", "off") // off | shadow (shadow: Jev answers are recorded next to the LLM diagnosis, never acted on) + var jc *jev.Client + if jevMode == "shadow" { + 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) + } + 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() { + log.Printf("jev shadow: typesafe key file empty; gates record 'typesafe api key is empty' until the secret is registered") + } // runbook ingest (needs the OpenAI key; retried in the background until the key exists) go func() { @@ -120,6 +130,11 @@ func main() { out := map[string]any{"model": acfg.Model, "prompt_version": pv, "labels": enums.Labels, "actions": enums.Actions} key, kerr := keyReader(env("OPENAI_KEY_FILE", "/run/secrets/openai-api-key"))() out["openai_key_present"] = kerr == nil && key != "" + out["jev_mode"] = jevMode + if jc != nil { + out["jev_key_present"] = jc.KeyPresent() + out["jev_model"] = jc.Model() + } out["rag_db"] = errText(store.Ping(c)) n, _ := store.DocCount(c) out["rag_chunks"] = n @@ -149,12 +164,20 @@ func main() { defer func() { inflight-- }() c, cancel := context.WithTimeout(context.Background(), 170*time.Second) // harness waits 3 minutes defer cancel() - res, err := ag.Diagnose(c, req) + prefetched := ag.Prefetch(c) + tri := ag.ShadowTriage(c, req, prefetched) // nil unless JEV_MODE=shadow; recorded only + res, err := ag.DiagnoseWith(c, req, prefetched) if err != nil { + ag.SaveShadow(context.Background(), req, "", nil, tri) writeJSON(w, 503, map[string]string{"error": err.Error()}) 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} + } + ag.SaveShadow(context.Background(), req, res.SessionID, res, tri, rev) writeJSON(w, 200, res) }) srv := &http.Server{Addr: env("AGENT_ADDR", ":8080"), Handler: logging(mux), ReadHeaderTimeout: 10 * time.Second} diff --git a/ai/internal/agent/agent.go b/ai/internal/agent/agent.go index 149b58f..4ebf5b7 100644 --- a/ai/internal/agent/agent.go +++ b/ai/internal/agent/agent.go @@ -17,6 +17,7 @@ import ( "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/diag" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/harness" + "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/jev" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/rag" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/tools" @@ -43,6 +44,7 @@ type Agent struct { enums diag.Enums promptV string http *http.Client + jev *jev.Client // nil = shadow gates off } func New(cfg Config, oai *openai.Client, tr *tools.Runner, hc *harness.Client, store *rag.Store, enums diag.Enums, promptVersion string) *Agent { @@ -61,6 +63,9 @@ 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 } + // Request is the harness -> AI VM contract (harness alerts.go diagnose()). type Request struct { Alert map[string]any `json:"alert"` @@ -72,14 +77,15 @@ type Request struct { // diagnosis event payload and reads root_cause_label / proposed_action from it) plus bookkeeping. type Result struct { diag.Diagnosis - SessionID string `json:"session_id"` - Model string `json:"model"` - PromptVersion string `json:"prompt_version"` - Turns int `json:"turns"` - ToolCalls int `json:"tool_calls"` - DurationMs int64 `json:"duration_ms"` - Disposition string `json:"disposition"` // proposal | escalated | none - ProposalToken string `json:"proposal_token,omitempty"` + SessionID string `json:"session_id"` + Model string `json:"model"` + PromptVersion string `json:"prompt_version"` + Turns int `json:"turns"` + ToolCalls int `json:"tool_calls"` + DurationMs int64 `json:"duration_ms"` + Disposition string `json:"disposition"` // proposal | escalated | none + ProposalToken string `json:"proposal_token,omitempty"` + Jev map[string]any `json:"jev,omitempty"` // shadow gate answers (recorded, never acted on) } type toolCallLog struct { @@ -97,7 +103,10 @@ func (a *Agent) toolDefs() []openai.Tool { return defs } -func (a *Agent) userMessage(ctx context.Context, req Request) string { +func (a *Agent) userMessage(ctx context.Context, req Request, prefetched string) string { + if prefetched == "" { + prefetched = a.Prefetch(ctx) + } alert, _ := json.MarshalIndent(req.Alert, "", " ") var b strings.Builder b.WriteString("## Topology\n") @@ -107,6 +116,20 @@ func (a *Agent) userMessage(ctx context.Context, req Request) string { b.WriteString("\n\n## Alert payload\n```json\n") b.Write(alert) b.WriteString("\n```\n\n## Last 10 minutes (pre-fetched)\n") + b.WriteString(prefetched) + if req.RunID != nil { + fmt.Fprintf(&b, "\nrun_id=%s", *req.RunID) + } + if req.ScenarioID != nil { + fmt.Fprintf(&b, " scenario_id=%s", *req.ScenarioID) + } + b.WriteString("\n\nInvestigate with the tools, then call submit_diagnosis exactly once.") + return b.String() +} + +// Prefetch returns the four pre-fetched 10-minute series (probe, p95, error ratio, legacy up) as markdown lines. +func (a *Agent) Prefetch(ctx context.Context) string { + var b strings.Builder for _, q := range []struct{ name, promql string }{ {"probe_success", `probe_success`}, {"api_p95_latency_s", `histogram_quantile(0.95, sum by (le) (rate(demo_http_request_duration_seconds_bucket[5m])))`}, @@ -119,25 +142,23 @@ func (a *Agent) userMessage(ctx context.Context, req Request) string { } fmt.Fprintf(&b, "- %s: %s\n", q.name, out) } - if req.RunID != nil { - fmt.Fprintf(&b, "\nrun_id=%s", *req.RunID) - } - if req.ScenarioID != nil { - fmt.Fprintf(&b, " scenario_id=%s", *req.ScenarioID) - } - b.WriteString("\n\nInvestigate with the tools, then call submit_diagnosis exactly once.") return b.String() } // Diagnose runs the tool loop and returns the validated diagnosis. The session is stored whatever the outcome. func (a *Agent) Diagnose(ctx context.Context, req Request) (*Result, error) { + return a.DiagnoseWith(ctx, req, "") +} + +// DiagnoseWith runs Diagnose with an already pre-fetched sample block (empty = fetch now). +func (a *Agent) DiagnoseWith(ctx context.Context, req Request, prefetched string) (*Result, error) { start := time.Now() sid := "ses-" + uuid.NewString()[:8] res := &Result{SessionID: sid, Model: a.cfg.Model, PromptVersion: a.promptV} var calls []toolCallLog input := []openai.Item{ {Type: "message", Role: "system", Content: a.cfg.SystemPrompt}, - {Type: "message", Role: "user", Content: a.userMessage(ctx, req)}, + {Type: "message", Role: "user", Content: a.userMessage(ctx, req, prefetched)}, } session := rag.Session{SessionID: sid, RunID: req.RunID, ScenarioID: req.ScenarioID, Model: a.cfg.Model, PromptVersion: a.promptV, Input: map[string]any{"alert": req.Alert, "user_message": input[1].Content}} finish := func(err error) { diff --git a/ai/internal/agent/shadow.go b/ai/internal/agent/shadow.go new file mode 100644 index 0000000..d23e40e --- /dev/null +++ b/ai/internal/agent/shadow.go @@ -0,0 +1,198 @@ +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 +} diff --git a/ai/internal/jev/client.go b/ai/internal/jev/client.go new file mode 100644 index 0000000..72f1c4f --- /dev/null +++ b/ai/internal/jev/client.go @@ -0,0 +1,124 @@ +// Package jev is a minimal client for TypeSafe's System One API (model Jev): POST /v1/systemone with a state and a map +// of typed questions (noul | choice | score), answered in parallel with calibrated probabilities. API reference: +// https://docs.typesafe.ai/api (checked 2026-10-08). Used only in shadow mode here: answers are recorded next to the +// LLM diagnosis and never change what the agent does. +package jev + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "strings" + "time" +) + +var ErrNoKey = errors.New("typesafe api key is empty") + +type Question struct { + Type string `json:"type"` // noul | choice | score + Instructions string `json:"instructions"` // the full question (the map key is not sent to the model) + Criteria any `json:"criteria,omitempty"` // noul: {true:..., false:...}; choice: map[option]description; score: []level +} + +type Answer struct { + Type string `json:"type"` + Noul *float64 `json:"noul,omitempty"` + Choice string `json:"choice,omitempty"` + Score *float64 `json:"score,omitempty"` + Probabilities map[string]float64 `json:"probabilities,omitempty"` + Confidence *float64 `json:"confidence,omitempty"` +} + +type Usage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` +} + +type Response struct { + Model string `json:"model"` + Answers map[string]Answer `json:"answers"` + Usage Usage `json:"usage"` +} + +type Client struct { + base string + model string + key func() (string, error) + http *http.Client +} + +// New builds a client. base defaults to https://api.typesafe.ai; model should be pinned (jev-1.13.0) so recorded +// answers stay comparable across campaigns (models page: aliases move on release). +func New(base, model string, key func() (string, error)) *Client { + if base == "" { + base = "https://api.typesafe.ai" + } + if model == "" { + model = "jev-1.13.0" + } + return &Client{base: strings.TrimRight(base, "/"), model: model, key: key, http: &http.Client{Timeout: 10 * time.Second}} +} + +func (c *Client) Model() string { return c.model } + +// KeyPresent reports whether a non-empty key can be read (for /selfcheck). +func (c *Client) KeyPresent() bool { + k, err := c.key() + return err == nil && strings.TrimSpace(k) != "" +} + +// Ask sends one request. 429/529 are retried once after the server's retry-after (or 1s); other errors return as is. +func (c *Client) Ask(ctx context.Context, state any, questions map[string]Question) (*Response, error) { + key, err := c.key() + if err != nil || strings.TrimSpace(key) == "" { + return nil, ErrNoKey + } + body, _ := json.Marshal(map[string]any{"model": c.model, "state": state, "questions": questions}) + var last error + for attempt := 0; attempt < 2; attempt++ { + req, _ := http.NewRequestWithContext(ctx, "POST", c.base+"/v1/systemone", bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(key)) + req.Header.Set("Content-Type", "application/json") + resp, err := c.http.Do(req) + if err != nil { + return nil, err + } + rb, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + resp.Body.Close() + if resp.StatusCode == 429 || resp.StatusCode == 529 { + last = fmt.Errorf("typesafe status %d: %s", resp.StatusCode, truncate(string(rb), 200)) + wait := time.Second + if ra := resp.Header.Get("retry-after"); ra != "" { + if d, perr := time.ParseDuration(ra + "s"); perr == nil && d < 5*time.Second { + wait = d + } + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(wait): + } + continue + } + if resp.StatusCode/100 != 2 { + return nil, fmt.Errorf("typesafe status %d: %s", resp.StatusCode, truncate(string(rb), 300)) + } + var out Response + if err := json.Unmarshal(rb, &out); err != nil { + return nil, fmt.Errorf("typesafe decode: %w", err) + } + return &out, nil + } + return nil, last +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "..." +} diff --git a/ai/internal/jev/client_test.go b/ai/internal/jev/client_test.go new file mode 100644 index 0000000..37fbf41 --- /dev/null +++ b/ai/internal/jev/client_test.go @@ -0,0 +1,43 @@ +package jev + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestAskRoundTrip(t *testing.T) { + var got map[string]any + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/systemone" || r.Header.Get("Authorization") != "Bearer k1" { + t.Errorf("bad request: %s %s", r.URL.Path, r.Header.Get("Authorization")) + } + _ = json.NewDecoder(r.Body).Decode(&got) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"model":"jev-1.13.0","answers":{"recovering":{"type":"noul","noul":0.91},"label":{"type":"choice","choice":"db_primary_down","probabilities":{"db_primary_down":0.8,"unknown":0.2},"confidence":0.6}},"usage":{"input_tokens":120,"output_tokens":4}}`)) + })) + defer srv.Close() + c := New(srv.URL, "", func() (string, error) { return "k1", nil }) + resp, err := c.Ask(context.Background(), map[string]any{"a": 1}, map[string]Question{ + "recovering": {Type: "noul", Instructions: "q1"}, + "label": {Type: "choice", Instructions: "q2", Criteria: map[string]string{"db_primary_down": "x", "unknown": "y"}}, + }) + if err != nil { + t.Fatal(err) + } + if got["model"] != "jev-1.13.0" || got["questions"].(map[string]any)["label"].(map[string]any)["type"] != "choice" { + t.Errorf("request body: %v", got) + } + if *resp.Answers["recovering"].Noul != 0.91 || resp.Answers["label"].Choice != "db_primary_down" || resp.Usage.InputTokens != 120 { + t.Errorf("response: %+v", resp) + } +} + +func TestAskNoKey(t *testing.T) { + c := New("http://127.0.0.1:1", "", func() (string, error) { return "", nil }) + if _, err := c.Ask(context.Background(), "s", nil); err != ErrNoKey { + t.Errorf("want ErrNoKey, got %v", err) + } +} diff --git a/ai/internal/rag/store.go b/ai/internal/rag/store.go index bafaeec..ce006df 100644 --- a/ai/internal/rag/store.go +++ b/ai/internal/rag/store.go @@ -68,6 +68,22 @@ CREATE TABLE IF NOT EXISTS ai.hitl_messages ( result jsonb, created_at timestamptz NOT NULL DEFAULT now() ); +CREATE TABLE IF NOT EXISTS ai.jev_shadow ( + id bigserial PRIMARY KEY, + session_id text, + run_id text, + scenario_id text, + gate text NOT NULL, + model text, + state jsonb NOT NULL, + questions jsonb NOT NULL, + answers jsonb, + llm jsonb, + latency_ms bigint, + input_tokens integer, + error text, + created_at timestamptz NOT NULL DEFAULT now() +); -- additive migrations (CREATE TABLE IF NOT EXISTS leaves existing tables untouched; seen 2026-10-08 on the AI VM) ALTER TABLE ai.hitl_messages ADD COLUMN IF NOT EXISTS diagnosis jsonb NOT NULL DEFAULT '{}'::jsonb;` @@ -371,3 +387,38 @@ func (s *Store) Pending(ctx context.Context) ([]HitlMessage, error) { } return out, rows.Err() } + +// ---- Jev shadow gates (2026-10-08) ---- + +type JevShadow struct { + SessionID string + RunID *string + ScenarioID *string + Gate string + Model string + State any + Questions any + Answers any + LLM any + LatencyMs int64 + InputTokens int + Error string +} + +func (s *Store) SaveJev(ctx context.Context, r JevShadow) error { + st, _ := json.Marshal(r.State) + q, _ := json.Marshal(r.Questions) + an, _ := json.Marshal(r.Answers) + llm, _ := json.Marshal(r.LLM) + _, err := s.pool.Exec(ctx, `INSERT INTO ai.jev_shadow (session_id,run_id,scenario_id,gate,model,state,questions,answers,llm,latency_ms,input_tokens,error) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,NULLIF($12,''))`, + nullIfEmpty(r.SessionID), r.RunID, r.ScenarioID, r.Gate, nullIfEmpty(r.Model), st, q, an, llm, r.LatencyMs, r.InputTokens, r.Error) + return err +} + +func nullIfEmpty(s string) *string { + if s == "" { + return nil + } + return &s +}