AskaEth
1d3965aa80
M19 复审抓到的真缺口：internal/services/gateway/proactive.go 的 SendProactive -> persistProactiveToMain 这条 Web 活跃客户端主动推送路由 完全绕过 turnreply.NormalizeParagraphs —— 同一条主动消息走 IM 是规整过的、 走 Web 实时推还是原样（半包 <p> 在客户端按「存在完整 <p> 就不再按空行回退」 的契约下整条塌成一坨）。 - 接线：SendProactive 在落库之前先 msg.Content = turnreply.NormalizeParagraphs(msg.Content)， 落库与 proactive_message 帧取同一个字符串 ⇒ 两端逐字相同；不在 gateway 另写实现。 - 新增 proactive_normalize_test.go：修前红（web 原样 / im 已规整），修后断言 ① 落库是规整后文本 ② 推帧 == 落库行 ③ 两条路由同源同文本 ④ <br>/<action> 语义不动。 - 既有两条断言按新口径更新（纯散文现在也被包成一个 <p>，与其它三处一致）。
283 lines
12 KiB
Go
283 lines
12 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"strings"
|
||
"testing"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/channelbridge"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
|
||
)
|
||
|
||
// 本文件锁定切片 P0 的**落库**那一半：主动消息无论走活跃客户端路由
|
||
// （gateway hub）还是 IM 桥路由（channelbridge），都必须成为主会话
|
||
// （`history.MainSessionID`）里**恰好一条** assistant 消息。
|
||
//
|
||
// 为什么需要它：修复前活跃客户端路由只广播 `proactive_message` 帧、一行都不写，
|
||
// 于是「帧被 2 个连接收下」被读成「已推送」，刷新后什么都不剩（主人报的正是
|
||
// 「后台思考的主动推送一个都没推出来」）。另一半风险是**双份** —— IM 适配器
|
||
// 出站消息本来就会被镜像回主会话（channelbridge/proactive.go 的 persistToMain），
|
||
// 若活跃路由也写一次、而一条推送同时走了两条路，主会话里就会出现两条。
|
||
// 因此这里两个路由各自跑一遍，逐个断言「恰好一条」，并额外断言
|
||
// 一条推送只会触发**一路**写入。
|
||
|
||
// proactiveStoreMsg 构造一条最小化的自主主动消息，等同于 backthink 的
|
||
// buildProactiveMessage 在无渠道决策（通用哨兵值）时的产物。
|
||
func proactiveStoreMsg(content string) adapter.UnifiedMessage {
|
||
return adapter.UnifiedMessage{
|
||
Platform: "cyrene-backthink",
|
||
ChannelType: adapter.ChannelDirect,
|
||
Content: content,
|
||
ContentType: adapter.ContentText,
|
||
Sender: adapter.Actor{Role: adapter.RoleSystem, DisplayName: "Cyrene"},
|
||
}
|
||
}
|
||
|
||
// mainSessionRows 读回主会话的落库行（最新优先）。
|
||
func mainSessionRows(t *testing.T, store *history.Store) []history.Message {
|
||
t.Helper()
|
||
rows, err := store.List(context.Background(), history.MainSessionID, 0, 50, "")
|
||
if err != nil {
|
||
t.Fatalf("list main session: %v", err)
|
||
}
|
||
return rows
|
||
}
|
||
|
||
// TestHub_SendProactive_PersistsAssistantMessageToMainSession 验证活跃客户端
|
||
// 路由把主动消息写成主会话里的一条 assistant 消息（「气泡恒落」的根：刷新后
|
||
// 仍在），且发出去的帧带上落库行的权威 id/时间戳 —— 客户端据此把实时气泡
|
||
// 对齐到历史行，而不是让它刷新即丢。
|
||
func TestHub_SendProactive_PersistsAssistantMessageToMainSession(t *testing.T) {
|
||
t.Parallel()
|
||
store := newHistoryStoreForGateway(t)
|
||
hb := newTestHub(nil, nil)
|
||
hb.history = store
|
||
owner := newOwnerConn(hb, 8)
|
||
hb.register(owner)
|
||
|
||
delivery, err := hb.SendProactive(context.Background(), proactiveStoreMsg("记得喝水哦"))
|
||
if err != nil {
|
||
t.Fatalf("SendProactive err = %v", err)
|
||
}
|
||
if delivery.Connections != 1 {
|
||
t.Fatalf("connections = %d, want 1", delivery.Connections)
|
||
}
|
||
if !delivery.Persisted {
|
||
t.Fatalf("Persisted = false, want true (主动消息必须落进主会话，否则刷新就没了)")
|
||
}
|
||
|
||
rows := mainSessionRows(t, store)
|
||
if len(rows) != 1 {
|
||
t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条（>1 = 双份）", len(rows))
|
||
}
|
||
row := rows[0]
|
||
if row.Role != history.RoleAssistant {
|
||
t.Fatalf("role = %q, want %q（她单方面开口的话，绝不能冒充主人的发言）", row.Role, history.RoleAssistant)
|
||
}
|
||
if row.SessionID != history.MainSessionID {
|
||
t.Fatalf("session = %q, want %q", row.SessionID, history.MainSessionID)
|
||
}
|
||
// 正文经共享规整器（turnreply.NormalizeParagraphs）：纯散文被包成一个 `<p>`
|
||
// ——与 Web 回合 / IM 出站 / IM 侧主动推送同一份口径（M25：本路由曾绕过它）。
|
||
if row.Content != "<p>记得喝水哦</p>" {
|
||
t.Fatalf("content = %q, want the normalized proactive text", row.Content)
|
||
}
|
||
if row.TurnID != "" {
|
||
t.Fatalf("turn_id = %q, want empty（主动推送不属于任何一路回合）", row.TurnID)
|
||
}
|
||
|
||
ev := readFakeFrame(t, owner)
|
||
if ev.Type != wsProactiveMessage {
|
||
t.Fatalf("frame type = %q, want %q", ev.Type, wsProactiveMessage)
|
||
}
|
||
data, ok := ev.Data.(map[string]any)
|
||
if !ok {
|
||
t.Fatalf("frame data type = %T, want map[string]any", ev.Data)
|
||
}
|
||
if data["id"] != row.ID {
|
||
t.Fatalf("frame id = %v, want the persisted row id %q", data["id"], row.ID)
|
||
}
|
||
if got, _ := data["createdAt"].(float64); int64(got) != row.CreatedAt {
|
||
t.Fatalf("frame createdAt = %v, want %d", data["createdAt"], row.CreatedAt)
|
||
}
|
||
}
|
||
|
||
// failingHistory 让 AppendFull 失败，用于验证「落库失败**可见**」：帧照发
|
||
// （实时气泡不能丢）、错误如实返回、Persisted=false（绝不谎报已推送）。
|
||
type failingHistory struct {
|
||
HistoryStore
|
||
err error
|
||
}
|
||
|
||
func (f failingHistory) AppendFull(context.Context, history.Message) (history.Message, error) {
|
||
return history.Message{}, f.err
|
||
}
|
||
|
||
// TestHub_SendProactive_PersistFailureIsVisible 验证落库失败时不做假成功：
|
||
// 错误被返回、Persisted=false、Detail 的措辞由调用方如实反映，而帧仍然发出。
|
||
func TestHub_SendProactive_PersistFailureIsVisible(t *testing.T) {
|
||
t.Parallel()
|
||
hb := newTestHub(nil, nil)
|
||
hb.history = failingHistory{err: errors.New("db down")}
|
||
owner := newOwnerConn(hb, 8)
|
||
hb.register(owner)
|
||
|
||
delivery, err := hb.SendProactive(context.Background(), proactiveStoreMsg("在吗"))
|
||
if err == nil {
|
||
t.Fatalf("err = nil, want the persist failure to be reported")
|
||
}
|
||
if delivery.Persisted {
|
||
t.Fatalf("Persisted = true, want false（没落库就是没落库）")
|
||
}
|
||
if delivery.Connections != 1 {
|
||
t.Fatalf("connections = %d, want 1（落库失败不该吞掉实时帧）", delivery.Connections)
|
||
}
|
||
var frameData any
|
||
if ev := readFakeFrame(t, owner); ev.Type != wsProactiveMessage {
|
||
t.Fatalf("frame type = %q, want %q", ev.Type, wsProactiveMessage)
|
||
} else {
|
||
frameData = ev.Data
|
||
}
|
||
data, ok := frameData.(map[string]any)
|
||
if !ok {
|
||
t.Fatalf("frame data type = %T, want map[string]any", frameData)
|
||
}
|
||
if _, hasID := data["id"]; hasID {
|
||
t.Fatalf("frame carries id %v, want the key omitted（没有落库行就没有权威 id）", data["id"])
|
||
}
|
||
}
|
||
|
||
// fakeIMAdapter 是 IM 路由测试所需的最小适配器（只用到连接态与出站）。
|
||
type fakeIMAdapter struct {
|
||
platform string
|
||
sends []adapter.UnifiedMessage
|
||
}
|
||
|
||
func (f *fakeIMAdapter) Name() string { return f.platform }
|
||
func (f *fakeIMAdapter) Config() map[string]any { return map[string]any{} }
|
||
func (f *fakeIMAdapter) Connect(context.Context) error { return nil }
|
||
func (f *fakeIMAdapter) Disconnect(context.Context) error { return nil }
|
||
func (f *fakeIMAdapter) IsConnected() bool { return true }
|
||
func (f *fakeIMAdapter) HealthCheck(context.Context) error { return nil }
|
||
func (f *fakeIMAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) {
|
||
<-ctx.Done()
|
||
}
|
||
func (f *fakeIMAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error {
|
||
f.sends = append(f.sends, resp.Messages...)
|
||
return nil
|
||
}
|
||
func (f *fakeIMAdapter) SendProactive(_ context.Context, msg adapter.UnifiedMessage) error {
|
||
f.sends = append(f.sends, msg)
|
||
return nil
|
||
}
|
||
func (f *fakeIMAdapter) SetMessageHandler(adapter.MessageHandler) {}
|
||
func (f *fakeIMAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() }
|
||
func (f *fakeIMAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
// stubBridgeChatter 满足 channelbridge.Chatter（本测试不会跑回合）。
|
||
type stubBridgeChatter struct{}
|
||
|
||
func (stubBridgeChatter) Run(context.Context, string, []model.Message) (*loop.Result, error) {
|
||
return &loop.Result{Text: "hi"}, nil
|
||
}
|
||
|
||
// stubBridgeSessions 满足 channelbridge.SessionStore（本测试不会走入站路径）。
|
||
type stubBridgeSessions struct{}
|
||
|
||
func (stubBridgeSessions) GetOrCreate(context.Context, adapter.UnifiedMessage, *user.OwnerProfile) (session.Session, error) {
|
||
return session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}, nil
|
||
}
|
||
|
||
// newProactiveTestBridge 装配一条接在**真历史存储**上的 IM 桥（一个已连接、
|
||
// 在白名单里的主人渠道），使 IM 路由的落库能被真库断言。
|
||
func newProactiveTestBridge(t *testing.T, store *history.Store, onPersisted *[]history.Message) (*channelbridge.Bridge, *fakeIMAdapter) {
|
||
t.Helper()
|
||
owner := &user.OwnerProfile{Accounts: []user.OwnerAccount{{Platform: "telegram", AccountID: "100"}}}
|
||
var hook func(history.Message)
|
||
if onPersisted != nil {
|
||
hook = func(m history.Message) { *onPersisted = append(*onPersisted, m) }
|
||
}
|
||
b, err := channelbridge.New(channelbridge.Options{
|
||
Log: gwTestLogger(),
|
||
Loop: stubBridgeChatter{},
|
||
Sessions: stubBridgeSessions{},
|
||
Owner: owner,
|
||
History: store,
|
||
OnPersisted: hook,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("new bridge: %v", err)
|
||
}
|
||
ad := &fakeIMAdapter{platform: "telegram"}
|
||
b.Add(ad)
|
||
return b, ad
|
||
}
|
||
|
||
// TestChannelBridge_SendProactive_MirrorsExactlyOnce 是**现状核查**的可执行
|
||
// 证据：IM 路由本来就会把主动消息镜像回主会话（且只镜像一次）。因此在这条
|
||
// 路由上再加一次落库就是双份 —— 活跃路由的落库正是靠「两条路由互斥」避开它的。
|
||
func TestChannelBridge_SendProactive_MirrorsExactlyOnce(t *testing.T) {
|
||
t.Parallel()
|
||
store := newHistoryStoreForGateway(t)
|
||
var persisted []history.Message
|
||
bridge, ad := newProactiveTestBridge(t, store, &persisted)
|
||
|
||
if err := bridge.SendProactive(context.Background(), proactiveStoreMsg("记得喝水哦")); err != nil {
|
||
t.Fatalf("SendProactive err = %v", err)
|
||
}
|
||
if len(ad.sends) != 1 {
|
||
t.Fatalf("IM sends = %d, want 1", len(ad.sends))
|
||
}
|
||
rows := mainSessionRows(t, store)
|
||
if len(rows) != 1 {
|
||
t.Fatalf("主会话里应恰好一条主动消息，得到 %d 条（>1 = 双份）", len(rows))
|
||
}
|
||
if rows[0].Role != history.RoleAssistant || !strings.Contains(rows[0].Content, "记得喝水哦") {
|
||
t.Fatalf("row = %+v, want an assistant message carrying the proactive text", rows[0])
|
||
}
|
||
if len(persisted) != 1 || persisted[0].ID != rows[0].ID {
|
||
t.Fatalf("OnPersisted 应恰好收到刚落地的那一条（Web 端据此实时镜像），得到 %+v", persisted)
|
||
}
|
||
}
|
||
|
||
// TestProactiveRoutes_EachRouteAddsExactlyOneRow 把两条路由对**同一条**主动
|
||
// 消息各跑一次：主会话里应恰好两条（每条路由各一条）。它同时钉住两件事 ——
|
||
// ① 两条路由都真的落库（不再有「只发帧不落库」的那条）；② 没有一条路由写两份
|
||
// （若活跃路由也走 IM 镜像那一次，这里会是 3 条）。
|
||
func TestProactiveRoutes_EachRouteAddsExactlyOneRow(t *testing.T) {
|
||
t.Parallel()
|
||
store := newHistoryStoreForGateway(t)
|
||
hb := newTestHub(nil, nil)
|
||
hb.history = store
|
||
hb.register(newOwnerConn(hb, 8))
|
||
bridge, _ := newProactiveTestBridge(t, store, nil)
|
||
|
||
msg := proactiveStoreMsg("两条路各跑一次")
|
||
if _, err := hb.SendProactive(context.Background(), msg); err != nil {
|
||
t.Fatalf("active route err = %v", err)
|
||
}
|
||
if got := len(mainSessionRows(t, store)); got != 1 {
|
||
t.Fatalf("活跃路由之后主会话有 %d 条，want 1", got)
|
||
}
|
||
if err := bridge.SendProactive(context.Background(), msg); err != nil {
|
||
t.Fatalf("im route err = %v", err)
|
||
}
|
||
rows := mainSessionRows(t, store)
|
||
if len(rows) != 2 {
|
||
t.Fatalf("两条路由各跑一次后主会话有 %d 条，want 2（3 条 = 有一条路由写了双份）", len(rows))
|
||
}
|
||
for _, row := range rows {
|
||
if row.Role != history.RoleAssistant {
|
||
t.Fatalf("row role = %q, want assistant", row.Role)
|
||
}
|
||
}
|
||
}
|