From 350a4d8b2d1e95402cbace2dc7bc4dfdaf707d8e Mon Sep 17 00:00:00 2001
From: chenhg5 <chenhonggui@wps.cn>
Date: Mon, 9 Mar 2026 19:12:24 +0800
Subject: [PATCH] fix(engine): prevent stale event processing in agent sessions

Add drainEvents utility to clear buffered events between turns and apply it in processInteractiveMessage and cmdCompress to avoid processing stale events from previous agent interactions.

generated by llmgit

Co-Authored-By: Claude <noreply@anthropic.com>
---
 core/engine.go | 25 +++++++++++++++++++++++++
 1 file changed, 25 insertions(+)

diff --git a/core/engine.go b/core/engine.go
index 6f18d76..0ecd310 100644
--- a/core/engine.go
+++ b/core/engine.go
@@ -792,6 +792,11 @@ func (e *Engine) processInteractiveMessage(p Platform, msg *Message, session *Se
 		}
 	}()
 
+	// Drain any stale events left in the channel from a previous turn.
+	// This prevents the next processInteractiveEvents from reading an old
+	// EventResult that was pushed after the previous turn already returned.
+	drainEvents(state.agentSession.Events())
+
 	sendStart := time.Now()
 	if err := state.agentSession.Send(msg.Content, msg.Images); err != nil {
 		slog.Error("failed to send prompt", "error", err)
@@ -2301,6 +2306,8 @@ func (e *Engine) cmdCompress(p Platform, msg *Message) {
 		state.replyCtx = msg.ReplyCtx
 		state.mu.Unlock()
 
+		drainEvents(state.agentSession.Events())
+
 		cmd := compressor.CompressCommand()
 		if err := state.agentSession.Send(cmd, nil); err != nil {
 			e.reply(p, msg.ReplyCtx, fmt.Sprintf(e.i18n.T(MsgError), err))
@@ -2718,6 +2725,24 @@ func (e *Engine) send(p Platform, replyCtx any, content string) {
 	}
 }
 
+// drainEvents discards any buffered events from the channel.
+// Called before a new turn to prevent stale events from a previous turn's
+// agent process from being mistaken for the new turn's response.
+func drainEvents(ch <-chan Event) {
+	drained := 0
+	for {
+		select {
+		case <-ch:
+			drained++
+		default:
+			if drained > 0 {
+				slog.Warn("drained stale events from previous turn", "count", drained)
+			}
+			return
+		}
+	}
+}
+
 // reply wraps p.Reply with error logging and slow-operation warnings.
 func (e *Engine) reply(p Platform, replyCtx any, content string) {
 	start := time.Now()