detached 31b43ea

fix(subagents): preserve completion delivery waves

commit
31b43ea95c5918e3160e4e87fe53ccb77ab0b94c
author
brice · · 2026-08-14 06:40 UTC
parent
949637b
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
12 files · +236 -63
M packages/agent/cli.go +2 -2
@@ -1481,9 +1481,9 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
14811481 // Keep the shared subagent runtime in the outer scope: session reloads,
14821482 // dashboard updates, agent rebuilds, and shutdown all use the same owner.
14831483 var runtime *subagentRuntime
1484- onResidentAccepted := func(spec subagents.ResidentChildSpec, _ string) {
1484+ onResidentAccepted := func(spec subagents.ResidentChildSpec, turnID, _ string) {
14851485 if iv != nil {
1486- iv.TrackResidentSubagent(spec.ID)
1486+ iv.TrackResidentSubagent(spec.ID, turnID)
14871487 }
14881488 }
14891489 onResidentCompletion := func(completion subagents.ResidentCompletion) {
@@ -1481,9 +1481,9 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
1481 // Keep the shared subagent runtime in the outer scope: session reloads,1481 // Keep the shared subagent runtime in the outer scope: session reloads,
1482 // dashboard updates, agent rebuilds, and shutdown all use the same owner.1482 // dashboard updates, agent rebuilds, and shutdown all use the same owner.
1483 var runtime *subagentRuntime1483 var runtime *subagentRuntime
1484- onResidentAccepted := func(spec subagents.ResidentChildSpec, _ string) {1484+ onResidentAccepted := func(spec subagents.ResidentChildSpec, turnID, _ string) {
1485 if iv != nil {1485 if iv != nil {
1486- iv.TrackResidentSubagent(spec.ID)1486+ iv.TrackResidentSubagent(spec.ID, turnID)
1487 }1487 }
1488 }1488 }
1489 onResidentCompletion := func(completion subagents.ResidentCompletion) {1489 onResidentCompletion := func(completion subagents.ResidentCompletion) {
M packages/agent/modes/interactive_host.go +6 -4
@@ -224,10 +224,12 @@ func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, u
224224 // the manager is dormant behind a sealed worker wave, the coordinator makes
225225 // this the next manager turn and carries any completed worker summary with it.
226226 if userInput && i.coordinatorAcceptsUserInput() {
227- if actions := i.applyCoordinator(orchestration.Event{Kind: orchestration.EventUserInput, Text: text}); len(actions) != 0 {
228- i.executeCoordinatorActions(actions)
229- return
230- }
227+ actions := i.applyCoordinator(orchestration.Event{Kind: orchestration.EventUserInput, Text: text})
228+ i.executeCoordinatorActions(actions)
229+ // The coordinator has accepted the input for its sealed worker wave.
230+ // It may intentionally defer an action until pending workers finish;
231+ // falling through would start a normal turn and discard that wave.
232+ return
231233 }
232234 i.maybeStartSessionTitle(i.runCtx, text)
233235 i.mu.Lock()
@@ -224,10 +224,12 @@ func (i *Interactive) submitOrQueue(text string, images []provider.ImageBlock, u
224 // the manager is dormant behind a sealed worker wave, the coordinator makes224 // the manager is dormant behind a sealed worker wave, the coordinator makes
225 // this the next manager turn and carries any completed worker summary with it.225 // this the next manager turn and carries any completed worker summary with it.
226 if userInput && i.coordinatorAcceptsUserInput() {226 if userInput && i.coordinatorAcceptsUserInput() {
227- if actions := i.applyCoordinator(orchestration.Event{Kind: orchestration.EventUserInput, Text: text}); len(actions) != 0 {227+ actions := i.applyCoordinator(orchestration.Event{Kind: orchestration.EventUserInput, Text: text})
228- i.executeCoordinatorActions(actions)228+ i.executeCoordinatorActions(actions)
229- return229+ // The coordinator has accepted the input for its sealed worker wave.
230- }230+ // It may intentionally defer an action until pending workers finish;
231+ // falling through would start a normal turn and discard that wave.
232+ return
231 }233 }
232 i.maybeStartSessionTitle(i.runCtx, text)234 i.maybeStartSessionTitle(i.runCtx, text)
233 i.mu.Lock()235 i.mu.Lock()
A packages/agent/modes/interactive_host_test.go +33 -0
@@ -0,0 +1,33 @@
1+package modes
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/zut/packages/agent/internal/orchestration"
7+ "github.com/bnema/zut/packages/agent/subagents"
8+ "github.com/bnema/zut/packages/core"
9+)
10+
11+func TestSubmitOrQueuePreservesSealedWorkerWaveWithPendingWorker(t *testing.T) {
12+ i := &Interactive{agent: &core.Agent{}}
13+ i.applyCoordinator(orchestration.Event{Kind: orchestration.EventManagerStarted})
14+ i.applyCoordinator(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: "worker#1"})
15+ i.applyCoordinator(orchestration.Event{Kind: orchestration.EventManagerFinished})
16+
17+ i.submitOrQueue("review the result", nil, true)
18+
19+ actions := i.applyCoordinator(orchestration.Event{
20+ Kind: orchestration.EventWorkerFinished,
21+ WorkerID: "worker#1",
22+ Completion: subagents.Completion{
23+ AgentID: "worker",
24+ Status: "completed",
25+ },
26+ })
27+ if len(actions) != 1 {
28+ t.Fatalf("completion actions = %d, want 1", len(actions))
29+ }
30+ if got := actions[0].Text; got != "review the result" {
31+ t.Fatalf("completion action text = %q, want queued user input", got)
32+ }
33+}
@@ -0,0 +1,33 @@
1+package modes
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/zut/packages/agent/internal/orchestration"
7+ "github.com/bnema/zut/packages/agent/subagents"
8+ "github.com/bnema/zut/packages/core"
9+)
10+
11+func TestSubmitOrQueuePreservesSealedWorkerWaveWithPendingWorker(t *testing.T) {
12+ i := &Interactive{agent: &core.Agent{}}
13+ i.applyCoordinator(orchestration.Event{Kind: orchestration.EventManagerStarted})
14+ i.applyCoordinator(orchestration.Event{Kind: orchestration.EventWorkerRegistered, WorkerID: "worker#1"})
15+ i.applyCoordinator(orchestration.Event{Kind: orchestration.EventManagerFinished})
16+
17+ i.submitOrQueue("review the result", nil, true)
18+
19+ actions := i.applyCoordinator(orchestration.Event{
20+ Kind: orchestration.EventWorkerFinished,
21+ WorkerID: "worker#1",
22+ Completion: subagents.Completion{
23+ AgentID: "worker",
24+ Status: "completed",
25+ },
26+ })
27+ if len(actions) != 1 {
28+ t.Fatalf("completion actions = %d, want 1", len(actions))
29+ }
30+ if got := actions[0].Text; got != "review the result" {
31+ t.Fatalf("completion action text = %q, want queued user input", got)
32+ }
33+}
M packages/agent/modes/interactive_resident_subagents.go +11 -3
@@ -2,6 +2,7 @@ package modes
22
33 import (
44 "context"
5+ "errors"
56 "fmt"
67 "os"
78 "path/filepath"
@@ -17,11 +18,13 @@ import (
1718 "github.com/bnema/zut/packages/tui"
1819 )
1920
20-func (i *Interactive) TrackResidentSubagent(childID string) {
21+func (i *Interactive) TrackResidentSubagent(childID, turnID string) {
2122 if i == nil {
2223 return
2324 }
24- i.ensureCompletionTracker().TrackResident()
25+ if !i.ensureCompletionTracker().TrackResident(childID, turnID) {
26+ return
27+ }
2528 i.registerCoordinatorWorker(childID)
2629 i.invalidate()
2730 i.requestCompletionDelivery()
@@ -33,8 +36,13 @@ func (i *Interactive) ReportResidentSubagent(completion subagents.ResidentComple
3336 status, errText := "completed", ""
3437 if completion.Err != nil {
3538 status, errText = "failed", completion.Err.Error()
39+ if errors.Is(completion.Err, context.Canceled) {
40+ status = "interrupted"
41+ }
42+ }
43+ if !i.ensureCompletionTracker().Report(subagents.Completion{AgentID: completion.ChildID, TurnID: completion.TurnID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary}) {
44+ return
3645 }
37- i.ensureCompletionTracker().Report(subagents.Completion{AgentID: completion.ChildID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary})
3846 i.reloadOpenResidentChildSession(completion.ChildID)
3947 i.invalidate()
4048 i.requestCompletionDelivery()
@@ -2,6 +2,7 @@ package modes
2 2
3 import (3 import (
4 "context"4 "context"
5+ "errors"
5 "fmt"6 "fmt"
6 "os"7 "os"
7 "path/filepath"8 "path/filepath"
@@ -17,11 +18,13 @@ import (
17 "github.com/bnema/zut/packages/tui"18 "github.com/bnema/zut/packages/tui"
18 )19 )
19 20
20-func (i *Interactive) TrackResidentSubagent(childID string) {21+func (i *Interactive) TrackResidentSubagent(childID, turnID string) {
21 if i == nil {22 if i == nil {
22 return23 return
23 }24 }
24- i.ensureCompletionTracker().TrackResident()25+ if !i.ensureCompletionTracker().TrackResident(childID, turnID) {
26+ return
27+ }
25 i.registerCoordinatorWorker(childID)28 i.registerCoordinatorWorker(childID)
26 i.invalidate()29 i.invalidate()
27 i.requestCompletionDelivery()30 i.requestCompletionDelivery()
@@ -33,8 +36,13 @@ func (i *Interactive) ReportResidentSubagent(completion subagents.ResidentComple
33 status, errText := "completed", ""36 status, errText := "completed", ""
34 if completion.Err != nil {37 if completion.Err != nil {
35 status, errText = "failed", completion.Err.Error()38 status, errText = "failed", completion.Err.Error()
39+ if errors.Is(completion.Err, context.Canceled) {
40+ status = "interrupted"
41+ }
42+ }
43+ if !i.ensureCompletionTracker().Report(subagents.Completion{AgentID: completion.ChildID, TurnID: completion.TurnID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary}) {
44+ return
36 }45 }
37- i.ensureCompletionTracker().Report(subagents.Completion{AgentID: completion.ChildID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary})
38 i.reloadOpenResidentChildSession(completion.ChildID)46 i.reloadOpenResidentChildSession(completion.ChildID)
39 i.invalidate()47 i.invalidate()
40 i.requestCompletionDelivery()48 i.requestCompletionDelivery()
M packages/agent/orchestration_modes.go +5 -2
@@ -210,10 +210,13 @@ func newOrchestratedRuntime(_ context.Context, args Args, r Resolved, cfg Config
210210 errText := ""
211211 if completion.Err != nil {
212212 status, errText = "failed", completion.Err.Error()
213+ if errors.Is(completion.Err, context.Canceled) {
214+ status = "interrupted"
215+ }
213216 }
214- tracker.Report(subagents.Completion{AgentID: completion.ChildID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary})
217+ tracker.Report(subagents.Completion{AgentID: completion.ChildID, TurnID: completion.TurnID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary})
215218 },
216- OnResidentSpawned: func(subagents.ResidentChildSpec, string) { tracker.TrackResident() },
219+ OnResidentSpawned: func(spec subagents.ResidentChildSpec, turnID, _ string) { tracker.TrackResident(spec.ID, turnID) },
217220 })
218221 }
219222
@@ -210,10 +210,13 @@ func newOrchestratedRuntime(_ context.Context, args Args, r Resolved, cfg Config
210 errText := ""210 errText := ""
211 if completion.Err != nil {211 if completion.Err != nil {
212 status, errText = "failed", completion.Err.Error()212 status, errText = "failed", completion.Err.Error()
213+ if errors.Is(completion.Err, context.Canceled) {
214+ status = "interrupted"
215+ }
213 }216 }
214- tracker.Report(subagents.Completion{AgentID: completion.ChildID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary})217+ tracker.Report(subagents.Completion{AgentID: completion.ChildID, TurnID: completion.TurnID, Status: status, Task: completion.Task, Error: errText, Summary: completion.Summary})
215 },218 },
216- OnResidentSpawned: func(subagents.ResidentChildSpec, string) { tracker.TrackResident() },219+ OnResidentSpawned: func(spec subagents.ResidentChildSpec, turnID, _ string) { tracker.TrackResident(spec.ID, turnID) },
217 })220 })
218 }221 }
219 222
M packages/agent/subagent_runtime.go +4 -7
@@ -44,10 +44,9 @@ type subagentRuntime struct {
4444 activeSession string
4545 webSearchGuard *webSearchSessionGuard
4646
47- onResidentSpawned func(subagents.ResidentChildSpec, string)
48- settingsMu sync.RWMutex
49- closeMu sync.Mutex
50- closed bool
47+ settingsMu sync.RWMutex
48+ closeMu sync.Mutex
49+ closed bool
5150 }
5251
5352 type subagentRuntimeConfiguration struct {
@@ -82,7 +81,7 @@ type subagentRuntimeConfig struct {
8281 ActiveModel func() string
8382 ActiveReasoning func() string
8483 ResidentCompletion func(subagents.ResidentCompletion)
85- OnResidentSpawned func(subagents.ResidentChildSpec, string)
84+ OnResidentSpawned func(subagents.ResidentChildSpec, string, string)
8685 }
8786
8887 func newSubagentRuntime(cfg subagentRuntimeConfig) *subagentRuntime {
@@ -103,7 +102,6 @@ func newSubagentRuntime(cfg subagentRuntimeConfig) *subagentRuntime {
103102 webSearchPolicy: cfg.WebSearchPolicy,
104103 policy: cfg.Policy,
105104 webSearchGuard: cfg.WebSearchGuard,
106- onResidentSpawned: cfg.OnResidentSpawned,
107105 }
108106 if rt.activeProvider == nil {
109107 rt.activeProvider = func() string {
@@ -398,7 +396,6 @@ func (rt *subagentRuntime) InjectTools(reg core.Registry) core.Registry {
398396 BuildResidentSpec: func(ctx context.Context, request tools.ResidentSpawnRequest) (subagents.ResidentChildSpec, error) {
399397 return rt.buildResidentChildSpec(ctx, request, reg)
400398 },
401- OnResidentSpawned: rt.onResidentSpawned,
402399 }
403400 reg[spawn.Name()] = spawn
404401 }
@@ -44,10 +44,9 @@ type subagentRuntime struct {
44 activeSession string44 activeSession string
45 webSearchGuard *webSearchSessionGuard45 webSearchGuard *webSearchSessionGuard
46 46
47- onResidentSpawned func(subagents.ResidentChildSpec, string)47+ settingsMu sync.RWMutex
48- settingsMu sync.RWMutex48+ closeMu sync.Mutex
49- closeMu sync.Mutex49+ closed bool
50- closed bool
51 }50 }
52 51
53 type subagentRuntimeConfiguration struct {52 type subagentRuntimeConfiguration struct {
@@ -82,7 +81,7 @@ type subagentRuntimeConfig struct {
82 ActiveModel func() string81 ActiveModel func() string
83 ActiveReasoning func() string82 ActiveReasoning func() string
84 ResidentCompletion func(subagents.ResidentCompletion)83 ResidentCompletion func(subagents.ResidentCompletion)
85- OnResidentSpawned func(subagents.ResidentChildSpec, string)84+ OnResidentSpawned func(subagents.ResidentChildSpec, string, string)
86 }85 }
87 86
88 func newSubagentRuntime(cfg subagentRuntimeConfig) *subagentRuntime {87 func newSubagentRuntime(cfg subagentRuntimeConfig) *subagentRuntime {
@@ -103,7 +102,6 @@ func newSubagentRuntime(cfg subagentRuntimeConfig) *subagentRuntime {
103 webSearchPolicy: cfg.WebSearchPolicy,102 webSearchPolicy: cfg.WebSearchPolicy,
104 policy: cfg.Policy,103 policy: cfg.Policy,
105 webSearchGuard: cfg.WebSearchGuard,104 webSearchGuard: cfg.WebSearchGuard,
106- onResidentSpawned: cfg.OnResidentSpawned,
107 }105 }
108 if rt.activeProvider == nil {106 if rt.activeProvider == nil {
109 rt.activeProvider = func() string {107 rt.activeProvider = func() string {
@@ -398,7 +396,6 @@ func (rt *subagentRuntime) InjectTools(reg core.Registry) core.Registry {
398 BuildResidentSpec: func(ctx context.Context, request tools.ResidentSpawnRequest) (subagents.ResidentChildSpec, error) {396 BuildResidentSpec: func(ctx context.Context, request tools.ResidentSpawnRequest) (subagents.ResidentChildSpec, error) {
399 return rt.buildResidentChildSpec(ctx, request, reg)397 return rt.buildResidentChildSpec(ctx, request, reg)
400 },398 },
401- OnResidentSpawned: rt.onResidentSpawned,
402 }399 }
403 reg[spawn.Name()] = spawn400 reg[spawn.Name()] = spawn
404 }401 }
M packages/agent/subagents/completion.go +40 -16
@@ -10,6 +10,7 @@ import (
1010 // Completion is one resident child's terminal turn outcome.
1111 type Completion struct {
1212 AgentID string
13+ TurnID string
1314 Status string
1415 Task string
1516 Error string
@@ -20,38 +21,61 @@ type Completion struct {
2021 // turn. It has no process or worker-event dependency.
2122 type CompletionTracker struct {
2223 mu sync.Mutex
23- pending int
24+ pending map[string]struct{}
2425 ready []Completion
2526 changed chan struct{}
2627 }
2728
2829 func NewCompletionTracker() *CompletionTracker {
29- return &CompletionTracker{changed: make(chan struct{})}
30+ return &CompletionTracker{changed: make(chan struct{}), pending: make(map[string]struct{})}
3031 }
3132
32-func (t *CompletionTracker) TrackResident() {
33+// TrackResident records an accepted resident turn. It returns false when the
34+// turn is incomplete or already tracked.
35+func (t *CompletionTracker) TrackResident(agentID, turnID string) bool {
3336 if t == nil {
34- return
37+ return false
38+ }
39+ key := completionKey(agentID, turnID)
40+ if key == "" {
41+ return false
3542 }
3643 t.mu.Lock()
37- t.pending++
44+ defer t.mu.Unlock()
45+ if _, exists := t.pending[key]; exists {
46+ return false
47+ }
48+ t.pending[key] = struct{}{}
3849 t.signalLocked()
39- t.mu.Unlock()
50+ return true
4051 }
4152
42-func (t *CompletionTracker) Report(completion Completion) {
53+// Report accepts one terminal completion for an accepted resident turn. It
54+// returns false for duplicate or unknown terminal reports.
55+func (t *CompletionTracker) Report(completion Completion) bool {
4356 if t == nil {
44- return
57+ return false
58+ }
59+ key := completionKey(completion.AgentID, completion.TurnID)
60+ if key == "" {
61+ return false
4562 }
4663 t.mu.Lock()
47- if t.pending == 0 {
48- t.mu.Unlock()
49- return
64+ defer t.mu.Unlock()
65+ if _, exists := t.pending[key]; !exists {
66+ return false
5067 }
51- t.pending--
68+ delete(t.pending, key)
5269 t.ready = append(t.ready, completion)
5370 t.signalLocked()
54- t.mu.Unlock()
71+ return true
72+}
73+
74+func completionKey(agentID, turnID string) string {
75+ if agentID == "" || turnID == "" {
76+ return ""
77+ }
78+ return agentID + "\x00" + turnID
5579 }
5680
5781 func (t *CompletionTracker) Pending() int {
@@ -60,7 +84,7 @@ func (t *CompletionTracker) Pending() int {
6084 }
6185 t.mu.Lock()
6286 defer t.mu.Unlock()
63- return t.pending
87+ return len(t.pending)
6488 }
6589
6690 // Reset drops buffered outcomes at a parent-turn cancellation boundary.
@@ -69,7 +93,7 @@ func (t *CompletionTracker) Reset() {
6993 return
7094 }
7195 t.mu.Lock()
72- t.pending = 0
96+ t.pending = make(map[string]struct{})
7397 t.ready = nil
7498 t.signalLocked()
7599 t.mu.Unlock()
@@ -81,7 +105,7 @@ func (t *CompletionTracker) WaitIdle(ctx context.Context) ([]Completion, error)
81105 }
82106 for {
83107 t.mu.Lock()
84- if t.pending == 0 {
108+ if len(t.pending) == 0 {
85109 ready := append([]Completion(nil), t.ready...)
86110 t.ready = nil
87111 t.mu.Unlock()
@@ -10,6 +10,7 @@ import (
10 // Completion is one resident child's terminal turn outcome.10 // Completion is one resident child's terminal turn outcome.
11 type Completion struct {11 type Completion struct {
12 AgentID string12 AgentID string
13+ TurnID string
13 Status string14 Status string
14 Task string15 Task string
15 Error string16 Error string
@@ -20,38 +21,61 @@ type Completion struct {
20 // turn. It has no process or worker-event dependency.21 // turn. It has no process or worker-event dependency.
21 type CompletionTracker struct {22 type CompletionTracker struct {
22 mu sync.Mutex23 mu sync.Mutex
23- pending int24+ pending map[string]struct{}
24 ready []Completion25 ready []Completion
25 changed chan struct{}26 changed chan struct{}
26 }27 }
27 28
28 func NewCompletionTracker() *CompletionTracker {29 func NewCompletionTracker() *CompletionTracker {
29- return &CompletionTracker{changed: make(chan struct{})}30+ return &CompletionTracker{changed: make(chan struct{}), pending: make(map[string]struct{})}
30 }31 }
31 32
32-func (t *CompletionTracker) TrackResident() {33+// TrackResident records an accepted resident turn. It returns false when the
34+// turn is incomplete or already tracked.
35+func (t *CompletionTracker) TrackResident(agentID, turnID string) bool {
33 if t == nil {36 if t == nil {
34- return37+ return false
38+ }
39+ key := completionKey(agentID, turnID)
40+ if key == "" {
41+ return false
35 }42 }
36 t.mu.Lock()43 t.mu.Lock()
37- t.pending++44+ defer t.mu.Unlock()
45+ if _, exists := t.pending[key]; exists {
46+ return false
47+ }
48+ t.pending[key] = struct{}{}
38 t.signalLocked()49 t.signalLocked()
39- t.mu.Unlock()50+ return true
40 }51 }
41 52
42-func (t *CompletionTracker) Report(completion Completion) {53+// Report accepts one terminal completion for an accepted resident turn. It
54+// returns false for duplicate or unknown terminal reports.
55+func (t *CompletionTracker) Report(completion Completion) bool {
43 if t == nil {56 if t == nil {
44- return57+ return false
58+ }
59+ key := completionKey(completion.AgentID, completion.TurnID)
60+ if key == "" {
61+ return false
45 }62 }
46 t.mu.Lock()63 t.mu.Lock()
47- if t.pending == 0 {64+ defer t.mu.Unlock()
48- t.mu.Unlock()65+ if _, exists := t.pending[key]; !exists {
49- return66+ return false
50 }67 }
51- t.pending--68+ delete(t.pending, key)
52 t.ready = append(t.ready, completion)69 t.ready = append(t.ready, completion)
53 t.signalLocked()70 t.signalLocked()
54- t.mu.Unlock()71+ return true
72+}
73+
74+func completionKey(agentID, turnID string) string {
75+ if agentID == "" || turnID == "" {
76+ return ""
77+ }
78+ return agentID + "\x00" + turnID
55 }79 }
56 80
57 func (t *CompletionTracker) Pending() int {81 func (t *CompletionTracker) Pending() int {
@@ -60,7 +84,7 @@ func (t *CompletionTracker) Pending() int {
60 }84 }
61 t.mu.Lock()85 t.mu.Lock()
62 defer t.mu.Unlock()86 defer t.mu.Unlock()
63- return t.pending87+ return len(t.pending)
64 }88 }
65 89
66 // Reset drops buffered outcomes at a parent-turn cancellation boundary.90 // Reset drops buffered outcomes at a parent-turn cancellation boundary.
@@ -69,7 +93,7 @@ func (t *CompletionTracker) Reset() {
69 return93 return
70 }94 }
71 t.mu.Lock()95 t.mu.Lock()
72- t.pending = 096+ t.pending = make(map[string]struct{})
73 t.ready = nil97 t.ready = nil
74 t.signalLocked()98 t.signalLocked()
75 t.mu.Unlock()99 t.mu.Unlock()
@@ -81,7 +105,7 @@ func (t *CompletionTracker) WaitIdle(ctx context.Context) ([]Completion, error)
81 }105 }
82 for {106 for {
83 t.mu.Lock()107 t.mu.Lock()
84- if t.pending == 0 {108+ if len(t.pending) == 0 {
85 ready := append([]Completion(nil), t.ready...)109 ready := append([]Completion(nil), t.ready...)
86 t.ready = nil110 t.ready = nil
87 t.mu.Unlock()111 t.mu.Unlock()
M packages/agent/subagents/completion_test.go +35 -2
@@ -15,9 +15,13 @@ func TestFormatCompletionUpdateIncludesFinalSummary(t *testing.T) {
1515
1616 func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) {
1717 tracker := NewCompletionTracker()
18- tracker.TrackResident()
18+ if !tracker.TrackResident("child", "turn-1") {
19+ t.Fatal("TrackResident returned false")
20+ }
1921 tracker.Reset()
20- tracker.Report(Completion{AgentID: "stale"})
22+ if tracker.Report(Completion{AgentID: "child", TurnID: "turn-1"}) {
23+ t.Fatal("Report returned true after reset")
24+ }
2125 got, err := tracker.WaitIdle(context.Background())
2226 if err != nil {
2327 t.Fatal(err)
@@ -26,3 +30,32 @@ func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) {
2630 t.Fatalf("stale completions = %#v", got)
2731 }
2832 }
33+
34+func TestCompletionTrackerMatchesTerminalReportsToAcceptedTurns(t *testing.T) {
35+ tracker := NewCompletionTracker()
36+ if !tracker.TrackResident("child", "turn-1") || !tracker.TrackResident("child", "turn-2") {
37+ t.Fatal("TrackResident returned false")
38+ }
39+ if tracker.Report(Completion{AgentID: "child", TurnID: "other"}) {
40+ t.Fatal("Report accepted an unknown turn")
41+ }
42+ if !tracker.Report(Completion{AgentID: "child", TurnID: "turn-1"}) {
43+ t.Fatal("Report rejected an accepted turn")
44+ }
45+ if tracker.Report(Completion{AgentID: "child", TurnID: "turn-1"}) {
46+ t.Fatal("Report accepted a duplicate terminal result")
47+ }
48+ if got := tracker.Pending(); got != 1 {
49+ t.Fatalf("Pending = %d, want 1", got)
50+ }
51+ if !tracker.Report(Completion{AgentID: "child", TurnID: "turn-2"}) {
52+ t.Fatal("Report rejected second accepted turn")
53+ }
54+ got, err := tracker.WaitIdle(context.Background())
55+ if err != nil {
56+ t.Fatal(err)
57+ }
58+ if len(got) != 2 {
59+ t.Fatalf("completions = %#v, want 2", got)
60+ }
61+}
@@ -15,9 +15,13 @@ func TestFormatCompletionUpdateIncludesFinalSummary(t *testing.T) {
15 15
16 func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) {16 func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) {
17 tracker := NewCompletionTracker()17 tracker := NewCompletionTracker()
18- tracker.TrackResident()18+ if !tracker.TrackResident("child", "turn-1") {
19+ t.Fatal("TrackResident returned false")
20+ }
19 tracker.Reset()21 tracker.Reset()
20- tracker.Report(Completion{AgentID: "stale"})22+ if tracker.Report(Completion{AgentID: "child", TurnID: "turn-1"}) {
23+ t.Fatal("Report returned true after reset")
24+ }
21 got, err := tracker.WaitIdle(context.Background())25 got, err := tracker.WaitIdle(context.Background())
22 if err != nil {26 if err != nil {
23 t.Fatal(err)27 t.Fatal(err)
@@ -26,3 +30,32 @@ func TestCompletionTrackerDropsReportsAfterCancellationReset(t *testing.T) {
26 t.Fatalf("stale completions = %#v", got)30 t.Fatalf("stale completions = %#v", got)
27 }31 }
28 }32 }
33+
34+func TestCompletionTrackerMatchesTerminalReportsToAcceptedTurns(t *testing.T) {
35+ tracker := NewCompletionTracker()
36+ if !tracker.TrackResident("child", "turn-1") || !tracker.TrackResident("child", "turn-2") {
37+ t.Fatal("TrackResident returned false")
38+ }
39+ if tracker.Report(Completion{AgentID: "child", TurnID: "other"}) {
40+ t.Fatal("Report accepted an unknown turn")
41+ }
42+ if !tracker.Report(Completion{AgentID: "child", TurnID: "turn-1"}) {
43+ t.Fatal("Report rejected an accepted turn")
44+ }
45+ if tracker.Report(Completion{AgentID: "child", TurnID: "turn-1"}) {
46+ t.Fatal("Report accepted a duplicate terminal result")
47+ }
48+ if got := tracker.Pending(); got != 1 {
49+ t.Fatalf("Pending = %d, want 1", got)
50+ }
51+ if !tracker.Report(Completion{AgentID: "child", TurnID: "turn-2"}) {
52+ t.Fatal("Report rejected second accepted turn")
53+ }
54+ got, err := tracker.WaitIdle(context.Background())
55+ if err != nil {
56+ t.Fatal(err)
57+ }
58+ if len(got) != 2 {
59+ t.Fatalf("completions = %#v, want 2", got)
60+ }
61+}
M packages/agent/subagents/resident_child.go +32 -13
@@ -30,6 +30,7 @@ type residentTurnResult struct {
3030 // ResidentCompletion is a typed in-process terminal turn notification.
3131 type ResidentCompletion struct {
3232 ChildID string
33+ TurnID string
3334 Task string
3435 Err error
3536 Summary string
@@ -256,36 +257,44 @@ func (c *ResidentChild) Resume(ctx context.Context, prompt string) error {
256257 return c.enqueue(ctx, residentPrompt{turnID: uuid.NewString(), prompt: prompt, ack: make(chan error, 1)})
257258 }
258259
259-func (c *ResidentChild) resumeAccepted(ctx context.Context, turnID, prompt string) error {
260- return c.enqueue(ctx, residentPrompt{turnID: turnID, prompt: prompt, ack: make(chan error, 1)})
260+// resumeAccepted queues durably accepted work and reports whether the child
261+// took ownership of its terminal completion. Once it has taken ownership, it
262+// emits the completion itself even when turn startup fails.
263+func (c *ResidentChild) resumeAccepted(ctx context.Context, turnID, prompt string) (bool, error) {
264+ return c.enqueueAccepted(ctx, residentPrompt{turnID: turnID, prompt: prompt, ack: make(chan error, 1)})
261265 }
262266
263267 func (c *ResidentChild) enqueue(ctx context.Context, request residentPrompt) error {
268+ _, err := c.enqueueAccepted(ctx, request)
269+ return err
270+}
271+
272+func (c *ResidentChild) enqueueAccepted(ctx context.Context, request residentPrompt) (bool, error) {
264273 if c == nil || c.runner == nil {
265- return errors.New("resident child: no runner")
274+ return false, errors.New("resident child: no runner")
266275 }
267276 if ctx == nil {
268277 ctx = context.Background()
269278 }
270279 request.prompt = strings.TrimSpace(request.prompt)
271280 if request.prompt == "" {
272- return errors.New("resident child: prompt is empty")
281+ return false, errors.New("resident child: prompt is empty")
273282 }
274283 if strings.TrimSpace(request.turnID) == "" {
275- return errors.New("resident child: missing turn ID")
284+ return false, errors.New("resident child: missing turn ID")
276285 }
277286 select {
278287 case <-ctx.Done():
279- return ctx.Err()
288+ return false, ctx.Err()
280289 case <-c.ctx.Done():
281- return errors.New("resident child: closed")
290+ return false, errors.New("resident child: closed")
282291 case c.inbox <- request:
283292 }
284293 select {
285294 case err := <-request.ack:
286- return err
295+ return true, err
287296 case <-c.ctx.Done():
288- return errors.New("resident child: closed before prompt acceptance")
297+ return true, errors.New("resident child: closed before prompt acceptance")
289298 }
290299 }
291300
@@ -325,6 +334,7 @@ func (c *ResidentChild) run() {
325334 }()
326335
327336 var queue []residentPrompt
337+ var interruptedPrompts []residentPrompt
328338 results := make(chan residentTurnResult, 1)
329339 var active residentPrompt
330340 running, interrupted := false, false
@@ -338,7 +348,7 @@ func (c *ResidentChild) run() {
338348 c.setState(ResidentFailed)
339349 c.cancel()
340350 if c.onCompletion != nil {
341- c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, Task: active.prompt, Err: terminalErr})
351+ c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: active.turnID, Task: active.prompt, Err: terminalErr})
342352 }
343353 return
344354 }
@@ -353,6 +363,11 @@ func (c *ResidentChild) run() {
353363 if interrupted && !running {
354364 c.live.Finish(ResidentInterrupted)
355365 c.setState(ResidentInterrupted)
366+ if c.onCompletion != nil {
367+ for _, prompt := range interruptedPrompts {
368+ c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: prompt.turnID, Task: prompt.prompt, Err: context.Canceled})
369+ }
370+ }
356371 return
357372 }
358373
@@ -366,10 +381,14 @@ func (c *ResidentChild) run() {
366381 continue
367382 }
368383 interrupted = true
369- if running && c.journal != nil {
370- _ = c.journal.RecordTurnInterrupted(c.spec, active.turnID)
384+ if running {
385+ interruptedPrompts = append(interruptedPrompts, active)
386+ if c.journal != nil {
387+ _ = c.journal.RecordTurnInterrupted(c.spec, active.turnID)
388+ }
371389 }
372390 for _, pending := range queue {
391+ interruptedPrompts = append(interruptedPrompts, pending)
373392 if c.journal != nil {
374393 _ = c.journal.RecordTurnInterrupted(c.spec, pending.turnID)
375394 }
@@ -418,7 +437,7 @@ func (c *ResidentChild) run() {
418437 summary = result.Summary
419438 }
420439 }
421- c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, Task: active.prompt, Err: terminalErr, Summary: summary})
440+ c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: result.turnID, Task: active.prompt, Err: terminalErr, Summary: summary})
422441 }
423442 if persistenceFailed {
424443 return
@@ -30,6 +30,7 @@ type residentTurnResult struct {
30 // ResidentCompletion is a typed in-process terminal turn notification.30 // ResidentCompletion is a typed in-process terminal turn notification.
31 type ResidentCompletion struct {31 type ResidentCompletion struct {
32 ChildID string32 ChildID string
33+ TurnID string
33 Task string34 Task string
34 Err error35 Err error
35 Summary string36 Summary string
@@ -256,36 +257,44 @@ func (c *ResidentChild) Resume(ctx context.Context, prompt string) error {
256 return c.enqueue(ctx, residentPrompt{turnID: uuid.NewString(), prompt: prompt, ack: make(chan error, 1)})257 return c.enqueue(ctx, residentPrompt{turnID: uuid.NewString(), prompt: prompt, ack: make(chan error, 1)})
257 }258 }
258 259
259-func (c *ResidentChild) resumeAccepted(ctx context.Context, turnID, prompt string) error {260+// resumeAccepted queues durably accepted work and reports whether the child
260- return c.enqueue(ctx, residentPrompt{turnID: turnID, prompt: prompt, ack: make(chan error, 1)})261+// took ownership of its terminal completion. Once it has taken ownership, it
262+// emits the completion itself even when turn startup fails.
263+func (c *ResidentChild) resumeAccepted(ctx context.Context, turnID, prompt string) (bool, error) {
264+ return c.enqueueAccepted(ctx, residentPrompt{turnID: turnID, prompt: prompt, ack: make(chan error, 1)})
261 }265 }
262 266
263 func (c *ResidentChild) enqueue(ctx context.Context, request residentPrompt) error {267 func (c *ResidentChild) enqueue(ctx context.Context, request residentPrompt) error {
268+ _, err := c.enqueueAccepted(ctx, request)
269+ return err
270+}
271+
272+func (c *ResidentChild) enqueueAccepted(ctx context.Context, request residentPrompt) (bool, error) {
264 if c == nil || c.runner == nil {273 if c == nil || c.runner == nil {
265- return errors.New("resident child: no runner")274+ return false, errors.New("resident child: no runner")
266 }275 }
267 if ctx == nil {276 if ctx == nil {
268 ctx = context.Background()277 ctx = context.Background()
269 }278 }
270 request.prompt = strings.TrimSpace(request.prompt)279 request.prompt = strings.TrimSpace(request.prompt)
271 if request.prompt == "" {280 if request.prompt == "" {
272- return errors.New("resident child: prompt is empty")281+ return false, errors.New("resident child: prompt is empty")
273 }282 }
274 if strings.TrimSpace(request.turnID) == "" {283 if strings.TrimSpace(request.turnID) == "" {
275- return errors.New("resident child: missing turn ID")284+ return false, errors.New("resident child: missing turn ID")
276 }285 }
277 select {286 select {
278 case <-ctx.Done():287 case <-ctx.Done():
279- return ctx.Err()288+ return false, ctx.Err()
280 case <-c.ctx.Done():289 case <-c.ctx.Done():
281- return errors.New("resident child: closed")290+ return false, errors.New("resident child: closed")
282 case c.inbox <- request:291 case c.inbox <- request:
283 }292 }
284 select {293 select {
285 case err := <-request.ack:294 case err := <-request.ack:
286- return err295+ return true, err
287 case <-c.ctx.Done():296 case <-c.ctx.Done():
288- return errors.New("resident child: closed before prompt acceptance")297+ return true, errors.New("resident child: closed before prompt acceptance")
289 }298 }
290 }299 }
291 300
@@ -325,6 +334,7 @@ func (c *ResidentChild) run() {
325 }()334 }()
326 335
327 var queue []residentPrompt336 var queue []residentPrompt
337+ var interruptedPrompts []residentPrompt
328 results := make(chan residentTurnResult, 1)338 results := make(chan residentTurnResult, 1)
329 var active residentPrompt339 var active residentPrompt
330 running, interrupted := false, false340 running, interrupted := false, false
@@ -338,7 +348,7 @@ func (c *ResidentChild) run() {
338 c.setState(ResidentFailed)348 c.setState(ResidentFailed)
339 c.cancel()349 c.cancel()
340 if c.onCompletion != nil {350 if c.onCompletion != nil {
341- c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, Task: active.prompt, Err: terminalErr})351+ c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: active.turnID, Task: active.prompt, Err: terminalErr})
342 }352 }
343 return353 return
344 }354 }
@@ -353,6 +363,11 @@ func (c *ResidentChild) run() {
353 if interrupted && !running {363 if interrupted && !running {
354 c.live.Finish(ResidentInterrupted)364 c.live.Finish(ResidentInterrupted)
355 c.setState(ResidentInterrupted)365 c.setState(ResidentInterrupted)
366+ if c.onCompletion != nil {
367+ for _, prompt := range interruptedPrompts {
368+ c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: prompt.turnID, Task: prompt.prompt, Err: context.Canceled})
369+ }
370+ }
356 return371 return
357 }372 }
358 373
@@ -366,10 +381,14 @@ func (c *ResidentChild) run() {
366 continue381 continue
367 }382 }
368 interrupted = true383 interrupted = true
369- if running && c.journal != nil {384+ if running {
370- _ = c.journal.RecordTurnInterrupted(c.spec, active.turnID)385+ interruptedPrompts = append(interruptedPrompts, active)
386+ if c.journal != nil {
387+ _ = c.journal.RecordTurnInterrupted(c.spec, active.turnID)
388+ }
371 }389 }
372 for _, pending := range queue {390 for _, pending := range queue {
391+ interruptedPrompts = append(interruptedPrompts, pending)
373 if c.journal != nil {392 if c.journal != nil {
374 _ = c.journal.RecordTurnInterrupted(c.spec, pending.turnID)393 _ = c.journal.RecordTurnInterrupted(c.spec, pending.turnID)
375 }394 }
@@ -418,7 +437,7 @@ func (c *ResidentChild) run() {
418 summary = result.Summary437 summary = result.Summary
419 }438 }
420 }439 }
421- c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, Task: active.prompt, Err: terminalErr, Summary: summary})440+ c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: result.turnID, Task: active.prompt, Err: terminalErr, Summary: summary})
422 }441 }
423 if persistenceFailed {442 if persistenceFailed {
424 return443 return
M packages/agent/subagents/resident_child_test.go +1 -1
@@ -110,7 +110,7 @@ func TestResidentChildReportsAndPreservesFailureWhenTurnStartCannotPersist(t *te
110110 }
111111 select {
112112 case completion := <-completed:
113- if completion.ChildID != "start-persistence-failure" || completion.Task != "task" {
113+ if completion.ChildID != "start-persistence-failure" || completion.TurnID != "turn" || completion.Task != "task" {
114114 t.Fatalf("completion = %#v", completion)
115115 }
116116 if completion.Err == nil || !strings.Contains(completion.Err.Error(), "persist resident child start state") {
@@ -110,7 +110,7 @@ func TestResidentChildReportsAndPreservesFailureWhenTurnStartCannotPersist(t *te
110 }110 }
111 select {111 select {
112 case completion := <-completed:112 case completion := <-completed:
113- if completion.ChildID != "start-persistence-failure" || completion.Task != "task" {113+ if completion.ChildID != "start-persistence-failure" || completion.TurnID != "turn" || completion.Task != "task" {
114 t.Fatalf("completion = %#v", completion)114 t.Fatalf("completion = %#v", completion)
115 }115 }
116 if completion.Err == nil || !strings.Contains(completion.Err.Error(), "persist resident child start state") {116 if completion.Err == nil || !strings.Contains(completion.Err.Error(), "persist resident child start state") {
M packages/agent/subagents/resident_manager.go +28 -12
@@ -36,7 +36,7 @@ type ResidentManager struct {
3636 allowedRoots []string
3737 dispatchMu sync.Mutex
3838 onCompletion func(ResidentCompletion)
39- onAccepted func(ResidentChildSpec, string)
39+ onAccepted func(ResidentChildSpec, string, string)
4040 onUpdate func(string)
4141 onHistoryUpdate func(string)
4242 onActivity func(bool)
@@ -51,9 +51,10 @@ func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion
5151 m.mu.Unlock()
5252 }
5353
54-// SetAcceptedObserver receives a prompt only after the authoritative
55-// child.accepted record has synchronized, and before it can be scheduled.
56-func (m *ResidentManager) SetAcceptedObserver(observer func(ResidentChildSpec, string)) {
54+// SetAcceptedObserver receives an accepted turn's ID and prompt only after
55+// the authoritative child.accepted record has synchronized and before it can
56+// be scheduled.
57+func (m *ResidentManager) SetAcceptedObserver(observer func(ResidentChildSpec, string, string)) {
5758 if m == nil {
5859 return
5960 }
@@ -274,7 +275,7 @@ func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, tas
274275 accepted := m.onAccepted
275276 m.mu.Unlock()
276277 if accepted != nil {
277- accepted(spec, task)
278+ accepted(spec, spec.InitialTurnID, task)
278279 }
279280 runner, err := m.factory(spec, journal)
280281 if err != nil {
@@ -284,7 +285,7 @@ func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, tas
284285 completion := m.onCompletion
285286 m.mu.Unlock()
286287 if completion != nil {
287- completion(ResidentCompletion{ChildID: spec.ID, Task: task, Err: err})
288+ completion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
288289 }
289290 return nil, err
290291 }
@@ -302,14 +303,20 @@ func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, tas
302303 }
303304 m.children[spec.ID] = child
304305 m.mu.Unlock()
305- if err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {
306- _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
306+ if childAccepted, err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {
307307 m.mu.Lock()
308308 if m.children[spec.ID] == child {
309309 delete(m.children, spec.ID)
310310 }
311+ completion := m.onCompletion
311312 m.mu.Unlock()
312313 _ = child.Close(context.Background())
314+ if !childAccepted {
315+ _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
316+ if completion != nil {
317+ completion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
318+ }
319+ }
313320 return nil, err
314321 }
315322 return child, nil
@@ -458,7 +465,8 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
458465 }
459466 turnID := uuid.NewString()
460467 if child.journal == nil {
461- return child.resumeAccepted(ctx, turnID, prompt)
468+ _, err := child.resumeAccepted(ctx, turnID, prompt)
469+ return err
462470 }
463471 if err := child.journal.AcceptFollowUp(child.spec, turnID, prompt); err != nil {
464472 return err
@@ -467,10 +475,18 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
467475 accepted := m.onAccepted
468476 m.mu.Unlock()
469477 if accepted != nil {
470- accepted(child.spec, prompt)
478+ accepted(child.spec, turnID, prompt)
471479 }
472- if err := child.resumeAccepted(ctx, turnID, prompt); err != nil {
473- _ = child.journal.RecordTurnInterrupted(child.spec, turnID)
480+ if childAccepted, err := child.resumeAccepted(ctx, turnID, prompt); err != nil {
481+ if !childAccepted {
482+ _ = child.journal.RecordTurnInterrupted(child.spec, turnID)
483+ m.mu.Lock()
484+ completion := m.onCompletion
485+ m.mu.Unlock()
486+ if completion != nil {
487+ completion(ResidentCompletion{ChildID: child.spec.ID, TurnID: turnID, Task: prompt, Err: err})
488+ }
489+ }
474490 return err
475491 }
476492 return nil
@@ -36,7 +36,7 @@ type ResidentManager struct {
36 allowedRoots []string36 allowedRoots []string
37 dispatchMu sync.Mutex37 dispatchMu sync.Mutex
38 onCompletion func(ResidentCompletion)38 onCompletion func(ResidentCompletion)
39- onAccepted func(ResidentChildSpec, string)39+ onAccepted func(ResidentChildSpec, string, string)
40 onUpdate func(string)40 onUpdate func(string)
41 onHistoryUpdate func(string)41 onHistoryUpdate func(string)
42 onActivity func(bool)42 onActivity func(bool)
@@ -51,9 +51,10 @@ func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion
51 m.mu.Unlock()51 m.mu.Unlock()
52 }52 }
53 53
54-// SetAcceptedObserver receives a prompt only after the authoritative54+// SetAcceptedObserver receives an accepted turn's ID and prompt only after
55-// child.accepted record has synchronized, and before it can be scheduled.55+// the authoritative child.accepted record has synchronized and before it can
56-func (m *ResidentManager) SetAcceptedObserver(observer func(ResidentChildSpec, string)) {56+// be scheduled.
57+func (m *ResidentManager) SetAcceptedObserver(observer func(ResidentChildSpec, string, string)) {
57 if m == nil {58 if m == nil {
58 return59 return
59 }60 }
@@ -274,7 +275,7 @@ func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, tas
274 accepted := m.onAccepted275 accepted := m.onAccepted
275 m.mu.Unlock()276 m.mu.Unlock()
276 if accepted != nil {277 if accepted != nil {
277- accepted(spec, task)278+ accepted(spec, spec.InitialTurnID, task)
278 }279 }
279 runner, err := m.factory(spec, journal)280 runner, err := m.factory(spec, journal)
280 if err != nil {281 if err != nil {
@@ -284,7 +285,7 @@ func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, tas
284 completion := m.onCompletion285 completion := m.onCompletion
285 m.mu.Unlock()286 m.mu.Unlock()
286 if completion != nil {287 if completion != nil {
287- completion(ResidentCompletion{ChildID: spec.ID, Task: task, Err: err})288+ completion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
288 }289 }
289 return nil, err290 return nil, err
290 }291 }
@@ -302,14 +303,20 @@ func (m *ResidentManager) Spawn(ctx context.Context, spec ResidentChildSpec, tas
302 }303 }
303 m.children[spec.ID] = child304 m.children[spec.ID] = child
304 m.mu.Unlock()305 m.mu.Unlock()
305- if err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {306+ if childAccepted, err := child.resumeAccepted(ctx, spec.InitialTurnID, task); err != nil {
306- _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
307 m.mu.Lock()307 m.mu.Lock()
308 if m.children[spec.ID] == child {308 if m.children[spec.ID] == child {
309 delete(m.children, spec.ID)309 delete(m.children, spec.ID)
310 }310 }
311+ completion := m.onCompletion
311 m.mu.Unlock()312 m.mu.Unlock()
312 _ = child.Close(context.Background())313 _ = child.Close(context.Background())
314+ if !childAccepted {
315+ _ = journal.RecordTurnInterrupted(spec, spec.InitialTurnID)
316+ if completion != nil {
317+ completion(ResidentCompletion{ChildID: spec.ID, TurnID: spec.InitialTurnID, Task: task, Err: err})
318+ }
319+ }
313 return nil, err320 return nil, err
314 }321 }
315 return child, nil322 return child, nil
@@ -458,7 +465,8 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
458 }465 }
459 turnID := uuid.NewString()466 turnID := uuid.NewString()
460 if child.journal == nil {467 if child.journal == nil {
461- return child.resumeAccepted(ctx, turnID, prompt)468+ _, err := child.resumeAccepted(ctx, turnID, prompt)
469+ return err
462 }470 }
463 if err := child.journal.AcceptFollowUp(child.spec, turnID, prompt); err != nil {471 if err := child.journal.AcceptFollowUp(child.spec, turnID, prompt); err != nil {
464 return err472 return err
@@ -467,10 +475,18 @@ func (m *ResidentManager) Resume(ctx context.Context, childID, prompt string) er
467 accepted := m.onAccepted475 accepted := m.onAccepted
468 m.mu.Unlock()476 m.mu.Unlock()
469 if accepted != nil {477 if accepted != nil {
470- accepted(child.spec, prompt)478+ accepted(child.spec, turnID, prompt)
471 }479 }
472- if err := child.resumeAccepted(ctx, turnID, prompt); err != nil {480+ if childAccepted, err := child.resumeAccepted(ctx, turnID, prompt); err != nil {
473- _ = child.journal.RecordTurnInterrupted(child.spec, turnID)481+ if !childAccepted {
482+ _ = child.journal.RecordTurnInterrupted(child.spec, turnID)
483+ m.mu.Lock()
484+ completion := m.onCompletion
485+ m.mu.Unlock()
486+ if completion != nil {
487+ completion(ResidentCompletion{ChildID: child.spec.ID, TurnID: turnID, Task: prompt, Err: err})
488+ }
489+ }
474 return err490 return err
475 }491 }
476 return nil492 return nil
M packages/agent/subagents/resident_manager_test.go +39 -1
@@ -67,6 +67,9 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) {
6767 }
6868 select {
6969 case completion := <-completed:
70+ if completion.TurnID == "" {
71+ t.Fatal("completion turn ID is empty")
72+ }
7073 if completion.Summary != "the actual child answer" {
7174 t.Fatalf("completion summary = %q", completion.Summary)
7275 }
@@ -75,6 +78,41 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) {
7578 }
7679 }
7780
81+func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) {
82+ started := make(chan struct{})
83+ completed := make(chan ResidentCompletion, 1)
84+ manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
85+ return func(ctx context.Context, _ string) error {
86+ close(started)
87+ <-ctx.Done()
88+ return ctx.Err()
89+ }, nil
90+ })
91+ manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion })
92+ t.Cleanup(func() { _ = manager.Close(context.Background()) })
93+
94+ child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "interrupted-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task")
95+ if err != nil {
96+ t.Fatal(err)
97+ }
98+ select {
99+ case <-started:
100+ case <-time.After(time.Second):
101+ t.Fatal("resident child did not start")
102+ }
103+ if err := child.Close(context.Background()); err != nil {
104+ t.Fatalf("Close: %v", err)
105+ }
106+ select {
107+ case completion := <-completed:
108+ if completion.ChildID != "interrupted-child" || completion.TurnID == "" || !errors.Is(completion.Err, context.Canceled) {
109+ t.Fatalf("completion = %#v", completion)
110+ }
111+ case <-time.After(time.Second):
112+ t.Fatal("interrupted turn did not report completion")
113+ }
114+}
115+
78116 func TestResidentManagerLiveReturnsChildSnapshot(t *testing.T) {
79117 started := make(chan struct{})
80118 release := make(chan struct{})
@@ -627,7 +665,7 @@ func TestResidentManagerNotifiesAcceptedObserverForFollowUps(t *testing.T) {
627665 })
628666 defer manager.Close(context.Background())
629667 accepted := make(chan string, 2)
630- manager.SetAcceptedObserver(func(_ ResidentChildSpec, prompt string) { accepted <- prompt })
668+ manager.SetAcceptedObserver(func(_ ResidentChildSpec, _ string, prompt string) { accepted <- prompt })
631669 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}
632670 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
633671 t.Fatal(err)
@@ -67,6 +67,9 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) {
67 }67 }
68 select {68 select {
69 case completion := <-completed:69 case completion := <-completed:
70+ if completion.TurnID == "" {
71+ t.Fatal("completion turn ID is empty")
72+ }
70 if completion.Summary != "the actual child answer" {73 if completion.Summary != "the actual child answer" {
71 t.Fatalf("completion summary = %q", completion.Summary)74 t.Fatalf("completion summary = %q", completion.Summary)
72 }75 }
@@ -75,6 +78,41 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) {
75 }78 }
76 }79 }
77 80
81+func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) {
82+ started := make(chan struct{})
83+ completed := make(chan ResidentCompletion, 1)
84+ manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentTurnRunner, error) {
85+ return func(ctx context.Context, _ string) error {
86+ close(started)
87+ <-ctx.Done()
88+ return ctx.Err()
89+ }, nil
90+ })
91+ manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion })
92+ t.Cleanup(func() { _ = manager.Close(context.Background()) })
93+
94+ child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "interrupted-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task")
95+ if err != nil {
96+ t.Fatal(err)
97+ }
98+ select {
99+ case <-started:
100+ case <-time.After(time.Second):
101+ t.Fatal("resident child did not start")
102+ }
103+ if err := child.Close(context.Background()); err != nil {
104+ t.Fatalf("Close: %v", err)
105+ }
106+ select {
107+ case completion := <-completed:
108+ if completion.ChildID != "interrupted-child" || completion.TurnID == "" || !errors.Is(completion.Err, context.Canceled) {
109+ t.Fatalf("completion = %#v", completion)
110+ }
111+ case <-time.After(time.Second):
112+ t.Fatal("interrupted turn did not report completion")
113+ }
114+}
115+
78 func TestResidentManagerLiveReturnsChildSnapshot(t *testing.T) {116 func TestResidentManagerLiveReturnsChildSnapshot(t *testing.T) {
79 started := make(chan struct{})117 started := make(chan struct{})
80 release := make(chan struct{})118 release := make(chan struct{})
@@ -627,7 +665,7 @@ func TestResidentManagerNotifiesAcceptedObserverForFollowUps(t *testing.T) {
627 })665 })
628 defer manager.Close(context.Background())666 defer manager.Close(context.Background())
629 accepted := make(chan string, 2)667 accepted := make(chan string, 2)
630- manager.SetAcceptedObserver(func(_ ResidentChildSpec, prompt string) { accepted <- prompt })668+ manager.SetAcceptedObserver(func(_ ResidentChildSpec, _ string, prompt string) { accepted <- prompt })
631 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}669 spec := ResidentChildSpec{ID: "observer", SessionID: "session", Provider: "openai", Model: "gpt-5"}
632 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {670 if _, err := manager.Spawn(context.Background(), spec, "initial"); err != nil {
633 t.Fatal(err)671 t.Fatal(err)

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 476

↑↓ moveenter openesc close