1package subagents
2
3import (
4 "bufio"
5 "errors"
6 "fmt"
7 "net"
8 "os"
9 "path/filepath"
10 "sync"
11 "syscall"
12 "time"
13)
14
15// Inbox is the supervisor-side handle on a per-agent unix socket.
16// The wire format is newline-delimited, versioned JSON commands. The child
17// listens on the same socket via Listener (below); the parent never reads
18// from it.
19//
20// Why a socket and not a FIFO: FIFOs are POSIX-only and have
21// awkward blocking-open semantics. A unix socket gives us:
22//
23// - the ability to fail fast if no child is listening yet,
24// so the supervisor gets a useful error to surface in the TUI;
25// - clean shutdown semantics when the child closes the
26// listener;
27// - works under tests with a temp dir; and
28// - portable enough; Windows support is a follow-up that can
29// swap in a named pipe behind this same interface.
30type Inbox struct {
31 path string
32
33 mu sync.Mutex
34 conn net.Conn // lazily dialed; one persistent connection per agent
35}
36
37// NewInbox returns a handle that will dial the socket at path on
38// the first SendCommand call. The socket file is expected to be
39// created by the child (Listener.Open below). This split lets the
40// parent build the Inbox before the child has booted; sends that
41// happen before the child is ready get a clear "not yet" error
42// the caller can retry or surface.
43func NewInbox(path string) *Inbox { return &Inbox{path: path} }
44
45// Path returns the absolute socket path. Used by the runner to
46// pass the path to the child as a flag.
47func (b *Inbox) Path() string { return b.path }
48
49// SendCommand sends a versioned JSONL command.
50func (b *Inbox) SendCommand(command Envelope) error {
51 line, err := MarshalJSONL(command)
52 if err != nil {
53 return err
54 }
55 return b.sendBytes(line)
56}
57
58func (b *Inbox) sendBytes(data []byte) error {
59 b.mu.Lock()
60 defer b.mu.Unlock()
61 if b.conn == nil {
62 c, err := dialUnix(b.path, 200*time.Millisecond)
63 if err != nil {
64 return err
65 }
66 b.conn = c
67 }
68 if _, err := b.conn.Write(data); err != nil {
69 // Drop the connection so the next call redials. The
70 // previous error is more informative than the redial's
71 // would be, so surface this one.
72 _ = b.conn.Close()
73 b.conn = nil
74 return err
75 }
76 return nil
77}
78
79// Close drops any persistent connection. Safe to call repeatedly.
80// Does not unlink the socket file — that's the child's job
81// (Listener.Close).
82func (b *Inbox) Close() error {
83 b.mu.Lock()
84 defer b.mu.Unlock()
85 if b.conn != nil {
86 err := b.conn.Close()
87 b.conn = nil
88 return err
89 }
90 return nil
91}
92
93// ErrNotReady is returned when the child hasn't yet
94// opened its listener. Callers can retry after a short backoff;
95// the TUI surfaces it as "agent <id> not listening yet".
96var ErrNotReady = errors.New("subagent: agent inbox not ready")
97
98// dialUnix wraps net.DialTimeout with a small retry window so the
99// happy path "child just started, supervisor sends right away"
100// works without forcing the caller to poll.
101//
102// Errors collapse onto ErrNotReady whenever the failure means "no
103// one is listening": either the socket file doesn't exist (child
104// hasn't booted) or it does exist but the dial was refused (the
105// previous child exited and left a stale inode). Both are
106// recoverable by Supervisor.Resume, and both should surface as the same
107// user-facing hint rather than as a path-leaking error string.
108func dialUnix(path string, timeout time.Duration) (net.Conn, error) {
109 deadline := time.Now().Add(timeout)
110 var lastErr error
111 for {
112 c, err := net.DialTimeout("unix", path, 50*time.Millisecond)
113 if err == nil {
114 return c, nil
115 }
116 lastErr = err
117 if time.Now().After(deadline) {
118 break
119 }
120 time.Sleep(20 * time.Millisecond)
121 }
122 if _, statErr := os.Stat(path); os.IsNotExist(statErr) {
123 return nil, ErrNotReady
124 }
125 if isNoListenerErr(lastErr) {
126 return nil, ErrNotReady
127 }
128 return nil, fmt.Errorf("subagent: dial %s: %w", path, lastErr)
129}
130
131// isNoListenerErr reports whether err means "the socket file exists
132// but no process is listening on it" — i.e. ECONNREFUSED on unix
133// domain sockets. We pattern-match on errno via errors.Is rather
134// than the error string so platform variants (linux vs darwin) all
135// fold to the same case.
136func isNoListenerErr(err error) bool {
137 return errors.Is(err, syscall.ECONNREFUSED)
138}
139
140// inboxLive reports whether a process currently accepts connections on path.
141// A Unix socket inode can survive its owner, so os.Stat alone cannot
142// distinguish a live worker from a stale path. Callers use this probe before
143// resume/remove operations to avoid racing a detached worker's session and
144// workspace ownership.
145const inboxProbeLine = "__zut_subagent_probe__"
146
147func inboxLive(path string) bool {
148 if path == "" {
149 return false
150 }
151 conn, err := net.DialTimeout("unix", path, 100*time.Millisecond)
152 if err != nil {
153 return false
154 }
155 // Listener treats this first line as a health probe and does not
156 // replace the current supervisor connection. A bare connect would be
157 // enough to detect liveness, but sending an explicit probe lets the
158 // listener distinguish this side-effect-free check from a takeover.
159 _, writeErr := conn.Write([]byte(inboxProbeLine + "\n"))
160 _ = conn.Close()
161 return writeErr == nil
162}
163
164// Listener is the child-side end of the inbox. The subagent-worker
165// daemon mode constructs one with Listen, then iterates Lines
166// to receive supervisor messages. Designed so a single goroutine
167// in the child can drive prompting without juggling raw net code.
168type Listener struct {
169 path string
170 ln net.Listener
171 // active is the most recent accepted connection. The protocol
172 // only expects one supervisor (the parent zut), so newer
173 // connections preempt older ones — the previous parent
174 // presumably crashed.
175 mu sync.Mutex
176 active net.Conn
177 conns map[net.Conn]struct{}
178 out chan string
179 done chan struct{}
180 wg sync.WaitGroup
181 closed bool
182 closeOnce sync.Once
183}
184
185// Listen creates the socket at path and starts accepting. The
186// caller must call Close to remove the socket file on shutdown.
187// A live endpoint is preserved and rejected; only a refused probe
188// is treated as a stale socket eligible for removal.
189func Listen(path string) (*Listener, error) {
190 dir := filepath.Dir(path)
191 if err := os.MkdirAll(dir, 0o700); err != nil {
192 return nil, fmt.Errorf("inbox dir: %w", err)
193 }
194 if err := os.Chmod(dir, 0o700); err != nil {
195 return nil, fmt.Errorf("inbox dir permissions: %w", err)
196 }
197 // Probe an existing endpoint before removing it. Unlinking an active
198 // Unix socket does not stop its listener, so blindly removing the path
199 // could let Resume start a second worker on the same session and event log.
200 if _, statErr := os.Stat(path); statErr == nil {
201 if inboxLive(path) {
202 return nil, fmt.Errorf("subagent inbox already has a live listener")
203 }
204 // A refused probe means the socket inode is stale. Remove only that
205 // endpoint; a live listener was handled above without unlinking it.
206 _ = os.Remove(path)
207 }
208 ln, err := net.Listen("unix", path)
209 if err != nil {
210 return nil, fmt.Errorf("inbox listen: %w", err)
211 }
212 // Unix sockets inherit the process umask but may otherwise be
213 // world-writable. The inbox carries prompts and control commands,
214 // so fail closed for other local users.
215 _ = os.Chmod(path, 0o600)
216 l := &Listener{
217 path: path,
218 ln: ln,
219 conns: make(map[net.Conn]struct{}),
220 out: make(chan string, 16),
221 done: make(chan struct{}),
222 }
223 go l.acceptLoop()
224 return l, nil
225}
226
227// Lines returns a channel of newline-stripped messages received
228// from the supervisor. The channel closes when Close is called.
229func (l *Listener) Lines() <-chan string { return l.out }
230
231// Close stops accepting, drops the active connection, removes
232// the socket file, and closes Lines. Idempotent.
233func (l *Listener) Close() error {
234 l.closeOnce.Do(func() {
235 l.mu.Lock()
236 l.closed = true
237 close(l.done)
238 for conn := range l.conns {
239 _ = conn.Close()
240 }
241 l.mu.Unlock()
242 _ = l.ln.Close()
243 _ = os.Remove(l.path)
244 })
245 return nil
246}
247
248func (l *Listener) acceptLoop() {
249 defer func() {
250 l.wg.Wait()
251 close(l.out)
252 }()
253 for {
254 c, err := l.ln.Accept()
255 if err != nil {
256 select {
257 case <-l.done:
258 return
259 default:
260 // Accept can fail transiently (signal); back off briefly.
261 time.Sleep(20 * time.Millisecond)
262 continue
263 }
264 }
265 l.mu.Lock()
266 if l.closed {
267 l.mu.Unlock()
268 _ = c.Close()
269 return
270 }
271 l.conns[c] = struct{}{}
272 l.wg.Add(1)
273 l.mu.Unlock()
274 go func() {
275 defer l.wg.Done()
276 l.readLoop(c)
277 }()
278 }
279}
280
281func (l *Listener) readLoop(c net.Conn) {
282 defer func() {
283 _ = c.Close()
284 l.mu.Lock()
285 delete(l.conns, c)
286 if l.active == c {
287 l.active = nil
288 }
289 l.mu.Unlock()
290 }()
291
292 br := bufio.NewReader(c)
293 firstLine := true
294 for {
295 line, err := br.ReadString('\n')
296 if line != "" {
297 message := trimRightNL(line)
298 if firstLine {
299 firstLine = false
300 if message == inboxProbeLine {
301 return
302 }
303 // The first real command claims ownership. Health probes do
304 // not preempt the current supervisor, while a replacement
305 // supervisor still gets the historical takeover behavior.
306 l.mu.Lock()
307 if l.active != nil && l.active != c {
308 _ = l.active.Close()
309 }
310 l.active = c
311 l.mu.Unlock()
312 }
313 select {
314 case l.out <- message:
315 case <-l.done:
316 return
317 }
318 }
319 if err != nil {
320 return
321 }
322 }
323}
324
325func trimRightNL(s string) string {
326 for len(s) > 0 && (s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
327 s = s[:len(s)-1]
328 }
329 return s
330}