diff --git a/internal/channel/session/keys.go b/internal/channel/session/keys.go new file mode 100644 index 0000000..1999e37 --- /dev/null +++ b/internal/channel/session/keys.go @@ -0,0 +1,54 @@ +package session + +// 本文件把「消息 → 会话」的键派生**导出**给别的层复用(M22 复审 P1-1)。 +// +// 为什么必须导出:会话的键一直是会话层自己算的(`channelID` + `platform \x00 channelID`), +// 而渠道桥接器的**每会话回合车道**需要「什么算同一个会话」这个判据与它**同源**。 +// 各算各的会漂移,而且漂移的后果不是排序不好看,是**同一个 session 两条车道、两个回合 +// 并发**——两条 run 共享 sessionID,checkpoint 行 id `{sessionID}:{step}` 互相覆盖、 +// `loadCheckpoint` 可能续到另一条在途 run 的快照而重放工具调用,flow 桶 +// `p.graphs[sessionID]` 被先收尾者删掉(loop 没有会话级互斥,IM 侧也没有网关那层 +// `session_turn_fence` 兜底)。 +// +// 两处真实反例(都来自复审实测): +// - kook:同一 guild 的两个 text channel —— `ChannelID` 不同(c1/c2)但 `GroupID` 相同, +// 会话层按 `group:` 判成**同一个** session;任何把 `ChannelID` 当会话判据的 +// 派生都会把它们拆成两条车道。 +// - 多账号:同一平台两个账号对同一对端 —— 账号名不同(`telegram:conf0`/`:conf1`)、 +// `platform` 常量相同、`ChannelID` 相同 ⇒ 也是**同一个** session;任何把「账号」当 +// 会话判据的派生都会把它们拆开。 +// +// 因此本文件只做一件事:把既有的派生**原样**导出,语义一个字不改(`channelID` 仍是唯一 +// 的实现,见 manager.go)。 + +import "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + +// sessionKeySep 分隔 (platform, channel_id) 两个维度。用 NUL 是因为平台名与渠道 id 都 +// 不可能包含它,拼接因此不会产生歧义(`a` + "b" 与 `ab` + "" 不会撞键)。 +const sessionKeySep = "\x00" + +// ChannelKey 返回一条入站消息所代表的**渠道**键(会话派生的第二个维度): +// 群消息按 `group:`(加前缀使群不会与同 id 的私聊渠道撞键),私聊按 +// `ChannelID`,没有显式渠道 id 时退回 `SenderID`。 +// +// 它就是 `SessionManager.GetOrCreate` 内部用的那一个函数(manager.go 的 `channelID`), +// 导出只为让别的层(渠道桥接器的回合车道)能与会话键同源,**不得**在别处另写一份。 +// 返回空串表示这条消息没有可用的渠道(`GetOrCreate` 会以 `ErrInvalidChannel` 拒掉它)。 +func ChannelKey(msg adapter.UnifiedMessage) string { return channelID(msg) } + +// Key 返回一条入站消息的**会话键** —— 与 `SessionManager` 的缓存键和 +// `channel_sessions` 表的唯一键**逐字相同**(`platform \x00 channelID`)。 +// +// 它是本包对外唯一的「什么算同一个会话」判据:任何需要按会话串行/分桶的调用方都用它, +// 不要自己拼 `platform|channel` 之类的近似键(那正是 M22 复审里「车道键与会话键不同源」 +// 的成因)。返回的键在渠道不可用时仍然非空(`platform\x00`),调用方若需要区分, +// 用 ChannelKey 判空。 +func Key(msg adapter.UnifiedMessage) string { + return sessionKey(msg.Platform, channelID(msg)) +} + +// sessionKey 拼接 (platform, channel_id) 两维度。本包内所有「会话键」的**唯一**出处 +// (GetOrCreate / List / 导出的 Key 都经这里),因此不可能各自漂移。 +func sessionKey(platform, cid string) string { + return platform + sessionKeySep + cid +} diff --git a/internal/channel/session/keys_test.go b/internal/channel/session/keys_test.go new file mode 100644 index 0000000..6ae7356 --- /dev/null +++ b/internal/channel/session/keys_test.go @@ -0,0 +1,89 @@ +package session + +// 本文件钉住 keys.go 导出的两个派生:`ChannelKey` 与会话层内部 `channelID` 是**同一个** +// 实现,`Key` 与会话缓存键**逐字相同**。它们是「什么算同一个会话」的唯一定义 —— 车道 +// (internal/services/channelbridge/lane.go)按会话串行就靠它,两处一旦分叉,同一个 +// session 就会出现两条车道、两个回合并发(M22 复审 P1-1)。 + +import ( + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +func TestChannelKeyDerivations(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + msg adapter.UnifiedMessage + want string + }{ + { + name: "群:按 group:,与频道无关", + msg: adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g1"}, + want: "group:g1", + }, + { + name: "群:缺 GroupID 时退回 ChannelID", + msg: adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1"}, + want: "c1", + }, + { + name: "私聊:按 ChannelID", + msg: adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect, ChannelID: "200", SenderID: "200"}, + want: "200", + }, + { + name: "私聊:缺 ChannelID 时退回 SenderID", + msg: adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect, SenderID: "200"}, + want: "200", + }, + { + name: "私聊:两者都缺 ⇒ 空(调用方据此以 ErrInvalidChannel 拒掉)", + msg: adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect}, + want: "", + }, + } + for _, c := range cases { + c := c + t.Run(c.name, func(t *testing.T) { + t.Parallel() + if got := ChannelKey(c.msg); got != c.want { + t.Fatalf("ChannelKey = %q, want %q", got, c.want) + } + }) + } +} + +// TestKeyIsTheSessionManagerCacheKey 钉住「导出的 Key 就是会话缓存键」: +// 形状 `platform \x00 channelID`(NUL 分隔,杜绝拼接歧义),且两维度都参与 —— +// 同平台不同渠道、不同平台同渠道都必须是**不同**的键。 +func TestKeyIsTheSessionManagerCacheKey(t *testing.T) { + t.Parallel() + + group1 := adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g1"} + group1OtherChannel := adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c2", GroupID: "g1"} + group2 := adapter.UnifiedMessage{Platform: "kook", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g2"} + otherPlatform := adapter.UnifiedMessage{Platform: "qq-obv11", ChannelType: adapter.ChannelGroup, ChannelID: "c1", GroupID: "g1"} + + if got, want := Key(group1), "kook\x00group:g1"; got != want { + t.Fatalf("Key = %q, want %q", got, want) + } + // 同一 guild 的两个 text channel 在会话层是**同一个**会话。 + if Key(group1) != Key(group1OtherChannel) { + t.Fatalf("same guild channels must share one session key: %q vs %q", Key(group1), Key(group1OtherChannel)) + } + // 不同群、不同平台必须是不同的会话。 + if Key(group1) == Key(group2) { + t.Fatalf("different groups must not share a session key: %q", Key(group1)) + } + if Key(group1) == Key(otherPlatform) { + t.Fatalf("different platforms must not share a session key: %q", Key(group1)) + } + // 渠道缺失时键仍非空,但 ChannelKey 能为空(调用方据此区分不可用渠道)。 + noChannel := adapter.UnifiedMessage{Platform: "telegram", ChannelType: adapter.ChannelDirect} + if Key(noChannel) == "" || ChannelKey(noChannel) != "" { + t.Fatalf("missing-channel case: Key=%q ChannelKey=%q", Key(noChannel), ChannelKey(noChannel)) + } +} diff --git a/internal/channel/session/manager.go b/internal/channel/session/manager.go index 5668078..c8340aa 100644 --- a/internal/channel/session/manager.go +++ b/internal/channel/session/manager.go @@ -110,7 +110,7 @@ func (m *SessionManager) GetOrCreate(ctx context.Context, msg adapter.UnifiedMes if cid == "" { return Session{}, fmt.Errorf("session: get-or-create %s: %w", msg.Platform, ErrInvalidChannel) } - key := msg.Platform + "\x00" + cid + key := sessionKey(msg.Platform, cid) // 快路径:已绑定且车道与本次解析一致——只读锁即可返回。 if s, ok := m.lookupCached(key); ok && s.ActorRole == actor.Role { @@ -155,7 +155,7 @@ func (m *SessionManager) List(ctx context.Context) ([]Session, error) { if err := rows.Scan(&platform, &cid, &sessionID, &roleStr); err != nil { return nil, fmt.Errorf("session: scan list: %w", err) } - key := platform + "\x00" + cid + key := sessionKey(platform, cid) if cached, ok := m.lookupCached(key); ok { sessions = append(sessions, cached) continue diff --git a/internal/services/channelbridge/channelbridge.go b/internal/services/channelbridge/channelbridge.go index 861b174..9110dbb 100644 --- a/internal/services/channelbridge/channelbridge.go +++ b/internal/services/channelbridge/channelbridge.go @@ -139,6 +139,15 @@ type Options struct { // Now 是安静时段判定所用的墙上时钟源。为 nil 时用 time.Now, // 使依赖时间的行为在测试中可注入。 Now func() time.Time + // LaneDepth 是**每会话**回合车道的排队深度(M22 切片 LANEQUEUE,见 lane.go): + // 群聊/陌生人车道满时**在门前拒收这一条**(WARN + 计数 + 审计,队列内容不动), + // 主人私聊永不拒收(溢到 spill 只排队)。零或负数取 + // defaultLaneDepth(那里写了取值依据)。 + LaneDepth int + // LaneStaleAfter 是**陈旧度闸**的阈值:轮到某条消息时它已等了超过这个时长, + // 就丢弃不补答(WARN + 计数 + 审计,绝不静默)。零或负数取 + // defaultLaneStaleAfter(那里写了取值依据)。主人私聊与非聊天事件不受该闸约束。 + LaneStaleAfter time.Duration // OwnerLaneFor 是 owner 车道结构化旋钮(design/18 §11「乙 = C」)的**读取**接缝: // 按 (platform, groupID) 返回该群生效的参与策略,返回 nil 表示没有覆盖 // (桥用自己的默认档:不对她说就不开口)。为 nil 时全部走默认档 + 桥内内存覆盖。 @@ -325,6 +334,10 @@ type Bridge struct { wg sync.WaitGroup sem chan struct{} + // lanes 是**每会话回合车道**的注册表(M22 切片 LANEQUEUE,见 lane.go): + // 读循环只投递不跑回合;同一会话内串行 FIFO、不同会话并行;超龄与背压丢弃都留痕。 + // 只在 enqueueInbound 里与 mu 同时出现,锁顺序恒为 mu → lanes.mu。 + lanes *laneSet } // New 依据 opts 构建 Bridge。当必需依赖缺失时返回错误, @@ -400,5 +413,7 @@ func New(opts Options) (*Bridge, error) { voiceAudioStore: opts.VoiceAudioStore, cancel: func() {}, sem: make(chan struct{}, maxConcurrent), + // 每会话车道:深度与陈旧度阈值可经 Options 覆盖(零值取 lane.go 里的默认值)。 + lanes: newLaneSet(opts.LaneDepth, opts.LaneStaleAfter, log), }, nil } diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index a9677b8..3c0bdec 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -59,9 +59,10 @@ func (b *Bridge) markReadStopped(a adapter.PlatformAdapter) { b.log.Warn("channel read loop ended, adapter is no longer connected", "channel", name) } -// onMessage 是面向适配器的入站回调。它在有界 goroutine 中运行 -// 流水线,使适配器的读取循环不会因缓慢的 agent 轮次而阻塞。 -// 适配器被显式携带,使流水线知道是哪个账号收到了 +// onMessage 是面向适配器的入站回调。它**只投递、不跑回合**:过完流控闸就把消息投进 +// 该会话自己的回合车道(lane.go)并立即返回,因此适配器的读循环(WS 单线程读 socket / +// iLink 长轮询 / webhook 处理器)永远不会被一轮缓慢的 agent 调用阻塞 —— 心跳与其它帧 +// 照常被读、被解析。适配器被显式携带,使流水线知道是哪个账号收到了 // 消息(多账号回复路由 + 按账号允许/拒绝,design/16 // §4.6 §5)。 func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter) { @@ -94,29 +95,17 @@ func (b *Bridge) onMessage(ctx context.Context, msg adapter.UnifiedMessage, a ad } } - select { - case b.sem <- struct{}{}: - case <-ctx.Done(): - return - } - - // 在 Stop 设置 closed 所用的同一把锁下 Add 工作项,使 - // 并发的 Stop 要么在等待前看到我们的 Add,要么我们 - // 观察到 closed 后退出,不泄露 goroutine。 - b.mu.Lock() - if b.closed { - b.mu.Unlock() - <-b.sem - return - } - b.wg.Add(1) - b.mu.Unlock() - - go func() { - defer b.wg.Done() - defer func() { <-b.sem }() - b.processFrom(ctx, msg, a) - }() + // M22(切片 LANEQUEUE)——**读与回水解耦**:流控闸在读循环里仍然是同步的 + // (一次滑动窗口判断,微秒级;洪泛正是在这里被挡掉,不该让它进车道占内存), + // 闸后立刻把消息投进**该会话自己的回合车道**并返回,读循环因此马上可以继续读 + // socket —— 心跳与其它帧不再被一轮模型调用阻塞(此前这里同步抢 16 个全局回合槽, + // 槽满即停止读 socket,这正是「消息一直排队、后面才回复前面的话」的现场)。 + // + // 入队非阻塞、有界(背压见 lane.go 的 offer):群聊/陌生人车道满时**在门前拒收 + // 这一条**并留痕(WARN + 计数 + 审计),队列里已有的条目一条不动;主人私聊永不 + // 拒收,溢到 spill 只排队。回合本身在 laneWorker 里跑,全局并发上限 + // (maxConcurrent)也搬到那里取。 + b.enqueueInbound(ctx, msg, a, actor) } // process 用平台与消息匹配的适配器运行入站流水线 diff --git a/internal/services/channelbridge/lane.go b/internal/services/channelbridge/lane.go new file mode 100644 index 0000000..77adc15 --- /dev/null +++ b/internal/services/channelbridge/lane.go @@ -0,0 +1,478 @@ +package channelbridge + +// 本文件承载入站的**回合车道**(M22 切片 LANEQUEUE)。 +// +// 报障现场(主人 2026-10-24):「群聊消息速率限制似乎会让消息一直排队,导致后面才回复 +// 前面的话。」病灶不在模型侧,而在**入站入口把回合跑在了适配器的读循环里**: +// `onMessage` 曾在读循环里**同步**抢全局回合槽(`b.sem`,16 个),槽满的那一刻读循环 +// 就停止读 socket —— 心跳不被回应、后续帧不被解析,等一轮跑完再按条重放,于是先回旧话、 +// 而且回得晚。本文件把「投递」与「跑回合」拆开,并给每个会话一条自己的车道。 +// +// - **读与回水解耦**:适配器的读循环只投递、不跑回合(`enqueueInbound` 立刻返回), +// 心跳与其它帧因此永远不会被一轮模型调用堵住。全局回合槽(`b.sem`)搬到车道 worker +// 里取,仍然限制「最多 16 个回合同时跑」,但等待发生在线程池里而不是读循环里。 +// - **每会话一条回合车道**:同一会话(群 / 私聊各自键,见 `laneKey`)的回合**串行**且 +// **先来先跑**(车道是一条有界 FIFO + 一个 worker),不同会话可并行。 +// - **陈旧度闸**:轮到某条时它已超龄 ⇒ 丢弃不补答,并留下 WARN + 计数 + 审计。 +// - **背压有界且诚实**:每会话车道深度有界。群聊 / 陌生人满时**在门前拒收这一条** +// (队列里已有的条目一条都不动);**主人私聊永不拒收**,只排队(见 offer)。 +// +// 本切片**不改「回什么」的语义**:该回的回、该静的仍然静,也没有新增任何「系统繁忙」 +// 式回复 —— 丢弃只走审计轨(与 flowgate 的 rate_limited / review_pruned 同一条流水)。 +// 静默窗口合并、回合中来了新消息就续跑一轮,属于下一片(M23),不在这里做。 +// +// 四条不变量(每条都在 lane_test.go 里有一条断言): +// +// ① 顺序:同一会话内的回复顺序与到达顺序一致(FIFO 车道 + 单 worker)。 +// ② 一致性:一轮内上下文是快照,同一会话绝不会有两个回合同时跑(车道串行; +// 主人私聊跨平台共用一条 `main` 车道,因为它们都路由进同一个 Web 主会话)。 +// ③ 完整性:被丢的只有「超龄」与「背压」两类,两者都留 WARN + 计数 + 审计。 +// ④ 最新性:回合上下文在**出队之后**才装配(processFrom 原样保留,见 runLaneItem), +// 所以一条排队中的消息看到的是它开始那一刻的最新历史,而不是入队时的快照。 + +import ( + "context" + "log/slog" + "sync" + "time" + + "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/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" +) + +const ( + // laneLogSource 是车道的子功能日志来源(AGENTS §10:日志必须标来源)。 + laneLogSource = "services/channelbridge/lane" + + // defaultLaneDepth 是**每会话**车道能排的条目数。取值依据: + // - 够吸收一次「连着发几条」的群聊突发(微信/QQ 里 5-8 条连发是常态, + // 尤其是几个人同时在说话而她正在答上一个问题); + // - 又小到车道永远不会变成「按条补答」的长队伍:一轮 IM 回合(含工具循环) + // 通常 3-20s,深 8 的车道在最坏情况下意味着约 2-3 分钟的滞后,那时排在后面的 + // 条目大多已过陈旧度闸,多留只会多丢; + // - 内存上界随之确定:条目数 ≤ 活跃会话数 × 8,绝不无界增长。 + defaultLaneDepth = 8 + + // defaultLaneStaleAfter 是**陈旧度闸**的默认阈值(可经 Options.LaneStaleAfter 覆盖, + // 见 channelbridge.go)。取值依据: + // - 一轮 IM 回合(含工具循环)实测多在 3-20s 这一档,30s ≈ 两倍上限; + // - 群聊里超过半分钟的「回答」在用户看来已经是「答上一句」,补它比不补更糟 + // (正是报障里的「后面才回复前面的话」); + // - 比它更短会误伤正常的「她正在打字」等待(一个人等她回答 10-20s 是正常的)。 + defaultLaneStaleAfter = 30 * time.Second + + // laneDropReasonStale / laneDropReasonOverflow 是两类**唯一**允许的丢弃原因, + // 进 onRejected 审计轨时用的原因标记(与 flowgate 的 rate_limited / + // review_pruned / capped 同一口径,面板与排障据此聚合)。 + laneDropReasonStale = "lane_stale" + laneDropReasonOverflow = "lane_overflow" + + // ownerLaneBacklogWarn 是主人私聊溢出区的告警水位:越过它(并按倍率 16/32/64…) + // 记一条 WARN,表明「她落后了这么多条主人私聊」—— 这是主人私聊车道的**唯一**可观测 + // 上界信号,它不丢任何一条(主人 2026-10-24 裁定:宁可晚回,不许不回)。 + ownerLaneBacklogWarn = 16 + + // ownerPrivateLaneKey 是主人私聊车道的键:那条 lane 对应的是「Web 主会话」这个 + // **目的地**(inboundTargetFor 把各平台的主人私聊都路由进去),因此与会话键故意不同。 + ownerPrivateLaneKey = "main|" + history.MainSessionID + + // laneKeySep 只用于「没有可用渠道」的兜底键(见 laneKey);真实车道的键逐字来自 + // session.Key,本来就是 `platform \x00 channelID`。 + laneKeySep = "\x00" +) + +// laneItem 是车道上的一条待办:入站消息 + 它的账号适配器 + 已解析出的 actor。 +// +// actor 在入口(onMessage)已经解析过一次(结构化判据,不读正文),这里带着走, +// 使陈旧度闸不必重解析,也不必为了判「主人私聊」而多一次会话查询。 +type laneItem struct { + msg adapter.UnifiedMessage + a adapter.PlatformAdapter + actor adapter.Actor + arrivedAt time.Time +} + +// lane 是一个会话的回合通道:有界 FIFO + 一个 worker(串行 ⇒ 不变量 ①②)。 +// +// neverDrop 只对**主人私聊**车道为真:它满了不拒收,条目改落 spill(溢出区,只排队)。 +// spill 恒为空的车道因此不占额外内存。 +type lane struct { + key string + ch chan laneItem + neverDrop bool + spill []laneItem +} + +// laneSet 是全部活跃车道的注册表。车道在**空闲时退休**(worker 见队列空即把自己摘掉), +// 因此表的大小只与「此刻有排队/在跑的会话数」相关。 +type laneSet struct { + depth int + staleAfter time.Duration + log *slog.Logger + + mu sync.Mutex + lanes map[string]*lane + droppedStale int64 + droppedOverflow int64 +} + +// newLaneSet 构造车道注册表;depth / staleAfter 为 0 时取上面的默认值。 +func newLaneSet(depth int, staleAfter time.Duration, log *slog.Logger) *laneSet { + if depth <= 0 { + depth = defaultLaneDepth + } + if staleAfter <= 0 { + staleAfter = defaultLaneStaleAfter + } + if log == nil { + log = logger.New(logger.Options{}) + } + return &laneSet{ + depth: depth, + staleAfter: staleAfter, + log: logger.WithSource(log, laneLogSource), + lanes: make(map[string]*lane), + } +} + +// LaneStats 是车道的计数快照:**被丢弃的入站有多少、为什么**(不变量 ③ 的可审计面), +// 加上主人私聊溢出区的当前积压。 +type LaneStats struct { + // ActiveLanes 是此刻有排队或在跑的会话车道数。 + ActiveLanes int + // DroppedStale 是因陈旧度闸被丢弃的条目累计数(超龄不补答)。 + DroppedStale int64 + // DroppedOverflow 是车道满时**在门前被拒收**的条目累计数。它只可能落在群聊 / + // 陌生人私聊上:主人私聊车道永不拒收(见 offer 的 neverDrop)。 + DroppedOverflow int64 + // Backlog 是主人私聊溢出区里此刻还排着的条目数(其余车道恒为 0)。 + Backlog int +} + +// LaneStats 返回车道计数的快照。它是「丢弃是诚实且有界」的进程内数字答案。 +// +// 生产可见性的口径(复审 P2-3):**不新开 HTTP 端点**(那是契约变更,归塔)。真正 +// 生产可见的是每一条丢弃 WARN 里带的 `dropped_stale_total` / `dropped_overflow_total` +// (累计计数随日志落地)以及 `onRejected` 审计轨;本方法只是给进程内调用方(管理面板 +// 将来接线、运维探针、测试)读同一组数字的窄口,不构成对外契约。 +func (b *Bridge) LaneStats() LaneStats { + if b.lanes == nil { + return LaneStats{} + } + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + backlog := 0 + for _, l := range b.lanes.lanes { + backlog += len(l.spill) + } + return LaneStats{ + ActiveLanes: len(b.lanes.lanes), + DroppedStale: b.lanes.droppedStale, + DroppedOverflow: b.lanes.droppedOverflow, + Backlog: backlog, + } +} + +// enqueueInbound 把一条**已过流控闸**的入站消息投进它所属会话的车道,并立刻返回 —— +// 这是「读与回水解耦」的全部:适配器的读循环调用它之后马上可以继续读 socket。 +// +// 🔴 这就是留给下一片(M23 静默窗口合并)的**入队 seam**:入参是一条**已过闸门的 +// prepared item**(Validate/ResolveActor/Admit 都已在本函数的调用方 onMessage 里做完, +// 连同解析结果 actor 一起带进来)。车窗车道头部(laneWorker)只做「等待 + 发车」, +// 绝不在这里做准入判断 —— M23 要把 processFrom 的闸门段再往前抽成 prepareInbound 时, +// 只需在调用方一侧重排,车道本身不用动。 +// +// 三件事在这里一次性做完: +// - 认领车道(没有就建一条并起 worker); +// - 入队(群聊/陌生人满则**在门前拒收这一条**并留痕;主人私聊**永不拒收**); +// - 与 Stop 的 `wg.Wait` 对齐:`b.wg.Add` 与 `closed` 的判定在同一把 `b.mu` 下, +// 因此 Stop 要么看见这个工作项、要么我们看见 closed 而不入队,二者必居其一。 +// +// 锁顺序:全树只有这一处同时取两把锁,顺序恒为 `b.mu` → `lanes.mu`。 +func (b *Bridge) enqueueInbound(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter, actor adapter.Actor) { + if b.lanes == nil { + return + } + item := laneItem{msg: msg, a: a, actor: actor, arrivedAt: b.now()} + key := laneKey(msg, actor) + // 主人私聊(私聊 + 主人在场,由 actor 元数据 + 白名单结构化判定)**永不丢弃**: + // 队列满时落到溢出区只排队(主人 2026-10-24 裁定:「宁可晚回,不许不回」)。 + neverDrop := ownerPrivateLane(msg, actor) + + b.mu.Lock() + if b.closed { + b.mu.Unlock() + b.lanes.log.Debug("bridge closed, inbound not queued", + "platform", msg.Platform, "channel", msg.ChannelID) + return + } + l, start, rejected := b.lanes.offer(key, item, neverDrop) + if start { + b.wg.Add(1) + } + b.mu.Unlock() + + if start { + go b.laneWorker(ctx, l) + } + if rejected { + b.dropInbound(item, laneDropReasonOverflow, "lane queue full") + } +} + +// offer 在**一个临界区**内完成「查/建车道 + 入队」,使 `retireLaneIfIdle` 不可能夹在 +// 中间把刚查到、还没写入的车道摘掉(那样条目会落进一条没有 worker 的死车道)。 +// +// 队列满时的策略分两档(都不是静默的): +// - **主人私聊**(neverDrop):绝不拒收,落到溢出区 `spill` 里**只排队**,并按倍率 +// (ownerLaneBacklogWarn 及其翻倍)留 WARN —— 一个白名单账号的积压只反映「她落后了」, +// 不构成第三方洪泛面; +// - **群聊 / 陌生人**:**在门前拒收这一条**(而不是挤掉队列里已有的某条)。队列里 +// 的条目因此一条都不会凭空消失 —— 下一片 M23 的 takeAll 依赖这一点;被拒收的那条 +// 由调用方留痕(WARN + 计数 + 审计)。 +// +// 🔴 **FIFO 的关键**:`spill` 一旦非空,后面的条目**一律继续追加到 spill**,不再回写 +// `ch`。否则「ch 腾出一个位置 + 此时正好有新条目到达」会让后到者超车(复审实测顺序 +// `[m1 m2 m3 m5 m4]`)。`nextLaneItem` 因此可以安全地「先取完 ch、再按序取 spill」: +// spill 里的一定晚于 ch 里的,而 spill 一旦开始就用到底(直到它被取空)。 +func (s *laneSet) offer(key string, item laneItem, neverDrop bool) (l *lane, start bool, rejected bool) { + s.mu.Lock() + defer s.mu.Unlock() + l = s.lanes[key] + if l == nil { + l = &lane{key: key, ch: make(chan laneItem, s.depth), neverDrop: neverDrop} + s.lanes[key] = l + start = true + } + if l.neverDrop && len(l.spill) > 0 { + s.spillLocked(l, item) + return l, start, false + } + select { + case l.ch <- item: + return l, start, false + default: + } + if l.neverDrop { + s.spillLocked(l, item) + return l, start, false + } + return l, start, true +} + +// spillLocked 把条目追加进主人私聊车道的溢出区,并按**倍率**告警 +// (16/32/64/128… 每翻一倍一条 WARN):固定水位只会在恰好等于时响一次,之后继续涨就 +// 再无信号,而「她落后了多少」正是主人侧唯一可观测的量。调用方必须持有 s.mu。 +func (s *laneSet) spillLocked(l *lane, item laneItem) { + l.spill = append(l.spill, item) + if n := len(l.spill); n >= ownerLaneBacklogWarn && isPowerOfTwo(n) { + s.log.Warn("owner private lane backlog is growing, still queueing", + "backlog", n, "lane", l.key, + "platform", item.msg.Platform, "channel", item.msg.ChannelID) + } +} + +// isPowerOfTwo 报告 n 是否为 2 的幂(n ≥ 1)。 +func isPowerOfTwo(n int) bool { return n > 0 && n&(n-1) == 0 } + +// ownerPrivateLane 报告这条消息是否属于「主人私聊」车道 —— 也就是**永不丢弃**的那条。 +// 判据与 laneKey 完全同源(私聊 + actor 元数据说他是主人),因此不会出现「键说是主人、 +// 丢弃策略说不是」的两套口径。这里绝不读正文:身份只来自 actor 元数据 + 白名单。 +func ownerPrivateLane(msg adapter.UnifiedMessage, actor adapter.Actor) bool { + return !msg.IsGroup() && actor.IsOwner() +} + +// nextLaneItem 按 FIFO 取下一条:先把缓冲区取完,再取主人私聊的溢出区(溢出区里的条目 +// 一定晚于缓冲区里的,所以顺序仍然是到达顺序)。ok=false 表示此刻真的没有待办。 +func (b *Bridge) nextLaneItem(l *lane) (laneItem, bool) { + select { + case item := <-l.ch: + return item, true + default: + } + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + if len(l.spill) == 0 { + return laneItem{}, false + } + item := l.spill[0] + copy(l.spill, l.spill[1:]) + l.spill = l.spill[:len(l.spill)-1] + return item, true +} + +// laneWorker 是某会话车道的唯一执行者:串行跑完每一条,队列空了就把车道交还注册表 +// (空闲车道不占内存,也不会把「这个人再说话」变慢)。 +func (b *Bridge) laneWorker(ctx context.Context, l *lane) { + defer b.wg.Done() + for { + if item, ok := b.nextLaneItem(l); ok { + b.runLaneItem(ctx, item) + continue + } + // 队列空了才谈退休:摘表与入队用同一把锁,因此「摘下」与「刚有人塞进来」不会 + // 交错成条目丢失(见 retireLaneIfIdle 的说明)。 + if b.retireLaneIfIdle(l) { + return + } + select { + case <-ctx.Done(): + // 桥停了:把车道摘掉,未处理的条目随进程取消一起作废(Stop 的语义)。 + b.retireLane(l) + return + case item := <-l.ch: + b.runLaneItem(ctx, item) + } + } +} + +// runLaneItem 跑车道上的一条:先取全局回合槽(**在线程池里等,不在读循环里等**), +// 再过陈旧度闸,最后交给原样的入站流水线。 +// +// 陈旧度在**取到槽之后**才判:排队等槽的那段时间同样是「让它等」,必须计入年龄, +// 否则一道繁忙的全局上限会把「等了很久」洗成「刚到」。 +func (b *Bridge) runLaneItem(ctx context.Context, item laneItem) { + select { + case b.sem <- struct{}{}: + case <-ctx.Done(): + return + } + defer func() { <-b.sem }() + + waited := b.now().Sub(item.arrivedAt) + if b.staleInbound(item, waited) { + return + } + // 「这条等了多久才开跑」是「她为什么回得晚」在生产日志里最直接的那个答案; + // 与它同时出现在日志里的 `msg="inbound frame"`(适配器的入站帧 DEBUG)则说明 + // 读循环在等的同时一直在读 —— 两行合起来就是「回合没有堵住心跳」的证据。 + b.lanes.log.Debug("lane turn started", + "waited", waited.String(), + "platform", item.msg.Platform, "channel", item.msg.ChannelID, + "sender", item.msg.SenderID, "group", item.msg.GroupID) + // 不变量 ④:上下文在这里(出队之后)才装配,因此一轮看到的是最新历史。 + b.processFrom(ctx, item.msg, item.a) +} + +// staleInbound 实现陈旧度闸:轮到时已超龄 ⇒ 丢弃不补答,并留 WARN + 计数 + 审计。 +// 返回 true 表示这条已被丢弃。waited 由调用方算好(含等全局回合槽的时间)。 +func (b *Bridge) staleInbound(item laneItem, waited time.Duration) bool { + if !laneStaleEligible(item) { + return false + } + if waited <= b.lanes.staleAfter { + return false + } + b.dropInbound(item, laneDropReasonStale, "waited past the staleness threshold") + return true +} + +// laneStaleEligible 报告某条入站是否受陈旧度闸约束。两类**豁免**各有硬理由: +// +// - **非聊天事件**(notice / request / meta_event / 解析失败兜底,ContentType=action): +// 它们不产生「回答旧话」这种失真,却是渠道事件留存(design/26 §5)与审计的唯一来源。 +// 按年龄丢掉它们只会让留存出现空洞。 +// - **主人私聊**:「回主人私聊」不是对群聊话题的应答,迟到的回答仍然是回答; +// 丢它等于对主人静默,直接违反「该回的回」这条红线。 +func laneStaleEligible(item laneItem) bool { + if item.msg.ContentType == adapter.ContentAction { + return false + } + if !item.msg.IsGroup() && item.actor.IsOwner() { + return false + } + return true +} + +// dropInbound 给「被丢弃的入站」留下**三样痕迹**,缺一不可: +// - 一条带 source 的 WARN(谁、哪个平台/会话/发送者、等了多久、为什么),**并且带上 +// 两类丢弃的累计计数** —— 计数因此不需要新端点就能在生产日志里被读到(复审 P2-3: +// 计数只活在进程内存里等于没有出口); +// - 按原因的累计计数(LaneStats,供进程内调用方/后续面板接线读,不新开端点); +// - onRejected 审计轨(与 flowgate 的 rate_limited / review_pruned 同一条)。 +// +// 它刻意**不**回一条「系统繁忙」:回什么属于回合语义,本切片不动它。 +func (b *Bridge) dropInbound(item laneItem, reason, detail string) { + stale, overflow := b.lanes.noteDrop(reason) + b.lanes.log.Warn("inbound dropped by session lane", + "reason", reason, "detail", detail, + "platform", item.msg.Platform, "channel", item.msg.ChannelID, + "sender", item.msg.SenderID, "message_id", item.msg.MessageID, + "waited", b.now().Sub(item.arrivedAt).String(), + "dropped_stale_total", stale, "dropped_overflow_total", overflow) + if b.onRejected != nil { + b.onRejected(item.msg, reason) + } +} + +// noteDrop 累加某一类丢弃的计数,并返回累加后的两类计数快照(写进 WARN 用)。 +func (s *laneSet) noteDrop(reason string) (stale, overflow int64) { + s.mu.Lock() + defer s.mu.Unlock() + switch reason { + case laneDropReasonStale: + s.droppedStale++ + case laneDropReasonOverflow: + s.droppedOverflow++ + } + return s.droppedStale, s.droppedOverflow +} + +// retireLaneIfIdle 在车道已空时把它从注册表摘掉并报告 true(worker 随即退出)。 +// +// 它必须与 `offer` 用同一把锁:两者交错的话会出现「offer 拿到车道 → worker 退休并删除 → +// offer 往一条没有 worker 的车道入队」的丢失。入队与「查空」在同一把锁下,就不存在这个窗口。 +func (b *Bridge) retireLaneIfIdle(l *lane) bool { + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + if len(l.ch) > 0 || len(l.spill) > 0 || b.lanes.lanes[l.key] != l { + return false + } + delete(b.lanes.lanes, l.key) + return true +} + +// retireLane 无条件把车道从注册表摘掉(桥停止时的清理;未处理的条目随取消作废)。 +func (b *Bridge) retireLane(l *lane) { + b.lanes.mu.Lock() + defer b.lanes.mu.Unlock() + if b.lanes.lanes[l.key] == l { + delete(b.lanes.lanes, l.key) + } +} + +// laneKey 决定「哪些消息共用一条车道」,也就是「什么算同一个会话」。它必须与 +// **会话键同源**(M22 复审 P1-1):两套各算各的判据会让同一个 session 出现两条车道、 +// 两个回合并发,而两条 run 共享 sessionID ⇒ checkpoint 行 id `{sessionID}:{step}` 互相 +// 覆盖、`loadCheckpoint` 可能续到另一条在途 run 的快照而重放工具调用,flow 桶 +// `p.graphs[sessionID]` 被先收尾者删掉(loop 没有会话级互斥,IM 侧也没有网关那层 +// `session_turn_fence`)。 +// +// - **主人私聊**:无论来自哪个平台,都被 inboundTargetFor 路由进同一个 Web 主会话 +// (history.MainSessionID),因此共用一条 `main` 车道。这里的键与会话键**故意不同** +// ——`main` 是那条路由的目的地,比渠道键更强(把各平台的主人私聊并成一条)。 +// - 其余:**逐字就是** `session.Key(msg)`(`platform \x00 channelID`,与会话层缓存键 +// 同一个出处),所以「会话层认为是一个会话」与「车道认为是一个会话」不可能分叉。 +// +// 🔴 **取舍(比旧实现少了一个维度,是有意的)**:键里**没有账号名**。同一平台的两个账号 +// 收到同一对端/同一个群的消息时,它们是**同一个 session**(会话键只含 platform 与 +// channel),因此必须共用一条车道 —— 分开就会让同一会话两个回合并发。副作用是同一对端 +// 的多个账号的入站会串行排队;回复出口不受影响:出站路由取的是**每条消息自己的** +// `msg.Platform`/`ChannelID`/`GroupID` 与它自己的适配器(`laneItem.a`),车道本身从不 +// 持有任何账号级状态。 +// +// 同理也**不要**在这里重新引入「频道」维度:kook 同一 guild 的两个 text channel 在会话层 +// 是同一个群会话(键按 `group:`),按 `ChannelID` 分车道同样会造成双车道。 +func laneKey(msg adapter.UnifiedMessage, actor adapter.Actor) string { + if ownerPrivateLane(msg, actor) { + return ownerPrivateLaneKey + } + if session.ChannelKey(msg) == "" { + // 没有任何渠道可用(私聊既无 ChannelID 也无 SenderID):会话层会以 + // ErrInvalidChannel 拒掉这条消息,这里只需要一个不会与真实会话撞键的稳定键。 + return "unbound" + laneKeySep + adapter.ConversationKey(msg) + } + return session.Key(msg) +} diff --git a/internal/services/channelbridge/lane_test.go b/internal/services/channelbridge/lane_test.go new file mode 100644 index 0000000..494d36c --- /dev/null +++ b/internal/services/channelbridge/lane_test.go @@ -0,0 +1,851 @@ +package channelbridge + +// 本文件是 **M22 切片 LANEQUEUE** 的断言集:每条不变量至少一条断言,见 lane.go 顶部的 +// 四条不变量清单。 +// +// - ① 顺序:同一会话内回复顺序 = 到达顺序(TestLaneSerializesOneSessionInArrivalOrder; +// 主人私聊溢出区另有一条 TestLaneOwnerPrivateSpillKeepsArrivalOrder); +// - ② 一致性:同一会话绝无两个回合同时跑,不同会话可并行 +// (TestLaneSerializesOneSessionInArrivalOrder + TestLaneRunsDifferentSessionsInParallel +// + TestLaneKeyNeverCollidesAcrossSessions —— 后者钉住「车道键与会话键同源」); +// - ③ 完整性:被丢的只有「超龄」与「背压」两类,两者都留计数 + 审计 + 日志 +// (TestLaneDropsStaleInboundAndTracesIt / TestLaneRejectsNewestWhenFull), +// 主人私聊与非聊天事件另有豁免(TestLaneStaleGateExemptions); +// - ④ 最新性:排队中的消息在**开跑之后**才装配上下文,因此看得到前一轮刚落的历史 +// (TestLaneTurnSeesContextAssembledAtStart)。 +// +// 另有 TestLaneDeliveryNeverBlocksTheReadLoop:把「读循环 + 回合」的旧形状与新形状放在 +// 一起对照 —— 全局回合槽占满时投递仍然立即返回,读循环照常读下一帧(心跳不再被堵)。 + +import ( + "context" + "fmt" + "log/slog" + "strings" + "sync" + "testing" + "time" + + "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" +) + +// laneAdapter 是并发安全的最小适配器:车道让多个会话**并行**发回复,因此这里必须带锁 +// (channelbridge_test.go 的 fakeAdapter 是单线程夹具,不能复用)。 +type laneAdapter struct { + name string + + mu sync.Mutex + sends []adapter.UnifiedMessage + handler adapter.MessageHandler +} + +func (a *laneAdapter) Name() string { return a.name } +func (a *laneAdapter) Config() map[string]any { return map[string]any{"bot_token": "tok"} } +func (a *laneAdapter) Connect(context.Context) error { return nil } +func (a *laneAdapter) Disconnect(context.Context) error { return nil } +func (a *laneAdapter) IsConnected() bool { return true } +func (a *laneAdapter) HealthCheck(context.Context) error { return nil } +func (a *laneAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) { + <-ctx.Done() +} +func (a *laneAdapter) SendMessage(_ context.Context, resp adapter.UnifiedResponse) error { + a.mu.Lock() + defer a.mu.Unlock() + a.sends = append(a.sends, resp.Messages...) + return nil +} +func (a *laneAdapter) SendProactive(_ context.Context, msg adapter.UnifiedMessage) error { + return a.SendMessage(context.Background(), adapter.UnifiedResponse{Messages: []adapter.UnifiedMessage{msg}}) +} +func (a *laneAdapter) SetMessageHandler(fn adapter.MessageHandler) { + a.mu.Lock() + a.handler = fn + a.mu.Unlock() +} +func (a *laneAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() } +func (a *laneAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { + return nil, nil +} + +// laneChatter 记录「谁在什么时候跑」并可在 Run 里阻塞,用来制造「回合还在跑」的窗口。 +type laneChatter struct { + // onRun 在**记录之后、返回之前**调用;非 nil 时可用它阻塞或推进注入的时钟。 + onRun func(user string) + + mu sync.Mutex + runs []laneRun + inFlight int + maxInFlight int + perSession map[string]int + maxPerSess map[string]int +} + +// laneRun 是一个已开跑的回合(当次用户文本 + 会话 + 模型可见消息)。 +type laneRun struct { + user string + session string + messages []model.Message +} + +func (c *laneChatter) Run(_ context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) { + use := lastUserText(msgs) + c.mu.Lock() + if c.perSession == nil { + c.perSession = map[string]int{} + c.maxPerSess = map[string]int{} + } + c.runs = append(c.runs, laneRun{user: use, session: sessionID, messages: msgs}) + c.inFlight++ + if c.inFlight > c.maxInFlight { + c.maxInFlight = c.inFlight + } + c.perSession[sessionID]++ + if c.perSession[sessionID] > c.maxPerSess[sessionID] { + c.maxPerSess[sessionID] = c.perSession[sessionID] + } + c.mu.Unlock() + + if c.onRun != nil { + c.onRun(use) + } + + c.mu.Lock() + c.inFlight-- + c.perSession[sessionID]-- + c.mu.Unlock() + return &loop.Result{Text: "reply:" + use}, nil +} + +func (c *laneChatter) userTexts() []string { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]string, 0, len(c.runs)) + for _, r := range c.runs { + out = append(out, r.user) + } + return out +} + +func (c *laneChatter) runCount() int { + c.mu.Lock() + defer c.mu.Unlock() + return len(c.runs) +} + +func (c *laneChatter) maxConcurrent() int { + c.mu.Lock() + defer c.mu.Unlock() + return c.maxInFlight +} + +func (c *laneChatter) maxConcurrentIn(sessionID string) int { + c.mu.Lock() + defer c.mu.Unlock() + return c.maxPerSess[sessionID] +} + +func (c *laneChatter) runAt(i int) laneRun { + c.mu.Lock() + defer c.mu.Unlock() + return c.runs[i] +} + +// waitRuns 等到至少 n 个回合开跑(或超时失败)。 +func (c *laneChatter) waitRuns(t *testing.T, n int, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if c.runCount() >= n { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("only %d of %d turns started within %s (lane never drained?)", c.runCount(), n, timeout) +} + +// laneSessions 按消息给出**稳定且互不相同**的隔离会话(同一发送者恒同一个会话)。 +type laneSessions struct { + owner bool +} + +func (s *laneSessions) GetOrCreate(_ context.Context, msg adapter.UnifiedMessage, _ *user.OwnerProfile) (session.Session, error) { + return session.Session{ + Platform: msg.Platform, + ChannelID: msg.ChannelID, + SessionID: "sess-" + msg.ChannelID + "-" + msg.SenderID, + Owner: s.owner, + IsGroup: msg.IsGroup(), + MemWritable: s.owner, + }, nil +} + +// laneClock 是可推进的注入时钟(陈旧度闸据此判定,测试因此不必真的等 30s)。 +type laneClock struct { + mu sync.Mutex + t time.Time +} + +func newLaneClock() *laneClock { return &laneClock{t: time.Unix(1_700_000_000, 0)} } + +func (c *laneClock) Now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *laneClock) advance(d time.Duration) { + c.mu.Lock() + c.t = c.t.Add(d) + c.mu.Unlock() +} + +// laneRecorder 收集 onRejected 审计(并发安全)。 +type laneRecorder struct { + mu sync.Mutex + rejected []string +} + +func (r *laneRecorder) record(_ adapter.UnifiedMessage, reason string) { + r.mu.Lock() + r.rejected = append(r.rejected, reason) + r.mu.Unlock() +} + +func (r *laneRecorder) reasons() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.rejected...) +} + +func lastUserText(msgs []model.Message) string { + for i := len(msgs) - 1; i >= 0; i-- { + if msgs[i].Role != model.RoleUser { + continue + } + if t := lastText(msgs[i]); t != "" { + return t + } + } + return "" +} + +func lastText(m model.Message) string { + for _, c := range m.Content { + if c.Type == model.ContentTypeText { + return c.Text + } + } + return "" +} + +// strangerMsg 是一条**陌生人私聊**(非群、非主人):唯一闸门是流控,因此它必然进回合, +// 且不同发送者 = 不同会话。它是「车道并行」与「陈旧度闸」最干净的夹具。 +func strangerMsg(sender, content string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "telegram", + ChannelID: sender, + ChannelType: adapter.ChannelDirect, + SenderID: sender, + Sender: adapter.Actor{Role: adapter.RoleMember}, + Content: content, + ContentType: adapter.ContentText, + } +} + +// ownerMsg 是一条**主人私聊**:车道键恒为 main|web(inboundTargetFor 把它路由进主会话)。 +func ownerMsg(content string) adapter.UnifiedMessage { + m := directMessage() + m.Content = content + return m +} + +func newLaneBridge(t *testing.T, opts Options, ad *laneAdapter) *Bridge { + t.Helper() + b, err := New(opts) + if err != nil { + t.Fatalf("New: %v", err) + } + if ad != nil { + b.Add(ad) + } + return b +} + +// waitLaneIdle 等到注册表里没有活跃车道(队列已排空)或超时。 +func waitLaneIdle(t *testing.T, b *Bridge, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + if b.LaneStats().ActiveLanes == 0 { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("lanes never went idle: %+v", b.LaneStats()) +} + +// TestLaneSerializesOneSessionInArrivalOrder 断言不变量 ①(顺序)与 ②(一致性): +// 同一个会话(这里是主人私聊 ⇒ 同一条 main 车道)的五条消息**先来先跑、绝不同时跑**。 +func TestLaneSerializesOneSessionInArrivalOrder(t *testing.T) { + t.Parallel() + + chat := &laneChatter{} + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{owner: true}, + Owner: testOwnerProfile("telegram", "100"), + }, nil) + ad := &laneAdapter{name: "telegram"} + + const n = 5 + for i := 0; i < n; i++ { + b.onMessage(context.Background(), ownerMsg(fmt.Sprintf("m%d", i)), ad) + } + chat.waitRuns(t, n, 5*time.Second) + + got := chat.userTexts() + want := []string{"m0", "m1", "m2", "m3", "m4"} + if len(got) != len(want) { + t.Fatalf("ran %d turns, want %d: %v", len(got), len(want), got) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("turn order = %v, want %v", got, want) + } + } + // ②:同一会话绝不并发(主人私聊跨平台共用 main 车道,因此这里是会话级而非账号级)。 + if m := chat.maxConcurrent(); m != 1 { + t.Errorf("max concurrent turns = %d, want 1 for one session", m) + } + if m := chat.maxConcurrentIn(history.MainSessionID); m != 1 { + t.Errorf("max concurrent turns in %s = %d, want 1", history.MainSessionID, m) + } +} + +// TestLaneRunsDifferentSessionsInParallel 断言 ②的另一半:不同会话**互不排队** —— +// 两个陌生人各自私聊她,两个回合必须同时在跑(串行的话下面会以 maxInFlight=1 失败)。 +func TestLaneRunsDifferentSessionsInParallel(t *testing.T) { + t.Parallel() + + var mu sync.Mutex + entered := 0 + both := make(chan struct{}) + chat := &laneChatter{} + chat.onRun = func(string) { + mu.Lock() + entered++ + if entered == 2 { + close(both) + } + mu.Unlock() + select { + case <-both: + case <-time.After(3 * time.Second): + // 串行了:返回让断言失败,而不是把测试挂死。 + } + } + b := newLaneBridge(t, Options{Loop: chat, Sessions: &laneSessions{}, Owner: &user.OwnerProfile{}}, nil) + ad := &laneAdapter{name: "telegram"} + + b.onMessage(context.Background(), strangerMsg("200", "a"), ad) + b.onMessage(context.Background(), strangerMsg("201", "b"), ad) + chat.waitRuns(t, 2, 5*time.Second) + + if m := chat.maxConcurrent(); m != 2 { + t.Fatalf("max concurrent turns across two sessions = %d, want 2 (lanes must not share)", m) + } +} + +// TestLaneDropsStaleInboundAndTracesIt 断言 ③ 的第一类丢弃:**轮到时已超龄 ⇒ 不补答**, +// 且丢得有声:计数 + onRejected 审计 + 带 source 的 WARN 日志。 +func TestLaneDropsStaleInboundAndTracesIt(t *testing.T) { + t.Parallel() + + logs := &safeBuffer{} + clock := newLaneClock() + rec := &laneRecorder{} + chat := &laneChatter{} + chat.onRun = func(user string) { + if user == "first" { + // 第一轮跑着的时候时间越过阈值:排在后面的那条等到自己开跑时已超龄。 + clock.advance(defaultLaneStaleAfter + time.Second) + } + } + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{}, + Owner: &user.OwnerProfile{}, + Now: clock.Now, + OnRejected: rec.record, + Log: slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})), + }, nil) + ad := &laneAdapter{name: "telegram"} + + b.onMessage(context.Background(), strangerMsg("200", "first"), ad) + b.onMessage(context.Background(), strangerMsg("200", "stale by the time it runs"), ad) + chat.waitRuns(t, 1, 5*time.Second) + waitLaneIdle(t, b, 2*time.Second) + + if stats := b.LaneStats(); stats.DroppedStale != 1 { + t.Fatalf("dropped stale = %d, want 1 (stats=%+v)", stats.DroppedStale, stats) + } + if reasons := rec.reasons(); len(reasons) != 1 || reasons[0] != laneDropReasonStale { + t.Fatalf("audit reasons = %v, want [%s]", reasons, laneDropReasonStale) + } + if got := chat.userTexts(); len(got) != 1 || got[0] != "first" { + t.Fatalf("turns = %v, want only the fresh first one (the stale one must not run)", got) + } + if out := logs.String(); !strings.Contains(out, "inbound dropped by session lane") || + !strings.Contains(out, laneDropReasonStale) { + t.Fatalf("drop left no trace in the log:\n%s", out) + } +} + +// TestLaneRejectsNewestWhenFull 断言 ③ 的第二类丢弃:车道满时**在门前**拒收这一条 +// (队列里已有的条目一条都不动 —— M23 的 takeAll 依赖队列完整),且拒收留痕 +// (计数 + 审计)。深度 1 时:m1 在跑、m2 已排上,m3 被拒收。 +// 注意语义是「拒收最新、队列不动」,**不是**「挤掉最旧」(复审 P2-2 的口径澄清)。 +func TestLaneRejectsNewestWhenFull(t *testing.T) { + t.Parallel() + + gate := make(chan struct{}) + chat := &laneChatter{} + chat.onRun = func(user string) { + if user == "m1" { + <-gate // 第一个回合卡住,后面的条目只能在车道上等 + } + } + rec := &laneRecorder{} + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{}, + Owner: &user.OwnerProfile{}, + LaneDepth: 1, + OnRejected: rec.record, + }, nil) + ad := &laneAdapter{name: "telegram"} + + b.onMessage(context.Background(), strangerMsg("200", "m1"), ad) + chat.waitRuns(t, 1, 5*time.Second) // m1 已在跑(车道空出来) + b.onMessage(context.Background(), strangerMsg("200", "m2"), ad) + b.onMessage(context.Background(), strangerMsg("200", "m3"), ad) // 深度 1:m3 在门前被拒收 + close(gate) + chat.waitRuns(t, 2, 5*time.Second) + waitLaneIdle(t, b, 2*time.Second) + + if stats := b.LaneStats(); stats.DroppedOverflow != 1 { + t.Fatalf("dropped overflow = %d, want 1 (stats=%+v)", stats.DroppedOverflow, stats) + } + if reasons := rec.reasons(); len(reasons) != 1 || reasons[0] != laneDropReasonOverflow { + t.Fatalf("audit reasons = %v, want [%s]", reasons, laneDropReasonOverflow) + } + if got := chat.userTexts(); len(got) != 2 || got[0] != "m1" || got[1] != "m2" { + t.Fatalf("turns = %v, want [m1 m2] (m3 is the one rejected at the gate)", got) + } +} + +// 🔴 TestLaneOwnerPrivateNeverDroppedOnOverflow 守住主人 2026-10-24 的裁定: +// **主人私聊永不因背压被吃掉**(「宁可晚回,不许不回」)。深度 1、第一轮卡住时连投四条, +// 四条必须一条不落地按到达顺序跑完 —— 溢出区只排队、绝不拒收。 +// 变异检验:把主人私聊塞进群聊那条拒收路径 ⇒ 本用例立刻红(只剩两条回合)。 +func TestLaneOwnerPrivateNeverDroppedOnOverflow(t *testing.T) { + t.Parallel() + + gate := make(chan struct{}) + chat := &laneChatter{} + chat.onRun = func(user string) { + if user == "m1" { + <-gate + } + } + rec := &laneRecorder{} + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{owner: true}, + Owner: testOwnerProfile("telegram", "100"), + LaneDepth: 1, + OnRejected: rec.record, + }, nil) + ad := &laneAdapter{name: "telegram"} + + b.onMessage(context.Background(), ownerMsg("m1"), ad) + chat.waitRuns(t, 1, 5*time.Second) + for _, m := range []string{"m2", "m3", "m4"} { + b.onMessage(context.Background(), ownerMsg(m), ad) + } + close(gate) + chat.waitRuns(t, 4, 5*time.Second) + waitLaneIdle(t, b, 2*time.Second) + + if stats := b.LaneStats(); stats.DroppedOverflow != 0 || stats.DroppedStale != 0 { + t.Fatalf("owner private must never be dropped: %+v", stats) + } + if reasons := rec.reasons(); len(reasons) != 0 { + t.Fatalf("owner private must not reach the audit drop track: %v", reasons) + } + got := chat.userTexts() + want := []string{"m1", "m2", "m3", "m4"} + if len(got) != len(want) { + t.Fatalf("turns = %v, want all %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("turn order = %v, want %v", got, want) + } + } +} + +// TestLaneStaleGateExemptions 钉住陈旧度闸的两类豁免:**主人私聊**(迟到的回答仍是回答) +// 与**非聊天事件**(留存与审计的唯一来源)。两者都不许被算进丢弃计数。 +func TestLaneStaleGateExemptions(t *testing.T) { + t.Parallel() + + clock := newLaneClock() + chat := &laneChatter{} + chat.onRun = func(user string) { + if user == "first" { + clock.advance(defaultLaneStaleAfter + time.Second) + } + } + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{owner: true}, + Owner: testOwnerProfile("telegram", "100"), + Now: clock.Now, + }, nil) + ad := &laneAdapter{name: "telegram"} + + b.onMessage(context.Background(), ownerMsg("first"), ad) + b.onMessage(context.Background(), ownerMsg("late owner private"), ad) + chat.waitRuns(t, 2, 5*time.Second) + + // 非聊天事件(ContentAction)同样豁免:它不产生回复,但必须被登记/留存。 + event := strangerMsg("200", "") + event.ContentType = adapter.ContentAction + b.onMessage(context.Background(), event, ad) + waitLaneIdle(t, b, 2*time.Second) + + if stats := b.LaneStats(); stats.DroppedStale != 0 || stats.DroppedOverflow != 0 { + t.Fatalf("exempt messages must never be counted as dropped: %+v", stats) + } + if got := chat.userTexts(); len(got) < 2 || got[1] != "late owner private" { + t.Fatalf("owner private turns = %v, want the late one to run despite its age", got) + } +} + +// TestLaneTurnSeesContextAssembledAtStart 断言 ④(最新性):一条在车道上排队的消息, +// 开跑时看到的是**当时**的历史 —— 前一轮刚落库的回复就在它的上下文里。 +// 若上下文在入队时就被快照,第二条绝看不到 reply:first。 +func TestLaneTurnSeesContextAssembledAtStart(t *testing.T) { + t.Parallel() + + gate := make(chan struct{}) + chat := &laneChatter{} + chat.onRun = func(user string) { + if user == "first" { + <-gate // 让第二条在车道上等,第一轮的落库全在它之后 + } + } + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{owner: true}, + Owner: testOwnerProfile("telegram", "100"), + History: newLaneHistory(), + }, nil) + ad := &laneAdapter{name: "telegram"} + + b.onMessage(context.Background(), ownerMsg("first"), ad) + chat.waitRuns(t, 1, 5*time.Second) + b.onMessage(context.Background(), ownerMsg("second"), ad) + close(gate) + chat.waitRuns(t, 2, 5*time.Second) + + var assistant []string + for _, m := range chat.runAt(1).messages { + if m.Role == model.RoleAssistant { + assistant = append(assistant, lastText(m)) + } + } + // 转录会把历史行渲染成段落(`

