1package subagents
2
3import (
4 "bufio"
5 "context"
6 "errors"
7 "fmt"
8 "net"
9 "os"
10 "path/filepath"
11 "sync"
12 "syscall"
13 "time"
14)
15
16// Inbox is the supervisor-side handle on a per-agent unix socket.
17// The wire format is newline-delimited, versioned JSON commands. The child
18// listens on the same socket via Listener (below); the parent never reads
19// from it.
20//
21// Why a socket and not a FIFO: FIFOs are POSIX-only and have
22// awkward blocking-open semantics. A unix socket gives us:
23//
24// - the ability to fail fast if no child is listening yet,
25// so the supervisor gets a useful error to surface in the TUI;
26// - clean shutdown semantics when the child closes the
27// listener;
28// - works under tests with a temp dir; and
29// - portable enough; Windows support is a follow-up that can
30// swap in a named pipe behind this same interface.
31type Inbox struct {
32 path string
33
34 mu sync.Mutex
35 conn net.Conn // lazily dialed; one persistent connection per agent
36
37 // send is a one-slot semaphore that serializes JSONL writes without making
38 // a caller wait behind another blocked write after its context is canceled.
39 send chan struct{}
40}
41
42// NewInbox returns a handle that will dial the socket at path on
43// the first SendCommand call. The socket file is expected to be
44// created by the child (Listener.Open below). This split lets the
45// parent build the Inbox before the child has booted; sends that
46// happen before the child is ready get a clear "not yet" error
47// the caller can retry or surface.
48func NewInbox(path string) *Inbox {
49 inbox := &Inbox{path: path, send: make(chan struct{}, 1)}
50 inbox.send <- struct{}{}
51 return inbox
52}
53
54// Path returns the absolute socket path. Used by the runner to
55// pass the path to the child as a flag.
56func (b *Inbox) Path() string { return b.path }
57
58// SendCommand sends a versioned JSONL command.
59func (b *Inbox) SendCommand(command Envelope) error {
60 return b.SendCommandContext(context.Background(), command)
61}
62
63// SendCommandContext sends a command while honoring cancellation during both
64// connection setup and a blocked write. Cancellation closes the persistent
65// connection so a later command redials instead of reusing a half-written
66// stream.
67func (b *Inbox) SendCommandContext(ctx context.Context, command Envelope) error {
68 line, err := MarshalJSONL(command)
69 if err != nil {
70 return err
71 }
72 return b.sendBytesContext(ctx, line)
73}
74
75func (b *Inbox) sendBytes(data []byte) error {
76 return b.sendBytesContext(context.Background(), data)
77}
78
79func (b *Inbox) sendBytesContext(ctx context.Context, data []byte) error {
80 if ctx == nil {
81 ctx = context.Background()
82 }
83 if err := ctx.Err(); err != nil {
84 return err
85 }
86 send, err := b.acquireSend(ctx)
87 if err != nil {
88 return err
89 }
90 defer releaseSend(send)
91 if err := ctx.Err(); err != nil {
92 return err
93 }
94
95 b.mu.Lock()
96 if b.conn == nil {
97 c, err := dialUnixContext(ctx, b.path, 200*time.Millisecond)
98 if err != nil {
99 b.mu.Unlock()
100 return err
101 }
102 b.conn = c
103 }
104 conn := b.conn
105 b.mu.Unlock()
106
107 stopCancel := context.AfterFunc(ctx, func() { _ = conn.Close() })
108 if deadline, ok := ctx.Deadline(); ok {
109 if err := conn.SetWriteDeadline(deadline); err != nil {
110 stopCancel()
111 return err
112 }
113 defer conn.SetWriteDeadline(time.Time{})
114 }
115 n, writeErr := conn.Write(data)
116 if writeErr != nil {
117 // Drop the connection so the next call redials. Once any bytes have
118 // crossed the socket boundary, the outcome is ambiguous; returning the
119 // caller's context error would invite a duplicate command with a new
120 // identity.
121 stopCancel()
122 b.dropConnection(conn)
123 if n > 0 {
124 return fmt.Errorf("%w: wrote %d of %d bytes: %v", ErrDeliveryUnknown, n, len(data), writeErr)
125 }
126 if ctxErr := ctx.Err(); ctxErr != nil {
127 return ctxErr
128 }
129 return writeErr
130 }
131 if n != len(data) {
132 stopCancel()
133 b.dropConnection(conn)
134 return fmt.Errorf("%w: wrote %d of %d bytes", ErrDeliveryUnknown, n, len(data))
135 }
136 if !stopCancel() {
137 // Cancellation closed this connection after the complete command was
138 // written. The worker can parse the newline-delimited command, so the
139 // delivery succeeded even though the context is now canceled.
140 b.mu.Lock()
141 if b.conn == conn {
142 b.conn = nil
143 }
144 b.mu.Unlock()
145 }
146 return nil
147}
148
149func (b *Inbox) dropConnection(conn net.Conn) {
150 b.mu.Lock()
151 defer b.mu.Unlock()
152 if b.conn == conn {
153 _ = b.conn.Close()
154 b.conn = nil
155 }
156}
157
158func (b *Inbox) acquireSend(ctx context.Context) (chan struct{}, error) {
159 b.mu.Lock()
160 if b.send == nil {
161 b.send = make(chan struct{}, 1)
162 b.send <- struct{}{}
163 }
164 send := b.send
165 b.mu.Unlock()
166 select {
167 case <-ctx.Done():
168 return nil, ctx.Err()
169 case <-send:
170 return send, nil
171 }
172}
173
174func releaseSend(send chan struct{}) {
175 send <- struct{}{}
176}
177
178// Close drops any persistent connection. Safe to call repeatedly.
179// Does not unlink the socket file — that's the child's job
180// (Listener.Close).
181func (b *Inbox) Close() error {
182 b.mu.Lock()
183 defer b.mu.Unlock()
184 if b.conn != nil {
185 err := b.conn.Close()
186 b.conn = nil
187 return err
188 }
189 return nil
190}
191
192// ErrNotReady is returned when the child hasn't yet
193// opened its listener. Callers can retry after a short backoff;
194// the TUI surfaces it as "agent <id> not listening yet".
195var ErrNotReady = errors.New("subagent: agent inbox not ready")
196
197// ErrDeliveryUnknown means a command write made progress but did not finish.
198// Callers must not replace the command with a fresh identity: the worker may
199// have received enough bytes to parse it, so retries must reuse the same
200// message id and be handled idempotently.
201var ErrDeliveryUnknown = errors.New("subagent: command delivery unknown")
202
203// dialUnixContext retries a Unix-socket connection briefly so the happy path
204// "child just started, supervisor sends right away" works without forcing the
205// caller to poll. Errors collapse onto ErrNotReady whenever the failure means
206// no worker is listening, without exposing the socket path to the caller.
207func dialUnixContext(ctx context.Context, path string, timeout time.Duration) (net.Conn, error) {
208 if ctx == nil {
209 ctx = context.Background()
210 }
211 deadline := time.Now().Add(timeout)
212 var lastErr error
213 for {
214 if err := ctx.Err(); err != nil {
215 return nil, err
216 }
217 attemptTimeout := 50 * time.Millisecond
218 if remaining := time.Until(deadline); remaining < attemptTimeout {
219 attemptTimeout = remaining
220 }
221 if attemptTimeout <= 0 {
222 break
223 }
224 c, err := (&net.Dialer{Timeout: attemptTimeout}).DialContext(ctx, "unix", path)
225 if err == nil {
226 return c, nil
227 }
228 if ctxErr := ctx.Err(); ctxErr != nil {
229 return nil, ctxErr
230 }
231 lastErr = err
232 if time.Now().After(deadline) {
233 break
234 }
235 wait := time.NewTimer(20 * time.Millisecond)
236 select {
237 case <-ctx.Done():
238 wait.Stop()
239 return nil, ctx.Err()
240 case <-wait.C:
241 }
242 }
243 if lastErr == nil {
244 return nil, ErrNotReady
245 }
246 if _, statErr := os.Stat(path); os.IsNotExist(statErr) {
247 return nil, ErrNotReady
248 }
249 if isNoListenerErr(lastErr) {
250 return nil, ErrNotReady
251 }
252 return nil, fmt.Errorf("subagent: dial %s: %w", path, lastErr)
253}
254
255// isNoListenerErr reports whether err means "the socket file exists
256// but no process is listening on it" — i.e. ECONNREFUSED on unix
257// domain sockets. We pattern-match on errno via errors.Is rather
258// than the error string so platform variants (linux vs darwin) all
259// fold to the same case.
260func isNoListenerErr(err error) bool {
261 return errors.Is(err, syscall.ECONNREFUSED)
262}
263
264// inboxLive reports whether a process currently accepts connections on path.
265// A Unix socket inode can survive its owner, so os.Stat alone cannot
266// distinguish a live worker from a stale path. Callers use this probe before
267// resume/remove operations to avoid racing a detached worker's session and
268// workspace ownership.
269const inboxProbeLine = "__zut_subagent_probe__"
270
271func inboxLive(path string) bool {
272 if path == "" {
273 return false
274 }
275 conn, err := net.DialTimeout("unix", path, 100*time.Millisecond)
276 if err != nil {
277 return false
278 }
279 // Listener treats this first line as a health probe and does not
280 // replace the current supervisor connection. A bare connect would be
281 // enough to detect liveness, but sending an explicit probe lets the
282 // listener distinguish this side-effect-free check from a takeover.
283 _, writeErr := conn.Write([]byte(inboxProbeLine + "\n"))
284 _ = conn.Close()
285 return writeErr == nil
286}
287
288// Listener is the child-side end of the inbox. The subagent-worker
289// daemon mode constructs one with Listen, then iterates Lines
290// to receive supervisor messages. Designed so a single goroutine
291// in the child can drive prompting without juggling raw net code.
292type Listener struct {
293 path string
294 ln net.Listener
295 // active is the most recent accepted connection. The protocol
296 // only expects one supervisor (the parent zut), so newer
297 // connections preempt older ones — the previous parent
298 // presumably crashed.
299 mu sync.Mutex
300 active net.Conn
301 conns map[net.Conn]struct{}
302 out chan string
303 done chan struct{}
304 wg sync.WaitGroup
305 closed bool
306 closeOnce sync.Once
307}
308
309// Listen creates the socket at path and starts accepting. The
310// caller must call Close to remove the socket file on shutdown.
311// A live endpoint is preserved and rejected; only a refused probe
312// is treated as a stale socket eligible for removal.
313func Listen(path string) (*Listener, error) {
314 dir := filepath.Dir(path)
315 if err := os.MkdirAll(dir, 0o700); err != nil {
316 return nil, fmt.Errorf("inbox dir: %w", err)
317 }
318 if err := os.Chmod(dir, 0o700); err != nil {
319 return nil, fmt.Errorf("inbox dir permissions: %w", err)
320 }
321 // Probe an existing endpoint before removing it. Unlinking an active
322 // Unix socket does not stop its listener, so blindly removing the path
323 // could let Resume start a second worker on the same session and event log.
324 if _, statErr := os.Stat(path); statErr == nil {
325 if inboxLive(path) {
326 return nil, fmt.Errorf("subagent inbox already has a live listener")
327 }
328 // A refused probe means the socket inode is stale. Remove only that
329 // endpoint; a live listener was handled above without unlinking it.
330 _ = os.Remove(path)
331 }
332 ln, err := net.Listen("unix", path)
333 if err != nil {
334 return nil, fmt.Errorf("inbox listen: %w", err)
335 }
336 // Unix sockets inherit the process umask but may otherwise be
337 // world-writable. The inbox carries prompts and control commands,
338 // so fail closed for other local users.
339 _ = os.Chmod(path, 0o600)
340 l := &Listener{
341 path: path,
342 ln: ln,
343 conns: make(map[net.Conn]struct{}),
344 out: make(chan string, 16),
345 done: make(chan struct{}),
346 }
347 go l.acceptLoop()
348 return l, nil
349}
350
351// Lines returns a channel of newline-stripped messages received
352// from the supervisor. The channel closes when Close is called.
353func (l *Listener) Lines() <-chan string { return l.out }
354
355// Close stops accepting, drops the active connection, removes
356// the socket file, and closes Lines. Idempotent.
357func (l *Listener) Close() error {
358 l.closeOnce.Do(func() {
359 l.mu.Lock()
360 l.closed = true
361 close(l.done)
362 for conn := range l.conns {
363 _ = conn.Close()
364 }
365 l.mu.Unlock()
366 _ = l.ln.Close()
367 _ = os.Remove(l.path)
368 })
369 return nil
370}
371
372func (l *Listener) acceptLoop() {
373 defer func() {
374 l.wg.Wait()
375 close(l.out)
376 }()
377 for {
378 c, err := l.ln.Accept()
379 if err != nil {
380 select {
381 case <-l.done:
382 return
383 default:
384 // Accept can fail transiently (signal); back off briefly.
385 time.Sleep(20 * time.Millisecond)
386 continue
387 }
388 }
389 l.mu.Lock()
390 if l.closed {
391 l.mu.Unlock()
392 _ = c.Close()
393 return
394 }
395 l.conns[c] = struct{}{}
396 l.wg.Add(1)
397 l.mu.Unlock()
398 go func() {
399 defer l.wg.Done()
400 l.readLoop(c)
401 }()
402 }
403}
404
405func (l *Listener) readLoop(c net.Conn) {
406 defer func() {
407 _ = c.Close()
408 l.mu.Lock()
409 delete(l.conns, c)
410 if l.active == c {
411 l.active = nil
412 }
413 l.mu.Unlock()
414 }()
415
416 br := bufio.NewReader(c)
417 firstLine := true
418 for {
419 line, err := br.ReadString('\n')
420 if line != "" {
421 message := trimRightNL(line)
422 if firstLine {
423 firstLine = false
424 if message == inboxProbeLine {
425 return
426 }
427 // The first real command claims ownership. Health probes do
428 // not preempt the current supervisor, while a replacement
429 // supervisor still gets the historical takeover behavior.
430 l.mu.Lock()
431 if l.active != nil && l.active != c {
432 _ = l.active.Close()
433 }
434 l.active = c
435 l.mu.Unlock()
436 }
437 select {
438 case l.out <- message:
439 case <-l.done:
440 return
441 }
442 }
443 if err != nil {
444 return
445 }
446 }
447}
448
449func trimRightNL(s string) string {
450 for len(s) > 0 && (s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
451 s = s[:len(s)-1]
452 }
453 return s
454}