Merge pull request #198 from 0xsegfaulted/feat/interrupt-support

feat: queue messages when agent is busy instead of discarding

core/engine.go

@@ -20,6 +20,7 @@ import (
 )
 
 const maxPlatformMessageLen = 4000
+const maxQueuedMessages = 5 // cap queued messages to bound memory usage
 
 const (
 	defaultThinkingMaxLen = 300
@@ -198,19 +199,35 @@ type workspaceInitFlow struct {
 	channelName string
 }
 
+// queuedMessage holds a message that arrived while the session was busy.
+// The message is NOT sent to agent stdin at queue time; the event loop
+// sends it after the current turn completes to avoid mid-turn interference.
+type queuedMessage struct {
+	platform      Platform
+	replyCtx      any
+	content       string
+	images        []ImageAttachment
+	files         []FileAttachment
+	fromVoice     bool
+	userID        string
+	msgPlatform   string // platform name for sender injection
+	msgSessionKey string // session key for extracting chat ID
+}
+
 // interactiveState tracks a running interactive agent session and its permission state.
 type interactiveState struct {
-	agentSession AgentSession
-	platform     Platform
-	replyCtx     any
-	workspaceDir string
-	mu           sync.Mutex
-	pending      *pendingPermission
-	approveAll   bool // when true, auto-approve all permission requests for this session
-	quiet        bool // when true, suppress thinking and tool progress for this session
-	fromVoice    bool // true if current turn originated from voice transcription
-	sideText     string
-	deleteMode   *deleteModeState
+	agentSession    AgentSession
+	platform        Platform
+	replyCtx        any
+	workspaceDir    string
+	mu              sync.Mutex
+	pending         *pendingPermission
+	pendingMessages []queuedMessage // messages queued while session was busy
+	approveAll      bool            // when true, auto-approve all permission requests for this session
+	quiet           bool            // when true, suppress thinking and tool progress for this session
+	fromVoice       bool            // true if current turn originated from voice transcription
+	sideText        string
+	deleteMode      *deleteModeState
 }
 
 type deleteModeState struct {
@@ -1046,6 +1063,18 @@ func (e *Engine) handleMessage(p Platform, msg *Message) {
 				}
 			}
 		}
+		// Session is busy — try to queue the message for the running turn
+		// so the agent processes it immediately after the current turn ends.
+		if e.queueMessageForBusySession(p, msg, interactiveKey) {
+			// Race guard: the drain loop in processInteractiveMessageWith may
+			// have just finished (session unlocked) between our TryLock failure
+			// and the queue append. Re-try TryLock — if it succeeds, no one is
+			// draining the queue so we must start a processor ourselves.
+			if session.TryLock() {
+				go e.drainOrphanedQueue(session, interactiveKey, agent, resolvedWorkspace)
+			}
+			return
+		}
 		e.reply(p, msg.ReplyCtx, e.i18n.T(MsgPreviousProcessing))
 		return
 	}
@@ -1059,6 +1088,125 @@ func (e *Engine) handleMessage(p Platform, msg *Message) {
 	go e.processInteractiveMessageWith(p, msg, session, agent, interactiveKey, resolvedWorkspace)
 }
 
