fix(core): add engine-level message dedup safety net (fixes #1667)

cmd/cc-connect/main.go

@@ -640,6 +640,21 @@ func main() {
 				})
 			}
 		}
+		// Wire engine-level message dedup (issue #1667 — WeChat server
+		// retransmits can slip past per-platform dedup; this catch-all keys
+		// by MessageID only). Default enabled=true with a 30s window to cover
+		// the typical retry interval reported by the issue reporter.
+		{
+			dedupEnabled := true
+			if cfg.Dedup.Enabled != nil {
+				dedupEnabled = *cfg.Dedup.Enabled
+			}
+			windowSecs := 30
+			if cfg.Dedup.WindowSecs != nil {
+				windowSecs = *cfg.Dedup.WindowSecs
+			}
+			engine.SetDedupConfig(dedupEnabled, time.Duration(windowSecs)*time.Second)
+		}
 		// Wire outgoing rate limiting
 		{
 			var maxPS float64
@@ -1824,6 +1839,19 @@ func reloadConfig(configPath, projName string, engine *core.Engine) (*core.Confi
 	// Reload filter_external_sessions
 	engine.SetFilterExternalSessions(proj.FilterExternalSessions != nil && *proj.FilterExternalSessions)
 
+	// Reload engine-level message dedup (issue #1667)
+	{
+		dedupEnabled := true
+		if cfg.Dedup.Enabled != nil {
+			dedupEnabled = *cfg.Dedup.Enabled
+		}
+		windowSecs := 30
+		if cfg.Dedup.WindowSecs != nil {
+			windowSecs = *cfg.Dedup.WindowSecs
+		}
+		engine.SetDedupConfig(dedupEnabled, time.Duration(windowSecs)*time.Second)
+	}
+
 	// Reload providers
 	if ps, ok := engine.GetAgent().(core.ProviderSwitcher); ok {
 		providers := make([]core.ProviderConfig, len(proj.Agent.Providers))

config.example.toml

@@ -287,6 +287,20 @@ level = "info" # debug, info, warn, error
 # Per-platform overrides / 每平台覆盖配置:
 # [outgoing_rate_limit.platforms.wecom]
 # max_per_second = 1
+
+# =============================================================================
+# Message Deduplication / 消息去重 (issue #1667)
+# Engine-level safety net that catches duplicate message IDs slipping past the
+# per-platform dedup layer. Useful for platforms (e.g. WeChat) whose server
+# retransmits the same message with a refreshed timestamp, breaking the
+# platform's composite dedup key. Default ON.
+# 引擎层消息 ID 去重安全网，兜底拦截漏过平台层去重的重复消息。
+# 适用于服务器端重传时刷新时间戳导致平台层去重键失效的平台（如微信）。
+# 默认开启。
+# =============================================================================
+# [dedup]
+# enabled = true             # Toggle the engine-level dedup (default: true) / 开关（默认 true）
+# window_secs = 30           # Dedup window in seconds (default: 30) / 去重时间窗秒数（默认 30）
 # [outgoing_rate_limit.platforms.telegram]
 # max_per_second = 25
 # [outgoing_rate_limit.platforms.feishu]

config/config.go

@@ -104,6 +104,7 @@ type Config struct {
 	InstantReply       InstantReplyConfig      `toml:"instant_reply"`       // immediate confirmation reply
 	RateLimit          RateLimitConfig         `toml:"rate_limit"`          // per-session rate limiting
 	OutgoingRateLimit  OutgoingRateLimitConfig `toml:"outgoing_rate_limit"` // outgoing message throttling
+	Dedup              DedupConfig             `toml:"dedup"`               // engine-level message-id dedup safety net (issue #1667)
 	Relay              RelayConfig             `toml:"relay"`               // bot-to-bot relay behavior
 	Cron               CronConfig              `toml:"cron"`
 	Queue              QueueConfig             `toml:"queue"`
@@ -241,6 +242,16 @@ type OutgoingRateLimitPlatConfig struct {
 	Burst        *int     `toml:"burst"`
 }
 
+// DedupConfig configures the engine-level message-id dedup safety net (issue
+// #1667). Per-platform dedup is the primary line of defense; this catch-all
+// matches by MessageID alone and silently drops duplicates that slipped past
+// the platform layer (e.g. WeChat server retransmits whose create_time_ms was
+// refreshed on retry, breaking the platform's composite dedup key).
+type DedupConfig struct {
+	Enabled    *bool `toml:"enabled"`     // default true; set false to disable
+	WindowSecs *int  `toml:"window_secs"` // dedup window in seconds; default 30
+}
+
 // UsersConfig controls per-user role assignments and policies within a project.
 type UsersConfig struct {
 	DefaultRole string                `toml:"default_role,omitempty"` // role for unmatched users; default "member"

core/dedup.go

@@ -5,6 +5,9 @@ import (
 	"time"
 )
 
+// dedupTTL is the default dedup window used when a zero-value MessageDedup is
+// constructed (preserves the historical behavior of every platform-level
+// MessageDedup embedded as `core.MessageDedup{}`).
 const dedupTTL = 60 * time.Second
 
 // StartTime is set once at process startup.
@@ -14,9 +17,26 @@ var StartTime = time.Now()
 
 // MessageDedup tracks recently seen message IDs to prevent duplicate processing.
 // Safe for concurrent use.
+//
+// The zero value uses a fixed 60s window (preserving the original behavior of
+// every platform that embeds this as `core.MessageDedup{}`). Use
+// NewMessageDedup to override the window — the engine uses this for the
+// cross-platform safety net described in issue #1667, where a per-platform
+// dedup key (e.g. WeChat's `from|msg_id|seq|create_time_ms|client_id`) misses
+// server retransmissions whose create_time_ms gets refreshed on retry.
 type MessageDedup struct {
 	mu   sync.Mutex
 	seen map[string]time.Time
+	ttl  time.Duration // 0 = use dedupTTL default
+}
+
+// NewMessageDedup returns a MessageDedup with a caller-supplied window.
+// Pass 0 or a negative value to get the package default (60s).
+func NewMessageDedup(ttl time.Duration) *MessageDedup {
+	if ttl <= 0 {
+		ttl = dedupTTL
+	}
+	return &MessageDedup{ttl: ttl}
 }
 
 // IsDuplicate returns true if msgID was already seen within the TTL window.
@@ -25,14 +45,18 @@ func (d *MessageDedup) IsDuplicate(msgID string) bool {
 	if msgID == "" {
 		return false
 	}
+	ttl := d.ttl
+	if ttl <= 0 {
+		ttl = dedupTTL
+	}
 	d.mu.Lock()
 	defer d.mu.Unlock()
 	if d.seen == nil {
 		d.seen = make(map[string]time.Time)
 	}
 	now := time.Now()
 	for k, t := range d.seen {
-		if now.Sub(t) > dedupTTL {
+		if now.Sub(t) > ttl {
 			delete(d.seen, k)
 		}
 	}

core/dedup_test.go

@@ -42,6 +42,47 @@ func TestMessageDedup_Concurrent(t *testing.T) {
 	}
 }
 
+func TestNewMessageDedup_ConfigurableWindow(t *testing.T) {
+	d := NewMessageDedup(20 * time.Millisecond)
+	if d.IsDuplicate("m1") {
+		t.Fatal("first call should not be a duplicate")
+	}
+	if !d.IsDuplicate("m1") {
+		t.Fatal("second call within window should be a duplicate")
+	}
+	time.Sleep(30 * time.Millisecond)
+	if d.IsDuplicate("m1") {
+		t.Fatal("after window expiry the same id should be accepted again")
+	}
+}
+
+func TestNewMessageDedup_DefaultOnZero(t *testing.T) {
+	d := NewMessageDedup(0)
+	if d.ttl != dedupTTL {
+		t.Errorf("expected default TTL %v, got %v", dedupTTL, d.ttl)
+	}
+}
+
+func TestNewMessageDedup_DefaultOnNegative(t *testing.T) {
+	d := NewMessageDedup(-5 * time.Second)
+	if d.ttl != dedupTTL {
+		t.Errorf("expected default TTL %v on negative input, got %v", dedupTTL, d.ttl)
+	}
+}
+
+func TestMessageDedup_ZeroValueStillUsesDefaultTTL(t *testing.T) {
+	// Backward-compat: every platform that embeds `core.MessageDedup{}` must
+	// continue to work with the original 60s window. First call primes,
+	// second call inside the window must be flagged duplicate.
+	var d MessageDedup
+	if d.IsDuplicate("z1") {
+		t.Fatal("first call should not be a duplicate")
+	}
+	if !d.IsDuplicate("z1") {
+		t.Fatal("second call within window should be a duplicate")
+	}
+}
+
 func TestIsOldMessage(t *testing.T) {
 	if IsOldMessage(time.Now()) {
 		t.Error("current time should not be considered old")

core/engine.go

@@ -382,6 +382,8 @@ type Engine struct {
 
 	rateLimiter      *RateLimiter
 	outgoingRL       *OutgoingRateLimiter
+	dedup            *MessageDedup
+	dedupEnabled     bool
 	streamPreview    StreamPreviewCfg
 	instantReply     InstantReplyCfg
 	references       ReferenceRenderCfg
@@ -1252,6 +1254,23 @@ func (e *Engine) SetOutgoingRateLimitCfg(defaults OutgoingRateLimitCfg, override
 	e.outgoingRL = NewOutgoingRateLimiter(defaults, overrides)
 }
 
+// SetDedupConfig enables or disables the engine-level message dedup safety net
+// that catches duplicates escaping per-platform dedup (issue #1667 — WeChat
+// server retransmits with a refreshed create_time_ms so the platform-level
+// dedup key changes, and both copies reach handleMessage).
+//
+// Pass enabled=false to disable dedup entirely (no cache is built). Pass
+// window=0 to use the package default (60s).
+func (e *Engine) SetDedupConfig(enabled bool, window time.Duration) {
+	if !enabled {
+		e.dedup = nil
+		e.dedupEnabled = false
+		return
+	}
+	e.dedup = NewMessageDedup(window)
+	e.dedupEnabled = true
+}
+
 // checkRateLimit returns true if the message is allowed, false if rate-limited.
 // It checks per-user role-based limits first, then falls back to the global limiter.
 func (e *Engine) checkRateLimit(msg *Message) bool {
@@ -2742,6 +2761,19 @@ func (e *Engine) handleMessage(p Platform, msg *Message) {
 		return
 	}
 
+	// Engine-level dedup safety net (issue #1667). Per-platform dedup is the
+	// primary line of defense, but a few platforms key their dedup by fields
+	// that server-side retransmission can refresh (e.g. WeChat's create_time_ms).
+	// When the platform copy slips through, this catch-all matches by MessageID
+	// alone and silently drops the second copy. Disabled when dedupEnabled is
+	// false or when the platform didn't supply a MessageID.
+	if e.dedupEnabled && e.dedup != nil && msg.MessageID != "" && e.dedup.IsDuplicate(msg.MessageID) {
+		slog.Info("message deduplicated by engine",
+			"platform", msg.Platform, "msg_id", msg.MessageID,
+			"session", msg.SessionKey, "user", msg.UserName)
+		return
+	}
+
 	slog.Info("message received",
 		"platform", msg.Platform, "msg_id", msg.MessageID,
 		"session", msg.SessionKey, "user", msg.UserName,