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>
This commit is contained in:
chenhg5andClaude committed 2026-03-09 22:40:14 +08:00
1 parent 44f2dba697
commit 350a4d8b2d
1 file changed
+25
+25
View File
@@ -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()