fix(subagents): report queued persistence failures
- commit
- 642e3ef8b4412e429a9315e4375149d65c5dd54a
- author
- brice · · 2026-08-14 06:55 UTC
- parent
- 31b43ea
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +57 -0
M packages/agent/subagents/resident_child.go +8 -0
| @@ -349,6 +349,9 @@ func (c *ResidentChild) run() { | |||
| 349 | 349 | c.cancel() | |
| 350 | 350 | if c.onCompletion != nil { | |
| 351 | 351 | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: active.turnID, Task: active.prompt, Err: terminalErr}) | |
| 352 | + | for _, pending := range queue { | |
| 353 | + | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: terminalErr}) | |
| 354 | + | } | |
| 352 | 355 | } | |
| 353 | 356 | return | |
| 354 | 357 | } | |
| @@ -438,6 +441,11 @@ func (c *ResidentChild) run() { | |||
| 438 | 441 | } | |
| 439 | 442 | } | |
| 440 | 443 | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: result.turnID, Task: active.prompt, Err: terminalErr, Summary: summary}) | |
| 444 | + | if persistenceFailed { | |
| 445 | + | for _, pending := range queue { | |
| 446 | + | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: terminalErr}) | |
| 447 | + | } | |
| 448 | + | } | |
| 441 | 449 | } | |
| 442 | 450 | if persistenceFailed { | |
| 443 | 451 | return | |
| @@ -349,6 +349,9 @@ func (c *ResidentChild) run() { | |||||
| 349 | c.cancel() | 349 | c.cancel() | ||
| 350 | if c.onCompletion != nil { | 350 | if c.onCompletion != nil { | ||
| 351 | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: active.turnID, Task: active.prompt, Err: terminalErr}) | 351 | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: active.turnID, Task: active.prompt, Err: terminalErr}) | ||
| 352 | + | for _, pending := range queue { | |||
| 353 | + | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: terminalErr}) | |||
| 354 | + | } | |||
| 352 | } | 355 | } | ||
| 353 | return | 356 | return | ||
| 354 | } | 357 | } | ||
| @@ -438,6 +441,11 @@ func (c *ResidentChild) run() { | |||||
| 438 | } | 441 | } | ||
| 439 | } | 442 | } | ||
| 440 | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: result.turnID, Task: active.prompt, Err: terminalErr, Summary: summary}) | 443 | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: result.turnID, Task: active.prompt, Err: terminalErr, Summary: summary}) | ||
| 444 | + | if persistenceFailed { | |||
| 445 | + | for _, pending := range queue { | |||
| 446 | + | c.onCompletion(ResidentCompletion{ChildID: c.spec.ID, TurnID: pending.turnID, Task: pending.prompt, Err: terminalErr}) | |||
| 447 | + | } | |||
| 448 | + | } | |||
| 441 | } | 449 | } | ||
| 442 | if persistenceFailed { | 450 | if persistenceFailed { | ||
| 443 | return | 451 | return | ||
M packages/agent/subagents/resident_manager_test.go +49 -0
| @@ -78,6 +78,55 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) { | |||
| 78 | 78 | } | |
| 79 | 79 | } | |
| 80 | 80 | ||
| 81 | + | func TestResidentManagerReportsQueuedTurnsWhenFinalResultCannotPersist(t *testing.T) { | |
| 82 | + | started := make(chan struct{}) | |
| 83 | + | release := make(chan struct{}) | |
| 84 | + | completed := make(chan ResidentCompletion, 2) | |
| 85 | + | var journal *ResidentJournal | |
| 86 | + | manager := NewResidentManager(t.TempDir(), func(_ ResidentChildSpec, childJournal *ResidentJournal) (ResidentTurnRunner, error) { | |
| 87 | + | journal = childJournal | |
| 88 | + | return func(context.Context, string) error { | |
| 89 | + | close(started) | |
| 90 | + | <-release | |
| 91 | + | return nil | |
| 92 | + | }, nil | |
| 93 | + | }) | |
| 94 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | |
| 95 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 96 | + | ||
| 97 | + | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "persistence-failure-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "initial task"); err != nil { | |
| 98 | + | t.Fatal(err) | |
| 99 | + | } | |
| 100 | + | select { | |
| 101 | + | case <-started: | |
| 102 | + | case <-time.After(time.Second): | |
| 103 | + | t.Fatal("resident child did not start") | |
| 104 | + | } | |
| 105 | + | if err := manager.Resume(context.Background(), "persistence-failure-child", "queued follow-up"); err != nil { | |
| 106 | + | t.Fatalf("Resume: %v", err) | |
| 107 | + | } | |
| 108 | + | if err := journal.Close(); err != nil { | |
| 109 | + | t.Fatalf("Close journal: %v", err) | |
| 110 | + | } | |
| 111 | + | close(release) | |
| 112 | + | ||
| 113 | + | seen := make(map[string]ResidentCompletion, 2) | |
| 114 | + | for range 2 { | |
| 115 | + | select { | |
| 116 | + | case completion := <-completed: | |
| 117 | + | seen[completion.Task] = completion | |
| 118 | + | case <-time.After(time.Second): | |
| 119 | + | t.Fatalf("terminal completions = %#v, want initial and queued turns", seen) | |
| 120 | + | } | |
| 121 | + | } | |
| 122 | + | for _, task := range []string{"initial task", "queued follow-up"} { | |
| 123 | + | completion, ok := seen[task] | |
| 124 | + | if !ok || completion.TurnID == "" || completion.Err == nil || !strings.Contains(completion.Err.Error(), "persist resident child terminal state") { | |
| 125 | + | t.Fatalf("completion for %q = %#v", task, completion) | |
| 126 | + | } | |
| 127 | + | } | |
| 128 | + | } | |
| 129 | + | ||
| 81 | 130 | func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) { | |
| 82 | 131 | started := make(chan struct{}) | |
| 83 | 132 | completed := make(chan ResidentCompletion, 1) | |
| @@ -78,6 +78,55 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) { | |||||
| 78 | } | 78 | } | ||
| 79 | } | 79 | } | ||
| 80 | 80 | ||||
| 81 | + | func TestResidentManagerReportsQueuedTurnsWhenFinalResultCannotPersist(t *testing.T) { | |||
| 82 | + | started := make(chan struct{}) | |||
| 83 | + | release := make(chan struct{}) | |||
| 84 | + | completed := make(chan ResidentCompletion, 2) | |||
| 85 | + | var journal *ResidentJournal | |||
| 86 | + | manager := NewResidentManager(t.TempDir(), func(_ ResidentChildSpec, childJournal *ResidentJournal) (ResidentTurnRunner, error) { | |||
| 87 | + | journal = childJournal | |||
| 88 | + | return func(context.Context, string) error { | |||
| 89 | + | close(started) | |||
| 90 | + | <-release | |||
| 91 | + | return nil | |||
| 92 | + | }, nil | |||
| 93 | + | }) | |||
| 94 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | |||
| 95 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |||
| 96 | + | ||||
| 97 | + | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "persistence-failure-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "initial task"); err != nil { | |||
| 98 | + | t.Fatal(err) | |||
| 99 | + | } | |||
| 100 | + | select { | |||
| 101 | + | case <-started: | |||
| 102 | + | case <-time.After(time.Second): | |||
| 103 | + | t.Fatal("resident child did not start") | |||
| 104 | + | } | |||
| 105 | + | if err := manager.Resume(context.Background(), "persistence-failure-child", "queued follow-up"); err != nil { | |||
| 106 | + | t.Fatalf("Resume: %v", err) | |||
| 107 | + | } | |||
| 108 | + | if err := journal.Close(); err != nil { | |||
| 109 | + | t.Fatalf("Close journal: %v", err) | |||
| 110 | + | } | |||
| 111 | + | close(release) | |||
| 112 | + | ||||
| 113 | + | seen := make(map[string]ResidentCompletion, 2) | |||
| 114 | + | for range 2 { | |||
| 115 | + | select { | |||
| 116 | + | case completion := <-completed: | |||
| 117 | + | seen[completion.Task] = completion | |||
| 118 | + | case <-time.After(time.Second): | |||
| 119 | + | t.Fatalf("terminal completions = %#v, want initial and queued turns", seen) | |||
| 120 | + | } | |||
| 121 | + | } | |||
| 122 | + | for _, task := range []string{"initial task", "queued follow-up"} { | |||
| 123 | + | completion, ok := seen[task] | |||
| 124 | + | if !ok || completion.TurnID == "" || completion.Err == nil || !strings.Contains(completion.Err.Error(), "persist resident child terminal state") { | |||
| 125 | + | t.Fatalf("completion for %q = %#v", task, completion) | |||
| 126 | + | } | |||
| 127 | + | } | |||
| 128 | + | } | |||
| 129 | + | ||||
| 81 | func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) { | 130 | func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) { | ||
| 82 | started := make(chan struct{}) | 131 | started := make(chan struct{}) | ||
| 83 | completed := make(chan ResidentCompletion, 1) | 132 | completed := make(chan ResidentCompletion, 1) | ||