1package tools
2
3import (
4 "context"
5 "encoding/json"
6 "strings"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/bnema/zut/packages/agent/subagents"
12 "github.com/bnema/zut/packages/core"
13 "github.com/bnema/zut/packages/provider"
14)
15
16// steerRuntime blocks its turn until the first steer arrives, then finishes.
17type steerRuntime struct {
18 started chan struct{}
19 steered chan string
20 finish func() error
21
22 mu sync.Mutex
23 running bool
24}
25
26func (r *steerRuntime) Run(ctx context.Context, _ string) error {
27 r.mu.Lock()
28 r.running = true
29 r.mu.Unlock()
30 defer func() { r.mu.Lock(); r.running = false; r.mu.Unlock() }()
31 r.started <- struct{}{}
32 select {
33 case <-ctx.Done():
34 return ctx.Err()
35 case <-r.steered:
36 if r.finish != nil {
37 return r.finish()
38 }
39 return nil
40 }
41}
42
43func (r *steerRuntime) Steer(prompt string) bool {
44 r.mu.Lock()
45 defer r.mu.Unlock()
46 if !r.running {
47 return false
48 }
49 r.steered <- prompt
50 return true
51}
52
53func (r *steerRuntime) PendingSteers() int { return 0 }
54func (r *steerRuntime) DrainSteers() []string { return nil }
55
56// Resume steers a running child by default, and its wait resolves on the
57// running turn that received the follow-up.
58func TestResidentResumeSteersRunningChildAndWaitsOnItsTurn(t *testing.T) {
59 runtime := &steerRuntime{started: make(chan struct{}, 1), steered: make(chan string, 1)}
60 manager := subagents.NewResidentManager(t.TempDir(), func(subagents.ResidentChildSpec, *subagents.ResidentJournal) (subagents.ResidentRuntime, error) {
61 return runtime, nil
62 })
63 t.Cleanup(func() { _ = manager.Close(context.Background()) })
64 if _, err := manager.Spawn(t.Context(), subagents.ResidentChildSpec{ID: "steer-tool", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"}, "investigate"); err != nil {
65 t.Fatal(err)
66 }
67 select {
68 case <-runtime.started:
69 case <-time.After(5 * time.Second):
70 t.Fatal("turn did not start")
71 }
72 facade := &SubagentTool{Resume: &SubagentResumeTool{ResidentManager: manager, Enabled: func() bool { return true }}}
73 result, err := facade.Execute(t.Context(), json.RawMessage(`{"action":"resume","agent_id":"steer-tool","prompt":"report now","wait":5}`), nil)
74 if err != nil || result.IsError {
75 t.Fatalf("resume = (%#v, %v)", result, err)
76 }
77 response, ok := result.Details.(subagentResumeResponse)
78 if !ok || response.Delivery != "steered" || response.Wait == nil || response.Wait.TimedOut || response.Wait.Status != string(subagents.ResidentCompleted) {
79 t.Fatalf("response = %#v", result.Details)
80 }
81}
82
83func 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
122func TestResidentResumeRejectsUnknownMode(t *testing.T) {
123 manager := subagents.NewResidentManager(t.TempDir(), func(subagents.ResidentChildSpec, *subagents.ResidentJournal) (subagents.ResidentRuntime, error) {
124 return subagents.ResidentTurnRunner(func(context.Context, string) error { return nil }), nil
125 })
126 t.Cleanup(func() { _ = manager.Close(context.Background()) })
127 resume := &SubagentResumeTool{ResidentManager: manager, Enabled: func() bool { return true }}
128 result, err := resume.Execute(t.Context(), json.RawMessage(`{"agent_id":"x","prompt":"p","mode":"now"}`), nil)
129 if err != nil || !result.IsError {
130 t.Fatalf("result = (%#v, %v), want mode error", result, err)
131 }
132}
133
134func TestResidentInterruptToolRequiresRunningTurn(t *testing.T) {
135 manager := subagents.NewResidentManager(t.TempDir(), func(subagents.ResidentChildSpec, *subagents.ResidentJournal) (subagents.ResidentRuntime, error) {
136 return subagents.ResidentTurnRunner(func(context.Context, string) error { return nil }), nil
137 })
138 t.Cleanup(func() { _ = manager.Close(context.Background()) })
139 done, cancel := manager.WatchCompletion("idle-child", "initial")
140 defer cancel()
141 if _, err := manager.Spawn(t.Context(), subagents.ResidentChildSpec{ID: "idle-child", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"}, "task"); err != nil {
142 t.Fatal(err)
143 }
144 <-done
145 interrupt := &SubagentInterruptTool{ResidentManager: manager, Enabled: func() bool { return true }}
146 result, err := interrupt.Execute(t.Context(), json.RawMessage(`{"agent_id":"idle-child"}`), nil)
147 if err != nil || !result.IsError {
148 t.Fatalf("result = (%#v, %v), want no-running-turn error", result, err)
149 }
150}
151
152// The facade dispatches interrupt to a running child and keeps it live.
153func TestSubagentFacadeInterruptsRunningTurn(t *testing.T) {
154 started := make(chan struct{}, 1)
155 manager := subagents.NewResidentManager(t.TempDir(), func(subagents.ResidentChildSpec, *subagents.ResidentJournal) (subagents.ResidentRuntime, error) {
156 return subagents.ResidentTurnRunner(func(ctx context.Context, _ string) error {
157 started <- struct{}{}
158 <-ctx.Done()
159 return ctx.Err()
160 }), nil
161 })
162 t.Cleanup(func() { _ = manager.Close(context.Background()) })
163 done, cancel := manager.WatchCompletion("running-child", "initial")
164 defer cancel()
165 if _, err := manager.Spawn(t.Context(), subagents.ResidentChildSpec{ID: "running-child", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"}, "task"); err != nil {
166 t.Fatal(err)
167 }
168 <-started
169 facade := &SubagentTool{Interrupt: &SubagentInterruptTool{ResidentManager: manager, Enabled: func() bool { return true }}}
170 result, err := facade.Execute(t.Context(), json.RawMessage(`{"action":"interrupt","agent_id":"running-child"}`), nil)
171 if err != nil || result.IsError {
172 t.Fatalf("interrupt = (%#v, %v)", result, err)
173 }
174 if response, ok := result.Details.(subagentActionResponse); !ok || response.Action != "interrupt_requested" {
175 t.Fatalf("response = %#v", result.Details)
176 }
177 select {
178 case completion := <-done:
179 if completion.Completion().Status != string(subagents.ResidentInterrupted) {
180 t.Fatalf("completion = %#v", completion)
181 }
182 case <-time.After(5 * time.Second):
183 t.Fatal("interrupted turn did not complete")
184 }
185 if manager.Get("running-child") == nil {
186 t.Fatal("interrupt removed the live child")
187 }
188 result, err = facade.Execute(t.Context(), json.RawMessage(`{"action":"interrupt","agent_id":"running-child","mode":"steer"}`), nil)
189 if err != nil || !result.IsError {
190 t.Fatalf("mode on interrupt = (%#v, %v), want foreign-field error", result, err)
191 }
192}