fix(feishu): drop stale redelivered user messages by create_time watermark (#1168)

Feishu may redeliver messages with a new message_id but the original
create_time after WS disconnect. Track a per-session watermark from
completed, in-flight, and queued user messages so older redeliveries
are discarded before processing or enqueueing.

Co-authored-by: Cursor <cursoragent@cursor.com>

core/engine.go

@@ -305,6 +305,7 @@ type queuedMessage struct {
 	msgPlatform       string // platform name for sender injection
 	msgSessionKey     string // session key for extracting chat ID
 	channelKey        string // platform-provided channel identifier (preferred over sessionKey extraction)
+	userMessageTimeMs int64  // Feishu create_time ms (optional); see Message.UserMessageTimeMs
 }
 
 // interactiveState tracks a running interactive agent session and its permission state.
@@ -340,6 +341,103 @@ type interactiveState struct {
 	// the next turn (e.g. after an abnormal exit). Defaults to true (safe);
 	// cleared to false only after a clean EventResult.
 	eventsNeedResync bool
+
+	// lastCompletedUserMessageTimeMs is the max platform user-message create time
+	// (ms) for which an agent turn has finished with EventResult.
+	lastCompletedUserMessageTimeMs int64
+	// currentTurnUserMessageTimeMs is the UserMessageTimeMs for the in-flight
+	// foreground turn (including a queued turn after EventResult).
+	currentTurnUserMessageTimeMs int64
+}
+
+// latestUserMessageWatermarkLocked returns the highest UserMessageTimeMs among
+// completed turns, the in-flight turn, and queued messages for this session.
+// state.mu must be held by the caller.
+func latestUserMessageWatermarkLocked(state *interactiveState) int64 {
+	if state == nil {
+		return 0
+	}
+	wm := state.lastCompletedUserMessageTimeMs
+	if state.currentTurnUserMessageTimeMs > wm {
+		wm = state.currentTurnUserMessageTimeMs
+	}
+	for _, q := range state.pendingMessages {
+		if q.userMessageTimeMs > wm {
+			wm = q.userMessageTimeMs
+		}
+	}
+	return wm
+}
+
+type userMessageWatermarkSnapshot struct {
+	maxMs       int64
+	completedMs int64
+	inFlightMs  int64
+	queuedMaxMs int64
+}
+
+// userMessageWatermarkSnapshotLocked captures the per-source watermark values.
+// state.mu must be held by the caller.
+func userMessageWatermarkSnapshotLocked(state *interactiveState) userMessageWatermarkSnapshot {
+	if state == nil {
+		return userMessageWatermarkSnapshot{}
+	}
+	snap := userMessageWatermarkSnapshot{
+		completedMs: state.lastCompletedUserMessageTimeMs,
+		inFlightMs:  state.currentTurnUserMessageTimeMs,
+	}
+	for _, q := range state.pendingMessages {
+		if q.userMessageTimeMs > snap.queuedMaxMs {
+			snap.queuedMaxMs = q.userMessageTimeMs
+		}
+	}
+	snap.maxMs = latestUserMessageWatermarkLocked(state)
+	return snap
+}
+
+// acceptReason describes which accepted message makes an older create_time stale.
+func (s userMessageWatermarkSnapshot) acceptReason() string {
+	if s.maxMs <= 0 {
+		return "none"
+	}
+	// Prefer the most specific active state when multiple sources tie.
+	if s.inFlightMs == s.maxMs && s.inFlightMs > 0 {
+		return "processing"
+	}
+	if s.queuedMaxMs == s.maxMs && s.queuedMaxMs > 0 {
+		return "queued"
+	}
+	if s.completedMs == s.maxMs && s.completedMs > 0 {
+		return "completed"
+	}
+	return "unknown"
+}
+
+func (e *Engine) logStaleUserMessageDropped(action string, msg *Message, interactiveKey string, snap userMessageWatermarkSnapshot) {
+	if msg == nil {
+		return
+	}
+	slog.Info("stale user message dropped: received from platform but create_time older than accepted watermark",
+		"action", action,
+		"platform", msg.Platform,
+		"session", msg.SessionKey,
+		"interactive_key", interactiveKey,
+		"msg_id", msg.MessageID,
+		"msg_create_time_ms", msg.UserMessageTimeMs,
+		"watermark_ms", snap.maxMs,
+		"watermark_reason", snap.acceptReason(),
+		"watermark_processing_ms", snap.inFlightMs,
+		"watermark_queued_max_ms", snap.queuedMaxMs,
+		"watermark_completed_ms", snap.completedMs,
+	)
+	slog.Debug("stale user message dropped detail",
+		"action", action,
+		"msg_id", msg.MessageID,
+		"msg_create_time_ms", msg.UserMessageTimeMs,
+		"watermark_ms", snap.maxMs,
+		"watermark_reason", snap.acceptReason(),
+		"content_len", len(msg.Content),
+	)
 }
 
 type pendingProviderAddState struct {
@@ -1847,6 +1945,74 @@ func (e *Engine) removeQueuedMessageByID(messageID string) (string, bool) {
 	return "", false
 }
 
+// isStaleUserMessageLocked reports whether timeMs is strictly older than the
+// latest user message already accepted for this session (completed, in-flight,
+// or queued). state.mu must be held by the caller. timeMs <= 0 is never stale.
+func (e *Engine) isStaleUserMessageLocked(state *interactiveState, timeMs int64) bool {
+	if timeMs <= 0 {
+		return false
+	}
+	wm := latestUserMessageWatermarkLocked(state)
+	return wm > 0 && timeMs < wm
+}
+
+// noteUserTurnCompleted advances lastCompletedUserMessageTimeMs after an
+// agent turn ends with EventResult.
+func (e *Engine) noteUserTurnCompleted(state *interactiveState) {
+	state.mu.Lock()
+	t := state.currentTurnUserMessageTimeMs
+	if t > 0 && t > state.lastCompletedUserMessageTimeMs {
+		state.lastCompletedUserMessageTimeMs = t
+	}
+	state.mu.Unlock()
+}
+
+// noteUserMessageAccepted records that a user message was accepted for this
+// session (direct processing). Queued messages update the watermark via
+// pendingMessages instead. This closes the window before processInteractiveMessageWith
+// starts where a redelivery could slip through while the session lock is held.
+func (e *Engine) noteUserMessageAccepted(interactiveKey string, timeMs int64) {
+	if timeMs <= 0 {
+		return
+	}
+	e.interactiveMu.Lock()
+	state, ok := e.interactiveStates[interactiveKey]
+	e.interactiveMu.Unlock()
+	if !ok || state == nil {
+		return
+	}
+	state.mu.Lock()
+	if timeMs > state.currentTurnUserMessageTimeMs {
+		state.currentTurnUserMessageTimeMs = timeMs
+	}
+	state.mu.Unlock()
+}
+
+// discardStaleUserMessageIfNeeded returns true when the message is dropped
+// because a newer user message is already in progress, queued, or completed
+// for this interactive session (e.g. Feishu redelivery with a new message_id
+// but an older create_time).
+func (e *Engine) discardStaleUserMessageIfNeeded(interactiveKey string, msg *Message) bool {
+	if msg == nil || msg.UserMessageTimeMs <= 0 {
+		return false
+	}
+	e.interactiveMu.Lock()
+	state, ok := e.interactiveStates[interactiveKey]
+	e.interactiveMu.Unlock()
+	if !ok || state == nil {
+		return false
+	}
+	state.mu.Lock()
+	stale := e.isStaleUserMessageLocked(state, msg.UserMessageTimeMs)
+	snap := userMessageWatermarkSnapshotLocked(state)
+	state.mu.Unlock()
+	if !stale {
+		return false
+	}
+	e.logStaleUserMessageDropped("reject_before_handle", msg, interactiveKey, snap)
+	return true
+}
+
 func (e *Engine) stopCurrentMessageIfRecalled(sessionKey string) bool {
 	e.interactiveMu.Lock()
 	state, ok := e.interactiveStates[sessionKey]
@@ -2125,6 +2291,10 @@ func (e *Engine) handleMessage(p Platform, msg *Message) {
 		return
 	}
 
+	if e.discardStaleUserMessageIfNeeded(interactiveKey, msg) {
+		return
+	}
+
 	session := sessions.GetOrCreateActive(msg.SessionKey)
 	sessions.UpdateUserMeta(msg.SessionKey, msg.UserName, msg.ChatName)
 	if !session.TryLock() {
@@ -2165,6 +2335,14 @@ sessionLocked:
 	// processor so messages arriving during session startup can be queued
 	// instead of dropped (issue #565).
 	e.ensureInteractiveStateForQueueing(interactiveKey, p, msg.ReplyCtx)
+	e.noteUserMessageAccepted(interactiveKey, msg.UserMessageTimeMs)
+	slog.Debug("user message accepted for processing",
+		"platform", msg.Platform,
+		"session", msg.SessionKey,
+		"interactive_key", interactiveKey,
+		"msg_id", msg.MessageID,
+		"msg_create_time_ms", msg.UserMessageTimeMs,
+	)
 
 	slog.Info("processing message",
 		"platform", msg.Platform,
@@ -2256,6 +2434,12 @@ func (e *Engine) queueMessageForBusySession(p Platform, msg *Message, interactiv
 	// EventResult that never arrives. Instead, the event loop sends the
 	// message after the current turn's EventResult is received.
 	state.mu.Lock()
+	if e.isStaleUserMessageLocked(state, msg.UserMessageTimeMs) {
+		snap := userMessageWatermarkSnapshotLocked(state)
+		state.mu.Unlock()
+		e.logStaleUserMessageDropped("reject_before_queue", msg, interactiveKey, snap)
+		return true
+	}
 	if len(state.pendingMessages) >= e.maxQueuedMessages {
 		depth := len(state.pendingMessages)
 		state.mu.Unlock()
@@ -2275,10 +2459,19 @@ func (e *Engine) queueMessageForBusySession(p Platform, msg *Message, interactiv
 		msgPlatform:       msg.Platform,
 		msgSessionKey:     msg.SessionKey,
 		channelKey:        msg.ChannelKey,
+		userMessageTimeMs: msg.UserMessageTimeMs,
 	})
 	queueDepth := len(state.pendingMessages)
 	state.mu.Unlock()
 
+	slog.Debug("user message accepted into queue",
+		"session", msg.SessionKey,
+		"interactive_key", interactiveKey,
+		"msg_id", msg.MessageID,
+		"msg_create_time_ms", msg.UserMessageTimeMs,
+		"queue_depth", queueDepth,
+	)
+
 	slog.Info("message queued for busy session",
 		"session", msg.SessionKey,
 		"user", msg.UserName,
@@ -2650,6 +2843,7 @@ func (e *Engine) processInteractiveMessageWith(p Platform, msg *Message, session
 	state.platform = p
 	state.replyCtx = msg.ReplyCtx
 	state.currentMessageID = msg.MessageID
+	state.currentTurnUserMessageTimeMs = msg.UserMessageTimeMs
 	state.mu.Unlock()
 	stopRecallMonitor := e.startMessageRecallMonitor(interactiveKey)
 	defer stopRecallMonitor()
@@ -4354,6 +4548,8 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess
 				"silent", isSilent,
 			)
 
+			e.noteUserTurnCompleted(state)
+
 			normalizedBaseResponse := strings.TrimSpace(baseResponse)
 			state.mu.Lock()
 			suppressDuplicate := normalizedBaseResponse != "" && normalizedBaseResponse == state.sideText
@@ -4542,6 +4738,17 @@ 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()
+			droppedStale := 0
+			for len(state.pendingMessages) > 0 && e.isStaleUserMessageLocked(state, state.pendingMessages[0].userMessageTimeMs) {
+				state.pendingMessages = state.pendingMessages[1:]
+				droppedStale++
+			}
+			if droppedStale > 0 {
+				slog.Info("dropped stale queued user messages after completed turn",
+					"session", sessionKey,
+					"dropped", droppedStale,
+				)
+			}
 			if len(state.pendingMessages) > 0 {
 				queued := state.pendingMessages[0]
 				state.pendingMessages = state.pendingMessages[1:]
@@ -4550,6 +4757,7 @@ func (e *Engine) processInteractiveEvents(state *interactiveState, session *Sess
 				state.replyCtx = queued.replyCtx
 				state.currentMessageID = queued.messageID
 				state.fromVoice = queued.fromVoice
+				state.currentTurnUserMessageTimeMs = queued.userMessageTimeMs
 				state.mu.Unlock()
 
 				// Stop the previous turn's typing indicator
@@ -4853,12 +5061,29 @@ func (e *Engine) drainPendingMessages(state *interactiveState, session *Session,
 			state.mu.Unlock()
 			return true
 		}
+		droppedStale := 0
+		for len(state.pendingMessages) > 0 && e.isStaleUserMessageLocked(state, state.pendingMessages[0].userMessageTimeMs) {
+			state.pendingMessages = state.pendingMessages[1:]
+			droppedStale++
+		}
+		if droppedStale > 0 {
+			slog.Info("dropped stale queued user messages while draining",
+				"session", sessionKey,
+				"dropped", droppedStale,
+			)
+		}
+		if len(state.pendingMessages) == 0 {
+			session.Unlock()
+			state.mu.Unlock()
+			return true
+		}
 		queued := state.pendingMessages[0]
 		state.pendingMessages = state.pendingMessages[1:]
 		state.platform = queued.platform
 		state.replyCtx = queued.replyCtx
 		state.currentMessageID = queued.messageID
 		state.fromVoice = queued.fromVoice
+		state.currentTurnUserMessageTimeMs = queued.userMessageTimeMs
 		state.mu.Unlock()
 
 		e.i18n.DetectAndSet(queued.content)

core/message.go

@@ -185,6 +185,11 @@ type Message struct {
 	ReplyCtx     any                 // platform-specific context needed for replying
 	FromVoice    bool                // true if message originated from voice transcription
 	ModeOverride string              // if set, temporarily override agent permission mode for this message
+	// UserMessageTimeMs is the platform message creation time in Unix milliseconds
+	// when known (e.g. Feishu im.message.message_received create_time). Used to
+	// drop late redeliveries that reuse a new message_id but an older create_time
+	// than a message already processed. Zero means unset (no ordering hint).
+	UserMessageTimeMs int64
 }
 
 // EventType distinguishes different kinds of agent output.

core/user_message_watermark_test.go

@@ -0,0 +1,202 @@
+package core
+
+import "testing"
+
+func TestIsStaleUserMessageLocked(t *testing.T) {
+	e := &Engine{}
+	s := &interactiveState{lastCompletedUserMessageTimeMs: 200}
+	if !e.isStaleUserMessageLocked(s, 199) {
+		t.Fatal("199 should be stale vs 200")
+	}
+	if e.isStaleUserMessageLocked(s, 200) {
+		t.Fatal("200 should not be stale vs 200 (strict <)")
+	}
+	if e.isStaleUserMessageLocked(s, 201) {
+		t.Fatal("201 should not be stale")
+	}
+	if e.isStaleUserMessageLocked(s, 0) {
+		t.Fatal("unset time never stale")
+	}
+	s.lastCompletedUserMessageTimeMs = 0
+	if e.isStaleUserMessageLocked(s, 50) {
+		t.Fatal("no watermark means not stale")
+	}
+
+	// In-flight turn (B processing) without completed watermark yet.
+	s = &interactiveState{currentTurnUserMessageTimeMs: 500}
+	if !e.isStaleUserMessageLocked(s, 499) {
+		t.Fatal("499 should be stale vs in-flight 500")
+	}
+	if e.isStaleUserMessageLocked(s, 500) {
+		t.Fatal("500 should not be stale vs in-flight 500")
+	}
+
+	// Queued newer message while session is busy.
+	s = &interactiveState{
+		currentTurnUserMessageTimeMs: 400,
+		pendingMessages: []queuedMessage{
+			{userMessageTimeMs: 600},
+		},
+	}
+	if !e.isStaleUserMessageLocked(s, 599) {
+		t.Fatal("599 should be stale vs queued 600")
+	}
+	if e.isStaleUserMessageLocked(s, 600) {
+		t.Fatal("600 should not be stale vs queued 600")
+	}
+}
+
+func TestNoteUserTurnCompleted(t *testing.T) {
+	e := &Engine{}
+	s := &interactiveState{currentTurnUserMessageTimeMs: 100, lastCompletedUserMessageTimeMs: 0}
+	e.noteUserTurnCompleted(s)
+	if s.lastCompletedUserMessageTimeMs != 100 {
+		t.Fatalf("last = %d", s.lastCompletedUserMessageTimeMs)
+	}
+	s.currentTurnUserMessageTimeMs = 90
+	e.noteUserTurnCompleted(s)
+	if s.lastCompletedUserMessageTimeMs != 100 {
+		t.Fatalf("older turn should not lower last, got %d", s.lastCompletedUserMessageTimeMs)
+	}
+	s.currentTurnUserMessageTimeMs = 150
+	e.noteUserTurnCompleted(s)
+	if s.lastCompletedUserMessageTimeMs != 150 {
+		t.Fatalf("last = %d", s.lastCompletedUserMessageTimeMs)
+	}
+}
+
+func TestDiscardStaleUserMessageIfNeeded(t *testing.T) {
+	e := newTestEngine()
+	sessionKey := "feishu:oc_chat:ou_user"
+	e.interactiveMu.Lock()
+	e.interactiveStates[sessionKey] = &interactiveState{
+		lastCompletedUserMessageTimeMs: 2_000,
+	}
+	e.interactiveMu.Unlock()
+
+	stale := &Message{
+		SessionKey:        sessionKey,
+		Platform:          "feishu",
+		MessageID:         "om_redelivery",
+		UserMessageTimeMs: 1_000,
+		Content:           "old task",
+	}
+	if !e.discardStaleUserMessageIfNeeded(sessionKey, stale) {
+		t.Fatal("expected redelivered older create_time to be discarded")
+	}
+
+	fresh := &Message{
+		SessionKey:        sessionKey,
+		Platform:          "feishu",
+		MessageID:         "om_new",
+		UserMessageTimeMs: 2_500,
+		Content:           "new task",
+	}
+	if e.discardStaleUserMessageIfNeeded(sessionKey, fresh) {
+		t.Fatal("message newer than watermark must not be discarded")
+	}
+
+	if e.discardStaleUserMessageIfNeeded(sessionKey, &Message{
+		SessionKey: sessionKey, Content: "no timestamp",
+	}) {
+		t.Fatal("zero UserMessageTimeMs must not be discarded")
+	}
+
+	// B in-flight: A redelivery must drop even before EventResult.
+	e.interactiveMu.Lock()
+	e.interactiveStates[sessionKey] = &interactiveState{
+		currentTurnUserMessageTimeMs: 5_000,
+	}
+	e.interactiveMu.Unlock()
+	if !e.discardStaleUserMessageIfNeeded(sessionKey, &Message{
+		SessionKey:        sessionKey,
+		UserMessageTimeMs: 4_000,
+		MessageID:         "om_a_redelivery",
+	}) {
+		t.Fatal("A redelivery should drop while B is in-flight")
+	}
+}
+
+func TestNoteUserMessageAccepted(t *testing.T) {
+	e := newTestEngine()
+	sessionKey := "feishu:oc_chat:ou_user"
+	e.interactiveMu.Lock()
+	e.interactiveStates[sessionKey] = &interactiveState{}
+	e.interactiveMu.Unlock()
+
+	e.noteUserMessageAccepted(sessionKey, 3_000)
+	e.interactiveMu.Lock()
+	state := e.interactiveStates[sessionKey]
+	e.interactiveMu.Unlock()
+	state.mu.Lock()
+	got := state.currentTurnUserMessageTimeMs
+	state.mu.Unlock()
+	if got != 3_000 {
+		t.Fatalf("currentTurn = %d, want 3000", got)
+	}
+
+	e.noteUserMessageAccepted(sessionKey, 2_000)
+	state.mu.Lock()
+	got = state.currentTurnUserMessageTimeMs
+	state.mu.Unlock()
+	if got != 3_000 {
+		t.Fatalf("older accept must not lower watermark, got %d", got)
+	}
+}
+
+func TestQueueMessageForBusySession_RejectsStaleBeforeEnqueue(t *testing.T) {
+	e := newTestEngine()
+	e.maxQueuedMessages = 5
+	sessionKey := "feishu:oc_chat:ou_user"
+	p := &stubPlatformEngine{n: "feishu"}
+
+	e.interactiveMu.Lock()
+	e.interactiveStates[sessionKey] = &interactiveState{
+		platform:                       p,
+		lastCompletedUserMessageTimeMs: 5_000,
+	}
+	e.interactiveMu.Unlock()
+
+	msg := &Message{
+		SessionKey:        sessionKey,
+		Platform:          "feishu",
+		MessageID:         "om_stale_queue",
+		UserMessageTimeMs: 3_000,
+		Content:           "stale while busy",
+		ReplyCtx:          "rctx",
+	}
+	if !e.queueMessageForBusySession(p, msg, sessionKey) {
+		t.Fatal("stale message should be handled (dropped) without enqueue failure")
+	}
+
+	e.interactiveMu.Lock()
+	state := e.interactiveStates[sessionKey]
+	e.interactiveMu.Unlock()
+	state.mu.Lock()
+	if len(state.pendingMessages) != 0 {
+		t.Fatalf("pending = %d, want 0 for stale message", len(state.pendingMessages))
+	}
+
+	// B in-flight (not completed): A redelivery must not enter the queue.
+	state.currentTurnUserMessageTimeMs = 8_000
+	state.lastCompletedUserMessageTimeMs = 0
+	state.pendingMessages = nil
+	state.mu.Unlock()
+
+	msg = &Message{
+		SessionKey:        sessionKey,
+		Platform:          "feishu",
+		MessageID:         "om_a_redelivery",
+		UserMessageTimeMs: 7_000,
+		Content:           "A while B processing",
+		ReplyCtx:          "rctx2",
+	}
+	if !e.queueMessageForBusySession(p, msg, sessionKey) {
+		t.Fatal("stale vs in-flight should return handled=true")
+	}
+	state.mu.Lock()
+	if len(state.pendingMessages) != 0 {
+		t.Fatalf("pending = %d, want 0 for stale vs in-flight", len(state.pendingMessages))
+	}
+	state.mu.Unlock()
+}

platform/feishu/feishu.go

@@ -1033,6 +1033,7 @@ func (p *Platform) onMessage(ctx context.Context, event *larkim.P2MessageReceive
 		return nil
 	}
 
+	var createTimeMs int64
 	if msg.CreateTime != nil {
 		if ms, err := strconv.ParseInt(*msg.CreateTime, 10, 64); err == nil {
 			msgTime := time.Unix(ms/1000, (ms%1000)*int64(time.Millisecond))
@@ -1040,6 +1041,7 @@ func (p *Platform) onMessage(ctx context.Context, event *larkim.P2MessageReceive
 				slog.Debug(p.tag()+": ignoring old message after restart", "create_time", *msg.CreateTime)
 				return nil
 			}
+			createTimeMs = ms
 		}
 	}
 
@@ -1128,7 +1130,7 @@ func (p *Platform) onMessage(ctx context.Context, event *larkim.P2MessageReceive
 	// blocked by IO-heavy operations (image/audio download, handler HTTP calls).
 	// The dedup and old-message checks above remain synchronous to guarantee
 	// correctness before spawning the goroutine.
-	go p.dispatchMessage(ctx, msgType, content, mentions, messageID, sessionKey, userID, chatID, rctx, parentID)
+	go p.dispatchMessage(ctx, msgType, content, mentions, messageID, sessionKey, userID, chatID, rctx, parentID, createTimeMs)
 
 	return nil
 }
@@ -1145,7 +1147,7 @@ func (p *Platform) replyUnauthorizedAccess(ctx context.Context, rctx replyContex
 // dispatchMessage handles the message content parsing, media download, and
 // handler invocation. It runs in its own goroutine so that onMessage returns
 // quickly and does not block the SDK event loop.
-func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string, mentions []*larkim.MentionEvent, messageID, sessionKey, userID, chatID string, rctx replyContext, parentID string) {
+func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string, mentions []*larkim.MentionEvent, messageID, sessionKey, userID, chatID string, rctx replyContext, parentID string, createTimeMs int64) {
 	if p.isMessageRecalled(messageID) {
 		slog.Debug(p.tag()+": recalled message ignored in async dispatch", "message_id", messageID)
 		return
@@ -1191,6 +1193,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 			MessageID: messageID,
 			UserID:    userID, UserName: userName, ChatName: chatName,
 			Content: text, ExtraContent: quoted.text, Images: quoted.images, ReplyCtx: rctx,
+			UserMessageTimeMs: createTimeMs,
 		})
 
 	case "image":
@@ -1215,6 +1218,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 			UserID:    userID, UserName: userName, ChatName: chatName,
 			Images:            []core.ImageAttachment{{MimeType: mimeType, Data: imgData}},
 			ReplyCtx:          rctx,
+			UserMessageTimeMs: createTimeMs,
 		})
 
 	case "audio":
@@ -1246,6 +1250,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 				Duration: audioBody.Duration / 1000,
 			},
 			ReplyCtx:          rctx,
+			UserMessageTimeMs: createTimeMs,
 		})
 
 	case "post":
@@ -1260,6 +1265,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 			UserID:    userID, UserName: userName, ChatName: chatName,
 			Content: text, ExtraContent: quoted.text, Images: append(quoted.images, images...),
 			ReplyCtx:          rctx,
+			UserMessageTimeMs: createTimeMs,
 		})
 
 	case "file":
@@ -1292,6 +1298,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 				FileName: fileBody.FileName,
 			}},
 			ReplyCtx:          rctx,
