interactive_resident_completion_test.go RawHistoryFolder
1package modes
2
3import (
4 "context"
5 "errors"
6 "strings"
7 "sync"
8 "sync/atomic"
9 "testing"
10 "time"
11
12 "github.com/bnema/zut/packages/agent/internal/orchestration"
13 "github.com/bnema/zut/packages/agent/subagents"
14 "github.com/bnema/zut/packages/core"
15)
16
17func TestResidentCompletionBatchDoesNotReportTerminalSiblingAsRunning(t *testing.T) {
18 ctx, cancel := context.WithCancel(context.Background())
19 defer cancel()
20 ag := core.NewAgent(nil, "model", "", nil)
21 // Own the delivery loop during setup: TrackResidentSubagent requests
22 // delivery immediately, and a manager-wave hold does not stop reports
23 // sliding into a busy parent. Both outcomes must be ready before draining
24 // to exercise the same-batch sibling filter, not two valid separate batches.
25 i := &Interactive{agent: ag, busy: true, runCtx: ctx, completionDeliveryRunning: true}
26 release := i.beginCompletionDeliveryHold()
27 defer release()
28 i.TrackResidentSubagent("first", "turn-1")
29 i.TrackResidentSubagent("second", "turn-2")
30
31 tracker := i.ensureCompletionTracker()
32 tracker.Report(subagents.Completion{AgentID: "first", TurnID: "turn-1", Status: "completed"})
33 tracker.Report(subagents.Completion{AgentID: "second", TurnID: "turn-2", Status: "completed"})
34 i.deliverCompletionUpdates()
35
36 queued := ag.PendingQueuedMessages()
37 if len(queued) != 2 {
38 t.Fatalf("queued completions = %#v, want 2", queued)
39 }
40 for _, message := range queued {
41 if strings.Contains(message.Text, "Still running:") {
42 t.Fatalf("terminal sibling reported as running: %q", message.Text)
43 }
44 }
45}
46
47// TestReleaseCompletionHoldDoesNotInvertCompletionLock pins the lock order
48// between the completion delivery hold and the coordinator state read. The
49// release path used to hold completionDeliveryMu while it called
50// goalContinuationMessage and interruptedPromptPending, which acquire i.mu,
51// while coordinatorHasPendingWorkers is legitimately called by a caller that
52// already holds i.mu. The two orders deadlocked.
53//
54// goalSampleGate makes the probe deterministic: the test waits until the
55// release is provably inside goal sampling before it reads coordinator state.
56// Under the inverted order that sample still owns completionDeliveryMu, so the
57// probe queues behind it and the bounded wait fails the test instead of
58// hanging the package.
59func TestReleaseCompletionHoldDoesNotInvertCompletionLock(t *testing.T) {
60 goal := &core.SessionGoal{ID: "goal-1", Status: core.GoalActive, Objective: "keep building"}
61 gate := newGoalSampleGate(func(int64) *core.SessionGoal { return goal })
62 defer gate.unblockNow()
63
64 i := &Interactive{agent: core.NewAgent(nil, "model", "", goalToolRegistry())}
65 i.cfg.CurrentGoal = gate.current
66 // A manager wake is not the subject of this test; stop it before it starts
67 // a turn.
68 i.cfg.PersistGoalRuntime = func(*core.SessionGoal) error { return errors.New("stop after wake") }
69 i.completionDeliveryHolds = 1
70
71 released := startGateRelease(i.releaseCompletionDeliveryHold)
72 enteredGate(t, gate, "releaseCompletionDeliveryHold")
73
74 // A caller holding i.mu that reads coordinator state takes
75 // completionDeliveryMu: that is the i.mu -> completionDeliveryMu order this
76 // test protects. Probing from that side must not queue behind the release.
77 i.mu.Lock()
78 pending := make(chan bool, 1)
79 go func() {
80 pending <- i.coordinatorHasPendingWorkers()
81 }()
82
83 select {
84 case <-pending:
85 case <-time.After(completionGateTimeout):
86 // Break the cycle before failing so the release can sample and relock.
87 i.mu.Unlock()
88 gate.unblockNow()
89 awaitGate(t, released, "releaseCompletionDeliveryHold")
90 t.Fatal("coordinatorHasPendingWorkers blocked behind releaseCompletionDeliveryHold")
91 }
92 i.mu.Unlock()
93 gate.unblockNow()
94 awaitGate(t, released, "releaseCompletionDeliveryHold")
95
96 i.completionDeliveryMu.Lock()
97 holds := i.completionDeliveryHolds
98 i.completionDeliveryMu.Unlock()
99 if holds != 0 {
100 t.Fatalf("completionDeliveryHolds = %d, want 0", holds)
101 }
102}
103
104// TestNonSealingReleaseCompletionHoldSkipsGoalSampling pins that only the
105// release which drains the last hold reads goal and interruption state. A
106// nested release (2 -> 1) or a duplicate release (already 0) must return
107// without acquiring i.mu, which those reads take.
108func TestNonSealingReleaseCompletionHoldSkipsGoalSampling(t *testing.T) {
109 for _, tc := range []struct {
110 name string
111 holds int
112 wantHolds int
113 }{
114 {name: "nested", holds: 2, wantHolds: 1},
115 {name: "duplicate", holds: 0, wantHolds: 0},
116 } {
117 t.Run(tc.name, func(t *testing.T) {
118 i := &Interactive{agent: core.NewAgent(nil, "model", "", nil)}
119 i.completionDeliveryHolds = tc.holds
120
121 i.mu.Lock()
122 done := startGateRelease(i.releaseCompletionDeliveryHold)
123 select {
124 case <-done:
125 case <-time.After(completionGateTimeout):
126 // Break the cycle before failing so the release goroutine exits.
127 i.mu.Unlock()
128 awaitGate(t, done, "non-sealing release")
129 t.Fatal("non-sealing release sampled goal state while i.mu was held")
130 }
131 i.mu.Unlock()
132
133 i.completionDeliveryMu.Lock()
134 holds := i.completionDeliveryHolds
135 i.completionDeliveryMu.Unlock()
136 if holds != tc.wantHolds {
137 t.Fatalf("completionDeliveryHolds = %d, want %d", holds, tc.wantHolds)
138 }
139 })
140 }
141}
142
143// TestStaleReleaseCompletionHoldDoesNotSealNewerWave covers a duplicate release
144// that arrives after the earlier wave already drained, while a newer manager
145// wave is active. The stale call carries no hold and must not seal that wave:
146// pre-fix it re-applied GoalChanged and ManagerFinished, which left the newer
147// wave looking finished.
148func TestStaleReleaseCompletionHoldDoesNotSealNewerWave(t *testing.T) {
149 i := &Interactive{agent: core.NewAgent(nil, "model", "", nil)}
150 // A newer manager wave is already active with no completion hold
151 // outstanding, as happens when completion delivery wakes the manager after
152 // the holds drained.
153 i.applyCoordinator(orchestration.Event{Kind: orchestration.EventManagerStarted})
154 i.applyCoordinator(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: "w1", AgentID: "agent-1"})
155 if i.coordinatorAcceptsUserInput() {
156 t.Fatal("setup: newer manager wave is already sealed")
157 }
158
159 i.releaseCompletionDeliveryHold()
160
161 if i.coordinatorAcceptsUserInput() {
162 t.Fatal("duplicate release sealed the newer manager wave")
163 }
164 if !i.coordinatorHasPendingWorkers() {
165 t.Fatal("duplicate release lost the newer wave's pending worker")
166 }
167}
168
169// goalSampleGate pins the goal-sampling window of a completion delivery
170// release without sleeps. The first CurrentGoal call signals entered and then
171// waits for unblock, so a test can deterministically act while that release is
172// between its hold transition and its relock.
173type goalSampleGate struct {
174 entered chan struct{}
175 unblock chan struct{}
176 // once lets tests close unblock more than once, including from a deferred
177 // cleanup that runs after the test already closed it.
178 once sync.Once
179 calls atomic.Int64
180 next func(call int64) *core.SessionGoal
181}
182
183func newGoalSampleGate(next func(call int64) *core.SessionGoal) *goalSampleGate {
184 return &goalSampleGate{
185 entered: make(chan struct{}),
186 unblock: make(chan struct{}),
187 next: next,
188 }
189}
190
191// unblockNow releases the sampling window exactly once, so a failure cleanup
192// can let a parked release finish without a double close.
193func (g *goalSampleGate) unblockNow() {
194 g.once.Do(func() { close(g.unblock) })
195}
196
197// current fulfils InteractiveConfig.CurrentGoal.
198func (g *goalSampleGate) current() *core.SessionGoal {
199 call := g.calls.Add(1)
200 if call == 1 {
201 close(g.entered)
202 <-g.unblock
203 }
204 return g.next(call)
205}
206
207// completionGateTimeout bounds every wait in the completion-hold gate tests so
208// an inverted lock order fails the test instead of hanging the package.
209const completionGateTimeout = 5 * time.Second
210
211// startGateRelease runs a potentially blocking hold, release, or coordinator
212// transition in its own goroutine and returns a channel closed when it
213// returns, so a test can bound the wait instead of hanging.
214func startGateRelease(release func()) <-chan struct{} {
215 done := make(chan struct{})
216 go func() {
217 defer close(done)
218 release()
219 }()
220 return done
221}
222
223// awaitGate waits for done to close, failing the test after
224// completionGateTimeout rather than blocking forever.
225func awaitGate(t *testing.T, done <-chan struct{}, what string) {
226 t.Helper()
227 select {
228 case <-done:
229 case <-time.After(completionGateTimeout):
230 t.Fatalf("%s did not finish within %s", what, completionGateTimeout)
231 }
232}
233
234// enteredGate waits until a parked release has provably reached goal sampling.
235func enteredGate(t *testing.T, gate *goalSampleGate, what string) {
236 t.Helper()
237 select {
238 case <-gate.entered:
239 case <-time.After(completionGateTimeout):
240 gate.unblockNow()
241 t.Fatalf("%s never sampled goal state", what)
242 }
243}
244
245// TestCancelledCoordinatorIsNotSealedByStaleRelease covers cancellation
246// invalidating an in-flight release. The release drained an older hold and is
247// sampling goal state when the coordinator is replaced, so its stale
248// GoalChanged/ManagerFinished snapshot must not seal or wake the replacement.
249func TestCancelledCoordinatorIsNotSealedByStaleRelease(t *testing.T) {
250 goal := &core.SessionGoal{ID: "goal-1", Status: core.GoalActive, Objective: "keep building"}
251 gate := newGoalSampleGate(func(int64) *core.SessionGoal { return goal })
252 defer gate.unblockNow()
253
254 // A manager wake reaches goal sampling only through PersistGoalRuntime, so
255 // recording that callback detects an unintended seal. Returning an error
256 // stops the continuation before it touches the provider.
257 woke := make(chan struct{}, 1)
258 i := &Interactive{agent: core.NewAgent(nil, "model", "", goalToolRegistry())}
259 i.cfg.CurrentGoal = gate.current
260 i.cfg.PersistGoalRuntime = func(*core.SessionGoal) error {
261 select {
262 case woke <- struct{}{}:
263 default:
264 }
265 return errors.New("stop after wake")
266 }
267
268 // The snapshot belongs to a wave that still owns a worker, so sealing it
269 // would wake the manager.
270 i.applyCoordinator(orchestration.Event{Kind: orchestration.EventManagerStarted})
271 i.applyCoordinator(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: "w1", AgentID: "agent-1"})
272 if !i.coordinatorHasPendingWorkers() {
273 t.Fatal("setup: outstanding wave has no pending worker")
274 }
275
276 releaseOld := i.beginCompletionDeliveryHold()
277 released := startGateRelease(releaseOld)
278 enteredGate(t, gate, "older release")
279
280 awaitGate(t, startGateRelease(i.cancelCoordinator), "cancelCoordinator")
281
282 gate.unblockNow()
283 awaitGate(t, released, "older release")
284
285 // The replacement coordinator never owned the older wave, so the stale
286 // release must leave it compact and must not wake the manager.
287 if i.coordinatorHasPendingWorkers() {
288 t.Fatal("stale release resurrected the cancelled wave's worker")
289 }
290 if i.coordinatorAcceptsUserInput() {
291 t.Fatal("stale release sealed the replacement coordinator")
292 }
293 select {
294 case <-woke:
295 t.Fatal("stale release woke the manager after cancellation")
296 default:
297 }
298
299 // Positive control: the replacement coordinator still seals and wakes for a
300 // wave that starts after cancellation, so the check above is meaningful.
301 awaitGate(t, startGateRelease(func() { i.beginCompletionDeliveryHold() }), "post-cancel hold")
302 awaitGate(t, startGateRelease(i.releaseCompletionDeliveryHold), "post-cancel release")
303 select {
304 case <-woke:
305 default:
306 t.Fatal("post-cancel wave did not seal and wake the manager")
307 }
308}
309
310// TestImplicitCoordinatorWaveIsNotSealedByStaleRelease covers a worker
311// registered outside a completion-delivery wave. That implicit wave replaces
312// whatever wave an older release drained, so once the worker finishes and the
313// wave is sealed with no pending work, the stale release must not fall through
314// to a goal wake when it relocks.
315func TestImplicitCoordinatorWaveIsNotSealedByStaleRelease(t *testing.T) {
316 goal := &core.SessionGoal{ID: "goal-1", Status: core.GoalActive, Objective: "keep building"}
317 gate := newGoalSampleGate(func(int64) *core.SessionGoal { return goal })
318 defer gate.unblockNow()
319
320 woke := make(chan struct{}, 1)
321 i := &Interactive{agent: core.NewAgent(nil, "model", "", goalToolRegistry())}
322 i.cfg.CurrentGoal = gate.current
323 i.cfg.PersistGoalRuntime = func(*core.SessionGoal) error {
324 select {
325 case woke <- struct{}{}:
326 default:
327 }
328 return errors.New("stop after wake")
329 }
330
331 releaseOld := i.beginCompletionDeliveryHold()
332 released := startGateRelease(releaseOld)
333 enteredGate(t, gate, "older release")
334
335 // A worker registered with no hold outstanding opens the implicit
336 // one-worker sealed wave.
337 awaitGate(t, startGateRelease(func() { i.registerCoordinatorWorker("w1") }), "registerCoordinatorWorker")
338 workerID := i.takeCoordinatorWorkerID("w1")
339 if workerID == "" {
340 t.Fatal("setup: implicit wave registered no worker")
341 }
342
343 // The worker finishes inside the older release's sampling window, leaving
344 // the implicit wave sealed with no pending work. Its own worker wake is not
345 // under test, so only the coordinator state is advanced here.
346 i.applyCoordinator(orchestration.Event{Kind: orchestration.EventWorkerFinished, WorkerID: workerID})
347 if i.coordinatorHasPendingWorkers() {
348 t.Fatal("setup: implicit wave still has pending work")
349 }
350
351 gate.unblockNow()
352 awaitGate(t, released, "older release")
353
354 select {
355 case <-woke:
356 t.Fatal("stale release sealed the newer implicit coordinator wave")
357 default:
358 }
359}
360
361// TestReleaseCompletionHoldLeavesActiveNewerWaveUnsealed covers a newer hold
362// that begins while an older release samples goal state and is still
363// outstanding when that release relocks. The older release must leave the
364// newer wave alone, and the newer release must still seal it.
365func TestReleaseCompletionHoldLeavesActiveNewerWaveUnsealed(t *testing.T) {
366 goal := &core.SessionGoal{ID: "goal-1", Status: core.GoalActive, Objective: "keep building"}
367 gate := newGoalSampleGate(func(int64) *core.SessionGoal { return goal })
368 defer gate.unblockNow()
369
370 i := &Interactive{agent: core.NewAgent(nil, "model", "", goalToolRegistry())}
371 i.cfg.CurrentGoal = gate.current
372 // A manager wake is not the subject of this test; stop it before it starts
373 // a turn.
374 i.cfg.PersistGoalRuntime = func(*core.SessionGoal) error { return errors.New("stop after wake") }
375
376 releaseOld := i.beginCompletionDeliveryHold()
377 i.registerCoordinatorWorker("w1")
378
379 released := startGateRelease(releaseOld)
380 enteredGate(t, gate, "older release")
381
382 var releaseNew func()
383 awaitGate(t, startGateRelease(func() { releaseNew = i.beginCompletionDeliveryHold() }), "newer hold")
384 if i.coordinatorAcceptsUserInput() {
385 t.Fatal("setup: newer hold wave is already sealed")
386 }
387 gate.unblockNow()
388 awaitGate(t, released, "older release")
389
390 i.completionDeliveryMu.Lock()
391 holds := i.completionDeliveryHolds
392 i.completionDeliveryMu.Unlock()
393 if holds != 1 {
394 t.Fatalf("completionDeliveryHolds = %d, want 1 (newer hold still active)", holds)
395 }
396 if i.coordinatorAcceptsUserInput() {
397 t.Fatal("older release sealed the newer wave it did not drain")
398 }
399
400 awaitGate(t, startGateRelease(releaseNew), "newer release")
401 if !i.coordinatorAcceptsUserInput() {
402 t.Fatal("newer release did not seal its own wave")
403 }
404 if !i.coordinatorHasPendingWorkers() {
405 t.Fatal("newer wave lost its pending worker")
406 }
407}
408
409// TestDrainedNewerWaveIsNotResealedByOlderRelease pins the wave generation
410// guard. While an older release samples goal state, a newer wave both begins
411// and fully drains, leaving the hold count at zero when the older release
412// relocks. The count alone would let that older release re-apply its stale
413// snapshot (including GoalChanged) over the wave that already sealed.
414func TestDrainedNewerWaveIsNotResealedByOlderRelease(t *testing.T) {
415 goal := &core.SessionGoal{ID: "goal-1", Status: core.GoalActive, Objective: "keep building"}
416 gate := newGoalSampleGate(func(call int64) *core.SessionGoal {
417 if call == 2 {
418 // The newer wave seals with no active goal, so its own seal must not
419 // wake the manager.
420 return nil
421 }
422 return goal
423 })
424 defer gate.unblockNow()
425
426 // A manager wake reaches goal sampling only through PersistGoalRuntime, so
427 // recording that callback detects an unintended seal. Returning an error
428 // stops the continuation before it touches the provider.
429 woke := make(chan struct{}, 1)
430 i := &Interactive{agent: core.NewAgent(nil, "model", "", goalToolRegistry())}
431 i.cfg.CurrentGoal = gate.current
432 i.cfg.PersistGoalRuntime = func(*core.SessionGoal) error {
433 select {
434 case woke <- struct{}{}:
435 default:
436 }
437 return errors.New("stop after wake")
438 }
439
440 releaseOld := i.beginCompletionDeliveryHold()
441 released := startGateRelease(releaseOld)
442 enteredGate(t, gate, "older release")
443
444 // A newer wave begins and fully drains inside the older release's sampling
445 // window.
446 awaitGate(t, startGateRelease(func() {
447 releaseNew := i.beginCompletionDeliveryHold()
448 releaseNew()
449 }), "newer hold wave")
450
451 i.completionDeliveryMu.Lock()
452 holds := i.completionDeliveryHolds
453 i.completionDeliveryMu.Unlock()
454 if holds != 0 {
455 t.Fatalf("completionDeliveryHolds = %d, want 0 after the newer wave drained", holds)
456 }
457 gate.unblockNow()
458 awaitGate(t, released, "older release")
459
460 // The older release relocked with zero holds but belongs to the previous
461 // generation, so it must not seal, and must not wake the manager for, the
462 // wave that the newer release already sealed.
463 select {
464 case <-woke:
465 t.Fatal("stale release sealed a drained newer wave")
466 default:
467 }
468
469 // Positive control: the same coordinator state does wake the manager when
470 // the goal really is active, so the check above is meaningful.
471 i.executeCoordinatorActions(i.applyCoordinator(orchestration.Event{Kind: orchestration.EventGoalChanged, GoalActive: true}))
472 i.executeCoordinatorActions(i.applyCoordinator(orchestration.Event{Kind: orchestration.EventManagerFinished}))
473 select {
474 case <-woke:
475 default:
476 t.Fatal("goal-active manager wake no longer reaches goal sampling")
477 }
478}
479
480func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) {
481 for _, tc := range []struct {
482 name string
483 err error
484 compacting bool
485 }{
486 {name: "success"},
487 {name: "failure", err: errors.New("review failed")},
488 {name: "compaction", compacting: true},
489 } {
490 t.Run(tc.name, func(t *testing.T) {
491 ctx, cancel := context.WithCancel(context.Background())
492 defer cancel()
493 ag := core.NewAgent(nil, "model", "", nil)
494 i := &Interactive{agent: ag, busy: true, compacting: tc.compacting, runCtx: ctx}
495 release := i.beginCompletionDeliveryHold()
496 defer release()
497 i.TrackResidentSubagent("finished", "turn-1")
498 i.TrackResidentSubagent("still-running", "turn-2")
499 i.ReportResidentSubagent(subagents.ResidentCompletion{ChildID: "finished", TurnID: "turn-1", Task: "review", Summary: "saved result", Err: tc.err})
500
501 deadline := time.NewTimer(2 * time.Second)
502 defer deadline.Stop()
503 tick := time.NewTicker(time.Millisecond)
504 defer tick.Stop()
505 for {
506 i.mu.Lock()
507 queued := ag.PendingQueuedMessages()
508 if tc.compacting {
509 queued = append([]core.QueuedMessage(nil), i.queued...)
510 }
511 i.mu.Unlock()
512 if len(queued) != 0 {
513 if len(queued) != 1 || !queued[0].HostEvent || !strings.Contains(queued[0].Text, "saved result") || !strings.Contains(queued[0].Text, "[auto-subagents update]") {
514 t.Fatalf("queued completions = %#v", queued)
515 }
516 if !strings.Contains(queued[0].Text, "Still running: still-running") {
517 t.Fatalf("completion update missing active sibling: %q", queued[0].Text)
518 }
519 if tc.err != nil && !strings.Contains(queued[0].Text, tc.err.Error()) {
520 t.Fatalf("missing failure: %q", queued[0].Text)
521 }
522 if i.ensureCompletionTracker().Pending() != 1 {
523 t.Fatal("unfinished child was not retained")
524 }
525 return
526 }
527 select {
528 case <-deadline.C:
529 t.Fatal("completion did not slide in while parent and sibling were active")
530 case <-tick.C:
531 }
532 }
533 })
534 }
535}