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 wg sync.WaitGroup
166}
167
168// Listen creates the socket at path and starts accepting. The
169// caller must call Close to remove the socket file on shutdown.
170// Returns the listener even if a stale socket exists from a
171// previous run: we unlink first, mirroring how most unix daemons
172// behave.
173func Listen(path string) (*Listener, error) {
174 if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
175 return nil, fmt.Errorf("inbox dir: %w", err)
176 }
177 // Best-effort cleanup of a stale socket. If the parent
178 // process is still alive and using it, Listen will fail
179 // below and the caller surfaces that as a real conflict.
180 _ = os.Remove(path)
181 ln, err := net.Listen("unix", path)
182 if err != nil {
183 return nil, fmt.Errorf("inbox listen: %w", err)
184 }
185 l := &Listener{
186 path: path,
187 ln: ln,
188 out: make(chan string, 16),
189 done: make(chan struct{}),
190 }
191 go l.acceptLoop()
192 return l, nil
193}
194
195// Lines returns a channel of newline-stripped messages received
196// from the supervisor. The channel closes when Close is called.
197func (l *Listener) Lines() <-chan string { return l.out }
198
199// Close stops accepting, drops the active connection, removes
200// the socket file, and closes Lines. Idempotent.
201func (l *Listener) Close() error {
202 select {
203 case <-l.done:
204 return nil
205 default:
206 close(l.done)
207 }
208 l.mu.Lock()
209 if l.active != nil {
210 _ = l.active.Close()
211 }
212 l.mu.Unlock()
213 _ = l.ln.Close()
214 _ = os.Remove(l.path)
215 return nil
216}
217
218func (l *Listener) acceptLoop() {
219 defer func() {
220 l.wg.Wait()
221 close(l.out)
222 }()
223 for {
224 c, err := l.ln.Accept()
225 if err != nil {
226 select {
227 case <-l.done:
228 return
229 default:
230 // Accept can fail transiently (signal); back off briefly.
231 time.Sleep(20 * time.Millisecond)
232 continue
233 }
234 }
235 l.mu.Lock()
236 if l.active != nil {
237 _ = l.active.Close()
238 }
239 l.active = c
240 l.mu.Unlock()
241 l.wg.Add(1)
242 go func() {
243 defer l.wg.Done()
244 l.readLoop(c)
245 }()
246 }
247}
248
249func (l *Listener) readLoop(c net.Conn) {
250 br := bufio.NewReader(c)
251 for {
252 line, err := br.ReadString('\n')
253 if line != "" {
254 select {
255 case l.out <- trimRightNL(line):
256 case <-l.done:
257 _ = c.Close()
258 return
259 }
260 }
261 if err != nil {
262 _ = c.Close()
263 l.mu.Lock()
264 if l.active == c {
265 l.active = nil
266 }
267 l.mu.Unlock()
268 return
269 }
270 }
271}
272
273func trimRightNL(s string) string {
274 for len(s) > 0 && (s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
275 s = s[:len(s)-1]
276 }
277 return s
278}