From 4089f78a26e2cc900a8d4f1270cb4e248ec75f87 Mon Sep 17 00:00:00 2001 From: 0xsegfaulted <37236264+0xsegfaulted@users.noreply.github.com> Date: Mon, 16 Mar 2026 10:49:21 +0800 Subject: [PATCH] feat: queue messages when agent is busy instead of discarding them When a user sends a message while the agent session is busy processing, instead of discarding it with "please wait", the message is now queued. After the current turn completes, queued messages are sent to the agent and processed in FIFO order. This enables users to interrupt or follow up without losing messages. Key design decisions: - Messages are queued as metadata only (deferred send) to avoid mid-turn stdin interference that causes hung sessions - If the agent dies between turns, the session is restarted before retrying queued messages - Typing indicators are properly managed across queued turns - Non-Claude-Code agents fall back to the previous "please wait" behavior Co-Authored-By: Claude Opus 4.6 (1M context) --- core/engine.go | 465 ++++++++++++++++++++++++++++++++++++++++++-- core/engine_test.go | 171 +++++++++++++++- core/i18n.go | 8 + 3 files changed, 624 insertions(+), 20 deletions(-) diff --git a/core/engine.go b/core/engine.go index 14a52d4..6eb823a 100644 --- a/core/engine.go +++ b/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 { @@ -1034,6 +1051,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 } @@ -1047,6 +1076,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 // ────────────────────────────────────────────────────────────── @@ -1310,7 +1458,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 @@ -1346,12 +1503,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() } @@ -1378,6 +1539,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)) @@ -1387,6 +1555,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 @@ -1409,7 +1583,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. @@ -1589,6 +1820,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() @@ -1612,13 +1848,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() @@ -1927,6 +2172,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: @@ -1935,6 +2272,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 } } @@ -1942,6 +2285,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 { @@ -1973,6 +2317,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 // ────────────────────────────────────────────────────────────── @@ -4087,7 +4445,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 @@ -4105,14 +4471,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() @@ -4137,11 +4504,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 @@ -4172,11 +4541,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{ @@ -4187,6 +4567,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 { diff --git a/core/engine_test.go b/core/engine_test.go index e104b7f..0aeab17 100644 --- a/core/engine_test.go +++ b/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) @@ -3391,6 +3391,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) { diff --git a/core/i18n.go b/core/i18n.go index 1ece909..a32498e 100644 --- a/core/i18n.go +++ b/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" @@ -516,6 +517,13 @@ var messages = map[MsgKey]map[Language]string{ LangJapanese: "⏳ 前のリクエストを処理中です。`/btw <メッセージ>` で現在のターンにコンテキストを追加できます。", LangSpanish: "⏳ La solicitud anterior aún se está procesando. Use `/btw ` 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 `\nExample: `/allow Bash`", LangChinese: "尚未预授权任何工具。\n用法: `/allow <工具名>`\n示例: `/allow Bash`",