diff --git a/ai/cmd/agent/main.go b/ai/cmd/agent/main.go index 7f5d8d1..b275567 100644 --- a/ai/cmd/agent/main.go +++ b/ai/cmd/agent/main.go @@ -7,12 +7,14 @@ import ( "log" "net/http" "os" + "strconv" "strings" "time" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/agent" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/anthropic" "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/diag" + "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/external" "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" @@ -20,6 +22,15 @@ import ( "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/tools" ) +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 +} + func env(k, d string) string { if v := os.Getenv(k); v != "" { return v @@ -108,6 +119,13 @@ func main() { }) ag.SetAnthropic(anth) log.Printf("anthropic auth: %s", anth.AuthMode()) + // external-* model ids (reference runs, 2026-10-16): each loop turn is served to an outside client over + // GET /external/next and answered through POST /external/reply (internal/external). The diagnosis deadline is + // EXTERNAL_DIAGNOSE_TIMEOUT_S (the harness waits a little longer for external-* models), one turn at most + // EXTERNAL_TURN_TIMEOUT_S. The endpoints are reachable only inside the VPC / over the IAP SSH tunnel. + ext := external.New(time.Duration(envInt("EXTERNAL_TURN_TIMEOUT_S", 1200)) * time.Second) + ag.SetExternal(ext) + externalDiagnoseTimeout := time.Duration(envInt("EXTERNAL_DIAGNOSE_TIMEOUT_S", 1500)) * time.Second 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" || jevMode == "enforce" { @@ -148,6 +166,7 @@ func main() { if f := anth.Federation(); f != nil { out["anthropic_federation"] = f.Probe(c, anth.BaseURL) // one real token exchange; the Console history page shows the deny reason on failure } + out["external_responder"] = ext.Status() out["jev_mode"] = jevMode if jc != nil { out["jev_key_present"] = jc.KeyPresent() @@ -180,7 +199,11 @@ func main() { } inflight++ defer func() { inflight-- }() - c, cancel := context.WithTimeout(context.Background(), 170*time.Second) // harness waits 3 minutes + timeout := 170 * time.Second // harness waits 3 minutes + if agent.IsExternal(req.Model) { + timeout = externalDiagnoseTimeout + } + c, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() prefetched := ag.Prefetch(c) tri := ag.ShadowTriage(c, req, prefetched) // nil unless JEV_MODE=shadow; recorded only @@ -199,6 +222,8 @@ func main() { ag.SaveShadow(context.Background(), req, res.SessionID, res, tri, rev) writeJSON(w, 200, res) }) + mux.HandleFunc("GET /external/next", ext.Next) + mux.HandleFunc("POST /external/reply", ext.ReplyHandler) srv := &http.Server{Addr: env("AGENT_ADDR", ":8080"), Handler: logging(mux), ReadHeaderTimeout: 10 * time.Second} log.Fatal(srv.ListenAndServe()) } diff --git a/ai/internal/agent/agent.go b/ai/internal/agent/agent.go index 81aae84..16df396 100644 --- a/ai/internal/agent/agent.go +++ b/ai/internal/agent/agent.go @@ -45,6 +45,7 @@ type Agent struct { cfg Config oai *openai.Client anth Responder // nil = claude-* models unavailable + ext Responder // nil = external-* models unavailable (internal/external: a client outside the VM answers each turn) tools *tools.Runner harness *harness.Client store *rag.Store @@ -74,7 +75,14 @@ func New(cfg Config, oai *openai.Client, tr *tools.Runner, hc *harness.Client, s // SetAnthropic enables claude-* model ids (Claude Messages API adapter). func (a *Agent) SetAnthropic(c Responder) { a.anth = c } -// provider picks the LLM client for a model id: claude-* -> Anthropic adapter, everything else -> OpenAI. +// SetExternal enables external-* model ids (turns served to an outside client over /external/next, reference runs). +func (a *Agent) SetExternal(c Responder) { a.ext = c } + +// IsExternal reports whether the model id is served by the external responder (longer deadlines apply). +func IsExternal(model string) bool { return strings.HasPrefix(model, "external-") } + +// provider picks the LLM client for a model id: claude-* -> Anthropic adapter, external-* -> external responder, +// everything else -> OpenAI. func (a *Agent) provider(model string) (Responder, error) { if strings.HasPrefix(model, "claude-") { if a.anth == nil { @@ -82,6 +90,12 @@ func (a *Agent) provider(model string) (Responder, error) { } return a.anth, nil } + if IsExternal(model) { + if a.ext == nil { + return nil, fmt.Errorf("model %s: external responder not configured", model) + } + return a.ext, nil + } return a.oai, nil } diff --git a/ai/internal/agent/agent_test.go b/ai/internal/agent/agent_test.go index a0a5b75..c188c36 100644 --- a/ai/internal/agent/agent_test.go +++ b/ai/internal/agent/agent_test.go @@ -127,3 +127,25 @@ func TestDiagnoseRoutesClaudeModel(t *testing.T) { t.Fatalf("usage: %v", res.Usage) } } + +// Model routing: external-* goes to the external responder when one is set, and is refused when none is configured. +func TestProviderRouting(t *testing.T) { + a := New(Config{}, nil, nil, nil, nil, diag.Enums{}, "v") + if _, err := a.provider("external-claude-mythos-5-1"); err == nil || !strings.Contains(err.Error(), "external responder not configured") { + t.Fatalf("expected unconfigured error, got %v", err) + } + fake := &fakeResponder{} + a.SetExternal(fake) + if p, err := a.provider("external-claude-mythos-5-1"); err != nil || p != Responder(fake) { + t.Fatalf("external routing: %v %v", p, err) + } + if !IsExternal("external-x") || IsExternal("claude-x") || IsExternal("gpt-6-luna") { + t.Fatal("IsExternal prefix check") + } +} + +type fakeResponder struct{} + +func (*fakeResponder) CreateResponse(context.Context, openai.Request) (*openai.Response, error) { + return nil, nil +} diff --git a/ai/internal/external/responder.go b/ai/internal/external/responder.go new file mode 100644 index 0000000..8238c8f --- /dev/null +++ b/ai/internal/external/responder.go @@ -0,0 +1,302 @@ +// Package external lets a client outside the AI VM act as the LLM for model ids starting with "external-": the +// diagnosis loop (internal/agent) is unchanged, but every turn is queued as a Turn and served to the client over +// GET /external/next (long poll); the client answers with POST /external/reply. Used for reference measurements with +// a model that has no API path (2026-10-16: Claude Mythos 5.1 driven by a Claude Code subagent). Token usage is not +// known for such turns and is reported as zero; reports mark those runs as reference records. +package external + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai" +) + +const maxConvCached = 64 + +// Turn is one request to the external client: the whole conversation so far (the loop sends the system prompt only +// on the first turn, so it is carried along), the tool definitions and whether the final submit_diagnosis is forced. +type Turn struct { + TurnID string `json:"turn_id"` + Session string `json:"session"` // first response id of the conversation ("" on the first turn) + Model string `json:"model"` + Turn int `json:"turn"` // 1-based index within the conversation + System string `json:"system"` + Input []openai.Item `json:"input"` // user messages, assistant function_call items, function_call_output items + Tools []openai.Tool `json:"tools"` + ForceSubmit bool `json:"force_submit"` // the loop asked for submit_diagnosis now (last turn) + MaxOutputTokens int `json:"max_output_tokens"` + IssuedAt time.Time `json:"issued_at"` + DeadlineAt time.Time `json:"deadline_at"` +} + +// Reply is the client's answer: function calls (name + arguments) and/or assistant text. +type Reply struct { + TurnID string `json:"turn_id"` + Output []replyItem `json:"output"` + Error string `json:"error,omitempty"` // client gave up on this turn +} + +type replyItem struct { + Type string `json:"type"` // function_call | message + Name string `json:"name,omitempty"` + Arguments json.RawMessage `json:"arguments,omitempty"` // JSON object or a JSON string holding one + Text string `json:"text,omitempty"` +} + +type pending struct { + turn Turn + reply chan Reply +} + +type conversation struct { + system string + items []openai.Item + turn int + first string +} + +// Responder implements agent.Responder; one instance serves every external-* diagnosis. +type Responder struct { + TurnTimeout time.Duration // how long one turn waits for the client (bounded by the caller's context as well) + + mu sync.Mutex + queue []*pending // issued, not yet taken by a client + taken map[string]*pending // taken, waiting for the reply + conv map[string]conversation + order []string + seq int + notify chan struct{} + lastPoll time.Time + lastTurn time.Time +} + +func New(turnTimeout time.Duration) *Responder { + if turnTimeout <= 0 { + turnTimeout = 20 * time.Minute + } + return &Responder{TurnTimeout: turnTimeout, taken: map[string]*pending{}, conv: map[string]conversation{}, notify: make(chan struct{}, 1)} +} + +// CreateResponse queues the turn and blocks until the client replies, the context ends or TurnTimeout passes. +func (r *Responder) CreateResponse(ctx context.Context, req openai.Request) (*openai.Response, error) { + var hist conversation + if req.PreviousResponseID != "" { + r.mu.Lock() + prev, ok := r.conv[req.PreviousResponseID] + r.mu.Unlock() + if !ok { + return nil, fmt.Errorf("external: unknown previous_response_id %q", req.PreviousResponseID) + } + hist = conversation{system: prev.system, items: append([]openai.Item{}, prev.items...), turn: prev.turn, first: prev.first} + } + for _, it := range req.Input { + if it.Type == "message" && it.Role == "system" { + if s, _ := it.Content.(string); s != "" { + if hist.system != "" { + hist.system += "\n\n" + } + hist.system += s + } + continue + } + hist.items = append(hist.items, it) + } + if len(hist.items) == 0 { + return nil, fmt.Errorf("external: no input items") + } + forced := false + if m, ok := req.ToolChoice.(map[string]any); ok && m["type"] == "function" { + forced = true + } + r.mu.Lock() + r.seq++ + id := fmt.Sprintf("ext_%d", r.seq) + now := time.Now() + p := &pending{turn: Turn{TurnID: id, Session: hist.first, Model: req.Model, Turn: hist.turn + 1, System: hist.system, Input: hist.items, Tools: req.Tools, + ForceSubmit: forced, MaxOutputTokens: req.MaxOutputTokens, IssuedAt: now, DeadlineAt: now.Add(r.TurnTimeout)}, reply: make(chan Reply, 1)} + r.queue = append(r.queue, p) + r.lastTurn = now + r.mu.Unlock() + select { + case r.notify <- struct{}{}: + default: + } + + var rep Reply + select { + case rep = <-p.reply: + case <-ctx.Done(): + r.drop(id) + return nil, fmt.Errorf("external: turn %s not answered before the diagnosis deadline (%v)", id, ctx.Err()) + case <-time.After(r.TurnTimeout): + r.drop(id) + return nil, fmt.Errorf("external: turn %s not answered within %s", id, r.TurnTimeout) + } + if rep.Error != "" { + return nil, fmt.Errorf("external: client error on turn %s: %s", id, rep.Error) + } + out := &openai.Response{ID: id, Status: "completed", Model: req.Model} + var text strings.Builder + for i, it := range rep.Output { + switch it.Type { + case "function_call": + out.Output = append(out.Output, openai.Item{Type: "function_call", CallID: fmt.Sprintf("%s_call_%d", id, i+1), ID: fmt.Sprintf("%s_call_%d", id, i+1), Name: it.Name, Arguments: argString(it.Arguments), Status: "completed"}) + case "message": + text.WriteString(it.Text) + } + } + if text.Len() > 0 { + out.Output = append([]openai.Item{{Type: "message", Role: "assistant", Content: []any{map[string]any{"type": "output_text", "text": text.String()}}}}, out.Output...) + } + // remember the conversation (assistant output items are replayed to the client on the next turn) + first := hist.first + if first == "" { + first = id + } + items := append(append([]openai.Item{}, hist.items...), out.Output...) + r.mu.Lock() + r.conv[id] = conversation{system: hist.system, items: items, turn: hist.turn + 1, first: first} + r.order = append(r.order, id) + for len(r.order) > maxConvCached { + delete(r.conv, r.order[0]) + r.order = r.order[1:] + } + r.mu.Unlock() + return out, nil +} + +// argString returns the function arguments as the JSON object text (a JSON string value is unwrapped). +func argString(raw json.RawMessage) string { + s := strings.TrimSpace(string(raw)) + if s == "" { + return "{}" + } + if strings.HasPrefix(s, `"`) { + var inner string + if json.Unmarshal(raw, &inner) == nil { + if strings.TrimSpace(inner) == "" { + return "{}" + } + return inner + } + } + return s +} + +func (r *Responder) drop(id string) { + r.mu.Lock() + defer r.mu.Unlock() + delete(r.taken, id) + for i, p := range r.queue { + if p.turn.TurnID == id { + r.queue = append(r.queue[:i], r.queue[i+1:]...) + return + } + } +} + +// Next is GET /external/next?wait=: the oldest queued turn (200), or 204 when none arrived within wait seconds. +func (r *Responder) Next(w http.ResponseWriter, req *http.Request) { + wait := 20 * time.Second + if v, err := strconv.Atoi(req.URL.Query().Get("wait")); err == nil && v >= 0 && v <= 60 { + wait = time.Duration(v) * time.Second + } + deadline := time.NewTimer(wait) + defer deadline.Stop() + for { + r.mu.Lock() + r.lastPoll = time.Now() + if len(r.queue) > 0 { + p := r.queue[0] + r.queue = r.queue[1:] + r.taken[p.turn.TurnID] = p + r.mu.Unlock() + writeJSON(w, 200, p.turn) + return + } + r.mu.Unlock() + select { + case <-r.notify: + case <-deadline.C: + w.WriteHeader(204) + return + case <-req.Context().Done(): + return + } + } +} + +// ReplyHandler is POST /external/reply with a Reply body; 404 when the turn is unknown (already answered, dropped +// after its deadline, or never taken through /external/next). +func (r *Responder) ReplyHandler(w http.ResponseWriter, req *http.Request) { + var rep Reply + if err := json.NewDecoder(http.MaxBytesReader(w, req.Body, 1<<20)).Decode(&rep); err != nil { + writeJSON(w, 400, map[string]string{"error": err.Error()}) + return + } + if rep.Error == "" { + n := 0 + for _, it := range rep.Output { + switch it.Type { + case "function_call": + if it.Name == "" { + writeJSON(w, 400, map[string]string{"error": "function_call without name"}) + return + } + if a := argString(it.Arguments); !json.Valid([]byte(a)) || !strings.HasPrefix(strings.TrimSpace(a), "{") { + writeJSON(w, 400, map[string]string{"error": "function_call arguments must be a JSON object"}) + return + } + n++ + case "message": + n++ + default: + writeJSON(w, 400, map[string]string{"error": "output item type must be function_call or message"}) + return + } + } + if n == 0 { + writeJSON(w, 400, map[string]string{"error": "output must hold at least one function_call or message"}) + return + } + } + r.mu.Lock() + p, ok := r.taken[rep.TurnID] + if ok { + delete(r.taken, rep.TurnID) + } + r.mu.Unlock() + if !ok { + writeJSON(w, 404, map[string]string{"error": "unknown or expired turn_id " + rep.TurnID}) + return + } + p.reply <- rep + writeJSON(w, 200, map[string]any{"ok": true, "turn_id": rep.TurnID}) +} + +// Status is shown by /selfcheck. +func (r *Responder) Status() map[string]any { + r.mu.Lock() + defer r.mu.Unlock() + out := map[string]any{"queued": len(r.queue), "awaiting_reply": len(r.taken), "turn_timeout_s": int(r.TurnTimeout.Seconds())} + if !r.lastPoll.IsZero() { + out["last_poll_age_s"] = int(time.Since(r.lastPoll).Seconds()) + } + if !r.lastTurn.IsZero() { + out["last_turn_age_s"] = int(time.Since(r.lastTurn).Seconds()) + } + return out +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(v) +} diff --git a/ai/internal/external/responder_test.go b/ai/internal/external/responder_test.go new file mode 100644 index 0000000..06cd2d2 --- /dev/null +++ b/ai/internal/external/responder_test.go @@ -0,0 +1,185 @@ +package external + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/selfheal-lab/selfheal-lab-apps/ai/internal/openai" +) + +// Two loop turns through the HTTP surface: the client takes turn 1 (system + user), answers with a tool call, +// then turn 2 (with previous_response_id) must replay the assistant call and the function_call_output and carry +// the forced-submit flag; the reply's arguments may be a JSON string or object. +func TestTurnsOverHTTP(t *testing.T) { + r := New(5 * time.Second) + mux := http.NewServeMux() + mux.HandleFunc("GET /external/next", r.Next) + mux.HandleFunc("POST /external/reply", r.ReplyHandler) + srv := httptest.NewServer(mux) + defer srv.Close() + tools := []openai.Tool{{Type: "function", Strict: true, Name: "get_probe_status", Parameters: map[string]any{"type": "object"}}} + type res struct { + resp *openai.Response + err error + } + ch := make(chan res, 1) + go func() { + resp, err := r.CreateResponse(context.Background(), openai.Request{Model: "external-claude-mythos-5-1", Tools: tools, ToolChoice: "auto", MaxOutputTokens: 4000, + Input: []openai.Item{{Type: "message", Role: "system", Content: "SYS"}, {Type: "message", Role: "user", Content: "alert"}}}) + ch <- res{resp, err} + }() + turn := poll(t, srv.URL) + if turn.TurnID != "ext_1" || turn.Session != "" || turn.Turn != 1 || turn.System != "SYS" || len(turn.Input) != 1 || turn.Input[0].Content != "alert" || len(turn.Tools) != 1 || turn.ForceSubmit || turn.MaxOutputTokens != 4000 { + t.Fatalf("turn 1: %+v", turn) + } + // second poll while nothing is queued -> 204 within the wait + if code := get(t, srv.URL+"/external/next?wait=0"); code != 204 { + t.Fatalf("empty queue must answer 204, got %d", code) + } + code, body := post(t, srv.URL+"/external/reply", `{"turn_id":"ext_1","output":[{"type":"message","text":"checking"},{"type":"function_call","name":"get_probe_status","arguments":"{}"}]}`) + if code != 200 { + t.Fatalf("reply 1: %d %s", code, body) + } + r1 := <-ch + if r1.err != nil { + t.Fatal(r1.err) + } + fcs := r1.resp.FunctionCalls() + if r1.resp.ID != "ext_1" || len(fcs) != 1 || fcs[0].Name != "get_probe_status" || fcs[0].Arguments != "{}" || fcs[0].CallID == "" || r1.resp.Usage.TotalTokens != 0 { + t.Fatalf("response 1: %+v", r1.resp) + } + if !strings.Contains(r1.resp.Text(), "checking") { + t.Fatalf("text not carried: %q", r1.resp.Text()) + } + + go func() { + resp, err := r.CreateResponse(context.Background(), openai.Request{Model: "external-claude-mythos-5-1", Tools: tools, PreviousResponseID: "ext_1", + ToolChoice: map[string]any{"type": "function", "name": "submit_diagnosis"}, + Input: []openai.Item{{Type: "function_call_output", CallID: fcs[0].CallID, Output: `{"alb":1}`}}}) + ch <- res{resp, err} + }() + turn2 := poll(t, srv.URL) + if turn2.TurnID != "ext_2" || turn2.Session != "ext_1" || turn2.Turn != 2 || turn2.System != "SYS" || !turn2.ForceSubmit { + t.Fatalf("turn 2 header: %+v", turn2) + } + // history: user alert / assistant text / assistant function_call / function_call_output + if len(turn2.Input) != 4 || turn2.Input[2].Type != "function_call" || turn2.Input[2].CallID != fcs[0].CallID || turn2.Input[3].Type != "function_call_output" || turn2.Input[3].Output != `{"alb":1}` { + t.Fatalf("turn 2 history: %s", mustJSON(turn2.Input)) + } + code, body = post(t, srv.URL+"/external/reply", `{"turn_id":"ext_2","output":[{"type":"function_call","name":"submit_diagnosis","arguments":{"root_cause_label":"x","confidence":0.9}}]}`) + if code != 200 { + t.Fatalf("reply 2: %d %s", code, body) + } + r2 := <-ch + if r2.err != nil { + t.Fatal(r2.err) + } + if got := r2.resp.FunctionCalls(); len(got) != 1 || got[0].Name != "submit_diagnosis" || !strings.Contains(got[0].Arguments, `"root_cause_label":"x"`) { + t.Fatalf("response 2: %+v", r2.resp.Output) + } + // an answered turn cannot be answered twice; malformed replies are refused before touching the queue + if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"ext_2","output":[{"type":"message","text":"again"}]}`); code != 404 { + t.Fatalf("second reply must be 404, got %d", code) + } + if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"ext_9","output":[{"type":"function_call","name":"x","arguments":"[1]"}]}`); code != 400 { + t.Fatalf("non-object arguments must be 400, got %d", code) + } + if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"ext_9","output":[]}`); code != 400 { + t.Fatalf("empty output must be 400, got %d", code) + } + st := r.Status() + if st["queued"] != 0 || st["awaiting_reply"] != 0 { + t.Fatalf("status: %v", st) + } +} + +// Deadlines: the diagnosis context ending drops the queued turn with an error; a client "error" reply surfaces as an +// error; an unknown previous_response_id is refused. +func TestDeadlinesAndErrors(t *testing.T) { + r := New(300 * time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + _, err := r.CreateResponse(ctx, openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}}) + if err == nil || !strings.Contains(err.Error(), "not answered before the diagnosis deadline") { + t.Fatalf("expected deadline error, got %v", err) + } + if st := r.Status(); st["queued"] != 0 { + t.Fatalf("dropped turn still queued: %v", st) + } + _, err = r.CreateResponse(context.Background(), openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}}) + if err == nil || !strings.Contains(err.Error(), "not answered within") { + t.Fatalf("expected turn timeout, got %v", err) + } + // error reply + r2 := New(5 * time.Second) + mux := http.NewServeMux() + mux.HandleFunc("GET /external/next", r2.Next) + mux.HandleFunc("POST /external/reply", r2.ReplyHandler) + srv := httptest.NewServer(mux) + defer srv.Close() + ch := make(chan error, 1) + go func() { + _, err := r2.CreateResponse(context.Background(), openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}}) + ch <- err + }() + turn := poll(t, srv.URL) + post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn.TurnID+`","error":"gave up"}`) + if err := <-ch; err == nil || !strings.Contains(err.Error(), "gave up") { + t.Fatalf("expected client error, got %v", err) + } + _, err = r2.CreateResponse(context.Background(), openai.Request{Model: "external-x", PreviousResponseID: "nope", Input: []openai.Item{{Type: "function_call_output", CallID: "a", Output: "b"}}}) + if err == nil || !strings.Contains(err.Error(), "unknown previous_response_id") { + t.Fatalf("expected unknown id error, got %v", err) + } +} + +func poll(t *testing.T, base string) Turn { + t.Helper() + for i := 0; i < 20; i++ { + resp, err := http.Get(base + "/external/next?wait=1") + if err != nil { + t.Fatal(err) + } + if resp.StatusCode == 200 { + var turn Turn + if err := json.NewDecoder(resp.Body).Decode(&turn); err != nil { + t.Fatal(err) + } + resp.Body.Close() + return turn + } + resp.Body.Close() + } + t.Fatal("no turn served") + return Turn{} +} + +func get(t *testing.T, u string) int { + t.Helper() + resp, err := http.Get(u) + if err != nil { + t.Fatal(err) + } + resp.Body.Close() + return resp.StatusCode +} + +func post(t *testing.T, u, body string) (int, string) { + t.Helper() + resp, err := http.Post(u, "application/json", bytes.NewReader([]byte(body))) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + var b bytes.Buffer + b.ReadFrom(resp.Body) + return resp.StatusCode, b.String() +} + +func mustJSON(v any) string { b, _ := json.Marshal(v); return string(b) } diff --git a/harness/alerts.go b/harness/alerts.go index 1b24115..091515c 100644 --- a/harness/alerts.go +++ b/harness/alerts.go @@ -10,6 +10,7 @@ import ( "log" "net/http" "sort" + "strings" "sync" "time" ) @@ -226,14 +227,26 @@ func (r *AlertRouter) cancelTimer(fp string) { } } +// diagnoseTimeout bounds one /diagnose call: 3 minutes for API models (the AI VM gives up at 170 s); external-* +// models (reference runs answered by a client outside the VM, apps 2026-10-16) get 26 minutes, one more than the AI +// VM's EXTERNAL_DIAGNOSE_TIMEOUT_S default, so the VM's own error reaches the diagnosis event first. +func diagnoseTimeout(model string) time.Duration { + if strings.HasPrefix(model, "external-") { + return 26 * time.Minute + } + return 3 * time.Minute +} + // diagnose asks the AI VM (design 6.3 strict schema). Until WP-12 the endpoint is absent; the failure is recorded as an event. func (r *AlertRouter) diagnose(alert map[string]any, runID, sid *string) { - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) - defer cancel() reqBody := map[string]any{"alert": alert, "run_id": runID, "scenario_id": sid} + model := "" if cur := r.app.runs.current(); cur != nil && runID != nil && cur.RunID == *runID && cur.Model != "" { - reqBody["model"] = cur.Model // campaign-level model selection (the AI VM falls back to its default when absent) + model = cur.Model + reqBody["model"] = model // campaign-level model selection (the AI VM falls back to its default when absent) } + ctx, cancel := context.WithTimeout(context.Background(), diagnoseTimeout(model)) + defer cancel() b, _ := json.Marshal(reqBody) req, _ := http.NewRequestWithContext(ctx, "POST", r.app.cfg.AIDiagnoseURL, bytes.NewReader(b)) req.Header.Set("Content-Type", "application/json") diff --git a/harness/alerts_test.go b/harness/alerts_test.go index f799b16..e7ee783 100644 --- a/harness/alerts_test.go +++ b/harness/alerts_test.go @@ -44,3 +44,20 @@ func TestDedupeSameAlertName(t *testing.T) { t.Fatal("dedupe off") } } + +// external-* models (reference runs served by a client outside the AI VM) get the long diagnose timeout; the model id +// itself passes the run request bound. +func TestDiagnoseTimeoutByModel(t *testing.T) { + if d := diagnoseTimeout("claude-sonnet-5-5"); d != 3*time.Minute { + t.Fatalf("api model timeout %v", d) + } + if d := diagnoseTimeout(""); d != 3*time.Minute { + t.Fatalf("default model timeout %v", d) + } + if d := diagnoseTimeout("external-claude-mythos-5-1"); d != 26*time.Minute { + t.Fatalf("external model timeout %v", d) + } + if !modelRe.MatchString("external-claude-mythos-5-1") { + t.Fatal("external model id must pass modelRe") + } +} diff --git a/harness/run.go b/harness/run.go index d9a8e9e..fee15f9 100644 --- a/harness/run.go +++ b/harness/run.go @@ -49,7 +49,8 @@ type RunRequest struct { AutoApproveAfterS *int `json:"auto_approve_after_s"` // measurement mode (model comparison without a human in the loop): nil/0 = off } -// modelRe bounds the model id (empty = AI VM default). The AI VM routes by prefix: claude-* -> Anthropic, else OpenAI. +// modelRe bounds the model id (empty = AI VM default). The AI VM routes by prefix: claude-* -> Anthropic, +// external-* -> external responder (reference runs, longer diagnose timeout in alerts.go), else OpenAI. var modelRe = regexp.MustCompile(`^[A-Za-z0-9._:-]{0,64}$`) func NewRunManager(a *App) *RunManager { return &RunManager{app: a} }