+// queueMessageForBusySession queues a message for later delivery when the
+// session is busy. The message is NOT sent to agent stdin at queue time;
+// the event loop sends it after the current turn's EventResult is received.
+// Returns true if the message was successfully queued, false otherwise.
+func (e *Engine) queueMessageForBusySession(p Platform, msg *Message, interactiveKey string) bool {
+	e.interactiveMu.Lock()
+	state, hasState := e.interactiveStates[interactiveKey]
+	e.interactiveMu.Unlock()
+
+	if !hasState || state == nil || state.agentSession == nil || !state.agentSession.Alive() {
+		return false
+	}
+
+	// Only queue metadata — do NOT send to agent stdin yet.
+	// The agent CLI may treat a mid-turn stdin message as part of the
+	// current turn, causing the event loop to hang waiting for a second
+	// EventResult that never arrives. Instead, the event loop sends the
+	// message after the current turn's EventResult is received.
+	state.mu.Lock()
+	if len(state.pendingMessages) >= maxQueuedMessages {
+		state.mu.Unlock()
+		return false // fall back to "previous processing" reply
+	}
+	state.pendingMessages = append(state.pendingMessages, queuedMessage{
+		platform:      p,
+		replyCtx:      msg.ReplyCtx,
+		content:       msg.Content,
+		images:        msg.Images,
+		files:         msg.Files,
+		fromVoice:     msg.FromVoice,
+		userID:        msg.UserID,
+		msgPlatform:   msg.Platform,
+		msgSessionKey: msg.SessionKey,
+	})
+	queueDepth := len(state.pendingMessages)
+	state.mu.Unlock()
+
+	slog.Info("message queued for busy session",
+		"session", msg.SessionKey,
+		"user", msg.UserName,
+		"queue_depth", queueDepth,
+	)
+	e.reply(p, msg.ReplyCtx, e.i18n.T(MsgMessageQueued))
+	return true
+}
+
+// drainOrphanedQueue is called when a message was queued but the drain loop
+// has already exited. It processes all pending messages in the state, similar
+// to the drain loop in processInteractiveMessageWith but as a standalone
+// goroutine.
+func (e *Engine) drainOrphanedQueue(session *Session, interactiveKey string, agent Agent, workspaceDir string) {
+	unlocked := false
+	defer func() {
+		if !unlocked {
+			session.Unlock()
+		}
+	}()
+
+	e.interactiveMu.Lock()
+	state, hasState := e.interactiveStates[interactiveKey]
+	e.interactiveMu.Unlock()
+
+	if !hasState || state == nil || state.agentSession == nil || !state.agentSession.Alive() {
+		// State disappeared — notify any queued senders that their messages are lost.
+		if hasState && state != nil {
+			e.notifyDroppedQueuedMessages(state, fmt.Errorf("agent session ended"))
+		}
+		return
+	}
+
+	for {
+		state.mu.Lock()
+		if len(state.pendingMessages) == 0 {
+			session.Unlock()
+			unlocked = true
+			state.mu.Unlock()
+			return
+		}
+		queued := state.pendingMessages[0]
+		state.pendingMessages = state.pendingMessages[1:]
+		state.platform = queued.platform
+		state.replyCtx = queued.replyCtx
+		state.fromVoice = queued.fromVoice
+		state.mu.Unlock()
+
+		e.i18n.DetectAndSet(queued.content)
+
+		queuedPrompt := queued.content
+		if e.injectSender && queued.userID != "" {
+			chatID := extractChannelID(queued.msgSessionKey)
+			queuedPrompt = fmt.Sprintf("[cc-connect sender_id=%s platform=%s chat_id=%s]\n%s", queued.userID, queued.msgPlatform, chatID, queued.content)
+		}
+
+		if state.agentSession == nil || !state.agentSession.Alive() {
+			e.send(queued.platform, queued.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), "agent session ended"))
+			e.notifyDroppedQueuedMessages(state, fmt.Errorf("agent session ended"))
+			return
+		}
+
+		drainEvents(state.agentSession.Events())
+
+		if err := state.agentSession.Send(queuedPrompt, queued.images, queued.files); err != nil {
+			slog.Error("failed to send orphaned queued message", "error", err, "session", interactiveKey)
+			e.send(queued.platform, queued.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), err))
+			e.notifyDroppedQueuedMessages(state, err)
+			return
+		}
+		session.AddHistory("user", queued.content)
+
+		var stopTyping func()
+		if ti, ok := queued.platform.(TypingIndicator); ok {
+			stopTyping = ti.StartTyping(e.ctx, queued.replyCtx)
+		}
+
+		slog.Info("processing orphaned queued message", "session", interactiveKey)
+		e.processInteractiveEvents(state, session, interactiveKey, "", time.Now(), stopTyping)
+	}
+}
+
 // ──────────────────────────────────────────────────────────────
 // Voice message handling
 // ──────────────────────────────────────────────────────────────
