Cyrene_v2/internal/services/channelbridge/lane.go

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 恒为空的车道因此不占额外内存。
//
// 车道分**两种类型**（M23 的红线级结构约束，见下面两个构造函数）：
//   - **direct 车道**（主人私聊）：到达即投递、一条一条跑。它**没有** batcher 字段值，
//     因此拿不到定时器工厂 —— 「主人私聊没有静默窗口」是**类型上不可达**，
//     不是配置判出来的（配置无权翻转，见 newDirectLane）。
//   - **batch 车道**（群聊 / 陌生人私聊）：持有一个 batchWindow，由窗口引擎决定何时发车。
type lane struct {
	key       string
	ch        chan laneItem
	neverDrop bool
	spill     []laneItem
	// front 是「回插队首」区：单批超过上限时，尾部**原样保序**地插回这里，
	// 成为下一批（M23 蓝图 §4-3：超量不是丢弃）。front 里的条目一定早于
	// ch 与 spill 里的，因此读取顺序恒为 front → ch → spill。
	front []laneItem
	// batcher 非 nil ⇒ 这是一条 batch 车道（持有静默窗口的定时器工厂）。
	// nil ⇒ direct 车道：窗口/定时器对它是**不可达类型**。
	batcher *batchWindow
	// lastTurnEnd 是该车道**上一轮回合结束**的时刻（只有 worker 读写 ⇒ 无需加锁）。
	// 它只服务 continuations_total：批内首条在上一轮结束前就已到达 ⇒ 这批是⑤续跑。
	lastTurnEnd time.Time
}

// newDirectLane 构造**主人私聊**的 direct 车道：它没有任何窗口状态，也不持有定时器
// 工厂（不传 batchWindow ⇒ 拿不到）。
//
// 🔴 车道类型由 **actor 元数据**（`IsOwner() && !IsGroupChat()`）决定，配置无权翻转：
// 唯一能选 direct 的地方是 enqueueInbound 的 ownerPrivateLane 判定。主人私聊是
// 「她和我说话」，2s 迟滞在私聊里没有任何收益（只有被拖慢的观感），
// 且 M22 已裁定主人私聊永不丢弃、不受陈旧度闸约束 —— 本函数是同一取向的延续。
func newDirectLane(key string, ch chan laneItem, neverDrop bool) *lane {
	return &lane{key: key, ch: ch, neverDrop: neverDrop}
}

// newBatchLane 构造**群聊 / 陌生人私聊**的 batch 车道：窗口引擎只在这里可达。
func newBatchLane(key string, ch chan laneItem, neverDrop bool, batcher *batchWindow) *lane {
	return &lane{key: key, ch: ch, neverDrop: neverDrop, batcher: batcher}
}

// laneSet 是全部活跃车道的注册表。车道在**空闲时退休**（worker 见队列空即把自己摘掉），
// 因此表的大小只与「此刻有排队/在跑的会话数」相关。
type laneSet struct {
	depth      int
	staleAfter time.Duration
	log        *slog.Logger
	// after 是静默窗口的定时器工厂（M23）：**只**在构造 batch 车道时被塞进
	// batchWindow，direct 车道拿不到它（见 newDirectLane）。
	after func(time.Duration) <-chan time.Time

	mu              sync.Mutex
	lanes           map[string]*lane
	droppedStale    int64
	droppedOverflow int64
}

// laneSpec 描述「这条车道该长什么样」。类型（direct / batch）由**actor 元数据**决定，
// 调用方（enqueueInbound）算好传进来；配置只能影响窗口参数，改不了类型。
type laneSpec struct {
	// neverDrop ⇒ 主人私聊：队列满时不拒收，落 spill 只排队。
	neverDrop bool
	// batch ⇒ 群聊/陌生人私聊：持有静默窗口（direct 车道不持有）。
	batch bool
	// capacity 是队列深度：batch 车道至少能装下一整批（否则「12 条成 1 轮」在
	// 结构上不可达 —— 第 9 条就会在门前被拒收）。
	capacity int
}

