1package subagents
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "net"
8 "os"
9 "path/filepath"
10 "runtime"
11 "sync"
12 "syscall"
13 "testing"
14 "time"
15)
16
17// shortSocketDir returns a tempdir short enough that we can append
18// "in.sock" and still fit under the unix-socket path cap. The Go
19// runner's t.TempDir() lives under a long /var/folders/... path on
20// macOS which blows the 104-byte limit by itself, so the inbox tests
21// can't use it.
22func shortSocketDir(t *testing.T) string {
23 t.Helper()
24 if runtime.GOOS == "windows" {
25 t.Skip("subagent inbox transport uses Unix-domain sockets; Windows support will use a named-pipe backend")
26 }
27 dir, err := os.MkdirTemp("/tmp", "zut-in-")
28 if err != nil {
29 t.Fatalf("mkdir: %v", err)
30 }
31 t.Cleanup(func() { os.RemoveAll(dir) })
32 return dir
33}
34
35// TestInboxRoundTrip is the happy path: child listens, supervisor
36// dials and sends three lines, child receives them in order, and
37// closing both sides is clean. Locks in the line-framed protocol.
38func TestInboxRoundTrip(t *testing.T) {
39 path := filepath.Join(shortSocketDir(t), "in.sock")
40 ln, err := Listen(path)
41 if err != nil {
42 t.Fatalf("Listen: %v", err)
43 }
44 defer ln.Close()
45
46 got := collectFor(t, ln.Lines(), 4, time.Second)
47 inbox := NewInbox(path)
48 defer inbox.Close()
49
50 want := []string{"user hello", "user world", "cancel"}
51 for _, msg := range want {
52 if err := inbox.sendBytes([]byte(msg + "\n")); err != nil {
53 t.Fatalf("sendBytes(%q): %v", msg, err)
54 }
55 }
56 if err := inbox.SendCommand(NewCommand(CommandTurnStart, "agent-1", "turn-1", TurnStartPayload{Prompt: "round trip"})); err != nil {
57 t.Fatalf("SendCommand: %v", err)
58 }
59
60 gotLines := <-got
61 if len(gotLines) != len(want)+1 {
62 t.Fatalf("got %d lines, want %d: %v", len(gotLines), len(want)+1, gotLines)
63 }
64 for i, w := range want {
65 if gotLines[i] != w {
66 t.Errorf("line %d = %q; want %q", i, gotLines[i], w)
67 }
68 }
69 parsed, err := ParseCommand(gotLines[len(want)])
70 if err != nil {
71 t.Fatalf("ParseCommand: %v", err)
72 }
73 var payload TurnStartPayload
74 if err := parsed.DecodePayload(&payload); err != nil {
75 t.Fatalf("DecodePayload: %v", err)
76 }
77 if payload.Prompt != "round trip" {
78 t.Fatalf("decoded prompt = %q, want %q", payload.Prompt, "round trip")
79 }
80}
81
82// TestInboxNotReady regression-tests the explicit "child hasn't
83// opened the socket yet" case. The TUI needs an error it can show
84// as "agent not listening yet" rather than a generic dial failure.
85func TestInboxNotReady(t *testing.T) {
86 path := filepath.Join(shortSocketDir(t), "in.sock")
87 inbox := NewInbox(path)
88 defer inbox.Close()
89 err := inbox.sendBytes([]byte("user hi\n"))
90 if err == nil {
91 t.Fatal("sendBytes on missing socket: want error")
92 }
93 if !errors.Is(err, ErrNotReady) {
94 t.Fatalf("sendBytes err = %v; want ErrNotReady", err)
95 }
96}
97
98func TestDialUnixContextNonPositiveTimeoutIsNotReady(t *testing.T) {
99 path := filepath.Join(t.TempDir(), "existing")
100 if err := os.WriteFile(path, nil, 0o600); err != nil {
101 t.Fatal(err)
102 }
103 for _, timeout := range []time.Duration{0, -time.Millisecond} {
104 _, err := dialUnixContext(context.Background(), path, timeout)
105 if !errors.Is(err, ErrNotReady) {
106 t.Fatalf("dial timeout %s error = %v, want ErrNotReady", timeout, err)
107 }
108 }
109}
110
111type blockingWriteConn struct {
112 net.Conn
113 started chan struct{}
114 release chan struct{}
115 startOnce sync.Once
116 closeOnce sync.Once
117}
118
119func (c *blockingWriteConn) Write(data []byte) (int, error) {
120 c.startOnce.Do(func() { close(c.started) })
121 <-c.release
122 return c.Conn.Write(data)
123}
124
125func (c *blockingWriteConn) Close() error {
126 c.closeOnce.Do(func() { close(c.release) })
127 return c.Conn.Close()
128}
129
130type cancelAfterWriteConn struct {
131 net.Conn
132 cancel context.CancelFunc
133}
134
135func (c *cancelAfterWriteConn) Write(data []byte) (int, error) {
136 c.cancel()
137 return len(data), nil
138}
139
140type partialWriteConn struct {
141 net.Conn
142 withError bool
143}
144
145func (c *partialWriteConn) Write(data []byte) (int, error) {
146 written := len(data) / 2
147 if written == 0 {
148 written = 1
149 }
150 if c.withError {
151 return written, errors.New("partial write")
152 }
153 return written, nil
154}
155
156type notifyingContext struct {
157 context.Context
158 doneCalled chan struct{}
159 once sync.Once
160}
161
162func (c *notifyingContext) Done() <-chan struct{} {
163 c.once.Do(func() { close(c.doneCalled) })
164 return c.Context.Done()
165}
166
167func TestInboxSendContextTreatsCompleteWriteBeforeCancellationAsDelivered(t *testing.T) {
168 local, remote := net.Pipe()
169 defer remote.Close()
170 ctx, cancel := context.WithCancel(context.Background())
171 inbox := NewInbox("unused")
172 inbox.conn = &cancelAfterWriteConn{Conn: local, cancel: cancel}
173 defer inbox.Close()
174
175 if err := inbox.sendBytesContext(ctx, []byte("follow-up\n")); err != nil {
176 t.Fatalf("complete write after cancellation = %v, want success", err)
177 }
178}
179
180func TestInboxSendContextTreatsPartialWritesAsUnknownDelivery(t *testing.T) {
181 for _, withError := range []bool{false, true} {
182 name := "short write"
183 if withError {
184 name = "partial write error"
185 }
186 t.Run(name, func(t *testing.T) {
187 local, remote := net.Pipe()
188 defer remote.Close()
189 inbox := NewInbox("unused")
190 inbox.conn = &partialWriteConn{Conn: local, withError: withError}
191 defer inbox.Close()
192
193 err := inbox.sendBytesContext(context.Background(), []byte("follow-up\n"))
194 if !errors.Is(err, ErrDeliveryUnknown) {
195 t.Fatalf("partial write error = %v, want ErrDeliveryUnknown", err)
196 }
197 })
198 }
199}
200
201func TestInboxSendContextCancelsBlockedWrite(t *testing.T) {
202 local, remote := net.Pipe()
203 defer remote.Close()
204 conn := &blockingWriteConn{
205 Conn: local,
206 started: make(chan struct{}),
207 release: make(chan struct{}),
208 }
209 inbox := NewInbox("unused")
210 inbox.conn = conn
211 defer inbox.Close()
212
213 ctx, cancel := context.WithCancel(context.Background())
214 sent := make(chan error, 1)
215 go func() { sent <- inbox.sendBytesContext(ctx, []byte("follow-up\n")) }()
216 select {
217 case <-conn.started:
218 case <-time.After(time.Second):
219 t.Fatal("write did not start")
220 }
221 cancel()
222 select {
223 case err := <-sent:
224 if !errors.Is(err, context.Canceled) {
225 t.Fatalf("blocked write error = %v, want context canceled", err)
226 }
227 case <-time.After(time.Second):
228 t.Fatal("blocked write did not honor context cancellation")
229 }
230}
231
232func TestInboxSendContextCancelsWhileAnotherWriteIsBlocked(t *testing.T) {
233 local, remote := net.Pipe()
234 defer remote.Close()
235 conn := &blockingWriteConn{
236 Conn: local,
237 started: make(chan struct{}),
238 release: make(chan struct{}),
239 }
240 inbox := NewInbox("unused")
241 inbox.conn = conn
242 defer inbox.Close()
243
244 first := make(chan error, 1)
245 go func() { first <- inbox.sendBytesContext(context.Background(), []byte("first\n")) }()
246 select {
247 case <-conn.started:
248 case <-time.After(time.Second):
249 t.Fatal("first write did not start")
250 }
251
252 base, cancel := context.WithCancel(context.Background())
253 secondCtx := ¬ifyingContext{Context: base, doneCalled: make(chan struct{})}
254 second := make(chan error, 1)
255 go func() { second <- inbox.sendBytesContext(secondCtx, []byte("second\n")) }()
256 select {
257 case <-secondCtx.doneCalled:
258 case <-time.After(time.Second):
259 t.Fatal("second write did not wait for the send semaphore")
260 }
261 cancel()
262 select {
263 case err := <-second:
264 if !errors.Is(err, context.Canceled) {
265 t.Fatalf("waiting write error = %v, want context canceled", err)
266 }
267 case <-time.After(time.Second):
268 t.Fatal("waiting write did not honor context cancellation")
269 }
270
271 _ = conn.Close()
272 select {
273 case <-first:
274 case <-time.After(time.Second):
275 t.Fatal("first write did not unblock")
276 }
277}
278
279// TestIsNoListenerErrMatchesECONNREFUSED documents the contract
280// that the user's screenshot regression depends on: any dial error
281// whose root cause is "the kernel refused the connect because nobody
282// is listening" must fold into ErrNotReady at the sendBytes layer,
283// so the TUI shows "agent not accepting input (press R to resume)"
284// instead of leaking the unix-socket path. We pattern-match via
285// errors.Is on syscall.ECONNREFUSED rather than string-matching the
286// error text so platform variants all hit the same branch.
287func TestIsNoListenerErrMatchesECONNREFUSED(t *testing.T) {
288 if !isNoListenerErr(syscall.ECONNREFUSED) {
289 t.Error("isNoListenerErr(ECONNREFUSED) = false; want true")
290 }
291 // Wrapped via fmt.Errorf — common form of the dial error.
292 if !isNoListenerErr(fmt.Errorf("dial unix /tmp/x: %w", syscall.ECONNREFUSED)) {
293 t.Error("isNoListenerErr(wrapped) = false; want true")
294 }
295 if isNoListenerErr(syscall.ECONNRESET) {
296 t.Error("isNoListenerErr(ECONNRESET) = true; want false (unrelated errno)")
297 }
298 if isNoListenerErr(nil) {
299 t.Error("isNoListenerErr(nil) = true; want false")
300 }
301}
302
303// TestInboxNewSupervisorPreemptsOld documents that only one
304// supervisor connection at a time is honoured. The subagent design
305// assumes a single parent zut owns each agent; if a second
306// parent dials, the listener boots the first one. The first
307// supervisor's already-delivered messages still land in the
308// channel — what we're guarding against is silent dual ownership,
309// not message loss.
310func TestInboxNewSupervisorPreemptsOld(t *testing.T) {
311 path := filepath.Join(shortSocketDir(t), "in.sock")
312 ln, err := Listen(path)
313 if err != nil {
314 t.Fatalf("Listen: %v", err)
315 }
316 defer ln.Close()
317
318 first := NewInbox(path)
319 if err := first.sendBytes([]byte("user one\n")); err != nil {
320 t.Fatalf("first send: %v", err)
321 }
322 // Wait for the listener to absorb "user one" before dialing the
323 // second supervisor; otherwise we race against the connection
324 // preemption and "user one" can be dropped (the first conn is
325 // closed by Accept before its read loop processes the byte).
326 waitN(t, ln.Lines(), 1, time.Second)
327
328 second := NewInbox(path)
329 defer second.Close()
330 if err := second.sendBytes([]byte("user two\n")); err != nil {
331 t.Fatalf("second send: %v", err)
332 }
333
334 secondLines := waitN(t, ln.Lines(), 1, time.Second)
335 if len(secondLines) != 1 || secondLines[0] != "user two" {
336 t.Fatalf("second supervisor not received: %v", secondLines)
337 }
338
339 // The original supervisor's connection was preempted by the
340 // listener accepting the second dial. The next sendBytes on the
341 // first handle sees a broken pipe and drops the cached connection;
342 // the call after that redials successfully. Both behaviours
343 // matter: the broken-pipe error must surface (so a real
344 // supervisor knows it lost ownership) AND the next call must
345 // recover so a flaky reconnect doesn't permanently wedge the
346 // inbox.
347 if err := first.sendBytes([]byte("user three\n")); err == nil {
348 t.Fatal("expected broken-pipe error on stale connection")
349 }
350 if err := first.sendBytes([]byte("user four\n")); err != nil {
351 t.Fatalf("first redial after eviction: %v", err)
352 }
353}
354
355func waitN(t *testing.T, ch <-chan string, n int, timeout time.Duration) []string {
356 t.Helper()
357 var got []string
358 deadline := time.After(timeout)
359 for len(got) < n {
360 select {
361 case msg := <-ch:
362 got = append(got, msg)
363 case <-deadline:
364 return got
365 }
366 }
367 return got
368}
369
370// collectFor consumes up to n items from ch within timeout and
371// pushes the slice onto the returned channel. Used by the tests
372// above so they don't race against the listener's accept goroutine.
373func collectFor(t *testing.T, ch <-chan string, n int, timeout time.Duration) <-chan []string {
374 t.Helper()
375 out := make(chan []string, 1)
376 go func() {
377 var got []string
378 deadline := time.After(timeout)
379 for len(got) < n {
380 select {
381 case msg, ok := <-ch:
382 if !ok {
383 out <- got
384 return
385 }
386 got = append(got, msg)
387 case <-deadline:
388 out <- got
389 return
390 }
391 }
392 out <- got
393 }()
394 return out
395}
396
397// TestInboxConcurrentSends serialises the sendBytes call so two
398// goroutines writing at once still produce two whole lines, never
399// a tearing of bytes between them.
400func TestInboxConcurrentSends(t *testing.T) {
401 path := filepath.Join(shortSocketDir(t), "in.sock")
402 ln, err := Listen(path)
403 if err != nil {
404 t.Fatalf("Listen: %v", err)
405 }
406 defer ln.Close()
407 inbox := NewInbox(path)
408 defer inbox.Close()
409
410 var wg sync.WaitGroup
411 const n = 50
412 wg.Add(n)
413 for i := 0; i < n; i++ {
414 go func(i int) {
415 defer wg.Done()
416 if err := inbox.sendBytes([]byte("user msg\n")); err != nil {
417 t.Errorf("send %d: %v", i, err)
418 }
419 }(i)
420 }
421 wg.Wait()
422
423 collected := collectFor(t, ln.Lines(), n, 2*time.Second)
424 got := <-collected
425 if len(got) != n {
426 t.Fatalf("got %d lines, want %d", len(got), n)
427 }
428 for i, g := range got {
429 if g != "user msg" {
430 t.Fatalf("line %d garbled: %q", i, g)
431 }
432 }
433}