From bcf7fe5de031671d9eec339f902eef6065e111c8 Mon Sep 17 00:00:00 2001 From: Aska Date: Tue, 8 Sep 2026 16:26:35 +0800 Subject: [PATCH] =?UTF-8?q?refactor(channelbridge):=20=E5=A4=9A=E6=B0=94?= =?UTF-8?q?=E6=B3=A1=E5=88=86=E5=89=B2+=E6=97=81=E7=99=BD=E5=89=A5?= =?UTF-8?q?=E7=A6=BB+=E9=9A=8F=E6=9C=BA=E5=BB=B6=E8=BF=9F=E4=B8=8A?= =?UTF-8?q?=E7=A7=BB=E4=B8=BAIM=E5=85=AC=E5=85=B1=E5=87=BA=E5=8F=A3?= =?UTF-8?q?=E2=80=94=E2=80=94sendReply/SendProactive=E7=BB=9F=E4=B8=80(?= =?UTF-8?q?=E5=BE=AE=E4=BF=A1=E4=B8=BB=E5=8A=A8=E6=8E=A8=E9=80=81=E4=B9=9F?= =?UTF-8?q?=E5=88=86=E5=89=B2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/services/channelbridge/bubbles.go | 53 ++++++ .../services/channelbridge/bubbles_test.go | 156 ++++++++++++++++++ .../services/channelbridge/channelbridge.go | 47 ++---- internal/services/channelbridge/proactive.go | 13 +- 4 files changed, 233 insertions(+), 36 deletions(-) create mode 100644 internal/services/channelbridge/bubbles.go create mode 100644 internal/services/channelbridge/bubbles_test.go diff --git a/internal/services/channelbridge/bubbles.go b/internal/services/channelbridge/bubbles.go new file mode 100644 index 0000000..97c8f10 --- /dev/null +++ b/internal/services/channelbridge/bubbles.go @@ -0,0 +1,53 @@ +package channelbridge + +import ( + "context" + "fmt" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// bubbleSender is the per-bubble outbound call the shared IM dispatcher uses. A +// reply exits through the adapter's SendMessage (UnifiedResponse) path, a proactive +// push through its SendProactive path — both collapse to "dispatch one chat bubble to +// the platform", which is what lets the reply and the proactive push share a single +// sanitize/split/gap pipeline instead of each re-implementing it. +type bubbleSender func(ctx context.Context, msg adapter.UnifiedMessage) error + +// sendIMBubbles is the single IM outbound exit shared by sendReply (reply path) and +// SendProactive (proactive push). It turns content into the plain chat bubbles an IM +// channel should receive: role-play/action/narration markup (rendered as muted italic +// on the Web) is stripped and each

paragraph is delivered as its own +// bubble/message (design/17 §2), with a multi-bubble burst spacing its sends 0.5-2s +// apart so it does not flood the channel. The Web frontend is never affected — it +// renders the persisted stream directly and does not route through here. +// +// target carries the platform/channel targeting fields and is copied onto every +// bubble (only Content/ContentType/Timestamp differ per bubble); sender is the +// adapter's outbound call. A content that strips to nothing (a pure action line) is +// dropped without a send, and a per-bubble failure is wrapped with its index. +func (b *Bridge) sendIMBubbles(ctx context.Context, target adapter.UnifiedMessage, sender bubbleSender, content string) error { + // splitBubbles applies stripNarration per bubble (so / and the + // star-drift role-play forms are removed for IM) and splits on

. It is the + // single sanitize entry point, shared with the old reply path. + bubbles := splitBubbles(content) + if len(bubbles) == 0 { + b.log.Debug("im outbound has no plain text, skipped", "platform", target.Platform, "channel", target.ChannelID) + return nil + } + for i, bubble := range bubbles { + if i > 0 { + time.Sleep(b.bubbleGapFn()) + } + out := target + out.Content = bubble + out.ContentType = adapter.ContentText + out.Timestamp = time.Now().UTC() + // Each bubble is its own send so the gap lands between messages. + if err := sender(ctx, out); err != nil { + return fmt.Errorf("channelbridge: send bubble #%d of %d: %w", i+1, len(bubbles), err) + } + } + return nil +} diff --git a/internal/services/channelbridge/bubbles_test.go b/internal/services/channelbridge/bubbles_test.go new file mode 100644 index 0000000..c3480a1 --- /dev/null +++ b/internal/services/channelbridge/bubbles_test.go @@ -0,0 +1,156 @@ +package channelbridge + +import ( + "context" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/internal/user" +) + +// gapCounter returns a bubbleGapFn that counts how many times the inter-bubble delay +// is consulted and returns zero, so the multi-bubble spacing can be asserted without +// sleeping a real 0.5-2s per gap. +func gapCounter(counter *int) func() time.Duration { + return func() time.Duration { + *counter++ + return 0 + } +} + +func TestSendIMBubbles_SplitsStripsAndGaps(t *testing.T) { + t.Parallel() + ad := &fakeAdapter{name: "kook", connected: true} + b, err := New(Options{Loop: &stubChatter{}, Sessions: &stubSessions{}}) + if err != nil { + t.Fatalf("New: %v", err) + } + gaps := 0 + b.bubbleGapFn = gapCounter(&gaps) + + target := adapter.UnifiedMessage{ + Platform: "kook", ChannelID: "329284282", ChannelType: adapter.ChannelDirect, + SenderID: "329284282", + } + content := "

弯起眼睛笑了当然看得到呀。

第二句话。

" + if err := b.sendIMBubbles(context.Background(), target, func(ctx context.Context, m adapter.UnifiedMessage) error { + return ad.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{m}}) + }, content); err != nil { + t.Fatalf("sendIMBubbles err = %v", err) + } + if len(ad.sends) != 2 { + t.Fatalf("sends = %d, want 2

bubbles", len(ad.sends)) + } + if ad.sends[0].Content != "当然看得到呀。" { + t.Fatalf("bubble[0] = %q, want the role-play-stripped text", ad.sends[0].Content) + } + if ad.sends[1].Content != "第二句话。" { + t.Fatalf("bubble[1] = %q, want the second paragraph", ad.sends[1].Content) + } + if ad.sends[0].Platform != "kook" || ad.sends[0].ChannelID != "329284282" { + t.Fatalf("bubble[0] target = %+v, want kook 329284282", ad.sends[0]) + } + // Two bubbles → exactly one inter-bubble delay is consulted (and applies to both + // reply and proactive paths, since this drives the shared dispatcher). + if gaps != 1 { + t.Fatalf("gap calls = %d, want 1 (2 bubbles → one gap)", gaps) + } +} + +func TestSendIMBubbles_SingleBubbleNoGap(t *testing.T) { + t.Parallel() + ad := &fakeAdapter{name: "weixin-ilink", connected: true} + b, err := New(Options{Loop: &stubChatter{}, Sessions: &stubSessions{}}) + if err != nil { + t.Fatalf("New: %v", err) + } + gaps := 0 + b.bubbleGapFn = gapCounter(&gaps) + + target := adapter.UnifiedMessage{ + Platform: "weixin-ilink", ChannelID: "o9cq@im.wechat", ChannelType: adapter.ChannelDirect, + SenderID: "o9cq@im.wechat", + } + if err := b.sendIMBubbles(context.Background(), target, func(ctx context.Context, m adapter.UnifiedMessage) error { + return ad.SendProactive(ctx, m) + }, "记得喝水哦"); err != nil { + t.Fatalf("sendIMBubbles err = %v", err) + } + if len(ad.sends) != 1 { + t.Fatalf("sends = %d, want 1 bubble", len(ad.sends)) + } + if gaps != 0 { + t.Fatalf("gap calls = %d, want 0 for a single bubble", gaps) + } +} + +func TestSendIMBubbles_PureActionDropped(t *testing.T) { + t.Parallel() + ad := &fakeAdapter{name: "kook", connected: true} + b, err := New(Options{Loop: &stubChatter{}, Sessions: &stubSessions{}}) + if err != nil { + t.Fatalf("New: %v", err) + } + target := adapter.UnifiedMessage{ + Platform: "kook", ChannelID: "1", ChannelType: adapter.ChannelDirect, SenderID: "1", + } + if err := b.sendIMBubbles(context.Background(), target, func(ctx context.Context, m adapter.UnifiedMessage) error { + return ad.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{m}}) + }, "拍拍你"); err != nil { + t.Fatalf("sendIMBubbles err = %v", err) + } + if len(ad.sends) != 0 { + t.Fatalf("sends = %d, want 0 for a pure action line", len(ad.sends)) + } +} + +// TestSendReply_SplitsBubblesAndStrips exercises the reply exit (sendReply) after the +// refactor: a multi-

reply is delivered as several stripped bubbles through the +// shared dispatcher, targeting the source conversation. +func TestSendReply_SplitsBubblesAndStrips(t *testing.T) { + t.Parallel() + b, _, _, ad := newTestBridge(t) + b.bubbleGapFn = func() time.Duration { return 0 } + in := directMessage() + content := "

弯起眼睛笑了当然看得到呀。

怎么突然这么问。

" + if err := b.sendReply(context.Background(), in, ad, content); err != nil { + t.Fatalf("sendReply err = %v", err) + } + if len(ad.sends) != 2 { + t.Fatalf("sends = %d, want 2 (reply splits into

bubbles)", len(ad.sends)) + } + if ad.sends[0].Content != "当然看得到呀。" || ad.sends[1].Content != "怎么突然这么问。" { + t.Fatalf("bubbles = %#v, want stripped narration + both paragraphs", ad.sends) + } + if ad.sends[0].Platform != "telegram" || ad.sends[0].ChannelID != "100" { + t.Fatalf("reply target = %+v, want telegram channel 100", ad.sends[0]) + } +} + +// TestSendProactive_SplitsWeChatBubbles verifies the proactive push now splits a +// multi-paragraph message into bubbles on WeChat (the reported regression: WeChat +// proactive pushes arrived as one un-split

blob). +func TestSendProactive_SplitsWeChatBubbles(t *testing.T) { + t.Parallel() + const acct = "o9cq802QPKPscBenJAp-SMZPBlnQ@im.wechat" + weixin := &fakeAdapter{name: "weixin-ilink", connected: true} + owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{{Platform: "weixin-ilink", AccountID: acct}}} + b := newProactiveBridge(t, owner, weixin) + b.bubbleGapFn = func() time.Duration { return 0 } + + msg := proactiveMsg("

第一段正文。

拍拍第二段。

") + if err := b.SendProactive(context.Background(), msg); err != nil { + t.Fatalf("SendProactive err = %v", err) + } + if len(weixin.sends) != 2 { + t.Fatalf("weixin sends = %d, want 2 (proactive splits into

bubbles)", len(weixin.sends)) + } + if weixin.sends[0].Content != "第一段正文。" || weixin.sends[1].Content != "第二段。" { + t.Fatalf("weixin bubbles = %#v, want stripped narration + both paragraphs", weixin.sends) + } + if weixin.sends[0].Platform != "weixin-ilink" || + weixin.sends[0].ChannelID != acct || weixin.sends[0].SenderID != acct { + t.Fatalf("weixin peer = %+v, want platform weixin-ilink account %s", weixin.sends[0], acct) + } +} diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 73c4ac2..c1bffb0 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -169,8 +169,12 @@ type Bridge struct { engage EngageConfig engageFor func(platform, groupID string) *EngageConfig engageRand func() float64 - engageMu sync.Mutex - engageGroups map[string]*engageGroupState + // bubbleGapFn is the injectable delay between IM outbound bubbles (0.5-2s), + // defaulting to the package bubbleGap. Tests inject a fixed short gap so the + // multi-bubble delay is asserted deterministically without sleeping a real 0.5-2s. + bubbleGapFn func() time.Duration + engageMu sync.Mutex + engageGroups map[string]*engageGroupState mu sync.RWMutex adapters map[string]adapter.PlatformAdapter @@ -242,6 +246,7 @@ func New(opts Options) (*Bridge, error) { engage: opts.Engage, engageFor: opts.EngageFor, engageRand: engageRand, + bubbleGapFn: bubbleGap, engageGroups: make(map[string]*engageGroupState), adapters: make(map[string]adapter.PlatformAdapter), connected: make(map[string]bool), @@ -660,7 +665,8 @@ func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.Sess // sendReply delivers the loop's text back to the channel's source conversation // through the account adapter that received the message (multi-account safe, // design/16 §4.6 §5). It mirrors the inbound message's channel targeting fields so -// the adapter resolves the same peer the message came from. +// the adapter resolves the same peer the message came from, and it exits through the +// shared IM bubble dispatcher (strip role-play/narration +

split + 0.5-2s gap). func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, a adapter.PlatformAdapter, text string) error { if a == nil { a = b.adapter(in.Platform) @@ -668,38 +674,9 @@ func (b *Bridge) sendReply(ctx context.Context, in adapter.UnifiedMessage, a ada if a == nil { return fmt.Errorf("channelbridge: no adapter for platform %q", in.Platform) } - // IM channels receive the reply as one or more bubbles: role-play/action/narration - // markup (rendered as muted italic on the Web) is stripped, and each

paragraph - // becomes its own bubble/message (design/17 §2). A reply that was only an action is - // dropped rather than sent empty. A multi-bubble reply spaces its sends 0.5-2s apart - // so a burst doesn't flood the channel; the Web frontend has no such delay. - bubbles := splitBubbles(text) - if len(bubbles) == 0 { - b.log.Debug("reply has no plain text, skipped", "platform", in.Platform, "channel", in.ChannelID) - return nil - } - for i, content := range bubbles { - if i > 0 { - time.Sleep(bubbleGap()) - } - out := adapter.UnifiedMessage{ - Platform: in.Platform, - ChannelType: in.ChannelType, - ChannelID: in.ChannelID, - GroupID: in.GroupID, - GroupName: in.GroupName, - SenderID: in.SenderID, - Sender: in.Sender, - Content: content, - ContentType: adapter.ContentText, - Timestamp: time.Now().UTC(), - } - // Each bubble is its own send so the gap lands between messages. - if err := a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{out}}); err != nil { - return fmt.Errorf("channelbridge: send bubble #%d of %d: %w", i+1, len(bubbles), err) - } - } - return nil + return b.sendIMBubbles(ctx, in, func(ctx context.Context, msg adapter.UnifiedMessage) error { + return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{msg}}) + }, text) } // adapter returns the registered adapter for a platform name, or nil. diff --git a/internal/services/channelbridge/proactive.go b/internal/services/channelbridge/proactive.go index 9b41bb7..a198329 100644 --- a/internal/services/channelbridge/proactive.go +++ b/internal/services/channelbridge/proactive.go @@ -31,6 +31,11 @@ const proactiveRecentCount = 20 // WeChat is a first-class candidate (it is whitelisted and, when connected, the // recent-active/fallback selection may pick it). // +// The push exits through the shared IM bubble dispatcher (sendIMBubbles), so a +// multi-paragraph message is split into bubbles, role-play/narration is stripped, and +// sends are spaced 0.5-2s apart — the same behaviour as a reply, and the reason +// WeChat/other IM proactive pushes no longer arrive as one un-split

blob. +// // When the owner profile has no whitelisted account that is backed by a connected // adapter it returns a clear error so the caller records the reason rather than // silently dropping the push. @@ -55,7 +60,13 @@ func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) // prefers, without duplicating the message. out.ChannelID = target.accountID out.SenderID = target.accountID - if err := ad.SendProactive(ctx, out); err != nil { + // The push exits through the shared IM bubble dispatcher, so a multi-paragraph + // proactive message is split into bubbles and role-play/narration is stripped, with + // the same 0.5-2s gap as a reply — WeChat proactive pushes are no longer sent as a + // single un-split message. + if err := b.sendIMBubbles(ctx, out, func(ctx context.Context, msg adapter.UnifiedMessage) error { + return ad.SendProactive(ctx, msg) + }, msg.Content); err != nil { return fmt.Errorf("channelbridge: send proactive via %q: %w", target.platform, err) } // Mirror the proactive message into the main conversation context (design/16