1package swarm
2
3import (
4 "context"
5 "errors"
6 "os"
7 "os/exec"
8 "path/filepath"
9 "runtime"
10 "strings"
11 "testing"
12 "time"
13)
14
15// TestRunnerEndToEndWithStubChild is the integration test for the
16// new daemon-mode runner. It compiles the stubchild binary in
17// testdata/cmd/stubchild, points an execRunner at it, and drives
18// a Swarm through one Spawn + one SendUserTurn + Stop cycle.
19//
20// What this proves:
21//
22// - The default argv shape (swarmAgentArgs) is one the child can
23// actually parse — locks the shape against silent breakage.
24// - The stdout JSONL parser ingests events and writes them to the
25// durable log.
26// - applyEventToSink turns events into Activity / Transcript
27// updates the dashboard reads.
28// - The supervisor inbox dials the child's socket and a follow-up
29// line round-trips back as a user_message event.
30// - Stop closes the inbox AND cancels the child's context so the
31// stub exits cleanly.
32//
33// The test is skipped on platforms without unix sockets (Windows).
34func TestRunnerEndToEndWithStubChild(t *testing.T) {
35 if runtime.GOOS == "windows" {
36 t.Skip("unix sockets not supported")
37 }
38 if testing.Short() {
39 t.Skip("skip end-to-end runner test in -short mode")
40 }
41
42 exe := buildStubChild(t)
43
44 root := t.TempDir()
45 repo := t.TempDir()
46 f := New(Config{
47 Root: root,
48 RepoRoot: repo,
49 NewRunner: func(a *Agent) Runner {
50 return &execRunner{
51 agent: a,
52 Command: swarmAgentArgs(swarmAgentArgsOpts{
53 Exe: exe,
54 Dir: a.Dir,
55 SessionPath: a.SessionPath,
56 InboxPath: a.InboxPath,
57 Task: a.Task,
58 Model: a.Model,
59 Provider: a.Provider,
60 }),
61 }
62 },
63 })
64 defer f.StopAll()
65
66 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
67 defer cancel()
68
69 a, err := f.Spawn(ctx, "first task")
70 if err != nil {
71 t.Fatalf("Spawn: %v", err)
72 }
73
74 // Wait until the durable log has at least one assistant_message
75 // from the initial task. That confirms stdout→log→follower.
76 waitFor := func(want string) {
77 t.Helper()
78 deadline := time.Now().Add(5 * time.Second)
79 for time.Now().Before(deadline) {
80 evs, _ := ReadEventLog(a.EventLogPath)
81 for _, ev := range evs {
82 if strings.Contains(eventText(ev), want) {
83 return
84 }
85 }
86 time.Sleep(20 * time.Millisecond)
87 }
88 evs, _ := ReadEventLog(a.EventLogPath)
89 t.Fatalf("timed out waiting for %q in event log; got %d events:\n%s\n%s",
90 want, len(evs), formatEvents(evs), dumpEventsVerbose(evs))
91 }
92 waitFor("echo: first task")
93
94 // Send a follow-up over the inbox. The stub echoes the text into
95 // an assistant_message we can poll for in the log.
96 if err := retrySend(f, a.ID, "user follow up", time.Second); err != nil {
97 t.Fatalf("SendInput: %v", err)
98 }
99 waitFor("echo: follow up")
100
101 // Shut the agent down gracefully via the inbox.
102 if err := f.SendInput(a.ID, "shutdown"); err != nil && !errors.Is(err, ErrNotReady) {
103 t.Fatalf("shutdown send: %v", err)
104 }
105 a.Wait()
106 if got := a.Status(); got != StatusDone && got != StatusKilled {
107 t.Fatalf("final status = %s; want done/killed", got)
108 }
109}
110
111// retrySend exists because the inbox dial races against the child
112// opening the socket. Production callers handle ErrNotReady with a
113// status message; tests retry within a small window.
114func retrySend(f *Swarm, id, msg string, timeout time.Duration) error {
115 deadline := time.Now().Add(timeout)
116 var lastErr error
117 for time.Now().Before(deadline) {
118 err := f.SendInput(id, msg)
119 if err == nil {
120 return nil
121 }
122 lastErr = err
123 if !errors.Is(err, ErrNotReady) {
124 return err
125 }
126 time.Sleep(30 * time.Millisecond)
127 }
128 return lastErr
129}
130
131func eventText(ev Event) string {
132 if ev.Type != "assistant_message" && ev.Type != "user_message" {
133 return ""
134 }
135 content, _ := ev.Data["content"].([]any)
136 var sb strings.Builder
137 for _, c := range content {
138 m, _ := c.(map[string]any)
139 if t, _ := m["type"].(string); t == "text" {
140 if txt, _ := m["text"].(string); txt != "" {
141 sb.WriteString(txt)
142 sb.WriteByte('\n')
143 }
144 }
145 }
146 return sb.String()
147}
148
149func dumpEventsVerbose(evs []Event) string {
150 var sb strings.Builder
151 for _, ev := range evs {
152 sb.WriteString(ev.Type)
153 sb.WriteString("\t")
154 for k, v := range ev.Data {
155 sb.WriteString(k)
156 sb.WriteString("=")
157 switch vv := v.(type) {
158 case string:
159 sb.WriteString(vv)
160 default:
161 sb.WriteString("<...>")
162 }
163 sb.WriteString(" ")
164 }
165 sb.WriteString("\n")
166 }
167 return sb.String()
168}
169
170func formatEvents(evs []Event) string {
171 var sb strings.Builder
172 for _, ev := range evs {
173 sb.WriteString(ev.Type)
174 sb.WriteString(" ")
175 sb.WriteString(ev.Time.Format(time.RFC3339Nano))
176 sb.WriteString("\n")
177 }
178 return sb.String()
179}
180
181func buildStubChild(t *testing.T) string {
182 t.Helper()
183 out := filepath.Join(t.TempDir(), "stubchild")
184 cmd := exec.Command("go", "build", "-o", out, "./testdata/cmd/stubchild")
185 // Pass through the test runner's env so `go build` can find
186 // HOME, PATH, GOCACHE, etc. CGO is disabled to keep the build
187 // hermetic across machines.
188 cmd.Env = append(os.Environ(), "CGO_ENABLED=0")
189 if b, err := cmd.CombinedOutput(); err != nil {
190 t.Fatalf("build stubchild: %v\n%s", err, b)
191 }
192 return out
193}