+			UserMessageTimeMs: createTimeMs,
 		})
 
 	case "merge_forward":
@@ -1308,6 +1315,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 			Images:            images,
 			Files:             files,
 			ReplyCtx:          rctx,
+			UserMessageTimeMs: createTimeMs,
 		}
 		p.dispatchCoreMessage(coreMsg)
 
@@ -1328,6 +1336,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 				MessageID: messageID,
 				UserID:    userID, UserName: userName, ChatName: chatName,
 				Content: "[sticker]", ExtraContent: quoted.text, ReplyCtx: rctx,
+				UserMessageTimeMs: createTimeMs,
 			})
 			return
 		}
@@ -1337,6 +1346,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 			UserID:    userID, UserName: userName, ChatName: chatName,
 			Images:            []core.ImageAttachment{{MimeType: mimeType, Data: imgData}},
 			ReplyCtx:          rctx,
+			UserMessageTimeMs: createTimeMs,
 		})
 
 	case "media":
@@ -1372,6 +1382,7 @@ func (p *Platform) dispatchMessage(ctx context.Context, msgType, content string,
 			MessageID: messageID,
 			UserID:    userID, UserName: userName, ChatName: chatName,
 			Content: text, ExtraContent: quoted.text, Images: images, ReplyCtx: rctx,
+			UserMessageTimeMs: createTimeMs,
 		})
 
 	default:

platform/feishu/feishu_test.go

@@ -80,6 +80,7 @@ func TestDispatchMessageDropsRecalledMessageBeforeHandler(t *testing.T) {
 		"",
 		replyContext{messageID: "om_drop", sessionKey: "feishu:ou_user:ou_user"},
 		"",
+		0,
 	)
 
 	if called {
@@ -185,6 +186,7 @@ func TestDispatchMessageIncludesQuotedImage(t *testing.T) {
 				"",
 				replyContext{messageID: "om_child", sessionKey: "feishu:oc_chat:ou_user"},
 				parentMessageID,
+				0,
 			)
 
 			select {
@@ -285,6 +287,7 @@ func TestDispatchMessageKeepsMentionOnlyQuotedText(t *testing.T) {
 		"oc_chat",
 		replyContext{messageID: "om_child", sessionKey: "feishu:oc_chat:ou_user"},
 		parentMessageID,
+		0,
 	)
 
 	select {