Cyrene_v2/internal/services/channelbridge/coalesce.go

package channelbridge

// 本文件承载**静默窗口合并**（M23）的窗口引擎：生效参数的解析、等待（滑动窗口 +
// 绝对上限）、发车原因与计数。它回答主人 2026-10-23/24 的原话「要让群聊聊天很自然，
// 很流畅的做法」——群里「一鼓作气打三句」不该让她回三次，也不该让她等很久才开口。
//
// 落点（M23 蓝图 §2 候选 C）：**车道工作者头部**，也就是 `enqueueInbound` 之后、
// 回合之前。理由：车道本身就是那条会话唯一的缓冲，窗口只是「何时把队列里的东西发车」；
// 不新造第二套缓冲，也不会把「本来就会被流控丢掉的消息」拉进窗口。
//
// 三条结构保证（不是靠自觉）：
//
//	① **窗口内零丢失**：消息只存在于车道队列（lane.ch / lane.spill / lane.front），
//	   入队后除了车道的单消费者 takeAll 没有任何移除路径；本文件与 coalesce_*.go
//	   的合并阶段没有「跳过某条」的分支（超量的尾部回插队首，见 requeueFront）。
//	② **一条批 = 一个 turnID = 一次 loop.Run = 一条助手回复**，但批内 N 条仍是
//	   **各自独立的 user 行**（design/15 §2.1 明文禁止把多条拼成一条）。
//	③ **主人私聊没有窗口**：那条车道是 direct 车道（`lane.batcher == nil`），
//	   连定时器工厂都拿不到（见 lane.go 的 newDirectLane / newBatchLane），
//	   类型由 actor 元数据决定，**配置无权翻转**。
//
// 时间基准：窗口用**到达时刻**（Bridge.now，可注入），**不**用 msg.Timestamp ——
// qq/telegram/kook 是平台秒级时间、weixin 是本机时钟，精度与漂移都不可靠。
// 陈旧度闸仍按 M22 口径用到达时刻判，两者互不掺杂。

import (
	"context"
	"sync/atomic"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter"
	"git.yeij.top/AskaEth/Cyrene_v2/internal/config/runtimeconfig"
)

const (
	// coalesceLogSource 是窗口引擎的子功能日志来源（AGENTS §10：日志必须标来源）。
	// 车道自身（背压/陈旧度）的来源归 M22（services/channelbridge/lane）。
	coalesceLogSource = "services/channelbridge/coalesce"

	// 发车原因（`flush_reason` 的闭集，进日志与计数）：
	//   quiet     —— 静默窗口到点（「她把话说完了」）；
	//   max_wait  —— 绝对上限到点（「说得没完没了也得先回一句」）；
	//   stop      —— 桥停止，未发车的批如实登记丢弃；
	//   all_stale —— 批内逐条都超龄（M22 口径：太旧不补答），整批不发车；
	//   off       —— 该车道/该会话没有窗口（direct 车道，或开关关闭），到达即发车。
	coalesceFlushQuiet    = "quiet"
	coalesceFlushMaxWait  = "max_wait"
	coalesceFlushStop     = "stop"
	coalesceFlushAllStale = "all_stale"
	coalesceFlushOff      = "off"
)

// batchWindow 是**只有 batch 车道**才持有的窗口状态：它持有定时器工厂。
//
// 这是 M23 最硬的一条结构约束（红线级）：direct 车道（主人私聊）的类型里**没有**
// 这个字段，也没有任何能拿到它的路径 ⇒「主人私聊不会建窗口」不是 `if enabled`
// 判出来的结果，而是**类型上不可达**。配置因此无权把它翻成 batch。
type batchWindow struct {
	// after 是定时器工厂（默认 time.After）。测试注入手动触发的实现，使
	// 「2.0s 静默」与「6s 绝对上限」都能被确定性地推进，且能数出调用次数 ——
	// TestOwnerPrivateNeverCreatesWindow / TestOwnerPrivateIgnoresEnabledConfig
	// 断言的正是「这个工厂一次都没被调用」。
	after func(time.Duration) <-chan time.Time
}

