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 if _, err := conn.Write(data); err != nil {
116 // Drop the connection so the next call redials. The previous error is
117 // more informative than the redial's would be, so surface this one.
118 stopCancel()
119 b.mu.Lock()
120 if b.conn == conn {
121 _ = b.conn.Close()
122 b.conn = nil
123 }
124 b.mu.Unlock()
125 if ctxErr := ctx.Err(); ctxErr != nil {
126 return ctxErr
127 }
128 return err
129 }
130 if !stopCancel() {
131 // Cancellation closed this connection after the write completed. Drop it
132 // so the next command redials instead of reusing a closed stream.
133 b.mu.Lock()
134 if b.conn == conn {
135 b.conn = nil
136 }
137 b.mu.Unlock()
138 return ctx.Err()
139 }
140 return nil
141}
142
143func (b *Inbox) acquireSend(ctx context.Context) (chan struct{}, error) {
144 b.mu.Lock()
145 if b.send == nil {
146 b.send = make(chan struct{}, 1)
147 b.send <- struct{}{}
148 }
149 send := b.send
150 b.mu.Unlock()
151 select {
152 case <-ctx.Done():
153 return nil, ctx.Err()
154 case <-send:
155 return send, nil
156 }
157}
158
159func releaseSend(send chan struct{}) {
160 send <- struct{}{}
161}
162
163// Close drops any persistent connection. Safe to call repeatedly.
164// Does not unlink the socket file — that's the child's job
165// (Listener.Close).
166func (b *Inbox) Close() error {
167 b.mu.Lock()
168 defer b.mu.Unlock()
169 if b.conn != nil {
170 err := b.conn.Close()
171 b.conn = nil
172 return err
173 }
174 return nil
175}
176
177// ErrNotReady is returned when the child hasn't yet
178// opened its listener. Callers can retry after a short backoff;
179// the TUI surfaces it as "agent <id> not listening yet".
180var ErrNotReady = errors.New("subagent: agent inbox not ready")
181
182// dialUnixContext retries a Unix-socket connection briefly so the happy path
183// "child just started, supervisor sends right away" works without forcing the
184// caller to poll. Errors collapse onto ErrNotReady whenever the failure means
185// no worker is listening, without exposing the socket path to the caller.
186func dialUnixContext(ctx context.Context, path string, timeout time.Duration) (net.Conn, error) {
187 if ctx == nil {
188 ctx = context.Background()
189 }
190 deadline := time.Now().Add(timeout)
191 var lastErr error
192 for {
193 if err := ctx.Err(); err != nil {
194 return nil, err
195 }
196 attemptTimeout := 50 * time.Millisecond
197 if remaining := time.Until(deadline); remaining < attemptTimeout {
198 attemptTimeout = remaining
199 }
200 if attemptTimeout <= 0 {
201 break
202 }
203 c, err := (&net.Dialer{Timeout: attemptTimeout}).DialContext(ctx, "unix", path)
204 if err == nil {
205 return c, nil
206 }
207 if ctxErr := ctx.Err(); ctxErr != nil {
208 return nil, ctxErr
209 }
210 lastErr = err
211 if time.Now().After(deadline) {
212 break
213 }
214 wait := time.NewTimer(20 * time.Millisecond)
215 select {
216 case <-ctx.Done():
217 wait.Stop()
218 return nil, ctx.Err()
219 case <-wait.C:
220 }
221 }
222 if lastErr == nil {
223 return nil, ErrNotReady
224 }
225 if _, statErr := os.Stat(path); os.IsNotExist(statErr) {
226 return nil, ErrNotReady
227 }
228 if isNoListenerErr(lastErr) {
229 return nil, ErrNotReady
230 }
231 return nil, fmt.Errorf("subagent: dial %s: %w", path, lastErr)
232}
233
234// isNoListenerErr reports whether err means "the socket file exists
235// but no process is listening on it" — i.e. ECONNREFUSED on unix
236// domain sockets. We pattern-match on errno via errors.Is rather
237// than the error string so platform variants (linux vs darwin) all
238// fold to the same case.
239func isNoListenerErr(err error) bool {
240 return errors.Is(err, syscall.ECONNREFUSED)
241}
242
243// inboxLive reports whether a process currently accepts connections on path.
244// A Unix socket inode can survive its owner, so os.Stat alone cannot
245// distinguish a live worker from a stale path. Callers use this probe before
246// resume/remove operations to avoid racing a detached worker's session and
247// workspace ownership.
248const inboxProbeLine = "__zut_subagent_probe__"
249
250func inboxLive(path string) bool {
251 if path == "" {
252 return false
253 }
254 conn, err := net.DialTimeout("unix", path, 100*time.Millisecond)
255 if err != nil {
256 return false
257 }
258 // Listener treats this first line as a health probe and does not
259 // replace the current supervisor connection. A bare connect would be
260 // enough to detect liveness, but sending an explicit probe lets the
261 // listener distinguish this side-effect-free check from a takeover.
262 _, writeErr := conn.Write([]byte(inboxProbeLine + "\n"))
263 _ = conn.Close()
264 return writeErr == nil
265}
266
267// Listener is the child-side end of the inbox. The subagent-worker
268// daemon mode constructs one with Listen, then iterates Lines
269// to receive supervisor messages. Designed so a single goroutine
270// in the child can drive prompting without juggling raw net code.
271type Listener struct {
272 path string
273 ln net.Listener
274 // active is the most recent accepted connection. The protocol
275 // only expects one supervisor (the parent zut), so newer
276 // connections preempt older ones — the previous parent
277 // presumably crashed.
278 mu sync.Mutex
279 active net.Conn
280 conns map[net.Conn]struct{}
281 out chan string
282 done chan struct{}
283 wg sync.WaitGroup
284 closed bool
285 closeOnce sync.Once
286}
287
288// Listen creates the socket at path and starts accepting. The
289// caller must call Close to remove the socket file on shutdown.
290// A live endpoint is preserved and rejected; only a refused probe
291// is treated as a stale socket eligible for removal.
292func Listen(path string) (*Listener, error) {
293 dir := filepath.Dir(path)
294 if err := os.MkdirAll(dir, 0o700); err != nil {
295 return nil, fmt.Errorf("inbox dir: %w", err)
296 }
297 if err := os.Chmod(dir, 0o700); err != nil {
298 return nil, fmt.Errorf("inbox dir permissions: %w", err)
299 }
300 // Probe an existing endpoint before removing it. Unlinking an active
301 // Unix socket does not stop its listener, so blindly removing the path
302 // could let Resume start a second worker on the same session and event log.
303 if _, statErr := os.Stat(path); statErr == nil {
304 if inboxLive(path) {
305 return nil, fmt.Errorf("subagent inbox already has a live listener")
306 }
307 // A refused probe means the socket inode is stale. Remove only that
308 // endpoint; a live listener was handled above without unlinking it.
309 _ = os.Remove(path)
310 }
311 ln, err := net.Listen("unix", path)
312 if err != nil {
313 return nil, fmt.Errorf("inbox listen: %w", err)
314 }
315 // Unix sockets inherit the process umask but may otherwise be
316 // world-writable. The inbox carries prompts and control commands,
317 // so fail closed for other local users.
318 _ = os.Chmod(path, 0o600)
319 l := &Listener{
320 path: path,
321 ln: ln,
322 conns: make(map[net.Conn]struct{}),
323 out: make(chan string, 16),
324 done: make(chan struct{}),
325 }
326 go l.acceptLoop()
327 return l, nil
328}
329
330// Lines returns a channel of newline-stripped messages received
331// from the supervisor. The channel closes when Close is called.
332func (l *Listener) Lines() <-chan string { return l.out }
333
334// Close stops accepting, drops the active connection, removes
335// the socket file, and closes Lines. Idempotent.
336func (l *Listener) Close() error {
337 l.closeOnce.Do(func() {
338 l.mu.Lock()
339 l.closed = true
340 close(l.done)
341 for conn := range l.conns {
342 _ = conn.Close()
343 }
344 l.mu.Unlock()
345 _ = l.ln.Close()
346 _ = os.Remove(l.path)
347 })
348 return nil
349}
350
351func (l *Listener) acceptLoop() {
352 defer func() {
353 l.wg.Wait()
354 close(l.out)
355 }()
356 for {
357 c, err := l.ln.Accept()
358 if err != nil {
359 select {
360 case <-l.done:
361 return
362 default:
363 // Accept can fail transiently (signal); back off briefly.
364 time.Sleep(20 * time.Millisecond)
365 continue
366 }
367 }
368 l.mu.Lock()
369 if l.closed {
370 l.mu.Unlock()
371 _ = c.Close()
372 return
373 }
374 l.conns[c] = struct{}{}
375 l.wg.Add(1)
376 l.mu.Unlock()
377 go func() {
378 defer l.wg.Done()
379 l.readLoop(c)
380 }()
381 }
382}
383
384func (l *Listener) readLoop(c net.Conn) {
385 defer func() {
386 _ = c.Close()
387 l.mu.Lock()
388 delete(l.conns, c)
389 if l.active == c {
390 l.active = nil
391 }
392 l.mu.Unlock()
393 }()
394
395 br := bufio.NewReader(c)
396 firstLine := true
397 for {
398 line, err := br.ReadString('\n')
399 if line != "" {
400 message := trimRightNL(line)
401 if firstLine {
402 firstLine = false
403 if message == inboxProbeLine {
404 return
405 }
406 // The first real command claims ownership. Health probes do
407 // not preempt the current supervisor, while a replacement
408 // supervisor still gets the historical takeover behavior.
409 l.mu.Lock()
410 if l.active != nil && l.active != c {
411 _ = l.active.Close()
412 }
413 l.active = c
414 l.mu.Unlock()
415 }
416 select {
417 case l.out <- message:
418 case <-l.done:
419 return
420 }
421 }
422 if err != nil {
423 return
424 }
425 }
426}
427
428func trimRightNL(s string) string {
429 for len(s) > 0 && (s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
430 s = s[:len(s)-1]
431 }
432 return s
433}