1// stubchild is a minimal swarm-agent stand-in used by the runner
2// end-to-end test. It speaks the daemon protocol the real zot
3// binary will implement next:
4//
5// - parses --swarm-agent <path>, --session <path>, --cwd <path>,
6// and an optional positional task,
7// - opens a unix-socket listener at the inbox path so the
8// supervisor's Inbox can dial through,
9// - emits well-formed JSONL events on stdout that the supervisor
10// mirrors into the durable event log,
11// - reads one line per supervisor message and echoes it back as
12// a "user_message" event followed by a fake "assistant_message"
13// so the runner sees the dialogue happen.
14//
15// The runner test compiles this binary into a tempdir, points
16// swarm's execRunner at it via Command, and asserts the events
17// flow through correctly without needing the full zot model
18// machinery.
19package main
20
21import (
22 "bufio"
23 "encoding/json"
24 "flag"
25 "fmt"
26 "net"
27 "os"
28 "sync"
29 "time"
30)
31
32func main() {
33 var (
34 inboxPath string
35 sessionPath string
36 cwd string
37 )
38 flag.StringVar(&inboxPath, "swarm-agent", "", "inbox socket path")
39 flag.StringVar(&sessionPath, "session", "", "session file path")
40 flag.StringVar(&cwd, "cwd", "", "working directory")
41 flag.Parse()
42
43 if inboxPath == "" {
44 fmt.Fprintln(os.Stderr, "stubchild: --swarm-agent required")
45 os.Exit(2)
46 }
47
48 emit := newEmitter()
49
50 // Open the inbox listener BEFORE doing any work so the supervisor
51 // can dial through as soon as it sees the agent_ready event. If we
52 // processed the initial task first, the parent's SendInput call
53 // would race the stub's net.Listen and trip ErrNotReady.
54 ln, err := net.Listen("unix", inboxPath)
55 if err != nil {
56 emit("error", map[string]any{"message": err.Error()})
57 os.Exit(1)
58 }
59 defer os.Remove(inboxPath)
60
61 emit("agent_ready", map[string]any{
62 "inbox": inboxPath,
63 "session": sessionPath,
64 "cwd": cwd,
65 })
66
67 // Initial task lives in the positional. Process it as the first
68 // user turn so the supervisor's "initial task" path is exercised.
69 if task := flag.Arg(0); task != "" {
70 runTurn(emit, task, 0)
71 }
72
73 turn := 1
74 for {
75 c, err := ln.Accept()
76 if err != nil {
77 return
78 }
79 br := bufio.NewReader(c)
80 for {
81 line, err := br.ReadString('\n')
82 if line != "" {
83 msg := trimNL(line)
84 switch {
85 case msg == "shutdown":
86 emit("agent_stopped", map[string]any{"reason": "shutdown"})
87 _ = c.Close()
88 return
89 case msg == "cancel":
90 emit("turn_end", map[string]any{"stop": "cancelled"})
91 case len(msg) > 5 && msg[:5] == "user ":
92 runTurn(emit, msg[5:], turn)
93 turn++
94 }
95 }
96 if err != nil {
97 _ = c.Close()
98 break
99 }
100 }
101 }
102}
103
104// runTurn fakes one model round-trip: turn_start, an echoed
105// user_message, an assistant_message that echoes back, and a
106// turn_end. Enough event variety that applyEventToSink has
107// something to interpret.
108func runTurn(emit emitter, text string, step int) {
109 emit("turn_start", map[string]any{"step": step})
110 emit("user_message", map[string]any{
111 "content": []any{
112 map[string]any{"type": "text", "text": text},
113 },
114 })
115 emit("assistant_message", map[string]any{
116 "content": []any{
117 map[string]any{"type": "text", "text": "echo: " + text},
118 },
119 })
120 emit("turn_end", map[string]any{"stop": "end"})
121}
122
123type emitter = func(string, map[string]any)
124
125func newEmitter() emitter {
126 var mu sync.Mutex
127 enc := json.NewEncoder(os.Stdout)
128 return func(typ string, data map[string]any) {
129 mu.Lock()
130 defer mu.Unlock()
131 if data == nil {
132 data = map[string]any{}
133 }
134 data["type"] = typ
135 data["time"] = time.Now().Format(time.RFC3339Nano)
136 _ = enc.Encode(data)
137 }
138}
139
140func trimNL(s string) string {
141 for len(s) > 0 && (s[len(s)-1] == '\n' || s[len(s)-1] == '\r') {
142 s = s[:len(s)-1]
143 }
144 return s
145}