fix(subagents): report parallel completions immediately (#203)
* fix(subagents): report parallel completions immediately * fix(subagents): filter completed sibling status * test(subagents): assert active sibling updates
- commit
- 9897fd745c418edc7fdb8c2ba2f677ddbe2ac987
- author
- bnema · · 2026-09-11 09:36 UTC
- committer
- GitHub · · 2026-09-11 09:36 UTC
- parent
- 0d89d16
- signature
- GPG Unverified
B5690EEEBB952194The key is unknown - refs
- tag: v0.36.0
7 files · +169 -26
M docs/subagents.md +5 -3
| @@ -234,9 +234,11 @@ Undelivered results remain queued after a provider or compaction error for the | |||
| 234 | 234 | next turn; they do not automatically retry the failed request. Explicit | |
| 235 | 235 | cancellation clears pending queue entries. The queue is in memory, while the | |
| 236 | 236 | child’s saved result remains available through `subagent_status`. | |
| 237 | - | When the primary is idle, remaining children are collected into one completion | |
| 238 | - | wave. Headless orchestration still waits for every accepted child before its | |
| 239 | - | next parent turn. | |
| 237 | + | When the primary is idle, each child completion immediately starts the next | |
| 238 | + | parent turn instead of waiting for its siblings. The host update names the | |
| 239 | + | finished child and lists sibling agent IDs that are still running. A completion | |
| 240 | + | that arrives while that parent turn is active slides into its queue. Headless | |
| 241 | + | orchestration still waits for every accepted child before its next parent turn. | |
| 240 | 242 | Successful completions include the final visible assistant summary, capped at | |
| 241 | 243 | 256 KiB; open the child session for the complete durable transcript. | |
| 242 | 244 | ||
| @@ -234,9 +234,11 @@ Undelivered results remain queued after a provider or compaction error for the | |||||
| 234 | next turn; they do not automatically retry the failed request. Explicit | 234 | next turn; they do not automatically retry the failed request. Explicit | ||
| 235 | cancellation clears pending queue entries. The queue is in memory, while the | 235 | cancellation clears pending queue entries. The queue is in memory, while the | ||
| 236 | child’s saved result remains available through `subagent_status`. | 236 | child’s saved result remains available through `subagent_status`. | ||
| 237 | - | When the primary is idle, remaining children are collected into one completion | 237 | + | When the primary is idle, each child completion immediately starts the next |
| 238 | - | wave. Headless orchestration still waits for every accepted child before its | 238 | + | parent turn instead of waiting for its siblings. The host update names the |
| 239 | - | next parent turn. | 239 | + | finished child and lists sibling agent IDs that are still running. A completion |
| 240 | + | that arrives while that parent turn is active slides into its queue. Headless | |||
| 241 | + | orchestration still waits for every accepted child before its next parent turn. | |||
| 240 | Successful completions include the final visible assistant summary, capped at | 242 | Successful completions include the final visible assistant summary, capped at | ||
| 241 | 256 KiB; open the child session for the complete durable transcript. | 243 | 256 KiB; open the child session for the complete durable transcript. | ||
| 242 | 244 | ||||
M packages/agent/internal/orchestration/coordinator.go +63 -11
| @@ -29,6 +29,7 @@ type queuedUserInput struct { | |||
| 29 | 29 | } | |
| 30 | 30 | ||
| 31 | 31 | type worker struct { | |
| 32 | + | agentID string | |
| 32 | 33 | delivered bool | |
| 33 | 34 | terminal bool | |
| 34 | 35 | completion subagents.Completion | |
| @@ -50,6 +51,7 @@ const ( | |||
| 50 | 51 | type Event struct { | |
| 51 | 52 | Kind EventKind | |
| 52 | 53 | WorkerID string | |
| 54 | + | AgentID string | |
| 53 | 55 | Completion subagents.Completion | |
| 54 | 56 | Text string | |
| 55 | 57 | Images []provider.ImageBlock | |
| @@ -74,11 +76,12 @@ const ( | |||
| 74 | 76 | ) | |
| 75 | 77 | ||
| 76 | 78 | type Action struct { | |
| 77 | - | Kind ActionKind | |
| 78 | - | Reason WakeReason | |
| 79 | - | Text string | |
| 80 | - | Images []provider.ImageBlock | |
| 81 | - | Completions []subagents.Completion | |
| 79 | + | Kind ActionKind | |
| 80 | + | Reason WakeReason | |
| 81 | + | Text string | |
| 82 | + | Images []provider.ImageBlock | |
| 83 | + | Completions []subagents.Completion | |
| 84 | + | ActiveAgentIDs []string | |
| 82 | 85 | } | |
| 83 | 86 | ||
| 84 | 87 | type Result struct{ Actions []Action } | |
| @@ -108,8 +111,7 @@ func (c *Coordinator) Apply(event Event) Result { | |||
| 108 | 111 | case EventManagerStarted: | |
| 109 | 112 | c.managerActive = true | |
| 110 | 113 | c.waveSealed = false | |
| 111 | - | c.workers = make(map[string]*worker) | |
| 112 | - | c.workerOrder = nil | |
| 114 | + | c.dropDeliveredWorkers() | |
| 113 | 115 | case EventManagerFinished: | |
| 114 | 116 | c.managerActive = false | |
| 115 | 117 | c.waveSealed = true | |
| @@ -122,7 +124,7 @@ func (c *Coordinator) Apply(event Event) Result { | |||
| 122 | 124 | return Result{} | |
| 123 | 125 | } | |
| 124 | 126 | if _, exists := c.workers[event.WorkerID]; !exists { | |
| 125 | - | c.workers[event.WorkerID] = &worker{} | |
| 127 | + | c.workers[event.WorkerID] = &worker{agentID: event.AgentID} | |
| 126 | 128 | c.workerOrder = append(c.workerOrder, event.WorkerID) | |
| 127 | 129 | } | |
| 128 | 130 | case EventWorkerFinished: | |
| @@ -134,10 +136,30 @@ func (c *Coordinator) Apply(event Event) Result { | |||
| 134 | 136 | w.completion = event.Completion | |
| 135 | 137 | if c.managerActive { | |
| 136 | 138 | w.delivered = true | |
| 137 | - | return Result{Actions: []Action{{Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{event.Completion}}}} | |
| 139 | + | return Result{Actions: []Action{{ | |
| 140 | + | Kind: ActionQueueManager, | |
| 141 | + | Reason: WakeWorkers, | |
| 142 | + | Completions: []subagents.Completion{event.Completion}, | |
| 143 | + | ActiveAgentIDs: c.activeAgentIDs(), | |
| 144 | + | }}} | |
| 138 | 145 | } | |
| 139 | - | if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() { | |
| 140 | - | return c.wakeIfIdle() | |
| 146 | + | if c.waveSealed { | |
| 147 | + | w.delivered = true | |
| 148 | + | action := Action{ | |
| 149 | + | Kind: ActionRunManager, | |
| 150 | + | Reason: WakeWorkers, | |
| 151 | + | Completions: []subagents.Completion{event.Completion}, | |
| 152 | + | ActiveAgentIDs: c.activeAgentIDs(), | |
| 153 | + | } | |
| 154 | + | if len(c.queuedUser) != 0 { | |
| 155 | + | input := c.queuedUser[0] | |
| 156 | + | c.queuedUser = c.queuedUser[1:] | |
| 157 | + | action.Reason = WakeUser | |
| 158 | + | action.Text = input.text | |
| 159 | + | action.Images = input.images | |
| 160 | + | } | |
| 161 | + | c.managerActive = true | |
| 162 | + | return Result{Actions: []Action{action}} | |
| 141 | 163 | } | |
| 142 | 164 | case EventUserInput: | |
| 143 | 165 | if event.Text == "" && len(event.Images) == 0 { | |
| @@ -200,6 +222,36 @@ func (c *Coordinator) hasPendingWorkers() bool { | |||
| 200 | 222 | return false | |
| 201 | 223 | } | |
| 202 | 224 | ||
| 225 | + | func (c *Coordinator) activeAgentIDs() []string { | |
| 226 | + | ids := make([]string, 0, len(c.workerOrder)) | |
| 227 | + | seen := make(map[string]struct{}) | |
| 228 | + | for _, workerID := range c.workerOrder { | |
| 229 | + | worker := c.workers[workerID] | |
| 230 | + | if worker == nil || worker.terminal || worker.agentID == "" { | |
| 231 | + | continue | |
| 232 | + | } | |
| 233 | + | if _, exists := seen[worker.agentID]; exists { | |
| 234 | + | continue | |
| 235 | + | } | |
| 236 | + | seen[worker.agentID] = struct{}{} | |
| 237 | + | ids = append(ids, worker.agentID) | |
| 238 | + | } | |
| 239 | + | return ids | |
| 240 | + | } | |
| 241 | + | ||
| 242 | + | func (c *Coordinator) dropDeliveredWorkers() { | |
| 243 | + | order := c.workerOrder[:0] | |
| 244 | + | for _, workerID := range c.workerOrder { | |
| 245 | + | worker := c.workers[workerID] | |
| 246 | + | if worker == nil || worker.delivered { | |
| 247 | + | delete(c.workers, workerID) | |
| 248 | + | continue | |
| 249 | + | } | |
| 250 | + | order = append(order, workerID) | |
| 251 | + | } | |
| 252 | + | c.workerOrder = order | |
| 253 | + | } | |
| 254 | + | ||
| 203 | 255 | func (c *Coordinator) completedWorkers() []subagents.Completion { | |
| 204 | 256 | if c.hasPendingWorkers() { | |
| 205 | 257 | return nil | |
| @@ -29,6 +29,7 @@ type queuedUserInput struct { | |||||
| 29 | } | 29 | } | ||
| 30 | 30 | ||||
| 31 | type worker struct { | 31 | type worker struct { | ||
| 32 | + | agentID string | |||
| 32 | delivered bool | 33 | delivered bool | ||
| 33 | terminal bool | 34 | terminal bool | ||
| 34 | completion subagents.Completion | 35 | completion subagents.Completion | ||
| @@ -50,6 +51,7 @@ const ( | |||||
| 50 | type Event struct { | 51 | type Event struct { | ||
| 51 | Kind EventKind | 52 | Kind EventKind | ||
| 52 | WorkerID string | 53 | WorkerID string | ||
| 54 | + | AgentID string | |||
| 53 | Completion subagents.Completion | 55 | Completion subagents.Completion | ||
| 54 | Text string | 56 | Text string | ||
| 55 | Images []provider.ImageBlock | 57 | Images []provider.ImageBlock | ||
| @@ -74,11 +76,12 @@ const ( | |||||
| 74 | ) | 76 | ) | ||
| 75 | 77 | ||||
| 76 | type Action struct { | 78 | type Action struct { | ||
| 77 | - | Kind ActionKind | 79 | + | Kind ActionKind |
| 78 | - | Reason WakeReason | 80 | + | Reason WakeReason |
| 79 | - | Text string | 81 | + | Text string |
| 80 | - | Images []provider.ImageBlock | 82 | + | Images []provider.ImageBlock |
| 81 | - | Completions []subagents.Completion | 83 | + | Completions []subagents.Completion |
| 84 | + | ActiveAgentIDs []string | |||
| 82 | } | 85 | } | ||
| 83 | 86 | ||||
| 84 | type Result struct{ Actions []Action } | 87 | type Result struct{ Actions []Action } | ||
| @@ -108,8 +111,7 @@ func (c *Coordinator) Apply(event Event) Result { | |||||
| 108 | case EventManagerStarted: | 111 | case EventManagerStarted: | ||
| 109 | c.managerActive = true | 112 | c.managerActive = true | ||
| 110 | c.waveSealed = false | 113 | c.waveSealed = false | ||
| 111 | - | c.workers = make(map[string]*worker) | 114 | + | c.dropDeliveredWorkers() |
| 112 | - | c.workerOrder = nil | |||
| 113 | case EventManagerFinished: | 115 | case EventManagerFinished: | ||
| 114 | c.managerActive = false | 116 | c.managerActive = false | ||
| 115 | c.waveSealed = true | 117 | c.waveSealed = true | ||
| @@ -122,7 +124,7 @@ func (c *Coordinator) Apply(event Event) Result { | |||||
| 122 | return Result{} | 124 | return Result{} | ||
| 123 | } | 125 | } | ||
| 124 | if _, exists := c.workers[event.WorkerID]; !exists { | 126 | if _, exists := c.workers[event.WorkerID]; !exists { | ||
| 125 | - | c.workers[event.WorkerID] = &worker{} | 127 | + | c.workers[event.WorkerID] = &worker{agentID: event.AgentID} |
| 126 | c.workerOrder = append(c.workerOrder, event.WorkerID) | 128 | c.workerOrder = append(c.workerOrder, event.WorkerID) | ||
| 127 | } | 129 | } | ||
| 128 | case EventWorkerFinished: | 130 | case EventWorkerFinished: | ||
| @@ -134,10 +136,30 @@ func (c *Coordinator) Apply(event Event) Result { | |||||
| 134 | w.completion = event.Completion | 136 | w.completion = event.Completion | ||
| 135 | if c.managerActive { | 137 | if c.managerActive { | ||
| 136 | w.delivered = true | 138 | w.delivered = true | ||
| 137 | - | return Result{Actions: []Action{{Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{event.Completion}}}} | 139 | + | return Result{Actions: []Action{{ |
| 140 | + | Kind: ActionQueueManager, | |||
| 141 | + | Reason: WakeWorkers, | |||
| 142 | + | Completions: []subagents.Completion{event.Completion}, | |||
| 143 | + | ActiveAgentIDs: c.activeAgentIDs(), | |||
| 144 | + | }}} | |||
| 138 | } | 145 | } | ||
| 139 | - | if !c.managerActive && c.waveSealed && !c.hasPendingWorkers() { | 146 | + | if c.waveSealed { |
| 140 | - | return c.wakeIfIdle() | 147 | + | w.delivered = true |
| 148 | + | action := Action{ | |||
| 149 | + | Kind: ActionRunManager, | |||
| 150 | + | Reason: WakeWorkers, | |||
| 151 | + | Completions: []subagents.Completion{event.Completion}, | |||
| 152 | + | ActiveAgentIDs: c.activeAgentIDs(), | |||
| 153 | + | } | |||
| 154 | + | if len(c.queuedUser) != 0 { | |||
| 155 | + | input := c.queuedUser[0] | |||
| 156 | + | c.queuedUser = c.queuedUser[1:] | |||
| 157 | + | action.Reason = WakeUser | |||
| 158 | + | action.Text = input.text | |||
| 159 | + | action.Images = input.images | |||
| 160 | + | } | |||
| 161 | + | c.managerActive = true | |||
| 162 | + | return Result{Actions: []Action{action}} | |||
| 141 | } | 163 | } | ||
| 142 | case EventUserInput: | 164 | case EventUserInput: | ||
| 143 | if event.Text == "" && len(event.Images) == 0 { | 165 | if event.Text == "" && len(event.Images) == 0 { | ||
| @@ -200,6 +222,36 @@ func (c *Coordinator) hasPendingWorkers() bool { | |||||
| 200 | return false | 222 | return false | ||
| 201 | } | 223 | } | ||
| 202 | 224 | ||||
| 225 | + | func (c *Coordinator) activeAgentIDs() []string { | |||
| 226 | + | ids := make([]string, 0, len(c.workerOrder)) | |||
| 227 | + | seen := make(map[string]struct{}) | |||
| 228 | + | for _, workerID := range c.workerOrder { | |||
| 229 | + | worker := c.workers[workerID] | |||
| 230 | + | if worker == nil || worker.terminal || worker.agentID == "" { | |||
| 231 | + | continue | |||
| 232 | + | } | |||
| 233 | + | if _, exists := seen[worker.agentID]; exists { | |||
| 234 | + | continue | |||
| 235 | + | } | |||
| 236 | + | seen[worker.agentID] = struct{}{} | |||
| 237 | + | ids = append(ids, worker.agentID) | |||
| 238 | + | } | |||
| 239 | + | return ids | |||
| 240 | + | } | |||
| 241 | + | ||||
| 242 | + | func (c *Coordinator) dropDeliveredWorkers() { | |||
| 243 | + | order := c.workerOrder[:0] | |||
| 244 | + | for _, workerID := range c.workerOrder { | |||
| 245 | + | worker := c.workers[workerID] | |||
| 246 | + | if worker == nil || worker.delivered { | |||
| 247 | + | delete(c.workers, workerID) | |||
| 248 | + | continue | |||
| 249 | + | } | |||
| 250 | + | order = append(order, workerID) | |||
| 251 | + | } | |||
| 252 | + | c.workerOrder = order | |||
| 253 | + | } | |||
| 254 | + | ||||
| 203 | func (c *Coordinator) completedWorkers() []subagents.Completion { | 255 | func (c *Coordinator) completedWorkers() []subagents.Completion { | ||
| 204 | if c.hasPendingWorkers() { | 256 | if c.hasPendingWorkers() { | ||
| 205 | return nil | 257 | return nil | ||
M packages/agent/internal/orchestration/coordinator_test.go +21 -9
| @@ -7,27 +7,31 @@ import ( | |||
| 7 | 7 | "github.com/bnema/zut/packages/provider" | |
| 8 | 8 | ) | |
| 9 | 9 | ||
| 10 | - | func TestCoordinatorSealsWorkerWaveAndWakesOnce(t *testing.T) { | |
| 10 | + | func TestCoordinatorReportsEachSealedWorkerCompletionImmediately(t *testing.T) { | |
| 11 | 11 | c := New() | |
| 12 | 12 | c.Apply(Event{Kind: EventManagerStarted}) | |
| 13 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"}) | |
| 14 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"}) | |
| 13 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a#1", AgentID: "a"}) | |
| 14 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b#2", AgentID: "b"}) | |
| 15 | 15 | assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}}) | |
| 16 | - | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b", Completion: subagents.Completion{AgentID: "b"}}), nil) | |
| 17 | - | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: subagents.Completion{AgentID: "a"}}), []Action{{ | |
| 16 | + | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b#2", Completion: subagents.Completion{AgentID: "b"}}), []Action{{ | |
| 18 | 17 | Kind: ActionRunManager, Reason: WakeWorkers, | |
| 19 | - | Completions: []subagents.Completion{{AgentID: "a"}, {AgentID: "b"}}, | |
| 18 | + | Completions: []subagents.Completion{{AgentID: "b"}}, ActiveAgentIDs: []string{"a"}, | |
| 19 | + | }}) | |
| 20 | + | c.Apply(Event{Kind: EventManagerStarted}) | |
| 21 | + | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a#1", Completion: subagents.Completion{AgentID: "a"}}), []Action{{ | |
| 22 | + | Kind: ActionQueueManager, Reason: WakeWorkers, | |
| 23 | + | Completions: []subagents.Completion{{AgentID: "a"}}, | |
| 20 | 24 | }}) | |
| 21 | 25 | } | |
| 22 | 26 | ||
| 23 | 27 | func TestCoordinatorSlidesCompletionIntoActiveManagerOnce(t *testing.T) { | |
| 24 | 28 | c := New() | |
| 25 | 29 | c.Apply(Event{Kind: EventManagerStarted}) | |
| 26 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"}) | |
| 27 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"}) | |
| 30 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a", AgentID: "a"}) | |
| 31 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b", AgentID: "b"}) | |
| 28 | 32 | completion := subagents.Completion{AgentID: "a", Status: "completed"} | |
| 29 | 33 | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), []Action{{ | |
| 30 | - | Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{completion}, | |
| 34 | + | Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{completion}, ActiveAgentIDs: []string{"b"}, | |
| 31 | 35 | }}) | |
| 32 | 36 | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), nil) | |
| 33 | 37 | assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}}) | |
| @@ -97,5 +101,13 @@ func assertActions(t *testing.T, got Result, want []Action) { | |||
| 97 | 101 | t.Fatalf("action %d completion %d = %#v, want %#v", i, j, got.Actions[i].Completions[j], want[i].Completions[j]) | |
| 98 | 102 | } | |
| 99 | 103 | } | |
| 104 | + | if len(got.Actions[i].ActiveAgentIDs) != len(want[i].ActiveAgentIDs) { | |
| 105 | + | t.Fatalf("action %d active agents = %#v, want %#v", i, got.Actions[i].ActiveAgentIDs, want[i].ActiveAgentIDs) | |
| 106 | + | } | |
| 107 | + | for j := range want[i].ActiveAgentIDs { | |
| 108 | + | if got.Actions[i].ActiveAgentIDs[j] != want[i].ActiveAgentIDs[j] { | |
| 109 | + | t.Fatalf("action %d active agent %d = %q, want %q", i, j, got.Actions[i].ActiveAgentIDs[j], want[i].ActiveAgentIDs[j]) | |
| 110 | + | } | |
| 111 | + | } | |
| 100 | 112 | } | |
| 101 | 113 | } | |
| @@ -7,27 +7,31 @@ import ( | |||||
| 7 | "github.com/bnema/zut/packages/provider" | 7 | "github.com/bnema/zut/packages/provider" | ||
| 8 | ) | 8 | ) | ||
| 9 | 9 | ||||
| 10 | - | func TestCoordinatorSealsWorkerWaveAndWakesOnce(t *testing.T) { | 10 | + | func TestCoordinatorReportsEachSealedWorkerCompletionImmediately(t *testing.T) { |
| 11 | c := New() | 11 | c := New() | ||
| 12 | c.Apply(Event{Kind: EventManagerStarted}) | 12 | c.Apply(Event{Kind: EventManagerStarted}) | ||
| 13 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"}) | 13 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a#1", AgentID: "a"}) |
| 14 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"}) | 14 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b#2", AgentID: "b"}) |
| 15 | assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}}) | 15 | assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}}) | ||
| 16 | - | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b", Completion: subagents.Completion{AgentID: "b"}}), nil) | 16 | + | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "b#2", Completion: subagents.Completion{AgentID: "b"}}), []Action{{ |
| 17 | - | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: subagents.Completion{AgentID: "a"}}), []Action{{ | |||
| 18 | Kind: ActionRunManager, Reason: WakeWorkers, | 17 | Kind: ActionRunManager, Reason: WakeWorkers, | ||
| 19 | - | Completions: []subagents.Completion{{AgentID: "a"}, {AgentID: "b"}}, | 18 | + | Completions: []subagents.Completion{{AgentID: "b"}}, ActiveAgentIDs: []string{"a"}, |
| 19 | + | }}) | |||
| 20 | + | c.Apply(Event{Kind: EventManagerStarted}) | |||
| 21 | + | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a#1", Completion: subagents.Completion{AgentID: "a"}}), []Action{{ | |||
| 22 | + | Kind: ActionQueueManager, Reason: WakeWorkers, | |||
| 23 | + | Completions: []subagents.Completion{{AgentID: "a"}}, | |||
| 20 | }}) | 24 | }}) | ||
| 21 | } | 25 | } | ||
| 22 | 26 | ||||
| 23 | func TestCoordinatorSlidesCompletionIntoActiveManagerOnce(t *testing.T) { | 27 | func TestCoordinatorSlidesCompletionIntoActiveManagerOnce(t *testing.T) { | ||
| 24 | c := New() | 28 | c := New() | ||
| 25 | c.Apply(Event{Kind: EventManagerStarted}) | 29 | c.Apply(Event{Kind: EventManagerStarted}) | ||
| 26 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a"}) | 30 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "a", AgentID: "a"}) |
| 27 | - | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b"}) | 31 | + | c.Apply(Event{Kind: EventWorkerRegistered, WorkerID: "b", AgentID: "b"}) |
| 28 | completion := subagents.Completion{AgentID: "a", Status: "completed"} | 32 | completion := subagents.Completion{AgentID: "a", Status: "completed"} | ||
| 29 | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), []Action{{ | 33 | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), []Action{{ | ||
| 30 | - | Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{completion}, | 34 | + | Kind: ActionQueueManager, Reason: WakeWorkers, Completions: []subagents.Completion{completion}, ActiveAgentIDs: []string{"b"}, |
| 31 | }}) | 35 | }}) | ||
| 32 | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), nil) | 36 | assertActions(t, c.Apply(Event{Kind: EventWorkerFinished, WorkerID: "a", Completion: completion}), nil) | ||
| 33 | assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}}) | 37 | assertActions(t, c.Apply(Event{Kind: EventManagerFinished}), []Action{{Kind: ActionWait}}) | ||
| @@ -97,5 +101,13 @@ func assertActions(t *testing.T, got Result, want []Action) { | |||||
| 97 | t.Fatalf("action %d completion %d = %#v, want %#v", i, j, got.Actions[i].Completions[j], want[i].Completions[j]) | 101 | t.Fatalf("action %d completion %d = %#v, want %#v", i, j, got.Actions[i].Completions[j], want[i].Completions[j]) | ||
| 98 | } | 102 | } | ||
| 99 | } | 103 | } | ||
| 104 | + | if len(got.Actions[i].ActiveAgentIDs) != len(want[i].ActiveAgentIDs) { | |||
| 105 | + | t.Fatalf("action %d active agents = %#v, want %#v", i, got.Actions[i].ActiveAgentIDs, want[i].ActiveAgentIDs) | |||
| 106 | + | } | |||
| 107 | + | for j := range want[i].ActiveAgentIDs { | |||
| 108 | + | if got.Actions[i].ActiveAgentIDs[j] != want[i].ActiveAgentIDs[j] { | |||
| 109 | + | t.Fatalf("action %d active agent %d = %q, want %q", i, j, got.Actions[i].ActiveAgentIDs[j], want[i].ActiveAgentIDs[j]) | |||
| 110 | + | } | |||
| 111 | + | } | |||
| 100 | } | 112 | } | ||
| 101 | } | 113 | } | ||
M packages/agent/modes/interactive_resident_completion_test.go +38 -0
| @@ -11,6 +11,41 @@ import ( | |||
| 11 | 11 | "github.com/bnema/zut/packages/core" | |
| 12 | 12 | ) | |
| 13 | 13 | ||
| 14 | + | func TestResidentCompletionBatchDoesNotReportTerminalSiblingAsRunning(t *testing.T) { | |
| 15 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 16 | + | defer cancel() | |
| 17 | + | ag := core.NewAgent(nil, "model", "", nil) | |
| 18 | + | i := &Interactive{agent: ag, busy: true, runCtx: ctx} | |
| 19 | + | release := i.beginCompletionDeliveryHold() | |
| 20 | + | defer release() | |
| 21 | + | i.TrackResidentSubagent("first", "turn-1") | |
| 22 | + | i.TrackResidentSubagent("second", "turn-2") | |
| 23 | + | ||
| 24 | + | tracker := i.ensureCompletionTracker() | |
| 25 | + | tracker.Report(subagents.Completion{AgentID: "first", TurnID: "turn-1", Status: "completed"}) | |
| 26 | + | tracker.Report(subagents.Completion{AgentID: "second", TurnID: "turn-2", Status: "completed"}) | |
| 27 | + | i.requestCompletionDelivery() | |
| 28 | + | ||
| 29 | + | deadline := time.NewTimer(2 * time.Second) | |
| 30 | + | defer deadline.Stop() | |
| 31 | + | for { | |
| 32 | + | queued := ag.PendingQueuedMessages() | |
| 33 | + | if len(queued) == 2 { | |
| 34 | + | for _, message := range queued { | |
| 35 | + | if strings.Contains(message.Text, "Still running:") { | |
| 36 | + | t.Fatalf("terminal sibling reported as running: %q", message.Text) | |
| 37 | + | } | |
| 38 | + | } | |
| 39 | + | return | |
| 40 | + | } | |
| 41 | + | select { | |
| 42 | + | case <-deadline.C: | |
| 43 | + | t.Fatalf("queued completions = %#v, want 2", queued) | |
| 44 | + | case <-time.After(time.Millisecond): | |
| 45 | + | } | |
| 46 | + | } | |
| 47 | + | } | |
| 48 | + | ||
| 14 | 49 | func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) { | |
| 15 | 50 | for _, tc := range []struct { | |
| 16 | 51 | name string | |
| @@ -47,6 +82,9 @@ func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) { | |||
| 47 | 82 | if len(queued) != 1 || !queued[0].HostEvent || !strings.Contains(queued[0].Text, "saved result") || !strings.Contains(queued[0].Text, "[auto-subagents update]") { | |
| 48 | 83 | t.Fatalf("queued completions = %#v", queued) | |
| 49 | 84 | } | |
| 85 | + | if !strings.Contains(queued[0].Text, "Still running: still-running") { | |
| 86 | + | t.Fatalf("completion update missing active sibling: %q", queued[0].Text) | |
| 87 | + | } | |
| 50 | 88 | if tc.err != nil && !strings.Contains(queued[0].Text, tc.err.Error()) { | |
| 51 | 89 | t.Fatalf("missing failure: %q", queued[0].Text) | |
| 52 | 90 | } | |
| @@ -11,6 +11,41 @@ import ( | |||||
| 11 | "github.com/bnema/zut/packages/core" | 11 | "github.com/bnema/zut/packages/core" | ||
| 12 | ) | 12 | ) | ||
| 13 | 13 | ||||
| 14 | + | func TestResidentCompletionBatchDoesNotReportTerminalSiblingAsRunning(t *testing.T) { | |||
| 15 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 16 | + | defer cancel() | |||
| 17 | + | ag := core.NewAgent(nil, "model", "", nil) | |||
| 18 | + | i := &Interactive{agent: ag, busy: true, runCtx: ctx} | |||
| 19 | + | release := i.beginCompletionDeliveryHold() | |||
| 20 | + | defer release() | |||
| 21 | + | i.TrackResidentSubagent("first", "turn-1") | |||
| 22 | + | i.TrackResidentSubagent("second", "turn-2") | |||
| 23 | + | ||||
| 24 | + | tracker := i.ensureCompletionTracker() | |||
| 25 | + | tracker.Report(subagents.Completion{AgentID: "first", TurnID: "turn-1", Status: "completed"}) | |||
| 26 | + | tracker.Report(subagents.Completion{AgentID: "second", TurnID: "turn-2", Status: "completed"}) | |||
| 27 | + | i.requestCompletionDelivery() | |||
| 28 | + | ||||
| 29 | + | deadline := time.NewTimer(2 * time.Second) | |||
| 30 | + | defer deadline.Stop() | |||
| 31 | + | for { | |||
| 32 | + | queued := ag.PendingQueuedMessages() | |||
| 33 | + | if len(queued) == 2 { | |||
| 34 | + | for _, message := range queued { | |||
| 35 | + | if strings.Contains(message.Text, "Still running:") { | |||
| 36 | + | t.Fatalf("terminal sibling reported as running: %q", message.Text) | |||
| 37 | + | } | |||
| 38 | + | } | |||
| 39 | + | return | |||
| 40 | + | } | |||
| 41 | + | select { | |||
| 42 | + | case <-deadline.C: | |||
| 43 | + | t.Fatalf("queued completions = %#v, want 2", queued) | |||
| 44 | + | case <-time.After(time.Millisecond): | |||
| 45 | + | } | |||
| 46 | + | } | |||
| 47 | + | } | |||
| 48 | + | ||||
| 14 | func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) { | 49 | func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) { | ||
| 15 | for _, tc := range []struct { | 50 | for _, tc := range []struct { | ||
| 16 | name string | 51 | name string | ||
| @@ -47,6 +82,9 @@ func TestResidentCompletionSlidesIntoBusyParent(t *testing.T) { | |||||
| 47 | if len(queued) != 1 || !queued[0].HostEvent || !strings.Contains(queued[0].Text, "saved result") || !strings.Contains(queued[0].Text, "[auto-subagents update]") { | 82 | if len(queued) != 1 || !queued[0].HostEvent || !strings.Contains(queued[0].Text, "saved result") || !strings.Contains(queued[0].Text, "[auto-subagents update]") { | ||
| 48 | t.Fatalf("queued completions = %#v", queued) | 83 | t.Fatalf("queued completions = %#v", queued) | ||
| 49 | } | 84 | } | ||
| 85 | + | if !strings.Contains(queued[0].Text, "Still running: still-running") { | |||
| 86 | + | t.Fatalf("completion update missing active sibling: %q", queued[0].Text) | |||
| 87 | + | } | |||
| 50 | if tc.err != nil && !strings.Contains(queued[0].Text, tc.err.Error()) { | 88 | if tc.err != nil && !strings.Contains(queued[0].Text, tc.err.Error()) { | ||
| 51 | t.Fatalf("missing failure: %q", queued[0].Text) | 89 | t.Fatalf("missing failure: %q", queued[0].Text) | ||
| 52 | } | 90 | } | ||
M packages/agent/modes/interactive_resident_subagents.go +16 -3
| @@ -124,7 +124,7 @@ func (i *Interactive) registerCoordinatorWorker(workerID string) { | |||
| 124 | 124 | i.coordinatorWorkerIDs = make(map[string][]string) | |
| 125 | 125 | } | |
| 126 | 126 | i.coordinatorWorkerIDs[workerID] = append(i.coordinatorWorkerIDs[workerID], registrationID) | |
| 127 | - | coordinator.Apply(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: registrationID}) | |
| 127 | + | coordinator.Apply(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: registrationID, AgentID: workerID}) | |
| 128 | 128 | var actions []orchestration.Action | |
| 129 | 129 | if implicitWave { | |
| 130 | 130 | actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions | |
| @@ -175,10 +175,10 @@ func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action) | |||
| 175 | 175 | prompt := action.Text | |
| 176 | 176 | if len(action.Completions) != 0 { | |
| 177 | 177 | instruction := "Briefly summarise the collective outcome for the user. Reference the agents by id. If any failed, suggest a follow-up; otherwise confirm completion. Do not spawn new sub-agents unless the user asks." | |
| 178 | - | if action.Kind == orchestration.ActionQueueManager { | |
| 178 | + | if action.Kind == orchestration.ActionQueueManager || len(action.ActiveAgentIDs) != 0 { | |
| 179 | 179 | instruction = "Treat these completion reports as worker evidence, not a new user request. Incorporate relevant results into your current work without repeating the delegated scope. Other workers may still be running; do not poll solely to wait for them." | |
| 180 | 180 | } | |
| 181 | - | update := subagents.FormatCompletionUpdate(action.Completions, instruction) | |
| 181 | + | update := subagents.FormatCompletionUpdateWithActive(action.Completions, action.ActiveAgentIDs, instruction) | |
| 182 | 182 | if prompt != "" { | |
| 183 | 183 | prompt = update + "\n\nQueued user request:\n" + prompt | |
| 184 | 184 | } else { | |
| @@ -277,6 +277,10 @@ func (i *Interactive) deliverCompletionUpdates() { | |||
| 277 | 277 | ||
| 278 | 278 | batch, err := tracker.WaitReady(i.completionWaitContext()) | |
| 279 | 279 | if err == nil && len(batch) != 0 { | |
| 280 | + | terminalAgentIDs := make(map[string]struct{}, len(batch)) | |
| 281 | + | for _, completion := range batch { | |
| 282 | + | terminalAgentIDs[completion.AgentID] = struct{}{} | |
| 283 | + | } | |
| 280 | 284 | var actions []orchestration.Action | |
| 281 | 285 | for _, completion := range batch { | |
| 282 | 286 | workerID := i.takeCoordinatorWorkerID(completion.AgentID) | |
| @@ -289,6 +293,15 @@ func (i *Interactive) deliverCompletionUpdates() { | |||
| 289 | 293 | Completion: completion, | |
| 290 | 294 | })...) | |
| 291 | 295 | } | |
| 296 | + | for index := range actions { | |
| 297 | + | active := actions[index].ActiveAgentIDs[:0] | |
| 298 | + | for _, agentID := range actions[index].ActiveAgentIDs { | |
| 299 | + | if _, terminal := terminalAgentIDs[agentID]; !terminal { | |
| 300 | + | active = append(active, agentID) | |
| 301 | + | } | |
| 302 | + | } | |
| 303 | + | actions[index].ActiveAgentIDs = active | |
| 304 | + | } | |
| 292 | 305 | i.executeCoordinatorActions(actions) | |
| 293 | 306 | } | |
| 294 | 307 | ||
| @@ -124,7 +124,7 @@ func (i *Interactive) registerCoordinatorWorker(workerID string) { | |||||
| 124 | i.coordinatorWorkerIDs = make(map[string][]string) | 124 | i.coordinatorWorkerIDs = make(map[string][]string) | ||
| 125 | } | 125 | } | ||
| 126 | i.coordinatorWorkerIDs[workerID] = append(i.coordinatorWorkerIDs[workerID], registrationID) | 126 | i.coordinatorWorkerIDs[workerID] = append(i.coordinatorWorkerIDs[workerID], registrationID) | ||
| 127 | - | coordinator.Apply(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: registrationID}) | 127 | + | coordinator.Apply(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: registrationID, AgentID: workerID}) |
| 128 | var actions []orchestration.Action | 128 | var actions []orchestration.Action | ||
| 129 | if implicitWave { | 129 | if implicitWave { | ||
| 130 | actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions | 130 | actions = coordinator.Apply(orchestration.Event{Kind: orchestration.EventManagerFinished}).Actions | ||
| @@ -175,10 +175,10 @@ func (i *Interactive) executeCoordinatorActions(actions []orchestration.Action) | |||||
| 175 | prompt := action.Text | 175 | prompt := action.Text | ||
| 176 | if len(action.Completions) != 0 { | 176 | if len(action.Completions) != 0 { | ||
| 177 | instruction := "Briefly summarise the collective outcome for the user. Reference the agents by id. If any failed, suggest a follow-up; otherwise confirm completion. Do not spawn new sub-agents unless the user asks." | 177 | instruction := "Briefly summarise the collective outcome for the user. Reference the agents by id. If any failed, suggest a follow-up; otherwise confirm completion. Do not spawn new sub-agents unless the user asks." | ||
| 178 | - | if action.Kind == orchestration.ActionQueueManager { | 178 | + | if action.Kind == orchestration.ActionQueueManager || len(action.ActiveAgentIDs) != 0 { |
| 179 | instruction = "Treat these completion reports as worker evidence, not a new user request. Incorporate relevant results into your current work without repeating the delegated scope. Other workers may still be running; do not poll solely to wait for them." | 179 | instruction = "Treat these completion reports as worker evidence, not a new user request. Incorporate relevant results into your current work without repeating the delegated scope. Other workers may still be running; do not poll solely to wait for them." | ||
| 180 | } | 180 | } | ||
| 181 | - | update := subagents.FormatCompletionUpdate(action.Completions, instruction) | 181 | + | update := subagents.FormatCompletionUpdateWithActive(action.Completions, action.ActiveAgentIDs, instruction) |
| 182 | if prompt != "" { | 182 | if prompt != "" { | ||
| 183 | prompt = update + "\n\nQueued user request:\n" + prompt | 183 | prompt = update + "\n\nQueued user request:\n" + prompt | ||
| 184 | } else { | 184 | } else { | ||
| @@ -277,6 +277,10 @@ func (i *Interactive) deliverCompletionUpdates() { | |||||
| 277 | 277 | ||||
| 278 | batch, err := tracker.WaitReady(i.completionWaitContext()) | 278 | batch, err := tracker.WaitReady(i.completionWaitContext()) | ||
| 279 | if err == nil && len(batch) != 0 { | 279 | if err == nil && len(batch) != 0 { | ||
| 280 | + | terminalAgentIDs := make(map[string]struct{}, len(batch)) | |||
| 281 | + | for _, completion := range batch { | |||
| 282 | + | terminalAgentIDs[completion.AgentID] = struct{}{} | |||
| 283 | + | } | |||
| 280 | var actions []orchestration.Action | 284 | var actions []orchestration.Action | ||
| 281 | for _, completion := range batch { | 285 | for _, completion := range batch { | ||
| 282 | workerID := i.takeCoordinatorWorkerID(completion.AgentID) | 286 | workerID := i.takeCoordinatorWorkerID(completion.AgentID) | ||
| @@ -289,6 +293,15 @@ func (i *Interactive) deliverCompletionUpdates() { | |||||
| 289 | Completion: completion, | 293 | Completion: completion, | ||
| 290 | })...) | 294 | })...) | ||
| 291 | } | 295 | } | ||
| 296 | + | for index := range actions { | |||
| 297 | + | active := actions[index].ActiveAgentIDs[:0] | |||
| 298 | + | for _, agentID := range actions[index].ActiveAgentIDs { | |||
| 299 | + | if _, terminal := terminalAgentIDs[agentID]; !terminal { | |||
| 300 | + | active = append(active, agentID) | |||
| 301 | + | } | |||
| 302 | + | } | |||
| 303 | + | actions[index].ActiveAgentIDs = active | |||
| 304 | + | } | |||
| 292 | i.executeCoordinatorActions(actions) | 305 | i.executeCoordinatorActions(actions) | ||
| 293 | } | 306 | } | ||
| 294 | 307 | ||||
M packages/agent/subagents/completion.go +13 -0
| @@ -148,6 +148,12 @@ func (t *CompletionTracker) wait(ctx context.Context, idle bool) ([]Completion, | |||
| 148 | 148 | func (t *CompletionTracker) signalLocked() { close(t.changed); t.changed = make(chan struct{}) } | |
| 149 | 149 | ||
| 150 | 150 | func FormatCompletionUpdate(batch []Completion, instruction string) string { | |
| 151 | + | return FormatCompletionUpdateWithActive(batch, nil, instruction) | |
| 152 | + | } | |
| 153 | + | ||
| 154 | + | // FormatCompletionUpdateWithActive formats terminal outcomes and identifies | |
| 155 | + | // sibling agents that are still running when this update is emitted. | |
| 156 | + | func FormatCompletionUpdateWithActive(batch []Completion, activeAgentIDs []string, instruction string) string { | |
| 151 | 157 | if len(batch) == 0 { | |
| 152 | 158 | return "" | |
| 153 | 159 | } | |
| @@ -170,6 +176,13 @@ func FormatCompletionUpdate(batch []Completion, instruction string) string { | |||
| 170 | 176 | } | |
| 171 | 177 | b.WriteByte('\n') | |
| 172 | 178 | } | |
| 179 | + | if len(activeAgentIDs) != 0 { | |
| 180 | + | b.WriteString("Still running:") | |
| 181 | + | for _, agentID := range activeAgentIDs { | |
| 182 | + | fmt.Fprintf(&b, " %s", agentID) | |
| 183 | + | } | |
| 184 | + | b.WriteByte('\n') | |
| 185 | + | } | |
| 173 | 186 | if instruction != "" { | |
| 174 | 187 | b.WriteString(instruction) | |
| 175 | 188 | } | |
| @@ -148,6 +148,12 @@ func (t *CompletionTracker) wait(ctx context.Context, idle bool) ([]Completion, | |||||
| 148 | func (t *CompletionTracker) signalLocked() { close(t.changed); t.changed = make(chan struct{}) } | 148 | func (t *CompletionTracker) signalLocked() { close(t.changed); t.changed = make(chan struct{}) } | ||
| 149 | 149 | ||||
| 150 | func FormatCompletionUpdate(batch []Completion, instruction string) string { | 150 | func FormatCompletionUpdate(batch []Completion, instruction string) string { | ||
| 151 | + | return FormatCompletionUpdateWithActive(batch, nil, instruction) | |||
| 152 | + | } | |||
| 153 | + | ||||
| 154 | + | // FormatCompletionUpdateWithActive formats terminal outcomes and identifies | |||
| 155 | + | // sibling agents that are still running when this update is emitted. | |||
| 156 | + | func FormatCompletionUpdateWithActive(batch []Completion, activeAgentIDs []string, instruction string) string { | |||
| 151 | if len(batch) == 0 { | 157 | if len(batch) == 0 { | ||
| 152 | return "" | 158 | return "" | ||
| 153 | } | 159 | } | ||
| @@ -170,6 +176,13 @@ func FormatCompletionUpdate(batch []Completion, instruction string) string { | |||||
| 170 | } | 176 | } | ||
| 171 | b.WriteByte('\n') | 177 | b.WriteByte('\n') | ||
| 172 | } | 178 | } | ||
| 179 | + | if len(activeAgentIDs) != 0 { | |||
| 180 | + | b.WriteString("Still running:") | |||
| 181 | + | for _, agentID := range activeAgentIDs { | |||
| 182 | + | fmt.Fprintf(&b, " %s", agentID) | |||
| 183 | + | } | |||
| 184 | + | b.WriteByte('\n') | |||
| 185 | + | } | |||
| 173 | if instruction != "" { | 186 | if instruction != "" { | ||
| 174 | b.WriteString(instruction) | 187 | b.WriteString(instruction) | ||
| 175 | } | 188 | } | ||
M packages/agent/subagents/completion_test.go +13 -0
| @@ -45,6 +45,19 @@ func TestFormatCompletionUpdateIncludesFinalSummary(t *testing.T) { | |||
| 45 | 45 | } | |
| 46 | 46 | } | |
| 47 | 47 | ||
| 48 | + | func TestFormatCompletionUpdateIdentifiesActiveSiblings(t *testing.T) { | |
| 49 | + | got := FormatCompletionUpdateWithActive( | |
| 50 | + | []Completion{{AgentID: "finished", Status: "completed"}}, | |
| 51 | + | []string{"worker-a", "worker-b"}, | |
| 52 | + | "continue", | |
| 53 | + | ) | |
| 54 | + | for _, want := range []string{"finished: completed", "Still running: worker-a worker-b", "continue"} { | |
| 55 | + | if !strings.Contains(got, want) { | |
| 56 | + | t.Fatalf("completion update %q missing %q", got, want) | |
| 57 | + | } | |
| 58 | + | } | |
| 59 | + | } | |
| 60 | + | ||
| 48 | 61 | func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) { | |
| 49 | 62 | tracker := NewCompletionTracker() | |
| 50 | 63 | if !tracker.TrackResident("child", "turn-1") { | |
| @@ -45,6 +45,19 @@ func TestFormatCompletionUpdateIncludesFinalSummary(t *testing.T) { | |||||
| 45 | } | 45 | } | ||
| 46 | } | 46 | } | ||
| 47 | 47 | ||||
| 48 | + | func TestFormatCompletionUpdateIdentifiesActiveSiblings(t *testing.T) { | |||
| 49 | + | got := FormatCompletionUpdateWithActive( | |||
| 50 | + | []Completion{{AgentID: "finished", Status: "completed"}}, | |||
| 51 | + | []string{"worker-a", "worker-b"}, | |||
| 52 | + | "continue", | |||
| 53 | + | ) | |||
| 54 | + | for _, want := range []string{"finished: completed", "Still running: worker-a worker-b", "continue"} { | |||
| 55 | + | if !strings.Contains(got, want) { | |||
| 56 | + | t.Fatalf("completion update %q missing %q", got, want) | |||
| 57 | + | } | |||
| 58 | + | } | |||
| 59 | + | } | |||
| 60 | + | ||||
| 48 | func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) { | 61 | func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) { | ||
| 49 | tracker := NewCompletionTracker() | 62 | tracker := NewCompletionTracker() | ||
| 50 | if !tracker.TrackResident("child", "turn-1") { | 63 | if !tracker.TrackResident("child", "turn-1") { | ||