1package agent
2
3import (
4 "encoding/json"
5 "net/http"
6 "net/http/httptest"
7 "strings"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "time"
12
13 "github.com/bnema/zut/packages/agent/subagents"
14 "github.com/bnema/zut/packages/provider"
15)
16
17// A follow-up steered while the child waits on the provider reaches the next
18// model request of the same turn and is persisted as a user message, so the
19// child reads it without waiting for its turn to end (issue 226).
20func TestResidentAgentRuntimeDeliversSteerMidTurn(t *testing.T) {
21 firstRequest := make(chan struct{})
22 releaseFirst := make(chan struct{})
23 var requests atomic.Int32
24 var mu sync.Mutex
25 var bodies []string
26 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
27 number := requests.Add(1)
28 var body map[string]any
29 if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
30 t.Errorf("decode request: %v", err)
31 }
32 encoded, _ := json.Marshal(body["messages"])
33 mu.Lock()
34 bodies = append(bodies, string(encoded))
35 mu.Unlock()
36 w.Header().Set("content-type", "text/event-stream")
37 if number == 1 {
38 close(firstRequest)
39 <-releaseFirst
40 writeOpenAIChunk(t, w, map[string]any{
41 "choices": []any{map[string]any{"index": 0, "delta": map[string]any{"tool_calls": []any{map[string]any{
42 "index": 0, "id": "call-1", "type": "function",
43 "function": map[string]any{"name": "read", "arguments": `{"path":"missing.txt"}`},
44 }}}, "finish_reason": "tool_calls"}},
45 })
46 return
47 }
48 writeOpenAIChunk(t, w, map[string]any{
49 "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "wrapped up"}, "finish_reason": "stop"}},
50 })
51 }))
52 defer server.Close()
53
54 runtime, journal, _ := newResidentChildTestRunner(t, "steer-child", Args{
55 Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: server.URL,
56 CWD: t.TempDir(), NoContextFiles: true, NoSkill: true,
57 })
58 if runtime.Steer("too early") {
59 t.Fatal("Steer accepted a follow-up with no running turn")
60 }
61 done := make(chan error, 1)
62 go func() { done <- runtime.Run(t.Context(), "investigate") }()
63 select {
64 case <-firstRequest:
65 case <-time.After(5 * time.Second):
66 t.Fatal("first provider request did not arrive")
67 }
68 if !runtime.Steer("stop investigating and report now") {
69 t.Fatal("Steer rejected a follow-up during a running turn")
70 }
71 if got := runtime.PendingSteers(); got != 1 {
72 t.Fatalf("pending steers = %d, want 1", got)
73 }
74 close(releaseFirst)
75 if err := <-done; err != nil {
76 t.Fatalf("Run: %v", err)
77 }
78 if got := runtime.PendingSteers(); got != 0 {
79 t.Fatalf("pending steers after turn = %d, want 0", got)
80 }
81 mu.Lock()
82 defer mu.Unlock()
83 if len(bodies) != 2 || !strings.Contains(bodies[1], "stop investigating and report now") {
84 t.Fatalf("provider requests = %#v, want the steer in the second request", bodies)
85 }
86 messages, err := subagents.ReadResidentTranscriptMessages(journal.Dir())
87 if err != nil {
88 t.Fatal(err)
89 }
90 var users []string
91 for _, message := range messages {
92 if message.Role != provider.RoleUser {
93 continue
94 }
95 for _, content := range message.Content {
96 if text, ok := content.(provider.TextBlock); ok {
97 users = append(users, text.Text)
98 }
99 }
100 }
101 if len(users) != 2 || users[1] != "stop investigating and report now" {
102 t.Fatalf("journaled user messages = %#v", users)
103 }
104}