@@ -1322,7 +1470,16 @@ func (e *Engine) processInteractiveMessage(p Platform, msg *Message, session *Se
 // It accepts an explicit agent, interactiveKey (for the interactiveStates map),
 // and workspaceDir so that multi-workspace mode can route to per-workspace agents.
 func (e *Engine) processInteractiveMessageWith(p Platform, msg *Message, session *Session, agent Agent, interactiveKey string, workspaceDir string) {
-	defer session.Unlock()
+	// session.Unlock() is NOT deferred here — it is called explicitly in
+	// the drain loop below while holding state.mu to close the race window
+	// between "queue is empty" and "session unlocked". A deferred fallback
+	// ensures the lock is released on early-return paths.
+	unlocked := false
+	defer func() {
+		if !unlocked {
+			session.Unlock()
+		}
+	}()
 
 	if e.ctx.Err() != nil {
 		return
@@ -1358,12 +1515,16 @@ func (e *Engine) processInteractiveMessageWith(p Platform, msg *Message, session
 		return
 	}
 
-	// Start typing indicator if platform supports it
+	// Start typing indicator if platform supports it.
+	// Ownership is transferred to processInteractiveEvents which manages
+	// stopping/restarting it across queued message turns.
 	var stopTyping func()
 	if ti, ok := p.(TypingIndicator); ok {
 		stopTyping = ti.StartTyping(e.ctx, msg.ReplyCtx)
 	}
 	defer func() {
+		// Stop typing if ownership was NOT transferred to processInteractiveEvents
+		// (i.e. an early return before that call).
 		if stopTyping != nil {
 			stopTyping()
 		}
@@ -1390,6 +1551,13 @@ func (e *Engine) processInteractiveMessageWith(p Platform, msg *Message, session
 		slog.Error("failed to send prompt", "error", err)
 
 		if !state.agentSession.Alive() {
+			// Preserve queued messages before cleanup so we can migrate
+			// them to the replacement state after restart.
+			state.mu.Lock()
+			savedQueue := state.pendingMessages
+			state.pendingMessages = nil
+			state.mu.Unlock()
+
 			e.cleanupInteractiveState(interactiveKey, state)
 			e.send(p, msg.ReplyCtx, e.i18n.T(MsgSessionRestarting))
 
@@ -1399,6 +1567,12 @@ func (e *Engine) processInteractiveMessageWith(p Platform, msg *Message, session
 				state.workspaceDir = workspaceDir
 				state.mu.Unlock()
 			}
+			// Restore queued messages on the new state.
+			if len(savedQueue) > 0 {
+				state.mu.Lock()
+				state.pendingMessages = append(state.pendingMessages, savedQueue...)
+				state.mu.Unlock()
+			}
 			if state.agentSession == nil {
 				e.reply(p, msg.ReplyCtx, fmt.Sprintf(e.i18n.T(MsgError), "failed to restart agent session"))
 				return
@@ -1421,7 +1595,64 @@ func (e *Engine) processInteractiveMessageWith(p Platform, msg *Message, session
 		slog.Warn("slow agent send", "elapsed", elapsed, "session", msg.SessionKey, "content_len", len(msg.Content))
 	}
 
-	e.processInteractiveEvents(state, session, interactiveKey, msg.MessageID, turnStart)
+	e.processInteractiveEvents(state, session, interactiveKey, msg.MessageID, turnStart, stopTyping)
+	stopTyping = nil // ownership transferred; prevent defer from double-stopping
+
+	// Guard against a narrow race: a message may have been queued between
+	// processInteractiveEvents observing an empty queue and returning here
+	// (session is still locked, so handleMessage's TryLock fails and routes
+	// the message to queueMessageForBusySession). Drain any such orphans.
+	//
+	// When the queue is empty we unlock the session while still holding
+	// state.mu so that any message arriving after this point will find
+	// TryLock() succeeds and goes through the normal handleMessage path.
+	for {
+		state.mu.Lock()
+		if len(state.pendingMessages) == 0 {
+			session.Unlock()
+			unlocked = true
+			state.mu.Unlock()
+			return
+		}
+		queued := state.pendingMessages[0]
+		state.pendingMessages = state.pendingMessages[1:]
+		state.platform = queued.platform
+		state.replyCtx = queued.replyCtx
+		state.fromVoice = queued.fromVoice
+		state.mu.Unlock()
+
+		e.i18n.DetectAndSet(queued.content)
+
+		queuedPrompt := queued.content
+		if e.injectSender && queued.userID != "" {
+			chatID := extractChannelID(queued.msgSessionKey)
+			queuedPrompt = fmt.Sprintf("[cc-connect sender_id=%s platform=%s chat_id=%s]\n%s", queued.userID, queued.msgPlatform, chatID, queued.content)
+		}
+
+		// Abort if the agent process has exited — notify remaining queued senders.
+		if state.agentSession == nil || !state.agentSession.Alive() {
+			e.send(queued.platform, queued.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), "agent session ended"))
+			e.notifyDroppedQueuedMessages(state, fmt.Errorf("agent session ended"))
+			return
+		}
+
+		// Drain stale events before starting a new turn (same as processInteractiveMessageWith).
+		drainEvents(state.agentSession.Events())
+
+		if err := state.agentSession.Send(queuedPrompt, queued.images, queued.files); err != nil {
+			slog.Error("failed to send orphaned queued message", "error", err, "session", interactiveKey)
+			e.send(queued.platform, queued.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), err))
+			e.notifyDroppedQueuedMessages(state, err)
+			return
+		}
+		session.AddHistory("user", queued.content)
+
+		var newStopTyping func()
+		if ti, ok := queued.platform.(TypingIndicator); ok {
+			newStopTyping = ti.StartTyping(e.ctx, queued.replyCtx)
+		}
+		e.processInteractiveEvents(state, session, interactiveKey, msg.MessageID, time.Now(), newStopTyping)
+	}
 }
 
 // getOrCreateWorkspaceAgent returns (or creates) a per-workspace agent and session manager.
@@ -1603,6 +1834,11 @@ func (e *Engine) cleanupInteractiveState(sessionKey string, expected ...*interac
 	delete(e.interactiveStates, sessionKey)
 	e.interactiveMu.Unlock()
 
+	// Notify senders of any queued messages that will never be processed.
+	if ok && state != nil {
+		e.notifyDroppedQueuedMessages(state, fmt.Errorf("session reset"))
+	}
+
 	if ok && state != nil && state.agentSession != nil {
 		slog.Debug("cleanupInteractiveState: closing agent session", "session", sessionKey)
 		closeStart := time.Now()
@@ -1626,13 +1862,22 @@ func (e *Engine) cleanupInteractiveState(sessionKey string, expected ...*interac
 
 const defaultEventIdleTimeout = 2 * time.Hour
 
-func (e *Engine) processInteractiveEvents(state *interactiveState, session *Session, sessionKey string, msgID string, turnStart time.Time) {
+func (e *Engine) processInteractiveEvents(state *interactiveState, session *Session, sessionKey string, msgID string, turnStart time.Time, stopTypingFn func()) {
 	var textParts []string
 	var segmentStart int // index into textParts: text before this has been sent/displayed
 	toolCount := 0
 	waitStart := time.Now()
 	firstEventLogged := false
 
+	// stopTyping tracks the current turn's typing indicator so it can be
+	// stopped when a queued message starts a new turn.
+	stopTyping := stopTypingFn
+	defer func() {
+		if stopTyping != nil {
+			stopTyping()
+		}
+	}()
+
 	state.mu.Lock()
 	sp := newStreamPreview(e.streamPreview, state.platform, state.replyCtx, e.ctx)
 	state.mu.Unlock()
@@ -1941,6 +2186,98 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess
 				}
 			}
 
+			// Check for queued messages — if present, continue the event loop
+			// for the next turn instead of returning.
+			state.mu.Lock()
+			if len(state.pendingMessages) > 0 {
+				queued := state.pendingMessages[0]
+				state.pendingMessages = state.pendingMessages[1:]
+				remainingQueue := len(state.pendingMessages)
+				state.platform = queued.platform
+				state.replyCtx = queued.replyCtx
+				state.fromVoice = queued.fromVoice
+				state.mu.Unlock()
+
+				// Stop the previous turn's typing indicator
+				if stopTyping != nil {
+					stopTyping()
+					stopTyping = nil
+				}
+				// Start a new typing indicator for the queued message's context
+				if ti, ok := queued.platform.(TypingIndicator); ok {
+					stopTyping = ti.StartTyping(e.ctx, queued.replyCtx)
+				}
+
+				// Drain stale events before starting the next turn. Between
+				// EventResult and Send(), the only buffered events would be
+				// stale leftovers (e.g. a deferred EventError from cmd.Wait()).
+				drainEvents(state.agentSession.Events())
+
+				// Build prompt content for the queued message
+				queuedPrompt := queued.content
+				if e.injectSender && queued.userID != "" {
+					chatID := extractChannelID(queued.msgSessionKey)
+					queuedPrompt = fmt.Sprintf("[cc-connect sender_id=%s platform=%s chat_id=%s]\n%s", queued.userID, queued.msgPlatform, chatID, queued.content)
+				}
+
+				// NOW send the queued message to agent stdin (not at queue time).
+				if err := state.agentSession.Send(queuedPrompt, queued.images, queued.files); err != nil {
+					slog.Error("failed to send queued message to agent", "error", err, "session", sessionKey)
+
+					// Send failed — stop typing, notify user, drain remaining queue.
+					// We intentionally do NOT attempt to restart the session here:
+					// the restart path introduces complex state management issues
+					// (stale session pointers, FIFO ordering, workspace agent
+					// recreation races). The user's next message will trigger a
+					// normal session restart via processInteractiveMessageWith.
+					if stopTyping != nil {
+						stopTyping()
+						stopTyping = nil
+					}
+					e.send(queued.platform, queued.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), err))
+					state.mu.Lock()
+					remaining := state.pendingMessages
+					state.pendingMessages = nil
+					state.mu.Unlock()
+					for _, q := range remaining {
+						e.send(q.platform, q.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), err))
+					}
+					return
+				}
+
+				// Detect language now (deferred from queue time to avoid
+				// flipping locale while the previous turn is still running).
+				e.i18n.DetectAndSet(queued.content)
+
+				// Reset per-turn state for the next turn
+				textParts = nil
+				segmentStart = 0
+				toolCount = 0
+				turnStart = time.Now()
+				firstEventLogged = false
+				waitStart = time.Now()
+				sp = newStreamPreview(e.streamPreview, queued.platform, queued.replyCtx, e.ctx)
+
+				session.AddHistory("user", queued.content)
+
+				if idleTimer != nil {
+					if !idleTimer.Stop() {
+						select {
+						case <-idleTimer.C:
+						default:
+						}
+					}
+					idleTimer.Reset(e.eventIdleTimeout)
+				}
+
+				slog.Info("processing queued message",
+					"session", sessionKey,
+					"remaining_queue", remainingQueue,
+				)
+				continue
+			}
+			state.mu.Unlock()
+
 			return
 
 		case EventError:
@@ -1949,6 +2286,12 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess
 				slog.Error("agent error", "error", event.Error)
 				e.send(p, replyCtx, fmt.Sprintf(e.i18n.T(MsgError), event.Error))
 			}
