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 notifyingContext struct {
131 context.Context
132 doneCalled chan struct{}
133 once sync.Once
134}
135
136func (c *notifyingContext) Done() <-chan struct{} {
137 c.once.Do(func() { close(c.doneCalled) })
138 return c.Context.Done()
139}
140
141func TestInboxSendContextCancelsBlockedWrite(t *testing.T) {
142 local, remote := net.Pipe()
143 defer remote.Close()
144 conn := &blockingWriteConn{
145 Conn: local,
146 started: make(chan struct{}),
147 release: make(chan struct{}),
148 }
149 inbox := NewInbox("unused")
150 inbox.conn = conn
151 defer inbox.Close()
152
153 ctx, cancel := context.WithCancel(context.Background())
154 sent := make(chan error, 1)
155 go func() { sent <- inbox.sendBytesContext(ctx, []byte("follow-up\n")) }()
156 select {
157 case <-conn.started:
158 case <-time.After(time.Second):
159 t.Fatal("write did not start")
160 }
161 cancel()
162 select {
163 case err := <-sent:
164 if !errors.Is(err, context.Canceled) {
165 t.Fatalf("blocked write error = %v, want context canceled", err)
166 }
167 case <-time.After(time.Second):
168 t.Fatal("blocked write did not honor context cancellation")
169 }
170}
171
172func TestInboxSendContextCancelsWhileAnotherWriteIsBlocked(t *testing.T) {
173 local, remote := net.Pipe()
174 defer remote.Close()
175 conn := &blockingWriteConn{
176 Conn: local,
177 started: make(chan struct{}),
178 release: make(chan struct{}),
179 }
180 inbox := NewInbox("unused")
181 inbox.conn = conn
182 defer inbox.Close()
183
184 first := make(chan error, 1)
185 go func() { first <- inbox.sendBytesContext(context.Background(), []byte("first\n")) }()
186 select {
187 case <-conn.started:
188 case <-time.After(time.Second):
189 t.Fatal("first write did not start")
190 }
191
192 base, cancel := context.WithCancel(context.Background())
193 secondCtx := ¬ifyingContext{Context: base, doneCalled: make(chan struct{})}
194 second := make(chan error, 1)
195 go func() { second <- inbox.sendBytesContext(secondCtx, []byte("second\n")) }()
196 select {
197 case <-secondCtx.doneCalled:
198 case <-time.After(time.Second):
199 t.Fatal("second write did not wait for the send semaphore")
200 }
201 cancel()
202 select {
203 case err := <-second:
204 if !errors.Is(err, context.Canceled) {
205 t.Fatalf("waiting write error = %v, want context canceled", err)
206 }
207 case <-time.After(time.Second):
208 t.Fatal("waiting write did not honor context cancellation")
209 }
210
211 _ = conn.Close()
212 select {
213 case <-first:
214 case <-time.After(time.Second):
215 t.Fatal("first write did not unblock")
216 }
217}
218
219// TestIsNoListenerErrMatchesECONNREFUSED documents the contract
220// that the user's screenshot regression depends on: any dial error
221// whose root cause is "the kernel refused the connect because nobody
222// is listening" must fold into ErrNotReady at the sendBytes layer,
223// so the TUI shows "agent not accepting input (press R to resume)"
224// instead of leaking the unix-socket path. We pattern-match via
225// errors.Is on syscall.ECONNREFUSED rather than string-matching the
226// error text so platform variants all hit the same branch.
227func TestIsNoListenerErrMatchesECONNREFUSED(t *testing.T) {
228 if !isNoListenerErr(syscall.ECONNREFUSED) {
229 t.Error("isNoListenerErr(ECONNREFUSED) = false; want true")
230 }
231 // Wrapped via fmt.Errorf — common form of the dial error.
232 if !isNoListenerErr(fmt.Errorf("dial unix /tmp/x: %w", syscall.ECONNREFUSED)) {
233 t.Error("isNoListenerErr(wrapped) = false; want true")
234 }
235 if isNoListenerErr(syscall.ECONNRESET) {
236 t.Error("isNoListenerErr(ECONNRESET) = true; want false (unrelated errno)")
237 }
238 if isNoListenerErr(nil) {
239 t.Error("isNoListenerErr(nil) = true; want false")
240 }
241}
242
243// TestInboxNewSupervisorPreemptsOld documents that only one
244// supervisor connection at a time is honoured. The subagent design
245// assumes a single parent zut owns each agent; if a second
246// parent dials, the listener boots the first one. The first
247// supervisor's already-delivered messages still land in the
248// channel — what we're guarding against is silent dual ownership,
249// not message loss.
250func TestInboxNewSupervisorPreemptsOld(t *testing.T) {
251 path := filepath.Join(shortSocketDir(t), "in.sock")
252 ln, err := Listen(path)
253 if err != nil {
254 t.Fatalf("Listen: %v", err)
255 }
256 defer ln.Close()
257
258 first := NewInbox(path)
259 if err := first.sendBytes([]byte("user one\n")); err != nil {
260 t.Fatalf("first send: %v", err)
261 }
262 // Wait for the listener to absorb "user one" before dialing the
263 // second supervisor; otherwise we race against the connection
264 // preemption and "user one" can be dropped (the first conn is
265 // closed by Accept before its read loop processes the byte).
266 waitN(t, ln.Lines(), 1, time.Second)
267
268 second := NewInbox(path)
269 defer second.Close()
270 if err := second.sendBytes([]byte("user two\n")); err != nil {
271 t.Fatalf("second send: %v", err)
272 }
273
274 secondLines := waitN(t, ln.Lines(), 1, time.Second)
275 if len(secondLines) != 1 || secondLines[0] != "user two" {
276 t.Fatalf("second supervisor not received: %v", secondLines)
277 }
278
279 // The original supervisor's connection was preempted by the
280 // listener accepting the second dial. The next sendBytes on the
281 // first handle sees a broken pipe and drops the cached connection;
282 // the call after that redials successfully. Both behaviours
283 // matter: the broken-pipe error must surface (so a real
284 // supervisor knows it lost ownership) AND the next call must
285 // recover so a flaky reconnect doesn't permanently wedge the
286 // inbox.
287 if err := first.sendBytes([]byte("user three\n")); err == nil {
288 t.Fatal("expected broken-pipe error on stale connection")
289 }
290 if err := first.sendBytes([]byte("user four\n")); err != nil {
291 t.Fatalf("first redial after eviction: %v", err)
292 }
293}
294
295func waitN(t *testing.T, ch <-chan string, n int, timeout time.Duration) []string {
296 t.Helper()
297 var got []string
298 deadline := time.After(timeout)
299 for len(got) < n {
300 select {
301 case msg := <-ch:
302 got = append(got, msg)
303 case <-deadline:
304 return got
305 }
306 }
307 return got
308}
309
310// collectFor consumes up to n items from ch within timeout and
311// pushes the slice onto the returned channel. Used by the tests
312// above so they don't race against the listener's accept goroutine.
313func collectFor(t *testing.T, ch <-chan string, n int, timeout time.Duration) <-chan []string {
314 t.Helper()
315 out := make(chan []string, 1)
316 go func() {
317 var got []string
318 deadline := time.After(timeout)
319 for len(got) < n {
320 select {
321 case msg, ok := <-ch:
322 if !ok {
323 out <- got
324 return
325 }
326 got = append(got, msg)
327 case <-deadline:
328 out <- got
329 return
330 }
331 }
332 out <- got
333 }()
334 return out
335}
336
337// TestInboxConcurrentSends serialises the sendBytes call so two
338// goroutines writing at once still produce two whole lines, never
339// a tearing of bytes between them.
340func TestInboxConcurrentSends(t *testing.T) {
341 path := filepath.Join(shortSocketDir(t), "in.sock")
342 ln, err := Listen(path)
343 if err != nil {
344 t.Fatalf("Listen: %v", err)
345 }
346 defer ln.Close()
347 inbox := NewInbox(path)
348 defer inbox.Close()
349
350 var wg sync.WaitGroup
351 const n = 50
352 wg.Add(n)
353 for i := 0; i < n; i++ {
354 go func(i int) {
355 defer wg.Done()
356 if err := inbox.sendBytes([]byte("user msg\n")); err != nil {
357 t.Errorf("send %d: %v", i, err)
358 }
359 }(i)
360 }
361 wg.Wait()
362
363 collected := collectFor(t, ln.Lines(), n, 2*time.Second)
364 got := <-collected
365 if len(got) != n {
366 t.Fatalf("got %d lines, want %d", len(got), n)
367 }
368 for i, g := range got {
369 if g != "user msg" {
370 t.Fatalf("line %d garbled: %q", i, g)
371 }
372 }
373}