1package modes
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "strings"
8 "sync"
9 "testing"
10 "time"
11
12 "github.com/bnema/zut/packages/agent/subagents"
13 agenttools "github.com/bnema/zut/packages/agent/tools"
14 "github.com/bnema/zut/packages/core"
15 "github.com/bnema/zut/packages/provider"
16)
17
18func goalToolRegistry() core.Registry {
19 return core.Registry{agenttools.UpdateGoalToolName: &agenttools.UpdateGoalTool{}}
20}
21
22type goalContinuationClient struct {
23 mu sync.Mutex
24 requests []provider.Request
25 onCall func(call int)
26}
27
28func (c *goalContinuationClient) Name() string { return "goal-continuation" }
29
30func (c *goalContinuationClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
31 c.mu.Lock()
32 c.requests = append(c.requests, req)
33 call := len(c.requests)
34 onCall := c.onCall
35 c.mu.Unlock()
36 if onCall != nil {
37 onCall(call)
38 }
39 out := make(chan provider.Event, 1)
40 out <- provider.EventDone{
41 Stop: provider.StopEnd,
42 Message: provider.Message{
43 Role: provider.RoleAssistant,
44 Content: []provider.Content{provider.TextBlock{Text: "working"}},
45 },
46 }
47 close(out)
48 return out, nil
49}
50
51type goalWorkerClient struct {
52 mu sync.Mutex
53 requests []provider.Request
54 calls chan int
55 onCall func(call int)
56}
57
58func (c *goalWorkerClient) Name() string { return "goal-worker" }
59
60func (c *goalWorkerClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
61 c.mu.Lock()
62 c.requests = append(c.requests, req)
63 call := len(c.requests)
64 onCall := c.onCall
65 c.mu.Unlock()
66 if onCall != nil {
67 onCall(call)
68 }
69 c.calls <- call
70
71 out := make(chan provider.Event, 1)
72 switch call {
73 case 1:
74 out <- provider.EventDone{
75 Stop: provider.StopToolUse,
76 Message: provider.Message{
77 Role: provider.RoleAssistant,
78 Content: []provider.Content{provider.ToolCallBlock{
79 ID: "worker",
80 Name: "start_worker",
81 Arguments: json.RawMessage(`{}`),
82 }},
83 },
84 }
85 default:
86 out <- provider.EventDone{
87 Stop: provider.StopEnd,
88 Message: provider.Message{
89 Role: provider.RoleAssistant,
90 Content: []provider.Content{provider.TextBlock{Text: "done"}},
91 },
92 }
93 }
94 close(out)
95 return out, nil
96}
97
98func (c *goalWorkerClient) request(call int) provider.Request {
99 c.mu.Lock()
100 defer c.mu.Unlock()
101 return c.requests[call-1]
102}
103
104type trackWorkerTool struct {
105 interactive *Interactive
106 worker *subagents.Agent
107}
108
109func (t *trackWorkerTool) Name() string { return "start_worker" }
110func (t *trackWorkerTool) Description() string { return "starts a worker" }
111func (t *trackWorkerTool) Schema() json.RawMessage {
112 return json.RawMessage(`{"type":"object"}`)
113}
114func (t *trackWorkerTool) Execute(_ context.Context, _ json.RawMessage, _ func(string)) (core.ToolResult, error) {
115 t.interactive.TrackSubagentWorker(t.worker, t.worker.Task, false)
116 return core.ToolResult{Content: []provider.Content{provider.TextBlock{Text: "worker started"}}}, nil
117}
118
119func TestActiveGoalWaitsForWorkerAndDeliversCompletion(t *testing.T) {
120 workerRelease := make(chan struct{})
121 var releaseWorkerOnce sync.Once
122 releaseWorker := func() {
123 releaseWorkerOnce.Do(func() { close(workerRelease) })
124 }
125 workerStarted := make(chan struct{})
126 supervisor := subagents.New(subagents.Config{
127 Root: t.TempDir(),
128 RepoRoot: t.TempDir(),
129 NewRunner: func(*subagents.Agent) subagents.Runner {
130 return subagents.RunnerFunc(func(context.Context, subagents.Sink) error {
131 close(workerStarted)
132 <-workerRelease
133 return nil
134 })
135 },
136 })
137 worker, err := supervisor.Spawn(context.Background(), "inspect the upstream worktree")
138 if err != nil {
139 t.Fatal(err)
140 }
141 <-workerStarted
142 t.Cleanup(func() {
143 releaseWorker()
144 worker.Wait()
145 })
146
147 var goalMu sync.Mutex
148 goal := &core.SessionGoal{Objective: "finish the upstream work", Status: core.GoalActive}
149 currentGoal := func() *core.SessionGoal {
150 goalMu.Lock()
151 defer goalMu.Unlock()
152 return cloneSessionGoal(goal)
153 }
154 persistGoal := func(next *core.SessionGoal) error {
155 goalMu.Lock()
156 goal = cloneSessionGoal(next)
157 goalMu.Unlock()
158 return nil
159 }
160 client := &goalWorkerClient{calls: make(chan int, 3), onCall: func(call int) {
161 if call == 3 {
162 _ = persistGoal(&core.SessionGoal{Objective: "finish the upstream work", Status: core.GoalPaused})
163 }
164 }}
165 workerTool := &trackWorkerTool{worker: worker}
166 interactive := NewInteractive(InteractiveConfig{
167 Agent: core.NewAgent(client, "model", "system", core.Registry{
168 "start_worker": workerTool,
169 agenttools.UpdateGoalToolName: &agenttools.UpdateGoalTool{},
170 }),
171 CurrentGoal: currentGoal,
172 PersistGoal: persistGoal,
173 })
174 workerTool.interactive = interactive
175 interactive.runCtx = context.Background()
176
177 interactive.runSlash(context.Background(), "/goal finish the upstream work")
178 for want := 1; want <= 2; want++ {
179 select {
180 case got := <-client.calls:
181 if got != want {
182 t.Fatalf("provider call = %d, want %d", got, want)
183 }
184 case <-time.After(2 * time.Second):
185 t.Fatalf("provider call %d did not start", want)
186 }
187 }
188 deadline := time.NewTimer(2 * time.Second)
189 defer deadline.Stop()
190 poll := time.NewTicker(time.Millisecond)
191 defer poll.Stop()
192 for !interactive.coordinatorAcceptsUserInput() {
193 select {
194 case <-deadline.C:
195 t.Fatal("active goal started another manager turn while its worker was pending")
196 case <-poll.C:
197 }
198 }
199 select {
200 case got := <-client.calls:
201 t.Fatalf("provider call %d started before the worker completed", got)
202 default:
203 }
204
205 releaseWorker()
206 select {
207 case got := <-client.calls:
208 if got != 3 {
209 t.Fatalf("provider call = %d, want completion delivery call 3", got)
210 }
211 case <-time.After(2 * time.Second):
212 t.Fatal("worker completion did not wake the manager")
213 }
214 request := client.request(3)
215 if len(request.Messages) == 0 {
216 t.Fatal("completion delivery request has no messages")
217 }
218 if got := userMessageText(request.Messages[len(request.Messages)-1]); !strings.Contains(got, "inspect the upstream worktree") {
219 t.Fatalf("completion delivery = %q", got)
220 }
221}
222
223func TestActiveGoalStartsAnotherTurnWhenThreadBecomesIdle(t *testing.T) {
224 var mu sync.Mutex
225 var goal *core.SessionGoal
226 currentGoal := func() *core.SessionGoal {
227 mu.Lock()
228 defer mu.Unlock()
229 return cloneSessionGoal(goal)
230 }
231 persistGoal := func(next *core.SessionGoal) error {
232 mu.Lock()
233 goal = cloneSessionGoal(next)
234 mu.Unlock()
235 return nil
236 }
237 client := &goalContinuationClient{onCall: func(call int) {
238 if call == 2 {
239 _ = persistGoal(&core.SessionGoal{Objective: "finish issue 97", Status: core.GoalPaused})
240 }
241 }}
242 turnFlushed := make(chan struct{}, 2)
243 interactive := NewInteractive(InteractiveConfig{
244 Agent: core.NewAgent(client, "model", "system", goalToolRegistry()),
245 CurrentGoal: currentGoal,
246 PersistGoal: persistGoal,
247 FlushSession: func() {
248 turnFlushed <- struct{}{}
249 },
250 })
251 interactive.runCtx = context.Background()
252 interactive.runSlash(context.Background(), "/goal finish issue 97")
253 for turn := 1; turn <= 2; turn++ {
254 select {
255 case <-turnFlushed:
256 case <-time.After(2 * time.Second):
257 t.Fatalf("turn %d did not finish", turn)
258 }
259 }
260
261 client.mu.Lock()
262 defer client.mu.Unlock()
263 if len(client.requests) != 2 {
264 t.Fatalf("provider requests = %d, want 2", len(client.requests))
265 }
266 messages := client.requests[1].Messages
267 if len(messages) == 0 || messages[len(messages)-1].Meta[goalContinueMetaKey] != "true" {
268 t.Fatalf("continuation request tail = %#v", messages)
269 }
270}
271
272func TestReplacementGoalContinuesAfterStaleTurn(t *testing.T) {
273 var mu sync.Mutex
274 goal := &core.SessionGoal{Objective: "replacement objective", Status: core.GoalActive}
275 currentGoal := func() *core.SessionGoal {
276 mu.Lock()
277 defer mu.Unlock()
278 return cloneSessionGoal(goal)
279 }
280 persistGoal := func(next *core.SessionGoal) error {
281 mu.Lock()
282 goal = cloneSessionGoal(next)
283 mu.Unlock()
284 return nil
285 }
286 client := &goalContinuationClient{onCall: func(call int) {
287 if call == 1 {
288 _ = persistGoal(&core.SessionGoal{Objective: "replacement objective", Status: core.GoalPaused})
289 }
290 }}
291 turnFlushed := make(chan struct{}, 1)
292 interactive := NewInteractive(InteractiveConfig{
293 Agent: core.NewAgent(client, "model", "system", goalToolRegistry()),
294 CurrentGoal: currentGoal,
295 PersistGoal: persistGoal,
296 FlushSession: func() {
297 turnFlushed <- struct{}{}
298 },
299 })
300 interactive.runCtx = context.Background()
301 interactive.mu.Lock()
302 interactive.busy = true
303 interactive.mu.Unlock()
304 stale := provider.Message{
305 Role: provider.RoleUser,
306 Content: []provider.Content{provider.TextBlock{Text: "original objective"}},
307 Meta: map[string]string{goalContinueMetaKey: "true"},
308 }
309
310 interactive.startGoalContinuation(context.Background(), stale)
311 select {
312 case <-turnFlushed:
313 case <-time.After(2 * time.Second):
314 t.Fatal("replacement goal did not start")
315 }
316
317 client.mu.Lock()
318 defer client.mu.Unlock()
319 if len(client.requests) != 1 {
320 t.Fatalf("provider requests = %d, want 1", len(client.requests))
321 }
322 tail := client.requests[0].Messages[len(client.requests[0].Messages)-1]
323 if got := userMessageText(tail); !strings.Contains(got, "replacement objective") || strings.Contains(got, "original objective") {
324 t.Fatalf("continuation message = %q", got)
325 }
326}
327
328func TestToolResultPersistenceRunsOutsideTUILock(t *testing.T) {
329 var transitionMu sync.RWMutex
330 transitionCompleted := false
331 callbackStarted := make(chan struct{})
332 allowPersistence := make(chan struct{})
333 transitionHeld := make(chan struct{})
334 transitionDone := make(chan struct{})
335 handled := make(chan struct{})
336 agent := core.NewAgent(nil, "model", "system", goalToolRegistry())
337 agent.CommitToolResult = func(_ string, _ core.ToolResult) error {
338 close(callbackStarted)
339 <-allowPersistence
340 transitionMu.RLock()
341 completed := transitionCompleted
342 transitionMu.RUnlock()
343 if !completed {
344 return errors.New("session transition did not complete")
345 }
346 return nil
347 }
348 interactive := NewInteractive(InteractiveConfig{Agent: agent})
349 commitErr := make(chan error, 1)
350
351 go func() {
352 commitErr <- agent.CommitToolResult("tool", core.ToolResult{})
353 interactive.handleEvent(core.EvToolResult{ID: "tool", Result: core.ToolResult{}})
354 close(handled)
355 }()
356 select {
357 case <-callbackStarted:
358 case <-time.After(2 * time.Second):
359 t.Fatal("tool-result callback did not start")
360 }
361 go func() {
362 transitionMu.Lock()
363 close(transitionHeld)
364 interactive.ApplySessionAgent(core.NewAgent(nil, "model", "system", goalToolRegistry()), "provider", "model")
365 transitionCompleted = true
366 transitionMu.Unlock()
367 close(transitionDone)
368 }()
369 select {
370 case <-transitionHeld:
371 case <-time.After(2 * time.Second):
372 t.Fatal("session transition did not acquire its lock")
373 }
374 select {
375 case <-transitionDone:
376 case <-time.After(2 * time.Second):
377 close(allowPersistence)
378 t.Fatal("session transition deadlocked with tool-result persistence")
379 }
380 close(allowPersistence)
381 select {
382 case <-handled:
383 case <-time.After(2 * time.Second):
384 t.Fatal("tool-result callback did not finish")
385 }
386 if err := <-commitErr; err != nil {
387 t.Fatalf("commit tool result: %v", err)
388 }
389}
390
391func TestGoalBadgeAcceptsNextPersistedManagerGoal(t *testing.T) {
392 interactive := NewInteractive(InteractiveConfig{
393 Agent: core.NewAgent(nil, "model", "system", goalToolRegistry()),
394 CurrentGoal: func() *core.SessionGoal {
395 return &core.SessionGoal{Objective: "implement the fix", Status: core.GoalActive, Owner: core.GoalOwnerManager}
396 },
397 })
398 interactive.mu.Lock()
399 interactive.goalStatus = core.GoalDone
400 interactive.mu.Unlock()
401
402 interactive.handleEvent(core.EvToolResult{ID: "goal", Result: core.ToolResult{
403 Details: agenttools.GoalUpdate{Status: core.GoalActive, Objective: "implement the fix"},
404 }})
405 interactive.mu.Lock()
406 defer interactive.mu.Unlock()
407 if interactive.goalStatus != core.GoalActive {
408 t.Fatalf("goal badge = %q, want active", interactive.goalStatus)
409 }
410}
411
412func TestGoalBadgeUsesDurableStatusAfterPersistenceFailure(t *testing.T) {
413 goal := &core.SessionGoal{Objective: "finish issue 97", Status: core.GoalActive}
414 interactive := NewInteractive(InteractiveConfig{
415 Agent: core.NewAgent(nil, "model", "system", goalToolRegistry()),
416 CurrentGoal: func() *core.SessionGoal { return cloneSessionGoal(goal) },
417 })
418 // Core converts a goal update to an error result when durable persistence
419 // fails, so the presentation layer must leave the active badge unchanged.
420 result := core.ToolResult{
421 Content: []provider.Content{provider.TextBlock{Text: "tool result state could not be persisted"}},
422 IsError: true,
423 }
424
425 interactive.handleEvent(core.EvToolResult{ID: "goal", Result: result})
426 interactive.mu.Lock()
427 defer interactive.mu.Unlock()
428 if interactive.goalStatus != core.GoalActive {
429 t.Fatalf("goal badge = %q, want durable active status", interactive.goalStatus)
430 }
431}
432
433func TestGoalSlashCommandStartsAndControlsGoal(t *testing.T) {
434 var goal *core.SessionGoal
435 interactive := NewInteractive(InteractiveConfig{
436 Agent: core.NewAgent(nil, "model", "system", goalToolRegistry()),
437 CurrentGoal: func() *core.SessionGoal { return cloneSessionGoal(goal) },
438 PersistGoal: func(next *core.SessionGoal) error {
439 goal = cloneSessionGoal(next)
440 return nil
441 },
442 })
443 interactive.busy = true // exercise controls without launching a provider turn
444
445 interactive.runSlash(context.Background(), "/goal finish issue 97")
446 if goal == nil || goal.Status != core.GoalActive || goal.Objective != "finish issue 97" {
447 t.Fatalf("started goal = %#v", goal)
448 }
449 if interactive.goalStatus != core.GoalActive {
450 t.Fatalf("started badge = %q", interactive.goalStatus)
451 }
452
453 interactive.runSlash(context.Background(), "/goal pause")
454 if goal.Status != core.GoalPaused {
455 t.Fatalf("paused goal = %#v", goal)
456 }
457 if interactive.goalStatus != core.GoalPaused {
458 t.Fatalf("paused badge = %q", interactive.goalStatus)
459 }
460 interactive.runSlash(context.Background(), "/goal resume")
461 if goal.Status != core.GoalActive {
462 t.Fatalf("resumed goal = %#v", goal)
463 }
464 if interactive.goalStatus != core.GoalActive {
465 t.Fatalf("resumed badge = %q", interactive.goalStatus)
466 }
467 interactive.runSlash(context.Background(), "/goal clear")
468 if goal != nil {
469 t.Fatalf("cleared goal = %#v", goal)
470 }
471 if interactive.goalStatus != "" {
472 t.Fatalf("cleared badge = %q", interactive.goalStatus)
473 }
474 interactive.runSlash(context.Background(), "/goal pause deployment safely")
475 if goal == nil || goal.Status != core.GoalActive || goal.Objective != "pause deployment safely" {
476 t.Fatalf("multiword control-like objective = %#v", goal)
477 }
478 goal.Status = core.GoalBlocked
479 goal.Reason = "waiting for input"
480 interactive.runSlash(context.Background(), "/goal")
481 if !strings.Contains(interactive.statusOK, "waiting for input") {
482 t.Fatalf("goal status = %q", interactive.statusOK)
483 }
484}
485
486func TestGoalHistoryShowsOwnerAndReason(t *testing.T) {
487 interactive := NewInteractive(InteractiveConfig{
488 CurrentGoal: func() *core.SessionGoal { return nil },
489 PersistGoal: func(*core.SessionGoal) error { return nil },
490 CurrentGoalHistory: func() []core.SessionGoal {
491 return []core.SessionGoal{
492 {Objective: "reproduce", Status: core.GoalDone, Owner: core.GoalOwnerManager},
493 {Objective: "await input", Status: core.GoalBlocked, Owner: core.GoalOwnerUser, Reason: "missing logs"},
494 }
495 },
496 })
497
498 interactive.runSlash(context.Background(), "/goal history")
499 if !strings.Contains(interactive.statusOK, "done (manager): reproduce") || !strings.Contains(interactive.statusOK, "blocked (user): await input (missing logs)") {
500 t.Fatalf("goal history = %q", interactive.statusOK)
501 }
502}
503
504func TestGoalControlRequiresSessionPersistence(t *testing.T) {
505 interactive := NewInteractive(InteractiveConfig{})
506 interactive.runSlash(context.Background(), "/goal pause")
507 if !strings.Contains(interactive.statusErr, "session persistence is unavailable") {
508 t.Fatalf("status error = %q", interactive.statusErr)
509 }
510}
511
512func TestGoalCommandDoesNotMutateLiveStateWhenPersistenceFails(t *testing.T) {
513 goal := &core.SessionGoal{Objective: "finish issue 97", Status: core.GoalActive}
514 interactive := NewInteractive(InteractiveConfig{
515 Agent: core.NewAgent(nil, "model", "system", goalToolRegistry()),
516 CurrentGoal: func() *core.SessionGoal { return goal },
517 PersistGoal: func(*core.SessionGoal) error { return errors.New("disk full") },
518 })
519 interactive.busy = true
520
521 interactive.runSlash(context.Background(), "/goal pause")
522 if goal.Status != core.GoalActive {
523 t.Fatalf("live goal status = %q, want active", goal.Status)
524 }
525 if interactive.goalStatus != core.GoalActive {
526 t.Fatalf("goal badge = %q, want active", interactive.goalStatus)
527 }
528}
529
530func TestGoalSlashCommandRequiresUpdateGoalTool(t *testing.T) {
531 var persisted bool
532 interactive := NewInteractive(InteractiveConfig{
533 Agent: core.NewAgent(nil, "model", "system", nil),
534 CurrentGoal: func() *core.SessionGoal { return nil },
535 PersistGoal: func(*core.SessionGoal) error {
536 persisted = true
537 return nil
538 },
539 })
540
541 interactive.runSlash(context.Background(), "/goal finish issue 97")
542 if persisted {
543 t.Fatal("goal was persisted without the update_goal tool")
544 }
545 if !strings.Contains(interactive.statusErr, "update_goal is unavailable") {
546 t.Fatalf("status error = %q", interactive.statusErr)
547 }
548}
549
550func TestGoalContinuationRequiresActiveIdleGoalAndYieldsToQueuedUser(t *testing.T) {
551 goal := &core.SessionGoal{Objective: "finish issue 97", Status: core.GoalActive}
552 agent := core.NewAgent(nil, "model", "system", goalToolRegistry())
553 interactive := &Interactive{
554 agent: agent,
555 cfg: InteractiveConfig{CurrentGoal: func() *core.SessionGoal { return cloneSessionGoal(goal) }},
556 }
557
558 message, ok := interactive.goalContinuationIfIdle()
559 if !ok {
560 t.Fatal("active idle goal did not request continuation")
561 }
562 if message.Meta[goalContinueMetaKey] != "true" || message.Role != provider.RoleUser {
563 t.Fatalf("continuation message = %#v", message)
564 }
565 if !isHiddenTranscriptMessage(message) {
566 t.Fatal("goal continuation was rendered as a user message")
567 }
568
569 agent.QueueMessage("user follow-up")
570 if _, ok := interactive.goalContinuationIfIdle(); ok {
571 t.Fatal("goal continuation bypassed queued user work")
572 }
573 agent.DrainQueuedMessages()
574 goal.Status = core.GoalPaused
575 if _, ok := interactive.goalContinuationIfIdle(); ok {
576 t.Fatal("paused goal requested continuation")
577 }
578}
579
580func cloneSessionGoal(goal *core.SessionGoal) *core.SessionGoal {
581 if goal == nil {
582 return nil
583 }
584 clone := *goal
585 return &clone
586}