+			// Only drop queued messages if the agent session is dead.
+			// Some agents (e.g. Codex) emit EventError for per-turn failures
+			// while keeping the session alive for subsequent turns.
+			if state.agentSession == nil || !state.agentSession.Alive() {
+				e.notifyDroppedQueuedMessages(state, event.Error)
+			}
 			return
 		}
 	}
@@ -1956,6 +2299,7 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess
 channelClosed:
 	// Channel closed - process exited unexpectedly
 	slog.Warn("agent process exited", "session_key", sessionKey)
+	e.notifyDroppedQueuedMessages(state, fmt.Errorf("agent process exited"))
 	e.cleanupInteractiveState(sessionKey, state)
 
 	if len(textParts) > 0 {
@@ -1987,6 +2331,20 @@ channelClosed:
 	}
 }
 
+// notifyDroppedQueuedMessages drains pendingMessages from the state and
+// sends an error notification to each queued message's sender. Called when
+// the event loop exits abnormally (EventError, channel closed) and queued
+// messages can no longer be delivered to the agent.
+func (e *Engine) notifyDroppedQueuedMessages(state *interactiveState, reason error) {
+	state.mu.Lock()
+	remaining := state.pendingMessages
+	state.pendingMessages = nil
+	state.mu.Unlock()
+	for _, q := range remaining {
+		e.send(q.platform, q.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), reason))
+	}
+}
+
 // ──────────────────────────────────────────────────────────────
 // Command handling
 // ──────────────────────────────────────────────────────────────
