fix(agent): await delegated workers before goal continuation

commit: 28c94cebed27340b5612d67aedebe6b32656ab5c
author: brice · 2mo ago · 2026-08-11 07:49 UTC
parent: 59c2524
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +200 -9

M packages/agent/internal/orchestration/coordinator.go +6 -0


@@ -82,6 +82,12 @@ func (c *Coordinator) AcceptsUserInput() bool {
82	82		return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0
83	83		}
84	84
	85	+	// HasPendingWorkers reports whether the current manager wave still owns work.
	86	+	// Callers use it to avoid starting a goal continuation outside the coordinator.
	87	+	func (c *Coordinator) HasPendingWorkers() bool {
	88	+	return c != nil && !c.stopped && c.hasPendingWorkers()
	89	+	}
	90	+
85	91		func (c *Coordinator) Apply(event Event) Result {
86	92		if c == nil \|\| c.stopped {
87	93		return Result{}


@@ -82,6 +82,12 @@ func (c *Coordinator) AcceptsUserInput() bool {
82	return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0	82		return c != nil && !c.stopped && !c.managerActive && c.waveSealed && len(c.workerOrder) != 0
83	}	83		}
84		84
		85	+	// HasPendingWorkers reports whether the current manager wave still owns work.
		86	+	// Callers use it to avoid starting a goal continuation outside the coordinator.
		87	+	func (c *Coordinator) HasPendingWorkers() bool {
		88	+	return c != nil && !c.stopped && c.hasPendingWorkers()
		89	+	}
		90	+
85	func (c *Coordinator) Apply(event Event) Result {	91		func (c *Coordinator) Apply(event Event) Result {
86	if c == nil \|\| c.stopped {	92		if c == nil \|\| c.stopped {
87	return Result{}	93		return Result{}

M packages/agent/modes/goal.go +1 -1


@@ -60,7 +60,7 @@ func (i *Interactive) goalContinuationIfIdle() (provider.Message, bool) {
60	60
61	61		func (i *Interactive) requestGoalContinuationIfIdle(parent context.Context) bool {
62	62		message, ok := i.goalContinuationMessage()
63		-	if !ok {
	63	+	if !ok \|\| i.coordinatorHasPendingWorkers() {
64	64		return false
65	65		}
66	66		i.mu.Lock()


@@ -60,7 +60,7 @@ func (i *Interactive) goalContinuationIfIdle() (provider.Message, bool) {
60			60
61		func (i *Interactive) requestGoalContinuationIfIdle(parent context.Context) bool {	61		func (i *Interactive) requestGoalContinuationIfIdle(parent context.Context) bool {
62		message, ok := i.goalContinuationMessage()	62		message, ok := i.goalContinuationMessage()
63	-	if !ok {	63	+	if !ok \|\| i.coordinatorHasPendingWorkers() {
64		return false	64		return false
65		}	65		}
66		i.mu.Lock()	66		i.mu.Lock()

M packages/agent/modes/goal_test.go +174 -0


@@ -2,12 +2,14 @@ package modes
2	2
3	3		import (
4	4		"context"
	5	+	"encoding/json"
5	6		"errors"
6	7		"strings"
7	8		"sync"
8	9		"testing"
9	10		"time"
10	11
	12	+	"github.com/bnema/zut/packages/agent/subagents"
11	13		agenttools "github.com/bnema/zut/packages/agent/tools"
12	14		"github.com/bnema/zut/packages/core"
13	15		"github.com/bnema/zut/packages/provider"

@@ -46,6 +48,178 @@ func (c *goalContinuationClient) Stream(_ context.Context, req provider.Request)
46	48		return out, nil
47	49		}
48	50
	51	+	type goalWorkerClient struct {
	52	+	mu sync.Mutex
	53	+	requests []provider.Request
	54	+	calls chan int
	55	+	onCall func(call int)
	56	+	}
	57	+
	58	+	func (c *goalWorkerClient) Name() string { return "goal-worker" }
	59	+
	60	+	func (c *goalWorkerClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
	61	+	c.mu.Lock()
	62	+	c.requests = append(c.requests, req)
	63	+	call := len(c.requests)
	64	+	onCall := c.onCall
	65	+	c.mu.Unlock()
	66	+	if onCall != nil {
	67	+	onCall(call)
	68	+	}
	69	+	c.calls <- call
	70	+
	71	+	out := make(chan provider.Event, 1)
	72	+	switch call {
	73	+	case 1:
	74	+	out <- provider.EventDone{
	75	+	Stop: provider.StopToolUse,
	76	+	Message: provider.Message{
	77	+	Role: provider.RoleAssistant,
	78	+	Content: []provider.Content{provider.ToolCallBlock{
	79	+	ID: "worker",
	80	+	Name: "start_worker",
	81	+	Arguments: json.RawMessage(`{}`),
	82	+	}},
	83	+	},
	84	+	}
	85	+	default:
	86	+	out <- provider.EventDone{
	87	+	Stop: provider.StopEnd,
	88	+	Message: provider.Message{
	89	+	Role: provider.RoleAssistant,
	90	+	Content: []provider.Content{provider.TextBlock{Text: "done"}},
	91	+	},
	92	+	}
	93	+	}
	94	+	close(out)
	95	+	return out, nil
	96	+	}
	97	+
	98	+	func (c *goalWorkerClient) request(call int) provider.Request {
	99	+	c.mu.Lock()
	100	+	defer c.mu.Unlock()
	101	+	return c.requests[call-1]
	102	+	}
	103	+
	104	+	type trackWorkerTool struct {
	105	+	interactive *Interactive
	106	+	worker *subagents.Agent
	107	+	}
	108	+
	109	+	func (t *trackWorkerTool) Name() string { return "start_worker" }
	110	+	func (t *trackWorkerTool) Description() string { return "starts a worker" }
	111	+	func (t *trackWorkerTool) Schema() json.RawMessage {
	112	+	return json.RawMessage(`{"type":"object"}`)
	113	+	}
	114	+	func (t *trackWorkerTool) Execute(_ context.Context, _ json.RawMessage, _ func(string)) (core.ToolResult, error) {
	115	+	t.interactive.TrackSubagentWorker(t.worker, t.worker.Task, false)
	116	+	return core.ToolResult{Content: []provider.Content{provider.TextBlock{Text: "worker started"}}}, nil
	117	+	}
	118	+
	119	+	func TestActiveGoalWaitsForWorkerAndDeliversCompletion(t *testing.T) {
	120	+	workerRelease := make(chan struct{})
	121	+	var releaseWorkerOnce sync.Once
	122	+	releaseWorker := func() {
	123	+	releaseWorkerOnce.Do(func() { close(workerRelease) })
	124	+	}
	125	+	workerStarted := make(chan struct{})
	126	+	supervisor := subagents.New(subagents.Config{
	127	+	Root: t.TempDir(),
	128	+	RepoRoot: t.TempDir(),
	129	+	NewRunner: func(*subagents.Agent) subagents.Runner {
	130	+	return subagents.RunnerFunc(func(context.Context, subagents.Sink) error {
	131	+	close(workerStarted)
	132	+	<-workerRelease
	133	+	return nil
	134	+	})
	135	+	},
	136	+	})
	137	+	worker, err := supervisor.Spawn(context.Background(), "inspect the upstream worktree")
	138	+	if err != nil {
	139	+	t.Fatal(err)
	140	+	}
	141	+	<-workerStarted
	142	+	t.Cleanup(func() {
	143	+	releaseWorker()
	144	+	worker.Wait()
	145	+	})
	146	+
	147	+	var goalMu sync.Mutex
	148	+	goal := &core.SessionGoal{Objective: "finish the upstream work", Status: core.GoalActive}
	149	+	currentGoal := func() *core.SessionGoal {
	150	+	goalMu.Lock()
	151	+	defer goalMu.Unlock()
	152	+	return cloneSessionGoal(goal)
	153	+	}
	154	+	persistGoal := func(next *core.SessionGoal) error {
	155	+	goalMu.Lock()
	156	+	goal = cloneSessionGoal(next)
	157	+	goalMu.Unlock()
	158	+	return nil
	159	+	}
	160	+	client := &goalWorkerClient{calls: make(chan int, 3), onCall: func(call int) {
	161	+	if call == 3 {
	162	+	_ = persistGoal(&core.SessionGoal{Objective: "finish the upstream work", Status: core.GoalPaused})
	163	+	}
	164	+	}}
	165	+	workerTool := &trackWorkerTool{worker: worker}
	166	+	interactive := NewInteractive(InteractiveConfig{
	167	+	Agent: core.NewAgent(client, "model", "system", core.Registry{
	168	+	"start_worker": workerTool,
	169	+	agenttools.UpdateGoalToolName: &agenttools.UpdateGoalTool{},
	170	+	}),
	171	+	CurrentGoal: currentGoal,
	172	+	PersistGoal: persistGoal,
	173	+	})
	174	+	workerTool.interactive = interactive
	175	+	interactive.runCtx = context.Background()
	176	+
	177	+	interactive.runSlash(context.Background(), "/goal finish the upstream work")
	178	+	for want := 1; want <= 2; want++ {
	179	+	select {
	180	+	case got := <-client.calls:
	181	+	if got != want {
	182	+	t.Fatalf("provider call = %d, want %d", got, want)
	183	+	}
	184	+	case <-time.After(2 * time.Second):
	185	+	t.Fatalf("provider call %d did not start", want)
	186	+	}
	187	+	}
	188	+	deadline := time.NewTimer(2 * time.Second)
	189	+	defer deadline.Stop()
	190	+	poll := time.NewTicker(time.Millisecond)
	191	+	defer poll.Stop()
	192	+	for !interactive.coordinatorAcceptsUserInput() {
	193	+	select {
	194	+	case <-deadline.C:
	195	+	t.Fatal("active goal started another manager turn while its worker was pending")
	196	+	case <-poll.C:
	197	+	}
	198	+	}
	199	+	select {
	200	+	case got := <-client.calls:
	201	+	t.Fatalf("provider call %d started before the worker completed", got)
	202	+	default:
	203	+	}
	204	+
	205	+	releaseWorker()
	206	+	select {
	207	+	case got := <-client.calls:
	208	+	if got != 3 {
	209	+	t.Fatalf("provider call = %d, want completion delivery call 3", got)
	210	+	}
	211	+	case <-time.After(2 * time.Second):
	212	+	t.Fatal("worker completion did not wake the manager")
	213	+	}
	214	+	request := client.request(3)
	215	+	if len(request.Messages) == 0 {
	216	+	t.Fatal("completion delivery request has no messages")
	217	+	}
	218	+	if got := userMessageText(request.Messages[len(request.Messages)-1]); !strings.Contains(got, "inspect the upstream worktree") {
	219	+	t.Fatalf("completion delivery = %q", got)
	220	+	}
	221	+	}
	222	+
49	223		func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) {
50	224		var mu sync.Mutex
51	225		var goal *core.SessionGoal


@@ -2,12 +2,14 @@ package modes
2		2
3	import (	3		import (
4	"context"	4		"context"
		5	+	"encoding/json"
5	"errors"	6		"errors"
6	"strings"	7		"strings"
7	"sync"	8		"sync"
8	"testing"	9		"testing"
9	"time"	10		"time"
10		11
		12	+	"github.com/bnema/zut/packages/agent/subagents"
11	agenttools "github.com/bnema/zut/packages/agent/tools"	13		agenttools "github.com/bnema/zut/packages/agent/tools"
12	"github.com/bnema/zut/packages/core"	14		"github.com/bnema/zut/packages/core"
13	"github.com/bnema/zut/packages/provider"	15		"github.com/bnema/zut/packages/provider"

@@ -46,6 +48,178 @@ func (c *goalContinuationClient) Stream(_ context.Context, req provider.Request)
46	return out, nil	48		return out, nil
47	}	49		}
48		50
		51	+	type goalWorkerClient struct {
		52	+	mu sync.Mutex
		53	+	requests []provider.Request
		54	+	calls chan int
		55	+	onCall func(call int)
		56	+	}
		57	+
		58	+	func (c *goalWorkerClient) Name() string { return "goal-worker" }
		59	+
		60	+	func (c *goalWorkerClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
		61	+	c.mu.Lock()
		62	+	c.requests = append(c.requests, req)
		63	+	call := len(c.requests)
		64	+	onCall := c.onCall
		65	+	c.mu.Unlock()
		66	+	if onCall != nil {
		67	+	onCall(call)
		68	+	}
		69	+	c.calls <- call
		70	+
		71	+	out := make(chan provider.Event, 1)
		72	+	switch call {
		73	+	case 1:
		74	+	out <- provider.EventDone{
		75	+	Stop: provider.StopToolUse,
		76	+	Message: provider.Message{
		77	+	Role: provider.RoleAssistant,
		78	+	Content: []provider.Content{provider.ToolCallBlock{
		79	+	ID: "worker",
		80	+	Name: "start_worker",
		81	+	Arguments: json.RawMessage(`{}`),
		82	+	}},
		83	+	},
		84	+	}
		85	+	default:
		86	+	out <- provider.EventDone{
		87	+	Stop: provider.StopEnd,
		88	+	Message: provider.Message{
		89	+	Role: provider.RoleAssistant,
		90	+	Content: []provider.Content{provider.TextBlock{Text: "done"}},
		91	+	},
		92	+	}
		93	+	}
		94	+	close(out)
		95	+	return out, nil
		96	+	}
		97	+
		98	+	func (c *goalWorkerClient) request(call int) provider.Request {
		99	+	c.mu.Lock()
		100	+	defer c.mu.Unlock()
		101	+	return c.requests[call-1]
		102	+	}
		103	+
		104	+	type trackWorkerTool struct {
		105	+	interactive *Interactive
		106	+	worker *subagents.Agent
		107	+	}
		108	+
		109	+	func (t *trackWorkerTool) Name() string { return "start_worker" }
		110	+	func (t *trackWorkerTool) Description() string { return "starts a worker" }
		111	+	func (t *trackWorkerTool) Schema() json.RawMessage {
		112	+	return json.RawMessage(`{"type":"object"}`)
		113	+	}
		114	+	func (t *trackWorkerTool) Execute(_ context.Context, _ json.RawMessage, _ func(string)) (core.ToolResult, error) {
		115	+	t.interactive.TrackSubagentWorker(t.worker, t.worker.Task, false)
		116	+	return core.ToolResult{Content: []provider.Content{provider.TextBlock{Text: "worker started"}}}, nil
		117	+	}
		118	+
		119	+	func TestActiveGoalWaitsForWorkerAndDeliversCompletion(t *testing.T) {
		120	+	workerRelease := make(chan struct{})
		121	+	var releaseWorkerOnce sync.Once
		122	+	releaseWorker := func() {
		123	+	releaseWorkerOnce.Do(func() { close(workerRelease) })
		124	+	}
		125	+	workerStarted := make(chan struct{})
		126	+	supervisor := subagents.New(subagents.Config{
		127	+	Root: t.TempDir(),
		128	+	RepoRoot: t.TempDir(),
		129	+	NewRunner: func(*subagents.Agent) subagents.Runner {
		130	+	return subagents.RunnerFunc(func(context.Context, subagents.Sink) error {
		131	+	close(workerStarted)
		132	+	<-workerRelease
		133	+	return nil
		134	+	})
		135	+	},
		136	+	})
		137	+	worker, err := supervisor.Spawn(context.Background(), "inspect the upstream worktree")
		138	+	if err != nil {
		139	+	t.Fatal(err)
		140	+	}
		141	+	<-workerStarted
		142	+	t.Cleanup(func() {
		143	+	releaseWorker()
		144	+	worker.Wait()
		145	+	})
		146	+
		147	+	var goalMu sync.Mutex
		148	+	goal := &core.SessionGoal{Objective: "finish the upstream work", Status: core.GoalActive}
		149	+	currentGoal := func() *core.SessionGoal {
		150	+	goalMu.Lock()
		151	+	defer goalMu.Unlock()
		152	+	return cloneSessionGoal(goal)
		153	+	}
		154	+	persistGoal := func(next *core.SessionGoal) error {
		155	+	goalMu.Lock()
		156	+	goal = cloneSessionGoal(next)
		157	+	goalMu.Unlock()
		158	+	return nil
		159	+	}
		160	+	client := &goalWorkerClient{calls: make(chan int, 3), onCall: func(call int) {
		161	+	if call == 3 {
		162	+	_ = persistGoal(&core.SessionGoal{Objective: "finish the upstream work", Status: core.GoalPaused})
		163	+	}
		164	+	}}
		165	+	workerTool := &trackWorkerTool{worker: worker}
		166	+	interactive := NewInteractive(InteractiveConfig{
		167	+	Agent: core.NewAgent(client, "model", "system", core.Registry{
		168	+	"start_worker": workerTool,
		169	+	agenttools.UpdateGoalToolName: &agenttools.UpdateGoalTool{},
		170	+	}),
		171	+	CurrentGoal: currentGoal,
		172	+	PersistGoal: persistGoal,
		173	+	})
		174	+	workerTool.interactive = interactive
		175	+	interactive.runCtx = context.Background()
		176	+
		177	+	interactive.runSlash(context.Background(), "/goal finish the upstream work")
		178	+	for want := 1; want <= 2; want++ {
		179	+	select {
		180	+	case got := <-client.calls:
		181	+	if got != want {
		182	+	t.Fatalf("provider call = %d, want %d", got, want)
		183	+	}
		184	+	case <-time.After(2 * time.Second):
		185	+	t.Fatalf("provider call %d did not start", want)
		186	+	}
		187	+	}
		188	+	deadline := time.NewTimer(2 * time.Second)
		189	+	defer deadline.Stop()
		190	+	poll := time.NewTicker(time.Millisecond)
		191	+	defer poll.Stop()
		192	+	for !interactive.coordinatorAcceptsUserInput() {
		193	+	select {
		194	+	case <-deadline.C:
		195	+	t.Fatal("active goal started another manager turn while its worker was pending")
		196	+	case <-poll.C:
		197	+	}
		198	+	}
		199	+	select {
		200	+	case got := <-client.calls:
		201	+	t.Fatalf("provider call %d started before the worker completed", got)
		202	+	default:
		203	+	}
		204	+
		205	+	releaseWorker()
		206	+	select {
		207	+	case got := <-client.calls:
		208	+	if got != 3 {
		209	+	t.Fatalf("provider call = %d, want completion delivery call 3", got)
		210	+	}
		211	+	case <-time.After(2 * time.Second):
		212	+	t.Fatal("worker completion did not wake the manager")
		213	+	}
		214	+	request := client.request(3)
		215	+	if len(request.Messages) == 0 {
		216	+	t.Fatal("completion delivery request has no messages")
		217	+	}
		218	+	if got := userMessageText(request.Messages[len(request.Messages)-1]); !strings.Contains(got, "inspect the upstream worktree") {
		219	+	t.Fatalf("completion delivery = %q", got)
		220	+	}
		221	+	}
		222	+
49	func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) {	223		func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) {
50	var mu sync.Mutex	224		var mu sync.Mutex
51	var goal *core.SessionGoal	225		var goal *core.SessionGoal

M packages/agent/modes/interactive.go +19 -8


@@ -7482,7 +7482,7 @@ func (i *Interactive) runCompact(parent context.Context, request compactContinua
7482	7482		handoff, persistHandoff = i.setCompactContinuationLocked(compactContinuationState{reason: continuationReason})
7483	7483		}
7484	7484		continueAutomatically = true
7485		-	case auto && goalActive:
	7485	+	case auto && goalActive && !i.coordinatorHasPendingWorkers():
7486	7486		continueGoal = true
7487	7487		handoff, persistHandoff = i.resetCompactContinuationLocked()
7488	7488		default:

@@ -7855,7 +7855,6 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
7855	7855		if terminalGoalError {
7856	7856		i.updateActiveGoal(core.GoalBlocked, "turn ended with an error")
7857	7857		}
7858		-	goalMessage, goalActive := i.goalContinuationMessage()
7859	7858		i.mu.Lock()
7860	7859		awaitingPre := i.awaitingStartupPre
7861	7860		// A newer explicit prompt may have cleared the handoff while the

@@ -7906,7 +7905,6 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
7906	7905		if !continueStatusRescue && (i.agent == nil \|\| ctx.Err() != nil \|\| err != nil \|\| awaitingPre \|\| hasNext \|\| continueQueued \|\| offer \|\| recoverContextOverflow \|\| (!shouldAutoCompact && !statusRescueActive)) {
7907	7906		handoff, persistHandoff = i.resetCompactContinuationLocked()
7908	7907		}
7909		-	continueGoal := goalActive && !awaitingPre && !hasNext && !continueQueued && !continueStatusRescue && !offer && !recoverContextOverflow && !shouldAutoCompact && err == nil && ctx.Err() == nil && lastStop == provider.StopEnd && lastTurnErr == nil
7910	7908		// The agent run can finish before the paced final text reaches the
7911	7909		// transcript. A compaction replaces that transcript, so it must never
7912	7910		// race the still-live stream frame; otherwise stale deltas can repaint

@@ -7914,8 +7912,8 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
7914	7912		if recoverContextOverflow \|\| shouldAutoCompact {
7915	7913		i.resetStreamingStateLocked()
7916	7914		}
7917		-	alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext \|\| agentQueued > 0 \|\| continueGoal, offer, recoverContextOverflow, shouldAutoCompact)
7918		-	i.busy = hasNext \|\| continueQueued \|\| continueStatusRescue \|\| continueGoal \|\| recoverContextOverflow \|\| shouldAutoCompact
	7915	+	alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext \|\| agentQueued > 0, offer, recoverContextOverflow, shouldAutoCompact)
	7916	+	i.busy = hasNext \|\| continueQueued \|\| continueStatusRescue \|\| recoverContextOverflow \|\| shouldAutoCompact
7919	7917		i.mu.Unlock()
7920	7918		if persistHandoff {
7921	7919		i.persistCompactHandoff(handoff)

@@ -7939,8 +7937,6 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im
7939	7937		i.startTurnRequest(parent, "", nil, true, false)
7940	7938		case continueStatusRescue:
7941	7939		i.startAutoCompactContinuation(parent)
7942		-	case continueGoal:
7943		-	i.startGoalContinuation(parent, goalMessage)
7944	7940		case offer:
7945	7941		i.openRescueDialog(rescueProv, rescueFprov, rescueModel, rescueWhy, prompt, rescueImgs)
7946	7942		case recoverContextOverflow:

@@ -8900,6 +8896,16 @@ func (i *Interactive) coordinatorAcceptsUserInput() bool {
8900	8896		return accepts
8901	8897		}
8902	8898
	8899	+	func (i *Interactive) coordinatorHasPendingWorkers() bool {
	8900	+	if i == nil {
	8901	+	return false
	8902	+	}
	8903	+	i.completionDeliveryMu.Lock()
	8904	+	pending := i.ensureCoordinatorLocked().HasPendingWorkers()
	8905	+	i.completionDeliveryMu.Unlock()
	8906	+	return pending
	8907	+	}
	8908	+
8903	8909		func (i *Interactive) applyCoordinator(event orchestration.Event) []orchestration.Action {
8904	8910		if i == nil {
8905	8911		return nil

@@ -9050,7 +9056,12 @@ func (i *Interactive) releaseCompletionDeliveryHold() {
9050	9056		i.completionDeliveryHolds--
9051	9057		}
9052	9058		if i.completionDeliveryHolds == 0 {
9053		-	actions = i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
	9059	+	// The coordinator must observe the goal before sealing the wave, so a
	9060	+	// pending worker wins the wake decision instead of a direct continuation.
	9061	+	_, goalActive := i.goalContinuationMessage()
	9062	+	coordinator := i.ensureCoordinatorLocked()
	9063	+	coordinator.Apply(orchestration.Event{Kind: orchestration.EventGoalChanged, GoalActive: goalActive})
	9064	+	actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
9054	9065		}
9055	9066		if i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning {
9056	9067		i.completionDeliveryRunning = true

 					handoff, persistHandoff = i.setCompactContinuationLocked(compactContinuationState{reason: continuationReason})
 				}
 				continueAutomatically = true
-			case auto && goalActive:
+			case auto && goalActive && !i.coordinatorHasPendingWorkers():
 				continueGoal = true
 				handoff, persistHandoff = i.resetCompactContinuationLocked()
 			default:
 		if terminalGoalError {
 			i.updateActiveGoal(core.GoalBlocked, "turn ended with an error")
 		}
-		goalMessage, goalActive := i.goalContinuationMessage()
 		i.mu.Lock()
 		awaitingPre := i.awaitingStartupPre
 		// A newer explicit prompt may have cleared the handoff while the
 		if !continueStatusRescue && (i.agent == nil || ctx.Err() != nil || err != nil || awaitingPre || hasNext || continueQueued || offer || recoverContextOverflow || (!shouldAutoCompact && !statusRescueActive)) {
 			handoff, persistHandoff = i.resetCompactContinuationLocked()
 		}
-		continueGoal := goalActive && !awaitingPre && !hasNext && !continueQueued && !continueStatusRescue && !offer && !recoverContextOverflow && !shouldAutoCompact && err == nil && ctx.Err() == nil && lastStop == provider.StopEnd && lastTurnErr == nil
 		// The agent run can finish before the paced final text reaches the
 		// transcript. A compaction replaces that transcript, so it must never
 		// race the still-live stream frame; otherwise stale deltas can repaint
 		if recoverContextOverflow || shouldAutoCompact {
 			i.resetStreamingStateLocked()
 		}
-		alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext || agentQueued > 0 || continueGoal, offer, recoverContextOverflow, shouldAutoCompact)
+		alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext || agentQueued > 0, offer, recoverContextOverflow, shouldAutoCompact)
-		i.busy = hasNext || continueQueued || continueStatusRescue || continueGoal || recoverContextOverflow || shouldAutoCompact
+		i.busy = hasNext || continueQueued || continueStatusRescue || recoverContextOverflow || shouldAutoCompact
 		i.mu.Unlock()
 		if persistHandoff {
 			i.persistCompactHandoff(handoff)
 			i.startTurnRequest(parent, "", nil, true, false)
 		case continueStatusRescue:
 			i.startAutoCompactContinuation(parent)
-		case continueGoal:
-			i.startGoalContinuation(parent, goalMessage)
 		case offer:
 			i.openRescueDialog(rescueProv, rescueFprov, rescueModel, rescueWhy, prompt, rescueImgs)
 		case recoverContextOverflow:
 	return accepts
 }
+func (i *Interactive) coordinatorHasPendingWorkers() bool {
+	if i == nil {
+		return false
+	}
+	i.completionDeliveryMu.Lock()
+	pending := i.ensureCoordinatorLocked().HasPendingWorkers()
+	i.completionDeliveryMu.Unlock()
+	return pending
+}
 func (i *Interactive) applyCoordinator(event orchestration.Event) []orchestration.Action {
 	if i == nil {
 		return nil
 		i.completionDeliveryHolds--
 	}
 	if i.completionDeliveryHolds == 0 {
-		actions = i.ensureCoordinatorLocked().Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
+		// The coordinator must observe the goal before sealing the wave, so a
+		// pending worker wins the wake decision instead of a direct continuation.
+		_, goalActive := i.goalContinuationMessage()
+		coordinator := i.ensureCoordinatorLocked()
+		coordinator.Apply(orchestration.Event{Kind: orchestration.EventGoalChanged, GoalActive: goalActive})
+		actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions
 	}
 	if i.completionDeliveryHolds == 0 && i.completionDeliveryRequest && !i.completionDeliveryRunning {
 		i.completionDeliveryRunning = true