Fix queued prompts after compaction
- commit
- 2c9dbfadd00c1c5db45441aa950e45a5469fd284
- author
- patriceckhart · · 2026-07-26 12:09 UTC
- parent
- 471dc8c
- signature
- unsigned
- refs
- tag: v0.3.11
2 files · +124 -4
A packages/agent/modes/compact_queue_test.go +108 -0
| @@ -0,0 +1,108 @@ | |||
| 1 | + | package modes | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "context" | |
| 5 | + | "sync" | |
| 6 | + | "testing" | |
| 7 | + | "time" | |
| 8 | + | ||
| 9 | + | "github.com/patriceckhart/zot/packages/core" | |
| 10 | + | "github.com/patriceckhart/zot/packages/provider" | |
| 11 | + | "github.com/patriceckhart/zot/packages/tui" | |
| 12 | + | ) | |
| 13 | + | ||
| 14 | + | type compactQueueClient struct { | |
| 15 | + | compactionStarted chan struct{} | |
| 16 | + | releaseCompaction chan struct{} | |
| 17 | + | followUpRequest chan provider.Request | |
| 18 | + | ||
| 19 | + | mu sync.Mutex | |
| 20 | + | calls int | |
| 21 | + | } | |
| 22 | + | ||
| 23 | + | func (c *compactQueueClient) Name() string { return "compact-queue-test" } | |
| 24 | + | ||
| 25 | + | func (c *compactQueueClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) { | |
| 26 | + | c.mu.Lock() | |
| 27 | + | c.calls++ | |
| 28 | + | call := c.calls | |
| 29 | + | c.mu.Unlock() | |
| 30 | + | ||
| 31 | + | out := make(chan provider.Event, 2) | |
| 32 | + | go func() { | |
| 33 | + | defer close(out) | |
| 34 | + | if call == 1 { | |
| 35 | + | close(c.compactionStarted) | |
| 36 | + | select { | |
| 37 | + | case <-ctx.Done(): | |
| 38 | + | out <- provider.EventDone{Stop: provider.StopAborted, Err: ctx.Err()} | |
| 39 | + | case <-c.releaseCompaction: | |
| 40 | + | out <- provider.EventTextDelta{Delta: "summary"} | |
| 41 | + | out <- provider.EventDone{Stop: provider.StopEnd} | |
| 42 | + | } | |
| 43 | + | return | |
| 44 | + | } | |
| 45 | + | ||
| 46 | + | c.followUpRequest <- req | |
| 47 | + | out <- provider.EventDone{ | |
| 48 | + | Stop: provider.StopEnd, | |
| 49 | + | Message: provider.Message{ | |
| 50 | + | Role: provider.RoleAssistant, | |
| 51 | + | Content: []provider.Content{provider.TextBlock{Text: "done"}}, | |
| 52 | + | }, | |
| 53 | + | } | |
| 54 | + | }() | |
| 55 | + | return out, nil | |
| 56 | + | } | |
| 57 | + | ||
| 58 | + | func TestPromptSubmittedDuringCompactionStartsFollowUpTurn(t *testing.T) { | |
| 59 | + | client := &compactQueueClient{ | |
| 60 | + | compactionStarted: make(chan struct{}), | |
| 61 | + | releaseCompaction: make(chan struct{}), | |
| 62 | + | followUpRequest: make(chan provider.Request, 1), | |
| 63 | + | } | |
| 64 | + | agent := core.NewAgent(client, "test-model", "", nil) | |
| 65 | + | agent.SetMessages([]provider.Message{ | |
| 66 | + | {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "one"}}}, | |
| 67 | + | {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "two"}}}, | |
| 68 | + | {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "three"}}}, | |
| 69 | + | {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}}, | |
| 70 | + | {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "five"}}}, | |
| 71 | + | }) | |
| 72 | + | interactive := NewInteractive(InteractiveConfig{Agent: agent}) | |
| 73 | + | interactive.runCtx = context.Background() | |
| 74 | + | ||
| 75 | + | interactive.runCompact(context.Background(), false) | |
| 76 | + | select { | |
| 77 | + | case <-client.compactionStarted: | |
| 78 | + | case <-time.After(2 * time.Second): | |
| 79 | + | t.Fatal("compaction did not start") | |
| 80 | + | } | |
| 81 | + | ||
| 82 | + | interactive.ed.SetValue("follow up") | |
| 83 | + | interactive.handleKey(context.Background(), tui.Key{Kind: tui.KeyEnter}) | |
| 84 | + | close(client.releaseCompaction) | |
| 85 | + | ||
| 86 | + | select { | |
| 87 | + | case req := <-client.followUpRequest: | |
| 88 | + | if !requestContainsUserText(req, "follow up") { | |
| 89 | + | t.Fatalf("follow-up request does not contain queued prompt: %#v", req.Messages) | |
| 90 | + | } | |
| 91 | + | case <-time.After(2 * time.Second): | |
| 92 | + | t.Fatal("queued prompt did not start a turn after compaction") | |
| 93 | + | } | |
| 94 | + | } | |
| 95 | + | ||
| 96 | + | func requestContainsUserText(req provider.Request, want string) bool { | |
| 97 | + | for _, message := range req.Messages { | |
| 98 | + | if message.Role != provider.RoleUser { | |
| 99 | + | continue | |
| 100 | + | } | |
| 101 | + | for _, content := range message.Content { | |
| 102 | + | if text, ok := content.(provider.TextBlock); ok && text.Text == want { | |
| 103 | + | return true | |
| 104 | + | } | |
| 105 | + | } | |
| 106 | + | } | |
| 107 | + | return false | |
| 108 | + | } | |
| @@ -0,0 +1,108 @@ | |||||
| 1 | + | package modes | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "context" | |||
| 5 | + | "sync" | |||
| 6 | + | "testing" | |||
| 7 | + | "time" | |||
| 8 | + | ||||
| 9 | + | "github.com/patriceckhart/zot/packages/core" | |||
| 10 | + | "github.com/patriceckhart/zot/packages/provider" | |||
| 11 | + | "github.com/patriceckhart/zot/packages/tui" | |||
| 12 | + | ) | |||
| 13 | + | ||||
| 14 | + | type compactQueueClient struct { | |||
| 15 | + | compactionStarted chan struct{} | |||
| 16 | + | releaseCompaction chan struct{} | |||
| 17 | + | followUpRequest chan provider.Request | |||
| 18 | + | ||||
| 19 | + | mu sync.Mutex | |||
| 20 | + | calls int | |||
| 21 | + | } | |||
| 22 | + | ||||
| 23 | + | func (c *compactQueueClient) Name() string { return "compact-queue-test" } | |||
| 24 | + | ||||
| 25 | + | func (c *compactQueueClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) { | |||
| 26 | + | c.mu.Lock() | |||
| 27 | + | c.calls++ | |||
| 28 | + | call := c.calls | |||
| 29 | + | c.mu.Unlock() | |||
| 30 | + | ||||
| 31 | + | out := make(chan provider.Event, 2) | |||
| 32 | + | go func() { | |||
| 33 | + | defer close(out) | |||
| 34 | + | if call == 1 { | |||
| 35 | + | close(c.compactionStarted) | |||
| 36 | + | select { | |||
| 37 | + | case <-ctx.Done(): | |||
| 38 | + | out <- provider.EventDone{Stop: provider.StopAborted, Err: ctx.Err()} | |||
| 39 | + | case <-c.releaseCompaction: | |||
| 40 | + | out <- provider.EventTextDelta{Delta: "summary"} | |||
| 41 | + | out <- provider.EventDone{Stop: provider.StopEnd} | |||
| 42 | + | } | |||
| 43 | + | return | |||
| 44 | + | } | |||
| 45 | + | ||||
| 46 | + | c.followUpRequest <- req | |||
| 47 | + | out <- provider.EventDone{ | |||
| 48 | + | Stop: provider.StopEnd, | |||
| 49 | + | Message: provider.Message{ | |||
| 50 | + | Role: provider.RoleAssistant, | |||
| 51 | + | Content: []provider.Content{provider.TextBlock{Text: "done"}}, | |||
| 52 | + | }, | |||
| 53 | + | } | |||
| 54 | + | }() | |||
| 55 | + | return out, nil | |||
| 56 | + | } | |||
| 57 | + | ||||
| 58 | + | func TestPromptSubmittedDuringCompactionStartsFollowUpTurn(t *testing.T) { | |||
| 59 | + | client := &compactQueueClient{ | |||
| 60 | + | compactionStarted: make(chan struct{}), | |||
| 61 | + | releaseCompaction: make(chan struct{}), | |||
| 62 | + | followUpRequest: make(chan provider.Request, 1), | |||
| 63 | + | } | |||
| 64 | + | agent := core.NewAgent(client, "test-model", "", nil) | |||
| 65 | + | agent.SetMessages([]provider.Message{ | |||
| 66 | + | {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "one"}}}, | |||
| 67 | + | {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "two"}}}, | |||
| 68 | + | {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "three"}}}, | |||
| 69 | + | {Role: provider.RoleAssistant, Content: []provider.Content{provider.TextBlock{Text: "four"}}}, | |||
| 70 | + | {Role: provider.RoleUser, Content: []provider.Content{provider.TextBlock{Text: "five"}}}, | |||
| 71 | + | }) | |||
| 72 | + | interactive := NewInteractive(InteractiveConfig{Agent: agent}) | |||
| 73 | + | interactive.runCtx = context.Background() | |||
| 74 | + | ||||
| 75 | + | interactive.runCompact(context.Background(), false) | |||
| 76 | + | select { | |||
| 77 | + | case <-client.compactionStarted: | |||
| 78 | + | case <-time.After(2 * time.Second): | |||
| 79 | + | t.Fatal("compaction did not start") | |||
| 80 | + | } | |||
| 81 | + | ||||
| 82 | + | interactive.ed.SetValue("follow up") | |||
| 83 | + | interactive.handleKey(context.Background(), tui.Key{Kind: tui.KeyEnter}) | |||
| 84 | + | close(client.releaseCompaction) | |||
| 85 | + | ||||
| 86 | + | select { | |||
| 87 | + | case req := <-client.followUpRequest: | |||
| 88 | + | if !requestContainsUserText(req, "follow up") { | |||
| 89 | + | t.Fatalf("follow-up request does not contain queued prompt: %#v", req.Messages) | |||
| 90 | + | } | |||
| 91 | + | case <-time.After(2 * time.Second): | |||
| 92 | + | t.Fatal("queued prompt did not start a turn after compaction") | |||
| 93 | + | } | |||
| 94 | + | } | |||
| 95 | + | ||||
| 96 | + | func requestContainsUserText(req provider.Request, want string) bool { | |||
| 97 | + | for _, message := range req.Messages { | |||
| 98 | + | if message.Role != provider.RoleUser { | |||
| 99 | + | continue | |||
| 100 | + | } | |||
| 101 | + | for _, content := range message.Content { | |||
| 102 | + | if text, ok := content.(provider.TextBlock); ok && text.Text == want { | |||
| 103 | + | return true | |||
| 104 | + | } | |||
| 105 | + | } | |||
| 106 | + | } | |||
| 107 | + | return false | |||
| 108 | + | } | |||
M packages/agent/modes/interactive.go +16 -4
| @@ -406,6 +406,11 @@ type Interactive struct { | |||
| 406 | 406 | // pre-turn fraction guard. Cleared by runCompact once shown. | |
| 407 | 407 | pendingPostCompactNote string | |
| 408 | 408 | ||
| 409 | + | // compacting is true for both manual and automatic compaction. Prompts | |
| 410 | + | // submitted while it is set stay in the host queue because Compact does | |
| 411 | + | // not run the agent loop that drains the agent-owned queue. | |
| 412 | + | compacting bool | |
| 413 | + | ||
| 409 | 414 | // autoCompacting is true while a model-triggered compaction is in | |
| 410 | 415 | // flight. Surfaced in the status bar so the user can tell a | |
| 411 | 416 | // condense pass from a regular assistant turn. | |
| @@ -2602,6 +2607,7 @@ func (i *Interactive) handleKey(ctx context.Context, k tui.Key) (done bool) { | |||
| 2602 | 2607 | // boundary instead of waiting for the whole run to finish. | |
| 2603 | 2608 | i.mu.Lock() | |
| 2604 | 2609 | busy := i.busy | |
| 2610 | + | compacting := i.compacting | |
| 2605 | 2611 | ag := i.agent | |
| 2606 | 2612 | i.mu.Unlock() | |
| 2607 | 2613 | if busy { | |
| @@ -2612,7 +2618,7 @@ func (i *Interactive) handleKey(ctx context.Context, k tui.Key) (done bool) { | |||
| 2612 | 2618 | i.invalidate() | |
| 2613 | 2619 | return false | |
| 2614 | 2620 | } | |
| 2615 | - | if ag != nil { | |
| 2621 | + | if ag != nil && !compacting { | |
| 2616 | 2622 | ag.QueueMessage(text) | |
| 2617 | 2623 | } else { | |
| 2618 | 2624 | i.mu.Lock() | |
| @@ -2963,10 +2969,13 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) { | |||
| 2963 | 2969 | return | |
| 2964 | 2970 | } | |
| 2965 | 2971 | if i.busy { | |
| 2966 | - | // Queue text only; images are dropped for queued prompts. | |
| 2972 | + | // Queue text only; images are dropped for queued prompts. Compaction | |
| 2973 | + | // uses the host queue because it has no active agent loop to drain | |
| 2974 | + | // Agent.QueueMessage entries. | |
| 2967 | 2975 | ag := i.agent | |
| 2976 | + | compacting := i.compacting | |
| 2968 | 2977 | i.mu.Unlock() | |
| 2969 | - | if ag != nil { | |
| 2978 | + | if ag != nil && !compacting { | |
| 2970 | 2979 | ag.QueueMessage(text) | |
| 2971 | 2980 | } else { | |
| 2972 | 2981 | i.mu.Lock() | |
| @@ -4143,10 +4152,11 @@ func (i *Interactive) runSlash(ctx context.Context, cmd string) (done bool) { | |||
| 4143 | 4152 | studyPrompt := buildStudyPrompt(strings.TrimSpace(strings.TrimPrefix(cmd, parts[0])), i.cfg.CWD) | |
| 4144 | 4153 | i.mu.Lock() | |
| 4145 | 4154 | busy := i.busy | |
| 4155 | + | compacting := i.compacting | |
| 4146 | 4156 | ag := i.agent | |
| 4147 | 4157 | i.mu.Unlock() | |
| 4148 | 4158 | if busy { | |
| 4149 | - | if ag != nil { | |
| 4159 | + | if ag != nil && !compacting { | |
| 4150 | 4160 | ag.QueueMessage(studyPrompt) | |
| 4151 | 4161 | } else { | |
| 4152 | 4162 | i.mu.Lock() | |
| @@ -5061,6 +5071,7 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||
| 5061 | 5071 | ctx, cancel := context.WithCancel(parent) | |
| 5062 | 5072 | i.mu.Lock() | |
| 5063 | 5073 | i.busy = true | |
| 5074 | + | i.compacting = true | |
| 5064 | 5075 | if auto { | |
| 5065 | 5076 | i.spin.StartFixed("condensing history") | |
| 5066 | 5077 | i.autoCompacting = true | |
| @@ -5085,6 +5096,7 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||
| 5085 | 5096 | _ = summary | |
| 5086 | 5097 | i.mu.Lock() | |
| 5087 | 5098 | i.busy = false | |
| 5099 | + | i.compacting = false | |
| 5088 | 5100 | i.resetStreamingStateLocked() | |
| 5089 | 5101 | i.cancelTurn = nil | |
| 5090 | 5102 | i.autoCompacting = false | |
| @@ -406,6 +406,11 @@ type Interactive struct { | |||||
| 406 | // pre-turn fraction guard. Cleared by runCompact once shown. | 406 | // pre-turn fraction guard. Cleared by runCompact once shown. | ||
| 407 | pendingPostCompactNote string | 407 | pendingPostCompactNote string | ||
| 408 | 408 | ||||
| 409 | + | // compacting is true for both manual and automatic compaction. Prompts | |||
| 410 | + | // submitted while it is set stay in the host queue because Compact does | |||
| 411 | + | // not run the agent loop that drains the agent-owned queue. | |||
| 412 | + | compacting bool | |||
| 413 | + | ||||
| 409 | // autoCompacting is true while a model-triggered compaction is in | 414 | // autoCompacting is true while a model-triggered compaction is in | ||
| 410 | // flight. Surfaced in the status bar so the user can tell a | 415 | // flight. Surfaced in the status bar so the user can tell a | ||
| 411 | // condense pass from a regular assistant turn. | 416 | // condense pass from a regular assistant turn. | ||
| @@ -2602,6 +2607,7 @@ func (i *Interactive) handleKey(ctx context.Context, k tui.Key) (done bool) { | |||||
| 2602 | // boundary instead of waiting for the whole run to finish. | 2607 | // boundary instead of waiting for the whole run to finish. | ||
| 2603 | i.mu.Lock() | 2608 | i.mu.Lock() | ||
| 2604 | busy := i.busy | 2609 | busy := i.busy | ||
| 2610 | + | compacting := i.compacting | |||
| 2605 | ag := i.agent | 2611 | ag := i.agent | ||
| 2606 | i.mu.Unlock() | 2612 | i.mu.Unlock() | ||
| 2607 | if busy { | 2613 | if busy { | ||
| @@ -2612,7 +2618,7 @@ func (i *Interactive) handleKey(ctx context.Context, k tui.Key) (done bool) { | |||||
| 2612 | i.invalidate() | 2618 | i.invalidate() | ||
| 2613 | return false | 2619 | return false | ||
| 2614 | } | 2620 | } | ||
| 2615 | - | if ag != nil { | 2621 | + | if ag != nil && !compacting { |
| 2616 | ag.QueueMessage(text) | 2622 | ag.QueueMessage(text) | ||
| 2617 | } else { | 2623 | } else { | ||
| 2618 | i.mu.Lock() | 2624 | i.mu.Lock() | ||
| @@ -2963,10 +2969,13 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) { | |||||
| 2963 | return | 2969 | return | ||
| 2964 | } | 2970 | } | ||
| 2965 | if i.busy { | 2971 | if i.busy { | ||
| 2966 | - | // Queue text only; images are dropped for queued prompts. | 2972 | + | // Queue text only; images are dropped for queued prompts. Compaction |
| 2973 | + | // uses the host queue because it has no active agent loop to drain | |||
| 2974 | + | // Agent.QueueMessage entries. | |||
| 2967 | ag := i.agent | 2975 | ag := i.agent | ||
| 2976 | + | compacting := i.compacting | |||
| 2968 | i.mu.Unlock() | 2977 | i.mu.Unlock() | ||
| 2969 | - | if ag != nil { | 2978 | + | if ag != nil && !compacting { |
| 2970 | ag.QueueMessage(text) | 2979 | ag.QueueMessage(text) | ||
| 2971 | } else { | 2980 | } else { | ||
| 2972 | i.mu.Lock() | 2981 | i.mu.Lock() | ||
| @@ -4143,10 +4152,11 @@ func (i *Interactive) runSlash(ctx context.Context, cmd string) (done bool) { | |||||
| 4143 | studyPrompt := buildStudyPrompt(strings.TrimSpace(strings.TrimPrefix(cmd, parts[0])), i.cfg.CWD) | 4152 | studyPrompt := buildStudyPrompt(strings.TrimSpace(strings.TrimPrefix(cmd, parts[0])), i.cfg.CWD) | ||
| 4144 | i.mu.Lock() | 4153 | i.mu.Lock() | ||
| 4145 | busy := i.busy | 4154 | busy := i.busy | ||
| 4155 | + | compacting := i.compacting | |||
| 4146 | ag := i.agent | 4156 | ag := i.agent | ||
| 4147 | i.mu.Unlock() | 4157 | i.mu.Unlock() | ||
| 4148 | if busy { | 4158 | if busy { | ||
| 4149 | - | if ag != nil { | 4159 | + | if ag != nil && !compacting { |
| 4150 | ag.QueueMessage(studyPrompt) | 4160 | ag.QueueMessage(studyPrompt) | ||
| 4151 | } else { | 4161 | } else { | ||
| 4152 | i.mu.Lock() | 4162 | i.mu.Lock() | ||
| @@ -5061,6 +5071,7 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||||
| 5061 | ctx, cancel := context.WithCancel(parent) | 5071 | ctx, cancel := context.WithCancel(parent) | ||
| 5062 | i.mu.Lock() | 5072 | i.mu.Lock() | ||
| 5063 | i.busy = true | 5073 | i.busy = true | ||
| 5074 | + | i.compacting = true | |||
| 5064 | if auto { | 5075 | if auto { | ||
| 5065 | i.spin.StartFixed("condensing history") | 5076 | i.spin.StartFixed("condensing history") | ||
| 5066 | i.autoCompacting = true | 5077 | i.autoCompacting = true | ||
| @@ -5085,6 +5096,7 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||||
| 5085 | _ = summary | 5096 | _ = summary | ||
| 5086 | i.mu.Lock() | 5097 | i.mu.Lock() | ||
| 5087 | i.busy = false | 5098 | i.busy = false | ||
| 5099 | + | i.compacting = false | |||
| 5088 | i.resetStreamingStateLocked() | 5100 | i.resetStreamingStateLocked() | ||
| 5089 | i.cancelTurn = nil | 5101 | i.cancelTurn = nil | ||
| 5090 | i.autoCompacting = false | 5102 | i.autoCompacting = false | ||