@@ -4131,7 +4489,15 @@ func (e *Engine) cmdCompress(p Platform, msg *Message) {
 	e.send(p, msg.ReplyCtx, e.i18n.T(MsgCompressing))
 
 	go func() {
-		defer session.Unlock()
+		// session.Unlock() is called inside drainQueuedMessagesAfterCompress
+		// while holding state.mu to close the race window. Deferred fallback
+		// ensures the lock is released on early-return paths.
+		compressUnlocked := false
+		defer func() {
+			if !compressUnlocked {
+				session.Unlock()
+			}
+		}()
 
 		state.mu.Lock()
 		state.platform = p
@@ -4149,14 +4515,15 @@ func (e *Engine) cmdCompress(p Platform, msg *Message) {
 			return
 		}
 
-		e.processCompressEvents(state, msg.SessionKey, p, msg.ReplyCtx)
+		e.processCompressEvents(state, session, msg.SessionKey, p, msg.ReplyCtx, &compressUnlocked)
 	}()
 }
 
 // processCompressEvents drains agent events after a compress command.
 // Unlike processInteractiveEvents it does NOT record history and treats
 // an empty result as success rather than "(empty response)".
-func (e *Engine) processCompressEvents(state *interactiveState, sessionKey string, p Platform, replyCtx any) {
+func (e *Engine) processCompressEvents(state *interactiveState, session *Session, sessionKey string, p Platform, replyCtx any, unlocked *bool) {
+
 	var textParts []string
 	events := state.agentSession.Events()
 
@@ -4181,11 +4548,13 @@ func (e *Engine) processCompressEvents(state *interactiveState, sessionKey strin
 				} else {
 					e.reply(p, replyCtx, e.i18n.T(MsgCompressDone))
 				}
+				e.notifyDroppedQueuedMessages(state, fmt.Errorf("agent process exited during compress"))
 				return
 			}
 		case <-idleCh:
 			e.send(p, replyCtx, fmt.Sprintf(e.i18n.T(MsgError), "compress timed out"))
 			e.cleanupInteractiveState(sessionKey, state)
+			e.notifyDroppedQueuedMessages(state, fmt.Errorf("compress timed out"))
 			return
 		case <-e.ctx.Done():
 			return
@@ -4216,11 +4585,22 @@ func (e *Engine) processCompressEvents(state *interactiveState, sessionKey strin
 			} else {
 				e.reply(p, replyCtx, e.i18n.T(MsgCompressDone))
 			}
+
+			// After compress succeeds, process any queued messages instead of dropping them.
+			e.drainQueuedMessagesAfterCompress(state, session, sessionKey, unlocked)
 			return
 		case EventError:
 			if event.Error != nil {
 				e.reply(p, replyCtx, fmt.Sprintf(e.i18n.T(MsgError), event.Error))
 			}
+			// Only drop queued messages if the agent is dead; some agents
+			// emit per-turn EventError while staying alive.
+			if !state.agentSession.Alive() {
+				e.notifyDroppedQueuedMessages(state, event.Error)
+			} else {
+				// Agent survived — try to process queued messages.
+				e.drainQueuedMessagesAfterCompress(state, session, sessionKey, unlocked)
+			}
 			return
 		case EventPermissionRequest:
 			_ = state.agentSession.RespondPermission(event.RequestID, PermissionResult{
@@ -4231,6 +4611,55 @@ func (e *Engine) processCompressEvents(state *interactiveState, sessionKey strin
 	}
 }
 
+// drainQueuedMessagesAfterCompress processes any messages that were queued
+// during a /compress operation. It sends each one to the agent and runs the
+// full interactive event loop for it.
+func (e *Engine) drainQueuedMessagesAfterCompress(state *interactiveState, session *Session, sessionKey string, unlocked *bool) {
+	for {
+		state.mu.Lock()
+		if len(state.pendingMessages) == 0 {
+			// Unlock session while holding state.mu to close the race window.
+			session.Unlock()
+			*unlocked = true
+			state.mu.Unlock()
+			return
+		}
+		queued := state.pendingMessages[0]
+		state.pendingMessages = state.pendingMessages[1:]
+		state.platform = queued.platform
+		state.replyCtx = queued.replyCtx
+		state.fromVoice = queued.fromVoice
+		state.mu.Unlock()
+
+		// Drain stale events from the previous turn (compress or prior queued
+		// turn) to avoid feeding leftover EventError/exit events into this turn.
+		drainEvents(state.agentSession.Events())
+
+		// Build prompt
+		queuedPrompt := queued.content
+		if e.injectSender && queued.userID != "" {
+			chatID := extractChannelID(queued.msgSessionKey)
+			queuedPrompt = fmt.Sprintf("[cc-connect sender_id=%s platform=%s chat_id=%s]\n%s", queued.userID, queued.msgPlatform, chatID, queued.content)
+		}
+
+		if err := state.agentSession.Send(queuedPrompt, queued.images, queued.files); err != nil {
+			slog.Error("failed to send queued message after compress", "error", err, "session", sessionKey)
+			e.send(queued.platform, queued.replyCtx, fmt.Sprintf(e.i18n.T(MsgError), err))
+			e.notifyDroppedQueuedMessages(state, err)
+			return
+		}
+
+		session.AddHistory("user", queued.content)
+
+		slog.Info("processing queued message after compress",
+			"session", sessionKey,
+		)
+
+		// Run the full event loop for this queued turn.
+		e.processInteractiveEvents(state, session, sessionKey, "", time.Now(), nil)
+	}
+}
+
 func (e *Engine) cmdAllow(p Platform, msg *Message, args []string) {
 	if len(args) == 0 {
 		if auth, ok := e.agent.(ToolAuthorizer); ok {

core/engine_test.go

@@ -369,7 +369,7 @@ func TestProcessInteractiveEvents_SuppressesDuplicateSideChannelText(t *testing.
 	}
 
 	agentSession.events <- Event{Type: EventResult, Content: sideText, Done: true}
-	e.processInteractiveEvents(state, session, sessionKey, "m1", time.Now())
+	e.processInteractiveEvents(state, session, sessionKey, "m1", time.Now(), nil)
 
 	if got := p.getSent(); len(got) != 1 || got[0] != sideText {
 		t.Fatalf("sent text = %#v, want one side-channel message", got)
@@ -399,7 +399,7 @@ func TestProcessInteractiveEvents_DoesNotSuppressDifferentFinalText(t *testing.T
 
 	finalText := "文件已发出，另外我也把使用方法整理好了。"
 	agentSession.events <- Event{Type: EventResult, Content: finalText, Done: true}
-	e.processInteractiveEvents(state, session, sessionKey, "m1", time.Now())
+	e.processInteractiveEvents(state, session, sessionKey, "m1", time.Now(), nil)
 
 	if got := p.getSent(); len(got) != 2 || got[0] == got[1] {
 		t.Fatalf("sent text = %#v, want side-channel and final reply", got)
@@ -3457,6 +3457,173 @@ func TestDrainEventsOpenChannel(t *testing.T) {
 	}
 }
 
+// --- Message queuing tests ---
+
+// queuingAgentSession records Send calls and emits events via a controllable channel.
+type queuingAgentSession struct {
+	controllableAgentSession
+	sendCalls []string
+	sendMu    sync.Mutex
+}
+
+func newQueuingSession(id string) *queuingAgentSession {
+	return &queuingAgentSession{
+		controllableAgentSession: controllableAgentSession{
+			sessionID: id,
+			alive:     true,
+			events:    make(chan Event, 16),
+			closed:    make(chan struct{}),
+		},
+	}
+}
+
+func (s *queuingAgentSession) Send(prompt string, _ []ImageAttachment, _ []FileAttachment) error {
+	s.sendMu.Lock()
+	s.sendCalls = append(s.sendCalls, prompt)
+	s.sendMu.Unlock()
+	return nil
+}
+
+func TestQueueMessageForBusySession_FIFODequeue(t *testing.T) {
+	p := &stubPlatformEngine{n: "test"}
+	sess := newQueuingSession("qs1")
+	agent := &controllableAgent{nextSession: sess}
+	e := NewEngine("test", agent, []Platform{p}, "", LangEnglish)
+
+	key := "test:user1"
+
+	// Set up an interactive state as if a turn is in progress.
+	state := &interactiveState{
+		agentSession: sess,
+		platform:     p,
+		replyCtx:     "ctx1",
+	}
+	e.interactiveMu.Lock()
+	e.interactiveStates[key] = state
+	e.interactiveMu.Unlock()
+
+	// Queue two messages while the session is "busy".
+	msg1 := &Message{SessionKey: key, Content: "msg1", ReplyCtx: "ctx-msg1"}
+	msg2 := &Message{SessionKey: key, Content: "msg2", ReplyCtx: "ctx-msg2"}
+
+	ok1 := e.queueMessageForBusySession(p, msg1, key)
+	ok2 := e.queueMessageForBusySession(p, msg2, key)
+
+	if !ok1 || !ok2 {
+		t.Fatal("expected both messages to be queued successfully")
+	}
+
+	// Since deferred-send, messages are NOT sent to agent stdin at queue
+	// time — only metadata is stored. Verify no Send calls occurred.
+	sess.sendMu.Lock()
+	if len(sess.sendCalls) != 0 {
+		t.Fatalf("sendCalls = %v, want [] (deferred send)", sess.sendCalls)
+	}
+	sess.sendMu.Unlock()
+
+	// Verify pending messages queue has correct FIFO order.
+	state.mu.Lock()
+	if len(state.pendingMessages) != 2 {
+		t.Fatalf("pendingMessages len = %d, want 2", len(state.pendingMessages))
+	}
+	if state.pendingMessages[0].content != "msg1" || state.pendingMessages[1].content != "msg2" {
+		t.Fatalf("pendingMessages = [%s, %s], want [msg1, msg2]",
+			state.pendingMessages[0].content, state.pendingMessages[1].content)
+	}
+	state.mu.Unlock()
+}
+
+func TestProcessInteractiveEvents_DrainsQueuedMessages(t *testing.T) {
+	p := &stubPlatformEngine{n: "test"}
+	sess := newQueuingSession("qs2")
+	agent := &controllableAgent{nextSession: sess}
+	e := NewEngine("test", agent, []Platform{p}, "", LangEnglish)
+
+	key := "test:user1"
+	session := e.sessions.GetOrCreateActive(key)
+
+	// Pre-populate the interactive state with one queued message.
+	state := &interactiveState{
+		agentSession: sess,
+		platform:     p,
+		replyCtx:     "ctx-turn1",
+		pendingMessages: []queuedMessage{
+			{platform: p, replyCtx: "ctx-turn2", content: "queued-msg"},
+		},
+	}
+	e.interactiveMu.Lock()
+	e.interactiveStates[key] = state
+	e.interactiveMu.Unlock()
+
+	// Simulate the agent completing turn 1 then turn 2.
+	// Turn 2 events are pushed only after Send() is called for the queued
+	// message, matching real-world timing where the agent doesn't produce
+	// events for a turn until it receives the prompt on stdin.
+	go func() {
+		// Turn 1 result
+		sess.events <- Event{Type: EventText, Content: "response1"}
+		sess.events <- Event{Type: EventResult, Content: "response1", Done: true}
+		// Wait for the queued message's Send() call before pushing turn 2 events.
+		sess.sendMu.Lock()
+		for len(sess.sendCalls) == 0 {
+			sess.sendMu.Unlock()
+			time.Sleep(5 * time.Millisecond)
+			sess.sendMu.Lock()
+		}
+		sess.sendMu.Unlock()
+		// Turn 2 result (for the queued message)
+		sess.events <- Event{Type: EventText, Content: "response2"}
+		sess.events <- Event{Type: EventResult, Content: "response2", Done: true}
+	}()
+
+	session.AddHistory("user", "initial-msg")
+
+	// processInteractiveEvents should handle both turns.
+	done := make(chan struct{})
+	go func() {
+		e.processInteractiveEvents(state, session, key, "msg1", time.Now(), nil)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// ok
+	case <-time.After(5 * time.Second):
+		t.Fatal("processInteractiveEvents did not complete in time")
+	}
+
+	// Verify queue is empty after processing.
+	state.mu.Lock()
+	remaining := len(state.pendingMessages)
+	state.mu.Unlock()
+	if remaining != 0 {
+		t.Fatalf("pendingMessages after processing = %d, want 0", remaining)
+	}
+
+	// Verify both turns recorded in session history.
+	history := session.GetHistory(100)
+	var assistantMsgs []string
+	for _, h := range history {
+		if h.Role == "assistant" {
+			assistantMsgs = append(assistantMsgs, h.Content)
+		}
+	}
+	if len(assistantMsgs) != 2 {
+		t.Fatalf("assistant history entries = %d, want 2", len(assistantMsgs))
+	}
+
+	// Verify the queued message was also added to history.
+	var userMsgs []string
+	for _, h := range history {
+		if h.Role == "user" {
+			userMsgs = append(userMsgs, h.Content)
+		}
+	}
+	if len(userMsgs) < 2 {
+		t.Fatalf("user history entries = %d, want >= 2", len(userMsgs))
+	}
+}
+
 // ── executeCardAction interactiveKey tests ───────────────────
 
 func TestExecuteCardAction_QuietUsesInteractiveKey(t *testing.T) {

core/i18n.go

@@ -125,6 +125,7 @@ const (
 	MsgExecutionStopped     MsgKey = "execution_stopped"
 	MsgNoExecution          MsgKey = "no_execution"
 	MsgPreviousProcessing   MsgKey = "previous_processing"
+	MsgMessageQueued        MsgKey = "message_queued"
 	MsgNoToolsAllowed       MsgKey = "no_tools_allowed"
 	MsgCurrentTools         MsgKey = "current_tools"
 	MsgCurrentSession       MsgKey = "current_session"
@@ -528,6 +529,13 @@ var messages = map[MsgKey]map[Language]string{
 		LangJapanese:           "⏳ 前のリクエストを処理中です。`/btw <メッセージ>` で現在のターンにコンテキストを追加できます。",
 		LangSpanish:            "⏳ La solicitud anterior aún se está procesando. Use `/btw <mensaje>` para agregar contexto al turno actual.",
 	},
+	MsgMessageQueued: {
+		LangEnglish:            "📬 Message received — will process after the current task finishes.",
+		LangChinese:            "📬 消息已收到，将在当前任务完成后处理。",
+		LangTraditionalChinese: "📬 訊息已收到，將在目前任務完成後處理。",
+		LangJapanese:           "📬 メッセージを受信しました。現在のタスク完了後に処理します。",
+		LangSpanish:            "📬 Mensaje recibido — se procesará después de que termine la tarea actual.",
+	},
 	MsgNoToolsAllowed: {
 		LangEnglish:            "No tools pre-allowed.\nUsage: `/allow <tool_name>`\nExample: `/allow Bash`",
 		LangChinese:            "尚未预授权任何工具。\n用法: `/allow <工具名>`\n示例: `/allow Bash`",