Deliver sliding-in messages during agent loop

commit: 36f190af3104b37ebd729a5a4d7a30561b9f084a
author: patriceckhart · 5mo ago · 2026-05-16 10:47 UTC
parent: 467ad0a
signature: unsigned

3 files · +337 -22

M internal/agent/modes/interactive.go +77 -22


@@ -1019,9 +1019,13 @@ func (i *Interactive) redraw() {
1019	1019		// in flight. Shown directly above the status bar so they're close
1020	1020		// to the editor but don't push the chat around.
1021	1021		var queue []string
1022		-	if len(i.queued) > 0 {
	1022	+	queued := append([]string(nil), i.queued...)
	1023	+	if i.agent != nil {
	1024	+	queued = append(queued, i.agent.PendingQueuedMessages()...)
	1025	+	}
	1026	+	if len(queued) > 0 {
1023	1027		queue = append(queue, "")
1024		-	for _, q := range i.queued {
	1028	+	for _, q := range queued {
1025	1029		label := i.cfg.Theme.FG256(i.cfg.Theme.Accent, " sliding in: ")
1026	1030		text := truncateLine(q, cols-17)
1027	1031		queue = append(queue, label+i.cfg.Theme.FG256(i.cfg.Theme.Muted, text))

@@ -1701,10 +1705,18 @@ func (i *Interactive) handleKey(ctx context.Context, k tui.Key) (done bool) {
1701	1705		// follow-up messages); a second press within ctrlCExitWindow
1702	1706		// exits. With both an empty editor and no queue the first
1703	1707		// press still just arms — require a deliberate double-tap.
1704		-	hadInput := !i.ed.IsEmpty() \|\| len(i.queued) > 0
	1708	+	ag := i.agent
	1709	+	pending := 0
	1710	+	if ag != nil {
	1711	+	pending = ag.QueuedMessageCount()
	1712	+	}
	1713	+	hadInput := !i.ed.IsEmpty() \|\| len(i.queued) > 0 \|\| pending > 0
1705	1714		if hadInput {
1706	1715		i.ed.Clear()
1707	1716		i.suggest.Reset()
	1717	+	if ag != nil {
	1718	+	ag.DrainQueuedMessages()
	1719	+	}
1708	1720		i.mu.Lock()
1709	1721		i.queued = nil
1710	1722		i.statusOK = "input cleared"

@@ -1792,10 +1804,18 @@ func (i *Interactive) handleKey(ctx context.Context, k tui.Key) (done bool) {
1792	1804		// the keypress falls through to the normal scroll behavior.
1793	1805		if k.Alt {
1794	1806		i.mu.Lock()
1795		-	if n := len(i.queued); n > 0 {
1796		-	text := i.queued[n-1]
1797		-	i.queued = i.queued[:n-1]
1798		-	i.mu.Unlock()
	1807	+	var text string
	1808	+	if i.agent != nil {
	1809	+	text, _ = i.agent.PopQueuedMessage()
	1810	+	}
	1811	+	if text == "" {
	1812	+	if n := len(i.queued); n > 0 {
	1813	+	text = i.queued[n-1]
	1814	+	i.queued = i.queued[:n-1]
	1815	+	}
	1816	+	}
	1817	+	i.mu.Unlock()
	1818	+	if text != "" {
1799	1819		i.ed.SetValue(text)
1800	1820		i.inputHistoryIndex = -1
1801	1821		i.invalidate()

@@ -1978,17 +1998,24 @@ func (i *Interactive) handleKey(ctx context.Context, k tui.Key) (done bool) {
1978	1998		if i.telegramBridge != nil && i.telegramBridge.Active() {
1979	1999		go i.telegramBridge.OnUserTyped(text)
1980	2000		}
1981		-	// If a turn is already in flight, queue this prompt instead of
1982		-	// starting a second one. The drain loop at the end of startTurn
1983		-	// will pick it up when the current turn finishes.
	2001	+	// If a turn is already in flight, queue this prompt inside the
	2002	+	// agent loop so it is delivered at the next safe model-call
	2003	+	// boundary instead of waiting for the whole run to finish.
1984	2004		i.mu.Lock()
1985		-	if i.busy {
1986		-	i.queued = append(i.queued, text)
1987		-	i.mu.Unlock()
	2005	+	busy := i.busy
	2006	+	ag := i.agent
	2007	+	i.mu.Unlock()
	2008	+	if busy {
	2009	+	if ag != nil {
	2010	+	ag.QueueMessage(text)
	2011	+	} else {
	2012	+	i.mu.Lock()
	2013	+	i.queued = append(i.queued, text)
	2014	+	i.mu.Unlock()
	2015	+	}
1988	2016		i.invalidate()
1989	2017		return false
1990	2018		}
1991		-	i.mu.Unlock()
1992	2019		i.startTurn(ctx, text)
1993	2020		}
1994	2021		return false

@@ -2179,8 +2206,15 @@ func (i *Interactive) SubmitOrQueue(text string, images []provider.ImageBlock) {
2179	2206		}
2180	2207		if i.busy {
2181	2208		// Queue text only; images are dropped for queued prompts.
2182		-	i.queued = append(i.queued, text)
	2209	+	ag := i.agent
2183	2210		i.mu.Unlock()
	2211	+	if ag != nil {
	2212	+	ag.QueueMessage(text)
	2213	+	} else {
	2214	+	i.mu.Lock()
	2215	+	i.queued = append(i.queued, text)
	2216	+	i.mu.Unlock()
	2217	+	}
2184	2218		i.invalidate()
2185	2219		return
2186	2220		}

@@ -2404,13 +2438,20 @@ func (i *Interactive) runSlash(ctx context.Context, cmd string) (done bool) {
2404	2438		// typing the prompt by hand.
2405	2439		const studyPrompt = "Read and understand everything in the current directory."
2406	2440		i.mu.Lock()
2407		-	if i.busy {
2408		-	i.queued = append(i.queued, studyPrompt)
2409		-	i.mu.Unlock()
	2441	+	busy := i.busy
	2442	+	ag := i.agent
	2443	+	i.mu.Unlock()
	2444	+	if busy {
	2445	+	if ag != nil {
	2446	+	ag.QueueMessage(studyPrompt)
	2447	+	} else {
	2448	+	i.mu.Lock()
	2449	+	i.queued = append(i.queued, studyPrompt)
	2450	+	i.mu.Unlock()
	2451	+	}
2410	2452		i.invalidate()
2411	2453		break
2412	2454		}
2413		-	i.mu.Unlock()
2414	2455		i.startTurn(ctx, studyPrompt)
2415	2456		case "/jail":
2416	2457		if i.cfg.Sandbox == nil {

@@ -3136,10 +3177,16 @@ func (i *Interactive) runCompact(parent context.Context, auto bool) {
3136	3177		i.statusOK = "compaction cancelled"
3137	3178		}
3138	3179		i.queued = nil // drop queue on cancel
	3180	+	if i.agent != nil {
	3181	+	i.agent.DrainQueuedMessages()
	3182	+	}
3139	3183		case err != nil:
3140	3184		i.statusErr = "compaction failed: " + err.Error()
3141	3185		i.statusOK = ""
3142	3186		i.queued = nil // drop queue on error
	3187	+	if i.agent != nil {
	3188	+	i.agent.DrainQueuedMessages()
	3189	+	}
3143	3190		default:
3144	3191		i.statusErr = ""
3145	3192		// Read token count from the compaction message meta.

@@ -3327,11 +3374,19 @@ func (i *Interactive) startTurnWithImages(parent context.Context, prompt string,
3327	3374		// user isn't bombarded with stale messages after an interrupt.
3328	3375		if ctx.Err() != nil \|\| err != nil {
3329	3376		i.queued = nil
	3377	+	if i.agent != nil {
	3378	+	i.agent.DrainQueuedMessages()
	3379	+	}
3330	3380		}
3331	3381		// Decide whether the next thing to do is an auto-compaction.
3332		-	// Only fires when the turn completed cleanly AND the queue is
3333		-	// empty (otherwise a queued message would race the condense).
3334		-	shouldAutoCompact := !hasNext && err == nil && ctx.Err() == nil && i.shouldAutoCompactLocked()
	3382	+	// Only fires when the turn completed cleanly AND no host-side
	3383	+	// or agent-side queued messages are waiting (otherwise a queued
	3384	+	// message would race the condense).
	3385	+	agentQueued := 0
	3386	+	if i.agent != nil {
	3387	+	agentQueued = i.agent.QueuedMessageCount()
	3388	+	}
	3389	+	shouldAutoCompact := !hasNext && agentQueued == 0 && err == nil && ctx.Err() == nil && i.shouldAutoCompactLocked()
3335	3390		i.mu.Unlock()
3336	3391		i.invalidate()
3337	3392		parent := i.runCtx

 	// in flight. Shown directly above the status bar so they're close
 	// to the editor but don't push the chat around.
 	var queue []string
-	if len(i.queued) > 0 {
+	queued := append([]string(nil), i.queued...)
+	if i.agent != nil {
+		queued = append(queued, i.agent.PendingQueuedMessages()...)
+	}
+	if len(queued) > 0 {
 		queue = append(queue, "")
-		for _, q := range i.queued {
+		for _, q := range queued {
 			label := i.cfg.Theme.FG256(i.cfg.Theme.Accent, "  sliding in: ")
 			text := truncateLine(q, cols-17)
 			queue = append(queue, label+i.cfg.Theme.FG256(i.cfg.Theme.Muted, text))
 		// follow-up messages); a second press within ctrlCExitWindow
 		// exits. With both an empty editor and no queue the first
 		// press still just arms — require a deliberate double-tap.
-		hadInput := !i.ed.IsEmpty() || len(i.queued) > 0
+		ag := i.agent
+		pending := 0
+		if ag != nil {
+			pending = ag.QueuedMessageCount()
+		}
+		hadInput := !i.ed.IsEmpty() || len(i.queued) > 0 || pending > 0
 		if hadInput {
 			i.ed.Clear()
 			i.suggest.Reset()
+			if ag != nil {
+				ag.DrainQueuedMessages()
+			}
 			i.mu.Lock()
 			i.queued = nil
 			i.statusOK = "input cleared"
 		// the keypress falls through to the normal scroll behavior.
 		if k.Alt {
 			i.mu.Lock()
-			if n := len(i.queued); n > 0 {
+			var text string
-				text := i.queued[n-1]
+			if i.agent != nil {
-				i.queued = i.queued[:n-1]
+				text, _ = i.agent.PopQueuedMessage()
-				i.mu.Unlock()
+			}
+			if text == "" {
+				if n := len(i.queued); n > 0 {
+					text = i.queued[n-1]
+					i.queued = i.queued[:n-1]
+				}
+			}
+			i.mu.Unlock()
+			if text != "" {
 				i.ed.SetValue(text)
 				i.inputHistoryIndex = -1
 				i.invalidate()
 		if i.telegramBridge != nil && i.telegramBridge.Active() {
 			go i.telegramBridge.OnUserTyped(text)
 		}
-		// If a turn is already in flight, queue this prompt instead of
+		// If a turn is already in flight, queue this prompt inside the
-		// starting a second one. The drain loop at the end of startTurn
+		// agent loop so it is delivered at the next safe model-call
-		// will pick it up when the current turn finishes.
+		// boundary instead of waiting for the whole run to finish.
 		i.mu.Lock()
-		if i.busy {
+		busy := i.busy
-			i.queued = append(i.queued, text)
+		ag := i.agent
-			i.mu.Unlock()
+		i.mu.Unlock()
+		if busy {
+			if ag != nil {
+				ag.QueueMessage(text)
+			} else {
+				i.mu.Lock()
+				i.queued = append(i.queued, text)
+				i.mu.Unlock()
+			}
 			i.invalidate()
 			return false
 		}
-		i.mu.Unlock()
 		i.startTurn(ctx, text)
 	}
 	return false
 	}
 	if i.busy {
 		// Queue text only; images are dropped for queued prompts.
-		i.queued = append(i.queued, text)
+		ag := i.agent
 		i.mu.Unlock()
+		if ag != nil {
+			ag.QueueMessage(text)
+		} else {
+			i.mu.Lock()
+			i.queued = append(i.queued, text)
+			i.mu.Unlock()
+		}
 		i.invalidate()
 		return
 	}
 		// typing the prompt by hand.
 		const studyPrompt = "Read and understand everything in the current directory."
 		i.mu.Lock()
-		if i.busy {
+		busy := i.busy
-			i.queued = append(i.queued, studyPrompt)
+		ag := i.agent
-			i.mu.Unlock()
+		i.mu.Unlock()
+		if busy {
+			if ag != nil {
+				ag.QueueMessage(studyPrompt)
+			} else {
+				i.mu.Lock()
+				i.queued = append(i.queued, studyPrompt)
+				i.mu.Unlock()
+			}
 			i.invalidate()
 			break
 		}
-		i.mu.Unlock()
 		i.startTurn(ctx, studyPrompt)
 	case "/jail":
 		if i.cfg.Sandbox == nil {
 				i.statusOK = "compaction cancelled"
 			}
 			i.queued = nil // drop queue on cancel
+			if i.agent != nil {
+				i.agent.DrainQueuedMessages()
+			}
 		case err != nil:
 			i.statusErr = "compaction failed: " + err.Error()
 			i.statusOK = ""
 			i.queued = nil // drop queue on error
+			if i.agent != nil {
+				i.agent.DrainQueuedMessages()
+			}
 		default:
 			i.statusErr = ""
 			// Read token count from the compaction message meta.
 		// user isn't bombarded with stale messages after an interrupt.
 		if ctx.Err() != nil || err != nil {
 			i.queued = nil
+			if i.agent != nil {
+				i.agent.DrainQueuedMessages()
+			}
 		}
 		// Decide whether the next thing to do is an auto-compaction.
-		// Only fires when the turn completed cleanly AND the queue is
+		// Only fires when the turn completed cleanly AND no host-side
-		// empty (otherwise a queued message would race the condense).
+		// or agent-side queued messages are waiting (otherwise a queued
-		shouldAutoCompact := !hasNext && err == nil && ctx.Err() == nil && i.shouldAutoCompactLocked()
+		// message would race the condense).
+		agentQueued := 0
+		if i.agent != nil {
+			agentQueued = i.agent.QueuedMessageCount()
+		}
+		shouldAutoCompact := !hasNext && agentQueued == 0 && err == nil && ctx.Err() == nil && i.shouldAutoCompactLocked()
 		i.mu.Unlock()
 		i.invalidate()
 		parent := i.runCtx

M internal/core/agent.go +108 -0


@@ -5,6 +5,7 @@ import (
5	5		"encoding/json"
6	6		"errors"
7	7		"fmt"
	8	+	"strings"
8	9		"sync"
9	10		"time"
10	11

@@ -81,6 +82,13 @@ type Agent struct {
81	82		// every keypress.
82	83		rev uint64
83	84		cost CostTracker
	85	+
	86	+	// queued holds user messages submitted while the agent is busy.
	87	+	// The loop appends them as normal user messages at safe
	88	+	// boundaries: before the next model call after a tool batch, or
	89	+	// after a text-only assistant turn finishes. It never interrupts
	90	+	// a running tool or cancels an in-flight provider request.
	91	+	queued []string
84	92		}
85	93
86	94		// NewAgent returns an Agent with sensible defaults.

@@ -94,6 +102,89 @@ func NewAgent(client provider.Client, model, system string, tools Registry) *Age
94	102		}
95	103		}
96	104
	105	+	// QueueMessage queues text to be injected as a user message at the
	106	+	// next safe boundary of the active agent loop. It is non-blocking in
	107	+	// the sense that it never waits for model/tool work; it only takes
	108	+	// the transcript mutex briefly. Empty/whitespace-only messages are
	109	+	// ignored.
	110	+	func (a *Agent) QueueMessage(text string) bool {
	111	+	text = strings.TrimSpace(text)
	112	+	if text == "" {
	113	+	return false
	114	+	}
	115	+	a.mu.Lock()
	116	+	a.queued = append(a.queued, text)
	117	+	a.mu.Unlock()
	118	+	return true
	119	+	}
	120	+
	121	+	// PendingQueuedMessages returns a snapshot of user messages waiting
	122	+	// to be injected. Used by hosts to render the visible "sliding in"
	123	+	// chips without consuming them.
	124	+	func (a *Agent) PendingQueuedMessages() []string {
	125	+	a.mu.Lock()
	126	+	defer a.mu.Unlock()
	127	+	out := make([]string, len(a.queued))
	128	+	copy(out, a.queued)
	129	+	return out
	130	+	}
	131	+
	132	+	// QueuedMessageCount returns the number of messages waiting to be
	133	+	// injected at the next safe boundary.
	134	+	func (a *Agent) QueuedMessageCount() int {
	135	+	a.mu.Lock()
	136	+	defer a.mu.Unlock()
	137	+	return len(a.queued)
	138	+	}
	139	+
	140	+	// PopQueuedMessage removes and returns the most recently queued
	141	+	// message. Hosts use this for the slide-back keybinding.
	142	+	func (a *Agent) PopQueuedMessage() (string, bool) {
	143	+	a.mu.Lock()
	144	+	defer a.mu.Unlock()
	145	+	n := len(a.queued)
	146	+	if n == 0 {
	147	+	return "", false
	148	+	}
	149	+	text := a.queued[n-1]
	150	+	a.queued = a.queued[:n-1]
	151	+	return text, true
	152	+	}
	153	+
	154	+	// DrainQueuedMessages discards and returns every queued message.
	155	+	// Hosts use this on explicit cancel/clear so stale follow-ups do
	156	+	// not run after the user aborted the turn.
	157	+	func (a *Agent) DrainQueuedMessages() []string {
	158	+	return a.drainQueuedMessages()
	159	+	}
	160	+
	161	+	func (a *Agent) drainQueuedMessages() []string {
	162	+	a.mu.Lock()
	163	+	defer a.mu.Unlock()
	164	+	out := make([]string, len(a.queued))
	165	+	copy(out, a.queued)
	166	+	a.queued = nil
	167	+	return out
	168	+	}
	169	+
	170	+	func (a *Agent) appendQueuedAsUser(texts []string, sink func(AgentEvent)) {
	171	+	for _, text := range texts {
	172	+	msg := provider.Message{
	173	+	Role: provider.RoleUser,
	174	+	Content: []provider.Content{provider.TextBlock{Text: text}},
	175	+	Time: time.Now(),
	176	+	}
	177	+	a.mu.Lock()
	178	+	a.messages = append(a.messages, msg)
	179	+	a.rev++
	180	+	a.mu.Unlock()
	181	+	a.fireMessageAppended(msg)
	182	+	if sink != nil {
	183	+	sink(EvUserMessage{Message: msg})
	184	+	}
	185	+	}
	186	+	}
	187	+
97	188		// Messages returns a copy of the current transcript.
98	189		func (a *Agent) Messages() []provider.Message {
99	190		a.mu.Lock()

@@ -208,6 +299,15 @@ func (a *Agent) wrapSink(sink func(AgentEvent)) func(AgentEvent) {
208	299
209	300		func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error {
210	301		for step := 1; a.MaxSteps <= 0 \|\| step <= a.MaxSteps; step++ {
	302	+	// Messages queued while the agent was busy are delivered
	303	+	// before the next model call. This is the safe boundary:
	304	+	// any previous tool batch has already completed and its
	305	+	// results have been appended, but no new provider request has
	306	+	// started yet.
	307	+	if pending := a.drainQueuedMessages(); len(pending) > 0 {
	308	+	a.appendQueuedAsUser(pending, sink)
	309	+	}
	310	+
211	311		sink(EvTurnStart{Step: step})
212	312		if a.BeforeTurn != nil {
213	313		if allowed, reason := a.BeforeTurn(step); !allowed {

@@ -260,6 +360,14 @@ func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error {
260	360		continue
261	361		}
262	362
	363	+	// If the assistant stopped without tool calls but a message was
	364	+	// queued while it was speaking, loop once more so that message
	365	+	// is appended and answered instead of waiting until a later
	366	+	// top-level prompt.
	367	+	if ctx.Err() == nil && a.QueuedMessageCount() > 0 {
	368	+	continue
	369	+	}
	370	+
263	371		// Terminal stop (end, length, error, aborted).
264	372		sink(EvDone{})
265	373		return nil


@@ -5,6 +5,7 @@ import (
5	"encoding/json"	5		"encoding/json"
6	"errors"	6		"errors"
7	"fmt"	7		"fmt"
		8	+	"strings"
8	"sync"	9		"sync"
9	"time"	10		"time"
10		11

@@ -81,6 +82,13 @@ type Agent struct {
81	// every keypress.	82		// every keypress.
82	rev uint64	83		rev uint64
83	cost CostTracker	84		cost CostTracker
		85	+
		86	+	// queued holds user messages submitted while the agent is busy.
		87	+	// The loop appends them as normal user messages at safe
		88	+	// boundaries: before the next model call after a tool batch, or
		89	+	// after a text-only assistant turn finishes. It never interrupts
		90	+	// a running tool or cancels an in-flight provider request.
		91	+	queued []string
84	}	92		}
85		93
86	// NewAgent returns an Agent with sensible defaults.	94		// NewAgent returns an Agent with sensible defaults.

@@ -94,6 +102,89 @@ func NewAgent(client provider.Client, model, system string, tools Registry) *Age
94	}	102		}
95	}	103		}
96		104
		105	+	// QueueMessage queues text to be injected as a user message at the
		106	+	// next safe boundary of the active agent loop. It is non-blocking in
		107	+	// the sense that it never waits for model/tool work; it only takes
		108	+	// the transcript mutex briefly. Empty/whitespace-only messages are
		109	+	// ignored.
		110	+	func (a *Agent) QueueMessage(text string) bool {
		111	+	text = strings.TrimSpace(text)
		112	+	if text == "" {
		113	+	return false
		114	+	}
		115	+	a.mu.Lock()
		116	+	a.queued = append(a.queued, text)
		117	+	a.mu.Unlock()
		118	+	return true
		119	+	}
		120	+
		121	+	// PendingQueuedMessages returns a snapshot of user messages waiting
		122	+	// to be injected. Used by hosts to render the visible "sliding in"
		123	+	// chips without consuming them.
		124	+	func (a *Agent) PendingQueuedMessages() []string {
		125	+	a.mu.Lock()
		126	+	defer a.mu.Unlock()
		127	+	out := make([]string, len(a.queued))
		128	+	copy(out, a.queued)
		129	+	return out
		130	+	}
		131	+
		132	+	// QueuedMessageCount returns the number of messages waiting to be
		133	+	// injected at the next safe boundary.
		134	+	func (a *Agent) QueuedMessageCount() int {
		135	+	a.mu.Lock()
		136	+	defer a.mu.Unlock()
		137	+	return len(a.queued)
		138	+	}
		139	+
		140	+	// PopQueuedMessage removes and returns the most recently queued
		141	+	// message. Hosts use this for the slide-back keybinding.
		142	+	func (a *Agent) PopQueuedMessage() (string, bool) {
		143	+	a.mu.Lock()
		144	+	defer a.mu.Unlock()
		145	+	n := len(a.queued)
		146	+	if n == 0 {
		147	+	return "", false
		148	+	}
		149	+	text := a.queued[n-1]
		150	+	a.queued = a.queued[:n-1]
		151	+	return text, true
		152	+	}
		153	+
		154	+	// DrainQueuedMessages discards and returns every queued message.
		155	+	// Hosts use this on explicit cancel/clear so stale follow-ups do
		156	+	// not run after the user aborted the turn.
		157	+	func (a *Agent) DrainQueuedMessages() []string {
		158	+	return a.drainQueuedMessages()
		159	+	}
		160	+
		161	+	func (a *Agent) drainQueuedMessages() []string {
		162	+	a.mu.Lock()
		163	+	defer a.mu.Unlock()
		164	+	out := make([]string, len(a.queued))
		165	+	copy(out, a.queued)
		166	+	a.queued = nil
		167	+	return out
		168	+	}
		169	+
		170	+	func (a *Agent) appendQueuedAsUser(texts []string, sink func(AgentEvent)) {
		171	+	for _, text := range texts {
		172	+	msg := provider.Message{
		173	+	Role: provider.RoleUser,
		174	+	Content: []provider.Content{provider.TextBlock{Text: text}},
		175	+	Time: time.Now(),
		176	+	}
		177	+	a.mu.Lock()
		178	+	a.messages = append(a.messages, msg)
		179	+	a.rev++
		180	+	a.mu.Unlock()
		181	+	a.fireMessageAppended(msg)
		182	+	if sink != nil {
		183	+	sink(EvUserMessage{Message: msg})
		184	+	}
		185	+	}
		186	+	}
		187	+
97	// Messages returns a copy of the current transcript.	188		// Messages returns a copy of the current transcript.
98	func (a *Agent) Messages() []provider.Message {	189		func (a *Agent) Messages() []provider.Message {
99	a.mu.Lock()	190		a.mu.Lock()

@@ -208,6 +299,15 @@ func (a *Agent) wrapSink(sink func(AgentEvent)) func(AgentEvent) {
208		299
209	func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error {	300		func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error {
210	for step := 1; a.MaxSteps <= 0 \|\| step <= a.MaxSteps; step++ {	301		for step := 1; a.MaxSteps <= 0 \|\| step <= a.MaxSteps; step++ {
		302	+	// Messages queued while the agent was busy are delivered
		303	+	// before the next model call. This is the safe boundary:
		304	+	// any previous tool batch has already completed and its
		305	+	// results have been appended, but no new provider request has
		306	+	// started yet.
		307	+	if pending := a.drainQueuedMessages(); len(pending) > 0 {
		308	+	a.appendQueuedAsUser(pending, sink)
		309	+	}
		310	+
211	sink(EvTurnStart{Step: step})	311		sink(EvTurnStart{Step: step})
212	if a.BeforeTurn != nil {	312		if a.BeforeTurn != nil {
213	if allowed, reason := a.BeforeTurn(step); !allowed {	313		if allowed, reason := a.BeforeTurn(step); !allowed {

@@ -260,6 +360,14 @@ func (a *Agent) runLoop(ctx context.Context, sink func(AgentEvent)) error {
260	continue	360		continue
261	}	361		}
262		362
		363	+	// If the assistant stopped without tool calls but a message was
		364	+	// queued while it was speaking, loop once more so that message
		365	+	// is appended and answered instead of waiting until a later
		366	+	// top-level prompt.
		367	+	if ctx.Err() == nil && a.QueuedMessageCount() > 0 {
		368	+	continue
		369	+	}
		370	+
263	// Terminal stop (end, length, error, aborted).	371		// Terminal stop (end, length, error, aborted).
264	sink(EvDone{})	372		sink(EvDone{})
265	return nil	373		return nil

A internal/core/queue_test.go +152 -0

@@ -0,0 +1,152 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"sync"
	7	+	"sync/atomic"
	8	+	"testing"
	9	+
	10	+	"github.com/patriceckhart/zot/internal/provider"
	11	+	)
	12	+
	13	+	type queueFakeClient struct {
	14	+	calls int32
	15	+	}
	16	+
	17	+	func (c *queueFakeClient) Name() string { return "queue-fake" }
	18	+
	19	+	func (c *queueFakeClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
	20	+	call := atomic.AddInt32(&c.calls, 1)
	21	+	out := make(chan provider.Event, 4)
	22	+	go func() {
	23	+	defer close(out)
	24	+	out <- provider.EventStart{Provider: "queue-fake", Model: req.Model}
	25	+	switch call {
	26	+	case 1:
	27	+	out <- provider.EventToolStart{ID: "t1", Name: "echo"}
	28	+	out <- provider.EventToolEnd{ID: "t1"}
	29	+	out <- provider.EventDone{Stop: provider.StopToolUse, Message: provider.Message{
	30	+	Role: provider.RoleAssistant,
	31	+	Content: []provider.Content{
	32	+	provider.TextBlock{Text: "using tool"},
	33	+	provider.ToolCallBlock{ID: "t1", Name: "echo", Arguments: json.RawMessage(`{}`)},
	34	+	},
	35	+	}}
	36	+	case 2:
	37	+	out <- provider.EventTextDelta{Delta: "saw queued"}
	38	+	out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
	39	+	Role: provider.RoleAssistant,
	40	+	Content: []provider.Content{provider.TextBlock{Text: "saw queued"}},
	41	+	}}
	42	+	default:
	43	+	out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
	44	+	Role: provider.RoleAssistant,
	45	+	Content: []provider.Content{provider.TextBlock{Text: "extra"}},
	46	+	}}
	47	+	}
	48	+	}()
	49	+	return out, nil
	50	+	}
	51	+
	52	+	// blockingTool waits until the test has queued a message, then
	53	+	// returns. This pins the core behaviour: queued user text is delivered
	54	+	// after the current tool batch finishes and before the next model call.
	55	+	type blockingTool struct {
	56	+	started chan struct{}
	57	+	release chan struct{}
	58	+	}
	59	+
	60	+	func (t *blockingTool) Name() string { return "echo" }
	61	+	func (t *blockingTool) Description() string { return "echoes" }
	62	+	func (t *blockingTool) Schema() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
	63	+
	64	+	func (t *blockingTool) Execute(ctx context.Context, args json.RawMessage, progress func(string)) (ToolResult, error) {
	65	+	close(t.started)
	66	+	select {
	67	+	case <-ctx.Done():
	68	+	return ToolResult{Content: []provider.Content{provider.TextBlock{Text: ctx.Err().Error()}}, IsError: true}, ctx.Err()
	69	+	case <-t.release:
	70	+	}
	71	+	return ToolResult{Content: []provider.Content{provider.TextBlock{Text: "tool ok"}}}, nil
	72	+	}
	73	+
	74	+	func TestQueuedMessageInjectedAfterToolBatchBeforeNextModelCall(t *testing.T) {
	75	+	client := &queueFakeClient{}
	76	+	tool := &blockingTool{started: make(chan struct{}), release: make(chan struct{})}
	77	+	a := NewAgent(client, "fake-model", "system", Registry{"echo": tool})
	78	+
	79	+	var (
	80	+	mu sync.Mutex
	81	+	texts []string
	82	+	)
	83	+	sink := func(ev AgentEvent) {
	84	+	switch e := ev.(type) {
	85	+	case EvUserMessage:
	86	+	mu.Lock()
	87	+	texts = append(texts, "user:"+extractText(e.Message))
	88	+	mu.Unlock()
	89	+	case EvAssistantMessage:
	90	+	mu.Lock()
	91	+	texts = append(texts, "asst:"+extractText(e.Message))
	92	+	mu.Unlock()
	93	+	}
	94	+	}
	95	+
	96	+	done := make(chan error, 1)
	97	+	go func() {
	98	+	done <- a.Prompt(context.Background(), "do X", nil, sink)
	99	+	}()
	100	+
	101	+	<-tool.started
	102	+	if !a.QueueMessage("also do Y") {
	103	+	t.Fatal("QueueMessage returned false")
	104	+	}
	105	+	close(tool.release)
	106	+
	107	+	if err := <-done; err != nil {
	108	+	t.Fatalf("Prompt returned %v", err)
	109	+	}
	110	+	if got := atomic.LoadInt32(&client.calls); got != 2 {
	111	+	t.Fatalf("Stream calls = %d; want 2", got)
	112	+	}
	113	+
	114	+	mu.Lock()
	115	+	defer mu.Unlock()
	116	+	if !queueTestContains(texts, "user:also do Y") {
	117	+	t.Fatalf("queued message was not emitted as user message; texts=%v", texts)
	118	+	}
	119	+	if !queueTestContains(texts, "asst:saw queued") {
	120	+	t.Fatalf("second assistant response missing; texts=%v", texts)
	121	+	}
	122	+	}
	123	+
	124	+	func queueTestContains(xs []string, want string) bool {
	125	+	for _, x := range xs {
	126	+	if x == want {
	127	+	return true
	128	+	}
	129	+	}
	130	+	return false
	131	+	}
	132	+
	133	+	func TestQueueMessageSnapshotPopAndDrain(t *testing.T) {
	134	+	a := NewAgent(nil, "fake", "", Registry{})
	135	+	if a.QueueMessage(" ") {
	136	+	t.Fatal("blank queue message accepted")
	137	+	}
	138	+	a.QueueMessage("one")
	139	+	a.QueueMessage("two")
	140	+	if got := a.PendingQueuedMessages(); len(got) != 2 \|\| got[0] != "one" \|\| got[1] != "two" {
	141	+	t.Fatalf("PendingQueuedMessages = %v; want [one two]", got)
	142	+	}
	143	+	if text, ok := a.PopQueuedMessage(); !ok \|\| text != "two" {
	144	+	t.Fatalf("PopQueuedMessage = %q,%v; want two,true", text, ok)
	145	+	}
	146	+	if got := a.DrainQueuedMessages(); len(got) != 1 \|\| got[0] != "one" {
	147	+	t.Fatalf("DrainQueuedMessages = %v; want [one]", got)
	148	+	}
	149	+	if got := a.QueuedMessageCount(); got != 0 {
	150	+	t.Fatalf("QueuedMessageCount = %d; want 0", got)
	151	+	}
	152	+	}

@@ -0,0 +1,152 @@
	1	+	package core
	2	+
	3	+	import (
	4	+	"context"
	5	+	"encoding/json"
	6	+	"sync"
	7	+	"sync/atomic"
	8	+	"testing"
	9	+
	10	+	"github.com/patriceckhart/zot/internal/provider"
	11	+	)
	12	+
	13	+	type queueFakeClient struct {
	14	+	calls int32
	15	+	}
	16	+
	17	+	func (c *queueFakeClient) Name() string { return "queue-fake" }
	18	+
	19	+	func (c *queueFakeClient) Stream(ctx context.Context, req provider.Request) (<-chan provider.Event, error) {
	20	+	call := atomic.AddInt32(&c.calls, 1)
	21	+	out := make(chan provider.Event, 4)
	22	+	go func() {
	23	+	defer close(out)
	24	+	out <- provider.EventStart{Provider: "queue-fake", Model: req.Model}
	25	+	switch call {
	26	+	case 1:
	27	+	out <- provider.EventToolStart{ID: "t1", Name: "echo"}
	28	+	out <- provider.EventToolEnd{ID: "t1"}
	29	+	out <- provider.EventDone{Stop: provider.StopToolUse, Message: provider.Message{
	30	+	Role: provider.RoleAssistant,
	31	+	Content: []provider.Content{
	32	+	provider.TextBlock{Text: "using tool"},
	33	+	provider.ToolCallBlock{ID: "t1", Name: "echo", Arguments: json.RawMessage(`{}`)},
	34	+	},
	35	+	}}
	36	+	case 2:
	37	+	out <- provider.EventTextDelta{Delta: "saw queued"}
	38	+	out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
	39	+	Role: provider.RoleAssistant,
	40	+	Content: []provider.Content{provider.TextBlock{Text: "saw queued"}},
	41	+	}}
	42	+	default:
	43	+	out <- provider.EventDone{Stop: provider.StopEnd, Message: provider.Message{
	44	+	Role: provider.RoleAssistant,
	45	+	Content: []provider.Content{provider.TextBlock{Text: "extra"}},
	46	+	}}
	47	+	}
	48	+	}()
	49	+	return out, nil
	50	+	}
	51	+
	52	+	// blockingTool waits until the test has queued a message, then
	53	+	// returns. This pins the core behaviour: queued user text is delivered
	54	+	// after the current tool batch finishes and before the next model call.
	55	+	type blockingTool struct {
	56	+	started chan struct{}
	57	+	release chan struct{}
	58	+	}
	59	+
	60	+	func (t *blockingTool) Name() string { return "echo" }
	61	+	func (t *blockingTool) Description() string { return "echoes" }
	62	+	func (t *blockingTool) Schema() json.RawMessage { return json.RawMessage(`{"type":"object"}`) }
	63	+
	64	+	func (t *blockingTool) Execute(ctx context.Context, args json.RawMessage, progress func(string)) (ToolResult, error) {
	65	+	close(t.started)
	66	+	select {
	67	+	case <-ctx.Done():
	68	+	return ToolResult{Content: []provider.Content{provider.TextBlock{Text: ctx.Err().Error()}}, IsError: true}, ctx.Err()
	69	+	case <-t.release:
	70	+	}
	71	+	return ToolResult{Content: []provider.Content{provider.TextBlock{Text: "tool ok"}}}, nil
	72	+	}
	73	+
	74	+	func TestQueuedMessageInjectedAfterToolBatchBeforeNextModelCall(t *testing.T) {
	75	+	client := &queueFakeClient{}
	76	+	tool := &blockingTool{started: make(chan struct{}), release: make(chan struct{})}
	77	+	a := NewAgent(client, "fake-model", "system", Registry{"echo": tool})
	78	+
	79	+	var (
	80	+	mu sync.Mutex
	81	+	texts []string
	82	+	)
	83	+	sink := func(ev AgentEvent) {
	84	+	switch e := ev.(type) {
	85	+	case EvUserMessage:
	86	+	mu.Lock()
	87	+	texts = append(texts, "user:"+extractText(e.Message))
	88	+	mu.Unlock()
	89	+	case EvAssistantMessage:
	90	+	mu.Lock()
	91	+	texts = append(texts, "asst:"+extractText(e.Message))
	92	+	mu.Unlock()
	93	+	}
	94	+	}
	95	+
	96	+	done := make(chan error, 1)
	97	+	go func() {
	98	+	done <- a.Prompt(context.Background(), "do X", nil, sink)
	99	+	}()
	100	+
	101	+	<-tool.started
	102	+	if !a.QueueMessage("also do Y") {
	103	+	t.Fatal("QueueMessage returned false")
	104	+	}
	105	+	close(tool.release)
	106	+
	107	+	if err := <-done; err != nil {
	108	+	t.Fatalf("Prompt returned %v", err)
	109	+	}
	110	+	if got := atomic.LoadInt32(&client.calls); got != 2 {
	111	+	t.Fatalf("Stream calls = %d; want 2", got)
	112	+	}
	113	+
	114	+	mu.Lock()
	115	+	defer mu.Unlock()
	116	+	if !queueTestContains(texts, "user:also do Y") {
	117	+	t.Fatalf("queued message was not emitted as user message; texts=%v", texts)
	118	+	}
	119	+	if !queueTestContains(texts, "asst:saw queued") {
	120	+	t.Fatalf("second assistant response missing; texts=%v", texts)
	121	+	}
	122	+	}
	123	+
	124	+	func queueTestContains(xs []string, want string) bool {
	125	+	for _, x := range xs {
	126	+	if x == want {
	127	+	return true
	128	+	}
	129	+	}
	130	+	return false
	131	+	}
	132	+
	133	+	func TestQueueMessageSnapshotPopAndDrain(t *testing.T) {
	134	+	a := NewAgent(nil, "fake", "", Registry{})
	135	+	if a.QueueMessage(" ") {
	136	+	t.Fatal("blank queue message accepted")
	137	+	}
	138	+	a.QueueMessage("one")
	139	+	a.QueueMessage("two")
	140	+	if got := a.PendingQueuedMessages(); len(got) != 2 \|\| got[0] != "one" \|\| got[1] != "two" {
	141	+	t.Fatalf("PendingQueuedMessages = %v; want [one two]", got)
	142	+	}
	143	+	if text, ok := a.PopQueuedMessage(); !ok \|\| text != "two" {
	144	+	t.Fatalf("PopQueuedMessage = %q,%v; want two,true", text, ok)
	145	+	}
	146	+	if got := a.DrainQueuedMessages(); len(got) != 1 \|\| got[0] != "one" {
	147	+	t.Fatalf("DrainQueuedMessages = %v; want [one]", got)
	148	+	}
	149	+	if got := a.QueuedMessageCount(); got != 0 {
	150	+	t.Fatalf("QueuedMessageCount = %d; want 0", got)
	151	+	}
	152	+	}