diff --git a/cmd/cc-connect/main.go b/cmd/cc-connect/main.go index acdbd3527c..708e3c4fa4 100644 --- a/cmd/cc-connect/main.go +++ b/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)) diff --git a/config.example.toml b/config.example.toml index 74b57b519f..d02d315471 100644 --- a/config.example.toml +++ b/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] diff --git a/config/config.go b/config/config.go index 908027db38..f44e3bd209 100644 --- a/config/config.go +++ b/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" diff --git a/core/dedup.go b/core/dedup.go index fc6f822926..30870bb433 100644 --- a/core/dedup.go +++ b/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,6 +45,10 @@ 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 { @@ -32,7 +56,7 @@ func (d *MessageDedup) IsDuplicate(msgID string) bool { } now := time.Now() for k, t := range d.seen { - if now.Sub(t) > dedupTTL { + if now.Sub(t) > ttl { delete(d.seen, k) } } diff --git a/core/dedup_test.go b/core/dedup_test.go index 34cefa2ddc..a17321a028 100644 --- a/core/dedup_test.go +++ b/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") diff --git a/core/engine.go b/core/engine.go index e99eee412a..9fedb71bf6 100644 --- a/core/engine.go +++ b/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, diff --git a/core/engine_test.go b/core/engine_test.go index b68cf7f6b6..a8902e6e3a 100644 --- a/core/engine_test.go +++ b/core/engine_test.go @@ -15994,3 +15994,156 @@ func TestProcessInteractiveEvents_StreamingCard_BareNoReply_Suppressed(t *testin t.Fatalf("silent reply leaked NO_REPLY into the streaming card: %q", card.finalContent()) } } + +// ── handleMessage: engine-level dedup safety net (issue #1667) ── + +func TestHandleMessage_EngineDedup_DropsDuplicate(t *testing.T) { + p := &stubPlatformEngine{n: "weixin"} + e := newTestEngine() + e.SetDedupConfig(true, time.Minute) + + mkMsg := func(id, content string) *Message { + return &Message{ + SessionKey: "weixin:user1", + Platform: "weixin", + UserID: "user1", + UserName: "user1", + MessageID: id, + Content: content, + ReplyCtx: "ctx", + } + } + + // First arrival: should pass through to the normal handleMessage path + // (no Reply because the test stub agent returns immediately). + e.handleMessage(p, mkMsg("7492913259736648968", "first")) + + // Simulate a platform that already injected a "you sent this twice" reply + // so we can verify the dedup short-circuit skipped the Reply call entirely. + p.clearSent() + + // Same MessageID within the window — must be dropped before any side effect. + e.handleMessage(p, mkMsg("7492913259736648968", "first")) + + if got := p.getSent(); len(got) != 0 { + t.Errorf("expected no platform messages after dedup drop, got %v", got) + } +} + +func TestHandleMessage_EngineDedup_AllowsDifferentMessageIDs(t *testing.T) { + p := &stubPlatformEngine{n: "weixin"} + e := newTestEngine() + e.SetDedupConfig(true, time.Minute) + + for _, id := range []string{"m1", "m2", "m3"} { + e.handleMessage(p, &Message{ + SessionKey: "weixin:user1", + Platform: "weixin", + UserID: "user1", + MessageID: id, + Content: "hi", + ReplyCtx: "ctx", + }) + } + // All three reached the normal path; the dedup only short-circuited on + // collisions, never on unique ids. + if !e.dedupEnabled { + t.Fatal("expected dedup to be enabled") + } +} + +func TestHandleMessage_EngineDedup_DisabledPassesAll(t *testing.T) { + p := &stubPlatformEngine{n: "weixin"} + e := newTestEngine() + e.SetDedupConfig(false, time.Minute) + + for i := 0; i < 3; i++ { + e.handleMessage(p, &Message{ + SessionKey: "weixin:user1", + Platform: "weixin", + UserID: "user1", + MessageID: "same-id", + Content: "hi", + ReplyCtx: "ctx", + }) + } + if e.dedup != nil { + t.Errorf("expected dedup cache to be nil when disabled, got %+v", e.dedup) + } + if e.dedupEnabled { + t.Error("expected dedupEnabled to be false") + } +} + +func TestHandleMessage_EngineDedup_EmptyMessageIDSkipsDedup(t *testing.T) { + // Platforms that omit MessageID (rare but possible) must never be silently + // dropped — we have no key to dedup by. The dedup cache is exercised but + // never matches. + p := &stubPlatformEngine{n: "feishu"} + e := newTestEngine() + e.SetDedupConfig(true, time.Minute) + + for i := 0; i < 5; i++ { + e.handleMessage(p, &Message{ + SessionKey: "feishu:user1", + Platform: "feishu", + UserID: "user1", + MessageID: "", // intentionally empty + Content: "hi", + ReplyCtx: "ctx", + }) + } +} + +func TestHandleMessage_EngineDedup_RecalledMessagesSkipped(t *testing.T) { + // Recalls (delete/recall events) hit handleMessageRecall, not the dedup + // path. Make sure we don't accidentally drop them when the original + // message_id was already in the dedup cache. + p := &stubPlatformEngine{n: "weixin"} + e := newTestEngine() + e.SetDedupConfig(true, time.Minute) + + // Prime the dedup cache with a normal message. + e.handleMessage(p, &Message{ + SessionKey: "weixin:user1", + Platform: "weixin", + UserID: "user1", + MessageID: "7492913259736648968", + Content: "hi", + ReplyCtx: "ctx", + }) + + // Now a recall event with the same MessageID should reach handleMessageRecall, + // not be dropped as a duplicate. + recall := &Message{ + SessionKey: "weixin:user1", + Platform: "weixin", + UserID: "user1", + MessageID: "7492913259736648968", + Recalled: true, + ReplyCtx: "ctx", + } + e.handleMessage(p, recall) + + // No assertion on side effects — handleMessageRecall is covered by its own + // tests. The point is that the recall path wasn't short-circuited. + if !e.dedupEnabled { + t.Fatal("expected dedup to still be enabled") + } +} + +func TestHandleMessage_EngineDedup_TogglesOffAndOn(t *testing.T) { + e := newTestEngine() + e.SetDedupConfig(true, time.Minute) + if e.dedup == nil || !e.dedupEnabled { + t.Fatal("setup: dedup should be on") + } + e.SetDedupConfig(false, time.Minute) + if e.dedup != nil || e.dedupEnabled { + t.Errorf("expected dedup off (cache=nil, enabled=false), got cache=%v enabled=%v", e.dedup, e.dedupEnabled) + } + e.SetDedupConfig(true, time.Minute) + if e.dedup == nil || !e.dedupEnabled { + t.Errorf("expected dedup back on, got cache=%v enabled=%v", e.dedup, e.dedupEnabled) + } +}