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 全绿。
263 lines
10 KiB
Go
263 lines
10 KiB
Go
package channelbridge
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
|
|
"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/turnreply"
|
|
)
|
|
|
|
// 本文件锁定切片 KB1 的渠道侧一半:**IM 回合也绝不能对用户静默**。
|
|
//
|
|
// 缺口形状(与网关侧实测同一根因):渠道桥自己驱动 loop 并自行落库,过去的
|
|
// deliverReply 见到空文本直接 return,于是模型给空回复(只有思维链、没有正文)
|
|
// 时,IM 侧就是「她没声了」:没有出站消息、历史里也只有那条用户消息。轮次用尽、
|
|
// 安全网终止、回合内部报错三条出口同样静默。
|
|
//
|
|
// 判定口径与网关逐字一致(同一份文案来自共享包 internal/runtime/turnreply),
|
|
// 边界差异只有取消:IM 没有网关的会话代次(interrupt / 续跑队列是 Web 会话的
|
|
// 概念),这里的取消判据是**处理上下文被取消**(桥接器 Stop 会取消 b.ctx)。
|
|
// 因此本文件用「被取消的回合不交付」覆盖网关侧「被 interrupt 取消 / 陈旧代次
|
|
// 不落库」那两条边界在 IM 上的等价物。
|
|
//
|
|
// 断言走真路径:真入站流水线(process → 会话解析 → actor → 落库 → 出站)+ 桩
|
|
// Chatter,只有模型那一侧是脚本化的。
|
|
|
|
// replyTurnBridge 装配一个主人私聊 IM 回合的桥接器:桩 Chatter + 记录型历史 +
|
|
// 已注册的 telegram 适配器。主人私聊 ⇒ 回复落 Web 主会话,因此可以直接读
|
|
// records 断言落库内容。
|
|
func replyTurnBridge(t *testing.T, chat *stubChatter) (*Bridge, *fakeAdapter, *captureHistory) {
|
|
t.Helper()
|
|
sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
|
|
hist := &captureHistory{}
|
|
ad := &fakeAdapter{name: "telegram"}
|
|
b, err := New(Options{
|
|
Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "100"), History: hist,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("New: %v", err)
|
|
}
|
|
b.Add(ad)
|
|
return b, ad, hist
|
|
}
|
|
|
|
// assertAssistantReply 断言历史里恰好一条 user + 恰好一条 assistant,且 assistant
|
|
// 内容逐字等于 want(用户消息必须原样留着 —— 回合跑过就该留下痕迹)。
|
|
func assertAssistantReply(t *testing.T, hist *captureHistory, want string) history.Message {
|
|
t.Helper()
|
|
recs := hist.records()
|
|
if len(recs) != 2 {
|
|
t.Fatalf("落库记录 = %d 条 (%+v), want 2 (user + assistant)", len(recs), recs)
|
|
}
|
|
if recs[0].Role != history.RoleUser || recs[0].Content != "hello" {
|
|
t.Fatalf("user 记录 = %+v, want 原样的 inbound 文本", recs[0])
|
|
}
|
|
if recs[1].Role != history.RoleAssistant {
|
|
t.Fatalf("第二条记录 role = %q, want assistant", recs[1].Role)
|
|
}
|
|
if strings.TrimSpace(recs[1].Content) == "" {
|
|
t.Fatal("落库的 assistant 内容为空 —— IM 侧就是「她没声了」")
|
|
}
|
|
if recs[1].Content != want {
|
|
t.Fatalf("assistant 内容 = %q, want %q", recs[1].Content, want)
|
|
}
|
|
return recs[1]
|
|
}
|
|
|
|
// TestTurnReply_EmptyReplyDeliversHonestChineseText 是必测①:IM 回合拿到空回复
|
|
// (空串与纯空白两种写法)时,必须落一条**非空的中文** assistant 消息,并经同一
|
|
// 适配器发出去 —— 而不是过去那条空消息/静默。
|
|
func TestTurnReply_EmptyReplyDeliversHonestChineseText(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cases := []struct {
|
|
name string
|
|
text string
|
|
}{
|
|
{"完全空串", ""},
|
|
{"纯空白", " "},
|
|
}
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
chat := &stubChatter{res: &loop.Result{Text: c.text}}
|
|
b, ad, hist := replyTurnBridge(t, chat)
|
|
|
|
b.process(context.Background(), directMessage())
|
|
|
|
if len(ad.sends) != 1 {
|
|
t.Fatalf("出站消息 = %d 条, want 1(空回复也必须有可见回复)", len(ad.sends))
|
|
}
|
|
got := ad.sends[0].Content
|
|
if got != turnreply.FallbackEmptyReply {
|
|
t.Fatalf("出站文案 = %q, want 共享包的诚实说明 %q", got, turnreply.FallbackEmptyReply)
|
|
}
|
|
if !strings.Contains(got, "抱歉") {
|
|
t.Fatalf("文案不是中文诚实说明: %q", got)
|
|
}
|
|
assertAssistantReply(t, hist, turnreply.FallbackEmptyReply)
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestTurnReply_TerminalStopsUseHonestChineseText 覆盖另外两条静默出口:轮次用尽
|
|
// 与关键工具循环安全网终止。安全网那一条尤其重要 —— loop 塞进 Result.Text 的是
|
|
// **内部英文诊断**,绝不能原样发给 IM 用户(与网关侧同一条断言)。
|
|
func TestTurnReply_TerminalStopsUseHonestChineseText(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
cases := []struct {
|
|
name string
|
|
res *loop.Result
|
|
want string
|
|
}{
|
|
{"轮次用尽", &loop.Result{Text: "", StopReason: loop.StopReasonMaxRounds}, turnreply.FallbackMaxRounds},
|
|
{"安全网终止",
|
|
&loop.Result{Text: "critical tool-loop repetition detected", StopReason: loop.StopReasonToolLoopTerminated},
|
|
turnreply.FallbackToolLoop},
|
|
}
|
|
for _, c := range cases {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
chat := &stubChatter{res: c.res}
|
|
b, ad, hist := replyTurnBridge(t, chat)
|
|
|
|
b.process(context.Background(), directMessage())
|
|
|
|
if len(ad.sends) != 1 {
|
|
t.Fatalf("出站消息 = %d 条, want 1", len(ad.sends))
|
|
}
|
|
if got := ad.sends[0].Content; got != c.want {
|
|
t.Fatalf("出站文案 = %q, want %q", got, c.want)
|
|
}
|
|
if strings.Contains(ad.sends[0].Content, "critical tool-loop") {
|
|
t.Fatalf("内部英文诊断文案漏给了 IM 用户: %q", ad.sends[0].Content)
|
|
}
|
|
assertAssistantReply(t, hist, c.want)
|
|
})
|
|
}
|
|
}
|
|
|
|
// TestTurnReply_RunErrorDeliversHonestChineseText 覆盖「失败回合」:loop 返回错误
|
|
// (模型 / 工具 / 编排报错)时,IM 用户仍要看到一条说明,而不是一片安静。
|
|
func TestTurnReply_RunErrorDeliversHonestChineseText(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
chat := &stubChatter{res: nil, err: errors.New("model: upstream 502")}
|
|
b, ad, hist := replyTurnBridge(t, chat)
|
|
|
|
b.process(context.Background(), directMessage())
|
|
|
|
if len(ad.sends) != 1 {
|
|
t.Fatalf("出站消息 = %d 条, want 1(回合报错也要留下说明)", len(ad.sends))
|
|
}
|
|
if got := ad.sends[0].Content; got != turnreply.FallbackError {
|
|
t.Fatalf("出站文案 = %q, want %q", got, turnreply.FallbackError)
|
|
}
|
|
assertAssistantReply(t, hist, turnreply.FallbackError)
|
|
}
|
|
|
|
// TestTurnReply_CanceledTurnDeliversNothing 是必测②(IM 侧的边界):被取消的回合
|
|
// **刻意什么都不交付** —— 用户意图就是「停下」,此刻补一句「我出问题了」反而违背
|
|
// 那条指令。两个取消信号都要认:回合以 context.Canceled 结束,以及处理上下文本身
|
|
// 已被取消(桥接器 Stop 取消 b.ctx 的形状)。
|
|
func TestTurnReply_CanceledTurnDeliversNothing(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
t.Run("回合以取消结束", func(t *testing.T) {
|
|
t.Parallel()
|
|
chat := &stubChatter{res: nil, err: context.Canceled}
|
|
b, ad, hist := replyTurnBridge(t, chat)
|
|
|
|
b.process(context.Background(), directMessage())
|
|
|
|
assertNothingDelivered(t, ad, hist)
|
|
})
|
|
|
|
t.Run("处理上下文已取消", func(t *testing.T) {
|
|
t.Parallel()
|
|
// 空回复 + 已取消的上下文:正是桥接器 Stop 打断在途回合的形状。
|
|
chat := &stubChatter{res: &loop.Result{Text: ""}}
|
|
b, ad, hist := replyTurnBridge(t, chat)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
cancel()
|
|
|
|
b.process(ctx, directMessage())
|
|
|
|
assertNothingDelivered(t, ad, hist)
|
|
})
|
|
}
|
|
|
|
// assertNothingDelivered 断言这个回合既没有出站消息,也没有落库 assistant 消息
|
|
// (只留下那条用户消息 —— 它是回合开始前就写好的)。
|
|
func assertNothingDelivered(t *testing.T, ad *fakeAdapter, hist *captureHistory) {
|
|
t.Helper()
|
|
if len(ad.sends) != 0 {
|
|
t.Fatalf("被取消的回合发了 %d 条消息, want 0: %+v", len(ad.sends), ad.sends)
|
|
}
|
|
recs := hist.records()
|
|
if len(recs) != 1 || recs[0].Role != history.RoleUser {
|
|
t.Fatalf("落库记录 = %+v, want 只有那条 user 消息", recs)
|
|
}
|
|
}
|
|
|
|
// TestTurnReply_NormalReplyPassedThroughVerbatim 是必测③:正常回合(有正文)
|
|
// 的出站内容与模型正文**逐字相同**(段落标记由 IM 出口按既有行为剥除),落库则是
|
|
// 同一条正文的**规范形态**(每段一个 <p>,写入侧规整,见
|
|
// turnreply_paragraph_test.go)——两者都不掺进兜底文案。
|
|
func TestTurnReply_NormalReplyPassedThroughVerbatim(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
const reply = "你好,我在。今天想聊点什么?"
|
|
// 规范形态:单段正文也会被包上 <p>,因此落库内容与模型原文相差的只有包裹。
|
|
const canonical = "<p>" + reply + "</p>"
|
|
chat := &stubChatter{res: &loop.Result{Text: reply, Mood: "平静"}}
|
|
b, ad, hist := replyTurnBridge(t, chat)
|
|
|
|
b.process(context.Background(), directMessage())
|
|
|
|
if len(ad.sends) != 1 {
|
|
t.Fatalf("出站消息 = %d 条, want 1", len(ad.sends))
|
|
}
|
|
if got := ad.sends[0].Content; got != reply {
|
|
t.Fatalf("出站文案 = %q, want 逐字等于模型正文(段落标记不投给 IM)", got)
|
|
}
|
|
rec := assertAssistantReply(t, hist, canonical)
|
|
if rec.Mood != "平静" {
|
|
t.Fatalf("assistant mood = %q, want 平静(真实结果的心情照旧带到消息上)", rec.Mood)
|
|
}
|
|
for _, fallback := range []string{turnreply.FallbackEmptyReply, turnreply.FallbackError} {
|
|
if strings.Contains(rec.Content, fallback) {
|
|
t.Fatalf("正常回复里混进了兜底文案: %q", rec.Content)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestTurnReply_EmptyInboundContentStillDropped 守住边界:无文本(且无附件)的
|
|
// 入站消息仍按既有规则丢弃 —— 本改动只补**回合的回复**,绝不放垃圾消息进库。
|
|
func TestTurnReply_EmptyInboundContentStillDropped(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
chat := &stubChatter{res: &loop.Result{Text: "不该出现"}}
|
|
b, ad, hist := replyTurnBridge(t, chat)
|
|
msg := directMessage()
|
|
msg.Content = " "
|
|
|
|
b.process(context.Background(), msg)
|
|
|
|
if chat.sessionID != "" {
|
|
t.Fatalf("空内容消息驱动了循环, session = %q", chat.sessionID)
|
|
}
|
|
if len(ad.sends) != 0 {
|
|
t.Fatalf("空内容消息发了 %d 条消息, want 0", len(ad.sends))
|
|
}
|
|
if recs := hist.records(); len(recs) != 0 {
|
|
t.Fatalf("空内容消息落库 = %+v, want 0 条", recs)
|
|
}
|
|
}
|