// coalesceWindow 是一条车道**本次发车**实际使用的窗口参数（已归一为 Duration）。
type coalesceWindow struct {
	window   time.Duration
	maxWait  time.Duration
	maxBatch int
}

// CoalesceOptions 是静默窗口合并的注入面（M23）。
//
// 时钟走 Options.Now（与陈旧度闸同一条注入时钟：一片里有两个时间源必然漂移）。
type CoalesceOptions struct {
	// For 按**车道键**返回该会话生效的窗口参数；nil 表示未接线（走代码默认）。
	// 每次发车前重读它 ⇒ 面板保存后热生效，不需要重启后端。
	//
	// 这里刻意用 runtimeconfig 的策略**类型**（而不是桥内再定义一份）：默认值与 clamp
	// 只有一处实现（`runtimeconfig.DefaultCoalescePolicy` / `ClampCoalescePolicy`），
	// 面板读到的规则与运行期真正生效的区间因此不可能分叉。桥只依赖这个**值类型**，
	// 不持有 store、不做 I/O（store 由组合根注入）。
	For func(laneKey string) *runtimeconfig.CoalescePolicy
	// After 是定时器工厂（测试缝）。nil ⇒ time.After。
	After func(time.Duration) <-chan time.Time
}

// coalesceCounters 是窗口引擎的进程内计数（原子，读侧无需锁）。
type coalesceCounters struct {
	windowsCreated     atomic.Int64
	batchesTotal       atomic.Int64
	batchMessagesTotal atomic.Int64
	flushQuiet         atomic.Int64
	flushMaxWait       atomic.Int64
	batchesSplit       atomic.Int64
	continuations      atomic.Int64
	coalescedDropped   atomic.Int64
	abandonedAtStop    atomic.Int64
	batchesAllStale    atomic.Int64
}

// CoalesceStats 是窗口引擎的计数快照（M23 蓝图 §8）。它给进程内调用方（测试、
// 将来的运维探针）读同一组数字的窄口，**不新开 HTTP 端点**（那是契约变更，归塔）。
//
// CoalescedDroppedTotal **必须恒为 0**：合并阶段没有任何丢弃点 —— 它存在的意义
// 就是让「有人不小心加了一条 continue」有一个能被断言的出口。
type CoalesceStats struct {
	// WindowsCreated 是「真正等过窗口」的批数（direct 车道与开关关闭时恒不增加）。
	WindowsCreated int64
	// BatchesTotal 是发过车的批数（含单条批）。
	BatchesTotal int64
	// BatchMessagesTotal 是这些批里的入站条数合计。
	BatchMessagesTotal int64
	// FlushQuietTotal / FlushMaxWaitTotal 按发车原因分开计数。
	FlushQuietTotal   int64
	FlushMaxWaitTotal int64
	// BatchesSplitTotal 是因单批上限被拆开、尾部回插队首的次数。
	BatchesSplitTotal int64
	// ContinuationsTotal 是「批内首条在该会话上一轮结束之前就已到达」的批数（⑤）。
	ContinuationsTotal int64
	// CoalescedDroppedTotal 必须恒为 0。
	CoalescedDroppedTotal int64
	// AbandonedAtStopTotal 是桥停止时未发车、随进程一起作废的入站条数（诚实登记）。
	AbandonedAtStopTotal int64
	// BatchesAllStaleTotal 是整批都超龄、因此一条都没跑的批数。
	BatchesAllStaleTotal int64
}

