1package subagents
2
3import (
4 "errors"
5 "fmt"
6 "os"
7 "path/filepath"
8 "runtime"
9 "sync"
10 "syscall"
11 "testing"
12 "time"
13)
14
15// shortSocketDir returns a tempdir short enough that we can append
16// "in.sock" and still fit under the unix-socket path cap. The Go
17// runner's t.TempDir() lives under a long /var/folders/... path on
18// macOS which blows the 104-byte limit by itself, so the inbox tests
19// can't use it.
20func shortSocketDir(t *testing.T) string {
21 t.Helper()
22 if runtime.GOOS == "windows" {
23 t.Skip("subagent inbox transport uses Unix-domain sockets; Windows support will use a named-pipe backend")
24 }
25 dir, err := os.MkdirTemp("/tmp", "zut-in-")
26 if err != nil {
27 t.Fatalf("mkdir: %v", err)
28 }
29 t.Cleanup(func() { os.RemoveAll(dir) })
30 return dir
31}
32
33// TestInboxRoundTrip is the happy path: child listens, supervisor
34// dials and sends three lines, child receives them in order, and
35// closing both sides is clean. Locks in the line-framed protocol.
36func TestInboxRoundTrip(t *testing.T) {
37 path := filepath.Join(shortSocketDir(t), "in.sock")
38 ln, err := Listen(path)
39 if err != nil {
40 t.Fatalf("Listen: %v", err)
41 }
42 defer ln.Close()
43
44 got := collectFor(t, ln.Lines(), 4, time.Second)
45 inbox := NewInbox(path)
46 defer inbox.Close()
47
48 want := []string{"user hello", "user world", "cancel"}
49 for _, msg := range want {
50 if err := inbox.sendBytes([]byte(msg + "\n")); err != nil {
51 t.Fatalf("sendBytes(%q): %v", msg, err)
52 }
53 }
54 if err := inbox.SendCommand(NewCommand(CommandTurnStart, "agent-1", "turn-1", TurnStartPayload{Prompt: "round trip"})); err != nil {
55 t.Fatalf("SendCommand: %v", err)
56 }
57
58 gotLines := <-got
59 if len(gotLines) != len(want)+1 {
60 t.Fatalf("got %d lines, want %d: %v", len(gotLines), len(want)+1, gotLines)
61 }
62 for i, w := range want {
63 if gotLines[i] != w {
64 t.Errorf("line %d = %q; want %q", i, gotLines[i], w)
65 }
66 }
67 parsed, err := ParseCommand(gotLines[len(want)])
68 if err != nil {
69 t.Fatalf("ParseCommand: %v", err)
70 }
71 var payload TurnStartPayload
72 if err := parsed.DecodePayload(&payload); err != nil {
73 t.Fatalf("DecodePayload: %v", err)
74 }
75 if payload.Prompt != "round trip" {
76 t.Fatalf("decoded prompt = %q, want %q", payload.Prompt, "round trip")
77 }
78}
79
80// TestInboxNotReady regression-tests the explicit "child hasn't
81// opened the socket yet" case. The TUI needs an error it can show
82// as "agent not listening yet" rather than a generic dial failure.
83func TestInboxNotReady(t *testing.T) {
84 path := filepath.Join(shortSocketDir(t), "in.sock")
85 inbox := NewInbox(path)
86 defer inbox.Close()
87 err := inbox.sendBytes([]byte("user hi\n"))
88 if err == nil {
89 t.Fatal("sendBytes on missing socket: want error")
90 }
91 if !errors.Is(err, ErrNotReady) {
92 t.Fatalf("sendBytes err = %v; want ErrNotReady", err)
93 }
94}
95
96// TestIsNoListenerErrMatchesECONNREFUSED documents the contract
97// that the user's screenshot regression depends on: any dial error
98// whose root cause is "the kernel refused the connect because nobody
99// is listening" must fold into ErrNotReady at the sendBytes layer,
100// so the TUI shows "agent not accepting input (press R to resume)"
101// instead of leaking the unix-socket path. We pattern-match via
102// errors.Is on syscall.ECONNREFUSED rather than string-matching the
103// error text so platform variants all hit the same branch.
104func TestIsNoListenerErrMatchesECONNREFUSED(t *testing.T) {
105 if !isNoListenerErr(syscall.ECONNREFUSED) {
106 t.Error("isNoListenerErr(ECONNREFUSED) = false; want true")
107 }
108 // Wrapped via fmt.Errorf — common form of the dial error.
109 if !isNoListenerErr(fmt.Errorf("dial unix /tmp/x: %w", syscall.ECONNREFUSED)) {
110 t.Error("isNoListenerErr(wrapped) = false; want true")
111 }
112 if isNoListenerErr(syscall.ECONNRESET) {
113 t.Error("isNoListenerErr(ECONNRESET) = true; want false (unrelated errno)")
114 }
115 if isNoListenerErr(nil) {
116 t.Error("isNoListenerErr(nil) = true; want false")
117 }
118}
119
120// TestInboxNewSupervisorPreemptsOld documents that only one
121// supervisor connection at a time is honoured. The subagent design
122// assumes a single parent zut owns each agent; if a second
123// parent dials, the listener boots the first one. The first
124// supervisor's already-delivered messages still land in the
125// channel — what we're guarding against is silent dual ownership,
126// not message loss.
127func TestInboxNewSupervisorPreemptsOld(t *testing.T) {
128 path := filepath.Join(shortSocketDir(t), "in.sock")
129 ln, err := Listen(path)
130 if err != nil {
131 t.Fatalf("Listen: %v", err)
132 }
133 defer ln.Close()
134
135 first := NewInbox(path)
136 if err := first.sendBytes([]byte("user one\n")); err != nil {
137 t.Fatalf("first send: %v", err)
138 }
139 // Wait for the listener to absorb "user one" before dialing the
140 // second supervisor; otherwise we race against the connection
141 // preemption and "user one" can be dropped (the first conn is
142 // closed by Accept before its read loop processes the byte).
143 waitN(t, ln.Lines(), 1, time.Second)
144
145 second := NewInbox(path)
146 defer second.Close()
147 if err := second.sendBytes([]byte("user two\n")); err != nil {
148 t.Fatalf("second send: %v", err)
149 }
150
151 secondLines := waitN(t, ln.Lines(), 1, time.Second)
152 if len(secondLines) != 1 || secondLines[0] != "user two" {
153 t.Fatalf("second supervisor not received: %v", secondLines)
154 }
155
156 // The original supervisor's connection was preempted by the
157 // listener accepting the second dial. The next sendBytes on the
158 // first handle sees a broken pipe and drops the cached connection;
159 // the call after that redials successfully. Both behaviours
160 // matter: the broken-pipe error must surface (so a real
161 // supervisor knows it lost ownership) AND the next call must
162 // recover so a flaky reconnect doesn't permanently wedge the
163 // inbox.
164 if err := first.sendBytes([]byte("user three\n")); err == nil {
165 t.Fatal("expected broken-pipe error on stale connection")
166 }
167 if err := first.sendBytes([]byte("user four\n")); err != nil {
168 t.Fatalf("first redial after eviction: %v", err)
169 }
170}
171
172func waitN(t *testing.T, ch <-chan string, n int, timeout time.Duration) []string {
173 t.Helper()
174 var got []string
175 deadline := time.After(timeout)
176 for len(got) < n {
177 select {
178 case msg := <-ch:
179 got = append(got, msg)
180 case <-deadline:
181 return got
182 }
183 }
184 return got
185}
186
187// collectFor consumes up to n items from ch within timeout and
188// pushes the slice onto the returned channel. Used by the tests
189// above so they don't race against the listener's accept goroutine.
190func collectFor(t *testing.T, ch <-chan string, n int, timeout time.Duration) <-chan []string {
191 t.Helper()
192 out := make(chan []string, 1)
193 go func() {
194 var got []string
195 deadline := time.After(timeout)
196 for len(got) < n {
197 select {
198 case msg, ok := <-ch:
199 if !ok {
200 out <- got
201 return
202 }
203 got = append(got, msg)
204 case <-deadline:
205 out <- got
206 return
207 }
208 }
209 out <- got
210 }()
211 return out
212}
213
214// TestInboxConcurrentSends serialises the sendBytes call so two
215// goroutines writing at once still produce two whole lines, never
216// a tearing of bytes between them.
217func TestInboxConcurrentSends(t *testing.T) {
218 path := filepath.Join(shortSocketDir(t), "in.sock")
219 ln, err := Listen(path)
220 if err != nil {
221 t.Fatalf("Listen: %v", err)
222 }
223 defer ln.Close()
224 inbox := NewInbox(path)
225 defer inbox.Close()
226
227 var wg sync.WaitGroup
228 const n = 50
229 wg.Add(n)
230 for i := 0; i < n; i++ {
231 go func(i int) {
232 defer wg.Done()
233 if err := inbox.sendBytes([]byte("user msg\n")); err != nil {
234 t.Errorf("send %d: %v", i, err)
235 }
236 }(i)
237 }
238 wg.Wait()
239
240 collected := collectFor(t, ln.Lines(), n, 2*time.Second)
241 got := <-collected
242 if len(got) != n {
243 t.Fatalf("got %d lines, want %d", len(got), n)
244 }
245 for i, g := range got {
246 if g != "user msg" {
247 t.Fatalf("line %d garbled: %q", i, g)
248 }
249 }
250}