Merge pull request #23 from bnema/fix/auto-subagent-delivery
fix(agent): deliver queued auto-subagent updates
2 files · +134 -34
M packages/agent/modes/auto_subagents_test.go +85 -0
| @@ -4,11 +4,13 @@ import ( | |||
| 4 | 4 | "context" | |
| 5 | 5 | "errors" | |
| 6 | 6 | "strings" | |
| 7 | + | "sync" | |
| 7 | 8 | "testing" | |
| 8 | 9 | "time" | |
| 9 | 10 | ||
| 10 | 11 | "github.com/patriceckhart/zot/packages/agent/subagents" | |
| 11 | 12 | "github.com/patriceckhart/zot/packages/core" | |
| 13 | + | "github.com/patriceckhart/zot/packages/provider" | |
| 12 | 14 | ) | |
| 13 | 15 | ||
| 14 | 16 | func TestAutoSubagentsSystemPromptTogglesProfileManifestWithDelegationGuidance(t *testing.T) { | |
| @@ -151,6 +153,89 @@ func TestCompleteSupervisorWatchReportsTurnOutcomeOnce(t *testing.T) { | |||
| 151 | 153 | } | |
| 152 | 154 | } | |
| 153 | 155 | ||
| 156 | + | type autoSubagentsQueueClient struct { | |
| 157 | + | requests chan provider.Request | |
| 158 | + | } | |
| 159 | + | ||
| 160 | + | func (c *autoSubagentsQueueClient) Name() string { return "auto-subagents-queue-test" } | |
| 161 | + | ||
| 162 | + | func (c *autoSubagentsQueueClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) { | |
| 163 | + | c.requests <- req | |
| 164 | + | out := make(chan provider.Event, 1) | |
| 165 | + | go func() { | |
| 166 | + | defer close(out) | |
| 167 | + | out <- provider.EventDone{ | |
| 168 | + | Stop: provider.StopEnd, | |
| 169 | + | Message: provider.Message{ | |
| 170 | + | Role: provider.RoleAssistant, | |
| 171 | + | Content: []provider.Content{provider.TextBlock{Text: "done"}}, | |
| 172 | + | }, | |
| 173 | + | } | |
| 174 | + | }() | |
| 175 | + | return out, nil | |
| 176 | + | } | |
| 177 | + | ||
| 178 | + | func TestAutoSubagentsUpdateQueuedAtTurnEndStartsFollowUp(t *testing.T) { | |
| 179 | + | client := &autoSubagentsQueueClient{requests: make(chan provider.Request, 2)} | |
| 180 | + | agent := core.NewAgent(client, "test-model", "", nil) | |
| 181 | + | interactive := NewInteractive(InteractiveConfig{Agent: agent}) | |
| 182 | + | interactive.runCtx = context.Background() | |
| 183 | + | ||
| 184 | + | var once sync.Once | |
| 185 | + | agent.OnEvent = func(ev core.AgentEvent) { | |
| 186 | + | if _, ok := ev.(core.EvDone); !ok { | |
| 187 | + | return | |
| 188 | + | } | |
| 189 | + | once.Do(func() { | |
| 190 | + | interactive.flushSupervisorSummary([]*subagentWatchEntry{{ | |
| 191 | + | agent: &subagents.Agent{}, | |
| 192 | + | task: "smoke test", | |
| 193 | + | }}) | |
| 194 | + | }) | |
| 195 | + | } | |
| 196 | + | ||
| 197 | + | interactive.startTurn(context.Background(), "initial") | |
| 198 | + | select { | |
| 199 | + | case <-client.requests: | |
| 200 | + | case <-time.After(2 * time.Second): | |
| 201 | + | t.Fatal("initial request did not start") | |
| 202 | + | } | |
| 203 | + | ||
| 204 | + | var followUp provider.Request | |
| 205 | + | select { | |
| 206 | + | case followUp = <-client.requests: | |
| 207 | + | case <-time.After(2 * time.Second): | |
| 208 | + | t.Fatal("queued auto-subagents update did not start a follow-up request") | |
| 209 | + | } | |
| 210 | + | foundUpdate := false | |
| 211 | + | for _, message := range followUp.Messages { | |
| 212 | + | for _, content := range message.Content { | |
| 213 | + | if text, ok := content.(provider.TextBlock); ok && strings.Contains(text.Text, "[auto-subagents update]") { | |
| 214 | + | foundUpdate = true | |
| 215 | + | } | |
| 216 | + | } | |
| 217 | + | } | |
| 218 | + | if !foundUpdate { | |
| 219 | + | t.Fatalf("follow-up request did not contain the auto-subagents update: %#v", followUp.Messages) | |
| 220 | + | } | |
| 221 | + | ||
| 222 | + | deadline := time.NewTimer(2 * time.Second) | |
| 223 | + | defer deadline.Stop() | |
| 224 | + | for { | |
| 225 | + | interactive.mu.Lock() | |
| 226 | + | busy := interactive.busy | |
| 227 | + | interactive.mu.Unlock() | |
| 228 | + | if !busy && agent.QueuedMessageCount() == 0 { | |
| 229 | + | return | |
| 230 | + | } | |
| 231 | + | select { | |
| 232 | + | case <-deadline.C: | |
| 233 | + | t.Fatal("interactive turn remained busy or retained a queued update") | |
| 234 | + | case <-time.After(time.Millisecond): | |
| 235 | + | } | |
| 236 | + | } | |
| 237 | + | } | |
| 238 | + | ||
| 154 | 239 | func newQueuedAutoSubagentsInteractive() *Interactive { | |
| 155 | 240 | return &Interactive{ | |
| 156 | 241 | agent: &core.Agent{}, | |
| @@ -4,11 +4,13 @@ import ( | |||||
| 4 | "context" | 4 | "context" | ||
| 5 | "errors" | 5 | "errors" | ||
| 6 | "strings" | 6 | "strings" | ||
| 7 | + | "sync" | |||
| 7 | "testing" | 8 | "testing" | ||
| 8 | "time" | 9 | "time" | ||
| 9 | 10 | ||||
| 10 | "github.com/patriceckhart/zot/packages/agent/subagents" | 11 | "github.com/patriceckhart/zot/packages/agent/subagents" | ||
| 11 | "github.com/patriceckhart/zot/packages/core" | 12 | "github.com/patriceckhart/zot/packages/core" | ||
| 13 | + | "github.com/patriceckhart/zot/packages/provider" | |||
| 12 | ) | 14 | ) | ||
| 13 | 15 | ||||
| 14 | func TestAutoSubagentsSystemPromptTogglesProfileManifestWithDelegationGuidance(t *testing.T) { | 16 | func TestAutoSubagentsSystemPromptTogglesProfileManifestWithDelegationGuidance(t *testing.T) { | ||
| @@ -151,6 +153,89 @@ func TestCompleteSupervisorWatchReportsTurnOutcomeOnce(t *testing.T) { | |||||
| 151 | } | 153 | } | ||
| 152 | } | 154 | } | ||
| 153 | 155 | ||||
| 156 | + | type autoSubagentsQueueClient struct { | |||
| 157 | + | requests chan provider.Request | |||
| 158 | + | } | |||
| 159 | + | ||||
| 160 | + | func (c *autoSubagentsQueueClient) Name() string { return "auto-subagents-queue-test" } | |||
| 161 | + | ||||
| 162 | + | func (c *autoSubagentsQueueClient) Stream(_ context.Context, req provider.Request) (<-chan provider.Event, error) { | |||
| 163 | + | c.requests <- req | |||
| 164 | + | out := make(chan provider.Event, 1) | |||
| 165 | + | go func() { | |||
| 166 | + | defer close(out) | |||
| 167 | + | out <- provider.EventDone{ | |||
| 168 | + | Stop: provider.StopEnd, | |||
| 169 | + | Message: provider.Message{ | |||
| 170 | + | Role: provider.RoleAssistant, | |||
| 171 | + | Content: []provider.Content{provider.TextBlock{Text: "done"}}, | |||
| 172 | + | }, | |||
| 173 | + | } | |||
| 174 | + | }() | |||
| 175 | + | return out, nil | |||
| 176 | + | } | |||
| 177 | + | ||||
| 178 | + | func TestAutoSubagentsUpdateQueuedAtTurnEndStartsFollowUp(t *testing.T) { | |||
| 179 | + | client := &autoSubagentsQueueClient{requests: make(chan provider.Request, 2)} | |||
| 180 | + | agent := core.NewAgent(client, "test-model", "", nil) | |||
| 181 | + | interactive := NewInteractive(InteractiveConfig{Agent: agent}) | |||
| 182 | + | interactive.runCtx = context.Background() | |||
| 183 | + | ||||
| 184 | + | var once sync.Once | |||
| 185 | + | agent.OnEvent = func(ev core.AgentEvent) { | |||
| 186 | + | if _, ok := ev.(core.EvDone); !ok { | |||
| 187 | + | return | |||
| 188 | + | } | |||
| 189 | + | once.Do(func() { | |||
| 190 | + | interactive.flushSupervisorSummary([]*subagentWatchEntry{{ | |||
| 191 | + | agent: &subagents.Agent{}, | |||
| 192 | + | task: "smoke test", | |||
| 193 | + | }}) | |||
| 194 | + | }) | |||
| 195 | + | } | |||
| 196 | + | ||||
| 197 | + | interactive.startTurn(context.Background(), "initial") | |||
| 198 | + | select { | |||
| 199 | + | case <-client.requests: | |||
| 200 | + | case <-time.After(2 * time.Second): | |||
| 201 | + | t.Fatal("initial request did not start") | |||
| 202 | + | } | |||
| 203 | + | ||||
| 204 | + | var followUp provider.Request | |||
| 205 | + | select { | |||
| 206 | + | case followUp = <-client.requests: | |||
| 207 | + | case <-time.After(2 * time.Second): | |||
| 208 | + | t.Fatal("queued auto-subagents update did not start a follow-up request") | |||
| 209 | + | } | |||
| 210 | + | foundUpdate := false | |||
| 211 | + | for _, message := range followUp.Messages { | |||
| 212 | + | for _, content := range message.Content { | |||
| 213 | + | if text, ok := content.(provider.TextBlock); ok && strings.Contains(text.Text, "[auto-subagents update]") { | |||
| 214 | + | foundUpdate = true | |||
| 215 | + | } | |||
| 216 | + | } | |||
| 217 | + | } | |||
| 218 | + | if !foundUpdate { | |||
| 219 | + | t.Fatalf("follow-up request did not contain the auto-subagents update: %#v", followUp.Messages) | |||
| 220 | + | } | |||
| 221 | + | ||||
| 222 | + | deadline := time.NewTimer(2 * time.Second) | |||
| 223 | + | defer deadline.Stop() | |||
| 224 | + | for { | |||
| 225 | + | interactive.mu.Lock() | |||
| 226 | + | busy := interactive.busy | |||
| 227 | + | interactive.mu.Unlock() | |||
| 228 | + | if !busy && agent.QueuedMessageCount() == 0 { | |||
| 229 | + | return | |||
| 230 | + | } | |||
| 231 | + | select { | |||
| 232 | + | case <-deadline.C: | |||
| 233 | + | t.Fatal("interactive turn remained busy or retained a queued update") | |||
| 234 | + | case <-time.After(time.Millisecond): | |||
| 235 | + | } | |||
| 236 | + | } | |||
| 237 | + | } | |||
| 238 | + | ||||
| 154 | func newQueuedAutoSubagentsInteractive() *Interactive { | 239 | func newQueuedAutoSubagentsInteractive() *Interactive { | ||
| 155 | return &Interactive{ | 240 | return &Interactive{ | ||
| 156 | agent: &core.Agent{}, | 241 | agent: &core.Agent{}, | ||
M packages/agent/modes/interactive.go +49 -34
| @@ -3595,19 +3595,17 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) { | |||
| 3595 | 3595 | i.maybeStartSessionTitle(i.runCtx, text) | |
| 3596 | 3596 | i.mu.Lock() | |
| 3597 | 3597 | if i.busy { | |
| 3598 | - | // Queue text only; images are dropped for queued prompts. Compaction | |
| 3599 | - | // uses the host queue because it has no active agent loop to drain | |
| 3600 | - | // Agent.QueueMessage entries. | |
| 3601 | - | ag := i.agent | |
| 3602 | - | compacting := i.compacting | |
| 3603 | - | i.mu.Unlock() | |
| 3604 | - | if ag != nil && !compacting { | |
| 3605 | - | ag.QueueMessage(text) | |
| 3598 | + | // Queue text only; images are dropped for queued prompts. Keep the | |
| 3599 | + | // interactive mutex held through Agent.QueueMessage so turn teardown | |
| 3600 | + | // cannot inspect the agent queue between the busy check and enqueue. | |
| 3601 | + | // Compaction uses the host queue because it has no active agent loop | |
| 3602 | + | // to drain Agent.QueueMessage entries. | |
| 3603 | + | if i.agent != nil && !i.compacting { | |
| 3604 | + | i.agent.QueueMessage(text) | |
| 3606 | 3605 | } else { | |
| 3607 | - | i.mu.Lock() | |
| 3608 | 3606 | i.queued = append(i.queued, text) | |
| 3609 | - | i.mu.Unlock() | |
| 3610 | 3607 | } | |
| 3608 | + | i.mu.Unlock() | |
| 3611 | 3609 | i.invalidate() | |
| 3612 | 3610 | return | |
| 3613 | 3611 | } | |
| @@ -5675,22 +5673,20 @@ func (i *Interactive) submitOrQueuePrompt(ctx context.Context, prompt string) { | |||
| 5675 | 5673 | i.invalidate() | |
| 5676 | 5674 | return | |
| 5677 | 5675 | } | |
| 5678 | - | busy := i.busy | |
| 5679 | - | compacting := i.compacting | |
| 5680 | - | ag := i.agent | |
| 5681 | - | i.mu.Unlock() | |
| 5682 | - | ||
| 5683 | - | if busy { | |
| 5684 | - | if !compacting { | |
| 5685 | - | ag.QueueMessage(prompt) | |
| 5686 | - | } else { | |
| 5687 | - | i.mu.Lock() | |
| 5676 | + | if i.busy { | |
| 5677 | + | // Keep the mutex held while enqueueing so the turn-completion | |
| 5678 | + | // goroutine cannot publish idle and inspect an empty queue between | |
| 5679 | + | // this check and the append. | |
| 5680 | + | if i.compacting { | |
| 5688 | 5681 | i.queued = append(i.queued, prompt) | |
| 5689 | - | i.mu.Unlock() | |
| 5682 | + | } else { | |
| 5683 | + | i.agent.QueueMessage(prompt) | |
| 5690 | 5684 | } | |
| 5685 | + | i.mu.Unlock() | |
| 5691 | 5686 | i.invalidate() | |
| 5692 | 5687 | return | |
| 5693 | 5688 | } | |
| 5689 | + | i.mu.Unlock() | |
| 5694 | 5690 | i.startTurn(ctx, prompt) | |
| 5695 | 5691 | } | |
| 5696 | 5692 | ||
| @@ -6121,11 +6117,11 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||
| 6121 | 6117 | summary, err := i.agent.Compact(ctx, keepTail, sink) | |
| 6122 | 6118 | _ = summary | |
| 6123 | 6119 | i.mu.Lock() | |
| 6124 | - | i.busy = false | |
| 6125 | - | i.compacting = false | |
| 6120 | + | // Keep busy/compacting asserted while cleanup and queue selection run. | |
| 6121 | + | // Completion updates can arrive in this window; clearing busy before | |
| 6122 | + | // inspecting the queues lets them race a new turn or be stranded. | |
| 6126 | 6123 | i.resetStreamingStateLocked() | |
| 6127 | 6124 | i.cancelTurn = nil | |
| 6128 | - | i.autoCompacting = false | |
| 6129 | 6125 | ||
| 6130 | 6126 | // Drain pending work after the transcript is clean. Overflow recovery | |
| 6131 | 6127 | // continues the user message already in the transcript. Pre-turn | |
| @@ -6203,12 +6199,13 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||
| 6203 | 6199 | continueAutomatically = true | |
| 6204 | 6200 | } | |
| 6205 | 6201 | } | |
| 6206 | - | if hasNext || continueExisting || continueAutomatically { | |
| 6207 | - | // Keep the host busy while handing the next turn back to the | |
| 6208 | - | // agent. Prompts submitted in this hand-off window then enter | |
| 6209 | - | // the agent queue instead of racing a second startTurn call. | |
| 6210 | - | i.busy = true | |
| 6211 | - | } | |
| 6202 | + | // Keep the host busy until the hand-off decision is committed under | |
| 6203 | + | // the mutex. A completion update arriving after the compaction result | |
| 6204 | + | // but before this assignment is then queued for the selected follow-up | |
| 6205 | + | // instead of starting a competing turn. | |
| 6206 | + | i.busy = hasNext || continueExisting || continueAutomatically | |
| 6207 | + | i.compacting = false | |
| 6208 | + | i.autoCompacting = false | |
| 6212 | 6209 | i.mu.Unlock() | |
| 6213 | 6210 | i.invalidate() | |
| 6214 | 6211 | ||
| @@ -6387,9 +6384,19 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||
| 6387 | 6384 | i.mu.Lock() | |
| 6388 | 6385 | needsPreCompact := !overflowRecoveryAttempted && !i.autoCompacting && i.shouldAutoCompactLocked() | |
| 6389 | 6386 | if needsPreCompact { | |
| 6390 | - | i.pendingCompactPrompt = prompt | |
| 6391 | - | i.pendingCompactImages = append([]provider.ImageBlock(nil), images...) | |
| 6392 | - | i.hasPendingCompactPrompt = true | |
| 6387 | + | // Reserve the compaction hand-off before releasing i.mu. A completion | |
| 6388 | + | // update arriving in the small gap before runCompact acquires the | |
| 6389 | + | // mutex must join the host queue rather than starting a competing turn. | |
| 6390 | + | i.busy = true | |
| 6391 | + | i.compacting = true | |
| 6392 | + | i.autoCompacting = true | |
| 6393 | + | if continueExisting { | |
| 6394 | + | i.continueAfterCompact = true | |
| 6395 | + | } else { | |
| 6396 | + | i.pendingCompactPrompt = prompt | |
| 6397 | + | i.pendingCompactImages = append([]provider.ImageBlock(nil), images...) | |
| 6398 | + | i.hasPendingCompactPrompt = true | |
| 6399 | + | } | |
| 6393 | 6400 | i.statusErr = "" | |
| 6394 | 6401 | i.extNotes = append(i.extNotes, autoCompactNoteLine(i.cfg.Theme, "context near limit — condensing history before sending...")) | |
| 6395 | 6402 | i.pendingPostCompactNote = "context auto-compacted; sending your last message" | |
| @@ -6463,7 +6470,11 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||
| 6463 | 6470 | err = i.agent.Prompt(ctx, prompt, images, sink) | |
| 6464 | 6471 | } | |
| 6465 | 6472 | i.mu.Lock() | |
| 6466 | - | i.busy = false | |
| 6473 | + | // Keep busy asserted through final cleanup and queue selection. A | |
| 6474 | + | // watcher may enqueue a completion summary concurrently with the | |
| 6475 | + | // provider's final event; publishing idle before inspecting queues can | |
| 6476 | + | // strand that summary in the core agent queue or start an overlapping | |
| 6477 | + | // turn. | |
| 6467 | 6478 | // Don't touch streamPending / streamFlushPending here — the | |
| 6468 | 6479 | // pacer may still be draining the final deltas and needs to | |
| 6469 | 6480 | // paint them even though Prompt has returned. It will reset | |
| @@ -6551,8 +6562,10 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||
| 6551 | 6562 | if i.agent != nil { | |
| 6552 | 6563 | agentQueued = i.agent.QueuedMessageCount() | |
| 6553 | 6564 | } | |
| 6565 | + | continueQueued := !awaitingPre && !hasNext && agentQueued > 0 && err == nil && ctx.Err() == nil | |
| 6554 | 6566 | shouldAutoCompact := !awaitingPre && !hasNext && agentQueued == 0 && err == nil && ctx.Err() == nil && i.shouldAutoCompactLocked() | |
| 6555 | 6567 | alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext || agentQueued > 0, offer, recoverContextOverflow, shouldAutoCompact) | |
| 6568 | + | i.busy = hasNext || continueQueued || recoverContextOverflow || shouldAutoCompact | |
| 6556 | 6569 | i.mu.Unlock() | |
| 6557 | 6570 | if alertReason != "" { | |
| 6558 | 6571 | i.scheduleMainAlert(alertReason) | |
| @@ -6569,6 +6582,8 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||
| 6569 | 6582 | switch { | |
| 6570 | 6583 | case hasNext: | |
| 6571 | 6584 | i.startTurn(parent, next) | |
| 6585 | + | case continueQueued: | |
| 6586 | + | i.startTurnRequest(parent, "", nil, true, false) | |
| 6572 | 6587 | case offer: | |
| 6573 | 6588 | i.openRescueDialog(rescueProv, rescueFprov, rescueModel, rescueWhy, prompt, rescueImgs) | |
| 6574 | 6589 | case recoverContextOverflow: | |
| @@ -3595,19 +3595,17 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) { | |||||
| 3595 | i.maybeStartSessionTitle(i.runCtx, text) | 3595 | i.maybeStartSessionTitle(i.runCtx, text) | ||
| 3596 | i.mu.Lock() | 3596 | i.mu.Lock() | ||
| 3597 | if i.busy { | 3597 | if i.busy { | ||
| 3598 | - | // Queue text only; images are dropped for queued prompts. Compaction | 3598 | + | // Queue text only; images are dropped for queued prompts. Keep the |
| 3599 | - | // uses the host queue because it has no active agent loop to drain | 3599 | + | // interactive mutex held through Agent.QueueMessage so turn teardown |
| 3600 | - | // Agent.QueueMessage entries. | 3600 | + | // cannot inspect the agent queue between the busy check and enqueue. |
| 3601 | - | ag := i.agent | 3601 | + | // Compaction uses the host queue because it has no active agent loop |
| 3602 | - | compacting := i.compacting | 3602 | + | // to drain Agent.QueueMessage entries. |
| 3603 | - | i.mu.Unlock() | 3603 | + | if i.agent != nil && !i.compacting { |
| 3604 | - | if ag != nil && !compacting { | 3604 | + | i.agent.QueueMessage(text) |
| 3605 | - | ag.QueueMessage(text) | |||
| 3606 | } else { | 3605 | } else { | ||
| 3607 | - | i.mu.Lock() | |||
| 3608 | i.queued = append(i.queued, text) | 3606 | i.queued = append(i.queued, text) | ||
| 3609 | - | i.mu.Unlock() | |||
| 3610 | } | 3607 | } | ||
| 3608 | + | i.mu.Unlock() | |||
| 3611 | i.invalidate() | 3609 | i.invalidate() | ||
| 3612 | return | 3610 | return | ||
| 3613 | } | 3611 | } | ||
| @@ -5675,22 +5673,20 @@ func (i *Interactive) submitOrQueuePrompt(ctx context.Context, prompt string) { | |||||
| 5675 | i.invalidate() | 5673 | i.invalidate() | ||
| 5676 | return | 5674 | return | ||
| 5677 | } | 5675 | } | ||
| 5678 | - | busy := i.busy | 5676 | + | if i.busy { |
| 5679 | - | compacting := i.compacting | 5677 | + | // Keep the mutex held while enqueueing so the turn-completion |
| 5680 | - | ag := i.agent | 5678 | + | // goroutine cannot publish idle and inspect an empty queue between |
| 5681 | - | i.mu.Unlock() | 5679 | + | // this check and the append. |
| 5682 | - | 5680 | + | if i.compacting { | |
| 5683 | - | if busy { | |||
| 5684 | - | if !compacting { | |||
| 5685 | - | ag.QueueMessage(prompt) | |||
| 5686 | - | } else { | |||
| 5687 | - | i.mu.Lock() | |||
| 5688 | i.queued = append(i.queued, prompt) | 5681 | i.queued = append(i.queued, prompt) | ||
| 5689 | - | i.mu.Unlock() | 5682 | + | } else { |
| 5683 | + | i.agent.QueueMessage(prompt) | |||
| 5690 | } | 5684 | } | ||
| 5685 | + | i.mu.Unlock() | |||
| 5691 | i.invalidate() | 5686 | i.invalidate() | ||
| 5692 | return | 5687 | return | ||
| 5693 | } | 5688 | } | ||
| 5689 | + | i.mu.Unlock() | |||
| 5694 | i.startTurn(ctx, prompt) | 5690 | i.startTurn(ctx, prompt) | ||
| 5695 | } | 5691 | } | ||
| 5696 | 5692 | ||||
| @@ -6121,11 +6117,11 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||||
| 6121 | summary, err := i.agent.Compact(ctx, keepTail, sink) | 6117 | summary, err := i.agent.Compact(ctx, keepTail, sink) | ||
| 6122 | _ = summary | 6118 | _ = summary | ||
| 6123 | i.mu.Lock() | 6119 | i.mu.Lock() | ||
| 6124 | - | i.busy = false | 6120 | + | // Keep busy/compacting asserted while cleanup and queue selection run. |
| 6125 | - | i.compacting = false | 6121 | + | // Completion updates can arrive in this window; clearing busy before |
| 6122 | + | // inspecting the queues lets them race a new turn or be stranded. | |||
| 6126 | i.resetStreamingStateLocked() | 6123 | i.resetStreamingStateLocked() | ||
| 6127 | i.cancelTurn = nil | 6124 | i.cancelTurn = nil | ||
| 6128 | - | i.autoCompacting = false | |||
| 6129 | 6125 | ||||
| 6130 | // Drain pending work after the transcript is clean. Overflow recovery | 6126 | // Drain pending work after the transcript is clean. Overflow recovery | ||
| 6131 | // continues the user message already in the transcript. Pre-turn | 6127 | // continues the user message already in the transcript. Pre-turn | ||
| @@ -6203,12 +6199,13 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) { | |||||
| 6203 | continueAutomatically = true | 6199 | continueAutomatically = true | ||
| 6204 | } | 6200 | } | ||
| 6205 | } | 6201 | } | ||
| 6206 | - | if hasNext || continueExisting || continueAutomatically { | 6202 | + | // Keep the host busy until the hand-off decision is committed under |
| 6207 | - | // Keep the host busy while handing the next turn back to the | 6203 | + | // the mutex. A completion update arriving after the compaction result |
| 6208 | - | // agent. Prompts submitted in this hand-off window then enter | 6204 | + | // but before this assignment is then queued for the selected follow-up |
| 6209 | - | // the agent queue instead of racing a second startTurn call. | 6205 | + | // instead of starting a competing turn. |
| 6210 | - | i.busy = true | 6206 | + | i.busy = hasNext || continueExisting || continueAutomatically |
| 6211 | - | } | 6207 | + | i.compacting = false |
| 6208 | + | i.autoCompacting = false | |||
| 6212 | i.mu.Unlock() | 6209 | i.mu.Unlock() | ||
| 6213 | i.invalidate() | 6210 | i.invalidate() | ||
| 6214 | 6211 | ||||
| @@ -6387,9 +6384,19 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||||
| 6387 | i.mu.Lock() | 6384 | i.mu.Lock() | ||
| 6388 | needsPreCompact := !overflowRecoveryAttempted && !i.autoCompacting && i.shouldAutoCompactLocked() | 6385 | needsPreCompact := !overflowRecoveryAttempted && !i.autoCompacting && i.shouldAutoCompactLocked() | ||
| 6389 | if needsPreCompact { | 6386 | if needsPreCompact { | ||
| 6390 | - | i.pendingCompactPrompt = prompt | 6387 | + | // Reserve the compaction hand-off before releasing i.mu. A completion |
| 6391 | - | i.pendingCompactImages = append([]provider.ImageBlock(nil), images...) | 6388 | + | // update arriving in the small gap before runCompact acquires the |
| 6392 | - | i.hasPendingCompactPrompt = true | 6389 | + | // mutex must join the host queue rather than starting a competing turn. |
| 6390 | + | i.busy = true | |||
| 6391 | + | i.compacting = true | |||
| 6392 | + | i.autoCompacting = true | |||
| 6393 | + | if continueExisting { | |||
| 6394 | + | i.continueAfterCompact = true | |||
| 6395 | + | } else { | |||
| 6396 | + | i.pendingCompactPrompt = prompt | |||
| 6397 | + | i.pendingCompactImages = append([]provider.ImageBlock(nil), images...) | |||
| 6398 | + | i.hasPendingCompactPrompt = true | |||
| 6399 | + | } | |||
| 6393 | i.statusErr = "" | 6400 | i.statusErr = "" | ||
| 6394 | i.extNotes = append(i.extNotes, autoCompactNoteLine(i.cfg.Theme, "context near limit — condensing history before sending...")) | 6401 | i.extNotes = append(i.extNotes, autoCompactNoteLine(i.cfg.Theme, "context near limit — condensing history before sending...")) | ||
| 6395 | i.pendingPostCompactNote = "context auto-compacted; sending your last message" | 6402 | i.pendingPostCompactNote = "context auto-compacted; sending your last message" | ||
| @@ -6463,7 +6470,11 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||||
| 6463 | err = i.agent.Prompt(ctx, prompt, images, sink) | 6470 | err = i.agent.Prompt(ctx, prompt, images, sink) | ||
| 6464 | } | 6471 | } | ||
| 6465 | i.mu.Lock() | 6472 | i.mu.Lock() | ||
| 6466 | - | i.busy = false | 6473 | + | // Keep busy asserted through final cleanup and queue selection. A |
| 6474 | + | // watcher may enqueue a completion summary concurrently with the | |||
| 6475 | + | // provider's final event; publishing idle before inspecting queues can | |||
| 6476 | + | // strand that summary in the core agent queue or start an overlapping | |||
| 6477 | + | // turn. | |||
| 6467 | // Don't touch streamPending / streamFlushPending here — the | 6478 | // Don't touch streamPending / streamFlushPending here — the | ||
| 6468 | // pacer may still be draining the final deltas and needs to | 6479 | // pacer may still be draining the final deltas and needs to | ||
| 6469 | // paint them even though Prompt has returned. It will reset | 6480 | // paint them even though Prompt has returned. It will reset | ||
| @@ -6551,8 +6562,10 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||||
| 6551 | if i.agent != nil { | 6562 | if i.agent != nil { | ||
| 6552 | agentQueued = i.agent.QueuedMessageCount() | 6563 | agentQueued = i.agent.QueuedMessageCount() | ||
| 6553 | } | 6564 | } | ||
| 6565 | + | continueQueued := !awaitingPre && !hasNext && agentQueued > 0 && err == nil && ctx.Err() == nil | |||
| 6554 | shouldAutoCompact := !awaitingPre && !hasNext && agentQueued == 0 && err == nil && ctx.Err() == nil && i.shouldAutoCompactLocked() | 6566 | shouldAutoCompact := !awaitingPre && !hasNext && agentQueued == 0 && err == nil && ctx.Err() == nil && i.shouldAutoCompactLocked() | ||
| 6555 | alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext || agentQueued > 0, offer, recoverContextOverflow, shouldAutoCompact) | 6567 | alertReason := mainAlertReason(ctx, err, lastTurnErr, lastStop, awaitingPre, hasNext || agentQueued > 0, offer, recoverContextOverflow, shouldAutoCompact) | ||
| 6568 | + | i.busy = hasNext || continueQueued || recoverContextOverflow || shouldAutoCompact | |||
| 6556 | i.mu.Unlock() | 6569 | i.mu.Unlock() | ||
| 6557 | if alertReason != "" { | 6570 | if alertReason != "" { | ||
| 6558 | i.scheduleMainAlert(alertReason) | 6571 | i.scheduleMainAlert(alertReason) | ||
| @@ -6569,6 +6582,8 @@ func (i *Interactive) startTurnRequest(parent context.Context, prompt string, im | |||||
| 6569 | switch { | 6582 | switch { | ||
| 6570 | case hasNext: | 6583 | case hasNext: | ||
| 6571 | i.startTurn(parent, next) | 6584 | i.startTurn(parent, next) | ||
| 6585 | + | case continueQueued: | |||
| 6586 | + | i.startTurnRequest(parent, "", nil, true, false) | |||
| 6572 | case offer: | 6587 | case offer: | ||
| 6573 | i.openRescueDialog(rescueProv, rescueFprov, rescueModel, rescueWhy, prompt, rescueImgs) | 6588 | i.openRescueDialog(rescueProv, rescueFprov, rescueModel, rescueWhy, prompt, rescueImgs) | ||
| 6574 | case recoverContextOverflow: | 6589 | case recoverContextOverflow: | ||