1package subagents
2
3import (
4 "context"
5 "errors"
6 "path/filepath"
7 "runtime"
8 "testing"
9 "time"
10)
11
12func TestRequiredWaitWakesOnlyAfterOutcomePersistence(t *testing.T) {
13 persisting := make(chan struct{})
14 release := make(chan struct{})
15 a := &Agent{
16 ID: "required-agent",
17 requirement: newRequirement(true, 1, time.Now()),
18 requirementChanged: make(chan struct{}),
19 persistFn: func(*Agent) error {
20 close(persisting)
21 <-release
22 return nil
23 },
24 }
25 resolved := make(chan struct{})
26 go func() {
27 a.resolveRequirement(1, nil, "", false)
28 close(resolved)
29 }()
30 <-persisting
31 a.lifecycleMu.Lock()
32 visible := a.visibleRequirementLocked()
33 a.lifecycleMu.Unlock()
34 if visible.State != RequirementPending {
35 t.Fatalf("visible requirement while persistence is blocked = %+v, want pending", visible)
36 }
37 waited := make(chan RequirementSnapshot, 1)
38 go func() {
39 got, _ := a.waitRequirement(context.Background())
40 waited <- got
41 }()
42 close(release)
43 <-resolved
44 if got := <-waited; got.State != RequirementSatisfied {
45 t.Fatalf("durable outcome = %+v", got)
46 }
47}
48
49func TestRequiredOutcomeUsesDelegatedTurnOrdinal(t *testing.T) {
50 a := &Agent{
51 requirement: newRequirement(true, 1, time.Now()),
52 requirementChanged: make(chan struct{}),
53 }
54 if err := updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
55 "step": float64(1), "lifetime_turns": float64(1), "current_run_turns": float64(1),
56 })); err != nil {
57 t.Fatal(err)
58 }
59 if err := updateAgentFromEvent(a, NewEvent("turn_end", map[string]any{"step": float64(1)})); err != nil {
60 t.Fatal(err)
61 }
62 if got := a.requirementSnapshot(); got.State != RequirementSatisfied || got.TargetTurn != 1 {
63 t.Fatalf("canonical first-turn outcome = %+v", got)
64 }
65
66 a.prepareRequired(0)
67 if err := updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
68 "step": float64(2), "lifetime_turns": float64(2), "current_run_turns": float64(1),
69 })); err != nil {
70 t.Fatal(err)
71 }
72 if err := updateAgentFromEvent(a, NewEvent("turn_end", map[string]any{"step": float64(2)})); err != nil {
73 t.Fatal(err)
74 }
75 if got := a.requirementSnapshot(); got.State != RequirementSatisfied || got.TargetTurn != 2 {
76 t.Fatalf("canonical retry outcome = %+v", got)
77 }
78
79 for _, step := range []int{0, 1} {
80 legacy := &Agent{
81 requirement: newRequirement(true, 1, time.Now()),
82 requirementChanged: make(chan struct{}),
83 }
84 if err := updateAgentFromEvent(legacy, NewEvent("turn_end", map[string]any{"step": float64(step)})); err != nil {
85 t.Fatal(err)
86 }
87 if got := legacy.requirementSnapshot(); got.State != RequirementSatisfied {
88 t.Fatalf("lifetime-free step %d outcome = %+v", step, got)
89 }
90 }
91}
92
93func TestRequiredWaitCompletesOnDelegatedTurnWhileWorkerStaysAlive(t *testing.T) {
94 if runtime.GOOS == "windows" {
95 t.Skip("unix sockets not supported")
96 }
97 if testing.Short() {
98 t.Skip("skip end-to-end runner test in -short mode")
99 }
100 exe := buildStubChild(t)
101 t.Setenv("ZUT_STUB_PROTOCOL", "1")
102 root := t.TempDir()
103 f := New(Config{
104 Root: filepath.Join(root, "subagents"),
105 RepoRoot: root,
106 NewRunner: func(a *Agent) Runner {
107 return &execRunner{
108 agent: a,
109 Command: subagentWorkerArgs(subagentWorkerArgsOpts{
110 Exe: exe, Dir: a.Dir, SessionPath: a.SessionPath, InboxPath: a.InboxPath, Task: a.Task,
111 }),
112 GracePeriod: time.Second,
113 }
114 },
115 })
116 t.Cleanup(f.StopAll)
117 a, err := f.SpawnReq(context.Background(), SpawnRequest{Task: "required review", Required: true})
118 if err != nil {
119 t.Fatal(err)
120 }
121 ctx, cancel := context.WithTimeout(context.Background(), runnerProcessTimeout)
122 defer cancel()
123 got, err := a.waitRequirement(ctx)
124 if err != nil {
125 events, _ := ReadEventLog(a.EventLogPath)
126 t.Fatalf("wait for delegated turn: %v\nsnapshot: %+v\nevents: %#v\nformatted: %s", err, a.Snapshot(), events, formatEvents(events))
127 }
128 if got.State != RequirementSatisfied {
129 t.Fatalf("requirement = %+v, want satisfied", got)
130 }
131 if snapshot := a.Snapshot(); snapshot.Status != StatusRunning {
132 t.Fatalf("worker lifecycle = (%s, %s), want live daemon", snapshot.Status, snapshot.TurnState)
133 }
134}
135
136func TestReloadFailsClosedWhenRequiredOutcomeWasNotObserved(t *testing.T) {
137 root := t.TempDir()
138 started := make(chan struct{})
139 first := New(Config{
140 Root: root, RepoRoot: root,
141 NewRunner: func(*Agent) Runner {
142 return RunnerFunc(func(ctx context.Context, _ Sink) error {
143 close(started)
144 <-ctx.Done()
145 return ctx.Err()
146 })
147 },
148 })
149 a, err := first.SpawnReq(context.Background(), SpawnRequest{Task: "required review", Required: true})
150 if err != nil {
151 t.Fatal(err)
152 }
153 <-started
154 t.Cleanup(first.StopAll)
155
156 second := New(Config{Root: root, RepoRoot: root})
157 loaded, errs := second.Reload()
158 if len(errs) != 0 || loaded != 1 {
159 t.Fatalf("reload loaded=%d errors=%v", loaded, errs)
160 }
161 got := second.Get(a.ID).Snapshot().Requirement
162 if got.State != RequirementIndeterminate || !got.Unmet() || got.ErrorCode != "required_outcome_unobserved" {
163 t.Fatalf("reconciled requirement = %+v, want durable indeterminate outcome", got)
164 }
165 meta, err := readAgentMeta(filepath.Join(root, "agents", a.ID))
166 if err != nil {
167 t.Fatal(err)
168 }
169 if meta.Requirement.State != RequirementIndeterminate {
170 t.Fatalf("persisted reconciled requirement = %+v", meta.Requirement)
171 }
172}
173
174func TestRequiredOutcomePersistsAcrossReload(t *testing.T) {
175 root := t.TempDir()
176 first := New(Config{
177 Root: root,
178 RepoRoot: root,
179 NewRunner: func(*Agent) Runner {
180 return RunnerFunc(func(context.Context, Sink) error { return errors.New("review failed") })
181 },
182 })
183 t.Cleanup(first.StopAll)
184
185 a, err := first.SpawnReq(context.Background(), SpawnRequest{Task: "required review", Required: true})
186 if err != nil {
187 t.Fatalf("spawn required worker: %v", err)
188 }
189 a.Wait()
190 got := a.Snapshot().Requirement
191 if !got.Required || got.State != RequirementFailed {
192 t.Fatalf("live requirement = %+v, want required failed", got)
193 }
194
195 second := New(Config{Root: root, RepoRoot: root})
196 loaded, errs := second.Reload()
197 if len(errs) != 0 {
198 t.Fatalf("reload errors: %v", errs)
199 }
200 if loaded != 1 {
201 t.Fatalf("loaded = %d, want 1", loaded)
202 }
203 reloaded := second.Get(a.ID)
204 if reloaded == nil {
205 t.Fatalf("reloaded worker %q not found", a.ID)
206 }
207 got = reloaded.Snapshot().Requirement
208 if !got.Required || got.State != RequirementFailed || got.TargetTurn != 1 {
209 t.Fatalf("reloaded requirement = %+v, want durable failed turn 1", got)
210 }
211
212 meta, err := readAgentMeta(filepath.Join(root, "agents", a.ID))
213 if err != nil {
214 t.Fatalf("read agent metadata: %v", err)
215 }
216 if !meta.Requirement.Required || meta.Requirement.State != RequirementFailed {
217 t.Fatalf("persisted requirement = %+v", meta.Requirement)
218 }
219}
220
221func TestRequirementNotificationPersistsAndRetryClearsIt(t *testing.T) {
222 root := t.TempDir()
223 first := New(Config{
224 Root: root, RepoRoot: root,
225 NewRunner: func(*Agent) Runner {
226 return RunnerFunc(func(context.Context, Sink) error { return nil })
227 },
228 })
229 t.Cleanup(first.StopAll)
230 a, err := first.SpawnReq(context.Background(), SpawnRequest{Task: "required validation", Required: true})
231 if err != nil {
232 t.Fatal(err)
233 }
234 a.Wait()
235 if err := first.MarkRequirementNotified(a.ID); err != nil {
236 t.Fatal(err)
237 }
238
239 second := New(Config{Root: root, RepoRoot: root})
240 if _, errs := second.Reload(); len(errs) != 0 {
241 t.Fatalf("reload errors: %v", errs)
242 }
243 reloaded := second.Get(a.ID)
244 if got := reloaded.requirementSnapshot(); !got.Notified || got.State != RequirementSatisfied {
245 t.Fatalf("reloaded notified requirement = %+v", got)
246 }
247 previous := reloaded.prepareRequired(0)
248 if previous.TargetTurn != 1 {
249 t.Fatalf("previous target = %d, want 1", previous.TargetTurn)
250 }
251 pending := reloaded.requirementSnapshot()
252 if pending.Notified || pending.TargetTurn != 2 {
253 t.Fatalf("retried requirement = %+v, want unnotified turn 2", pending)
254 }
255 reloaded.prepareRequired(0)
256 if got := reloaded.requirementSnapshot(); got.TargetTurn != 2 {
257 t.Fatalf("idempotent prepare target = %d, want 2", got.TargetTurn)
258 }
259 reloaded.resolveRequirement(2, nil, "", false)
260 if got := reloaded.requirementSnapshot(); got.Notified || got.State != RequirementSatisfied {
261 t.Fatalf("resolved retry = %+v, want unnotified satisfaction", got)
262 }
263}
264
265func TestRequiredTerminalOutcomesRemainUnmet(t *testing.T) {
266 tests := []struct {
267 name string
268 run func(context.Context) error
269 want RequirementState
270 }{
271 {name: "failed", run: func(context.Context) error { return errors.New("boom") }, want: RequirementFailed},
272 {name: "timed out", run: func(context.Context) error { return context.DeadlineExceeded }, want: RequirementTimedOut},
273 {name: "canceled", run: func(context.Context) error { return context.Canceled }, want: RequirementCanceled},
274 }
275 for _, tt := range tests {
276 t.Run(tt.name, func(t *testing.T) {
277 f := newTestSupervisor(t, func(*Agent) Runner {
278 return RunnerFunc(func(ctx context.Context, _ Sink) error { return tt.run(ctx) })
279 })
280 a, err := f.SpawnReq(context.Background(), SpawnRequest{Task: tt.name, Required: true})
281 if err != nil {
282 t.Fatal(err)
283 }
284 a.Wait()
285 got := a.Snapshot().Requirement
286 if got.State != tt.want || !got.Unmet() {
287 t.Fatalf("requirement = %+v, want unmet %s", got, tt.want)
288 }
289 })
290 }
291}
292
293func TestWaitRequiredIsEventDrivenAndRetryCanSatisfy(t *testing.T) {
294 started := make(chan struct{})
295 release := make(chan struct{})
296 attempt := 0
297 f := newTestSupervisor(t, func(*Agent) Runner {
298 attempt++
299 current := attempt
300 return RunnerFunc(func(ctx context.Context, _ Sink) error {
301 if current == 1 {
302 close(started)
303 select {
304 case <-release:
305 return errors.New("first attempt failed")
306 case <-ctx.Done():
307 return ctx.Err()
308 }
309 }
310 return nil
311 })
312 })
313 a, err := f.SpawnReq(context.Background(), SpawnRequest{Task: "required finalizer", Required: true})
314 if err != nil {
315 t.Fatal(err)
316 }
317 <-started
318
319 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
320 defer cancel()
321 waited := make(chan RequirementSnapshot, 1)
322 waitErr := make(chan error, 1)
323 go func() {
324 result, err := a.waitRequirement(ctx)
325 if err != nil {
326 waitErr <- err
327 return
328 }
329 waited <- result
330 }()
331 select {
332 case got := <-waited:
333 t.Fatalf("wait returned before the worker finished: %+v", got)
334 case err := <-waitErr:
335 t.Fatalf("wait failed before the worker finished: %v", err)
336 default:
337 }
338
339 close(release)
340 select {
341 case err := <-waitErr:
342 t.Fatalf("wait failed: %v", err)
343 case got := <-waited:
344 if got.State != RequirementFailed || !got.Unmet() {
345 t.Fatalf("first outcome = %+v, want unmet failure", got)
346 }
347 case <-ctx.Done():
348 t.Fatal("event-driven requirement wait did not wake")
349 }
350 a.Wait()
351
352 resumed, err := f.ResumeRequiredWithPrompt(context.Background(), a.ID, "retry the finalization")
353 if err != nil {
354 t.Fatalf("required retry: %v", err)
355 }
356 resumed.Wait()
357 got, err := resumed.waitRequirement(context.Background())
358 if err != nil {
359 t.Fatalf("wait for retry: %v", err)
360 }
361 if got.State != RequirementSatisfied || got.Unmet() || got.TargetTurn != 2 {
362 t.Fatalf("retry outcome = %+v, want satisfied turn 2", got)
363 }
364}