diff --git a/core/engine.go b/core/engine.go index 1aba81e..dc710d2 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 { @@ -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 { diff --git a/core/engine_test.go b/core/engine_test.go index 0b9fc1e..a8560b1 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) @@ -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) { diff --git a/core/i18n.go b/core/i18n.go index 52f0e08..3bf6880 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" @@ -528,6 +529,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`",