// newLaneSet 构造车道注册表；depth / staleAfter 为 0 时取上面的默认值，
// after 为 nil 时取 time.After。
func newLaneSet(depth int, staleAfter time.Duration, after func(time.Duration) <-chan time.Time, log *slog.Logger) *laneSet {
	if depth <= 0 {
		depth = defaultLaneDepth
	}
	if staleAfter <= 0 {
		staleAfter = defaultLaneStaleAfter
	}
	if after == nil {
		after = time.After
	}
	if log == nil {
		log = logger.New(logger.Options{})
	}
	return &laneSet{
		depth:      depth,
		staleAfter: staleAfter,
		log:        logger.WithSource(log, laneLogSource),
		after:      after,
		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 一起带进来）。车道头部只做「等（窗口）+ 发车」，准入判断不在这里。
//
// 三件事在这里一次性做完：
//   - 认领车道（没有就建一条并起 worker）—— **类型由 actor 元数据决定**：
//     主人私聊 ⇒ direct（无窗口），其余 ⇒ batch（有窗口）；
//   - 入队（群聊/陌生人满则**在门前拒收这一条**并留痕；主人私聊**永不拒收**）；
//   - 与 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)
	spec := b.laneSpecFor(key, item, neverDrop)

	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, spec)
	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")
	}
}

// laneSpecFor 决定一条车道的形状：**类型只由 actor 元数据决定**
// （`IsOwner() && !IsGroupChat()` ⇒ direct），窗口参数只影响容量。
//
// 容量：batch 车道至少要能装下一整批，否则「一批 N 条」在结构上不可达
// （队列深度 8 时第 9 条就会在门前被拒收，合并就成了摆设）。因此
// capacity = max(默认深度, 生效的单批上限)；direct 车道用默认深度。
// 内存口径因此是「活跃会话数 × 批上限」，而批上限本身被 clamp 在 [1,50]。
func (b *Bridge) laneSpecFor(key string, item laneItem, neverDrop bool) laneSpec {
	spec := laneSpec{neverDrop: neverDrop, batch: !neverDrop, capacity: b.lanes.depth}
	if neverDrop {
		return spec
	}
	if w, ok := b.coalesceWindowFor(key, laneStrangerPrivate(item)); ok && w.maxBatch > spec.capacity {
		spec.capacity = w.maxBatch
	}
	return spec
}

// offer 在**一个临界区**内完成「查/建车道 + 入队」，使 `retireLaneIfIdle` 不可能夹在
// 中间把刚查到、还没写入的车道摘掉（那样条目会落进一条没有 worker 的死车道）。
//
// 队列满时的策略分两档（都不是静默的）：
//   - **主人私聊**（neverDrop）：绝不拒收，落到溢出区 `spill` 里**只排队**，并按倍率
//     （ownerLaneBacklogWarn 及其翻倍）留 WARN —— 一个白名单账号的积压只反映「她落后了」，
//     不构成第三方洪泛面；
//   - **群聊 / 陌生人**：**在门前拒收这一条**（而不是挤掉队列里已有的某条）。队列里
//     的条目因此一条都不会凭空消失 —— 窗口的 takeAll 依赖这一点；被拒收的那条
//     由调用方留痕（WARN + 计数 + 审计）。
//
// 🔴 **FIFO 的关键**：`spill` 一旦非空，后面的条目**一律继续追加到 spill**，不再回写
// `ch`。否则「ch 腾出一个位置 + 此时正好有新条目到达」会让后到者超车（复审实测顺序
// `[m1 m2 m3 m5 m4]`）。`nextLaneItem` 因此可以安全地「先取完 ch、再按序取 spill」：
// spill 里的一定晚于 ch 里的，而 spill 一旦开始就用到底（直到它被取空）。
//
// 同理，`front`（回插队首区）里的一定**早于** ch 与 spill 里的（只有 worker 自己
// 在发车前把超量的尾部放进去），所以读取顺序恒为 front → ch → spill。
func (s *laneSet) offer(key string, item laneItem, spec laneSpec) (l *lane, start bool, rejected bool) {
	s.mu.Lock()
	defer s.mu.Unlock()
	l = s.lanes[key]
	if l == nil {
		capacity := spec.capacity
		if capacity <= 0 {
			capacity = s.depth
		}
		ch := make(chan laneItem, capacity)
		if spec.batch {
			l = newBatchLane(key, ch, spec.neverDrop, &batchWindow{after: s.after})
		} else {
			l = newDirectLane(key, ch, spec.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 取下一条：先取回插队首区（front），再把缓冲区取完，最后取
// 主人私聊的溢出区（三者的先后由 offer / requeueFront 的写入口径保证）。
// ok=false 表示此刻真的没有待办。
func (b *Bridge) nextLaneItem(l *lane) (laneItem, bool) {
	b.lanes.mu.Lock()
	if len(l.front) > 0 {
		item := l.front[0]
		l.front = l.front[1:]
		b.lanes.mu.Unlock()
		return item, true
	}
	b.lanes.mu.Unlock()
	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
}

// takeAll 把车道上**此刻**排着的条目整片取出（顺序 = 到达顺序），供静默窗口的
// 批组装使用。它是合并阶段**唯一**的取队列入口（M23 蓝图 §4-1）：
//
//   - 只有车道的单消费者（该车道唯一的 worker）调用它 ⇒ 不存在两个消费者各抓一半；
//   - 取出即「这批就是这些」，等待期间新到的继续落进队列，下一轮 takeAll 再取；
//   - 它**不**做任何筛选：不丢、不重排、不改内容（窗口内零丢失的结构保证）。
func (b *Bridge) takeAll(l *lane) []laneItem {
	b.lanes.mu.Lock()
	out := l.front
	l.front = nil
	b.lanes.mu.Unlock()
	for {
		select {
		case item := <-l.ch:
			out = append(out, item)
			continue
		default:
		}
		b.lanes.mu.Lock()
		if len(l.spill) > 0 {
			out = append(out, l.spill...)
			l.spill = nil
		}
		b.lanes.mu.Unlock()
		return out
	}
}

// requeueFront 把一批**尚未发车**的条目按原顺序插回队首，成为下一批。
//
// 它是「单批上限不是丢弃」的落点（M23 蓝图 §4-3）：超量的**尾部**原样回插、保序、
// 在锁内、在发车之前。回插到 front 而不是写 ch：ch 有界且写满即拒收，
// 而 front 无界但只由 worker 在发车前写入，写入量与「刚拆出来的尾部」相等。
func (b *Bridge) requeueFront(l *lane, items []laneItem) {
	if len(items) == 0 {
		return
	}
	b.lanes.mu.Lock()
	merged := make([]laneItem, 0, len(items)+len(l.front))
	merged = append(merged, items...)
	merged = append(merged, l.front...)
	l.front = merged
	b.lanes.mu.Unlock()
}

// laneWorker 是某会话车道的唯一执行者：串行跑完每一条，队列空了就把车道交还注册表
// （空闲车道不占内存，也不会把「这个人再说话」变慢）。
//
// **回合中续跑（⑤）就落在这里**：不存在任何 "turn in flight" 标志 —— 「回合在跑」
// 在结构上就等于「worker 正忙」，跑回合期间新消息落进同一个队列，回合结束回到循环头，
// takeAll 自然把这段期间攒下的消息整片取出 ⇒ **一轮续跑**（不是 N 轮，也不是 0 条）。
func (b *Bridge) laneWorker(ctx context.Context, l *lane) {
	defer b.wg.Done()
	for {
		// 桥已停：队列里还没发车的条目**如实登记**（绝不静默消失），然后退休。
		if ctx.Err() != nil {
			b.abandonLaneAtStop(l)
			b.retireLane(l)
			return
		}
		if item, ok := b.nextLaneItem(l); ok {
			b.dispatchLaneItem(ctx, l, item)
			continue
		}
		// 队列空了才谈退休：摘表与入队用同一把锁，因此「摘下」与「刚有人塞进来」不会
		// 交错成条目丢失（见 retireLaneIfIdle 的说明）。
		if b.retireLaneIfIdle(l) {
			return
		}
		select {
		case <-ctx.Done():
			// 桥停了：把车道摘掉，未处理的条目随进程取消一起作废（Stop 的语义）。
			b.retireLane(l)
			return
		case item := <-l.ch:
			b.dispatchLaneItem(ctx, l, item)
		}
	}
}

// dispatchLaneItem 按**车道类型**分派：direct（主人私聊）一条一条跑；batch 走窗口
// 引擎（见 coalesce_lane.go 的 runBatchLaneItem）。
//
// 分派依据是 `l.batcher == nil` —— 一个**结构**判据，不是配置开关：direct 车道的
// 类型里根本没有 batcher 字段值，窗口代码对它不可达。
func (b *Bridge) dispatchLaneItem(ctx context.Context, l *lane, item laneItem) {
	if l.batcher == nil {
		b.runDirectLaneItem(ctx, l, item)
		return
	}
	b.runBatchLaneItem(ctx, l, item)
}

// runDirectLaneItem 跑 direct 车道（= 主人私聊）的一条：先取全局回合槽（**在线程池里
// 等，不在读循环里等**），再过陈旧度闸，然后走逐条闸门 + 单条回合。**没有任何等待窗口**。
func (b *Bridge) runDirectLaneItem(ctx context.Context, l *lane, item laneItem) {
	if !b.acquireTurn(ctx) {
		return
	}
	defer b.releaseTurn()

	waited := b.now().Sub(item.arrivedAt)
	if b.staleInbound(item, waited) {
		return
	}
	b.logLaneStart(item, waited, 1)
	p, ok := b.prepareInbound(ctx, item)
	if !ok {
		l.lastTurnEnd = b.now()
		return
	}
	b.runBatch(ctx, []preparedInbound{p})
	l.lastTurnEnd = b.now()
}

// acquireTurn 取一个全局回合槽（maxConcurrent 个）。ctx 取消时返回 false：桥停了
// 就不再开新回合（carried 条目由 Stop 的诚实登记负责，见 coalesce_lane.go）。
func (b *Bridge) acquireTurn(ctx context.Context) bool {
	select {
	case b.sem <- struct{}{}:
		return true
	case <-ctx.Done():
		return false
	}
}

// releaseTurn 归还一个全局回合槽。
func (b *Bridge) releaseTurn() { <-b.sem }

// 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` / `requeueFront` 用同一把锁：与 offer 交错的话会出现「offer 拿到
// 车道 → worker 退休并删除 → offer 往一条没有 worker 的车道入队」的丢失。入队与
// 「查空」在同一把锁下，就不存在这个窗口。
func (b *Bridge) retireLaneIfIdle(l *lane) bool {
	b.lanes.mu.Lock()
	defer b.lanes.mu.Unlock()
	if len(l.front) > 0 || len(l.ch) > 0 || len(l.spill) > 0 || b.lanes.lanes[l.key] != l {
		return false
	}
	delete(b.lanes.lanes, l.key)
	return true
}

// abandonLaneAtStop 登记「桥停止时车道里还排着、因此永远不会被回答的条目」。
//
// 它与 coalesce_lane.go 的 abandonAtStop 分工明确：那边管**已经取在手上、正在等窗口**
// 的那一批，这里管**还躺在队列里**的（front/ch/spill）。两处都带 source 落 WARN 并计入
// abandoned_at_stop_total —— 进程要停了没法「晚回」，唯一能做对的是**说出来**。
func (b *Bridge) abandonLaneAtStop(l *lane) {
	b.lanes.mu.Lock()
	remaining := len(l.front) + len(l.ch) + len(l.spill)
	b.lanes.mu.Unlock()
	if remaining == 0 {
		return
	}
	total := b.coalesceStats.abandonedAtStop.Add(int64(remaining))
	b.lanes.log.Warn("lane queue abandoned at bridge stop",
		"lane", l.key, "count", remaining, "abandoned_at_stop_total", total)
}

// logLaneStart 记「这一批等了多久才开始跑」—— 「她为什么回得晚」在生产日志里最直接
// 的那个答案。与它同时出现在日志里的 `msg="inbound frame"`（适配器的入站帧 DEBUG）
// 则说明读循环在等的同时一直在读：两行合起来就是「回合没有堵住心跳」的证据。
func (b *Bridge) logLaneStart(item laneItem, waited time.Duration, messages int) {
	b.lanes.log.Debug("lane turn started",
		"waited", waited.String(), "messages", messages,
		"platform", item.msg.Platform, "channel", item.msg.ChannelID,
		"sender", item.msg.SenderID, "group", item.msg.GroupID)
}

// laneStrangerPrivate 报告一条入站是否属于**陌生人私聊**（非群 + 非主人）。
//
// 它与 `isStrangerPrivate(sess)` 同源（同一判据的两面）：会话层从 actor 派生
// `sess.Owner`，这里直接用 actor 元数据，因此不需要先查会话就能判定「这条车道
// 要不要受 applyToStrangers 管」。主人私聊走不到这里（那是 direct 车道）。
func laneStrangerPrivate(item laneItem) bool {
	return !item.msg.IsGroup() && !item.actor.IsOwner()
}

// 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:<GroupID>`），按 `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)
}