From 2a2f91a32cb591557681490f9e25f4957c28c6df Mon Sep 17 00:00:00 2001 From: seonNoh Date: Fri, 16 Oct 2026 22:22:33 +0900 Subject: [PATCH] fix(ai): hand an unanswered external turn out again after 3 minutes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 참고 측정 1세트 첫 진단에서 클라이언트가 /external/next 로 가져간 2번째 턴의 출력을 잃어 답이 오지 않았고, 턴은 awaiting_reply 에 남아 진단이 데드라인까지 멈췄다. 가져간 뒤 RequeueAfter(기본 3분) 동안 답이 없는 턴은 새 reply_token 으로 큐 앞에 되돌려 다음 폴링이 다시 받게 한다(requeued 횟수 기록, 이전 토큰은 거부). 출력 유실은 재시도 한 번의 비용으로 끝나고 진단은 이어진다. --- ai/internal/external/responder.go | 36 ++++++++++++++++++++---- ai/internal/external/responder_test.go | 39 ++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 5 deletions(-) diff --git a/ai/internal/external/responder.go b/ai/internal/external/responder.go index ab0cc37..37762e6 100644 --- a/ai/internal/external/responder.go +++ b/ai/internal/external/responder.go @@ -33,6 +33,7 @@ const maxConvCached = 64 type Turn struct { TurnID string `json:"turn_id"` ReplyToken string `json:"reply_token"` // random, handed out with the turn, required by /external/reply + Requeued int `json:"requeued"` // times this turn was handed out again after an unanswered take 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 @@ -61,8 +62,9 @@ type replyItem struct { } type pending struct { - turn Turn - reply chan Reply + turn Turn + reply chan Reply + takenAt time.Time // zero while queued } type conversation struct { @@ -74,7 +76,8 @@ type conversation struct { // 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) + TurnTimeout time.Duration // how long one turn waits for the client (bounded by the caller's context as well) + RequeueAfter time.Duration // a taken turn with no reply after this long goes back to the queue with a new token mu sync.Mutex queue []*pending // issued, not yet taken by a client @@ -91,7 +94,7 @@ 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)} + return &Responder{TurnTimeout: turnTimeout, RequeueAfter: 3 * time.Minute, 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. @@ -200,6 +203,27 @@ func argString(raw json.RawMessage) string { return s } +// requeueStale (mu held) returns taken turns whose reply never came (client lost the output, crashed, was replaced) +// to the front of the queue with a fresh reply token, so the next poll serves them again; the old token is void. +func (r *Responder) requeueStale() { + if r.RequeueAfter <= 0 { + return + } + var stale []*pending + for id, p := range r.taken { + if time.Since(p.takenAt) >= r.RequeueAfter { + delete(r.taken, id) + p.turn.ReplyToken = randHex(16) + p.turn.Requeued++ + p.takenAt = time.Time{} + stale = append(stale, p) + } + } + if len(stale) > 0 { + r.queue = append(stale, r.queue...) + } +} + func (r *Responder) drop(id string) { r.mu.Lock() defer r.mu.Unlock() @@ -223,9 +247,11 @@ func (r *Responder) Next(w http.ResponseWriter, req *http.Request) { for { r.mu.Lock() r.lastPoll = time.Now() + r.requeueStale() if len(r.queue) > 0 { p := r.queue[0] r.queue = r.queue[1:] + p.takenAt = time.Now() r.taken[p.turn.TurnID] = p r.mu.Unlock() writeJSON(w, 200, p.turn) @@ -300,7 +326,7 @@ func (r *Responder) ReplyHandler(w http.ResponseWriter, req *http.Request) { 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())} + out := map[string]any{"queued": len(r.queue), "awaiting_reply": len(r.taken), "turn_timeout_s": int(r.TurnTimeout.Seconds()), "requeue_after_s": int(r.RequeueAfter.Seconds())} if !r.lastPoll.IsZero() { out["last_poll_age_s"] = int(time.Since(r.lastPoll).Seconds()) } diff --git a/ai/internal/external/responder_test.go b/ai/internal/external/responder_test.go index d9c9cfb..558327b 100644 --- a/ai/internal/external/responder_test.go +++ b/ai/internal/external/responder_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -137,6 +138,44 @@ func TestDeadlinesAndErrors(t *testing.T) { } } +// 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++ {