…

`),因此按包含判定,不按字面相等。 + joined := strings.Join(assistant, "\n") + if !strings.Contains(joined, "reply:first") { + t.Fatalf("second turn assistant transcript = %q, want it to contain reply:first", joined) + } +} + +// TestLaneDeliveryNeverBlocksTheReadLoop 是**读与回水解耦**的回归断言。 +// +// 旧形状(报障现场):适配器的读循环调用 onMessage,onMessage 同步抢 16 个全局回合槽, +// 槽满即**停止读 socket** —— 心跳不被回、后续帧不被解析,消息堆在连接缓冲里,跑完一轮 +// 再按条重放(「后面才回复前面的话」)。 +// +// 新形状:投递立刻返回(进车道排队),全局回合槽在车道 worker 里取。这里把 16 个槽全部 +// 占满,然后走一遍「读循环 → onMessage(一帧) → 心跳帧」:心跳必须立刻被读到, +// 而那一帧的回合一个都不许开跑(槽确实还满着)。 +func TestLaneDeliveryNeverBlocksTheReadLoop(t *testing.T) { + t.Parallel() + + chat := &laneChatter{} + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{owner: true}, + Owner: testOwnerProfile("telegram", "100"), + }, nil) + ad := &laneAdapter{name: "telegram"} + + // 占满全局回合槽:此刻任何一个回合都无法开跑。 + for i := 0; i < maxConcurrent; i++ { + b.sem <- struct{}{} + } + + // 读循环的形状:串行处理每一帧;「心跳」是同一循环里的一帧(真实的 qq 适配器在 + // readConn 里对 opPing 回 opPong,位置就在 dispatch 之前两行)。 + type frame struct { + msg *adapter.UnifiedMessage + pong chan struct{} + } + frames := make(chan frame, 4) + loopCtx, stopReader := context.WithCancel(context.Background()) + defer stopReader() + readerDone := make(chan struct{}) + go func() { + defer close(readerDone) + for { + select { + case <-loopCtx.Done(): + return + case f := <-frames: + if f.pong != nil { + close(f.pong) + continue + } + b.onMessage(loopCtx, *f.msg, ad) // ← 投递必须立即返回 + } + } + }() + + slow := ownerMsg("slow turn") + frames <- frame{msg: &slow} + pong := make(chan struct{}) + frames <- frame{pong: pong} + select { + case <-pong: + case <-time.After(2 * time.Second): + t.Fatal("read loop never reached the heartbeat frame: delivery is blocking the reader") + } + // 槽还满着 ⇒ 回合一个都不该开跑;投递却已经返回并把消息排进了车道。 + if n := chat.runCount(); n != 0 { + t.Fatalf("turns started = %d while every global slot was taken, want 0", n) + } + + // 让位:释放槽,回合照旧跑完(排队不等于丢弃)。 + for i := 0; i < maxConcurrent; i++ { + <-b.sem + } + chat.waitRuns(t, 1, 5*time.Second) + stopReader() + <-readerDone +} + +// laneHistory 是最小的历史存储替身:只实现入站流水线真正用到的两条(AppendFull / Recent), +// 其余方法由内嵌的 nil 接口兜住(被调用即 panic —— 测试因此不会静默走错路)。 +type laneHistory struct { + history.HistoryStore + + mu sync.Mutex + rows map[string][]history.Message +} + +func newLaneHistory() *laneHistory { + return &laneHistory{rows: map[string][]history.Message{}} +} + +func (h *laneHistory) AppendFull(_ context.Context, msg history.Message) (history.Message, error) { + h.mu.Lock() + defer h.mu.Unlock() + h.rows[msg.SessionID] = append(h.rows[msg.SessionID], msg) + return msg, nil +} + +func (h *laneHistory) Recent(_ context.Context, sessionID string, n int) ([]history.Message, error) { + h.mu.Lock() + defer h.mu.Unlock() + rows := h.rows[sessionID] + if len(rows) > n { + rows = rows[len(rows)-n:] + } + return append([]history.Message(nil), rows...), nil +} + +// kookGroupMsg 是一条 kook 群消息的**生产形状**:`ChannelID` 是 text channel(频道), +// `GroupID` 是 guild。发送者是白名单里的主人且这条**指向她本人**(`AddressedToMe`, +// 结构化字段),因此它必定进回合 —— 与真人「主人在群里 @ 她」同一条路。 +// +// 它会进回合而**不受**非主人群聊的最小打扰分级影响,这正是我们要的:本用例要观察的 +// 只有「车道怎么分」,不是「要不要回答」。 +func kookGroupMsg(channelID, groupID, content string) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "kook", + ChannelID: channelID, + ChannelType: adapter.ChannelGroup, + GroupID: groupID, + SenderID: "100", + Sender: adapter.Actor{Role: adapter.RoleOwner}, + Content: content, + ContentType: adapter.ContentText, + AddressedToMe: true, + } +} + +// TestLaneKeyNeverCollidesAcrossSessions 钉住复审 P1-1 的修法:**车道键与会话键同源** +// (`laneKey` 逐字就是 `session.Key`),因此「同一个 session」绝不可能有两条车道。 +// +// 表里必须含复审给出的两条**真实反例**(它们在旧实现 `a.Name()+"|"+ConversationKey` +// 下会分成两条车道、两个回合并发): +// - kook:同一 guild 的两个 text channel(c1/c2)—— 会话层按 `group:` 判成 +// 同一个群会话,旧键却按 `ChannelID` 拆开; +// - telegram:同一平台两个账号对同一对端(`telegram:conf0` / `telegram:conf1`)—— +// 会话键只含 platform + channel,旧键却把账号名算进车道。 +// +// 断言方式是**行为级**的(不只看键字符串):第一条回合卡住时,第二条必须还在排队; +// 而控制行(不同群)必须能真的并行,证明本修法没有把所有会话串成一条车道。 +func TestLaneKeyNeverCollidesAcrossSessions(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + // 两条消息:同一 session 的必须共用一条车道;控制行则是两个不同 session。 + first, second adapter.UnifiedMessage + // adapterNames 是两个账号的注册名(多账号场景不同,单账号/单频道场景相同)。 + firstAdapter, secondAdapter string + // sessionOwner 是会话替身报告的身份车道:主人消息 true、陌生人私聊 false + // (与真实 SessionManager 由 actor 派生的口径一致;它决定最小打扰分级的分支)。 + sessionOwner bool + // wantParallel 为 true 表示这两条属于**不同** session,必须能同时跑。 + wantParallel bool + }{ + { + name: "kook: 同 guild 的两个 text channel 是同一个群会话", + first: kookGroupMsg("c1", "g1", "a"), + second: kookGroupMsg("c2", "g1", "b"), + firstAdapter: "kook:conf0", + secondAdapter: "kook:conf0", + sessionOwner: true, + }, + { + name: "telegram: 同平台两个账号对同一对端是同一个会话", + first: strangerMsg("200", "a"), + second: strangerMsg("200", "b"), + firstAdapter: "telegram:conf0", + secondAdapter: "telegram:conf1", + sessionOwner: false, + }, + { + name: "控制组: 不同群仍是不同会话(必须能并行)", + first: kookGroupMsg("c1", "g1", "a"), + second: kookGroupMsg("c2", "g2", "b"), + firstAdapter: "kook:conf0", + secondAdapter: "kook:conf0", + sessionOwner: true, + wantParallel: true, + }, + } + + for _, c := range cases { + c := c + t.Run(c.name, func(t *testing.T) { + t.Parallel() + gate := make(chan struct{}) + chat := &laneChatter{} + chat.onRun = func(string) { <-gate } // 每个回合都卡住,便于观察「谁在跑」 + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{owner: c.sessionOwner}, + Owner: testOwnerProfile("kook", "100"), + }, nil) + first := &laneAdapter{name: c.firstAdapter} + second := &laneAdapter{name: c.secondAdapter} + + b.onMessage(context.Background(), c.first, first) + chat.waitRuns(t, 1, 5*time.Second) // 第一条已在跑(并卡住) + b.onMessage(context.Background(), c.second, second) + + if c.wantParallel { + chat.waitRuns(t, 2, 3*time.Second) // 不同会话:第二条立刻开跑 + } else { + // 同一会话:第二条必须还在排队 —— 给它足够的时间「如果会跑就跑起来」。 + assertRunCountStays(t, chat, 1, 200*time.Millisecond) + } + close(gate) + chat.waitRuns(t, 2, 5*time.Second) + }) + } +} + +// assertRunCountStays 在窗口期内反复确认回合数没有超过 want(用于断言「第二条没跑起来」)。 +func assertRunCountStays(t *testing.T, c *laneChatter, want int, window time.Duration) { + t.Helper() + deadline := time.Now().Add(window) + for time.Now().Before(deadline) { + if n := c.runCount(); n > want { + t.Fatalf("turns started = %d, want at most %d: the second message must wait in the same lane", n, want) + } + time.Sleep(time.Millisecond) + } +} + +// TestLaneOwnerPrivateSpillKeepsArrivalOrder 断言不变量 ①对**溢出区**同样成立: +// 主人私聊车道满了以后,新到的条目必须排在溢出区**末尾**,绝不能因为 `ch` 腾出一个位置 +// 就超车到已在排队的条目前面。 +// +// 复审实测的旧行为是 `[m1 m2 m3 m5 m4]`(到达顺序 m1..m5):m4 落进 spill 后,m1 结束让 +// `ch` 空出一格,m5 于是先进 `ch` 而先跑。修法是「spill 非空后一律追加 spill」。 +func TestLaneOwnerPrivateSpillKeepsArrivalOrder(t *testing.T) { + t.Parallel() + + release := make(chan struct{}, 8) + chat := &laneChatter{} + chat.onRun = func(string) { <-release } // 每个回合都由测试逐个放行 + b := newLaneBridge(t, Options{ + Loop: chat, + Sessions: &laneSessions{owner: true}, + Owner: testOwnerProfile("telegram", "100"), + LaneDepth: 2, + }, nil) + ad := &laneAdapter{name: "telegram"} + + b.onMessage(context.Background(), ownerMsg("m1"), ad) + chat.waitRuns(t, 1, 5*time.Second) // m1 在跑(卡住) + b.onMessage(context.Background(), ownerMsg("m2"), ad) + b.onMessage(context.Background(), ownerMsg("m3"), ad) // 深度 2:ch 满 + b.onMessage(context.Background(), ownerMsg("m4"), ad) // 落 spill + + release <- struct{}{} // 放 m1 走 + chat.waitRuns(t, 2, 5*time.Second) // m2 开跑 ⇒ ch 空出一格 + b.onMessage(context.Background(), ownerMsg("m5"), ad) // 关键交错:此刻新条目到达 + + for i := 0; i < 4; i++ { // 逐条放行剩下的 m2..m5 + release <- struct{}{} + } + chat.waitRuns(t, 5, 5*time.Second) + waitLaneIdle(t, b, 2*time.Second) + + got := chat.userTexts() + want := []string{"m1", "m2", "m3", "m4", "m5"} + if len(got) != len(want) { + t.Fatalf("turns = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("turn order = %v, want %v (spill must never be overtaken)", got, want) + } + } + if stats := b.LaneStats(); stats.DroppedOverflow != 0 || stats.DroppedStale != 0 { + t.Fatalf("owner private must never be dropped: %+v", stats) + } +}