feat(core): implement mid-turn queue drain for agent execution

packages/cli/src/ui/AppContainer.tsx

@@ -697,6 +697,7 @@ export const AppContainer = (props: AppContainerProps) => {
   }, [config, historyManager, settings.merged]);
 
   const cancelHandlerRef = useRef<() => void>(() => {});
+  const midTurnDrainRef = useRef<(() => string[]) | null>(null);
 
   const {
     streamingState,
@@ -728,6 +729,7 @@ export const AppContainer = (props: AppContainerProps) => {
     setEmbeddedShellFocused,
     terminalWidth,
     terminalHeight,
+    midTurnDrainRef,
   );
 
   // Track whether suggestions are visible for Tab key handling
@@ -751,6 +753,19 @@ export const AppContainer = (props: AppContainerProps) => {
       submitQuery,
     });
 
+  // Bridge message queue to mid-turn drain via ref (avoids stale closure).
+  const messageQueueRef = useRef(messageQueue);
+  messageQueueRef.current = messageQueue;
+  midTurnDrainRef.current = () => {
+    const queue = messageQueueRef.current;
+    if (queue.length === 0) return [];
+    // Synchronously clear the ref to prevent duplicate drains if called
+    // again before React re-renders.
+    messageQueueRef.current = [];
+    clearQueue();
+    return [...queue];
+  };
+
   // Callback for handling final submit (must be after addMessage from useMessageQueue)
   const handleFinalSubmit = useCallback(
     (submittedValue: string) => {

packages/cli/src/ui/components/agent-view/AgentComposer.tsx

@@ -21,9 +21,9 @@ import { Box, Text, useStdin } from 'ink';
 import { useCallback, useEffect, useMemo, useState } from 'react';
 import {
   AgentStatus,
-  isTerminalStatus,
   ApprovalMode,
   APPROVAL_MODES,
+  AgentEventType,
 } from '@qwen-code/qwen-code-core';
 import {
   useAgentViewState,
@@ -184,32 +184,40 @@ export const AgentComposer: React.FC<AgentComposerProps> = ({ agentId }) => {
     [buffer, agentTabBarFocused, setAgentTabBarFocused],
   );
 
-  // ── Message queue (accumulate while streaming, flush as one prompt on idle) ──
+  // ── Message queue (mid-turn drain: always enqueue, track display locally) ──
 
-  const [messageQueue, setMessageQueue] = useState<string[]>([]);
+  const [pendingMessages, setPendingMessages] = useState<string[]>([]);
 
-  // When agent becomes idle (and not terminal), flush queued messages.
+  // Clear display when messages are consumed mid-turn or agent goes idle.
   useEffect(() => {
-    if (
-      streamingState === StreamingState.Idle &&
-      messageQueue.length > 0 &&
-      status !== undefined &&
-      !isTerminalStatus(status)
-    ) {
-      const combined = messageQueue.join('\n');
-      setMessageQueue([]);
-      interactiveAgent?.enqueueMessage(combined);
-    }
-  }, [streamingState, messageQueue, interactiveAgent, status]);
+    const emitter = interactiveAgent?.getEventEmitter();
+    if (!emitter) return;
+
+    const clearDisplay = () => setPendingMessages([]);
+    const clearOnIdle = (event: { newStatus?: AgentStatus }) => {
+      if (event.newStatus === AgentStatus.IDLE) {
+        setPendingMessages([]);
+      }
+    };
+    emitter.on(AgentEventType.QUEUE_MESSAGES_CONSUMED, clearDisplay);
+    emitter.on(AgentEventType.STATUS_CHANGE, clearOnIdle);
+
+    return () => {
+      emitter.off(AgentEventType.QUEUE_MESSAGES_CONSUMED, clearDisplay);
+      emitter.off(AgentEventType.STATUS_CHANGE, clearOnIdle);
+    };
+  }, [interactiveAgent]);
 
   const handleSubmit = useCallback(
     (text: string) => {
       const trimmed = text.trim();
       if (!trimmed || !interactiveAgent) return;
-      if (streamingState === StreamingState.Idle) {
-        interactiveAgent.enqueueMessage(trimmed);
-      } else {
-        setMessageQueue((prev) => [...prev, trimmed]);
+      // Always enqueue directly — mid-turn drain picks up messages
+      // during tool execution rather than waiting for idle.
+      interactiveAgent.enqueueMessage(trimmed);
+      // Track for display when agent is busy
+      if (streamingState !== StreamingState.Idle) {
+        setPendingMessages((prev) => [...prev, trimmed]);
       }
     },
     [interactiveAgent, streamingState],
@@ -279,7 +287,7 @@ export const AgentComposer: React.FC<AgentComposerProps> = ({ agentId }) => {
           </Box>
         )}
 
-        <QueuedMessageDisplay messageQueue={messageQueue} />
+        <QueuedMessageDisplay messageQueue={pendingMessages} />
 
         {/* Input prompt — always visible, like the main Composer */}
         <BaseTextInput

packages/cli/src/ui/hooks/useGeminiStream.ts

@@ -172,6 +172,7 @@ export const useGeminiStream = (
   setShellInputFocused: (value: boolean) => void,
   terminalWidth: number,
   terminalHeight: number,
+  midTurnDrainRef?: React.RefObject<(() => string[]) | null>,
 ) => {
   const [initError, setInitError] = useState<string | null>(null);
   const abortControllerRef = useRef<AbortController | null>(null);
@@ -1561,6 +1562,17 @@ export const useGeminiStream = (
         return;
       }
 
+      // Mid-turn queue drain: inject queued user messages alongside tool
+      // results so the model sees them in the next API call.
+      const drained = midTurnDrainRef?.current?.() ?? [];
+      if (drained.length > 0) {
+        for (const msg of drained) {
+          responsesToSend.push({
+            text: `\n[User message received during tool execution]: ${msg}`,
+          });
+        }
+      }
+
       submitQuery(responsesToSend, SendMessageType.ToolResult, prompt_ids[0]);
     },
     [
@@ -1571,6 +1583,7 @@ export const useGeminiStream = (
       performMemoryRefresh,
       modelSwitchedFromQuotaError,
       config,
+      midTurnDrainRef,
     ],
   );
 

packages/core/src/agents/runtime/agent-core.ts

@@ -84,6 +84,12 @@ export interface ReasoningLoopOptions {
   maxTimeMinutes?: number;
   /** Start time in ms (for timeout calculation). Defaults to Date.now(). */
   startTimeMs?: number;
+  /**
+   * Optional callback invoked after tool execution completes, before the
+   * next API call. Returns queued user messages to inject into the current
+   * turn so the model sees them immediately (mid-turn queue drain).
+   */
+  midTurnDrain?: () => string[];
 }
 
 /**
@@ -490,6 +496,20 @@ export class AgentCore {
           toolsList,
           currentResponseId,
         );
+
+        // Mid-turn queue drain: inject queued user messages alongside tool
+        // results so the model sees them in the next API call.
+        // Skip if aborted — messages stay in queue for the next round.
+        if (options?.midTurnDrain && !abortController.signal.aborted) {
+          const drained = options.midTurnDrain();
+          if (drained.length > 0 && currentMessages[0]?.parts) {
+            for (const msg of drained) {
+              currentMessages[0].parts.push({
+                text: `\n[User message received during tool execution]: ${msg}`,
+              });
+            }
+          }
+        }
       } else {
         // No tool calls — treat this as the model's final answer.
         if (roundText && roundText.trim().length > 0) {

packages/core/src/agents/runtime/agent-events.ts

@@ -37,7 +37,8 @@ export type AgentEvent =
   | 'usage_metadata'
   | 'finish'
   | 'error'
-  | 'status_change';
+  | 'status_change'
+  | 'queue_messages_consumed';
 
 export enum AgentEventType {
   START = 'start',
@@ -54,6 +55,7 @@ export enum AgentEventType {
   FINISH = 'finish',
   ERROR = 'error',
   STATUS_CHANGE = 'status_change',
+  QUEUE_MESSAGES_CONSUMED = 'queue_messages_consumed',
 }
 
 // ─── Event Payloads ─────────────────────────────────────────
@@ -172,6 +174,15 @@ export interface AgentErrorEvent {
   timestamp: number;
 }
 
+export interface AgentQueueDrainEvent {
+  subagentId: string;
+  /** The messages that were consumed from the queue. */
+  messages: string[];
+  /** Number of messages consumed. */
+  count: number;
+  timestamp: number;
+}
+
 export interface AgentStatusChangeEvent {
   agentId: string;
   previousStatus: AgentStatus;
@@ -200,6 +211,7 @@ export interface AgentEventMap {
   [AgentEventType.FINISH]: AgentFinishEvent;
   [AgentEventType.ERROR]: AgentErrorEvent;
   [AgentEventType.STATUS_CHANGE]: AgentStatusChangeEvent;
+  [AgentEventType.QUEUE_MESSAGES_CONSUMED]: AgentQueueDrainEvent;
 }
 
 // ─── Event Emitter ──────────────────────────────────────────