// CoalesceStats 返回窗口引擎计数的快照。
func (b *Bridge) CoalesceStats() CoalesceStats {
	c := b.coalesceStats
	if c == nil {
		return CoalesceStats{}
	}
	return CoalesceStats{
		WindowsCreated:        c.windowsCreated.Load(),
		BatchesTotal:          c.batchesTotal.Load(),
		BatchMessagesTotal:    c.batchMessagesTotal.Load(),
		FlushQuietTotal:       c.flushQuiet.Load(),
		FlushMaxWaitTotal:     c.flushMaxWait.Load(),
		BatchesSplitTotal:     c.batchesSplit.Load(),
		ContinuationsTotal:    c.continuations.Load(),
		CoalescedDroppedTotal: c.coalescedDropped.Load(),
		AbandonedAtStopTotal:  c.abandonedAtStop.Load(),
		BatchesAllStaleTotal:  c.batchesAllStale.Load(),
	}
}

// coalesceWindowFor 解析某条车道**此刻**生效的窗口：代码默认 → 注入的 store
// （面板配置文件）→ 归一/夹紧 → 按会话类型决定「这条车道要不要窗口」。
//
// 返回 ok=false 的三种情形都表示「到达即发车」（行为回到 M22）：
//   - 开关关闭（`enabled:false`，也是「回退到步 1 行为」的唯一旋钮）；
//   - 陌生人私聊且 `applyToStrangers:false`；
//   - 配置算出窗口 0（归一后不会出现，防御性兜底）。
//
// 🔴 它**不**决定车道类型：主人私聊根本不进这里（direct 车道没有 batcher）。
func (b *Bridge) coalesceWindowFor(laneKey string, strangerPrivate bool) (coalesceWindow, bool) {
	p := runtimeconfig.DefaultCoalescePolicy()
	if b.coalesceFor != nil {
		if got := b.coalesceFor(laneKey); got != nil {
			p = *got
		}
	}
	p, _ = runtimeconfig.ClampCoalescePolicy(p)
	if !p.Enabled {
		return coalesceWindow{}, false
	}
	if strangerPrivate && !p.ApplyToStrangers {
		return coalesceWindow{}, false
	}
	windowMs := p.WindowMs
	if strangerPrivate && p.StrangerWindowMs > 0 {
		windowMs = p.StrangerWindowMs
	}
	w := coalesceWindow{
		window:   time.Duration(windowMs) * time.Millisecond,
		maxWait:  time.Duration(p.MaxWaitMs) * time.Millisecond,
		maxBatch: p.MaxBatchMessages,
	}
	if w.window <= 0 || w.maxWait < w.window || w.maxBatch <= 0 {
		return coalesceWindow{}, false
	}
	return w, true
}

// laneWindowAdmits 是**窗口准入的唯一判据**（M23 任务书：「窗口只收 chat/image/voice/file；
// **不收**系统通知 / action 帧 / 我们自己出站回执」）。
//
// 它只读消息自身的**结构化事实** —— ContentType 与「是不是平台帧」（isSystemNoticeMessage
// 同样只按 ContentType 判定），**绝不解析正文**、绝不猜语义。拒绝的两类理由：
//
//   - **平台帧 / 系统通知 / 我们自己的 `message_sent` 回执**（ContentType=action）：延迟
//     2s 只为合并它们毫无收益，更糟的是**一条回执就能把窗口一直续住**（它们和群聊共用
//     同一条车道键）；
//   - 其余非聊天类型：本来就没有「等她把话说完」的语义。
//
// 不进窗口的条目**不是被丢弃**：它们走「到达即发车」的 solo 路径（见 coalesce_lane.go
// 的 runBatchLaneItem / coalesceWait 的 solos）。
func laneWindowAdmits(msg adapter.UnifiedMessage) bool {
	if isSystemNoticeMessage(msg) {
		return false
	}
	switch msg.ContentType {
	case adapter.ContentText, adapter.ContentImage, adapter.ContentVoice, adapter.ContentFile:
		return true
	default:
		return false
	}
}

