feat(ai): external responder for external-* model ids (reference runs)
모델 id가 external- 로 시작하면 ai-agent 가 LLM API 를 부르지 않고 각 턴(시스템 프롬프트·대화 이력·도구 정의·강제 제출 여부)을 GET /external/next 로 내어주고 POST /external/reply 로 답을 받는다. 진단 루프·프롬프트·도구·턴 상한은 다른 모델과 동일하므로, API 경로가 없는 모델(Claude Mythos 5.1 을 Claude Code 서브에이전트가 구동)의 참고 측정에 쓴다. 토큰 사용량은 알 수 없어 0 으로 기록한다. external-* 진단은 EXTERNAL_DIAGNOSE_TIMEOUT_S(기본 1500초), 한 턴은 EXTERNAL_TURN_TIMEOUT_S(기본 1200초)까지 기다리고, 하네스는 같은 모델에 26분을 기다린다. /selfcheck 에 external_responder 상태(대기 턴·마지막 폴링 시각)를 추가.
This commit is contained in:
1 parent
85d99324a1
commit
ed3d41d399
8 files changed
+585
-6
No files matched your search
+26
-1
@@ -7,12 +7,14 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
"os"
|
"os"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/agent"
|
"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/anthropic"
|
||||||
"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/diag"
|
"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/harness"
|
||||||
"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/jev"
|
"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/openai"
|
||||||
@@ -20,6 +22,15 @@ import (
|
|||||||
"github.com/selfheal-lab/selfheal-lab-apps/ai/internal/tools"
|
"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 {
|
func env(k, d string) string {
|
||||||
if v := os.Getenv(k); v != "" {
|
if v := os.Getenv(k); v != "" {
|
||||||
return v
|
return v
|
||||||
@@ -108,6 +119,13 @@ func main() {
|
|||||||
})
|
})
|
||||||
ag.SetAnthropic(anth)
|
ag.SetAnthropic(anth)
|
||||||
log.Printf("anthropic auth: %s", anth.AuthMode())
|
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)
|
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
|
var jc *jev.Client
|
||||||
if jevMode == "shadow" || jevMode == "enforce" {
|
if jevMode == "shadow" || jevMode == "enforce" {
|
||||||
@@ -148,6 +166,7 @@ func main() {
|
|||||||
if f := anth.Federation(); f != nil {
|
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["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
|
out["jev_mode"] = jevMode
|
||||||
if jc != nil {
|
if jc != nil {
|
||||||
out["jev_key_present"] = jc.KeyPresent()
|
out["jev_key_present"] = jc.KeyPresent()
|
||||||
@@ -180,7 +199,11 @@ func main() {
|
|||||||
}
|
}
|
||||||
inflight++
|
inflight++
|
||||||
defer func() { 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()
|
defer cancel()
|
||||||
prefetched := ag.Prefetch(c)
|
prefetched := ag.Prefetch(c)
|
||||||
tri := ag.ShadowTriage(c, req, prefetched) // nil unless JEV_MODE=shadow; recorded only
|
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)
|
ag.SaveShadow(context.Background(), req, res.SessionID, res, tri, rev)
|
||||||
writeJSON(w, 200, res)
|
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}
|
srv := &http.Server{Addr: env("AGENT_ADDR", ":8080"), Handler: logging(mux), ReadHeaderTimeout: 10 * time.Second}
|
||||||
log.Fatal(srv.ListenAndServe())
|
log.Fatal(srv.ListenAndServe())
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -45,6 +45,7 @@ type Agent struct {
|
|||||||
cfg Config
|
cfg Config
|
||||||
oai *openai.Client
|
oai *openai.Client
|
||||||
anth Responder // nil = claude-* models unavailable
|
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
|
tools *tools.Runner
|
||||||
harness *harness.Client
|
harness *harness.Client
|
||||||
store *rag.Store
|
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).
|
// SetAnthropic enables claude-* model ids (Claude Messages API adapter).
|
||||||
func (a *Agent) SetAnthropic(c Responder) { a.anth = c }
|
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) {
|
func (a *Agent) provider(model string) (Responder, error) {
|
||||||
if strings.HasPrefix(model, "claude-") {
|
if strings.HasPrefix(model, "claude-") {
|
||||||
if a.anth == nil {
|
if a.anth == nil {
|
||||||
@@ -82,6 +90,12 @@ func (a *Agent) provider(model string) (Responder, error) {
|
|||||||
}
|
}
|
||||||
return a.anth, nil
|
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
|
return a.oai, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -127,3 +127,25 @@ func TestDiagnoseRoutesClaudeModel(t *testing.T) {
|
|||||||
t.Fatalf("usage: %v", res.Usage)
|
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
|
||||||
|
}
|
||||||
Vendored
+302
@@ -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=<s>: 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)
|
||||||
|
}
|
||||||
+185
@@ -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) }
|
||||||
+16
-3
@@ -10,6 +10,7 @@ import (
|
|||||||
"log"
|
"log"
|
||||||
"net/http"
|
"net/http"
|
||||||
"sort"
|
"sort"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"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.
|
// 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) {
|
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}
|
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 != "" {
|
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)
|
b, _ := json.Marshal(reqBody)
|
||||||
req, _ := http.NewRequestWithContext(ctx, "POST", r.app.cfg.AIDiagnoseURL, bytes.NewReader(b))
|
req, _ := http.NewRequestWithContext(ctx, "POST", r.app.cfg.AIDiagnoseURL, bytes.NewReader(b))
|
||||||
req.Header.Set("Content-Type", "application/json")
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
|||||||
@@ -44,3 +44,20 @@ func TestDedupeSameAlertName(t *testing.T) {
|
|||||||
t.Fatal("dedupe off")
|
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")
|
||||||
|
}
|
||||||
|
}
|
||||||
+2
-1
@@ -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
|
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}$`)
|
var modelRe = regexp.MustCompile(`^[A-Za-z0-9._:-]{0,64}$`)
|
||||||
|
|
||||||
func NewRunManager(a *App) *RunManager { return &RunManager{app: a} }
|
func NewRunManager(a *App) *RunManager { return &RunManager{app: a} }
|
||||||
|
|||||||
Reference in new issue
Block a user