fix(subagents): prevent completion delivery deadlock (#210)

commit: 69d07a579e10d3ea8f148b9cae7f380422db173a
author: bnema · 36d ago · 2026-09-16 17:12 UTC
committer: GitHub · 36d ago · 2026-09-16 17:12 UTC
parent: 969e52a
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

3 files · +504 -10

M packages/agent/modes/interactive.go +6 -0


@@ -871,6 +871,12 @@ type Interactive struct {
871	871		completionDeliveryRunning bool
872	872		completionDeliveryRequest bool
873	873		completionDeliveryHolds int
	874	+	// completionDeliveryGeneration identifies the current hold wave. It is
	875	+	// bumped whenever a new wave starts (a first hold, an implicit one-worker
	876	+	// wave, or a coordinator replacement after cancellation), so a release can
	877	+	// tell whether a newer wave replaced the one it drained while it sampled
	878	+	// goal state.
	879	+	completionDeliveryGeneration uint64
874	880
875	881		// pendingFork is true when the user ran /session fork: the next
876	882		// jump-picker selection should branch off that message instead


@@ -871,6 +871,12 @@ type Interactive struct {
871	completionDeliveryRunning bool	871		completionDeliveryRunning bool
872	completionDeliveryRequest bool	872		completionDeliveryRequest bool
873	completionDeliveryHolds int	873		completionDeliveryHolds int
		874	+	// completionDeliveryGeneration identifies the current hold wave. It is
		875	+	// bumped whenever a new wave starts (a first hold, an implicit one-worker
		876	+	// wave, or a coordinator replacement after cancellation), so a release can
		877	+	// tell whether a newer wave replaced the one it drained while it sampled
		878	+	// goal state.
		879	+	completionDeliveryGeneration uint64
874		880
875	// pendingFork is true when the user ran /session fork: the next	881		// pendingFork is true when the user ran /session fork: the next
876	// jump-picker selection should branch off that message instead	882		// jump-picker selection should branch off that message instead

M packages/agent/modes/interactive_resident_completion_test.go +436 -0


@@ -4,9 +4,12 @@ import (
4	4		"context"
5	5		"errors"
6	6		"strings"
	7	+	"sync"
	8	+	"sync/atomic"
7	9		"testing"
8	10		"time"
9	11
	12	+	"github.com/bnema/zut/packages/agent/internal/orchestration"
10	13		"github.com/bnema/zut/packages/agent/subagents"
11	14		"github.com/bnema/zut/packages/core"
12	15		)

@@ -46,6 +49,439 @@ func TestResidentCompletionBatchDoesNotReportTerminalSiblingAsRunning(t *testing
46	49		}
47	50		}
48	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.
	64	+	func 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.
	113	+	func 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.
	153	+	func 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.
	178	+	type 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	+
	188	+	func 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.
	198	+	func (g *goalSampleGate) unblockNow() {
	199	+	g.once.Do(func() { close(g.unblock) })
	200	+	}
	201	+
	202	+	// current fulfils InteractiveConfig.CurrentGoal.
	203	+	func (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.
	214	+	const 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.
	219	+	func 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.
	230	+	func 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.
	240	+	func 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.
	254	+	func 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.
	320	+	func 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.
	370	+	func 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.
	419	+	func 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	+
49	485		func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) {
50	486		for _, tc := range []struct {
51	487		name string


@@ -4,9 +4,12 @@ import (
4	"context"	4		"context"
5	"errors"	5		"errors"
6	"strings"	6		"strings"
		7	+	"sync"
		8	+	"sync/atomic"
7	"testing"	9		"testing"
8	"time"	10		"time"
9		11
		12	+	"github.com/bnema/zut/packages/agent/internal/orchestration"
10	"github.com/bnema/zut/packages/agent/subagents"	13		"github.com/bnema/zut/packages/agent/subagents"
11	"github.com/bnema/zut/packages/core"	14		"github.com/bnema/zut/packages/core"
12	)	15		)

@@ -46,6 +49,439 @@ func TestResidentCompletionBatchDoesNotReportTerminalSiblingAsRunning(t *testing
46	}	49		}
47	}	50		}
48		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.
		64	+	func 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.
		113	+	func 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.
		153	+	func 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.
		178	+	type 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	+
		188	+	func 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.
		198	+	func (g *goalSampleGate) unblockNow() {
		199	+	g.once.Do(func() { close(g.unblock) })
		200	+	}
		201	+
		202	+	// current fulfils InteractiveConfig.CurrentGoal.
		203	+	func (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.
		214	+	const 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.
		219	+	func 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.
		230	+	func 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.
		240	+	func 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.
		254	+	func 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.
		320	+	func 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.
		370	+	func 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.
		419	+	func 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	+
49	func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) {	485		func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) {
50	for _, tc := range []struct {	486		for _, tc := range []struct {
51	name string	487		name string

M packages/agent/modes/interactive_resident_subagents.go +62 -10


@@ -116,6 +116,10 @@ func (i *Interactive) registerCoordinatorWorker(workerID string) {
116	116		coordinator := i.ensureCoordinatorLocked()
117	117		implicitWave := i.completionDeliveryHolds == 0
118	118		if implicitWave {
	119	+	// A one-worker sealed wave replaces whatever wave an older release may
	120	+	// still be holding a sampled snapshot for. Bump the generation so that
	121	+	// release cannot seal this newer wave when it relocks.
	122	+	i.completionDeliveryGeneration++
119	123		coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
120	124		}
121	125		i.coordinatorWorkerSeq++

@@ -144,6 +148,11 @@ func (i *Interactive) cancelCoordinator() {
144	148		if i.completionTracker != nil {
145	149		i.completionTracker.Reset()
146	150		}
	151	+	// Cancellation also invalidates any snapshot a release sampled before it.
	152	+	// Bump the wave generation rather than zeroing the hold count: outstanding
	153	+	// release closures must still balance that count, and the generation is
	154	+	// what stops a stale snapshot from sealing the replacement coordinator.
	155	+	i.completionDeliveryGeneration++
147	156		i.turnCoordinator = orchestration.New()
148	157		i.coordinatorWorkerIDs = nil
149	158		i.completionDeliveryMu.Unlock()

@@ -221,6 +230,12 @@ func (i *Interactive) beginCompletionDeliveryHold() func() {
221	230		return func() {}
222	231		}
223	232		i.completionDeliveryMu.Lock()
	233	+	if i.completionDeliveryHolds == 0 {
	234	+	// A wave starts only from zero holds. Bumping the generation here lets a
	235	+	// release that drained an older wave detect that a newer wave replaced
	236	+	// it while it sampled goal state.
	237	+	i.completionDeliveryGeneration++
	238	+	}
224	239		i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
225	240		i.completionDeliveryHolds++
226	241		i.completionDeliveryMu.Unlock()

@@ -230,21 +245,58 @@ func (i *Interactive) releaseCompletionDeliveryHold() {
230	245		if i == nil {
231	246		return
232	247		}
233		-	start := false
234		-	var actions []orchestration.Action
235	248		i.completionDeliveryMu.Lock()
	249	+	sealing := false
	250	+	var sealingGeneration uint64
236	251		if i.completionDeliveryHolds > 0 {
237	252		i.completionDeliveryHolds--
	253	+	if i.completionDeliveryHolds == 0 {
	254	+	sealing = true
	255	+	sealingGeneration = i.completionDeliveryGeneration
	256	+	}
238	257		}
239		-	if i.completionDeliveryHolds == 0 {
240		-	// The coordinator must observe the goal before sealing the wave, so a
241		-	// pending worker wins the wake decision instead of a direct continuation.
242		-	// An Esc-armed reassessment prompt suppresses the wake: the next
243		-	// ordinary user message owns the turn, not autonomous work.
244		-	_, goalActive := i.goalContinuationMessage()
245		-	if i.interruptedPromptPending() {
246		-	goalActive = false
	258	+	if !sealing {
	259	+	// A nested or duplicate release only returns a hold; it must not sample
	260	+	// goal state or seal a wave it does not own. Starting pending delivery
	261	+	// stays available because that is the existing behaviour whenever the
	262	+	// wave is already idle.
	263	+	start := i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning
	264	+	if start {
	265	+	i.completionDeliveryRunning = true
	266	+	}
	267	+	i.completionDeliveryMu.Unlock()
	268	+	if start {
	269	+	go i.deliverCompletionUpdates()
247	270		}
	271	+	return
	272	+	}
	273	+	i.completionDeliveryMu.Unlock()
	274	+
	275	+	// Only the release that drained the last hold seals the wave, so this
	276	+	// snapshot belongs to exactly that wave. Read the goal and interruption
	277	+	// state before taking completionDeliveryMu: both helpers acquire i.mu, and
	278	+	// callers legitimately hold i.mu while they read coordinator state, so
	279	+	// acquiring i.mu while holding completionDeliveryMu would invert the lock
	280	+	// order and deadlock.
	281	+	//
	282	+	// The coordinator must observe the goal before sealing the wave, so a
	283	+	// pending worker wins the wake decision instead of a direct continuation.
	284	+	// An Esc-armed reassessment prompt suppresses the wake: the next ordinary
	285	+	// user message owns the turn, not autonomous work.
	286	+	_, goalActive := i.goalContinuationMessage()
	287	+	if i.interruptedPromptPending() {
	288	+	goalActive = false
	289	+	}
	290	+
	291	+	start := false
	292	+	var actions []orchestration.Action
	293	+	i.completionDeliveryMu.Lock()
	294	+	// A newer wave may have replaced this one while goal state was sampled, so
	295	+	// the hold count alone is not enough: a newer hold can begin and fully
	296	+	// drain, leaving the count at zero, and sealing then would re-apply this
	297	+	// stale snapshot over the wave that already sealed. The captured generation
	298	+	// identifies exactly the wave this release drained.
	299	+	if i.completionDeliveryHolds == 0 && i.completionDeliveryGeneration == sealingGeneration {
248	300		coordinator := i.ensureCoordinatorLocked()
249	301		coordinator.Apply(orchestration.Event{Kind: orchestration.EventGoalChanged, GoalActive: goalActive})
250	302		actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions


@@ -116,6 +116,10 @@ func (i *Interactive) registerCoordinatorWorker(workerID string) {
116		coordinator := i.ensureCoordinatorLocked()	116		coordinator := i.ensureCoordinatorLocked()
117		implicitWave := i.completionDeliveryHolds == 0	117		implicitWave := i.completionDeliveryHolds == 0
118		if implicitWave {	118		if implicitWave {
			119	+	// A one-worker sealed wave replaces whatever wave an older release may
			120	+	// still be holding a sampled snapshot for. Bump the generation so that
			121	+	// release cannot seal this newer wave when it relocks.
			122	+	i.completionDeliveryGeneration++
119		coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})	123		coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
120		}	124		}
121		i.coordinatorWorkerSeq++	125		i.coordinatorWorkerSeq++

@@ -144,6 +148,11 @@ func (i *Interactive) cancelCoordinator() {
144		if i.completionTracker != nil {	148		if i.completionTracker != nil {
145		i.completionTracker.Reset()	149		i.completionTracker.Reset()
146		}	150		}
			151	+	// Cancellation also invalidates any snapshot a release sampled before it.
			152	+	// Bump the wave generation rather than zeroing the hold count: outstanding
			153	+	// release closures must still balance that count, and the generation is
			154	+	// what stops a stale snapshot from sealing the replacement coordinator.
			155	+	i.completionDeliveryGeneration++
147		i.turnCoordinator = orchestration.New()	156		i.turnCoordinator = orchestration.New()
148		i.coordinatorWorkerIDs = nil	157		i.coordinatorWorkerIDs = nil
149		i.completionDeliveryMu.Unlock()	158		i.completionDeliveryMu.Unlock()

@@ -221,6 +230,12 @@ func (i *Interactive) beginCompletionDeliveryHold() func() {
221		return func() {}	230		return func() {}
222		}	231		}
223		i.completionDeliveryMu.Lock()	232		i.completionDeliveryMu.Lock()
			233	+	if i.completionDeliveryHolds == 0 {
			234	+	// A wave starts only from zero holds. Bumping the generation here lets a
			235	+	// release that drained an older wave detect that a newer wave replaced
			236	+	// it while it sampled goal state.
			237	+	i.completionDeliveryGeneration++
			238	+	}
224		i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})	239		i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerStarted})
225		i.completionDeliveryHolds++	240		i.completionDeliveryHolds++
226		i.completionDeliveryMu.Unlock()	241		i.completionDeliveryMu.Unlock()

@@ -230,21 +245,58 @@ func (i *Interactive) releaseCompletionDeliveryHold() {
230		if i == nil {	245		if i == nil {
231		return	246		return
232		}	247		}
233	-	start := false
234	-	var actions []orchestration.Action
235		i.completionDeliveryMu.Lock()	248		i.completionDeliveryMu.Lock()
			249	+	sealing := false
			250	+	var sealingGeneration uint64
236		if i.completionDeliveryHolds > 0 {	251		if i.completionDeliveryHolds > 0 {
237		i.completionDeliveryHolds--	252		i.completionDeliveryHolds--
			253	+	if i.completionDeliveryHolds == 0 {
			254	+	sealing = true
			255	+	sealingGeneration = i.completionDeliveryGeneration
			256	+	}
238		}	257		}
239	-	if i.completionDeliveryHolds == 0 {	258	+	if !sealing {
240	-	// The coordinator must observe the goal before sealing the wave, so a	259	+	// A nested or duplicate release only returns a hold; it must not sample
241	-	// pending worker wins the wake decision instead of a direct continuation.	260	+	// goal state or seal a wave it does not own. Starting pending delivery
242	-	// An Esc-armed reassessment prompt suppresses the wake: the next	261	+	// stays available because that is the existing behaviour whenever the
243	-	// ordinary user message owns the turn, not autonomous work.	262	+	// wave is already idle.
244	-	_, goalActive := i.goalContinuationMessage()	263	+	start := i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning
245	-	if i.interruptedPromptPending() {	264	+	if start {
246	-	goalActive = false	265	+	i.completionDeliveryRunning = true
			266	+	}
			267	+	i.completionDeliveryMu.Unlock()
			268	+	if start {
			269	+	go i.deliverCompletionUpdates()
247		}	270		}
			271	+	return
			272	+	}
			273	+	i.completionDeliveryMu.Unlock()
			274	+
			275	+	// Only the release that drained the last hold seals the wave, so this
			276	+	// snapshot belongs to exactly that wave. Read the goal and interruption
			277	+	// state before taking completionDeliveryMu: both helpers acquire i.mu, and
			278	+	// callers legitimately hold i.mu while they read coordinator state, so
			279	+	// acquiring i.mu while holding completionDeliveryMu would invert the lock
			280	+	// order and deadlock.
			281	+	//
			282	+	// The coordinator must observe the goal before sealing the wave, so a
			283	+	// pending worker wins the wake decision instead of a direct continuation.
			284	+	// An Esc-armed reassessment prompt suppresses the wake: the next ordinary
			285	+	// user message owns the turn, not autonomous work.
			286	+	_, goalActive := i.goalContinuationMessage()
			287	+	if i.interruptedPromptPending() {
			288	+	goalActive = false
			289	+	}
			290	+
			291	+	start := false
			292	+	var actions []orchestration.Action
			293	+	i.completionDeliveryMu.Lock()
			294	+	// A newer wave may have replaced this one while goal state was sampled, so
			295	+	// the hold count alone is not enough: a newer hold can begin and fully
			296	+	// drain, leaving the count at zero, and sealing then would re-apply this
			297	+	// stale snapshot over the wave that already sealed. The captured generation
			298	+	// identifies exactly the wave this release drained.
			299	+	if i.completionDeliveryHolds == 0 && i.completionDeliveryGeneration == sealingGeneration {
248		coordinator := i.ensureCoordinatorLocked()	300		coordinator := i.ensureCoordinatorLocked()
249		coordinator.Apply(orchestration.Event{Kind: orchestration.EventGoalChanged, GoalActive: goalActive})	301		coordinator.Apply(orchestration.Event{Kind: orchestration.EventGoalChanged, GoalActive: goalActive})
250		actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions	302		actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions