fix(backthink): 她主动找你终于看得见 —— 主动消息落主会话 + 前端消费 + 非前台浏览器通知
CI / Backend (build / vet / test / cross-compile) (push) Has been cancelled
CI / Frontend (tsc + vite build) (push) Has been cancelled

修主人报的硬故障「后台思考的主动推送一个都没有推出来」(根因两条,缺一不可):
- 后端 gateway/proactive.go 只广播 proactive_message 帧、**从不写会话历史** ⇒ 刷新补不回来
- 前端**零消费点** ⇒ 帧被静默丢弃(既无气泡也无通知)

后端:新增 proactive_persist.go(**先落库、再发帧**;web/assistant/mode=chat;未接线时空操作且不谎称已落库)
  载荷新增 id/createdAt(落库行权威 id);PushResult.Detail 三态如实分开报告「投递到几个连接」与「是否落库」
前端:handlers/proactive.ts(气泡**恒落** + 按落库 id 去重 + **非前台才补通知** + 拒绝授权/畸形载荷可见降级)
  + lib/notify/proactive.ts(纯逻辑 isBackground/decideProactiveNotice/有界去重表)
  + handlers/task.ts(另三个被丢的帧 task_done/automation_result/subagent_result)
  + ws/resync.ts + useWsResync.ts(**重连重跑本页引导读取**,首次连接不跑)
双份核查(带三层证据):现状不双份 —— 路由分叉「一次 Push 至多一个 sink 成功」,且 IM 路由本来就把主动消息镜像回主会话恰好一次
契约 v0.72;变异实测:只发帧不落库 / 写成 user 角色 / 去重失效 / 前台也弹 / IM 成功再打活跃 ⇒ 全部打红
This commit is contained in:
AskaEth committed 2026-10-05 16:02:36 +08:00
1 parent 1bf85a955b
commit 84c124f5a9
18 files changed
+1545 -46

No files matched your search

