1package modes
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/bnema/zut/packages/core"
12 "github.com/bnema/zut/packages/provider"
13 "github.com/bnema/zut/packages/tui"
14)
15
16type compactQueueClient struct {
17 compactionStarted chan struct{}
18 releaseCompaction chan struct{}
19 followUpRequest chan provider.Request
20 onFollowUp func()
21
22 mu sync.Mutex
23 calls int
24}
25
26func (c *compactQueueClient) Name() string { return "compact-queue-test" }
27
28func (c *compactQueueClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
29 c.mu.Lock()
30 c.calls++
31 call := c.calls
32 c.mu.Unlock()
33
34 out := make(chan provider.Event, 2)
35 go func() {
36 defer close(out)
37 if call == 1 {
38 close(c.compactionStarted)
39 select {
40 case <-ctx.Done():
41 out <- provider.EventDone{Stop: provider.StopAborted, Err: ctx.Err()}
42 case <-c.releaseCompaction:
43 out <- provider.EventTextDelta{Delta: "summary"}
44 out <- provider.EventDone{Stop: provider.StopEnd}
45 }
46 return
47 }
48
49 if c.onFollowUp != nil {
50 c.onFollowUp()
51 }
52 c.followUpRequest <- req
53 out <- provider.EventDone{
54 Stop: provider.StopEnd,
55 Message: provider.Message{
56 Role: provider.RoleAssistant,
57 Content: []provider.Content{provider.TextBlock{Text: "done"}},
58 },
59 }
60 }()
61 return out, nil
62}
63
64type autoCompactPacerClient struct {
65 compactionStarted chan struct{}
66 releaseCompaction chan struct{}
67
68 mu sync.Mutex
69 calls int
70}
71
72type goalLengthClient struct {
73 requests chan provider.Request
74 onThird func()
75
76 mu sync.Mutex
77 calls int
78}
79
80func (c *goalLengthClient) Name() string { return "goal-length-test" }
81
82func (c *goalLengthClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
83 c.mu.Lock()
84 c.calls++
85 call := c.calls
86 c.mu.Unlock()
87
88 out := make(chan provider.Event, 2)
89 go func() {
90 defer close(out)
91 c.requests <- req
92 switch call {
93 case 1:
94 out <- provider.EventDone{
95 Stop: provider.StopLength,
96 Message: provider.Message{
97 Role: provider.RoleAssistant,
98 Content: []provider.Content{provider.TextBlock{Text: "partial work"}},
99 },
100 }
101 case 2:
102 out <- provider.EventTextDelta{Delta: "durable summary"}
103 out <- provider.EventDone{Stop: provider.StopEnd}
104 default:
105 if c.onThird != nil {
106 c.onThird()
107 }
108 out <- provider.EventDone{
109 Stop: provider.StopEnd,
110 Message: provider.Message{
111 Role: provider.RoleAssistant,
112 Content: []provider.Content{provider.TextBlock{Text: "continued"}},
113 },
114 }
115 }
116 }()
117 return out, nil
118}
119
120func (c *autoCompactPacerClient) Name() string { return "auto-compact-pacer-test" }
121
122func (c *autoCompactPacerClient) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Event, error) {
123 c.mu.Lock()
124 c.calls++
125 call := c.calls
126 c.mu.Unlock()
127
128 out := make(chan provider.Event, 3)
129 go func() {
130 defer close(out)
131 switch call {
132 case 1:
133 const response = "a final response that is still buffered for typewriter rendering"
134 out <- provider.EventTextDelta{Delta: response}
135 out <- provider.EventUsage{Usage: provider.Usage{InputTokens: 150000}}
136 out <- provider.EventDone{
137 Stop: provider.StopEnd,
138 Message: provider.Message{
139 Role: provider.RoleAssistant,
140 Content: []provider.Content{provider.TextBlock{Text: response}},
141 },
142 }
143 case 2:
144 close(c.compactionStarted)
145 select {
146 case <-ctx.Done():
147 out <- provider.EventDone{Stop: provider.StopAborted, Err: ctx.Err()}
148 case <-c.releaseCompaction:
149 out <- provider.EventTextDelta{Delta: "summary"}
150 out <- provider.EventDone{Stop: provider.StopEnd}
151 }
152 }
153 }()
154 return out, nil
155}
156
157func TestAutoCompactionSettlesFinalStreamingStateBeforeCompacting(t *testing.T) {
158 client := &autoCompactPacerClient{
159 compactionStarted: make(chan struct{}),
160 releaseCompaction: make(chan struct{}),
161 }
162 agent := core.NewAgent(client, "test-model", "", nil)
163 threshold := 70
164 interactive := NewInteractive(InteractiveConfig{
165 Agent: agent,
166 Provider: "anthropic",
167 Model: "claude-sonnet-4-5-20250929",
168 AutoCompactThreshold: &threshold,
169 })
170 interactive.runCtx = context.Background()
171
172 // Do not start the pacer: the test controls the unpainted final delta so
173 // it can prove compaction never races with a still-live streaming frame.
174 interactive.startTurn(context.Background(), "initial request")
175 select {
176 case <-client.compactionStarted:
177 case <-time.After(2 * time.Second):
178 t.Fatal("auto-compaction did not start")
179 }
180
181 interactive.mu.Lock()
182 streaming, pending := interactive.streaming.Len(), len(interactive.streamPending)
183 streamOn, flushPending := interactive.streamOn, interactive.streamFlushPending
184 interactive.mu.Unlock()
185 if streaming != 0 || pending != 0 || streamOn || flushPending {
186 t.Fatalf("auto-compaction started with live stream state: rendered=%d pending=%d on=%t flush=%t", streaming, pending, streamOn, flushPending)
187 }
188
189 close(client.releaseCompaction)
190}
191
192func TestActiveGoalLengthStopCompactsAndContinuesAsGoal(t *testing.T) {
193 var goalMu sync.Mutex
194 goal := &core.SessionGoal{ID: "goal-1", Objective: "finish the goal", Status: core.GoalActive}
195 currentGoal := func() *core.SessionGoal {
196 goalMu.Lock()
197 defer goalMu.Unlock()
198 return cloneSessionGoal(goal)
199 }
200 persistGoal := func(next *core.SessionGoal) error {
201 goalMu.Lock()
202 goal = cloneSessionGoal(next)
203 goalMu.Unlock()
204 return nil
205 }
206 client := &goalLengthClient{
207 requests: make(chan provider.Request, 3),
208 onThird: func() {
209 _ = persistGoal(&core.SessionGoal{ID: "goal-1", Objective: "finish the goal", Status: core.GoalPaused})
210 },
211 }
212 agent := core.NewAgent(client, "test-model", "", goalToolRegistry())
213 persistedHandoff := make(chan string, 4)
214 interactive := NewInteractive(InteractiveConfig{
215 Agent: agent,
216 CurrentGoal: currentGoal,
217 PersistGoal: persistGoal,
218 PersistCompactHandoff: func(state json.RawMessage) error {
219 persistedHandoff <- string(state)
220 return nil
221 },
222 })
223 interactive.runCtx = context.Background()
224
225 run, err := interactive.startGoalRun(currentGoal())
226 if err != nil || run == nil {
227 t.Fatalf("start goal run = (%v, %v)", run, err)
228 }
229 interactive.startGoalContinuation(context.Background(), mustGoalContinuationMessage(t, interactive), run)
230
231 first := receiveRequest(t, client.requests)
232 if first.Messages[len(first.Messages)-1].Meta[goalContinueMetaKey] != "true" {
233 t.Fatalf("initial request tail = %#v, want goal continuation", first.Messages)
234 }
235 select {
236 case handoff := <-persistedHandoff:
237 if handoff != `{"version":1,"reason":"goal_compaction_pending"}` {
238 t.Fatalf("pre-compaction handoff = %q, want durable goal owner", handoff)
239 }
240 case <-time.After(2 * time.Second):
241 t.Fatal("goal handoff was not persisted before compaction")
242 }
243 compaction := receiveRequest(t, client.requests)
244 if compaction.MaxTokens != 4096 {
245 t.Fatalf("second request MaxTokens = %d, want compaction request", compaction.MaxTokens)
246 }
247 continued := receiveRequest(t, client.requests)
248 if continued.Messages[len(continued.Messages)-1].Meta[goalContinueMetaKey] != "true" {
249 t.Fatalf("post-compaction request tail = %#v, want leased goal continuation", continued.Messages)
250 }
251
252 waitInteractiveIdle(t, interactive)
253}
254
255func TestRestoredPendingGoalCompactionRunsBeforeGoalLease(t *testing.T) {
256 var goalMu sync.Mutex
257 goal := &core.SessionGoal{ID: "goal-1", Objective: "finish the goal", Status: core.GoalActive}
258 currentGoal := func() *core.SessionGoal {
259 goalMu.Lock()
260 defer goalMu.Unlock()
261 return cloneSessionGoal(goal)
262 }
263 persistGoal := func(next *core.SessionGoal) error {
264 goalMu.Lock()
265 goal = cloneSessionGoal(next)
266 goalMu.Unlock()
267 return nil
268 }
269 client := &goalLengthClient{
270 requests: make(chan provider.Request, 2),
271 calls: 1,
272 onThird: func() {
273 _ = persistGoal(&core.SessionGoal{ID: "goal-1", Objective: "finish the goal", Status: core.GoalPaused})
274 },
275 }
276 agent := core.NewAgent(client, "test-model", "", goalToolRegistry())
277 agent.SetMessages([]provider.Message{
278 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "original goal work"}}},
279 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "truncated"}}},
280 })
281 interactive := NewInteractive(InteractiveConfig{
282 Agent: agent,
283 CurrentGoal: currentGoal,
284 PersistGoal: persistGoal,
285 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"goal_compaction_pending"}`),
286 })
287 interactive.runCtx = context.Background()
288
289 interactive.startRestoredCompactHandoff(context.Background())
290 compaction := receiveRequest(t, client.requests)
291 if compaction.MaxTokens != 4096 {
292 t.Fatalf("restored first request MaxTokens = %d, want compaction", compaction.MaxTokens)
293 }
294 request := receiveRequest(t, client.requests)
295 if request.Messages[len(request.Messages)-1].Meta[goalContinueMetaKey] != "true" {
296 t.Fatalf("restored request tail = %#v, want goal continuation", request.Messages)
297 }
298 waitInteractiveIdle(t, interactive)
299}
300
301func TestRestoredGoalCompactHandoffResumesWithGoalLease(t *testing.T) {
302 var goalMu sync.Mutex
303 goal := &core.SessionGoal{ID: "goal-1", Objective: "finish the goal", Status: core.GoalActive}
304 currentGoal := func() *core.SessionGoal {
305 goalMu.Lock()
306 defer goalMu.Unlock()
307 return cloneSessionGoal(goal)
308 }
309 persistGoal := func(next *core.SessionGoal) error {
310 goalMu.Lock()
311 goal = cloneSessionGoal(next)
312 goalMu.Unlock()
313 return nil
314 }
315 client := &goalLengthClient{
316 requests: make(chan provider.Request, 1),
317 calls: 2,
318 onThird: func() {
319 _ = persistGoal(&core.SessionGoal{ID: "goal-1", Objective: "finish the goal", Status: core.GoalPaused})
320 },
321 }
322 agent := core.NewAgent(client, "test-model", "", goalToolRegistry())
323 interactive := NewInteractive(InteractiveConfig{
324 Agent: agent,
325 CurrentGoal: currentGoal,
326 PersistGoal: persistGoal,
327 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"goal"}`),
328 })
329 interactive.runCtx = context.Background()
330
331 interactive.startRestoredCompactHandoff(context.Background())
332 request := receiveRequest(t, client.requests)
333 if request.Messages[len(request.Messages)-1].Meta[goalContinueMetaKey] != "true" {
334 t.Fatalf("restored request tail = %#v, want goal continuation", request.Messages)
335 }
336 waitInteractiveIdle(t, interactive)
337}
338
339func mustGoalContinuationMessage(t *testing.T, interactive *Interactive) provider.Message {
340 t.Helper()
341 message, ok := interactive.goalContinuationMessage()
342 if !ok {
343 t.Fatal("goal continuation message unavailable")
344 }
345 return message
346}
347
348func receiveRequest(t *testing.T, requests <-chan provider.Request) provider.Request {
349 t.Helper()
350 select {
351 case req := <-requests:
352 return req
353 case <-time.After(2 * time.Second):
354 t.Fatal("timed out waiting for provider request")
355 return provider.Request{}
356 }
357}
358
359func TestActiveGoalContinuesAfterAutomaticCompaction(t *testing.T) {
360 var goalMu sync.Mutex
361 goal := &core.SessionGoal{Objective: "finish the goal", Status: core.GoalActive}
362 currentGoal := func() *core.SessionGoal {
363 goalMu.Lock()
364 defer goalMu.Unlock()
365 return cloneSessionGoal(goal)
366 }
367 persistGoal := func(next *core.SessionGoal) error {
368 goalMu.Lock()
369 goal = cloneSessionGoal(next)
370 goalMu.Unlock()
371 return nil
372 }
373 client := &compactQueueClient{
374 compactionStarted: make(chan struct{}),
375 releaseCompaction: make(chan struct{}),
376 followUpRequest: make(chan provider.Request, 1),
377 onFollowUp: func() {
378 // Pause before publishing the follow-up so the loop stops after one turn.
379 _ = persistGoal(&core.SessionGoal{Objective: "finish the goal", Status: core.GoalPaused})
380 },
381 }
382 agent := core.NewAgent(client, "test-model", "", goalToolRegistry())
383 agent.SetMessages([]provider.Message{
384 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "request"}}},
385 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "first result"}}},
386 })
387 interactive := NewInteractive(InteractiveConfig{Agent: agent, CurrentGoal: currentGoal, PersistGoal: persistGoal})
388 interactive.runCtx = context.Background()
389
390 interactive.runCompact(context.Background(), compactContinuationRequest{origin: compactOriginAfterTurnThreshold, lastStop: provider.StopEnd})
391 select {
392 case <-client.compactionStarted:
393 case <-time.After(2 * time.Second):
394 t.Fatal("automatic compaction did not start")
395 }
396 close(client.releaseCompaction)
397
398 select {
399 case req := <-client.followUpRequest:
400 if len(req.Messages) == 0 || req.Messages[len(req.Messages)-1].Meta[goalContinueMetaKey] != "true" {
401 t.Fatalf("goal continuation tail = %#v", req.Messages)
402 }
403 case <-time.After(2 * time.Second):
404 t.Fatal("active goal did not continue after automatic compaction")
405 }
406 waitInteractiveIdle(t, interactive)
407}
408
409func TestActiveGoalDoesNotContinueAfterManualCompaction(t *testing.T) {
410 goal := &core.SessionGoal{Objective: "finish the goal", Status: core.GoalActive}
411 client := &compactQueueClient{
412 compactionStarted: make(chan struct{}),
413 releaseCompaction: make(chan struct{}),
414 followUpRequest: make(chan provider.Request, 1),
415 }
416 agent := core.NewAgent(client, "test-model", "", goalToolRegistry())
417 agent.SetMessages([]provider.Message{
418 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "request"}}},
419 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "result"}}},
420 })
421 interactive := NewInteractive(InteractiveConfig{
422 Agent: agent,
423 CurrentGoal: func() *core.SessionGoal { return cloneSessionGoal(goal) },
424 PersistGoal: func(next *core.SessionGoal) error {
425 goal = cloneSessionGoal(next)
426 return nil
427 },
428 })
429 interactive.runCtx = context.Background()
430
431 interactive.runCompact(context.Background(), compactContinuationRequest{origin: compactOriginManual})
432 select {
433 case <-client.compactionStarted:
434 case <-time.After(2 * time.Second):
435 t.Fatal("manual compaction did not start")
436 }
437 close(client.releaseCompaction)
438 waitInteractiveIdle(t, interactive)
439 select {
440 case req := <-client.followUpRequest:
441 t.Fatalf("manual compaction started goal continuation: %#v", req)
442 default:
443 }
444}
445
446func TestPromptSubmittedDuringCompactionStartsFollowUpTurn(t *testing.T) {
447 client := &compactQueueClient{
448 compactionStarted: make(chan struct{}),
449 releaseCompaction: make(chan struct{}),
450 followUpRequest: make(chan provider.Request, 1),
451 }
452 agent := core.NewAgent(client, "test-model", "", nil)
453 agent.SetMessages([]provider.Message{
454 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "one"}}},
455 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "two"}}},
456 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "three"}}},
457 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},
458 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "five"}}},
459 })
460 interactive := NewInteractive(InteractiveConfig{Agent: agent})
461 interactive.runCtx = context.Background()
462
463 interactive.runCompact(context.Background(), compactContinuationRequest{origin: compactOriginManual})
464 select {
465 case <-client.compactionStarted:
466 case <-time.After(2 * time.Second):
467 t.Fatal("compaction did not start")
468 }
469
470 interactive.ed.SetValue("follow up")
471 interactive.handleKey(context.Background(), tui.Key{Kind: tui.KeyEnter})
472 close(client.releaseCompaction)
473
474 select {
475 case req := <-client.followUpRequest:
476 if !requestContainsUserText(req, "follow up") {
477 t.Fatalf("follow-up request does not contain queued prompt: %#v", req.Messages)
478 }
479 case <-time.After(2 * time.Second):
480 t.Fatal("queued prompt did not start a turn after compaction")
481 }
482}
483
484type thresholdAutoCompactClient struct {
485 compactionStarted chan struct{}
486 releaseCompaction chan struct{}
487 followUpRequest chan provider.Request
488 firstStop provider.StopReason
489 firstContent []provider.Content
490 followUpContents []string
491 blockFollowUpCall int
492 releaseFollowUp <-chan struct{}
493
494 mu sync.Mutex
495 calls int
496 followUps int
497}
498
499func (c *thresholdAutoCompactClient) Name() string { return "threshold-auto-compact-test" }
500
501func (c *thresholdAutoCompactClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
502 c.mu.Lock()
503 c.calls++
504 call := c.calls
505 c.mu.Unlock()
506
507 out := make(chan provider.Event, 3)
508 go func() {
509 defer close(out)
510 switch call {
511 case 1:
512 stop := c.firstStop
513 if stop == "" {
514 stop = provider.StopEnd
515 }
516 content := c.firstContent
517 if len(content) == 0 {
518 content = []provider.Content{provider.ReasoningBlock{Summary: "still working"}}
519 }
520 out <- provider.EventUsage{Usage: provider.Usage{InputTokens: 150000}}
521 out <- provider.EventDone{
522 Stop: stop,
523 Message: provider.Message{
524 Role: provider.RoleAssistant,
525 Content: content,
526 },
527 }
528 case 2:
529 close(c.compactionStarted)
530 select {
531 case <-ctx.Done():
532 out <- provider.EventDone{Stop: provider.StopAborted, Err: ctx.Err()}
533 case <-c.releaseCompaction:
534 out <- provider.EventTextDelta{Delta: "summary"}
535 out <- provider.EventDone{Stop: provider.StopEnd}
536 }
537 default:
538 c.mu.Lock()
539 c.followUps++
540 followUp := call - 2
541 text := "continued"
542 if followUp > 0 && followUp <= len(c.followUpContents) {
543 text = c.followUpContents[followUp-1]
544 }
545 c.mu.Unlock()
546 c.followUpRequest <- req
547 if c.blockFollowUpCall == call {
548 select {
549 case <-ctx.Done():
550 out <- provider.EventDone{Stop: provider.StopAborted, Err: ctx.Err()}
551 return
552 case <-c.releaseFollowUp:
553 }
554 }
555 out <- provider.EventDone{
556 Stop: provider.StopEnd,
557 Message: provider.Message{
558 Role: provider.RoleAssistant,
559 Content: []provider.Content{provider.TextBlock{Text: text}},
560 },
561 }
562 }
563 }()
564 return out, nil
565}
566
567type restoredHandoffClient struct {
568 requests chan provider.Request
569 replies []string
570
571 mu sync.Mutex
572 calls int
573}
574
575func (c *restoredHandoffClient) Name() string { return "restored-handoff-test" }
576
577func (c *restoredHandoffClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
578 c.mu.Lock()
579 call := c.calls
580 c.calls++
581 text := "done"
582 if call < len(c.replies) {
583 text = c.replies[call]
584 }
585 c.mu.Unlock()
586
587 out := make(chan provider.Event, 1)
588 go func() {
589 defer close(out)
590 c.requests <- req
591 out <- provider.EventDone{
592 Stop: provider.StopEnd,
593 Message: provider.Message{
594 Role: provider.RoleAssistant,
595 Content: []provider.Content{provider.TextBlock{Text: text}},
596 },
597 }
598 }()
599 return out, nil
600}
601
602type heldCompletionClient struct {
603 started chan struct{}
604 release chan struct{}
605}
606
607func (c *heldCompletionClient) Name() string { return "held-completion-test" }
608
609func (c *heldCompletionClient) Stream(ctx context.Context, _ provider.Request) (<-chan provider.Event, error) {
610 out := make(chan provider.Event, 1)
611 go func() {
612 defer close(out)
613 close(c.started)
614 select {
615 case <-ctx.Done():
616 out <- provider.EventDone{Stop: provider.StopAborted, Err: ctx.Err()}
617 case <-c.release:
618 out <- provider.EventDone{
619 Stop: provider.StopEnd,
620 Message: provider.Message{
621 Role: provider.RoleAssistant,
622 Content: []provider.Content{provider.TextBlock{Text: "Next I will inspect the remaining call sites."}},
623 },
624 }
625 }
626 }()
627 return out, nil
628}
629
630func TestRestoredStatusHandoffContinuesExistingPromptWithoutDuplication(t *testing.T) {
631 client := &restoredHandoffClient{
632 requests: make(chan provider.Request, 2),
633 replies: []string{"Next I will run the targeted tests.", "All required work is complete."},
634 }
635 agent := core.NewAgent(client, "test-model", "", nil)
636 agent.SetMessages([]provider.Message{{
637 Role: provider.RoleUser,
638 Content: []provider.Content{provider.TextBlock{Text: autoCompactContinuationPrompt}},
639 Meta: map[string]string{autoCompactContinueMetaKey: "true"},
640 }})
641 interactive := NewInteractive(InteractiveConfig{
642 Agent: agent,
643 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"status_rescue","rescue_attempts":1}`),
644 })
645 interactive.runCtx = context.Background()
646
647 interactive.startRestoredCompactHandoff(context.Background())
648 for attempt := 1; attempt <= 2; attempt++ {
649 select {
650 case req := <-client.requests:
651 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != attempt {
652 t.Fatalf("request %d continuation prompt count = %d, want %d: %#v", attempt, got, attempt, req.Messages)
653 }
654 case <-time.After(2 * time.Second):
655 t.Fatalf("restored handoff request %d did not start", attempt)
656 }
657 }
658 waitInteractiveIdle(t, interactive)
659}
660
661func TestRestoredHandoffAppendsMissingContinuationPrompt(t *testing.T) {
662 client := &restoredHandoffClient{requests: make(chan provider.Request, 1)}
663 agent := core.NewAgent(client, "test-model", "", nil)
664 interactive := NewInteractive(InteractiveConfig{
665 Agent: agent,
666 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"structural_tail"}`),
667 })
668 interactive.runCtx = context.Background()
669
670 interactive.startRestoredCompactHandoff(context.Background())
671 select {
672 case req := <-client.requests:
673 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
674 t.Fatalf("restored request continuation prompt count = %d, want 1: %#v", got, req.Messages)
675 }
676 case <-time.After(2 * time.Second):
677 t.Fatal("restored handoff did not append a missing continuation prompt")
678 }
679 waitInteractiveIdle(t, interactive)
680}
681
682func TestRestoredHandoffAfterFinalReplyIsDiscarded(t *testing.T) {
683 persisted := make(chan string, 1)
684 client := &restoredHandoffClient{requests: make(chan provider.Request, 1)}
685 agent := core.NewAgent(client, "test-model", "", nil)
686 agent.SetMessages([]provider.Message{
687 {
688 Role: provider.RoleUser,
689 Content: []provider.Content{provider.TextBlock{Text: autoCompactContinuationPrompt}},
690 Meta: map[string]string{autoCompactContinueMetaKey: "true"},
691 },
692 {
693 Role: provider.RoleAssistant,
694 Content: []provider.Content{provider.TextBlock{Text: "All required work is complete."}},
695 },
696 })
697 interactive := NewInteractive(InteractiveConfig{
698 Agent: agent,
699 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"status_rescue","rescue_attempts":1}`),
700 PersistCompactHandoff: func(state json.RawMessage) error {
701 select {
702 case persisted <- string(state):
703 default:
704 }
705 return nil
706 },
707 })
708
709 interactive.startRestoredCompactHandoff(context.Background())
710 interactive.mu.Lock()
711 busy := interactive.busy
712 interactive.mu.Unlock()
713 if busy {
714 t.Fatal("discarded restored handoff left the turn slot reserved")
715 }
716 select {
717 case state := <-persisted:
718 if state != "" {
719 t.Fatalf("cleared handoff = %q, want empty", state)
720 }
721 case <-time.After(2 * time.Second):
722 t.Fatal("stale handoff was not cleared")
723 }
724 select {
725 case req := <-client.requests:
726 t.Fatalf("stale handoff started a provider request: %#v", req.Messages)
727 default:
728 }
729}
730
731func TestRestoredHandoffReservesTurnBeforePersistence(t *testing.T) {
732 client := &restoredHandoffClient{
733 requests: make(chan provider.Request, 1),
734 replies: []string{"All required work is complete."},
735 }
736 agent := core.NewAgent(client, "test-model", "", nil)
737 agent.SetMessages([]provider.Message{{
738 Role: provider.RoleUser,
739 Content: []provider.Content{provider.TextBlock{Text: autoCompactContinuationPrompt}},
740 Meta: map[string]string{autoCompactContinueMetaKey: "true"},
741 }})
742 persistenceStarted := make(chan struct{})
743 releasePersistence := make(chan struct{})
744 released := false
745 defer func() {
746 if !released {
747 close(releasePersistence)
748 }
749 }()
750 interactive := NewInteractive(InteractiveConfig{
751 Agent: agent,
752 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"status_rescue","rescue_attempts":1}`),
753 PersistCompactHandoff: func(json.RawMessage) error {
754 close(persistenceStarted)
755 <-releasePersistence
756 return nil
757 },
758 })
759 interactive.runCtx = context.Background()
760 interactive.mu.Lock()
761 interactive.queued = []core.QueuedMessage{{Text: "newer user request"}}
762 interactive.mu.Unlock()
763
764 done := make(chan struct{})
765 go func() {
766 interactive.startRestoredCompactHandoff(context.Background())
767 close(done)
768 }()
769 select {
770 case <-persistenceStarted:
771 case <-time.After(2 * time.Second):
772 t.Fatal("restored handoff did not reach persistence")
773 }
774 interactive.mu.Lock()
775 busy := interactive.busy
776 interactive.mu.Unlock()
777 if !busy {
778 t.Fatal("restored handoff did not reserve the turn slot before persistence")
779 }
780 close(releasePersistence)
781 released = true
782 <-done
783 select {
784 case <-client.requests:
785 case <-time.After(2 * time.Second):
786 t.Fatal("restored handoff did not continue after persistence")
787 }
788 waitInteractiveIdle(t, interactive)
789}
790
791func TestStatusRescueClearsHandoffWhenAgentIsRemovedDuringTurn(t *testing.T) {
792 client := &heldCompletionClient{started: make(chan struct{}), release: make(chan struct{})}
793 agent := core.NewAgent(client, "test-model", "", nil)
794 agent.SetMessages([]provider.Message{{
795 Role: provider.RoleUser,
796 Content: []provider.Content{provider.TextBlock{Text: autoCompactContinuationPrompt}},
797 Meta: map[string]string{autoCompactContinueMetaKey: "true"},
798 }})
799 persisted := make(chan string, 1)
800 interactive := NewInteractive(InteractiveConfig{
801 Agent: agent,
802 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"status_rescue","rescue_attempts":1}`),
803 PersistCompactHandoff: func(state json.RawMessage) error {
804 select {
805 case persisted <- string(state):
806 default:
807 }
808 return nil
809 },
810 })
811 interactive.runCtx = context.Background()
812
813 interactive.startTurnRequest(context.Background(), "", nil, true, false)
814 select {
815 case <-client.started:
816 case <-time.After(2 * time.Second):
817 t.Fatal("continuation did not start")
818 }
819 interactive.mu.Lock()
820 interactive.agent = nil
821 interactive.mu.Unlock()
822 close(client.release)
823
824 select {
825 case state := <-persisted:
826 if state != "" {
827 t.Fatalf("cleared handoff = %q, want empty", state)
828 }
829 case <-time.After(2 * time.Second):
830 t.Fatal("agent removal did not clear the persisted handoff")
831 }
832 waitInteractiveIdle(t, interactive)
833}
834
835func TestRestoredHandoffDefersToQueuedUserPrompt(t *testing.T) {
836 client := &restoredHandoffClient{requests: make(chan provider.Request, 1)}
837 agent := core.NewAgent(client, "test-model", "", nil)
838 agent.SetMessages([]provider.Message{{
839 Role: provider.RoleUser,
840 Content: []provider.Content{provider.TextBlock{Text: autoCompactContinuationPrompt}},
841 Meta: map[string]string{autoCompactContinueMetaKey: "true"},
842 }})
843 interactive := NewInteractive(InteractiveConfig{
844 Agent: agent,
845 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"status_rescue","rescue_attempts":1}`),
846 })
847 interactive.runCtx = context.Background()
848 interactive.mu.Lock()
849 interactive.queued = []core.QueuedMessage{{Text: "newer user request"}}
850 interactive.mu.Unlock()
851
852 interactive.startRestoredCompactHandoff(context.Background())
853 select {
854 case req := <-client.requests:
855 if got := requestUserTextCount(req, "newer user request"); got != 1 {
856 t.Fatalf("newer user request count = %d, want 1: %#v", got, req.Messages)
857 }
858 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
859 t.Fatalf("restored continuation prompt count = %d, want only the existing prompt: %#v", got, req.Messages)
860 }
861 case <-time.After(2 * time.Second):
862 t.Fatal("queued user prompt did not run instead of restored handoff")
863 }
864 waitInteractiveIdle(t, interactive)
865}
866
867func TestRestoredExhaustedStatusHandoffDoesNotContinueAgain(t *testing.T) {
868 client := &restoredHandoffClient{
869 requests: make(chan provider.Request, 2),
870 replies: []string{"Next I will run the targeted tests."},
871 }
872 agent := core.NewAgent(client, "test-model", "", nil)
873 agent.SetMessages([]provider.Message{{
874 Role: provider.RoleUser,
875 Content: []provider.Content{provider.TextBlock{Text: autoCompactContinuationPrompt}},
876 Meta: map[string]string{autoCompactContinueMetaKey: "true"},
877 }})
878 interactive := NewInteractive(InteractiveConfig{
879 Agent: agent,
880 InitialCompactHandoff: json.RawMessage(`{"version":1,"reason":"status_rescue","rescue_attempts":2}`),
881 })
882 interactive.runCtx = context.Background()
883
884 interactive.startRestoredCompactHandoff(context.Background())
885 select {
886 case req := <-client.requests:
887 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
888 t.Fatalf("restored request continuation prompt count = %d, want 1: %#v", got, req.Messages)
889 }
890 case <-time.After(2 * time.Second):
891 t.Fatal("restored handoff did not start")
892 }
893 waitInteractiveIdle(t, interactive)
894 select {
895 case req := <-client.requests:
896 t.Fatalf("exhausted handoff started another request: %#v", req.Messages)
897 default:
898 }
899}
900
901func TestThresholdAutoCompactionContinuesMostRecentIntent(t *testing.T) {
902 client := &thresholdAutoCompactClient{
903 compactionStarted: make(chan struct{}),
904 releaseCompaction: make(chan struct{}),
905 followUpRequest: make(chan provider.Request, 1),
906 }
907 agent := core.NewAgent(client, "test-model", "", nil)
908 threshold := 70
909 interactive := NewInteractive(InteractiveConfig{
910 Agent: agent,
911 Provider: "anthropic",
912 Model: "claude-sonnet-4-5-20250929",
913 AutoCompactThreshold: &threshold,
914 })
915 interactive.runCtx = context.Background()
916
917 interactive.startTurn(context.Background(), "initial request")
918 select {
919 case <-client.compactionStarted:
920 case <-time.After(2 * time.Second):
921 t.Fatal("threshold auto-compaction did not start")
922 }
923 close(client.releaseCompaction)
924
925 select {
926 case req := <-client.followUpRequest:
927 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
928 t.Fatalf("continuation request contains auto-continue prompt %d times, want 1: %#v", got, req.Messages)
929 }
930 case <-time.After(2 * time.Second):
931 t.Fatal("threshold auto-compaction did not continue the most recent intent")
932 }
933}
934
935func TestThresholdAutoCompactionRescuesStatusAfterCompaction(t *testing.T) {
936 persistedHandoff := make(chan string, 1)
937 client := &thresholdAutoCompactClient{
938 compactionStarted: make(chan struct{}),
939 releaseCompaction: make(chan struct{}),
940 followUpRequest: make(chan provider.Request, 1),
941 firstContent: []provider.Content{
942 provider.TextBlock{Text: "I have completed this pass. Next I will inspect the remaining call sites."},
943 },
944 }
945 agent := core.NewAgent(client, "test-model", "", nil)
946 threshold := 70
947 interactive := NewInteractive(InteractiveConfig{
948 Agent: agent,
949 Provider: "anthropic",
950 Model: "claude-sonnet-4-5-20250929",
951 AutoCompactThreshold: &threshold,
952 PersistCompactHandoff: func(state json.RawMessage) error {
953 select {
954 case persistedHandoff <- string(state):
955 default:
956 }
957 return nil
958 },
959 })
960 interactive.runCtx = context.Background()
961
962 interactive.startTurn(context.Background(), "initial request")
963 select {
964 case <-client.compactionStarted:
965 case <-time.After(2 * time.Second):
966 t.Fatal("threshold auto-compaction did not start")
967 }
968 close(client.releaseCompaction)
969
970 select {
971 case state := <-persistedHandoff:
972 want := `{"version":1,"reason":"status_rescue","rescue_attempts":1}`
973 if state != want {
974 t.Fatalf("persisted handoff = %q, want %q", state, want)
975 }
976 case <-time.After(2 * time.Second):
977 t.Fatal("status rescue handoff was not persisted")
978 }
979
980 select {
981 case req := <-client.followUpRequest:
982 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
983 t.Fatalf("status rescue request contains auto-continue prompt %d times, want 1: %#v", got, req.Messages)
984 }
985 case <-time.After(2 * time.Second):
986 t.Fatal("status rescue continuation did not start")
987 }
988 waitInteractiveIdle(t, interactive)
989
990 client.mu.Lock()
991 calls := client.calls
992 client.mu.Unlock()
993 if calls != 3 {
994 t.Fatalf("provider called %d times, want initial request, compaction, and one status rescue", calls)
995 }
996}
997
998func TestThresholdAutoCompactionBoundsSecondStatusRescue(t *testing.T) {
999 client := &thresholdAutoCompactClient{
1000 compactionStarted: make(chan struct{}),
1001 releaseCompaction: make(chan struct{}),
1002 followUpRequest: make(chan provider.Request, 2),
1003 firstContent: []provider.Content{
1004 provider.TextBlock{Text: "Next I will inspect the remaining call sites."},
1005 },
1006 followUpContents: []string{
1007 "Then I will run the targeted tests.",
1008 "done",
1009 },
1010 }
1011 agent := core.NewAgent(client, "test-model", "", nil)
1012 threshold := 70
1013 interactive := NewInteractive(InteractiveConfig{
1014 Agent: agent,
1015 Provider: "anthropic",
1016 Model: "claude-sonnet-4-5-20250929",
1017 AutoCompactThreshold: &threshold,
1018 })
1019 interactive.runCtx = context.Background()
1020
1021 interactive.startTurn(context.Background(), "initial request")
1022 select {
1023 case <-client.compactionStarted:
1024 case <-time.After(2 * time.Second):
1025 t.Fatal("threshold auto-compaction did not start")
1026 }
1027 close(client.releaseCompaction)
1028
1029 for attempt := 1; attempt <= 2; attempt++ {
1030 select {
1031 case req := <-client.followUpRequest:
1032 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != attempt {
1033 t.Fatalf("status rescue %d contains auto-continue prompt %d times, want %d: %#v", attempt, got, attempt, req.Messages)
1034 }
1035 case <-time.After(2 * time.Second):
1036 t.Fatalf("bounded status rescue %d did not start", attempt)
1037 }
1038 }
1039 waitInteractiveIdle(t, interactive)
1040
1041 client.mu.Lock()
1042 calls, followUps := client.calls, client.followUps
1043 client.mu.Unlock()
1044 if calls != 4 || followUps != 2 {
1045 t.Fatalf("provider calls/follow-ups = %d/%d, want 4/2", calls, followUps)
1046 }
1047}
1048
1049func TestThresholdAutoCompactionStopsAfterStatusRescueCompletion(t *testing.T) {
1050 client := &thresholdAutoCompactClient{
1051 compactionStarted: make(chan struct{}),
1052 releaseCompaction: make(chan struct{}),
1053 followUpRequest: make(chan provider.Request, 2),
1054 firstContent: []provider.Content{
1055 provider.TextBlock{Text: "Next I will inspect the remaining call sites."},
1056 },
1057 followUpContents: []string{"Then I will run the targeted tests.", "All required work is complete."},
1058 }
1059 agent := core.NewAgent(client, "test-model", "", nil)
1060 threshold := 70
1061 interactive := NewInteractive(InteractiveConfig{
1062 Agent: agent,
1063 Provider: "anthropic",
1064 Model: "claude-sonnet-4-5-20250929",
1065 AutoCompactThreshold: &threshold,
1066 })
1067 interactive.runCtx = context.Background()
1068
1069 interactive.startTurn(context.Background(), "initial request")
1070 select {
1071 case <-client.compactionStarted:
1072 case <-time.After(2 * time.Second):
1073 t.Fatal("threshold auto-compaction did not start")
1074 }
1075 close(client.releaseCompaction)
1076 for attempt := 1; attempt <= 2; attempt++ {
1077 select {
1078 case <-client.followUpRequest:
1079 case <-time.After(2 * time.Second):
1080 t.Fatalf("status rescue %d did not start", attempt)
1081 }
1082 }
1083 waitInteractiveIdle(t, interactive)
1084
1085 client.mu.Lock()
1086 calls := client.calls
1087 client.mu.Unlock()
1088 if calls != 4 {
1089 t.Fatalf("provider called %d times after completion, want 4", calls)
1090 }
1091 select {
1092 case <-client.followUpRequest:
1093 t.Fatal("completion after bounded rescue started an extra request")
1094 default:
1095 }
1096}
1097
1098func TestThresholdStatusRescueDefersToNewerUserPrompt(t *testing.T) {
1099 client := &thresholdAutoCompactClient{
1100 compactionStarted: make(chan struct{}),
1101 releaseCompaction: make(chan struct{}),
1102 followUpRequest: make(chan provider.Request, 1),
1103 firstContent: []provider.Content{
1104 provider.TextBlock{Text: "Next I will inspect the remaining call sites."},
1105 },
1106 }
1107 agent := core.NewAgent(client, "test-model", "", nil)
1108 threshold := 70
1109 interactive := NewInteractive(InteractiveConfig{
1110 Agent: agent,
1111 Provider: "anthropic",
1112 Model: "claude-sonnet-4-5-20250929",
1113 AutoCompactThreshold: &threshold,
1114 })
1115 interactive.runCtx = context.Background()
1116
1117 interactive.startTurn(context.Background(), "initial request")
1118 select {
1119 case <-client.compactionStarted:
1120 case <-time.After(2 * time.Second):
1121 t.Fatal("threshold auto-compaction did not start")
1122 }
1123 interactive.SubmitOrQueue("newer user request", nil)
1124 close(client.releaseCompaction)
1125
1126 select {
1127 case req := <-client.followUpRequest:
1128 if got := requestUserTextCount(req, "newer user request"); got != 1 {
1129 t.Fatalf("newer request count = %d, want 1: %#v", got, req.Messages)
1130 }
1131 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 0 {
1132 t.Fatalf("superseded status rescue count = %d, want 0: %#v", got, req.Messages)
1133 }
1134 case <-time.After(2 * time.Second):
1135 t.Fatal("newer user request did not run")
1136 }
1137 waitInteractiveIdle(t, interactive)
1138}
1139
1140func TestThresholdStatusRescueQueuedUserCancelsFollowUpHandoff(t *testing.T) {
1141 releaseFollowUp := make(chan struct{})
1142 client := &thresholdAutoCompactClient{
1143 compactionStarted: make(chan struct{}),
1144 releaseCompaction: make(chan struct{}),
1145 followUpRequest: make(chan provider.Request, 3),
1146 firstContent: []provider.Content{
1147 provider.TextBlock{Text: "Next I will inspect the remaining call sites."},
1148 },
1149 followUpContents: []string{
1150 "Then I will run the targeted tests.",
1151 "Next I will inspect the user's newly requested work.",
1152 },
1153 blockFollowUpCall: 3,
1154 releaseFollowUp: releaseFollowUp,
1155 }
1156 agent := core.NewAgent(client, "test-model", "", nil)
1157 threshold := 70
1158 interactive := NewInteractive(InteractiveConfig{
1159 Agent: agent,
1160 Provider: "anthropic",
1161 Model: "claude-sonnet-4-5-20250929",
1162 AutoCompactThreshold: &threshold,
1163 })
1164 interactive.runCtx = context.Background()
1165
1166 interactive.startTurn(context.Background(), "initial request")
1167 select {
1168 case <-client.compactionStarted:
1169 case <-time.After(2 * time.Second):
1170 t.Fatal("threshold auto-compaction did not start")
1171 }
1172 close(client.releaseCompaction)
1173
1174 select {
1175 case req := <-client.followUpRequest:
1176 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
1177 t.Fatalf("first status rescue contains auto-continue prompt %d times, want 1: %#v", got, req.Messages)
1178 }
1179 case <-time.After(2 * time.Second):
1180 t.Fatal("first status rescue did not start")
1181 }
1182
1183 interactive.SubmitOrQueue("newer user request", nil)
1184 close(releaseFollowUp)
1185
1186 select {
1187 case req := <-client.followUpRequest:
1188 if got := requestUserTextCount(req, "newer user request"); got != 1 {
1189 t.Fatalf("newer request count = %d, want 1: %#v", got, req.Messages)
1190 }
1191 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
1192 t.Fatalf("superseded follow-up rescue count = %d, want 1 existing prompt: %#v", got, req.Messages)
1193 }
1194 case <-time.After(2 * time.Second):
1195 t.Fatal("newer user request did not run after the status rescue")
1196 }
1197 waitInteractiveIdle(t, interactive)
1198
1199 client.mu.Lock()
1200 calls := client.calls
1201 client.mu.Unlock()
1202 if calls != 4 {
1203 t.Fatalf("provider called %d times, want initial request, compaction, status rescue, and newer user request", calls)
1204 }
1205 select {
1206 case req := <-client.followUpRequest:
1207 t.Fatalf("superseded handoff started an extra continuation: %#v", req.Messages)
1208 default:
1209 }
1210}
1211
1212func TestThresholdAutoCompactionDoesNotRescueClearText(t *testing.T) {
1213 client := &thresholdAutoCompactClient{
1214 compactionStarted: make(chan struct{}),
1215 releaseCompaction: make(chan struct{}),
1216 followUpRequest: make(chan provider.Request, 1),
1217 firstContent: []provider.Content{
1218 provider.TextBlock{Text: "All required work is complete."},
1219 },
1220 }
1221 agent := core.NewAgent(client, "test-model", "", nil)
1222 threshold := 70
1223 interactive := NewInteractive(InteractiveConfig{
1224 Agent: agent,
1225 Provider: "anthropic",
1226 Model: "claude-sonnet-4-5-20250929",
1227 AutoCompactThreshold: &threshold,
1228 })
1229 interactive.runCtx = context.Background()
1230
1231 interactive.startTurn(context.Background(), "initial request")
1232 select {
1233 case <-client.compactionStarted:
1234 case <-time.After(2 * time.Second):
1235 t.Fatal("threshold auto-compaction did not start")
1236 }
1237 close(client.releaseCompaction)
1238 waitInteractiveIdle(t, interactive)
1239
1240 client.mu.Lock()
1241 calls := client.calls
1242 client.mu.Unlock()
1243 if calls != 2 {
1244 t.Fatalf("provider called %d times for clear text, want initial request and compaction", calls)
1245 }
1246 select {
1247 case req := <-client.followUpRequest:
1248 t.Fatalf("clear text unexpectedly started follow-up: %#v", req.Messages)
1249 default:
1250 }
1251}
1252
1253func TestThresholdAutoCompactionContinuesTruncatedText(t *testing.T) {
1254 client := &thresholdAutoCompactClient{
1255 compactionStarted: make(chan struct{}),
1256 releaseCompaction: make(chan struct{}),
1257 followUpRequest: make(chan provider.Request, 1),
1258 firstStop: provider.StopLength,
1259 firstContent: []provider.Content{provider.TextBlock{Text: "partial answer"}},
1260 }
1261 agent := core.NewAgent(client, "test-model", "", nil)
1262 threshold := 70
1263 interactive := NewInteractive(InteractiveConfig{
1264 Agent: agent,
1265 Provider: "anthropic",
1266 Model: "claude-sonnet-4-5-20250929",
1267 AutoCompactThreshold: &threshold,
1268 })
1269 interactive.runCtx = context.Background()
1270
1271 interactive.startTurn(context.Background(), "initial request")
1272 select {
1273 case <-client.compactionStarted:
1274 case <-time.After(2 * time.Second):
1275 t.Fatal("threshold auto-compaction did not start after truncated text")
1276 }
1277 close(client.releaseCompaction)
1278
1279 select {
1280 case req := <-client.followUpRequest:
1281 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
1282 t.Fatalf("truncated continuation request contains auto-continue prompt %d times, want 1: %#v", got, req.Messages)
1283 }
1284 case <-time.After(2 * time.Second):
1285 t.Fatal("threshold auto-compaction did not continue truncated text")
1286 }
1287}
1288
1289func TestThresholdAutoCompactionForcedContinuationPrecedesQueuedPrompt(t *testing.T) {
1290 client := &thresholdAutoCompactClient{
1291 compactionStarted: make(chan struct{}),
1292 releaseCompaction: make(chan struct{}),
1293 followUpRequest: make(chan provider.Request, 2),
1294 firstStop: provider.StopLength,
1295 firstContent: []provider.Content{provider.TextBlock{Text: "partial answer"}},
1296 }
1297 agent := core.NewAgent(client, "test-model", "", nil)
1298 threshold := 70
1299 interactive := NewInteractive(InteractiveConfig{
1300 Agent: agent,
1301 Provider: "anthropic",
1302 Model: "claude-sonnet-4-5-20250929",
1303 AutoCompactThreshold: &threshold,
1304 })
1305 interactive.runCtx = context.Background()
1306
1307 interactive.startTurn(context.Background(), "initial request")
1308 select {
1309 case <-client.compactionStarted:
1310 case <-time.After(2 * time.Second):
1311 t.Fatal("threshold auto-compaction did not start after truncated text")
1312 }
1313 interactive.SubmitOrQueue("queued follow-up", nil)
1314 close(client.releaseCompaction)
1315
1316 select {
1317 case req := <-client.followUpRequest:
1318 if got := requestUserTextCount(req, autoCompactContinuationPrompt); got != 1 {
1319 t.Fatalf("first request contains auto-continue prompt %d times, want 1: %#v", got, req.Messages)
1320 }
1321 if got := requestUserTextCount(req, "queued follow-up"); got != 0 {
1322 t.Fatalf("forced continuation request included queued prompt %d times, want 0: %#v", got, req.Messages)
1323 }
1324 case <-time.After(2 * time.Second):
1325 t.Fatal("threshold auto-compaction did not continue truncated text")
1326 }
1327
1328 select {
1329 case req := <-client.followUpRequest:
1330 if got := requestUserTextCount(req, "queued follow-up"); got != 1 {
1331 t.Fatalf("second request contains queued prompt %d times, want 1: %#v", got, req.Messages)
1332 }
1333 case <-time.After(2 * time.Second):
1334 t.Fatal("queued prompt did not run after forced continuation")
1335 }
1336
1337 deadline := time.NewTimer(2 * time.Second)
1338 defer deadline.Stop()
1339 poll := time.NewTicker(time.Millisecond)
1340 defer poll.Stop()
1341 for {
1342 interactive.mu.Lock()
1343 idle := !interactive.busy
1344 interactive.mu.Unlock()
1345 if idle {
1346 break
1347 }
1348 select {
1349 case <-deadline.C:
1350 t.Fatal("interactive did not settle after forced continuation and queued prompt")
1351 case <-poll.C:
1352 }
1353 }
1354
1355 client.mu.Lock()
1356 calls := client.calls
1357 followUps := client.followUps
1358 client.mu.Unlock()
1359 if calls != 4 {
1360 t.Fatalf("provider called %d times, want initial request, compaction, forced continuation, and queued prompt", calls)
1361 }
1362 if followUps != 2 {
1363 t.Fatalf("follow-up requests = %d, want forced continuation and queued prompt", followUps)
1364 }
1365}
1366
1367func TestPreTurnCompactionPreservesPromptImages(t *testing.T) {
1368 client := &compactQueueClient{
1369 compactionStarted: make(chan struct{}),
1370 releaseCompaction: make(chan struct{}),
1371 followUpRequest: make(chan provider.Request, 1),
1372 }
1373 agent := core.NewAgent(client, "test-model", "", nil)
1374 agent.SetMessages([]provider.Message{{
1375 Role: provider.RoleUser,
1376 Content: []provider.Content{provider.TextBlock{Text: "one"}},
1377 }})
1378 threshold := 70
1379 interactive := NewInteractive(InteractiveConfig{
1380 Agent: agent,
1381 Provider: "anthropic",
1382 Model: "claude-sonnet-4-5",
1383 AutoCompactThreshold: &threshold,
1384 })
1385 interactive.runCtx = context.Background()
1386 interactive.lastCtxInput = 150000
1387 image := provider.ImageBlock{MimeType: "image/png", Data: []byte("image-data")}
1388
1389 interactive.startTurnWithImages(context.Background(), "follow up", []provider.ImageBlock{image})
1390 select {
1391 case <-client.compactionStarted:
1392 case <-time.After(2 * time.Second):
1393 t.Fatal("pre-turn compaction did not start")
1394 }
1395 close(client.releaseCompaction)
1396
1397 select {
1398 case req := <-client.followUpRequest:
1399 if got := requestUserTextCount(req, "follow up"); got != 1 {
1400 t.Fatalf("follow-up request contains prompt %d times, want 1: %#v", got, req.Messages)
1401 }
1402 if got := requestImageCount(req, image); got != 1 {
1403 t.Fatalf("follow-up request contains image %d times, want 1: %#v", got, req.Messages)
1404 }
1405 case <-time.After(2 * time.Second):
1406 t.Fatal("pre-turn compaction did not start the pending image request")
1407 }
1408}
1409
1410type contextRecoveryClient struct {
1411 mu sync.Mutex
1412 calls int
1413 retried chan provider.Request
1414 overflowRetry bool
1415}
1416
1417func (c *contextRecoveryClient) Name() string { return "context-recovery-test" }
1418
1419func (c *contextRecoveryClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) {
1420 c.mu.Lock()
1421 c.calls++
1422 call := c.calls
1423 c.mu.Unlock()
1424
1425 out := make(chan provider.Event, 2)
1426 go func() {
1427 defer close(out)
1428 switch call {
1429 case 1:
1430 out <- provider.EventDone{
1431 Stop: provider.StopError,
1432 Err: errors.New("provider error: Your input exceeds the context window of this model. Please adjust your input and try again."),
1433 }
1434 case 2:
1435 out <- provider.EventTextDelta{Delta: "summary"}
1436 out <- provider.EventDone{Stop: provider.StopEnd}
1437 default:
1438 c.retried <- req
1439 if c.overflowRetry {
1440 out <- provider.EventDone{
1441 Stop: provider.StopError,
1442 Err: errors.New("context window exceeded after compaction"),
1443 }
1444 return
1445 }
1446 out <- provider.EventDone{
1447 Stop: provider.StopEnd,
1448 Message: provider.Message{
1449 Role: provider.RoleAssistant,
1450 Content: []provider.Content{provider.TextBlock{Text: "done"}},
1451 },
1452 }
1453 }
1454 }()
1455 return out, nil
1456}
1457
1458func TestContextWindowErrorCompactsAndRetriesPromptOnce(t *testing.T) {
1459 client := &contextRecoveryClient{retried: make(chan provider.Request, 1)}
1460 agent := core.NewAgent(client, "test-model", "", nil)
1461 agent.SetMessages([]provider.Message{
1462 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "one"}}},
1463 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "two"}}},
1464 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "three"}}},
1465 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},
1466 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "five"}}},
1467 })
1468 interactive := NewInteractive(InteractiveConfig{Agent: agent})
1469 interactive.runCtx = context.Background()
1470
1471 image := provider.ImageBlock{MimeType: "image/png", Data: []byte("image-data")}
1472 interactive.startTurnWithImages(context.Background(), "retry me", []provider.ImageBlock{image})
1473
1474 select {
1475 case req := <-client.retried:
1476 if got := requestUserTextCount(req, "retry me"); got != 1 {
1477 t.Fatalf("retried request contains original prompt %d times, want 1: %#v", got, req.Messages)
1478 }
1479 if got := requestImageCount(req, image); got != 1 {
1480 t.Fatalf("retried request contains original image %d times, want 1: %#v", got, req.Messages)
1481 }
1482 case <-time.After(2 * time.Second):
1483 t.Fatal("context-window error did not compact and retry the prompt")
1484 }
1485}
1486
1487func TestContextWindowRecoveryStopsAfterOneRetry(t *testing.T) {
1488 client := &contextRecoveryClient{
1489 retried: make(chan provider.Request, 2),
1490 overflowRetry: true,
1491 }
1492 agent := core.NewAgent(client, "test-model", "", nil)
1493 agent.SetMessages([]provider.Message{
1494 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "one"}}},
1495 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "two"}}},
1496 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "three"}}},
1497 {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}},
1498 {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "five"}}},
1499 })
1500 interactive := NewInteractive(InteractiveConfig{Agent: agent})
1501 interactive.runCtx = context.Background()
1502
1503 interactive.startTurn(context.Background(), "still too large")
1504 select {
1505 case <-client.retried:
1506 case <-time.After(2 * time.Second):
1507 t.Fatal("context-window recovery did not retry")
1508 }
1509
1510 deadline := time.NewTimer(2 * time.Second)
1511 defer deadline.Stop()
1512 poll := time.NewTicker(time.Millisecond)
1513 defer poll.Stop()
1514 for {
1515 interactive.mu.Lock()
1516 idleWithError := !interactive.busy && interactive.statusErr != ""
1517 interactive.mu.Unlock()
1518 if idleWithError {
1519 break
1520 }
1521 select {
1522 case <-deadline.C:
1523 t.Fatal("second context-window error did not settle")
1524 case <-poll.C:
1525 }
1526 }
1527
1528 client.mu.Lock()
1529 calls := client.calls
1530 client.mu.Unlock()
1531 if calls != 3 {
1532 t.Fatalf("provider called %d times, want initial request, compaction, and one retry", calls)
1533 }
1534}
1535
1536func requestContainsUserText(req provider.Request, want string) bool {
1537 return requestUserTextCount(req, want) > 0
1538}
1539
1540func requestUserTextCount(req provider.Request, want string) int {
1541 count := 0
1542 for _, message := range req.Messages {
1543 if message.Role != provider.RoleUser {
1544 continue
1545 }
1546 for _, content := range message.Content {
1547 if text, ok := content.(provider.TextBlock); ok && text.Text == want {
1548 count++
1549 }
1550 }
1551 }
1552 return count
1553}
1554
1555func requestImageCount(req provider.Request, want provider.ImageBlock) int {
1556 count := 0
1557 for _, message := range req.Messages {
1558 if message.Role != provider.RoleUser {
1559 continue
1560 }
1561 for _, content := range message.Content {
1562 if image, ok := content.(provider.ImageBlock); ok && image.MimeType == want.MimeType && string(image.Data) == string(want.Data) {
1563 count++
1564 }
1565 }
1566 }
1567 return count
1568}