package main import ( "context" "encoding/json" "log" "net/http" "path/filepath" "strings" "sync" "testing" "time" "github.com/stretchr/testify/require" ) func TestTurnRunsToolsUntilTheFinalAnswer(t *testing.T) { server := newModelFake(t) server.reply = func(w http.ResponseWriter, call ModelFakeCall) { if call.number == 0 { modelWriteEvents(w, modelCompletedEvent([]any{ modelReasoningItem("rs_1", "opaque thought"), modelFunctionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`), }, 100, 0, 5)) return } modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("It said hi.")}, 120, 100, 5)) } test := newTestTurnLoop(t, server) _, err := test.loop.Submit(terminalSource, "Run echo hi.") require.NoError(t, err) test.events.WaitFor(t, "idle", 1) require.Equal(t, 2, server.Calls()) first := server.Call(0).body second := server.Call(1).body instructions, err := test.agent.Prompt() require.NoError(t, err) require.Equal(t, instructions, first["instructions"]) require.Equal(t, test.agent.config.Model, first["model"]) // The second request begins with the whole first one. firstInput := first["input"].([]any) secondInput := second["input"].([]any) require.Equal(t, firstInput, secondInput[:len(firstInput)]) require.Equal(t, []any{ map[string]any{"type": "reasoning", "encrypted_content": "opaque thought", "summary": []any{}}, map[string]any{"type": "function_call", "call_id": "call_1", "name": "shell", "arguments": `{"command":"echo hi"}`}, map[string]any{"type": "function_call_output", "call_id": "call_1", "output": "hi\n"}, }, secondInput[len(firstInput):]) require.Equal(t, []string{ "user: Run echo hi.", `tool: shell {"command":"echo hi"}`, "echo: hi\n", "reply: It said hi.", }, test.Log(t)) require.Equal(t, []string{"user", "busy", "tool", "echo", "reply", "idle"}, test.events.Kinds()) var turns []string for _, record := range modelReadUsage(t, test.usagePath) { turns = append(turns, record.Turn) } require.Equal(t, []string{"first", "next"}, turns) } func TestFirstRequestHoldsTheViewTheStateAndTheMessage(t *testing.T) { server := newModelFake(t) test := newTestTurnLoop(t, server) _, err := test.memory.Append("user", "an earlier message") require.NoError(t, err) _, err = test.loop.Submit(terminalSource, "The new message.") require.NoError(t, err) test.events.WaitFor(t, "idle", 1) require.Equal(t, []string{ "\n0+1|user: an earlier message\n", "\n\nNow: Thursday 2026-01-01 12:00 +00:00", "The new message.", }, turnInputTexts(t, server.Call(0))) } func TestMessageDuringATurnJoinsTheNextRequest(t *testing.T) { server := newModelFake(t) arrived := make(chan struct{}) release := make(chan struct{}) server.reply = func(w http.ResponseWriter, call ModelFakeCall) { if call.number == 0 { close(arrived) <-release modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5)) return } modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("done")}, 100, 0, 5)) } test := newTestTurnLoop(t, server) _, err := test.loop.Submit(terminalSource, "first") require.NoError(t, err) <-arrived _, err = test.loop.Submit(terminalSource, "second, during the turn") require.NoError(t, err) close(release) test.events.WaitFor(t, "idle", 1) input := server.Call(1).body["input"].([]any) require.Equal(t, map[string]any{ "role": "user", "content": []any{map[string]any{"type": "input_text", "text": "second, during the turn"}}, }, input[len(input)-1]) time.Sleep(50 * time.Millisecond) require.Equal(t, 2, server.Calls()) require.Equal(t, 1, test.events.Count("busy")) } func TestMessageDuringTheFinalRequestStartsANewTurn(t *testing.T) { server := newModelFake(t) arrived := make(chan struct{}) release := make(chan struct{}) server.reply = func(w http.ResponseWriter, call ModelFakeCall) { if call.number == 0 { close(arrived) <-release } modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("answer")}, 100, 0, 5)) } test := newTestTurnLoop(t, server) _, err := test.loop.Submit(terminalSource, "first") require.NoError(t, err) <-arrived _, err = test.loop.Submit(terminalSource, "second") require.NoError(t, err) close(release) test.events.WaitFor(t, "idle", 2) require.Equal(t, 2, server.Calls()) texts := turnInputTexts(t, server.Call(1)) require.Equal(t, "second", texts[len(texts)-1]) require.Contains(t, strings.Join(texts, ""), "|reply: answer\n") } func TestTurnWaitsForTheSummaries(t *testing.T) { server := newModelFake(t) summarizer := newFakeSummarizer() summarizer.gate = make(chan struct{}) test := newTestTurnLoopWith(t, server, summarizer) _, err := test.memory.Append("echo", strings.Repeat("long output ", 100)) require.NoError(t, err) _, err = test.loop.Submit(terminalSource, "go on") require.NoError(t, err) time.Sleep(50 * time.Millisecond) require.Equal(t, 0, server.Calls()) close(summarizer.gate) test.events.WaitFor(t, "idle", 1) require.Equal(t, 1, server.Calls()) require.NotContains(t, turnInputTexts(t, server.Call(0))[0], viewPending) } func TestInterruptEndsTheTurn(t *testing.T) { server := newModelFake(t) arrived := make(chan struct{}) release := make(chan struct{}) server.reply = func(w http.ResponseWriter, call ModelFakeCall) { if call.number == 0 { close(arrived) <-release } modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("answer")}, 100, 0, 5)) } test := newTestTurnLoop(t, server) _, err := test.loop.Submit(terminalSource, "a long task") require.NoError(t, err) <-arrived require.True(t, test.loop.Interrupt()) test.events.WaitFor(t, "idle", 1) close(release) require.Equal(t, "interrupted", test.events.Last("error")) require.Equal(t, []string{"user: a long task"}, test.Log(t)) require.False(t, test.loop.Interrupt()) _, err = test.loop.Submit(terminalSource, "again") require.NoError(t, err) test.events.WaitFor(t, "idle", 2) require.Equal(t, "reply: answer", test.Log(t)[2]) require.Empty(t, test.logs.String()) } func TestFailedCallEndsTheTurnWithoutARetry(t *testing.T) { server := newModelFake(t) server.reply = func(w http.ResponseWriter, _ ModelFakeCall) { http.Error(w, "busy", http.StatusServiceUnavailable) } test := newTestTurnLoop(t, server) _, err := test.loop.Submit(terminalSource, "hello") require.NoError(t, err) test.events.WaitFor(t, "idle", 1) time.Sleep(50 * time.Millisecond) require.Equal(t, 1, server.Calls()) require.Contains(t, test.events.Last("error"), "HTTP 503") require.Equal(t, 1, strings.Count(test.logs.String(), "\n")) require.Contains(t, test.logs.String(), "turn failed: ") require.Contains(t, test.logs.String(), "HTTP 503") require.NotContains(t, test.logs.String(), "hello") } func TestViewMessagesOnlyGrowBetweenTurns(t *testing.T) { server := newModelFake(t) test := newTestTurnLoop(t, server) for n, text := range []string{"one", "two", "three"} { _, err := test.loop.Submit(terminalSource, text) require.NoError(t, err) test.events.WaitFor(t, "idle", n+1) } first := server.Call(1).body["input"].([]any) second := server.Call(2).body["input"].([]any) require.Len(t, first, 4) require.Len(t, second, 5) require.Equal(t, first[:2], second[:2]) require.Equal(t, []string{ "\n", "0+1|user: one\n1+1|reply: a line\n", "2+1|user: two\n3+1|reply: a line\n", "\n\nNow: Thursday 2026-01-01 12:00 +00:00", "three", }, turnInputTexts(t, server.Call(2))) } // TurnTestLoop is a loop that runs in the background of a test, with its parts. type TurnTestLoop struct { agent *Agent memory *Memory loop *TurnLoop events *TurnEventRecorder logs *SyncBuffer usagePath string } // Log returns the logged messages as "kind: text". func (l *TurnTestLoop) Log(t *testing.T) []string { t.Helper() var lines []string for id := range l.memory.Len() { message, err := l.memory.Message(id) require.NoError(t, err) lines = append(lines, message.Kind+": "+message.Text) } return lines } // TurnEventRecorder keeps the events of a loop. type TurnEventRecorder struct { mutex sync.Mutex events []Event } // WaitFor waits until count events of kind came. func (r *TurnEventRecorder) WaitFor(t *testing.T, kind string, count int) { t.Helper() require.Eventually(t, func() bool { return r.Count(kind) >= count }, 10*time.Second, time.Millisecond) } func (r *TurnEventRecorder) Kinds() []string { r.mutex.Lock() defer r.mutex.Unlock() kinds := make([]string, len(r.events)) for n, event := range r.events { kinds[n] = event.Kind } return kinds } func (r *TurnEventRecorder) Count(kind string) int { count := 0 for _, each := range r.Kinds() { if each == kind { count++ } } return count } // Last returns the text of the last event of kind. func (r *TurnEventRecorder) Last(kind string) string { return r.LastEvent(kind).Text } // LastEvent returns the last event of kind. func (r *TurnEventRecorder) LastEvent(kind string) Event { r.mutex.Lock() defer r.mutex.Unlock() for n := len(r.events) - 1; n >= 0; n-- { if r.events[n].Kind == kind { return r.events[n] } } return Event{} } func (r *TurnEventRecorder) Add(event Event) { r.mutex.Lock() defer r.mutex.Unlock() r.events = append(r.events, event) } func newTestTurnLoop(t *testing.T, server *ModelFake) *TurnTestLoop { t.Helper() return newTestTurnLoopWith(t, server, newFakeSummarizer()) } // turnInputTexts returns the text of each user message of the input of call. func turnInputTexts(t *testing.T, call ModelFakeCall) []string { t.Helper() var body struct { Input []struct { Role string `json:"role"` Content []struct { Text string `json:"text"` } `json:"content"` } `json:"input"` } require.NoError(t, json.Unmarshal(call.raw, &body)) var texts []string for _, item := range body.Input { if item.Role != "user" { continue } var text strings.Builder for _, part := range item.Content { text.WriteString(part.Text) } texts = append(texts, text.String()) } return texts } // newTestTurnLoopWith returns a running loop on a new agent whose builds use // summarizer and whose commands run in a temporary directory. func newTestTurnLoopWith(t *testing.T, server *ModelFake, summarizer Summarizer) *TurnTestLoop { t.Helper() agent, err := openAgent(t.TempDir()) require.NoError(t, err) agent.config.Workdir = t.TempDir() clock := newTestClock() memory, err := openMemory(t.Context(), t.TempDir(), clock.Now, summarizer, discardLog()) require.NoError(t, err) test := &TurnTestLoop{ agent: agent, memory: memory, events: &TurnEventRecorder{}, logs: &SyncBuffer{}, usagePath: filepath.Join(t.TempDir(), "usage.jsonl"), } client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), test.usagePath) whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), clock) test.loop, err = newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, clock.Now), client, clock.Now, log.New(test.logs, "", 0), test.events.Add) require.NoError(t, err) ctx, cancel := context.WithCancel(t.Context()) done := make(chan error) go func() { done <- test.loop.Run(ctx) }() t.Cleanup(func() { cancel() require.ErrorIs(t, <-done, context.Canceled) require.NoError(t, memory.Close()) }) return test }