fix(subagents): deliver waited reports through one channel (#236)
* fix(subagents): deliver waited reports through one channel * fix(subagents): require host acceptance before omitting reports * docs(subagents): clarify report acceptance and cancellation
- commit
- 4ce882c48d3aaf0f4c36c96653d91ebd21f891b4
- author
- bnema · · 2026-09-30 11:44 UTC
- committer
- GitHub · · 2026-09-30 11:44 UTC
- parent
- 1ddba1e
- signature
- GPG Unverified
B5690EEEBB952194The key is unknown
23 files · +474 -35
M docs/subagents.md +13 -2
| @@ -250,6 +250,9 @@ all of them. | |||
| 250 | 250 | - `stop` accepts `agent_id` and stops one live child. | |
| 251 | 251 | - `resume` accepts `agent_id`, `prompt`, optional `mode`, and optional | |
| 252 | 252 | `wait`. `prompt` is an explicit follow-up prompt for an existing child. | |
| 253 | + | Do not use `resume` solely to wait for existing work or ask for its result | |
| 254 | + | again: yield for the automatic completion update. `queue` schedules another | |
| 255 | + | turn and produces another completion. | |
| 253 | 256 | After a terminal failure, inspect the saved result before resuming; resume | |
| 254 | 257 | continues the retained session. `mode` controls how a follow-up reaches a | |
| 255 | 258 | running child: | |
| @@ -263,8 +266,9 @@ all of them. | |||
| 263 | 266 | `steered` or `queued`. `wait` uses the same 1–300 second bound as `spawn` | |
| 264 | 267 | and waits for the turn that answers the follow-up, returning its outcome or | |
| 265 | 268 | reporting the timeout while the child stays active. A returned outcome can | |
| 266 | - | be `dropped` when a queued follow-up never started, and it lists | |
| 267 | - | `undelivered` steers the child never read. | |
| 269 | + | be `dropped` when a queued follow-up never started. When the host accepts | |
| 270 | + | the completion report, the summary and `undelivered` steers arrive in the | |
| 271 | + | completion update; otherwise the outcome includes them. | |
| 268 | 272 | - `interrupt` accepts `agent_id` and cancels only the child's running turn. | |
| 269 | 273 | Unlike `stop`, the child stays live with its transcript, so a later | |
| 270 | 274 | `resume` such as "report what you have" continues with full context. | |
| @@ -281,6 +285,13 @@ completion arrives through the host’s typed completion update; the `subagent` | |||
| 281 | 285 | tool's `status` action returns immediately and does not wait for completion. Do | |
| 282 | 286 | not use sleep loops, repeated status calls, journal files, or terminal UI | |
| 283 | 287 | inspection as a completion signal. | |
| 288 | + | When the host accepts a completion report, a completed `spawn` or `resume` | |
| 289 | + | wait returns terminal status and `report_delivery: host_update`; the full report | |
| 290 | + | arrives only through the completion update, not again in the tool result. | |
| 291 | + | If no host observer accepts the report, the tool result includes it inline. | |
| 292 | + | An expired wait does not suppress later completion delivery. Cancelling the | |
| 293 | + | parent turn discards pending completion updates; cancelling only a tool wait | |
| 294 | + | leaves host tracking intact. | |
| 284 | 295 | In interactive mode, a result received while the primary is busy enters the | |
| 285 | 296 | visible **sliding in** queue and reaches the model at its next safe boundary, | |
| 286 | 297 | without interrupting a model request or tool call and without waiting for other | |
| @@ -250,6 +250,9 @@ all of them. | |||||
| 250 | - `stop` accepts `agent_id` and stops one live child. | 250 | - `stop` accepts `agent_id` and stops one live child. | ||
| 251 | - `resume` accepts `agent_id`, `prompt`, optional `mode`, and optional | 251 | - `resume` accepts `agent_id`, `prompt`, optional `mode`, and optional | ||
| 252 | `wait`. `prompt` is an explicit follow-up prompt for an existing child. | 252 | `wait`. `prompt` is an explicit follow-up prompt for an existing child. | ||
| 253 | + | Do not use `resume` solely to wait for existing work or ask for its result | |||
| 254 | + | again: yield for the automatic completion update. `queue` schedules another | |||
| 255 | + | turn and produces another completion. | |||
| 253 | After a terminal failure, inspect the saved result before resuming; resume | 256 | After a terminal failure, inspect the saved result before resuming; resume | ||
| 254 | continues the retained session. `mode` controls how a follow-up reaches a | 257 | continues the retained session. `mode` controls how a follow-up reaches a | ||
| 255 | running child: | 258 | running child: | ||
| @@ -263,8 +266,9 @@ all of them. | |||||
| 263 | `steered` or `queued`. `wait` uses the same 1–300 second bound as `spawn` | 266 | `steered` or `queued`. `wait` uses the same 1–300 second bound as `spawn` | ||
| 264 | and waits for the turn that answers the follow-up, returning its outcome or | 267 | and waits for the turn that answers the follow-up, returning its outcome or | ||
| 265 | reporting the timeout while the child stays active. A returned outcome can | 268 | reporting the timeout while the child stays active. A returned outcome can | ||
| 266 | - | be `dropped` when a queued follow-up never started, and it lists | 269 | + | be `dropped` when a queued follow-up never started. When the host accepts |
| 267 | - | `undelivered` steers the child never read. | 270 | + | the completion report, the summary and `undelivered` steers arrive in the |
| 271 | + | completion update; otherwise the outcome includes them. | |||
| 268 | - `interrupt` accepts `agent_id` and cancels only the child's running turn. | 272 | - `interrupt` accepts `agent_id` and cancels only the child's running turn. | ||
| 269 | Unlike `stop`, the child stays live with its transcript, so a later | 273 | Unlike `stop`, the child stays live with its transcript, so a later | ||
| 270 | `resume` such as "report what you have" continues with full context. | 274 | `resume` such as "report what you have" continues with full context. | ||
| @@ -281,6 +285,13 @@ completion arrives through the host’s typed completion update; the `subagent` | |||||
| 281 | tool's `status` action returns immediately and does not wait for completion. Do | 285 | tool's `status` action returns immediately and does not wait for completion. Do | ||
| 282 | not use sleep loops, repeated status calls, journal files, or terminal UI | 286 | not use sleep loops, repeated status calls, journal files, or terminal UI | ||
| 283 | inspection as a completion signal. | 287 | inspection as a completion signal. | ||
| 288 | + | When the host accepts a completion report, a completed `spawn` or `resume` | |||
| 289 | + | wait returns terminal status and `report_delivery: host_update`; the full report | |||
| 290 | + | arrives only through the completion update, not again in the tool result. | |||
| 291 | + | If no host observer accepts the report, the tool result includes it inline. | |||
| 292 | + | An expired wait does not suppress later completion delivery. Cancelling the | |||
| 293 | + | parent turn discards pending completion updates; cancelling only a tool wait | |||
| 294 | + | leaves host tracking intact. | |||
| 284 | In interactive mode, a result received while the primary is busy enters the | 295 | In interactive mode, a result received while the primary is busy enters the | ||
| 285 | visible **sliding in** queue and reaches the model at its next safe boundary, | 296 | visible **sliding in** queue and reaches the model at its next safe boundary, | ||
| 286 | without interrupting a model request or tool call and without waiting for other | 297 | without interrupting a model request or tool call and without waiting for other | ||
M packages/agent/cli.go +2 -4
| @@ -1673,10 +1673,8 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||
| 1673 | 1673 | iv.TrackResidentSubagent(spec.ID, turnID) | |
| 1674 | 1674 | } | |
| 1675 | 1675 | } | |
| 1676 | - | onResidentCompletion := func(completion subagents.ResidentCompletion) { | |
| 1677 | - | if iv != nil { | |
| 1678 | - | iv.ReportResidentSubagent(completion) | |
| 1679 | - | } | |
| 1676 | + | onResidentCompletion := func(completion subagents.ResidentCompletion) bool { | |
| 1677 | + | return iv != nil && iv.ReportResidentSubagent(completion) | |
| 1680 | 1678 | } | |
| 1681 | 1679 | activeProviderForSubagents := func() string { | |
| 1682 | 1680 | persistMu.Lock() | |
| @@ -1673,10 +1673,8 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro | |||||
| 1673 | iv.TrackResidentSubagent(spec.ID, turnID) | 1673 | iv.TrackResidentSubagent(spec.ID, turnID) | ||
| 1674 | } | 1674 | } | ||
| 1675 | } | 1675 | } | ||
| 1676 | - | onResidentCompletion := func(completion subagents.ResidentCompletion) { | 1676 | + | onResidentCompletion := func(completion subagents.ResidentCompletion) bool { |
| 1677 | - | if iv != nil { | 1677 | + | return iv != nil && iv.ReportResidentSubagent(completion) |
| 1678 | - | iv.ReportResidentSubagent(completion) | |||
| 1679 | - | } | |||
| 1680 | } | 1678 | } | ||
| 1681 | activeProviderForSubagents := func() string { | 1679 | activeProviderForSubagents := func() string { | ||
| 1682 | persistMu.Lock() | 1680 | persistMu.Lock() | ||
M packages/agent/modes/interactive.go +0 -3
| @@ -2142,9 +2142,6 @@ type telegramSenderAdapter struct { | |||
| 2142 | 2142 | // TrackResidentSubagent reserves completion delivery after the manager has | |
| 2143 | 2143 | // durably accepted a resident child prompt and before it is scheduled. | |
| 2144 | 2144 | ||
| 2145 | - | // ReportResidentSubagent delivers a typed resident terminal outcome through | |
| 2146 | - | // the same coordinator path used by legacy worker completions. | |
| 2147 | - | ||
| 2148 | 2145 | // registerCoordinatorWorker associates a tracker registration with the open | |
| 2149 | 2146 | // manager wave. Direct embedders that register a worker outside startTurn get | |
| 2150 | 2147 | // a one-worker sealed wave, retaining the exported tracker API's old behavior. | |
| @@ -2142,9 +2142,6 @@ type telegramSenderAdapter struct { | |||||
| 2142 | // TrackResidentSubagent reserves completion delivery after the manager has | 2142 | // TrackResidentSubagent reserves completion delivery after the manager has | ||
| 2143 | // durably accepted a resident child prompt and before it is scheduled. | 2143 | // durably accepted a resident child prompt and before it is scheduled. | ||
| 2144 | 2144 | ||||
| 2145 | - | // ReportResidentSubagent delivers a typed resident terminal outcome through | |||
| 2146 | - | // the same coordinator path used by legacy worker completions. | |||
| 2147 | - | ||||
| 2148 | // registerCoordinatorWorker associates a tracker registration with the open | 2145 | // registerCoordinatorWorker associates a tracker registration with the open | ||
| 2149 | // manager wave. Direct embedders that register a worker outside startTurn get | 2146 | // manager wave. Direct embedders that register a worker outside startTurn get | ||
| 2150 | // a one-worker sealed wave, retaining the exported tracker API's old behavior. | 2147 | // a one-worker sealed wave, retaining the exported tracker API's old behavior. | ||
A packages/agent/modes/interactive_completion_delivery_test.go +83 -0
| @@ -0,0 +1,83 @@ | |||
| 1 | + | package modes | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "encoding/json" | |
| 6 | + | "strings" | |
| 7 | + | "testing" | |
| 8 | + | "time" | |
| 9 | + | ||
| 10 | + | "github.com/bnema/zut/packages/agent/subagents" | |
| 11 | + | "github.com/bnema/zut/packages/agent/tools" | |
| 12 | + | "github.com/bnema/zut/packages/core" | |
| 13 | + | "github.com/bnema/zut/packages/provider" | |
| 14 | + | ) | |
| 15 | + | ||
| 16 | + | func TestInteractiveWaitReportAcceptance(t *testing.T) { | |
| 17 | + | for _, reset := range []bool{false, true} { | |
| 18 | + | name := "accepted" | |
| 19 | + | if reset { | |
| 20 | + | name = "cancelled tracking" | |
| 21 | + | } | |
| 22 | + | t.Run(name, func(t *testing.T) { | |
| 23 | + | ctx, cancel := context.WithCancel(t.Context()) | |
| 24 | + | defer cancel() | |
| 25 | + | ag := core.NewAgent(nil, "test", "", nil) | |
| 26 | + | i := &Interactive{agent: ag, busy: true, runCtx: ctx} | |
| 27 | + | release := i.beginCompletionDeliveryHold() | |
| 28 | + | defer release() | |
| 29 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |
| 30 | + | return subagents.ResidentTurnRunner(func(context.Context, string) error { | |
| 31 | + | if reset { | |
| 32 | + | i.cancelCoordinator() | |
| 33 | + | } | |
| 34 | + | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "interactive report"}}}}) | |
| 35 | + | }), nil | |
| 36 | + | }) | |
| 37 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 38 | + | manager.SetAcceptedObserver(func(spec subagents.ResidentChildSpec, turn, _ string) { i.TrackResidentSubagent(spec.ID, turn) }) | |
| 39 | + | manager.SetCompletionObserver(i.ReportResidentSubagent) | |
| 40 | + | spawn := &tools.SubagentSpawnTool{ | |
| 41 | + | ResidentManager: manager, Enabled: func() bool { return true }, | |
| 42 | + | BuildResidentSpec: func(context.Context, tools.ResidentSpawnRequest) (subagents.ResidentChildSpec, error) { | |
| 43 | + | return subagents.ResidentChildSpec{ID: "worker", SessionID: "session", Provider: "openai", Model: "test"}, nil | |
| 44 | + | }, | |
| 45 | + | } | |
| 46 | + | result, err := spawn.Execute(ctx, json.RawMessage(`{"task":"report","wait":5}`), nil) | |
| 47 | + | if err != nil || result.IsError { | |
| 48 | + | t.Fatalf("spawn = (%#v, %v)", result, err) | |
| 49 | + | } | |
| 50 | + | encoded, err := json.Marshal(result.Content) | |
| 51 | + | if err != nil { | |
| 52 | + | t.Fatal(err) | |
| 53 | + | } | |
| 54 | + | if strings.Contains(string(encoded), "interactive report") != reset { | |
| 55 | + | t.Fatalf("reset=%t tool result=%s", reset, encoded) | |
| 56 | + | } | |
| 57 | + | if reset { | |
| 58 | + | if strings.Contains(string(encoded), "host_update") || ag.QueuedMessageCount() != 0 { | |
| 59 | + | t.Fatalf("rejected completion claimed host delivery: %s", encoded) | |
| 60 | + | } | |
| 61 | + | return | |
| 62 | + | } | |
| 63 | + | deadline := time.NewTimer(5 * time.Second) | |
| 64 | + | defer deadline.Stop() | |
| 65 | + | tick := time.NewTicker(time.Millisecond) | |
| 66 | + | defer tick.Stop() | |
| 67 | + | for { | |
| 68 | + | queued := ag.PendingQueuedMessages() | |
| 69 | + | if len(queued) != 0 { | |
| 70 | + | if len(queued) != 1 || !queued[0].HostEvent || strings.Count(queued[0].Text, "interactive report") != 1 || !strings.Contains(queued[0].Text, "[auto-subagents update]") { | |
| 71 | + | t.Fatalf("queued host report = %#v", queued) | |
| 72 | + | } | |
| 73 | + | return | |
| 74 | + | } | |
| 75 | + | select { | |
| 76 | + | case <-deadline.C: | |
| 77 | + | t.Fatal("accepted completion did not slide in") | |
| 78 | + | case <-tick.C: | |
| 79 | + | } | |
| 80 | + | } | |
| 81 | + | }) | |
| 82 | + | } | |
| 83 | + | } | |
| @@ -0,0 +1,83 @@ | |||||
| 1 | + | package modes | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "encoding/json" | |||
| 6 | + | "strings" | |||
| 7 | + | "testing" | |||
| 8 | + | "time" | |||
| 9 | + | ||||
| 10 | + | "github.com/bnema/zut/packages/agent/subagents" | |||
| 11 | + | "github.com/bnema/zut/packages/agent/tools" | |||
| 12 | + | "github.com/bnema/zut/packages/core" | |||
| 13 | + | "github.com/bnema/zut/packages/provider" | |||
| 14 | + | ) | |||
| 15 | + | ||||
| 16 | + | func TestInteractiveWaitReportAcceptance(t *testing.T) { | |||
| 17 | + | for _, reset := range []bool{false, true} { | |||
| 18 | + | name := "accepted" | |||
| 19 | + | if reset { | |||
| 20 | + | name = "cancelled tracking" | |||
| 21 | + | } | |||
| 22 | + | t.Run(name, func(t *testing.T) { | |||
| 23 | + | ctx, cancel := context.WithCancel(t.Context()) | |||
| 24 | + | defer cancel() | |||
| 25 | + | ag := core.NewAgent(nil, "test", "", nil) | |||
| 26 | + | i := &Interactive{agent: ag, busy: true, runCtx: ctx} | |||
| 27 | + | release := i.beginCompletionDeliveryHold() | |||
| 28 | + | defer release() | |||
| 29 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |||
| 30 | + | return subagents.ResidentTurnRunner(func(context.Context, string) error { | |||
| 31 | + | if reset { | |||
| 32 | + | i.cancelCoordinator() | |||
| 33 | + | } | |||
| 34 | + | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "interactive report"}}}}) | |||
| 35 | + | }), nil | |||
| 36 | + | }) | |||
| 37 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |||
| 38 | + | manager.SetAcceptedObserver(func(spec subagents.ResidentChildSpec, turn, _ string) { i.TrackResidentSubagent(spec.ID, turn) }) | |||
| 39 | + | manager.SetCompletionObserver(i.ReportResidentSubagent) | |||
| 40 | + | spawn := &tools.SubagentSpawnTool{ | |||
| 41 | + | ResidentManager: manager, Enabled: func() bool { return true }, | |||
| 42 | + | BuildResidentSpec: func(context.Context, tools.ResidentSpawnRequest) (subagents.ResidentChildSpec, error) { | |||
| 43 | + | return subagents.ResidentChildSpec{ID: "worker", SessionID: "session", Provider: "openai", Model: "test"}, nil | |||
| 44 | + | }, | |||
| 45 | + | } | |||
| 46 | + | result, err := spawn.Execute(ctx, json.RawMessage(`{"task":"report","wait":5}`), nil) | |||
| 47 | + | if err != nil || result.IsError { | |||
| 48 | + | t.Fatalf("spawn = (%#v, %v)", result, err) | |||
| 49 | + | } | |||
| 50 | + | encoded, err := json.Marshal(result.Content) | |||
| 51 | + | if err != nil { | |||
| 52 | + | t.Fatal(err) | |||
| 53 | + | } | |||
| 54 | + | if strings.Contains(string(encoded), "interactive report") != reset { | |||
| 55 | + | t.Fatalf("reset=%t tool result=%s", reset, encoded) | |||
| 56 | + | } | |||
| 57 | + | if reset { | |||
| 58 | + | if strings.Contains(string(encoded), "host_update") || ag.QueuedMessageCount() != 0 { | |||
| 59 | + | t.Fatalf("rejected completion claimed host delivery: %s", encoded) | |||
| 60 | + | } | |||
| 61 | + | return | |||
| 62 | + | } | |||
| 63 | + | deadline := time.NewTimer(5 * time.Second) | |||
| 64 | + | defer deadline.Stop() | |||
| 65 | + | tick := time.NewTicker(time.Millisecond) | |||
| 66 | + | defer tick.Stop() | |||
| 67 | + | for { | |||
| 68 | + | queued := ag.PendingQueuedMessages() | |||
| 69 | + | if len(queued) != 0 { | |||
| 70 | + | if len(queued) != 1 || !queued[0].HostEvent || strings.Count(queued[0].Text, "interactive report") != 1 || !strings.Contains(queued[0].Text, "[auto-subagents update]") { | |||
| 71 | + | t.Fatalf("queued host report = %#v", queued) | |||
| 72 | + | } | |||
| 73 | + | return | |||
| 74 | + | } | |||
| 75 | + | select { | |||
| 76 | + | case <-deadline.C: | |||
| 77 | + | t.Fatal("accepted completion did not slide in") | |||
| 78 | + | case <-tick.C: | |||
| 79 | + | } | |||
| 80 | + | } | |||
| 81 | + | }) | |||
| 82 | + | } | |||
| 83 | + | } | |||
M packages/agent/modes/interactive_resident_subagents.go +8 -3
| @@ -27,16 +27,21 @@ func (i *Interactive) TrackResidentSubagent(childID, turnID string) { | |||
| 27 | 27 | i.invalidate() | |
| 28 | 28 | i.requestCompletionDelivery() | |
| 29 | 29 | } | |
| 30 | - | func (i *Interactive) ReportResidentSubagent(completion subagents.ResidentCompletion) { | |
| 30 | + | ||
| 31 | + | // ReportResidentSubagent reports whether the outcome was accepted for host | |
| 32 | + | // delivery. On false, the caller must retain its own report. Parent cancellation | |
| 33 | + | // can discard accepted reports that have not yet entered the agent queue. | |
| 34 | + | func (i *Interactive) ReportResidentSubagent(completion subagents.ResidentCompletion) bool { | |
| 31 | 35 | if i == nil { | |
| 32 | - | return | |
| 36 | + | return false | |
| 33 | 37 | } | |
| 34 | 38 | if !i.ensureCompletionTracker().Report(completion.Completion()) { | |
| 35 | - | return | |
| 39 | + | return false | |
| 36 | 40 | } | |
| 37 | 41 | i.reloadOpenResidentChildSession(completion.ChildID) | |
| 38 | 42 | i.invalidate() | |
| 39 | 43 | i.requestCompletionDelivery() | |
| 44 | + | return true | |
| 40 | 45 | } | |
| 41 | 46 | func (i *Interactive) reloadOpenResidentChildSession(childID string) { | |
| 42 | 47 | if i == nil { | |
| @@ -27,16 +27,21 @@ func (i *Interactive) TrackResidentSubagent(childID, turnID string) { | |||||
| 27 | i.invalidate() | 27 | i.invalidate() | ||
| 28 | i.requestCompletionDelivery() | 28 | i.requestCompletionDelivery() | ||
| 29 | } | 29 | } | ||
| 30 | - | func (i *Interactive) ReportResidentSubagent(completion subagents.ResidentCompletion) { | 30 | + | |
| 31 | + | // ReportResidentSubagent reports whether the outcome was accepted for host | |||
| 32 | + | // delivery. On false, the caller must retain its own report. Parent cancellation | |||
| 33 | + | // can discard accepted reports that have not yet entered the agent queue. | |||
| 34 | + | func (i *Interactive) ReportResidentSubagent(completion subagents.ResidentCompletion) bool { | |||
| 31 | if i == nil { | 35 | if i == nil { | ||
| 32 | - | return | 36 | + | return false |
| 33 | } | 37 | } | ||
| 34 | if !i.ensureCompletionTracker().Report(completion.Completion()) { | 38 | if !i.ensureCompletionTracker().Report(completion.Completion()) { | ||
| 35 | - | return | 39 | + | return false |
| 36 | } | 40 | } | ||
| 37 | i.reloadOpenResidentChildSession(completion.ChildID) | 41 | i.reloadOpenResidentChildSession(completion.ChildID) | ||
| 38 | i.invalidate() | 42 | i.invalidate() | ||
| 39 | i.requestCompletionDelivery() | 43 | i.requestCompletionDelivery() | ||
| 44 | + | return true | |||
| 40 | } | 45 | } | ||
| 41 | func (i *Interactive) reloadOpenResidentChildSession(childID string) { | 46 | func (i *Interactive) reloadOpenResidentChildSession(childID string) { | ||
| 42 | if i == nil { | 47 | if i == nil { | ||
M packages/agent/modes/resident_subagents_dialog_test.go +1 -1
| @@ -39,7 +39,7 @@ func TestResidentSubagentsDialogShowsUsageMetadata(t *testing.T) { | |||
| 39 | 39 | return journal.RecordAgentEvent(core.EvUsage{Usage: usage, Cumulative: usage}) | |
| 40 | 40 | }), nil | |
| 41 | 41 | }) | |
| 42 | - | manager.SetCompletionObserver(func(subagents.ResidentCompletion) { completed <- struct{}{} }) | |
| 42 | + | manager.SetCompletionObserver(func(subagents.ResidentCompletion) bool { completed <- struct{}{}; return true }) | |
| 43 | 43 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 44 | 44 | for index := range 2 { | |
| 45 | 45 | id := fmt.Sprintf("usage-child-%d", index) | |
| @@ -39,7 +39,7 @@ func TestResidentSubagentsDialogShowsUsageMetadata(t *testing.T) { | |||||
| 39 | return journal.RecordAgentEvent(core.EvUsage{Usage: usage, Cumulative: usage}) | 39 | return journal.RecordAgentEvent(core.EvUsage{Usage: usage, Cumulative: usage}) | ||
| 40 | }), nil | 40 | }), nil | ||
| 41 | }) | 41 | }) | ||
| 42 | - | manager.SetCompletionObserver(func(subagents.ResidentCompletion) { completed <- struct{}{} }) | 42 | + | manager.SetCompletionObserver(func(subagents.ResidentCompletion) bool { completed <- struct{}{}; return true }) |
| 43 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 43 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 44 | for index := range 2 { | 44 | for index := range 2 { | ||
| 45 | id := fmt.Sprintf("usage-child-%d", index) | 45 | id := fmt.Sprintf("usage-child-%d", index) | ||
M packages/agent/orchestration_modes.go +2 -2
| @@ -208,8 +208,8 @@ func newOrchestratedRuntime(_ context.Context, args Args, r Resolved, cfg Config | |||
| 208 | 208 | ContextWindow: r.ContextWindow, | |
| 209 | 209 | Policy: subagentPolicyFromConfig(cfg.Subagents), | |
| 210 | 210 | WebSearchPolicy: webSearchPolicyForRegistry(r.WebSearchPolicy, r.ToolRegistry), | |
| 211 | - | ResidentCompletion: func(completion subagents.ResidentCompletion) { | |
| 212 | - | tracker.Report(completion.Completion()) | |
| 211 | + | ResidentCompletion: func(completion subagents.ResidentCompletion) bool { | |
| 212 | + | return tracker.Report(completion.Completion()) | |
| 213 | 213 | }, | |
| 214 | 214 | OnResidentSpawned: func(spec subagents.ResidentChildSpec, turnID, _ string) { tracker.TrackResident(spec.ID, turnID) }, | |
| 215 | 215 | }) | |
| @@ -208,8 +208,8 @@ func newOrchestratedRuntime(_ context.Context, args Args, r Resolved, cfg Config | |||||
| 208 | ContextWindow: r.ContextWindow, | 208 | ContextWindow: r.ContextWindow, | ||
| 209 | Policy: subagentPolicyFromConfig(cfg.Subagents), | 209 | Policy: subagentPolicyFromConfig(cfg.Subagents), | ||
| 210 | WebSearchPolicy: webSearchPolicyForRegistry(r.WebSearchPolicy, r.ToolRegistry), | 210 | WebSearchPolicy: webSearchPolicyForRegistry(r.WebSearchPolicy, r.ToolRegistry), | ||
| 211 | - | ResidentCompletion: func(completion subagents.ResidentCompletion) { | 211 | + | ResidentCompletion: func(completion subagents.ResidentCompletion) bool { |
| 212 | - | tracker.Report(completion.Completion()) | 212 | + | return tracker.Report(completion.Completion()) |
| 213 | }, | 213 | }, | ||
| 214 | OnResidentSpawned: func(spec subagents.ResidentChildSpec, turnID, _ string) { tracker.TrackResident(spec.ID, turnID) }, | 214 | OnResidentSpawned: func(spec subagents.ResidentChildSpec, turnID, _ string) { tracker.TrackResident(spec.ID, turnID) }, | ||
| 215 | }) | 215 | }) | ||
A packages/agent/subagent_delivery_test.go +54 -0
| @@ -0,0 +1,54 @@ | |||
| 1 | + | package agent | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "encoding/json" | |
| 6 | + | "net/http" | |
| 7 | + | "net/http/httptest" | |
| 8 | + | "strings" | |
| 9 | + | "testing" | |
| 10 | + | ||
| 11 | + | "github.com/bnema/zut/packages/agent/subagents" | |
| 12 | + | "github.com/bnema/zut/packages/core" | |
| 13 | + | ) | |
| 14 | + | ||
| 15 | + | func TestOrchestratedWaitDeliversReportOnlyThroughContinuation(t *testing.T) { | |
| 16 | + | t.Setenv("ZUT_HOME", t.TempDir()) | |
| 17 | + | server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | |
| 18 | + | w.Header().Set("content-type", "text/event-stream") | |
| 19 | + | writeOpenAIChunk(t, w, map[string]any{ | |
| 20 | + | "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "synthetic child report"}, "finish_reason": "stop"}}, | |
| 21 | + | }) | |
| 22 | + | })) | |
| 23 | + | defer server.Close() | |
| 24 | + | tracker := subagents.NewCompletionTracker() | |
| 25 | + | args := Args{CWD: t.TempDir(), Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: server.URL, NoContextFiles: true, NoSkill: true, NoLSP: true} | |
| 26 | + | runtime := newOrchestratedRuntime(t.Context(), args, Resolved{CWD: args.CWD, Provider: args.Provider, Model: args.Model, BaseURL: args.BaseURL}, Config{}, tracker) | |
| 27 | + | t.Cleanup(func() { _ = runtime.ResidentManager().Close(context.Background()) }) | |
| 28 | + | facade := runtime.InjectTools(core.Registry{})["subagent"] | |
| 29 | + | if facade == nil { | |
| 30 | + | t.Fatal("subagent tool unavailable") | |
| 31 | + | } | |
| 32 | + | result, err := facade.Execute(t.Context(), json.RawMessage(`{"action":"spawn","task":"return a report","wait":5}`), nil) | |
| 33 | + | if err != nil || result.IsError { | |
| 34 | + | t.Fatalf("spawn = (%#v, %v)", result, err) | |
| 35 | + | } | |
| 36 | + | encoded, err := json.Marshal(result.Content) | |
| 37 | + | if err != nil { | |
| 38 | + | t.Fatal(err) | |
| 39 | + | } | |
| 40 | + | if strings.Contains(string(encoded), "synthetic child report") || !strings.Contains(string(encoded), "host_update") { | |
| 41 | + | t.Fatalf("tool report = %s", encoded) | |
| 42 | + | } | |
| 43 | + | calls := 0 | |
| 44 | + | got, err := runHeadlessContinuation(t.Context(), tracker, "initial", func(_ context.Context, prompt string) (string, error) { | |
| 45 | + | calls++ | |
| 46 | + | if strings.Count(prompt, "[auto-subagents update]") != 1 || strings.Count(prompt, "synthetic child report") != 1 { | |
| 47 | + | t.Errorf("continuation report = %q", prompt) | |
| 48 | + | } | |
| 49 | + | return "synthesis", nil | |
| 50 | + | }) | |
| 51 | + | if err != nil || calls != 1 || got != "synthesis" { | |
| 52 | + | t.Fatalf("continuation = (%q, %v), calls=%d", got, err, calls) | |
| 53 | + | } | |
| 54 | + | } | |
| @@ -0,0 +1,54 @@ | |||||
| 1 | + | package agent | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "encoding/json" | |||
| 6 | + | "net/http" | |||
| 7 | + | "net/http/httptest" | |||
| 8 | + | "strings" | |||
| 9 | + | "testing" | |||
| 10 | + | ||||
| 11 | + | "github.com/bnema/zut/packages/agent/subagents" | |||
| 12 | + | "github.com/bnema/zut/packages/core" | |||
| 13 | + | ) | |||
| 14 | + | ||||
| 15 | + | func TestOrchestratedWaitDeliversReportOnlyThroughContinuation(t *testing.T) { | |||
| 16 | + | t.Setenv("ZUT_HOME", t.TempDir()) | |||
| 17 | + | server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | |||
| 18 | + | w.Header().Set("content-type", "text/event-stream") | |||
| 19 | + | writeOpenAIChunk(t, w, map[string]any{ | |||
| 20 | + | "choices": []any{map[string]any{"index": 0, "delta": map[string]string{"content": "synthetic child report"}, "finish_reason": "stop"}}, | |||
| 21 | + | }) | |||
| 22 | + | })) | |||
| 23 | + | defer server.Close() | |||
| 24 | + | tracker := subagents.NewCompletionTracker() | |||
| 25 | + | args := Args{CWD: t.TempDir(), Provider: "openai", Model: "gpt-4o", APIKey: "synthetic", BaseURL: server.URL, NoContextFiles: true, NoSkill: true, NoLSP: true} | |||
| 26 | + | runtime := newOrchestratedRuntime(t.Context(), args, Resolved{CWD: args.CWD, Provider: args.Provider, Model: args.Model, BaseURL: args.BaseURL}, Config{}, tracker) | |||
| 27 | + | t.Cleanup(func() { _ = runtime.ResidentManager().Close(context.Background()) }) | |||
| 28 | + | facade := runtime.InjectTools(core.Registry{})["subagent"] | |||
| 29 | + | if facade == nil { | |||
| 30 | + | t.Fatal("subagent tool unavailable") | |||
| 31 | + | } | |||
| 32 | + | result, err := facade.Execute(t.Context(), json.RawMessage(`{"action":"spawn","task":"return a report","wait":5}`), nil) | |||
| 33 | + | if err != nil || result.IsError { | |||
| 34 | + | t.Fatalf("spawn = (%#v, %v)", result, err) | |||
| 35 | + | } | |||
| 36 | + | encoded, err := json.Marshal(result.Content) | |||
| 37 | + | if err != nil { | |||
| 38 | + | t.Fatal(err) | |||
| 39 | + | } | |||
| 40 | + | if strings.Contains(string(encoded), "synthetic child report") || !strings.Contains(string(encoded), "host_update") { | |||
| 41 | + | t.Fatalf("tool report = %s", encoded) | |||
| 42 | + | } | |||
| 43 | + | calls := 0 | |||
| 44 | + | got, err := runHeadlessContinuation(t.Context(), tracker, "initial", func(_ context.Context, prompt string) (string, error) { | |||
| 45 | + | calls++ | |||
| 46 | + | if strings.Count(prompt, "[auto-subagents update]") != 1 || strings.Count(prompt, "synthetic child report") != 1 { | |||
| 47 | + | t.Errorf("continuation report = %q", prompt) | |||
| 48 | + | } | |||
| 49 | + | return "synthesis", nil | |||
| 50 | + | }) | |||
| 51 | + | if err != nil || calls != 1 || got != "synthesis" { | |||
| 52 | + | t.Fatalf("continuation = (%q, %v), calls=%d", got, err, calls) | |||
| 53 | + | } | |||
| 54 | + | } | |||
M packages/agent/subagent_factory_test.go +1 -1
| @@ -434,7 +434,7 @@ func TestResidentChildRunnerResumesWithRetainedHistory(t *testing.T) { | |||
| 434 | 434 | }) | |
| 435 | 435 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 436 | 436 | completed := make(chan subagents.ResidentCompletion, 2) | |
| 437 | - | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) { completed <- c }) | |
| 437 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { completed <- c; return true }) | |
| 438 | 438 | spec := subagents.ResidentChildSpec{ID: "resume-child", SessionID: "session", Provider: "openai", Model: "gpt-4o"} | |
| 439 | 439 | if _, err := manager.Spawn(t.Context(), spec, "review"); err != nil { | |
| 440 | 440 | t.Fatal(err) | |
| @@ -434,7 +434,7 @@ func TestResidentChildRunnerResumesWithRetainedHistory(t *testing.T) { | |||||
| 434 | }) | 434 | }) | ||
| 435 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 435 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 436 | completed := make(chan subagents.ResidentCompletion, 2) | 436 | completed := make(chan subagents.ResidentCompletion, 2) | ||
| 437 | - | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) { completed <- c }) | 437 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { completed <- c; return true }) |
| 438 | spec := subagents.ResidentChildSpec{ID: "resume-child", SessionID: "session", Provider: "openai", Model: "gpt-4o"} | 438 | spec := subagents.ResidentChildSpec{ID: "resume-child", SessionID: "session", Provider: "openai", Model: "gpt-4o"} | ||
| 439 | if _, err := manager.Spawn(t.Context(), spec, "review"); err != nil { | 439 | if _, err := manager.Spawn(t.Context(), spec, "review"); err != nil { | ||
| 440 | t.Fatal(err) | 440 | t.Fatal(err) | ||
M packages/agent/subagent_runtime.go +1 -1
| @@ -85,7 +85,7 @@ type subagentRuntimeConfig struct { | |||
| 85 | 85 | ActiveProvider func() string | |
| 86 | 86 | ActiveModel func() string | |
| 87 | 87 | ActiveReasoning func() string | |
| 88 | - | ResidentCompletion func(subagents.ResidentCompletion) | |
| 88 | + | ResidentCompletion func(subagents.ResidentCompletion) bool | |
| 89 | 89 | OnResidentSpawned func(subagents.ResidentChildSpec, string, string) | |
| 90 | 90 | } | |
| 91 | 91 | ||
| @@ -85,7 +85,7 @@ type subagentRuntimeConfig struct { | |||||
| 85 | ActiveProvider func() string | 85 | ActiveProvider func() string | ||
| 86 | ActiveModel func() string | 86 | ActiveModel func() string | ||
| 87 | ActiveReasoning func() string | 87 | ActiveReasoning func() string | ||
| 88 | - | ResidentCompletion func(subagents.ResidentCompletion) | 88 | + | ResidentCompletion func(subagents.ResidentCompletion) bool |
| 89 | OnResidentSpawned func(subagents.ResidentChildSpec, string, string) | 89 | OnResidentSpawned func(subagents.ResidentChildSpec, string, string) | ||
| 90 | } | 90 | } | ||
| 91 | 91 | ||||
M packages/agent/subagents/resident_cancellation_test.go +1 -1
| @@ -17,7 +17,7 @@ func TestResidentCancellationPreservesInterruptedState(t *testing.T) { | |||
| 17 | 17 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 18 | 18 | spec := ResidentChildSpec{ID: "canceled", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test", Required: true} | |
| 19 | 19 | completions := make(chan ResidentCompletion, 1) | |
| 20 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | |
| 20 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) | |
| 21 | 21 | spawnCtx, cancel := context.WithCancel(t.Context()) | |
| 22 | 22 | defer cancel() | |
| 23 | 23 | if _, err := manager.Spawn(spawnCtx, spec, "investigate"); err != nil { | |
| @@ -17,7 +17,7 @@ func TestResidentCancellationPreservesInterruptedState(t *testing.T) { | |||||
| 17 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 17 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 18 | spec := ResidentChildSpec{ID: "canceled", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test", Required: true} | 18 | spec := ResidentChildSpec{ID: "canceled", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test", Required: true} | ||
| 19 | completions := make(chan ResidentCompletion, 1) | 19 | completions := make(chan ResidentCompletion, 1) | ||
| 20 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | 20 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) |
| 21 | spawnCtx, cancel := context.WithCancel(t.Context()) | 21 | spawnCtx, cancel := context.WithCancel(t.Context()) | ||
| 22 | defer cancel() | 22 | defer cancel() | ||
| 23 | if _, err := manager.Spawn(spawnCtx, spec, "investigate"); err != nil { | 23 | if _, err := manager.Spawn(spawnCtx, spec, "investigate"); err != nil { | ||
M packages/agent/subagents/resident_child.go +3 -0
| @@ -77,6 +77,9 @@ type ResidentCompletion struct { | |||
| 77 | 77 | Task string | |
| 78 | 78 | Err error | |
| 79 | 79 | Summary string | |
| 80 | + | // HostUpdate is set by the manager after the host completion observer | |
| 81 | + | // accepts this outcome. Tool waits then omit the full report. | |
| 82 | + | HostUpdate bool | |
| 80 | 83 | // NotStarted marks a queued follow-up that was dropped before the child | |
| 81 | 84 | // ever ran it, for example because the child was stopped or interrupted. | |
| 82 | 85 | NotStarted bool | |
| @@ -77,6 +77,9 @@ type ResidentCompletion struct { | |||||
| 77 | Task string | 77 | Task string | ||
| 78 | Err error | 78 | Err error | ||
| 79 | Summary string | 79 | Summary string | ||
| 80 | + | // HostUpdate is set by the manager after the host completion observer | |||
| 81 | + | // accepts this outcome. Tool waits then omit the full report. | |||
| 82 | + | HostUpdate bool | |||
| 80 | // NotStarted marks a queued follow-up that was dropped before the child | 83 | // NotStarted marks a queued follow-up that was dropped before the child | ||
| 81 | // ever ran it, for example because the child was stopped or interrupted. | 84 | // ever ran it, for example because the child was stopped or interrupted. | ||
| 82 | NotStarted bool | 85 | NotStarted bool | ||
M packages/agent/subagents/resident_failure_recovery_test.go +2 -2
| @@ -71,7 +71,7 @@ func TestResidentFailureRecoveryKeepsRequiredUnmetUntilResume(t *testing.T) { | |||
| 71 | 71 | manager := NewResidentManager(root, factory) | |
| 72 | 72 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 73 | 73 | completions := make(chan ResidentCompletion, 2) | |
| 74 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | |
| 74 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) | |
| 75 | 75 | if _, err := manager.Spawn(t.Context(), spec, "inspect and verify README"); err != nil { | |
| 76 | 76 | t.Fatal(err) | |
| 77 | 77 | } | |
| @@ -106,7 +106,7 @@ func TestResidentFailureRecoveryKeepsRequiredUnmetUntilResume(t *testing.T) { | |||
| 106 | 106 | t.Fatal(err) | |
| 107 | 107 | } | |
| 108 | 108 | manager = NewResidentManager(root, factory) | |
| 109 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | |
| 109 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) | |
| 110 | 110 | if errs := manager.Reconcile(); len(errs) != 0 { | |
| 111 | 111 | t.Fatal(errs) | |
| 112 | 112 | } | |
| @@ -71,7 +71,7 @@ func TestResidentFailureRecoveryKeepsRequiredUnmetUntilResume(t *testing.T) { | |||||
| 71 | manager := NewResidentManager(root, factory) | 71 | manager := NewResidentManager(root, factory) | ||
| 72 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 72 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 73 | completions := make(chan ResidentCompletion, 2) | 73 | completions := make(chan ResidentCompletion, 2) | ||
| 74 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | 74 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) |
| 75 | if _, err := manager.Spawn(t.Context(), spec, "inspect and verify README"); err != nil { | 75 | if _, err := manager.Spawn(t.Context(), spec, "inspect and verify README"); err != nil { | ||
| 76 | t.Fatal(err) | 76 | t.Fatal(err) | ||
| 77 | } | 77 | } | ||
| @@ -106,7 +106,7 @@ func TestResidentFailureRecoveryKeepsRequiredUnmetUntilResume(t *testing.T) { | |||||
| 106 | t.Fatal(err) | 106 | t.Fatal(err) | ||
| 107 | } | 107 | } | ||
| 108 | manager = NewResidentManager(root, factory) | 108 | manager = NewResidentManager(root, factory) | ||
| 109 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | 109 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) |
| 110 | if errs := manager.Reconcile(); len(errs) != 0 { | 110 | if errs := manager.Reconcile(); len(errs) != 0 { | ||
| 111 | t.Fatal(errs) | 111 | t.Fatal(errs) | ||
| 112 | } | 112 | } | ||
M packages/agent/subagents/resident_journal_test.go +1 -1
| @@ -728,7 +728,7 @@ func TestReconcileResidentJournalTranslatesV2BudgetHistory(t *testing.T) { | |||
| 728 | 728 | manager := NewResidentManager(root, factory) | |
| 729 | 729 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 730 | 730 | completions := make(chan ResidentCompletion, 1) | |
| 731 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | |
| 731 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) | |
| 732 | 732 | if errs := manager.Reconcile(); len(errs) != 0 { | |
| 733 | 733 | t.Fatal(errs) | |
| 734 | 734 | } | |
| @@ -728,7 +728,7 @@ func TestReconcileResidentJournalTranslatesV2BudgetHistory(t *testing.T) { | |||||
| 728 | manager := NewResidentManager(root, factory) | 728 | manager := NewResidentManager(root, factory) | ||
| 729 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 729 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 730 | completions := make(chan ResidentCompletion, 1) | 730 | completions := make(chan ResidentCompletion, 1) | ||
| 731 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | 731 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) |
| 732 | if errs := manager.Reconcile(); len(errs) != 0 { | 732 | if errs := manager.Reconcile(); len(errs) != 0 { | ||
| 733 | t.Fatal(errs) | 733 | t.Fatal(errs) | ||
| 734 | } | 734 | } | ||
M packages/agent/subagents/resident_manager.go +9 -5
| @@ -36,7 +36,7 @@ type ResidentManager struct { | |||
| 36 | 36 | queueTimeout time.Duration | |
| 37 | 37 | allowedRoots []string | |
| 38 | 38 | dispatchMu sync.Mutex | |
| 39 | - | onCompletion func(ResidentCompletion) | |
| 39 | + | onCompletion func(ResidentCompletion) bool | |
| 40 | 40 | completionWaiters map[string][]chan ResidentCompletion | |
| 41 | 41 | onAccepted func(ResidentChildSpec, string, string) | |
| 42 | 42 | onUpdate func(string) | |
| @@ -44,7 +44,9 @@ type ResidentManager struct { | |||
| 44 | 44 | onActivity func(bool) | |
| 45 | 45 | } | |
| 46 | 46 | ||
| 47 | - | func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion)) { | |
| 47 | + | // SetCompletionObserver installs the host report channel. Return true only | |
| 48 | + | // when the host accepts the full report; false keeps tool waits self-contained. | |
| 49 | + | func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion) bool) { | |
| 48 | 50 | if m == nil { | |
| 49 | 51 | return | |
| 50 | 52 | } | |
| @@ -102,13 +104,15 @@ func (m *ResidentManager) reportCompletion(completion ResidentCompletion) { | |||
| 102 | 104 | delete(m.completionWaiters, key) | |
| 103 | 105 | observer := m.onCompletion | |
| 104 | 106 | m.mu.Unlock() | |
| 107 | + | // The host must accept the report before a tool wait can omit its copy. | |
| 108 | + | // Observers run without manager locks, as before. | |
| 109 | + | if observer != nil { | |
| 110 | + | completion.HostUpdate = observer(completion) | |
| 111 | + | } | |
| 105 | 112 | for _, waiter := range waiters { | |
| 106 | 113 | waiter <- completion | |
| 107 | 114 | close(waiter) | |
| 108 | 115 | } | |
| 109 | - | if observer != nil { | |
| 110 | - | observer(completion) | |
| 111 | - | } | |
| 112 | 116 | } | |
| 113 | 117 | ||
| 114 | 118 | // SetAcceptedObserver receives an accepted turn's ID and prompt only after | |
| @@ -36,7 +36,7 @@ type ResidentManager struct { | |||||
| 36 | queueTimeout time.Duration | 36 | queueTimeout time.Duration | ||
| 37 | allowedRoots []string | 37 | allowedRoots []string | ||
| 38 | dispatchMu sync.Mutex | 38 | dispatchMu sync.Mutex | ||
| 39 | - | onCompletion func(ResidentCompletion) | 39 | + | onCompletion func(ResidentCompletion) bool |
| 40 | completionWaiters map[string][]chan ResidentCompletion | 40 | completionWaiters map[string][]chan ResidentCompletion | ||
| 41 | onAccepted func(ResidentChildSpec, string, string) | 41 | onAccepted func(ResidentChildSpec, string, string) | ||
| 42 | onUpdate func(string) | 42 | onUpdate func(string) | ||
| @@ -44,7 +44,9 @@ type ResidentManager struct { | |||||
| 44 | onActivity func(bool) | 44 | onActivity func(bool) | ||
| 45 | } | 45 | } | ||
| 46 | 46 | ||||
| 47 | - | func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion)) { | 47 | + | // SetCompletionObserver installs the host report channel. Return true only |
| 48 | + | // when the host accepts the full report; false keeps tool waits self-contained. | |||
| 49 | + | func (m *ResidentManager) SetCompletionObserver(observer func(ResidentCompletion) bool) { | |||
| 48 | if m == nil { | 50 | if m == nil { | ||
| 49 | return | 51 | return | ||
| 50 | } | 52 | } | ||
| @@ -102,13 +104,15 @@ func (m *ResidentManager) reportCompletion(completion ResidentCompletion) { | |||||
| 102 | delete(m.completionWaiters, key) | 104 | delete(m.completionWaiters, key) | ||
| 103 | observer := m.onCompletion | 105 | observer := m.onCompletion | ||
| 104 | m.mu.Unlock() | 106 | m.mu.Unlock() | ||
| 107 | + | // The host must accept the report before a tool wait can omit its copy. | |||
| 108 | + | // Observers run without manager locks, as before. | |||
| 109 | + | if observer != nil { | |||
| 110 | + | completion.HostUpdate = observer(completion) | |||
| 111 | + | } | |||
| 105 | for _, waiter := range waiters { | 112 | for _, waiter := range waiters { | ||
| 106 | waiter <- completion | 113 | waiter <- completion | ||
| 107 | close(waiter) | 114 | close(waiter) | ||
| 108 | } | 115 | } | ||
| 109 | - | if observer != nil { | |||
| 110 | - | observer(completion) | |||
| 111 | - | } | |||
| 112 | } | 116 | } | ||
| 113 | 117 | ||||
| 114 | // SetAcceptedObserver receives an accepted turn's ID and prompt only after | 118 | // SetAcceptedObserver receives an accepted turn's ID and prompt only after | ||
M packages/agent/subagents/resident_manager_test.go +4 -4
| @@ -135,7 +135,7 @@ func TestResidentManagerResumeWithTurnGeneratesIDForBlank(t *testing.T) { | |||
| 135 | 135 | case <-time.After(5 * time.Second): | |
| 136 | 136 | t.Fatal("initial turn did not complete") | |
| 137 | 137 | } | |
| 138 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completions <- completion }) | |
| 138 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completions <- completion; return true }) | |
| 139 | 139 | if err := manager.resumeWithTurn(context.Background(), "blank-turn-child", "continue", " "); err != nil { | |
| 140 | 140 | t.Fatalf("resumeWithTurn: %v", err) | |
| 141 | 141 | } | |
| @@ -156,7 +156,7 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) { | |||
| 156 | 156 | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "the actual child answer"}}}}) | |
| 157 | 157 | }), nil | |
| 158 | 158 | }) | |
| 159 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | |
| 159 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completed <- completion; return true }) | |
| 160 | 160 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 161 | 161 | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "summary-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil { | |
| 162 | 162 | t.Fatal(err) | |
| @@ -187,7 +187,7 @@ func TestResidentManagerReportsQueuedTurnsWhenFinalResultCannotPersist(t *testin | |||
| 187 | 187 | return nil | |
| 188 | 188 | }), nil | |
| 189 | 189 | }) | |
| 190 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | |
| 190 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completed <- completion; return true }) | |
| 191 | 191 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 192 | 192 | ||
| 193 | 193 | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "persistence-failure-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "initial task"); err != nil { | |
| @@ -233,7 +233,7 @@ func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) { | |||
| 233 | 233 | return ctx.Err() | |
| 234 | 234 | }), nil | |
| 235 | 235 | }) | |
| 236 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | |
| 236 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completed <- completion; return true }) | |
| 237 | 237 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 238 | 238 | ||
| 239 | 239 | child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "interrupted-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task") | |
| @@ -135,7 +135,7 @@ func TestResidentManagerResumeWithTurnGeneratesIDForBlank(t *testing.T) { | |||||
| 135 | case <-time.After(5 * time.Second): | 135 | case <-time.After(5 * time.Second): | ||
| 136 | t.Fatal("initial turn did not complete") | 136 | t.Fatal("initial turn did not complete") | ||
| 137 | } | 137 | } | ||
| 138 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completions <- completion }) | 138 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completions <- completion; return true }) |
| 139 | if err := manager.resumeWithTurn(context.Background(), "blank-turn-child", "continue", " "); err != nil { | 139 | if err := manager.resumeWithTurn(context.Background(), "blank-turn-child", "continue", " "); err != nil { | ||
| 140 | t.Fatalf("resumeWithTurn: %v", err) | 140 | t.Fatalf("resumeWithTurn: %v", err) | ||
| 141 | } | 141 | } | ||
| @@ -156,7 +156,7 @@ func TestResidentManagerCompletionCarriesFinalSummary(t *testing.T) { | |||||
| 156 | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "the actual child answer"}}}}) | 156 | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "the actual child answer"}}}}) | ||
| 157 | }), nil | 157 | }), nil | ||
| 158 | }) | 158 | }) | ||
| 159 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | 159 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completed <- completion; return true }) |
| 160 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 160 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 161 | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "summary-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil { | 161 | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "summary-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task"); err != nil { | ||
| 162 | t.Fatal(err) | 162 | t.Fatal(err) | ||
| @@ -187,7 +187,7 @@ func TestResidentManagerReportsQueuedTurnsWhenFinalResultCannotPersist(t *testin | |||||
| 187 | return nil | 187 | return nil | ||
| 188 | }), nil | 188 | }), nil | ||
| 189 | }) | 189 | }) | ||
| 190 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | 190 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completed <- completion; return true }) |
| 191 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 191 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 192 | 192 | ||||
| 193 | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "persistence-failure-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "initial task"); err != nil { | 193 | if _, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "persistence-failure-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "initial task"); err != nil { | ||
| @@ -233,7 +233,7 @@ func TestResidentManagerReportsAcceptedInterruptedTurn(t *testing.T) { | |||||
| 233 | return ctx.Err() | 233 | return ctx.Err() | ||
| 234 | }), nil | 234 | }), nil | ||
| 235 | }) | 235 | }) | ||
| 236 | - | manager.SetCompletionObserver(func(completion ResidentCompletion) { completed <- completion }) | 236 | + | manager.SetCompletionObserver(func(completion ResidentCompletion) bool { completed <- completion; return true }) |
| 237 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 237 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 238 | 238 | ||||
| 239 | child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "interrupted-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task") | 239 | child, err := manager.Spawn(context.Background(), ResidentChildSpec{ID: "interrupted-child", SessionID: "child-session", Provider: "openai", Model: "gpt-5"}, "task") | ||
M packages/agent/subagents/resident_steer_test.go +1 -1
| @@ -85,7 +85,7 @@ func newSteerTestManager(t *testing.T, runtime ResidentRuntime) (*ResidentManage | |||
| 85 | 85 | t.Helper() | |
| 86 | 86 | manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentRuntime, error) { return runtime, nil }) | |
| 87 | 87 | completions := make(chan ResidentCompletion, 8) | |
| 88 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | |
| 88 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) | |
| 89 | 89 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 90 | 90 | return manager, completions | |
| 91 | 91 | } | |
| @@ -85,7 +85,7 @@ func newSteerTestManager(t *testing.T, runtime ResidentRuntime) (*ResidentManage | |||||
| 85 | t.Helper() | 85 | t.Helper() | ||
| 86 | manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentRuntime, error) { return runtime, nil }) | 86 | manager := NewResidentManager(t.TempDir(), func(ResidentChildSpec, *ResidentJournal) (ResidentRuntime, error) { return runtime, nil }) | ||
| 87 | completions := make(chan ResidentCompletion, 8) | 87 | completions := make(chan ResidentCompletion, 8) | ||
| 88 | - | manager.SetCompletionObserver(func(c ResidentCompletion) { completions <- c }) | 88 | + | manager.SetCompletionObserver(func(c ResidentCompletion) bool { completions <- c; return true }) |
| 89 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | 89 | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | ||
| 90 | return manager, completions | 90 | return manager, completions | ||
| 91 | } | 91 | } | ||
M packages/agent/tools/subagent.go +2 -2
| @@ -26,7 +26,7 @@ type SubagentTool struct { | |||
| 26 | 26 | // subagentSpawnGuidance is the operational contract for delegation. It is the | |
| 27 | 27 | // facade description tail and the internal spawn tool description, so the text | |
| 28 | 28 | // cannot drift between them. | |
| 29 | - | const subagentSpawnGuidance = "Delegate a concrete, bounded scope to a resident sub-agent. For proactive delegation, use an independent sidecar only when the parent has useful non-overlapping work; keep immediate blockers local. A worker owns its scope until completion, so never duplicate it in the parent. If delegation owns the blocking task, end or yield the parent turn. Omit wait to return immediately and receive completion through [auto-subagents update]; set wait to an explicit 1–300 second value only when this turn should wait for the initial task. Set required=true when the outcome is mandatory before the parent's terminal response; failures remain recoverable through the resume action. Never use bash sleep, watch, tail -f, polling loops, repeated status calls, dashboard, metadata, or file checks solely to wait." | |
| 29 | + | const subagentSpawnGuidance = "Delegate a concrete, bounded scope to a resident sub-agent. For proactive delegation, use an independent sidecar only when the parent has useful non-overlapping work; keep immediate blockers local. A worker owns its scope until completion, so never duplicate it in the parent. If delegation owns the blocking task, end or yield the parent turn. Omit wait to return immediately and receive completion through [auto-subagents update]; set wait to an explicit 1–300 second value only when this turn should wait for the initial task. Set required=true when the outcome is mandatory before the parent's terminal response; failures remain recoverable through the resume action. Do not use resume solely to wait for an existing task or request its result again; yield for the automatic completion update instead. Resume requires a real new instruction; queue schedules a separate turn and another completion. Never use bash sleep, watch, tail -f, polling loops, repeated status calls, dashboard, metadata, or file checks solely to wait." | |
| 30 | 30 | ||
| 31 | 31 | // subagentSchema merges the action schemas into the flat, model-facing schema | |
| 32 | 32 | // the project keeps for provider compatibility. Every property is annotated | |
| @@ -79,7 +79,7 @@ const subagentSchema = `{ | |||
| 79 | 79 | "type": "integer", | |
| 80 | 80 | "minimum": 1, | |
| 81 | 81 | "maximum": 300, | |
| 82 | - | "description": "Optional explicit number of seconds to wait for this sub-agent's initial task to finish. Omit to return immediately. The sub-agent continues in the background if this wait expires. With resume it waits for the accepted follow-up turn instead. Only valid when action is spawn or resume." | |
| 82 | + | "description": "Optional explicit number of seconds to wait for this sub-agent's initial task to finish. Omit to return immediately. The sub-agent continues in the background if this wait expires. With resume it waits for the accepted follow-up turn instead. Do not use resume solely to wait for an already running task; yield for its automatic completion update. When the host delivers reports, a completed wait returns status and report_delivery=host_update without repeating the full report. Only valid when action is spawn or resume." | |
| 83 | 83 | }, | |
| 84 | 84 | "isolation": { | |
| 85 | 85 | "type": "string", | |
| @@ -26,7 +26,7 @@ type SubagentTool struct { | |||||
| 26 | // subagentSpawnGuidance is the operational contract for delegation. It is the | 26 | // subagentSpawnGuidance is the operational contract for delegation. It is the | ||
| 27 | // facade description tail and the internal spawn tool description, so the text | 27 | // facade description tail and the internal spawn tool description, so the text | ||
| 28 | // cannot drift between them. | 28 | // cannot drift between them. | ||
| 29 | - | const subagentSpawnGuidance = "Delegate a concrete, bounded scope to a resident sub-agent. For proactive delegation, use an independent sidecar only when the parent has useful non-overlapping work; keep immediate blockers local. A worker owns its scope until completion, so never duplicate it in the parent. If delegation owns the blocking task, end or yield the parent turn. Omit wait to return immediately and receive completion through [auto-subagents update]; set wait to an explicit 1–300 second value only when this turn should wait for the initial task. Set required=true when the outcome is mandatory before the parent's terminal response; failures remain recoverable through the resume action. Never use bash sleep, watch, tail -f, polling loops, repeated status calls, dashboard, metadata, or file checks solely to wait." | 29 | + | const subagentSpawnGuidance = "Delegate a concrete, bounded scope to a resident sub-agent. For proactive delegation, use an independent sidecar only when the parent has useful non-overlapping work; keep immediate blockers local. A worker owns its scope until completion, so never duplicate it in the parent. If delegation owns the blocking task, end or yield the parent turn. Omit wait to return immediately and receive completion through [auto-subagents update]; set wait to an explicit 1–300 second value only when this turn should wait for the initial task. Set required=true when the outcome is mandatory before the parent's terminal response; failures remain recoverable through the resume action. Do not use resume solely to wait for an existing task or request its result again; yield for the automatic completion update instead. Resume requires a real new instruction; queue schedules a separate turn and another completion. Never use bash sleep, watch, tail -f, polling loops, repeated status calls, dashboard, metadata, or file checks solely to wait." |
| 30 | 30 | ||||
| 31 | // subagentSchema merges the action schemas into the flat, model-facing schema | 31 | // subagentSchema merges the action schemas into the flat, model-facing schema | ||
| 32 | // the project keeps for provider compatibility. Every property is annotated | 32 | // the project keeps for provider compatibility. Every property is annotated | ||
| @@ -79,7 +79,7 @@ const subagentSchema = `{ | |||||
| 79 | "type": "integer", | 79 | "type": "integer", | ||
| 80 | "minimum": 1, | 80 | "minimum": 1, | ||
| 81 | "maximum": 300, | 81 | "maximum": 300, | ||
| 82 | - | "description": "Optional explicit number of seconds to wait for this sub-agent's initial task to finish. Omit to return immediately. The sub-agent continues in the background if this wait expires. With resume it waits for the accepted follow-up turn instead. Only valid when action is spawn or resume." | 82 | + | "description": "Optional explicit number of seconds to wait for this sub-agent's initial task to finish. Omit to return immediately. The sub-agent continues in the background if this wait expires. With resume it waits for the accepted follow-up turn instead. Do not use resume solely to wait for an already running task; yield for its automatic completion update. When the host delivers reports, a completed wait returns status and report_delivery=host_update without repeating the full report. Only valid when action is spawn or resume." |
| 83 | }, | 83 | }, | ||
| 84 | "isolation": { | 84 | "isolation": { | ||
| 85 | "type": "string", | 85 | "type": "string", | ||
A packages/agent/tools/subagent_delivery_test.go +223 -0
| @@ -0,0 +1,223 @@ | |||
| 1 | + | package tools | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "encoding/json" | |
| 6 | + | "errors" | |
| 7 | + | "fmt" | |
| 8 | + | "strings" | |
| 9 | + | "sync" | |
| 10 | + | "testing" | |
| 11 | + | "time" | |
| 12 | + | ||
| 13 | + | "github.com/bnema/zut/packages/agent/subagents" | |
| 14 | + | "github.com/bnema/zut/packages/core" | |
| 15 | + | "github.com/bnema/zut/packages/provider" | |
| 16 | + | ) | |
| 17 | + | ||
| 18 | + | // A waited turn still reaches the host, but its report must not also be | |
| 19 | + | // repeated in the tool result. Without a host, the tool remains self-contained. | |
| 20 | + | func TestResidentWaitResultDelivery(t *testing.T) { | |
| 21 | + | for _, action := range []string{"spawn", "resume"} { | |
| 22 | + | for _, route := range []string{"inline", "accepted", "rejected"} { | |
| 23 | + | host := route == "accepted" | |
| 24 | + | for _, turnErr := range []error{nil, errors.New("synthetic failure"), context.Canceled} { | |
| 25 | + | t.Run(fmt.Sprintf("%s/host=%s/%s", action, route, completionStatus(turnErr)), func(t *testing.T) { | |
| 26 | + | const summary = "synthetic worker report" | |
| 27 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |
| 28 | + | return subagents.ResidentTurnRunner(func(_ context.Context, prompt string) error { | |
| 29 | + | if prompt == "initial" { | |
| 30 | + | return nil | |
| 31 | + | } | |
| 32 | + | if err := journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: summary}}}}); err != nil { | |
| 33 | + | return err | |
| 34 | + | } | |
| 35 | + | return turnErr | |
| 36 | + | }), nil | |
| 37 | + | }) | |
| 38 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 39 | + | facade := newDeliveryFacade(t, manager, action) | |
| 40 | + | updates := make(chan subagents.ResidentCompletion, 2) | |
| 41 | + | if route != "inline" { | |
| 42 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { | |
| 43 | + | if host { | |
| 44 | + | updates <- c | |
| 45 | + | } | |
| 46 | + | return host | |
| 47 | + | }) | |
| 48 | + | } | |
| 49 | + | ||
| 50 | + | args := json.RawMessage(`{"action":"spawn","task":"report","wait":5}`) | |
| 51 | + | if action == "resume" { | |
| 52 | + | args = json.RawMessage(`{"action":"resume","agent_id":"report-child","prompt":"report","mode":"queue","wait":5}`) | |
| 53 | + | } | |
| 54 | + | result, err := facade.Execute(t.Context(), args, nil) | |
| 55 | + | if err != nil || result.IsError { | |
| 56 | + | t.Fatalf("Execute = (%#v, %v)", result, err) | |
| 57 | + | } | |
| 58 | + | text := toolResultText(t, result) | |
| 59 | + | if strings.Contains(text, summary) == host { | |
| 60 | + | t.Fatalf("host=%t, tool report = %q", host, text) | |
| 61 | + | } | |
| 62 | + | if !strings.Contains(text, completionStatus(turnErr)) { | |
| 63 | + | t.Fatalf("missing terminal status in %q", text) | |
| 64 | + | } | |
| 65 | + | if host { | |
| 66 | + | if !strings.Contains(text, "host_update") { | |
| 67 | + | t.Fatalf("missing delivery indication in %q", text) | |
| 68 | + | } | |
| 69 | + | select { | |
| 70 | + | case c := <-updates: | |
| 71 | + | if c.Summary != summary || !errors.Is(c.Err, turnErr) { | |
| 72 | + | t.Fatalf("host report = %#v", c) | |
| 73 | + | } | |
| 74 | + | case <-time.After(5 * time.Second): | |
| 75 | + | t.Fatal("host lost the report") | |
| 76 | + | } | |
| 77 | + | select { | |
| 78 | + | case c := <-updates: | |
| 79 | + | t.Fatalf("duplicate host report: %#v", c) | |
| 80 | + | default: | |
| 81 | + | } | |
| 82 | + | } | |
| 83 | + | }) | |
| 84 | + | } | |
| 85 | + | } | |
| 86 | + | } | |
| 87 | + | } | |
| 88 | + | ||
| 89 | + | // Both action tests share the same durable child setup. Resume needs an idle | |
| 90 | + | // initial turn before the tested follow-up; observers are installed afterwards. | |
| 91 | + | func newDeliveryFacade(t *testing.T, manager *subagents.ResidentManager, action string) *SubagentTool { | |
| 92 | + | t.Helper() | |
| 93 | + | spec := subagents.ResidentChildSpec{ID: "report-child", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"} | |
| 94 | + | if action == "resume" { | |
| 95 | + | done, cancel := manager.WatchCompletion(spec.ID, spec.InitialTurnID) | |
| 96 | + | defer cancel() | |
| 97 | + | if _, err := manager.Spawn(t.Context(), spec, "initial"); err != nil { | |
| 98 | + | t.Fatal(err) | |
| 99 | + | } | |
| 100 | + | select { | |
| 101 | + | case <-done: | |
| 102 | + | case <-time.After(5 * time.Second): | |
| 103 | + | t.Fatal("initial turn did not finish") | |
| 104 | + | } | |
| 105 | + | } | |
| 106 | + | return &SubagentTool{ | |
| 107 | + | Spawn: &SubagentSpawnTool{ | |
| 108 | + | ResidentManager: manager, Enabled: func() bool { return true }, | |
| 109 | + | BuildResidentSpec: func(context.Context, ResidentSpawnRequest) (subagents.ResidentChildSpec, error) { return spec, nil }, | |
| 110 | + | }, | |
| 111 | + | Resume: &SubagentResumeTool{ResidentManager: manager, Enabled: func() bool { return true }}, | |
| 112 | + | } | |
| 113 | + | } | |
| 114 | + | ||
| 115 | + | func completionStatus(err error) string { | |
| 116 | + | if errors.Is(err, context.Canceled) { | |
| 117 | + | return "interrupted" | |
| 118 | + | } | |
| 119 | + | if err != nil { | |
| 120 | + | return "failed" | |
| 121 | + | } | |
| 122 | + | return "completed" | |
| 123 | + | } | |
| 124 | + | ||
| 125 | + | // Abandoning a bounded wait never removes the host's completion route. | |
| 126 | + | func TestResidentAbandonedWaitPreservesHostReport(t *testing.T) { | |
| 127 | + | for _, action := range []string{"spawn", "resume"} { | |
| 128 | + | for _, cancelWait := range []bool{false, true} { | |
| 129 | + | t.Run(fmt.Sprintf("%s/cancel=%t", action, cancelWait), func(t *testing.T) { | |
| 130 | + | started := make(chan struct{}, 1) | |
| 131 | + | release := make(chan struct{}) | |
| 132 | + | var once sync.Once | |
| 133 | + | unblock := func() { once.Do(func() { close(release) }) } | |
| 134 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |
| 135 | + | return subagents.ResidentTurnRunner(func(ctx context.Context, prompt string) error { | |
| 136 | + | if prompt == "initial" { | |
| 137 | + | return nil | |
| 138 | + | } | |
| 139 | + | started <- struct{}{} | |
| 140 | + | select { | |
| 141 | + | case <-release: | |
| 142 | + | case <-ctx.Done(): | |
| 143 | + | return ctx.Err() | |
| 144 | + | } | |
| 145 | + | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "late report"}}}}) | |
| 146 | + | }), nil | |
| 147 | + | }) | |
| 148 | + | t.Cleanup(func() { unblock(); _ = manager.Close(context.Background()) }) | |
| 149 | + | facade := newDeliveryFacade(t, manager, action) | |
| 150 | + | updates := make(chan subagents.ResidentCompletion, 2) | |
| 151 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { updates <- c; return true }) | |
| 152 | + | ||
| 153 | + | seconds := 1 | |
| 154 | + | if cancelWait { | |
| 155 | + | seconds = 300 | |
| 156 | + | } | |
| 157 | + | args := fmt.Sprintf(`{"action":"spawn","task":"report","wait":%d}`, seconds) | |
| 158 | + | if action == "resume" { | |
| 159 | + | args = fmt.Sprintf(`{"action":"resume","agent_id":"report-child","prompt":"report","mode":"queue","wait":%d}`, seconds) | |
| 160 | + | } | |
| 161 | + | ctx, cancel := context.WithCancel(t.Context()) | |
| 162 | + | defer cancel() | |
| 163 | + | type toolOutcome struct { | |
| 164 | + | result core.ToolResult | |
| 165 | + | err error | |
| 166 | + | } | |
| 167 | + | done := make(chan toolOutcome, 1) | |
| 168 | + | go func() { | |
| 169 | + | result, err := facade.Execute(ctx, json.RawMessage(args), nil) | |
| 170 | + | done <- toolOutcome{result, err} | |
| 171 | + | }() | |
| 172 | + | select { | |
| 173 | + | case <-started: | |
| 174 | + | case <-time.After(5 * time.Second): | |
| 175 | + | t.Fatal("turn did not start") | |
| 176 | + | } | |
| 177 | + | if cancelWait { | |
| 178 | + | cancel() | |
| 179 | + | } | |
| 180 | + | select { | |
| 181 | + | case outcome := <-done: | |
| 182 | + | if cancelWait { | |
| 183 | + | if !errors.Is(outcome.err, context.Canceled) { | |
| 184 | + | t.Fatalf("wait error = %v", outcome.err) | |
| 185 | + | } | |
| 186 | + | } else { | |
| 187 | + | if outcome.err != nil || outcome.result.IsError { | |
| 188 | + | t.Fatalf("wait outcome = %#v", outcome) | |
| 189 | + | } | |
| 190 | + | if action == "resume" { | |
| 191 | + | response, ok := outcome.result.Details.(subagentResumeResponse) | |
| 192 | + | if !ok || response.Wait == nil || !response.Wait.TimedOut { | |
| 193 | + | t.Fatalf("wait outcome = %#v", outcome) | |
| 194 | + | } | |
| 195 | + | } else if !strings.Contains(toolResultText(t, outcome.result), "timed out") { | |
| 196 | + | t.Fatalf("wait outcome = %#v", outcome) | |
| 197 | + | } | |
| 198 | + | } | |
| 199 | + | case <-time.After(5 * time.Second): | |
| 200 | + | t.Fatal("wait did not return") | |
| 201 | + | } | |
| 202 | + | unblock() | |
| 203 | + | select { | |
| 204 | + | case c := <-updates: | |
| 205 | + | if c.Err != nil || c.Summary != "late report" { | |
| 206 | + | t.Fatalf("late completion = %#v", c) | |
| 207 | + | } | |
| 208 | + | case <-time.After(5 * time.Second): | |
| 209 | + | t.Fatal("late host report was lost") | |
| 210 | + | } | |
| 211 | + | }) | |
| 212 | + | } | |
| 213 | + | } | |
| 214 | + | } | |
| 215 | + | ||
| 216 | + | func TestSubagentGuidanceRejectsResumeAsWait(t *testing.T) { | |
| 217 | + | facade := &SubagentTool{} | |
| 218 | + | for _, text := range []string{facade.Description(), string(facade.Schema())} { | |
| 219 | + | if !strings.Contains(text, "Do not use resume solely to wait") { | |
| 220 | + | t.Fatalf("missing resume-as-wait warning: %s", text) | |
| 221 | + | } | |
| 222 | + | } | |
| 223 | + | } | |
| @@ -0,0 +1,223 @@ | |||||
| 1 | + | package tools | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "encoding/json" | |||
| 6 | + | "errors" | |||
| 7 | + | "fmt" | |||
| 8 | + | "strings" | |||
| 9 | + | "sync" | |||
| 10 | + | "testing" | |||
| 11 | + | "time" | |||
| 12 | + | ||||
| 13 | + | "github.com/bnema/zut/packages/agent/subagents" | |||
| 14 | + | "github.com/bnema/zut/packages/core" | |||
| 15 | + | "github.com/bnema/zut/packages/provider" | |||
| 16 | + | ) | |||
| 17 | + | ||||
| 18 | + | // A waited turn still reaches the host, but its report must not also be | |||
| 19 | + | // repeated in the tool result. Without a host, the tool remains self-contained. | |||
| 20 | + | func TestResidentWaitResultDelivery(t *testing.T) { | |||
| 21 | + | for _, action := range []string{"spawn", "resume"} { | |||
| 22 | + | for _, route := range []string{"inline", "accepted", "rejected"} { | |||
| 23 | + | host := route == "accepted" | |||
| 24 | + | for _, turnErr := range []error{nil, errors.New("synthetic failure"), context.Canceled} { | |||
| 25 | + | t.Run(fmt.Sprintf("%s/host=%s/%s", action, route, completionStatus(turnErr)), func(t *testing.T) { | |||
| 26 | + | const summary = "synthetic worker report" | |||
| 27 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |||
| 28 | + | return subagents.ResidentTurnRunner(func(_ context.Context, prompt string) error { | |||
| 29 | + | if prompt == "initial" { | |||
| 30 | + | return nil | |||
| 31 | + | } | |||
| 32 | + | if err := journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: summary}}}}); err != nil { | |||
| 33 | + | return err | |||
| 34 | + | } | |||
| 35 | + | return turnErr | |||
| 36 | + | }), nil | |||
| 37 | + | }) | |||
| 38 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |||
| 39 | + | facade := newDeliveryFacade(t, manager, action) | |||
| 40 | + | updates := make(chan subagents.ResidentCompletion, 2) | |||
| 41 | + | if route != "inline" { | |||
| 42 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { | |||
| 43 | + | if host { | |||
| 44 | + | updates <- c | |||
| 45 | + | } | |||
| 46 | + | return host | |||
| 47 | + | }) | |||
| 48 | + | } | |||
| 49 | + | ||||
| 50 | + | args := json.RawMessage(`{"action":"spawn","task":"report","wait":5}`) | |||
| 51 | + | if action == "resume" { | |||
| 52 | + | args = json.RawMessage(`{"action":"resume","agent_id":"report-child","prompt":"report","mode":"queue","wait":5}`) | |||
| 53 | + | } | |||
| 54 | + | result, err := facade.Execute(t.Context(), args, nil) | |||
| 55 | + | if err != nil || result.IsError { | |||
| 56 | + | t.Fatalf("Execute = (%#v, %v)", result, err) | |||
| 57 | + | } | |||
| 58 | + | text := toolResultText(t, result) | |||
| 59 | + | if strings.Contains(text, summary) == host { | |||
| 60 | + | t.Fatalf("host=%t, tool report = %q", host, text) | |||
| 61 | + | } | |||
| 62 | + | if !strings.Contains(text, completionStatus(turnErr)) { | |||
| 63 | + | t.Fatalf("missing terminal status in %q", text) | |||
| 64 | + | } | |||
| 65 | + | if host { | |||
| 66 | + | if !strings.Contains(text, "host_update") { | |||
| 67 | + | t.Fatalf("missing delivery indication in %q", text) | |||
| 68 | + | } | |||
| 69 | + | select { | |||
| 70 | + | case c := <-updates: | |||
| 71 | + | if c.Summary != summary || !errors.Is(c.Err, turnErr) { | |||
| 72 | + | t.Fatalf("host report = %#v", c) | |||
| 73 | + | } | |||
| 74 | + | case <-time.After(5 * time.Second): | |||
| 75 | + | t.Fatal("host lost the report") | |||
| 76 | + | } | |||
| 77 | + | select { | |||
| 78 | + | case c := <-updates: | |||
| 79 | + | t.Fatalf("duplicate host report: %#v", c) | |||
| 80 | + | default: | |||
| 81 | + | } | |||
| 82 | + | } | |||
| 83 | + | }) | |||
| 84 | + | } | |||
| 85 | + | } | |||
| 86 | + | } | |||
| 87 | + | } | |||
| 88 | + | ||||
| 89 | + | // Both action tests share the same durable child setup. Resume needs an idle | |||
| 90 | + | // initial turn before the tested follow-up; observers are installed afterwards. | |||
| 91 | + | func newDeliveryFacade(t *testing.T, manager *subagents.ResidentManager, action string) *SubagentTool { | |||
| 92 | + | t.Helper() | |||
| 93 | + | spec := subagents.ResidentChildSpec{ID: "report-child", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"} | |||
| 94 | + | if action == "resume" { | |||
| 95 | + | done, cancel := manager.WatchCompletion(spec.ID, spec.InitialTurnID) | |||
| 96 | + | defer cancel() | |||
| 97 | + | if _, err := manager.Spawn(t.Context(), spec, "initial"); err != nil { | |||
| 98 | + | t.Fatal(err) | |||
| 99 | + | } | |||
| 100 | + | select { | |||
| 101 | + | case <-done: | |||
| 102 | + | case <-time.After(5 * time.Second): | |||
| 103 | + | t.Fatal("initial turn did not finish") | |||
| 104 | + | } | |||
| 105 | + | } | |||
| 106 | + | return &SubagentTool{ | |||
| 107 | + | Spawn: &SubagentSpawnTool{ | |||
| 108 | + | ResidentManager: manager, Enabled: func() bool { return true }, | |||
| 109 | + | BuildResidentSpec: func(context.Context, ResidentSpawnRequest) (subagents.ResidentChildSpec, error) { return spec, nil }, | |||
| 110 | + | }, | |||
| 111 | + | Resume: &SubagentResumeTool{ResidentManager: manager, Enabled: func() bool { return true }}, | |||
| 112 | + | } | |||
| 113 | + | } | |||
| 114 | + | ||||
| 115 | + | func completionStatus(err error) string { | |||
| 116 | + | if errors.Is(err, context.Canceled) { | |||
| 117 | + | return "interrupted" | |||
| 118 | + | } | |||
| 119 | + | if err != nil { | |||
| 120 | + | return "failed" | |||
| 121 | + | } | |||
| 122 | + | return "completed" | |||
| 123 | + | } | |||
| 124 | + | ||||
| 125 | + | // Abandoning a bounded wait never removes the host's completion route. | |||
| 126 | + | func TestResidentAbandonedWaitPreservesHostReport(t *testing.T) { | |||
| 127 | + | for _, action := range []string{"spawn", "resume"} { | |||
| 128 | + | for _, cancelWait := range []bool{false, true} { | |||
| 129 | + | t.Run(fmt.Sprintf("%s/cancel=%t", action, cancelWait), func(t *testing.T) { | |||
| 130 | + | started := make(chan struct{}, 1) | |||
| 131 | + | release := make(chan struct{}) | |||
| 132 | + | var once sync.Once | |||
| 133 | + | unblock := func() { once.Do(func() { close(release) }) } | |||
| 134 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |||
| 135 | + | return subagents.ResidentTurnRunner(func(ctx context.Context, prompt string) error { | |||
| 136 | + | if prompt == "initial" { | |||
| 137 | + | return nil | |||
| 138 | + | } | |||
| 139 | + | started <- struct{}{} | |||
| 140 | + | select { | |||
| 141 | + | case <-release: | |||
| 142 | + | case <-ctx.Done(): | |||
| 143 | + | return ctx.Err() | |||
| 144 | + | } | |||
| 145 | + | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "late report"}}}}) | |||
| 146 | + | }), nil | |||
| 147 | + | }) | |||
| 148 | + | t.Cleanup(func() { unblock(); _ = manager.Close(context.Background()) }) | |||
| 149 | + | facade := newDeliveryFacade(t, manager, action) | |||
| 150 | + | updates := make(chan subagents.ResidentCompletion, 2) | |||
| 151 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { updates <- c; return true }) | |||
| 152 | + | ||||
| 153 | + | seconds := 1 | |||
| 154 | + | if cancelWait { | |||
| 155 | + | seconds = 300 | |||
| 156 | + | } | |||
| 157 | + | args := fmt.Sprintf(`{"action":"spawn","task":"report","wait":%d}`, seconds) | |||
| 158 | + | if action == "resume" { | |||
| 159 | + | args = fmt.Sprintf(`{"action":"resume","agent_id":"report-child","prompt":"report","mode":"queue","wait":%d}`, seconds) | |||
| 160 | + | } | |||
| 161 | + | ctx, cancel := context.WithCancel(t.Context()) | |||
| 162 | + | defer cancel() | |||
| 163 | + | type toolOutcome struct { | |||
| 164 | + | result core.ToolResult | |||
| 165 | + | err error | |||
| 166 | + | } | |||
| 167 | + | done := make(chan toolOutcome, 1) | |||
| 168 | + | go func() { | |||
| 169 | + | result, err := facade.Execute(ctx, json.RawMessage(args), nil) | |||
| 170 | + | done <- toolOutcome{result, err} | |||
| 171 | + | }() | |||
| 172 | + | select { | |||
| 173 | + | case <-started: | |||
| 174 | + | case <-time.After(5 * time.Second): | |||
| 175 | + | t.Fatal("turn did not start") | |||
| 176 | + | } | |||
| 177 | + | if cancelWait { | |||
| 178 | + | cancel() | |||
| 179 | + | } | |||
| 180 | + | select { | |||
| 181 | + | case outcome := <-done: | |||
| 182 | + | if cancelWait { | |||
| 183 | + | if !errors.Is(outcome.err, context.Canceled) { | |||
| 184 | + | t.Fatalf("wait error = %v", outcome.err) | |||
| 185 | + | } | |||
| 186 | + | } else { | |||
| 187 | + | if outcome.err != nil || outcome.result.IsError { | |||
| 188 | + | t.Fatalf("wait outcome = %#v", outcome) | |||
| 189 | + | } | |||
| 190 | + | if action == "resume" { | |||
| 191 | + | response, ok := outcome.result.Details.(subagentResumeResponse) | |||
| 192 | + | if !ok || response.Wait == nil || !response.Wait.TimedOut { | |||
| 193 | + | t.Fatalf("wait outcome = %#v", outcome) | |||
| 194 | + | } | |||
| 195 | + | } else if !strings.Contains(toolResultText(t, outcome.result), "timed out") { | |||
| 196 | + | t.Fatalf("wait outcome = %#v", outcome) | |||
| 197 | + | } | |||
| 198 | + | } | |||
| 199 | + | case <-time.After(5 * time.Second): | |||
| 200 | + | t.Fatal("wait did not return") | |||
| 201 | + | } | |||
| 202 | + | unblock() | |||
| 203 | + | select { | |||
| 204 | + | case c := <-updates: | |||
| 205 | + | if c.Err != nil || c.Summary != "late report" { | |||
| 206 | + | t.Fatalf("late completion = %#v", c) | |||
| 207 | + | } | |||
| 208 | + | case <-time.After(5 * time.Second): | |||
| 209 | + | t.Fatal("late host report was lost") | |||
| 210 | + | } | |||
| 211 | + | }) | |||
| 212 | + | } | |||
| 213 | + | } | |||
| 214 | + | } | |||
| 215 | + | ||||
| 216 | + | func TestSubagentGuidanceRejectsResumeAsWait(t *testing.T) { | |||
| 217 | + | facade := &SubagentTool{} | |||
| 218 | + | for _, text := range []string{facade.Description(), string(facade.Schema())} { | |||
| 219 | + | if !strings.Contains(text, "Do not use resume solely to wait") { | |||
| 220 | + | t.Fatalf("missing resume-as-wait warning: %s", text) | |||
| 221 | + | } | |||
| 222 | + | } | |||
| 223 | + | } | |||
M packages/agent/tools/subagent_resume.go +6 -2
| @@ -116,8 +116,12 @@ func (t *SubagentResumeTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||
| 116 | 116 | result := completion.Completion() | |
| 117 | 117 | outcome.Status = result.Status | |
| 118 | 118 | outcome.Error = result.Error | |
| 119 | - | outcome.Summary = result.Summary | |
| 120 | - | outcome.Undelivered = result.Undelivered | |
| 119 | + | if completion.HostUpdate { | |
| 120 | + | outcome.ReportDelivery = subagentReportHostUpdate | |
| 121 | + | } else { | |
| 122 | + | outcome.Summary = result.Summary | |
| 123 | + | outcome.Undelivered = result.Undelivered | |
| 124 | + | } | |
| 121 | 125 | } | |
| 122 | 126 | } | |
| 123 | 127 | if updated, ok := t.ResidentManager.SnapshotFor(snapshot.ID); ok { | |
| @@ -116,8 +116,12 @@ func (t *SubagentResumeTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||||
| 116 | result := completion.Completion() | 116 | result := completion.Completion() | ||
| 117 | outcome.Status = result.Status | 117 | outcome.Status = result.Status | ||
| 118 | outcome.Error = result.Error | 118 | outcome.Error = result.Error | ||
| 119 | - | outcome.Summary = result.Summary | 119 | + | if completion.HostUpdate { |
| 120 | - | outcome.Undelivered = result.Undelivered | 120 | + | outcome.ReportDelivery = subagentReportHostUpdate |
| 121 | + | } else { | |||
| 122 | + | outcome.Summary = result.Summary | |||
| 123 | + | outcome.Undelivered = result.Undelivered | |||
| 124 | + | } | |||
| 121 | } | 125 | } | ||
| 122 | } | 126 | } | ||
| 123 | if updated, ok := t.ResidentManager.SnapshotFor(snapshot.ID); ok { | 127 | if updated, ok := t.ResidentManager.SnapshotFor(snapshot.ID); ok { | ||
M packages/agent/tools/subagent_spawn.go +6 -0
| @@ -229,6 +229,12 @@ func (t *SubagentSpawnTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||
| 229 | 229 | if waitTimedOut { | |
| 230 | 230 | fmt.Fprintf(&sb, "wait: timed out after %d seconds\n", *a.Wait) | |
| 231 | 231 | sb.WriteString("\nThe accepted sub-agent remains active in the background. It owns the delegated scope: do not repeat that work in the parent. Continue only with a previously selected non-overlapping task, or end/yield for the host-event-driven [auto-subagents update].") | |
| 232 | + | } else if completion.HostUpdate { | |
| 233 | + | fmt.Fprintf(&sb, "report_delivery: %s\n", subagentReportHostUpdate) | |
| 234 | + | if completion.Err != nil { | |
| 235 | + | fmt.Fprintf(&sb, "error: %s\n", completion.Err) | |
| 236 | + | } | |
| 237 | + | sb.WriteString("\nThe full report arrives through [auto-subagents update]. Do not resume or poll solely to retrieve it.") | |
| 232 | 238 | } else if completion.Err != nil { | |
| 233 | 239 | fmt.Fprintf(&sb, "error: %s\n", completion.Err) | |
| 234 | 240 | if completion.Summary != "" { | |
| @@ -229,6 +229,12 @@ func (t *SubagentSpawnTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||||
| 229 | if waitTimedOut { | 229 | if waitTimedOut { | ||
| 230 | fmt.Fprintf(&sb, "wait: timed out after %d seconds\n", *a.Wait) | 230 | fmt.Fprintf(&sb, "wait: timed out after %d seconds\n", *a.Wait) | ||
| 231 | sb.WriteString("\nThe accepted sub-agent remains active in the background. It owns the delegated scope: do not repeat that work in the parent. Continue only with a previously selected non-overlapping task, or end/yield for the host-event-driven [auto-subagents update].") | 231 | sb.WriteString("\nThe accepted sub-agent remains active in the background. It owns the delegated scope: do not repeat that work in the parent. Continue only with a previously selected non-overlapping task, or end/yield for the host-event-driven [auto-subagents update].") | ||
| 232 | + | } else if completion.HostUpdate { | |||
| 233 | + | fmt.Fprintf(&sb, "report_delivery: %s\n", subagentReportHostUpdate) | |||
| 234 | + | if completion.Err != nil { | |||
| 235 | + | fmt.Fprintf(&sb, "error: %s\n", completion.Err) | |||
| 236 | + | } | |||
| 237 | + | sb.WriteString("\nThe full report arrives through [auto-subagents update]. Do not resume or poll solely to retrieve it.") | |||
| 232 | } else if completion.Err != nil { | 238 | } else if completion.Err != nil { | ||
| 233 | fmt.Fprintf(&sb, "error: %s\n", completion.Err) | 239 | fmt.Fprintf(&sb, "error: %s\n", completion.Err) | ||
| 234 | if completion.Summary != "" { | 240 | if completion.Summary != "" { | ||
M packages/agent/tools/subagent_steer_test.go +46 -0
| @@ -3,17 +3,21 @@ package tools | |||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | 5 | "encoding/json" | |
| 6 | + | "strings" | |
| 6 | 7 | "sync" | |
| 7 | 8 | "testing" | |
| 8 | 9 | "time" | |
| 9 | 10 | ||
| 10 | 11 | "github.com/bnema/zut/packages/agent/subagents" | |
| 12 | + | "github.com/bnema/zut/packages/core" | |
| 13 | + | "github.com/bnema/zut/packages/provider" | |
| 11 | 14 | ) | |
| 12 | 15 | ||
| 13 | 16 | // steerRuntime blocks its turn until the first steer arrives, then finishes. | |
| 14 | 17 | type steerRuntime struct { | |
| 15 | 18 | started chan struct{} | |
| 16 | 19 | steered chan string | |
| 20 | + | finish func() error | |
| 17 | 21 | ||
| 18 | 22 | mu sync.Mutex | |
| 19 | 23 | running bool | |
| @@ -29,6 +33,9 @@ func (r *steerRuntime) Run(ctx context.Context, _ string) error { | |||
| 29 | 33 | case <-ctx.Done(): | |
| 30 | 34 | return ctx.Err() | |
| 31 | 35 | case <-r.steered: | |
| 36 | + | if r.finish != nil { | |
| 37 | + | return r.finish() | |
| 38 | + | } | |
| 32 | 39 | return nil | |
| 33 | 40 | } | |
| 34 | 41 | } | |
| @@ -73,6 +80,45 @@ func TestResidentResumeSteersRunningChildAndWaitsOnItsTurn(t *testing.T) { | |||
| 73 | 80 | } | |
| 74 | 81 | } | |
| 75 | 82 | ||
| 83 | + | func TestResidentSteeredWaitUsesAcceptedHostReport(t *testing.T) { | |
| 84 | + | runtime := &steerRuntime{started: make(chan struct{}, 1), steered: make(chan string, 1)} | |
| 85 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |
| 86 | + | runtime.finish = func() error { | |
| 87 | + | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "steered report"}}}}) | |
| 88 | + | } | |
| 89 | + | return runtime, nil | |
| 90 | + | }) | |
| 91 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 92 | + | tracker := subagents.NewCompletionTracker() | |
| 93 | + | manager.SetAcceptedObserver(func(spec subagents.ResidentChildSpec, turn, _ string) { tracker.TrackResident(spec.ID, turn) }) | |
| 94 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { return tracker.Report(c.Completion()) }) | |
| 95 | + | if _, err := manager.Spawn(t.Context(), subagents.ResidentChildSpec{ID: "steered-host", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"}, "investigate"); err != nil { | |
| 96 | + | t.Fatal(err) | |
| 97 | + | } | |
| 98 | + | select { | |
| 99 | + | case <-runtime.started: | |
| 100 | + | case <-time.After(5 * time.Second): | |
| 101 | + | t.Fatal("turn did not start") | |
| 102 | + | } | |
| 103 | + | facade := &SubagentTool{Resume: &SubagentResumeTool{ResidentManager: manager, Enabled: func() bool { return true }}} | |
| 104 | + | result, err := facade.Execute(t.Context(), json.RawMessage(`{"action":"resume","agent_id":"steered-host","prompt":"report now","wait":5}`), nil) | |
| 105 | + | if err != nil || result.IsError { | |
| 106 | + | t.Fatalf("resume = (%#v, %v)", result, err) | |
| 107 | + | } | |
| 108 | + | response, ok := result.Details.(subagentResumeResponse) | |
| 109 | + | if !ok || response.Delivery != "steered" || response.Wait == nil || response.Wait.ReportDelivery != "host_update" || response.Wait.Summary != "" { | |
| 110 | + | t.Fatalf("response = %#v", result.Details) | |
| 111 | + | } | |
| 112 | + | batch, err := tracker.WaitIdle(t.Context()) | |
| 113 | + | if err != nil || len(batch) != 1 || batch[0].TurnID != "initial" { | |
| 114 | + | t.Fatalf("host batch = %#v, error = %v", batch, err) | |
| 115 | + | } | |
| 116 | + | update := subagents.FormatCompletionUpdate(batch, "") | |
| 117 | + | if strings.Count(update, "steered report") != 1 || strings.Contains(toolResultText(t, result), "steered report") { | |
| 118 | + | t.Fatalf("report delivery: update=%q, result=%#v", update, result) | |
| 119 | + | } | |
| 120 | + | } | |
| 121 | + | ||
| 76 | 122 | func TestResidentResumeRejectsUnknownMode(t *testing.T) { | |
| 77 | 123 | manager := subagents.NewResidentManager(t.TempDir(), func(subagents.ResidentChildSpec, *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |
| 78 | 124 | return subagents.ResidentTurnRunner(func(context.Context, string) error { return nil }), nil | |
| @@ -3,17 +3,21 @@ package tools | |||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | "encoding/json" | 5 | "encoding/json" | ||
| 6 | + | "strings" | |||
| 6 | "sync" | 7 | "sync" | ||
| 7 | "testing" | 8 | "testing" | ||
| 8 | "time" | 9 | "time" | ||
| 9 | 10 | ||||
| 10 | "github.com/bnema/zut/packages/agent/subagents" | 11 | "github.com/bnema/zut/packages/agent/subagents" | ||
| 12 | + | "github.com/bnema/zut/packages/core" | |||
| 13 | + | "github.com/bnema/zut/packages/provider" | |||
| 11 | ) | 14 | ) | ||
| 12 | 15 | ||||
| 13 | // steerRuntime blocks its turn until the first steer arrives, then finishes. | 16 | // steerRuntime blocks its turn until the first steer arrives, then finishes. | ||
| 14 | type steerRuntime struct { | 17 | type steerRuntime struct { | ||
| 15 | started chan struct{} | 18 | started chan struct{} | ||
| 16 | steered chan string | 19 | steered chan string | ||
| 20 | + | finish func() error | |||
| 17 | 21 | ||||
| 18 | mu sync.Mutex | 22 | mu sync.Mutex | ||
| 19 | running bool | 23 | running bool | ||
| @@ -29,6 +33,9 @@ func (r *steerRuntime) Run(ctx context.Context, _ string) error { | |||||
| 29 | case <-ctx.Done(): | 33 | case <-ctx.Done(): | ||
| 30 | return ctx.Err() | 34 | return ctx.Err() | ||
| 31 | case <-r.steered: | 35 | case <-r.steered: | ||
| 36 | + | if r.finish != nil { | |||
| 37 | + | return r.finish() | |||
| 38 | + | } | |||
| 32 | return nil | 39 | return nil | ||
| 33 | } | 40 | } | ||
| 34 | } | 41 | } | ||
| @@ -73,6 +80,45 @@ func TestResidentResumeSteersRunningChildAndWaitsOnItsTurn(t *testing.T) { | |||||
| 73 | } | 80 | } | ||
| 74 | } | 81 | } | ||
| 75 | 82 | ||||
| 83 | + | func TestResidentSteeredWaitUsesAcceptedHostReport(t *testing.T) { | |||
| 84 | + | runtime := &steerRuntime{started: make(chan struct{}, 1), steered: make(chan string, 1)} | |||
| 85 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | |||
| 86 | + | runtime.finish = func() error { | |||
| 87 | + | return journal.RecordAgentEvent(core.EvAssistantMessage{Message: provider.Message{Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "steered report"}}}}) | |||
| 88 | + | } | |||
| 89 | + | return runtime, nil | |||
| 90 | + | }) | |||
| 91 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |||
| 92 | + | tracker := subagents.NewCompletionTracker() | |||
| 93 | + | manager.SetAcceptedObserver(func(spec subagents.ResidentChildSpec, turn, _ string) { tracker.TrackResident(spec.ID, turn) }) | |||
| 94 | + | manager.SetCompletionObserver(func(c subagents.ResidentCompletion) bool { return tracker.Report(c.Completion()) }) | |||
| 95 | + | if _, err := manager.Spawn(t.Context(), subagents.ResidentChildSpec{ID: "steered-host", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"}, "investigate"); err != nil { | |||
| 96 | + | t.Fatal(err) | |||
| 97 | + | } | |||
| 98 | + | select { | |||
| 99 | + | case <-runtime.started: | |||
| 100 | + | case <-time.After(5 * time.Second): | |||
| 101 | + | t.Fatal("turn did not start") | |||
| 102 | + | } | |||
| 103 | + | facade := &SubagentTool{Resume: &SubagentResumeTool{ResidentManager: manager, Enabled: func() bool { return true }}} | |||
| 104 | + | result, err := facade.Execute(t.Context(), json.RawMessage(`{"action":"resume","agent_id":"steered-host","prompt":"report now","wait":5}`), nil) | |||
| 105 | + | if err != nil || result.IsError { | |||
| 106 | + | t.Fatalf("resume = (%#v, %v)", result, err) | |||
| 107 | + | } | |||
| 108 | + | response, ok := result.Details.(subagentResumeResponse) | |||
| 109 | + | if !ok || response.Delivery != "steered" || response.Wait == nil || response.Wait.ReportDelivery != "host_update" || response.Wait.Summary != "" { | |||
| 110 | + | t.Fatalf("response = %#v", result.Details) | |||
| 111 | + | } | |||
| 112 | + | batch, err := tracker.WaitIdle(t.Context()) | |||
| 113 | + | if err != nil || len(batch) != 1 || batch[0].TurnID != "initial" { | |||
| 114 | + | t.Fatalf("host batch = %#v, error = %v", batch, err) | |||
| 115 | + | } | |||
| 116 | + | update := subagents.FormatCompletionUpdate(batch, "") | |||
| 117 | + | if strings.Count(update, "steered report") != 1 || strings.Contains(toolResultText(t, result), "steered report") { | |||
| 118 | + | t.Fatalf("report delivery: update=%q, result=%#v", update, result) | |||
| 119 | + | } | |||
| 120 | + | } | |||
| 121 | + | ||||
| 76 | func TestResidentResumeRejectsUnknownMode(t *testing.T) { | 122 | func TestResidentResumeRejectsUnknownMode(t *testing.T) { | ||
| 77 | manager := subagents.NewResidentManager(t.TempDir(), func(subagents.ResidentChildSpec, *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | 123 | manager := subagents.NewResidentManager(t.TempDir(), func(subagents.ResidentChildSpec, *subagents.ResidentJournal) (subagents.ResidentRuntime, error) { | ||
| 78 | return subagents.ResidentTurnRunner(func(context.Context, string) error { return nil }), nil | 124 | return subagents.ResidentTurnRunner(func(context.Context, string) error { return nil }), nil | ||
M packages/agent/tools/subagent_stop.go +5 -0
| @@ -32,8 +32,13 @@ type subagentWaitOutcome struct { | |||
| 32 | 32 | Error string `json:"error,omitempty"` | |
| 33 | 33 | Summary string `json:"summary,omitempty"` | |
| 34 | 34 | Undelivered []string `json:"undelivered,omitempty"` | |
| 35 | + | // ReportDelivery identifies the full report's host channel when the wait | |
| 36 | + | // returns status only. Without a host, Summary and Undelivered stay inline. | |
| 37 | + | ReportDelivery string `json:"report_delivery,omitempty"` | |
| 35 | 38 | } | |
| 36 | 39 | ||
| 40 | + | const subagentReportHostUpdate = "host_update" | |
| 41 | + | ||
| 37 | 42 | type subagentActionResponse struct { | |
| 38 | 43 | Action string `json:"action"` | |
| 39 | 44 | Agent subagentStatusEntry `json:"agent"` | |
| @@ -32,8 +32,13 @@ type subagentWaitOutcome struct { | |||||
| 32 | Error string `json:"error,omitempty"` | 32 | Error string `json:"error,omitempty"` | ||
| 33 | Summary string `json:"summary,omitempty"` | 33 | Summary string `json:"summary,omitempty"` | ||
| 34 | Undelivered []string `json:"undelivered,omitempty"` | 34 | Undelivered []string `json:"undelivered,omitempty"` | ||
| 35 | + | // ReportDelivery identifies the full report's host channel when the wait | |||
| 36 | + | // returns status only. Without a host, Summary and Undelivered stay inline. | |||
| 37 | + | ReportDelivery string `json:"report_delivery,omitempty"` | |||
| 35 | } | 38 | } | ||
| 36 | 39 | ||||
| 40 | + | const subagentReportHostUpdate = "host_update" | |||
| 41 | + | ||||
| 37 | type subagentActionResponse struct { | 42 | type subagentActionResponse struct { | ||
| 38 | Action string `json:"action"` | 43 | Action string `json:"action"` | ||
| 39 | Agent subagentStatusEntry `json:"agent"` | 44 | Agent subagentStatusEntry `json:"agent"` | ||