1package swarm
2
3import (
4 "bufio"
5 "errors"
6 "fmt"
7 "io"
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// that the parent zot uses to send follow-up prompts (and other
18// control lines) to a running swarm agent.
19//
20// Protocol on the wire: one UTF-8 message per line, newline-
21// terminated. Three message kinds:
22//
23// user <text>... — append <text> as the next user turn
24// cancel — cancel the agent's in-flight turn
25// shutdown — graceful exit; child will write a final
26// EvTurnEnd-equivalent JSON event and quit
27//
28// The child listens on the same socket via Listener (below) and
29// translates the messages into core.Agent.Prompt calls. The
30// parent never reads from the socket — it's a one-way channel
31// driven by the supervisor.
32//
33// Why a socket and not a FIFO: FIFOs are POSIX-only and have
34// awkward blocking-open semantics. A unix socket gives us:
35//
36// - the ability to fail fast if no child is listening yet,
37// so SendInput returns a useful error to the TUI;
38// - clean shutdown semantics when the child closes the
39// listener;
40// - works under tests with a temp dir; and
41// - portable enough; Windows support is a follow-up that can
42// swap in a named pipe behind this same interface.
43type Inbox struct {
44 path string
45
46 mu sync.Mutex
47 conn net.Conn // lazily dialed; one persistent connection per agent
48}
49
50// NewInbox returns a handle that will dial the socket at path on
51// the first SendInput call. The socket file is expected to be
52// created by the child (Listener.Open below). This split lets the
53// parent build the Inbox before the child has booted; sends that
54// happen before the child is ready get a clear "not yet" error
55// the caller can retry or surface.
56func NewInbox(path string) *Inbox { return &Inbox{path: path} }
57
58// Path returns the absolute socket path. Used by the runner to
59// pass the path to the child as a flag.
60func (b *Inbox) Path() string { return b.path }
61
62// SendInput writes msg to the inbox as one line. msg must already
63// include the leading kind ("user ", "cancel", "shutdown") — the
64// caller chooses the shape so we don't proliferate methods. The
65// socket is reused across calls; transient dial failures yield
66// ErrNotReady so the TUI can retry or report "agent not listening
67// yet" rather than crashing.
68func (b *Inbox) SendInput(msg string) error {
69 b.mu.Lock()
70 defer b.mu.Unlock()
71 if b.conn == nil {
72 c, err := dialUnix(b.path, 200*time.Millisecond)
73 if err != nil {
74 return err
75 }
76 b.conn = c
77 }
78 if _, err := io.WriteString(b.conn, msg+"\n"); err != nil {
79 // Drop the connection so the next call redials. The
80 // previous error is more informative than the redial's
81 // would be, so surface this one.
82 _ = b.conn.Close()
83 b.conn = nil
84 return err
85 }
86 return nil
87}
88
89// Close drops any persistent connection. Safe to call repeatedly.
90// Does not unlink the socket file — that's the child's job
91// (Listener.Close).
92func (b *Inbox) Close() error {
93 b.mu.Lock()
94 defer b.mu.Unlock()
95 if b.conn != nil {
96 err := b.conn.Close()
97 b.conn = nil
98 return err
99 }
100 return nil
101}
102
103// ErrNotReady is returned by SendInput when the child hasn't yet
104// opened its listener. Callers can retry after a short backoff;
105// the TUI surfaces it as "agent <id> not listening yet".
106var ErrNotReady = errors.New("swarm: agent inbox not ready")
107
108// dialUnix wraps net.DialTimeout with a small retry window so the
109// happy path "child just started, supervisor sends right away"
110// works without forcing the caller to poll.
111//
112// Errors collapse onto ErrNotReady whenever the failure means "no
113// one is listening": either the socket file doesn't exist (child
114// hasn't booted) or it does exist but the dial was refused (the
115// previous child exited and left a stale inode). Both are
116// recoverable by Swarm.Resume, and both should surface as the same
117// user-facing hint rather than as a path-leaking error string.
118func dialUnix(path string, timeout time.Duration) (net.Conn, error) {
119 deadline := time.Now().Add(timeout)
120 var lastErr error
121 for {
122 c, err := net.DialTimeout("unix", path, 50*time.Millisecond)
123 if err == nil {
124 return c, nil
125 }
126 lastErr = err
127 if time.Now().After(deadline) {
128 break
129 }
130 time.Sleep(20 * time.Millisecond)
131 }
132 if _, statErr := os.Stat(path); os.IsNotExist(statErr) {
133 return nil, ErrNotReady
134 }
135 if isNoListenerErr(lastErr) {
136 return nil, ErrNotReady
137 }
138 return nil, fmt.Errorf("swarm: dial %s: %w", path, lastErr)
139}
140
141// isNoListenerErr reports whether err means "the socket file exists
142// but no process is listening on it" — i.e. ECONNREFUSED on unix
143// domain sockets. We pattern-match on errno via errors.Is rather
144// than the error string so platform variants (linux vs darwin) all
145// fold to the same case.
146func isNoListenerErr(err error) bool {
147 return errors.Is(err, syscall.ECONNREFUSED)
148}
149
150// Listener is the child-side end of the inbox. The swarm-agent
151// daemon mode constructs one with Listen, then iterates Lines
152// to receive supervisor messages. Designed so a single goroutine
153// in the child can drive prompting without juggling raw net code.
154type Listener struct {
155 path string
156 ln net.Listener
157 // active is the most recent accepted connection. The protocol
158 // only expects one supervisor (the parent zot), so newer
159 // connections preempt older ones — the previous parent
160 // presumably crashed.
161 mu sync.Mutex
162 active net.Conn
163 out chan string
164 done chan struct{}
165}
166
167// Listen creates the socket at path and starts accepting. The
168// caller must call Close to remove the socket file on shutdown.
169// Returns the listener even if a stale socket exists from a
170// previous run: we unlink first, mirroring how most unix daemons
171// behave.
172func Listen(path string) (*Listener, error) {
173 if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
174 return nil, fmt.Errorf("inbox dir: %w", err)
175 }
176 // Best-effort cleanup of a stale socket. If the parent
177 // process is still alive and using it, Listen will fail
178 // below and the caller surfaces that as a real conflict.
179 _ = os.Remove(path)
180 ln, err := net.Listen("unix", path)
181 if err != nil {
182 return nil, fmt.Errorf("inbox listen: %w", err)
183 }
184 l := &Listener{
185 path: path,
186 ln: ln,
187 out: make(chan string, 16),
188 done: make(chan struct{}),
189 }
190 go l.acceptLoop()
191 return l, nil
192}
193
194// Lines returns a channel of newline-stripped messages received
195// from the supervisor. The channel closes when Close is called.
196func (l *Listener) Lines() <-chan string { return l.out }
197
198// Close stops accepting, drops the active connection, removes
199// the socket file, and closes Lines. Idempotent.
200func (l *Listener) Close() error {
201 select {
202 case <-l.done:
203 return nil
204 default:
205 close(l.done)
206 }
207 l.mu.Lock()
208 if l.active != nil {
209 _ = l.active.Close()
210 }
211 l.mu.Unlock()
212 _ = l.ln.Close()
213 _ = os.Remove(l.path)
214 return nil
215}
216
217func (l *Listener) acceptLoop() {
218 defer close(l.out)
219 for {
220 c, err := l.ln.Accept()
221 if err != nil {
222 select {
223 case <-l.done:
224 return
225 default:
226 // Accept can fail transiently (signal); back off briefly.
227 time.Sleep(20 * time.Millisecond)
228 continue
229 }
230 }
231 l.mu.Lock()
232 if l.active != nil {
233 _ = l.active.Close()
234 }
235 l.active = c
236 l.mu.Unlock()
237 go l.readLoop(c)
238 }
239}
240
241func (l *Listener) readLoop(c net.Conn) {
242 br := bufio.NewReader(c)
243 for {
244 line, err := br.ReadString('\n')
245 if line != "" {
246 select {
247 case l.out <- trimRightNL(line):
248 case <-l.done:
249 _ = c.Close()
250 return
251 }
252 }
253 if err != nil {
254 _ = c.Close()
255 l.mu.Lock()
256 if l.active == c {
257 l.active = nil
258 }
259 l.mu.Unlock()
260 return
261 }
262 }
263}
264
265func trimRightNL(s string) string {
266 for len(s) > 0 && (s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
267 s = s[:len(s)-1]
268 }
269 return s
270}