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 Worktree: MemWorktree(filepath.Join(root, "wt")),
50 NewRunner: func(a *Agent) Runner {
51 return &execRunner{
52 agent: a,
53 Command: swarmAgentArgs(swarmAgentArgsOpts{
54 Exe: exe,
55 Dir: a.Dir,
56 SessionPath: a.SessionPath,
57 InboxPath: a.InboxPath,
58 Task: a.Task,
59 Model: a.Model,
60 Provider: a.Provider,
61 }),
62 }
63 },
64 })
65 defer f.StopAll()
66
67 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
68 defer cancel()
69
70 a, err := f.Spawn(ctx, "first task")
71 if err != nil {
72 t.Fatalf("Spawn: %v", err)
73 }
74
75 // Wait until the durable log has at least one assistant_message
76 // from the initial task. That confirms stdout→log→follower.
77 waitFor := func(want string) {
78 t.Helper()
79 deadline := time.Now().Add(5 * time.Second)
80 for time.Now().Before(deadline) {
81 evs, _ := ReadEventLog(a.EventLogPath)
82 for _, ev := range evs {
83 if strings.Contains(eventText(ev), want) {
84 return
85 }
86 }
87 time.Sleep(20 * time.Millisecond)
88 }
89 evs, _ := ReadEventLog(a.EventLogPath)
90 t.Fatalf("timed out waiting for %q in event log; got %d events:\n%s\n%s",
91 want, len(evs), formatEvents(evs), dumpEventsVerbose(evs))
92 }
93 waitFor("echo: first task")
94
95 // Send a follow-up over the inbox. The stub echoes the text into
96 // an assistant_message we can poll for in the log.
97 if err := retrySend(f, a.ID, "user follow up", time.Second); err != nil {
98 t.Fatalf("SendInput: %v", err)
99 }
100 waitFor("echo: follow up")
101
102 // Shut the agent down gracefully via the inbox.
103 if err := f.SendInput(a.ID, "shutdown"); err != nil && !errors.Is(err, ErrNotReady) {
104 t.Fatalf("shutdown send: %v", err)
105 }
106 a.Wait()
107 if got := a.Status(); got != StatusDone && got != StatusKilled {
108 t.Fatalf("final status = %s; want done/killed", got)
109 }
110}
111
112// retrySend exists because the inbox dial races against the child
113// opening the socket. Production callers handle ErrNotReady with a
114// status message; tests retry within a small window.
115func retrySend(f *Swarm, id, msg string, timeout time.Duration) error {
116 deadline := time.Now().Add(timeout)
117 var lastErr error
118 for time.Now().Before(deadline) {
119 err := f.SendInput(id, msg)
120 if err == nil {
121 return nil
122 }
123 lastErr = err
124 if !errors.Is(err, ErrNotReady) {
125 return err
126 }
127 time.Sleep(30 * time.Millisecond)
128 }
129 return lastErr
130}
131
132func eventText(ev Event) string {
133 if ev.Type != "assistant_message" && ev.Type != "user_message" {
134 return ""
135 }
136 content, _ := ev.Data["content"].([]any)
137 var sb strings.Builder
138 for _, c := range content {
139 m, _ := c.(map[string]any)
140 if t, _ := m["type"].(string); t == "text" {
141 if txt, _ := m["text"].(string); txt != "" {
142 sb.WriteString(txt)
143 sb.WriteByte('\n')
144 }
145 }
146 }
147 return sb.String()
148}
149
150func dumpEventsVerbose(evs []Event) string {
151 var sb strings.Builder
152 for _, ev := range evs {
153 sb.WriteString(ev.Type)
154 sb.WriteString("\t")
155 for k, v := range ev.Data {
156 sb.WriteString(k)
157 sb.WriteString("=")
158 switch vv := v.(type) {
159 case string:
160 sb.WriteString(vv)
161 default:
162 sb.WriteString("<...>")
163 }
164 sb.WriteString(" ")
165 }
166 sb.WriteString("\n")
167 }
168 return sb.String()
169}
170
171func formatEvents(evs []Event) string {
172 var sb strings.Builder
173 for _, ev := range evs {
174 sb.WriteString(ev.Type)
175 sb.WriteString(" ")
176 sb.WriteString(ev.Time.Format(time.RFC3339Nano))
177 sb.WriteString("\n")
178 }
179 return sb.String()
180}
181
182func buildStubChild(t *testing.T) string {
183 t.Helper()
184 out := filepath.Join(t.TempDir(), "stubchild")
185 cmd := exec.Command("go", "build", "-o", out, "./testdata/cmd/stubchild")
186 // Pass through the test runner's env so `go build` can find
187 // HOME, PATH, GOCACHE, etc. CGO is disabled to keep the build
188 // hermetic across machines.
189 cmd.Env = append(os.Environ(), "CGO_ENABLED=0")
190 if b, err := cmd.CombinedOutput(); err != nil {
191 t.Fatalf("build stubchild: %v\n%s", err, b)
192 }
193 return out
194}