selfheal-lab-apps/ai/internal/external/responder_test.go

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)
	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)
	}
}

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) }