// splitWindowItems 把一片条目按**窗口准入**分成两半：进窗口的（保持原序）与不进窗口的
// （同样保持原序）。两半都会被处理 —— 这是分类，不是剔除。
func splitWindowItems(items []laneItem) (admitted, others []laneItem) {
	for _, item := range items {
		if laneWindowAdmits(item.msg) {
			admitted = append(admitted, item)
			continue
		}
		others = append(others, item)
	}
	return admitted, others
}

// coalesceWait 在车道头部等一段静默期，把等待期间到达的**可进窗口**的消息并进这一批，
// 并返回（这一批、不进窗口的条目、发车原因）。
//
// 语义（M23 蓝图 §4）：
//   - **滑动**：新消息把计时重置到 w.window（「等她把话说完」）；
//   - **绝对上限**：w.maxWait 锚定在批内**第一条**的到达时刻 —— 一个不停说话的群
//     因此不会让计时永不归零（宁可晚回，不许不回）；
//   - **准入**：只有 `laneWindowAdmits` 为真的条目参与攒批与重置计时；通知 / action 帧 /
//     出站回执只被**收进 solos**（原序），既不改计时也不进批 —— 否则「一条我们自己的
//     回执把窗口续住」就是生产上的真实现场。
//
// 它只做「等」与「收」：不判内容、不丢弃、不改顺序（收进来的顺序就是到达顺序：
// 车道 FIFO ⇒ takeAll 的切片本身有序）。
func (b *Bridge) coalesceWait(ctx context.Context, l *lane, held []laneItem, w coalesceWindow) (items, solos []laneItem, reason string) {
	items = held
	anchor := items[0].arrivedAt
	lastArrival := items[len(items)-1].arrivedAt
	b.coalesceStats.windowsCreated.Add(1)
	for {
		now := b.now()
		// 绝对上限先判：到点必发车（哪怕静默窗口因为不停有新消息而一直没归零）。
		maxLeft := w.maxWait - now.Sub(anchor)
		if maxLeft <= 0 {
			return items, solos, coalesceFlushMaxWait
		}
		quietLeft := w.window - now.Sub(lastArrival)
		if quietLeft <= 0 {
			return items, solos, coalesceFlushQuiet
		}
		wait := quietLeft
		if maxLeft < wait {
			wait = maxLeft
		}
		select {
		case <-l.batcher.after(wait):
			seen, others := splitWindowItems(b.takeAll(l))
			solos = append(solos, others...)
			if len(seen) == 0 {
				// 没有新的**聊天**消息：到点即发车，原因由「哪条上限先到」决定
				// （不进窗口的帧不推迟这个判断）。
				if w.maxWait-b.now().Sub(anchor) <= w.window-b.now().Sub(lastArrival) {
					return items, solos, coalesceFlushMaxWait
				}
				return items, solos, coalesceFlushQuiet
			}
			items = append(items, seen...)
			lastArrival = items[len(items)-1].arrivedAt
		case item := <-l.ch:
			if !laneWindowAdmits(item.msg) {
				// 不进窗口：只登记到 solos，**绝不重置计时**。
				solos = append(solos, item)
				continue
			}
			items = append(items, item)
			seen, others := splitWindowItems(b.takeAll(l))
			items = append(items, seen...)
			solos = append(solos, others...)
			lastArrival = items[len(items)-1].arrivedAt
		case <-ctx.Done():
			return items, solos, coalesceFlushStop
		}
	}
}

// noteFlushReason 把发车原因记进对应的计数（未知原因不计数：宁可少一个数，
// 也不要把它归到一个「看起来像」的桶里）。
func (b *Bridge) noteFlushReason(reason string) {
	switch reason {
	case coalesceFlushQuiet:
		b.coalesceStats.flushQuiet.Add(1)
	case coalesceFlushMaxWait:
		b.coalesceStats.flushMaxWait.Add(1)
	case coalesceFlushAllStale:
		b.coalesceStats.batchesAllStale.Add(1)
	}
}