refactor(channelbridge): 多气泡分割+旁白剥离+随机延迟上移为IM公共出口——sendReply/SendProactive统一(微信主动推送也分割)

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 <p> 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 <action>/<narration> and the
+	// star-drift role-play forms are removed for IM) and splits on <p>. 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
+}

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 := "<p><action>弯起眼睛笑了</action>当然看得到呀。</p><p>第二句话。</p>"
+	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 <p> 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}})
+	}, "<action>拍拍你</action>"); 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-<p> 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 := "<p><action>弯起眼睛笑了</action>当然看得到呀。</p><p>怎么突然这么问。</p>"
+	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 <p> 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 <p> 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("<p>第一段正文。</p><p><action>拍拍</action>第二段。</p>")
+	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 <p> 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)
+	}
+}

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
+	// bubbleGapFn is the injectable delay between IM outbound bubbles (0.5-2s),
-	engageGroups  map[string]*engageGroupState
+	// 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 + <p> 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
+	return b.sendIMBubbles(ctx, in, func(ctx context.Context, msg adapter.UnifiedMessage) error {
-	// markup (rendered as muted italic on the Web) is stripped, and each <p> paragraph
+		return a.SendMessage(ctx, adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{msg}})
-	// becomes its own bubble/message (design/17 §2). A reply that was only an action is
+	}, text)
-	// 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
 }
 
 // adapter returns the registered adapter for a platform name, or nil.

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 <p> 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