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