P1-1(channelbridge 的 -race 门仍红,红在用例不确定性而不是竞态): - coalesceTimers 改成**截止时刻**语义:每次 after(d) 记 at = 注入时钟 + d, 推钟只把时钟推进到「当前最早的截止时刻」再触发它。旧写法按 FIFO 逐个触发并 按请求时长累加,实测 12 条消息内攒到 5 个废弃定时器把时钟推 +10s(窗口 2s、 绝对上限 6s)⇒ 循环头走「到点直接发车、不 takeAll」的分支,队列尾部落到下一批。 - 推钟前强制「车道排空」这条不变量(pump/advanceAndFire/waitIdle): 时钟只表示时间过去,它一转就可能跳过 takeAll,因此先等 worker 把货收进本批。 - sendSettled:投递后按**序号**等「这次投递之后新挂的定时器」(旧的都已被 worker 放弃、再没人监听),供「投递与推钟交替」的用例(每 500ms 一条的绝对上限用例)。 - fireLatest/advanceAndFire 带 done 谓词重试,避免「登记与触发之间续臂」导致 推了钟却没人醒;waitIdle 一边等一边推窗口(worker 停在窗口里时永远走不到退休)。 - 用例侧只补相位与等待,不动断言:TwelveBurst 读历史前等 assistant 行落库; 准入用例断言「最新一次续臂」(原来按下标取 d[1],隐含臂数假设); stale 用例两条都投完再推时钟(车道不会在两条之间退休)。 P2-1(替身锁无法用变异证实)选 A:删掉 stubChatter/stubSessions 的锁与访问器, 退回原字段访问,并写清理由 —— 21 个用例全走同步便捷路径 (b.process → processFrom → runBatch → Loop.Run),DEV HEAD 上 -race 数十次跑 (含把锁删掉的变异)DATA RACE 恒为 0;若将来从车道 worker 触碰这些字段必须加锁 并补 -race 用例。 证据:go test -race -count=3 ./internal/services/channelbridge/ 连跑 6 轮 18/18 全绿、DATA RACE=0;coalesce 子集 -count=80 全绿;整包 -race -count=20 全绿; 全树 go build/vet/test -count=1 全绿。
670 lines
25 KiB
Go
670 lines
25 KiB
Go
package channelbridge
|
||
|
||
import (
|
||
"context"
|
||
"strings"
|
||
"sync"
|
||
"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/user"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/kook"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/webhook"
|
||
)
|
||
|
||
// fakeAdapter 是最小的 PlatformAdapter,记录出站发送,
|
||
// 使桥接器的回复路径无需网络 I/O 即可断言。
|
||
type fakeAdapter struct {
|
||
name string
|
||
connected bool
|
||
cfg map[string]any
|
||
sends []adapter.UnifiedMessage
|
||
handler adapter.MessageHandler
|
||
sendErr error
|
||
}
|
||
|
||
func (f *fakeAdapter) Name() string { return f.name }
|
||
func (f *fakeAdapter) Config() map[string]any {
|
||
if f.cfg != nil {
|
||
return f.cfg
|
||
}
|
||
return map[string]any{"bot_token": "configured"}
|
||
}
|
||
func (f *fakeAdapter) Connect(context.Context) error { f.connected = true; return nil }
|
||
func (f *fakeAdapter) Disconnect(context.Context) error {
|
||
f.connected = false
|
||
return nil
|
||
}
|
||
func (f *fakeAdapter) IsConnected() bool { return f.connected }
|
||
func (f *fakeAdapter) HealthCheck(context.Context) error { return nil }
|
||
func (f *fakeAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) {
|
||
<-ctx.Done()
|
||
}
|
||
func (f *fakeAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error {
|
||
for _, m := range resp.Messages {
|
||
f.sends = append(f.sends, m)
|
||
}
|
||
return f.sendErr
|
||
}
|
||
func (f *fakeAdapter) SendProactive(_ context.Context, msg adapter.UnifiedMessage) error {
|
||
f.sends = append(f.sends, msg)
|
||
return f.sendErr
|
||
}
|
||
func (f *fakeAdapter) SetMessageHandler(fn adapter.MessageHandler) { f.handler = fn }
|
||
func (f *fakeAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() }
|
||
func (f *fakeAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { return nil, nil }
|
||
|
||
// stubChatter 记录传给 agent 循环的会话 id 与消息。
|
||
//
|
||
// 不加锁是**有意的**:用到它的 21 个用例全部走同步便捷路径
|
||
// (`b.process → processFrom → runBatch → Loop.Run`,见 inbound.go 的 processFrom 注释),
|
||
// 写与读发生在同一个 goroutine 上,没有跨 goroutine 的 happens-before 缺口
|
||
// (M33 实测:`-race` 数十次跑、含把锁删掉的变异,DATA RACE 恒为 0)。
|
||
// 🔴 若将来有用例把它放上车道(laneWorker)再读这些字段,**必须**加锁(sync.Mutex)
|
||
// 并补一条 -race 用例 —— 届时的竞态是真的,不能只靠运气。
|
||
type stubChatter struct {
|
||
sessionID string
|
||
messages []model.Message
|
||
runCtx context.Context
|
||
res *loop.Result
|
||
err error
|
||
}
|
||
|
||
func (s *stubChatter) Run(ctx context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) {
|
||
s.sessionID = sessionID
|
||
s.messages = msgs
|
||
s.runCtx = ctx
|
||
return s.res, s.err
|
||
}
|
||
|
||
// stubSessions 记录入站消息并返回固定会话。
|
||
//
|
||
// 与 stubChatter 同理:只在同步路径上被调用,因此不加锁;若将来从车道 worker 触碰,
|
||
// 必须补锁 + 补一条 -race 用例。
|
||
type stubSessions struct {
|
||
msg adapter.UnifiedMessage
|
||
owner *user.OwnerProfile
|
||
sess session.Session
|
||
err error
|
||
}
|
||
|
||
func (s *stubSessions) GetOrCreate(_ context.Context, msg adapter.UnifiedMessage, owner *user.OwnerProfile) (session.Session, error) {
|
||
s.msg = msg
|
||
s.owner = owner
|
||
if s.err != nil {
|
||
return session.Session{}, s.err
|
||
}
|
||
return s.sess, nil
|
||
}
|
||
|
||
// directMessage 是一条主人私聊消息:platform telegram / 发送者 100。它是否是主人由**白名单**
|
||
// 决定——夹具的主人档案(testOwnerProfile("telegram", "100"))把这个账号结构性地写进去
|
||
// (裁决 06.4.2:owner 车道只认白名单)。这里仍带上适配器上报的 role=owner 标记,用来断言
|
||
// 「白名单命中时适配器标记不影响判主」;反过来,未进白名单的账号即使带该标记也不是主人。
|
||
// newTestBridge 的单账号 stub session 据此报 Owner=true,与生产 SessionManager 的派生一致。
|
||
func directMessage() adapter.UnifiedMessage {
|
||
return adapter.UnifiedMessage{
|
||
Platform: "telegram",
|
||
ChannelID: "100",
|
||
ChannelType: adapter.ChannelDirect,
|
||
SenderID: "100",
|
||
Sender: adapter.Actor{Role: adapter.RoleOwner},
|
||
Content: "hello",
|
||
ContentType: adapter.ContentText,
|
||
}
|
||
}
|
||
|
||
func newTestBridge(t *testing.T) (*Bridge, *stubChatter, *stubSessions, *fakeAdapter) {
|
||
t.Helper()
|
||
chat := &stubChatter{res: &loop.Result{Text: "hi back"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
|
||
ad := &fakeAdapter{name: "telegram"}
|
||
|
||
// 主人私聊要进主会话,夹具必须把发送者结构性地写进白名单(owner 车道只认白名单,
|
||
// 裁决 06.4.2);适配器上报的 RoleOwner 标记本身不再判主。
|
||
b, err := New(Options{Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "100")})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
b.Add(ad)
|
||
return b, chat, sess, ad
|
||
}
|
||
|
||
func TestProcessRunsPipelineAndReplies(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b, chat, sess, ad := newTestBridge(t)
|
||
ctx := context.Background()
|
||
|
||
b.process(ctx, directMessage())
|
||
|
||
if sess.msg.SenderID != "100" {
|
||
t.Errorf("resolved sender = %q, want 100", sess.msg.SenderID)
|
||
}
|
||
// 主人私聊被路由进唯一的 Web 主会话(design/16 §4.4)。
|
||
if chat.sessionID != history.MainSessionID {
|
||
t.Errorf("loop session id = %q, want %q", chat.sessionID, history.MainSessionID)
|
||
}
|
||
// IM 平台上的主人私聊会镜像到主会话。来源由
|
||
// 最前面的一行系统上下文承载,用户文本保持不加装饰
|
||
// (design/16 §4.6.1):元数据绝不进入用户的自由文本通道。
|
||
var sys, userText string
|
||
for _, m := range chat.messages {
|
||
if m.Role == model.RoleSystem {
|
||
sys = m.Content[0].Text
|
||
} else if m.Role == model.RoleUser {
|
||
userText = m.Content[0].Text
|
||
}
|
||
}
|
||
if userText != "hello" {
|
||
t.Errorf("loop user text = %q, want undecorated 'hello'", userText)
|
||
}
|
||
if !strings.Contains(sys, "来源:telegram · 私聊") {
|
||
t.Errorf("loop origin system = %q, want telegram private origin", sys)
|
||
}
|
||
if len(ad.sends) != 1 {
|
||
t.Fatalf("send count = %d, want 1", len(ad.sends))
|
||
}
|
||
reply := ad.sends[0]
|
||
if reply.Content != "hi back" || reply.ContentType != adapter.ContentText {
|
||
t.Errorf("reply = %+v, want text 'hi back'", reply)
|
||
}
|
||
if reply.ChannelID != "100" || reply.Platform != "telegram" {
|
||
t.Errorf("reply target = %+v, want platform telegram channel 100", reply)
|
||
}
|
||
}
|
||
|
||
func TestProcessSkipsEmptyText(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b, chat, _, ad := newTestBridge(t)
|
||
msg := directMessage()
|
||
msg.Content = " "
|
||
|
||
b.process(context.Background(), msg)
|
||
|
||
if chat.sessionID != "" {
|
||
t.Errorf("loop ran for empty content, session = %q", chat.sessionID)
|
||
}
|
||
if len(ad.sends) != 0 {
|
||
t.Errorf("send count = %d, want 0 for empty content", len(ad.sends))
|
||
}
|
||
}
|
||
|
||
func TestProcessGroupMessageKeepsGroupTarget(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b, chat, _, ad := newTestBridge(t)
|
||
// 被点名(提及)的非主人群消息按 design/12 §6 策略二
|
||
// 会得到回复,因此它仍能覆盖群回复目标选择。
|
||
msg := adapter.UnifiedMessage{
|
||
Platform: "telegram",
|
||
ChannelID: "-20",
|
||
ChannelType: adapter.ChannelGroup,
|
||
GroupID: "-20",
|
||
GroupName: "Gang",
|
||
SenderID: "9",
|
||
Sender: adapter.Actor{Role: adapter.RoleMember},
|
||
Mentions: []adapter.Mention{{UserID: "9"}},
|
||
Content: "hello all",
|
||
ContentType: adapter.ContentText,
|
||
}
|
||
chat.res = &loop.Result{Text: "group ack"}
|
||
|
||
b.process(context.Background(), msg)
|
||
|
||
if len(ad.sends) != 1 {
|
||
t.Fatalf("send count = %d, want 1", len(ad.sends))
|
||
}
|
||
reply := ad.sends[0]
|
||
if reply.ChannelType != adapter.ChannelGroup || reply.GroupID != "-20" {
|
||
t.Errorf("group reply target = %+v, want group -20", reply)
|
||
}
|
||
}
|
||
|
||
// captureHistory 记录每次 Append,用于断言桥接器把主人私聊
|
||
// 消息写入 Web 主会话(design/16 §4.4)。
|
||
type captureHistory struct {
|
||
mu sync.Mutex
|
||
appends []history.Message
|
||
}
|
||
|
||
func (c *captureHistory) Append(_ context.Context, msg history.Message) error {
|
||
_, err := c.AppendFull(context.Background(), msg)
|
||
return err
|
||
}
|
||
|
||
func (c *captureHistory) AppendFull(_ context.Context, msg history.Message) (history.Message, error) {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
c.appends = append(c.appends, msg)
|
||
return msg, nil
|
||
}
|
||
|
||
func (c *captureHistory) List(context.Context, string, int64, int, string) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
func (c *captureHistory) ListAfter(context.Context, string, int64, int, string) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
func (c *captureHistory) Recent(context.Context, string, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
func (c *captureHistory) ListPlatformLog(context.Context, string, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
func (c *captureHistory) ListAdapterLog(context.Context, string, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
func (c *captureHistory) ListAllLog(context.Context, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
func (c *captureHistory) DeleteSession(context.Context, string) error { return nil }
|
||
func (c *captureHistory) Search(context.Context, string, string, int64, int) ([]history.Message, error) {
|
||
return nil, nil
|
||
}
|
||
func (c *captureHistory) UpdateMessage(_ context.Context, id, content string) (history.Message, error) {
|
||
return history.Message{ID: id, Content: content}, nil
|
||
}
|
||
func (c *captureHistory) DeleteMessage(context.Context, string) error { return nil }
|
||
func (c *captureHistory) RenameSession(context.Context, string, string) error {
|
||
return nil
|
||
}
|
||
func (c *captureHistory) SetPinned(context.Context, string, bool) error { return nil }
|
||
func (c *captureHistory) NewSubSession(context.Context, string) (string, error) {
|
||
return "", nil
|
||
}
|
||
func (c *captureHistory) ListSessions(context.Context) ([]history.SessionMeta, error) {
|
||
return nil, nil
|
||
}
|
||
|
||
func (c *captureHistory) records() []history.Message {
|
||
c.mu.Lock()
|
||
defer c.mu.Unlock()
|
||
return append([]history.Message(nil), c.appends...)
|
||
}
|
||
|
||
func TestOwnerPrivatePersistsToMainSessionWithPlatform(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "微信回复"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
|
||
hist := &captureHistory{}
|
||
var persisted []history.Message
|
||
b, err := New(Options{
|
||
Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "100"),
|
||
History: hist,
|
||
OnPersisted: func(m history.Message) { persisted = append(persisted, m) },
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
// 注册假装接收该消息的账号适配器,使持久化的
|
||
// 消息流带上适配器注册名(例如 "telegram")。
|
||
b.Add(&fakeAdapter{name: "telegram"})
|
||
|
||
msg := directMessage() // Platform 字段 = "telegram"
|
||
b.process(context.Background(), msg)
|
||
|
||
recs := hist.records()
|
||
if len(recs) != 2 {
|
||
t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs))
|
||
}
|
||
// 两者都落在 Web 主会话并带渠道平台 + 适配器,
|
||
// 且 IM 消息总是以 chat 模式持久化(design/16 §5.1)。
|
||
if recs[0].SessionID != history.MainSessionID || recs[0].Role != history.RoleUser ||
|
||
recs[0].Content != "hello" || recs[0].Platform != "telegram" || recs[0].AdapterID != "telegram" ||
|
||
recs[0].Mode != "chat" {
|
||
t.Fatalf("user record = %+v, want web/telegram/hello/telegram/chat", recs[0])
|
||
}
|
||
// 助手正文的落库内容是**规范形态**(写入侧段落包裹规整,见
|
||
// turnreply_paragraph_test.go):这里只关心归属字段,因此只要求包裹正确。
|
||
if recs[1].SessionID != history.MainSessionID || recs[1].Role != history.RoleAssistant ||
|
||
recs[1].Content != "<p>微信回复</p>" || recs[1].Platform != "telegram" || recs[1].AdapterID != "telegram" ||
|
||
recs[1].Mode != "chat" {
|
||
t.Fatalf("assistant record = %+v, want web/telegram/<p>微信回复</p>/telegram/chat", recs[1])
|
||
}
|
||
// 每条已持久化的消息都交给了广播钩子。
|
||
if len(persisted) != 2 {
|
||
t.Fatalf("persisted notifications = %d, want 2", len(persisted))
|
||
}
|
||
// 且循环针对主会话运行,而不是隔离的 sess-1。
|
||
if chat.sessionID != history.MainSessionID {
|
||
t.Fatalf("loop session id = %q, want %q", chat.sessionID, history.MainSessionID)
|
||
}
|
||
}
|
||
|
||
func TestOwnerPrivatePersistsToMainWithDeviceID(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "微信回复"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
|
||
hist := &captureHistory{}
|
||
b, err := New(Options{
|
||
Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "100"),
|
||
History: hist,
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
|
||
// directMessage() → Platform telegram / ChannelID 100 / SenderID 100,因此
|
||
// deviceID 构建为 <platform>:<channelID>:<senderID>(design/16 §4.5)。
|
||
b.process(context.Background(), directMessage())
|
||
|
||
recs := hist.records()
|
||
if len(recs) != 2 {
|
||
t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs))
|
||
}
|
||
// 用户轮次与助手回复都带上推导出的 deviceID 与
|
||
// 渠道平台,使 Web 客户端能标注来源端。
|
||
for i, rec := range recs {
|
||
if rec.DeviceID != "telegram:100:100" {
|
||
t.Fatalf("record[%d] DeviceID = %q, want telegram:100:100", i, rec.DeviceID)
|
||
}
|
||
if rec.Platform != "telegram" {
|
||
t.Fatalf("record[%d] Platform = %q, want telegram", i, rec.Platform)
|
||
}
|
||
}
|
||
if recs[0].Role != history.RoleUser || recs[0].Content != "hello" {
|
||
t.Fatalf("user record = %+v", recs[0])
|
||
}
|
||
if recs[1].Role != history.RoleAssistant || recs[1].Content != "<p>微信回复</p>" {
|
||
t.Fatalf("assistant record = %+v", recs[1])
|
||
}
|
||
}
|
||
|
||
func TestGroupMessageNotPersistedToMainSession(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "群聊回复"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: true, MemWritable: false}}
|
||
hist := &captureHistory{}
|
||
var persisted []history.Message
|
||
b, err := New(Options{
|
||
Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist,
|
||
OnPersisted: func(m history.Message) { persisted = append(persisted, m) },
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
|
||
// 被点名(提及)的群消息 → 回复(design/12 §6 策略二),仍
|
||
// 隔离在群会话内。
|
||
msg := adapter.UnifiedMessage{
|
||
Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup,
|
||
GroupID: "-20", GroupName: "Gang", SenderID: "9",
|
||
Sender: adapter.Actor{Role: adapter.RoleMember}, Mentions: []adapter.Mention{{UserID: "9"}},
|
||
Content: "hello all", ContentType: adapter.ContentText,
|
||
}
|
||
b.process(context.Background(), msg)
|
||
|
||
// design/16 §4.6:群聊/陌生人会话保持隔离——它不会
|
||
// 持久化进 Web 主会话,循环也针对隔离的
|
||
// 会话 id 运行。消息只落在隔离会话自有的平台日志
|
||
// (适配器日志页),因此没有任何内容到达 Web 主会话。
|
||
recs := hist.records()
|
||
if len(recs) != 2 {
|
||
t.Fatalf("group appends = %d, want 2 (isolated user + assistant)", len(recs))
|
||
}
|
||
for i, rec := range recs {
|
||
if rec.SessionID != "sess-g" {
|
||
t.Fatalf("record[%d] SessionID = %q, want isolated sess-g (not web)", i, rec.SessionID)
|
||
}
|
||
if rec.Platform != "telegram" {
|
||
t.Fatalf("record[%d] Platform = %q, want telegram", i, rec.Platform)
|
||
}
|
||
// 隔离的 IM 消息仍为 chat 模式(design/16 §5.1)。
|
||
if rec.Mode != "chat" {
|
||
t.Fatalf("record[%d] Mode = %q, want chat", i, rec.Mode)
|
||
}
|
||
}
|
||
if chat.sessionID != "sess-g" {
|
||
t.Fatalf("group loop session id = %q, want sess-g", chat.sessionID)
|
||
}
|
||
// 每次隔离写入也会交给广播钩子,使 gateway 能为 IM
|
||
// 适配器日志页发出 `channel_persisted`——但消息本身
|
||
// 严格留在隔离会话中,绝不泄露到主/`web`
|
||
// 会话(design/16 §4.6 §3)。
|
||
if len(persisted) != 2 {
|
||
t.Fatalf("isolated persisted notifications = %d, want 2", len(persisted))
|
||
}
|
||
for i, p := range persisted {
|
||
if p.SessionID != "sess-g" {
|
||
t.Fatalf("persisted[%d] SessionID = %q, want isolated sess-g", i, p.SessionID)
|
||
}
|
||
if p.SessionID == history.MainSessionID {
|
||
t.Fatalf("persisted[%d] leaked to main web session", i)
|
||
}
|
||
}
|
||
}
|
||
|
||
func TestBlacklistGroupMessageDroppedBeforeLoop(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "不该回复"}}
|
||
// 一个陌生人的群会话(若不拦截本应被处理)。
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}}
|
||
ad := &fakeAdapter{name: "telegram", cfg: map[string]any{
|
||
"bot_token": "configured",
|
||
"denyGroups": []string{"-100"},
|
||
}}
|
||
var rejected []string
|
||
b, err := New(Options{
|
||
Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: &captureHistory{},
|
||
OnRejected: func(_ adapter.UnifiedMessage, reason string) { rejected = append(rejected, reason) },
|
||
})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
b.Add(ad)
|
||
|
||
msg := adapter.UnifiedMessage{
|
||
Platform: "telegram", ChannelID: "-100", ChannelType: adapter.ChannelGroup,
|
||
GroupID: "-100", SenderID: "9", SenderName: "张三",
|
||
Sender: adapter.Actor{Role: adapter.RoleMember}, Content: "spam",
|
||
ContentType: adapter.ContentText,
|
||
}
|
||
b.process(context.Background(), msg)
|
||
|
||
// 被拉黑的群消息绝不进入 agent 循环,也绝不发送
|
||
// 回复;拒绝原因暴露给审计轨迹。
|
||
if chat.sessionID != "" || len(chat.messages) != 0 {
|
||
t.Fatalf("loop ran for blacklisted message: session=%q messages=%+v", chat.sessionID, chat.messages)
|
||
}
|
||
if len(ad.sends) != 0 {
|
||
t.Fatalf("blacklisted message sent %d replies, want 0", len(ad.sends))
|
||
}
|
||
if len(rejected) != 1 || rejected[0] != "group_denied" {
|
||
t.Fatalf("rejection reasons = %v, want [group_denied]", rejected)
|
||
}
|
||
}
|
||
|
||
func TestBlacklistUserMessageDroppedBeforeLoop(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "不该回复"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-s", IsGroup: false, Owner: false, MemWritable: false}}
|
||
ad := &fakeAdapter{name: "telegram", cfg: map[string]any{
|
||
"bot_token": "configured",
|
||
"allowUsers": []string{"111"},
|
||
}}
|
||
b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: &captureHistory{}})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
b.Add(ad)
|
||
|
||
msg := adapter.UnifiedMessage{
|
||
Platform: "telegram", ChannelID: "200", ChannelType: adapter.ChannelDirect,
|
||
SenderID: "200", Sender: adapter.Actor{Role: adapter.RoleMember}, Content: "hi",
|
||
ContentType: adapter.ContentText,
|
||
}
|
||
b.process(context.Background(), msg)
|
||
|
||
// 来自不在允许白名单中的用户的私聊消息被丢弃;主人
|
||
// 绝不受影响(这里是陌生人)。
|
||
if chat.sessionID != "" || len(chat.messages) != 0 {
|
||
t.Fatalf("loop ran for non-whitelisted user: session=%q messages=%+v", chat.sessionID, chat.messages)
|
||
}
|
||
if len(ad.sends) != 0 {
|
||
t.Fatalf("non-whitelisted user sent %d replies, want 0", len(ad.sends))
|
||
}
|
||
}
|
||
|
||
func TestGroupMessageThirdPartySenderNameAndBoundary(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "收到"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}}
|
||
hist := &captureHistory{}
|
||
b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
|
||
// 被点名(提及)的群消息 → 回复(design/12 §6 策略二),因此
|
||
// 会注入第三方边界与 [第三方 · 张三] 前缀。
|
||
msg := adapter.UnifiedMessage{
|
||
Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup,
|
||
GroupID: "-20", GroupName: "Gang", SenderID: "9", SenderName: "张三",
|
||
Sender: adapter.Actor{Role: adapter.RoleMember}, Mentions: []adapter.Mention{{UserID: "9"}},
|
||
Content: "hello all", ContentType: adapter.ContentText,
|
||
}
|
||
b.process(context.Background(), msg)
|
||
|
||
// 循环看到最前面有一条第三方边界系统消息,加上已带
|
||
// [第三方 · 张三] 前缀的用户消息(历史中保留原始
|
||
// 内容,因此不丢失来源数据)。
|
||
if len(chat.messages) != 2 {
|
||
t.Fatalf("loop messages = %d, want 2 (boundary + user)", len(chat.messages))
|
||
}
|
||
if chat.messages[0].Role != model.RoleSystem || !strings.Contains(chat.messages[0].Content[0].Text, "非管理员账户") {
|
||
t.Fatalf("expected third-party boundary system message, got %+v", chat.messages[0])
|
||
}
|
||
user := chat.messages[1]
|
||
// 第三方说话者由结构化的 SenderName 字段承载,用户文本保持
|
||
// 不加装饰(design/16 §4.6.1);来源/边界放在系统消息里。
|
||
if user.SenderName != "张三" || user.Content[0].Text != "hello all" {
|
||
t.Fatalf("loop user message = %+v, want SenderName 张三 with undecorated text", user)
|
||
}
|
||
// 历史保留原始内容并携带发送者名,使适配器日志
|
||
// 页能标注消息流而不改动来源消息。
|
||
recs := hist.records()
|
||
if len(recs) != 2 || recs[0].Content != "hello all" || recs[0].SenderName != "张三" {
|
||
t.Fatalf("isolated records = %+v, want raw content with SenderName 张三", recs)
|
||
}
|
||
}
|
||
|
||
func TestOwnerInGroupIsIsolatedButNotLabeledThirdParty(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
chat := &stubChatter{res: &loop.Result{Text: "收到"}}
|
||
// 主人在群里发言:会话仍是受限的(记忆只读),
|
||
// 但主人绝不会被标为第三方(design/16 §4.6 §1-2)。
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: true, MemWritable: false}}
|
||
hist := &captureHistory{}
|
||
b, err := New(Options{Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "owner_uid"), History: hist})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
|
||
msg := adapter.UnifiedMessage{
|
||
Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup,
|
||
GroupID: "-20", SenderID: "owner_uid", SenderName: "主人",
|
||
Sender: adapter.Actor{Role: adapter.RoleOwner}, Mentions: []adapter.Mention{{UserID: "owner_uid"}},
|
||
Content: "我在群里", ContentType: adapter.ContentText,
|
||
}
|
||
// 这条主人群消息**指向她本人**(含提及):owner 车道默认「不对她说就不开口」
|
||
// (design/18 §11),不带指向信号的主人群消息只进上下文、不驱动回合;
|
||
// 带 @ 信号才会进循环,本用例因此仍在断言「进循环时主人不被标为第三方」。
|
||
b.process(context.Background(), msg)
|
||
|
||
// 受限上下文 → 仍会注入第三方边界系统消息。
|
||
if len(chat.messages) != 2 || chat.messages[0].Role != model.RoleSystem {
|
||
t.Fatalf("loop messages = %d, want 2 (boundary + user)", len(chat.messages))
|
||
}
|
||
// 主人的消息不会被标为第三方,也不携带 SenderName;用户
|
||
// 文本保持不加装饰,群的来源放在系统消息里
|
||
// (design/16 §4.6.1)。
|
||
user := chat.messages[1]
|
||
if user.SenderName != "" || user.Content[0].Text != "我在群里" {
|
||
t.Fatalf("owner group message = %+v, want undecorated text, no SenderName, no [第三方] prefix", user)
|
||
}
|
||
if !strings.Contains(chat.messages[0].Content[0].Text, "群#-20") {
|
||
t.Fatalf("origin/boundary system = %q, want group#-20", chat.messages[0].Content[0].Text)
|
||
}
|
||
// 主人在群里的发言仍落在隔离会话(不是 Web 主会话)。
|
||
recs := hist.records()
|
||
if len(recs) != 2 || recs[0].SessionID != "sess-g" {
|
||
t.Fatalf("owner group records = %+v, want isolated sess-g, not web", recs)
|
||
}
|
||
}
|
||
|
||
func TestNoHistoryStoreKeepsBridgeUnchanged(t *testing.T) {
|
||
t.Parallel()
|
||
// 未装配历史存储的桥接器不持久化任何内容,但仍运行/回复。
|
||
b, chat, _, ad := newTestBridge(t)
|
||
b.process(context.Background(), directMessage())
|
||
if len(ad.sends) != 1 {
|
||
t.Fatalf("send count = %d, want 1", len(ad.sends))
|
||
}
|
||
if chat.sessionID != history.MainSessionID {
|
||
t.Fatalf("loop session id = %q, want %q", chat.sessionID, history.MainSessionID)
|
||
}
|
||
}
|
||
|
||
// newWebhookBridge 构建一个裸桥接器(无适配器),接上其他测试
|
||
// 使用的桩,便于 WebhookHandlers 测试自行添加适配器。
|
||
func newWebhookBridge(t *testing.T) *Bridge {
|
||
t.Helper()
|
||
chat := &stubChatter{res: &loop.Result{Text: "hi"}}
|
||
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
|
||
b, err := New(Options{Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}})
|
||
if err != nil {
|
||
t.Fatalf("New: %v", err)
|
||
}
|
||
return b
|
||
}
|
||
|
||
func TestWebhookHandlersCollectsKookWebhook(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b := newWebhookBridge(t)
|
||
b.Add(kook.New(kook.Config{BotToken: "t", ReceiveMode: "webhook", WebhookPath: "/webhook/kook"}))
|
||
|
||
handlers := b.WebhookHandlers()
|
||
h, ok := handlers["/webhook/kook"]
|
||
if !ok || h == nil {
|
||
t.Fatalf("kook webhook handler not mounted; got %v", handlers)
|
||
}
|
||
}
|
||
|
||
func TestWebhookHandlersSkipsWSModeKook(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b := newWebhookBridge(t)
|
||
b.Add(kook.New(kook.Config{BotToken: "t", ReceiveMode: "ws"}))
|
||
|
||
if got := b.WebhookHandlers(); got != nil {
|
||
t.Fatalf("ws-mode kook exposed a webhook handler, got %v", got)
|
||
}
|
||
}
|
||
|
||
func TestWebhookHandlersMountsWebhookChannel(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
b := newWebhookBridge(t)
|
||
b.Add(webhook.New(webhook.WithConfig(webhook.Config{EndpointPath: "/cb"})))
|
||
|
||
handlers := b.WebhookHandlers()
|
||
h, ok := handlers["/cb"]
|
||
if !ok || h == nil {
|
||
t.Fatalf("webhook channel handler not mounted; got %v", handlers)
|
||
}
|
||
}
|