seon
2a2f91a32c
참고 측정 1세트 첫 진단에서 클라이언트가 /external/next 로 가져간 2번째 턴의 출력을 잃어 답이 오지 않았고, 턴은 awaiting_reply 에 남아 진단이 데드라인까지 멈췄다. 가져간 뒤 RequeueAfter(기본 3분) 동안 답이 없는 턴은 새 reply_token 으로 큐 앞에 되돌려 다음 폴링이 다시 받게 한다(requeued 횟수 기록, 이전 토큰은 거부). 출력 유실은 재시도 한 번의 비용으로 끝나고 진단은 이어진다.
223 lines
9.4 KiB
Go
223 lines
9.4 KiB
Go
package external
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"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)
|
|
srv := httptest.NewServer(r.Handler())
|
|
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 !strings.HasPrefix(turn.TurnID, "ext_") || len(turn.TurnID) != 20 || len(turn.ReplyToken) != 32 || 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)
|
|
}
|
|
// a reply without the turn's token is refused (the turn stays open)
|
|
if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn.TurnID+`","reply_token":"wrong","output":[{"type":"message","text":"x"}]}`); code != 403 {
|
|
t.Fatalf("wrong token must be 403, got %d", code)
|
|
}
|
|
code, body := post(t, srv.URL+"/external/reply", `{"turn_id":"`+turn.TurnID+`","reply_token":"`+turn.ReplyToken+`","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 != turn.TurnID || 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: turn.TurnID,
|
|
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 == turn.TurnID || turn2.Session != turn.TurnID || 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":"`+turn2.TurnID+`","reply_token":"`+turn2.ReplyToken+`","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":"`+turn2.TurnID+`","reply_token":"`+turn2.ReplyToken+`","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)
|
|
srv := httptest.NewServer(r2.Handler())
|
|
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+`","reply_token":"`+turn.ReplyToken+`","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)
|
|
}
|
|
}
|
|
|
|
// A taken turn whose reply never comes is handed out again after RequeueAfter with a new token; the old token is
|
|
// refused and the new one completes the turn.
|
|
func TestRequeueUnansweredTurn(t *testing.T) {
|
|
r := New(10 * time.Second)
|
|
r.RequeueAfter = 200 * time.Millisecond
|
|
srv := httptest.NewServer(r.Handler())
|
|
defer srv.Close()
|
|
ch := make(chan error, 1)
|
|
go func() {
|
|
resp, err := r.CreateResponse(context.Background(), openai.Request{Model: "external-x", Input: []openai.Item{{Type: "message", Role: "user", Content: "hi"}}})
|
|
if err == nil && len(resp.FunctionCalls()) != 1 {
|
|
err = fmt.Errorf("unexpected response %+v", resp)
|
|
}
|
|
ch <- err
|
|
}()
|
|
first := poll(t, srv.URL) // taken, never answered
|
|
if first.Requeued != 0 {
|
|
t.Fatalf("fresh turn marked requeued: %+v", first)
|
|
}
|
|
if code := get(t, srv.URL+"/external/next?wait=0"); code != 204 {
|
|
t.Fatalf("taken turn must not be served again at once, got %d", code)
|
|
}
|
|
time.Sleep(250 * time.Millisecond)
|
|
second := poll(t, srv.URL)
|
|
if second.TurnID != first.TurnID || second.ReplyToken == first.ReplyToken || second.Requeued != 1 {
|
|
t.Fatalf("requeue: first=%+v second=%+v", first, second)
|
|
}
|
|
if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"`+first.TurnID+`","reply_token":"`+first.ReplyToken+`","output":[{"type":"function_call","name":"x","arguments":{}}]}`); code != 403 {
|
|
t.Fatalf("old token must be refused, got %d", code)
|
|
}
|
|
if code, _ := post(t, srv.URL+"/external/reply", `{"turn_id":"`+second.TurnID+`","reply_token":"`+second.ReplyToken+`","output":[{"type":"function_call","name":"x","arguments":{}}]}`); code != 200 {
|
|
t.Fatalf("new token must be accepted, got %d", code)
|
|
}
|
|
if err := <-ch; err != nil {
|
|
t.Fatal(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) }
|