+9 -7
View File
@@ -72,19 +72,21 @@ func (p *mockPusher) count() int { p.mu.Lock(); defer p.mu.Unlock(); return p.ca
// mockActive 实现 ActiveClient。
type mockActive struct {
mu sync.Mutex
n int
err error
calls int
last adapter.UnifiedMessage
mu sync.Mutex
n int
persisted bool
err error
calls int
last adapter.UnifiedMessage
}
func (m *mockActive) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) {
func (m *mockActive) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (ProactiveDelivery, error) {
m.mu.Lock()
m.calls++
m.last = msg
n, persisted, err := m.n, m.persisted, m.err
m.mu.Unlock()
return m.n, m.err
return ProactiveDelivery{Connections: n, Persisted: persisted}, err
}
// mockIMAdapter 实现 IMAdapter。
+57 -11
View File
@@ -69,13 +69,35 @@ func (p PushResult) String() string {
return fmt.Sprintf("pushed=%v route=%s bypass=%v detail=%q", p.Pushed, p.Route, p.Bypassed, p.Detail)
}
// ProactiveDelivery 是一次活跃客户端投递的结果(切片 P0「她主动找你」)。
// 它把**两件本就独立**的事分开报告:帧送到了几条连接,以及这条消息有没有
// 成为主会话里的一条真实消息。
//
// 为什么要分开:修复前活跃客户端路由只广播 WS 帧、不落库,于是面板上的
// 「delivered to 2 active client(s)」被读成「已推送」,而刷新后什么都没剩
// —— 那正是主人报的「后台思考的主动推送一个都没有推出来」。落库与投递各自
// 成败,记录里就必须各自可见。
type ProactiveDelivery struct {
// Connections 是接受该帧的在线 owner/admin 连接数。0 意味着此刻没有
// 在线目标;它**不代表消息丢了**——消息可能已经落库(见 Persisted)。
Connections int
// Persisted 报告该消息是否已成为主会话历史(`history.MainSessionID`)里的
// 一条真实消息,因而刷新/重载后仍在。
Persisted bool
}
// ActiveClient 把主动消息投递给所有当前活跃的 Web/原生
// 客户端,并报告有多少个收到。它是首选的推送目标。
// 客户端,并报告投递与落库两件事。它是首选的推送目标。
// 生产中由 gateway 的 WebSocket hub 实现;测试注入 mock。
type ActiveClient interface {
// SendProactive 把 msg 推给每个活跃客户端,并返回
// 接受它的客户端数量。
SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error)
// ProactiveDelivery。
//
// 实现契约(切片 P0):**先落库、再发帧**,且发帧的载荷带上落库行的 id ——
// 客户端据此把实时气泡对齐到历史行(按 id 去重、刷新后仍在)。落库失败时
// 仍要发出帧(实时气泡不能丢),但必须把错误返回且 Persisted=false,
// 让调用方如实记录「发出去了、但没留下记录」。
SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (ProactiveDelivery, error)
}
// IMAdapter 是 IM 桥降级:它把主动消息经 IM
@@ -246,19 +268,43 @@ func (p *pusher) exportAsync(msg adapter.UnifiedMessage) {
}()
}
// pushActive 只投递给活跃 Web/原生客户端。
// pushActive 只投递给活跃 Web/原生客户端。它按 ProactiveDelivery 如实报告
// 「送到了几条连接」与「有没有落进主会话」(切片 P0):投递数为 0 时消息仍可能
// 已经是一条真实记录,两者绝不互相冒充。
func (p *pusher) pushActive(ctx context.Context, msg adapter.UnifiedMessage) PushResult {
if p.active != nil {
n, err := p.active.SendProactive(ctx, msg)
if err != nil {
p.log.Warn("backthink active client push failed", "error", err)
} else if n > 0 {
return PushResult{Pushed: true, Route: RouteActiveClient, Detail: fmt.Sprintf("delivered to %d active client(s)", n)}
}
if p.active == nil {
return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client"}
}
d, err := p.active.SendProactive(ctx, msg)
if err != nil {
p.log.Warn("backthink active client push failed", "error", err)
}
if d.Connections > 0 {
return PushResult{Pushed: true, Route: RouteActiveClient, Detail: activeDeliveryDetail(d)}
}
if d.Persisted {
// 此刻没有在线客户端,但这条话已经成为主会话里的一条真实记录:打开/
// 刷新页面就能看到,只是没有实时帧可发。如实分别报告这两件事,面板才不会
// 把「没推出去」与「没留下」混为一谈。
return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client, recorded in main session"}
}
if err != nil {
return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client", Err: err.Error()}
}
return PushResult{Pushed: false, Route: RouteNone, Detail: "no active client"}
}
// activeDeliveryDetail 是活跃客户端路由的 Detail 措辞:投递数与「有没有落进
// 主会话」两件事实都在里面,且**绝不**只说「已推送」(那正是被误读成「刷新还在」
// 的原话)。
func activeDeliveryDetail(d ProactiveDelivery) string {
state := "not persisted to main session"
if d.Persisted {
state = "persisted to main session"
}
return fmt.Sprintf("delivered to %d active client(s), %s", d.Connections, state)
}
// pushIM 只投递给 IM 桥。
func (p *pusher) pushIM(ctx context.Context, msg adapter.UnifiedMessage) PushResult {
if p.im == nil {
@@ -0,0 +1,117 @@
package backthink
import (
"context"
"errors"
"strings"
"testing"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
)
// 本文件锁定切片 P0「她主动找你」的**活跃客户端路由口径**:
//
// - PushResult.Detail 必须把「送到了几条连接」与「有没有落进主会话」**分开**
// 报告 —— 修复前那句「delivered to 2 active client(s)」被读者当成「已推送
// (刷新也在)」,而当时一行都没落库;
// - 一次推送**至多让一个 sink 成功**(IM 桥与活跃客户端互斥),这正是
// 「主会话里不会出现双份」的结构性依据:两条路由各自落库一次,而一次推送
// 只走其中一条。
// TestPusher_ActiveDetailReportsPersistence 验证 Detail 分别报告投递数与落库。
func TestPusher_ActiveDetailReportsPersistence(t *testing.T) {
t.Parallel()
tests := []struct {
name string
active *mockActive
wantPushed bool
wantIn string
wantNotIn string
}{
{
name: "投递成功且已落库",
active: &mockActive{n: 2, persisted: true},
wantPushed: true,
wantIn: "persisted to main session",
wantNotIn: "not persisted",
},
{
name: "投递成功但未落库(故障可见)",
active: &mockActive{n: 2, persisted: false},
wantPushed: true,
wantIn: "not persisted to main session",
},
{
name: "无在线客户端但已落库",
active: &mockActive{n: 0, persisted: true},
wantPushed: false,
wantIn: "recorded in main session",
},
}
for _, tc := range tests {
tc := tc
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
p := newPusher(tc.active, nil, &mockGate{allow: true}, discardLog())
res := p.Push(context.Background(), PushRequest{
ChannelHint: channelWeb,
Message: adapter.UnifiedMessage{Content: "记得喝水哦"},
})
if res.Pushed != tc.wantPushed {
t.Fatalf("Pushed = %v, want %v (res=%+v)", res.Pushed, tc.wantPushed, res)
}
if !strings.Contains(res.Detail, tc.wantIn) {
t.Fatalf("Detail = %q, want it to contain %q", res.Detail, tc.wantIn)
}
if tc.wantNotIn != "" && strings.Contains(res.Detail, tc.wantNotIn) {
t.Fatalf("Detail = %q, must not contain %q", res.Detail, tc.wantNotIn)
}
})
}
}
// TestPusher_OnePushSucceedsOnAtMostOneSink 验证一次推送**至多让一个 sink 成功**
// —— 互斥是「主会话里不会双份」的结构性依据。IM 失败后的活跃客户端降级是同一
// 次推送内的兜底:失败的 IM 什么都没写,因此整体仍只有一条记录。
func TestPusher_OnePushSucceedsOnAtMostOneSink(t *testing.T) {
t.Parallel()
tests := []struct {
name string
hint string
imErr error
activeN int
wantIMCalls int
wantActiveCall int
}{
{name: "web 提示只打活跃客户端", hint: channelWeb, activeN: 1, wantIMCalls: 0, wantActiveCall: 1},
{name: "auto 提示 IM 可用时只打 IM", hint: channelAuto, activeN: 2, wantIMCalls: 1, wantActiveCall: 0},
{name: "IM 失败时降级到活跃客户端", hint: channelAuto, imErr: errors.New("no owner IM channel"), activeN: 1, wantIMCalls: 1, wantActiveCall: 1},
}
for _, tc := range tests {
tc := tc
t.Run(tc.name, func(t *testing.T) {
t.Parallel()
active := &mockActive{n: tc.activeN, persisted: true}
im := &mockIMAdapter{err: tc.imErr}
p := newPusher(active, im, &mockGate{allow: true}, discardLog())
res := p.Push(context.Background(), PushRequest{
ChannelHint: tc.hint,
Message: adapter.UnifiedMessage{Content: "记得喝水哦"},
})
if im.count() != tc.wantIMCalls || active.calls != tc.wantActiveCall {
t.Fatalf("sink calls: im=%d (want %d) active=%d (want %d)",
im.count(), tc.wantIMCalls, active.calls, tc.wantActiveCall)
}
// 成功的 sink 至多一个:能落库的路径因此也至多一条。
imOK := tc.imErr == nil && tc.wantIMCalls == 1
activeOK := tc.activeN > 0 && tc.wantActiveCall == 1
if imOK && activeOK {
t.Fatalf("both sinks succeeded in one push (res=%+v)", res)
}
if !imOK && !activeOK && res.Pushed {
t.Fatalf("Pushed = true without a successful sink (res=%+v)", res)
}
})
}
}
+44 -13
View File
@@ -8,6 +8,8 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
"git.yeij.top/AskaEth/Cyrene_v2/internal/services/backthink"
)
// wsProactiveMessage 是自主主动推送的 WS 事件类型
@@ -27,21 +29,46 @@ type proactivePayload struct {
Sender string `json:"sender,omitempty"`
Timestamp time.Time `json:"timestamp,omitempty"`
MessageID string `json:"messageId,omitempty"`
// ID 是这条主动消息在**主会话历史**里的权威行 id(切片 P0「气泡恒落」)。
// 客户端据此把实时气泡对齐到落库行:与握手回放 / REST 历史拉取按 id 去重
// (同一条绝不渲染两个气泡),刷新后仍在。落库失败时该键**省略** ——
// 客户端因此知道这条只有实时帧、没有历史行,而不是拿到一个悬空的 id。
ID string `json:"id,omitempty"`
// CreatedAt 是落库行的时间戳(Unix 毫秒),与 ID 同时出现/同时省略。它是
// 权威的排序时间(与 `timestamp` 是同一次推送的两个视角:前者是历史行的
// 时间,后者是这条统一消息自身的时间)。
CreatedAt int64 `json:"createdAt,omitempty"`
}
// SendProactive 把 msg 投递给每个存活的 owner/admin 连接,并返回
// 有多少个接受了它。计数为零意味着没有 owner/admin 客户端在线,
// 因此调用方回退到 IM bridge。只有 owner/admin 连接会被作为
// 目标(红线:主动推送仅 owner);viewer 或未认证连接
// 绝不被触达。它绝不阻塞:发送缓冲满时对该
// 慢客户端丢弃该帧,而不是卡住后台思考推送。
func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) {
// SendProactive 把 msg 投递给每个存活的 owner/admin 连接,并返回投递与落库
// 两件事的结果(切片 P0,`backthink.ProactiveDelivery`)。
//
// 顺序是刻意的 —— **先落库、再发帧**:主动消息必须成为主会话
// (`history.MainSessionID`)里的一条真实消息(assistant 角色),否则刷新/
// 重载后什么都补不回来(这正是「气泡恒落」的根);而帧要带上落库行的 id,
// 客户端才能把实时气泡对齐到历史行。
//
// 落库失败**不静默**:记一条带来源的 WARN,并把错误返回给调用方,同时
// Persisted=false —— 帧照发(实时气泡不能丢),但记录里绝不说「已推送」。
//
// 计数为零意味着没有 owner/admin 客户端在线;消息此时仍会作为一条真实记录
// 留在主会话里(下次打开就能看到),调用方据此回退到 IM bridge 把它**投出去**。
// 只有 owner/admin 连接会被作为目标(红线:主动推送仅 owner);viewer 或未认证
// 连接绝不被触达。它绝不阻塞:发送缓冲满时对该慢客户端丢弃该帧,
// 而不是卡住后台思考推送。
func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (backthink.ProactiveDelivery, error) {
if err := ctx.Err(); err != nil {
return 0, err
return backthink.ProactiveDelivery{}, err
}
b, err := json.Marshal(wsEvent{Type: wsProactiveMessage, Data: proactivePayloadFrom(msg)})
persisted, persistErr := h.persistProactiveToMain(ctx, msg)
if persistErr != nil {
h.log.Warn("persist proactive message to main session failed",
"err", persistErr, "platform", msg.Platform)
}
b, err := json.Marshal(wsEvent{Type: wsProactiveMessage, Data: proactivePayloadFrom(msg, persisted)})
if err != nil {
return 0, fmt.Errorf("gateway: marshal proactive: %w", err)
return backthink.ProactiveDelivery{Persisted: persisted.ID != ""},
fmt.Errorf("gateway: marshal proactive: %w", err)
}
h.mu.RLock()
defer h.mu.RUnlock()
@@ -56,17 +83,19 @@ func (h *hub) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (in
default:
}
}
return n, nil
return backthink.ProactiveDelivery{Connections: n, Persisted: persisted.ID != ""}, persistErr
}
// Server.SendProactive 通过委托给 hub 满足后台思考的 ActiveClient
// 接缝,因此组合根能把网关自身交给汇聚点设置器。
func (s *Server) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (int, error) {
func (s *Server) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) (backthink.ProactiveDelivery, error) {
return s.hub.SendProactive(ctx, msg)
}
// proactivePayloadFrom 把 UnifiedMessage 投影为消毒后的线上载荷。
func proactivePayloadFrom(msg adapter.UnifiedMessage) proactivePayload {
// persisted 是刚落库的主会话消息(未接线历史存储或落库失败时为零值),
// 它的 id/时间戳随之进入载荷,使客户端能把实时气泡对齐到历史行。
func proactivePayloadFrom(msg adapter.UnifiedMessage, persisted history.Message) proactivePayload {
return proactivePayload{
Platform: msg.Platform,
ChannelID: msg.ChannelID,
@@ -76,6 +105,8 @@ func proactivePayloadFrom(msg adapter.UnifiedMessage) proactivePayload {
Sender: msg.Sender.DisplayName,
Timestamp: msg.Timestamp,
MessageID: msg.MessageID,
ID: persisted.ID,
CreatedAt: persisted.CreatedAt,
}
}
@@ -0,0 +1,51 @@
package gateway
import (
"context"
"fmt"
"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history"
)
// 本文件承载主动推送的**落库**那一半(切片 P0「她主动找你」)。它从
// proactive.go 拆出来,好让「帧长什么样」与「记录怎么写」各自成篇
// (AGENTS §10 模块按职责拆源文件)。
//
// 为什么必须有这一半:修复前活跃客户端路由只把 `proactive_message` 帧
// 广播给在线连接(fire-and-forget),**一行都不写** ⇒ 刷新/重载后什么都补不回来,
// 界面既不显示气泡也补不出历史。主人的原话是「后台思考的主动推送一个都没推出来」。
// 主动消息是**她真实说过的一句话**,因此它必须与 IM 镜像路径走同一条去向:
// 主会话历史里的一条 assistant 消息。
// persistProactiveToMain 把一条主动消息写进主会话(`history.MainSessionID`),
// 并返回落库后的完整消息(含权威 id/时间戳),供帧载荷携带。
//
// 三条口径:
// - **角色恒为 assistant**:这是她单方面开口说的话,不是主人的输入 —— 塞成
// user 角色会污染「谁说了什么」,也会在历史里冒充主人发言(红线)。
// - **正文用调用方已经净化过的文本**(backthink 的 stripInternalMarkers 出口),
// 系统元数据(决策标记、渠道选择)绝不进正文,也绝不新增任何隐藏字段 ——
// 注入面因此与「气泡里看到的那句话」逐字相同。
// - **turn_id 留空**:主动推送不属于任何一路回合(她单方面开口,没有对应的
// 入站回合),空值诚实地表示这一点(与 channelbridge 的同一取舍)。
//
// 未接线历史存储时它是空操作并返回零值(`Persisted=false`)—— 与编排器 /
// IM 镜像路径同一取舍:没存储不是错误,但**绝不能谎称已落库**。写库失败返回
// 包装后的错误,由调用方让它可见(WARN + 返回给推送服务)。
func (h *hub) persistProactiveToMain(ctx context.Context, msg adapter.UnifiedMessage) (history.Message, error) {
if h.history == nil {
return history.Message{}, nil
}
full, err := h.history.AppendFull(ctx, history.Message{
SessionID: history.MainSessionID,
Role: history.RoleAssistant,
Content: msg.Content,
Platform: msg.Platform,
Mode: modeChat,
})
if err != nil {
return history.Message{}, fmt.Errorf("gateway: persist proactive to main session: %w", err)
}
return full, nil
}
@@ -0,0 +1,280 @@
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)
}
if row.Content != "记得喝水哦" {
t.Fatalf("content = %q, want the proactive text verbatim", 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)
}
}
}
+6 -6
View File
@@ -39,8 +39,8 @@ func TestHub_SendProactive_OwnerAndAdmin(t *testing.T) {
if err != nil {
t.Fatalf("SendProactive err = %v", err)
}
if n != 2 {
t.Fatalf("count = %d, want 2 (owner+admin; viewer and bare excluded)", n)
if n.Connections != 2 {
t.Fatalf("count = %d, want 2 (owner+admin; viewer and bare excluded)", n.Connections)
}
for _, c := range []*wsConn{owner, admin} {
@@ -82,8 +82,8 @@ func TestHub_SendProactive_NoOnlineOwner(t *testing.T) {
if err != nil {
t.Fatalf("SendProactive err = %v", err)
}
if n != 0 {
t.Fatalf("count = %d, want 0", n)
if n.Connections != 0 {
t.Fatalf("count = %d, want 0", n.Connections)
}
}
@@ -113,8 +113,8 @@ func TestServer_SendProactive_DelegatesToHub(t *testing.T) {
if err != nil {
t.Fatalf("SendProactive err = %v", err)
}
if n != 1 {
t.Fatalf("count = %d, want 1", n)
if n.Connections != 1 {
t.Fatalf("count = %d, want 1", n.Connections)
}
ev := readFakeFrame(t, c)
if ev.Type != "proactive_message" {