Cyrene_v2/internal/services/channelbridge/coalesce_test.go

package channelbridge

// coalesce_test.go 钉住 M23 的**静默窗口合并**（④）本身：滑动等待、到期发车、
// 绝对上限、零丢失/顺序/快照一致、开关回退、单批上限拆分。
//
// 对应 M23 蓝图 §12 的 #1/#2/#3/#4/#5/#6/#7/#8/#15/#16。

import (
	"fmt"
	"strings"
	"testing"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
)

// TestCoalesceWaitsForQuietThenFlushes（#1）断言**滑动**语义：第一条到达后 1.5s 又来一条，
// 计时重置；静默满 2.0s 才发车，且这一批同时带走两条。
//
// 若实现是 fixed 窗口（不重置），那么在 m1 到达 2.0s 时就会先发一次车（只剩 m1）——
// 下面「advance(1500) 之后一轮都不许跑」与「重置后再满 2.0s 才发一次、批内两条」
// 两点同时钉住它。
func TestCoalesceWaitsForQuietThenFlushes(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1"))
	// 时钟走到 m1 之后 1.5s：窗口还没到点，一轮都不该跑（不是 fixed 窗口的 2.0s 边界）。
	f.clock.advance(1500 * time.Millisecond)
	if n := chat.runCount(); n != 0 {
		t.Fatalf("turns started = %d after 1.5s, want 0 (window is 2.0s)", n)
	}
	// 窗口内的第二条：计时必须重置（新定时器 = 从 m2 起再等满一个窗口）。
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m2"))
	f.timers.advanceAndFire(t, f.clock)
	f.waitRuns(t, 1)

	if n := chat.runCount(); n != 1 {
		t.Fatalf("turns started = %d, want exactly 1 (two messages, one turn)", n)
	}
	got := userTextsIn(chat.runAt(0).messages)
	if len(got) != 2 || got[0] != "m1" || got[1] != "m2" {
		t.Fatalf("batch user messages = %v, want [m1 m2]", got)
	}
	if stats := f.b.CoalesceStats(); stats.FlushQuietTotal != 1 || stats.WindowsCreated != 1 {
		t.Fatalf("stats = %+v, want one quiet flush on one window", stats)
	}
}

// TestCoalesceFlushesOnWindowExpiry（#2）断言「只来一条」时窗口到期即发车：
// 到点**之前**一轮都不许跑（不提前），到点之后恰好一轮（不延后）。
func TestCoalesceFlushesOnWindowExpiry(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "only"))
	if n := chat.runCount(); n != 0 {
		t.Fatalf("turns started = %d before the window expired, want 0", n)
	}
	f.timers.advanceAndFire(t, f.clock)
	f.waitRuns(t, 1)

	if n := chat.runCount(); n != 1 {
		t.Fatalf("turns started = %d, want 1", n)
	}
	if got := userTextsIn(chat.runAt(0).messages); len(got) != 1 || got[0] != "only" {
		t.Fatalf("batch user messages = %v, want [only]", got)
	}
	if stats := f.b.CoalesceStats(); stats.FlushMaxWaitTotal != 0 {
		t.Fatalf("flush reason = max_wait, want quiet (stats=%+v)", stats)
	}
}

// TestCoalesceTwelveBurstOneTurn（#3）是塔要的确定性判据：12 条 2s 内到达 ⇒
// **一次 Run**、模型可见 user 行 12 条、落库 12 user + 1 assistant、turn_id 全同。
func TestCoalesceTwelveBurstOneTurn(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	hist := newLaneHistory()
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Sessions: coalesceGroupSessions{},
		History:  hist,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	const n = 12
	for i := 0; i < n; i++ {
		f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%d", i)))
	}
	f.pump(t, func() bool { return chat.runCount() >= 1 })

	if got := chat.runCount(); got != 1 {
		t.Fatalf("turns started = %d, want exactly 1 for a 12-message burst", got)
	}
	if got := userTextsIn(chat.runAt(0).messages); len(got) != n {
		t.Fatalf("batch user messages = %d, want %d: %v", len(got), n, got)
	}
	rows := hist.snapshot("sess--20")
	userRows, assistantRows := 0, 0
	turnIDs := map[string]struct{}{}
	for _, row := range rows {
		switch row.Role {
		case "user":
			userRows++
			turnIDs[row.TurnID] = struct{}{}
		case "assistant":
			assistantRows++
		}
	}
	if userRows != n || assistantRows != 1 {
		t.Fatalf("persisted rows = %d user + %d assistant, want %d + 1", userRows, assistantRows, n)
	}
	if len(turnIDs) != 1 {
		t.Fatalf("distinct user turnIDs = %d, want 1 (one batch = one turn)", len(turnIDs))
	}
	if stats := f.b.CoalesceStats(); stats.CoalescedDroppedTotal != 0 || stats.BatchMessagesTotal != n {
		t.Fatalf("stats = %+v, want %d messages and zero coalesced drops", stats, n)
	}
}

// TestCoalesceNeverDropsInWindow（#4）断言窗口期一条都不丢：20 条连发（超过默认车道深度
// 8，因此也顺带钉住「batch 车道的容量必须够装一整批」），最终 20 条**全部**按到达顺序
// 出现在某一批里，且没有任何合并期丢弃。
func TestCoalesceNeverDropsInWindow(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	const n = 20
	want := make([]string, 0, n)
	for i := 0; i < n; i++ {
		text := fmt.Sprintf("m%02d", i)
		want = append(want, text)
		f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", text))
	}
	f.pump(t, func() bool { return chat.runCount() >= 1 })

	var got []string
	for i := 0; i < chat.runCount(); i++ {
		got = append(got, userTextsIn(chat.runAt(i).messages)...)
	}
	if strings.Join(got, ",") != strings.Join(want, ",") {
		t.Fatalf("batch messages = %v, want %v (nothing dropped, order kept)", got, want)
	}
	stats := f.b.CoalesceStats()
	if stats.CoalescedDroppedTotal != 0 {
		t.Fatalf("coalesced_dropped_total = %d, want 0", stats.CoalescedDroppedTotal)
	}
	if lane := f.b.LaneStats(); lane.DroppedOverflow != 0 || lane.DroppedStale != 0 {
		t.Fatalf("lane stats = %+v, want no drops (batch lane must fit a whole batch)", lane)
	}
}

// TestCoalescePreservesArrivalOrder（#5）断言跨发送者的顺序：模型可见消息与落库行的顺序
// 都逐字等于到达顺序（两个群友交替说话），且每条仍是**独立的 user 消息**（不拼成一条）。
func TestCoalescePreservesArrivalOrder(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	hist := newLaneHistory()
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Sessions: coalesceGroupSessions{},
		History:  hist,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	steps := []struct{ sender, name, text string }{
		{"9", "张三", "m1"}, {"10", "李四", "m2"}, {"9", "张三", "m3"},
		{"10", "李四", "m4"}, {"9", "张三", "m5"}, {"10", "李四", "m6"},
	}
	for _, s := range steps {
		f.send(coalesceGroupMsg("telegram", "-20", s.sender, s.name, s.text))
	}
	f.pump(t, func() bool { return chat.runCount() >= 1 })

	if got := chat.runCount(); got != 1 {
		t.Fatalf("turns started = %d, want 1", got)
	}
	msgs := chat.runAt(0).messages
	var users []model.Message
	for _, m := range msgs {
		if m.Role == model.RoleUser {
			users = append(users, m)
		}
	}
	if len(users) != len(steps) {
		t.Fatalf("user messages = %d, want %d (one row per message, never concatenated)", len(users), len(steps))
	}
	for i, s := range steps {
		if lastText(users[i]) != s.text {
			t.Fatalf("user[%d] = %q, want %q (arrival order)", i, lastText(users[i]), s.text)
		}
		if users[i].SenderName != s.name {
			t.Fatalf("user[%d] SenderName = %q, want %q", i, users[i].SenderName, s.name)
		}
	}
	var persisted []string
	for _, row := range hist.snapshot("sess--20") {
		if row.Role == "user" {
			persisted = append(persisted, row.Content)
		}
	}
	if strings.Join(persisted, ",") != "m1,m2,m3,m4,m5,m6" {
		t.Fatalf("persisted user rows = %v, want arrival order", persisted)
	}
}

// TestCoalesceBatchesAreConsistentSnapshots（#6）断言批与批之间互不污染：
// 同一批里的消息**不会**出现在这一批自己的转录里（各自恰好一次），而前一批的行会作为
// 历史进入下一批 —— 这正是「播种在读之前、落库在后」的既有顺序。
func TestCoalesceBatchesAreConsistentSnapshots(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	hist := newLaneHistory()
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Sessions: coalesceGroupSessions{},
		History:  hist,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "a1"))
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "a2"))
	f.pump(t, func() bool { return chat.runCount() >= 1 })
	f.waitIdle(t)
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "b1"))
	f.pump(t, func() bool { return chat.runCount() >= 2 })

	first := userTextsIn(chat.runAt(0).messages)
	if strings.Join(first, ",") != "a1,a2" {
		t.Fatalf("first batch user messages = %v, want [a1 a2] exactly once each", first)
	}
	second := userTextsIn(chat.runAt(1).messages)
	if strings.Join(second, ",") != "a1,a2,b1" {
		t.Fatalf("second batch user messages = %v, want [a1 a2 b1] (previous batch as transcript, own rows once)", second)
	}
	if joined := joinTexts(chat.runAt(1).messages); !strings.Contains(joined, "reply:a2") {
		t.Fatalf("second batch is missing the previous assistant reply:\n%s", joined)
	}
}

// TestCoalesceBatchUsesLatestContext（#7）断言最新性：续跑那一批的上下文里含上一轮刚落库的
// 助手回复（她接得上自己刚才说的话）。
func TestCoalesceBatchUsesLatestContext(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	hist := newLaneHistory()
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Sessions: coalesceGroupSessions{},
		History:  hist,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "first"))
	f.pump(t, func() bool { return chat.runCount() >= 1 })
	f.waitIdle(t)
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "second"))
	f.pump(t, func() bool { return chat.runCount() >= 2 })

	if joined := joinTexts(chat.runAt(1).messages); !strings.Contains(joined, "reply:first") {
		t.Fatalf("continuation context does not contain the previous reply:\n%s", joined)
	}
}

// TestCoalesceMaxWaitForcesFlush（#8）断言绝对上限：每 500ms 来一条、连续 6s，
// 到 6s 必发车（滑动计时被一条条重置也压不住），绝不无限等下去。
func TestCoalesceMaxWaitForcesFlush(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1"))
	for i := 1; i < 12; i++ {
		f.clock.advance(500 * time.Millisecond)
		f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%d", i+1)))
	}
	// 最后那个定时器请求的正是「距离 6s 绝对上限还差多少」——按它推完最后 500ms，
	// 循环头因此看见 maxLeft<=0 并按 max_wait 发车（滑动窗口被一条条重置也压不住它）。
	f.timers.advanceAndFire(t, f.clock)
	f.waitRuns(t, 1)
	if n := chat.runCount(); n != 1 {
		t.Fatalf("turns started = %d, want 1 (max wait forces one flush)", n)
	}
	if got := userTextsIn(chat.runAt(0).messages); len(got) != 12 {
		t.Fatalf("batch user messages = %d, want 12", len(got))
	}
	if stats := f.b.CoalesceStats(); stats.FlushMaxWaitTotal != 1 {
		t.Fatalf("stats = %+v, want flush_reason=max_wait", stats)
	}
}

// TestCoalesceDisabledMatchesLaneBehavior（#15）断言开关回退：`enabled:false` ⇒ 12 条
// 就是 12 轮（到达即投递、一条一轮），**一个定时器都不建**，落库 12 user + 12 assistant。
//
// LaneDepth 取 16 是刻意的：关窗之后车道深度回到 M22 口径（默认 8），而本用例要断言
// 「逐条跑」而不是「背压丢了几条」，因此给它一条足够深的车道。
func TestCoalesceDisabledMatchesLaneBehavior(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	hist := newLaneHistory()
	f := newCoalesceBridge(t, Options{
		Loop:      chat,
		Sessions:  coalesceGroupSessions{},
		History:   hist,
		LaneDepth: 16,
		Coalesce:  CoalesceOptions{For: coalesceOffFor},
	})
	const n = 12
	for i := 0; i < n; i++ {
		f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%d", i)))
	}
	f.waitRuns(t, n)
	f.waitIdle(t)

	if got := chat.runCount(); got != n {
		t.Fatalf("turns started = %d, want %d (feature off = one turn per message)", got, n)
	}
	if calls := f.timers.callCount(); calls != 0 {
		t.Fatalf("coalesce timers created = %d, want 0 when the feature is off", calls)
	}
	stats := f.b.CoalesceStats()
	if stats.WindowsCreated != 0 || stats.BatchesTotal != n {
		t.Fatalf("stats = %+v, want 0 windows and %d batches", stats, n)
	}
	userRows, assistantRows := 0, 0
	for _, row := range hist.snapshot("sess--20") {
		switch row.Role {
		case "user":
			userRows++
		case "assistant":
			assistantRows++
		}
	}
	if userRows != n || assistantRows != n {
		t.Fatalf("persisted rows = %d user + %d assistant, want %d + %d", userRows, assistantRows, n, n)
	}
}

// TestCoalesceBatchOverflowSplitsNotDrops（#16）断言单批上限：30 条（> 24）入队 ⇒
// 两批 24 + 6，顺序不变、零丢弃、被拆出去的尾部**排在最前**（成为下一批）。
func TestCoalesceBatchOverflowSplitsNotDrops(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:      chat,
		LaneDepth: 40, // 让 30 条都排得下，从而让「拆分」这条分支可达
		Coalesce:  CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	const n = 30
	for i := 0; i < n; i++ {
		f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("m%02d", i)))
	}
	f.pump(t, func() bool { return chat.runCount() >= 1 })
	// 被回插队尾的那 6 条**已经等过了**窗口（时钟在第一批到点时已走到窗口末尾），
	// 因此它们下一批立刻发车，不需要再推进一个窗口。
	f.waitRuns(t, 2)

	if got := chat.runCount(); got != 2 {
		t.Fatalf("turns started = %d, want 2 (24 + 6)", got)
	}
	first := userTextsIn(chat.runAt(0).messages)
	second := userTextsIn(chat.runAt(1).messages)
	if len(first) != 24 || len(second) != 6 {
		t.Fatalf("batch sizes = %d + %d, want 24 + 6", len(first), len(second))
	}
	var all []string
	all = append(all, first...)
	all = append(all, second...)
	for i, text := range all {
		if text != fmt.Sprintf("m%02d", i) {
			t.Fatalf("message[%d] = %q, want m%02d (order must survive the split)", i, text, i)
		}
	}
	if stats := f.b.CoalesceStats(); stats.BatchesSplitTotal != 1 || stats.CoalescedDroppedTotal != 0 {
		t.Fatalf("stats = %+v, want exactly one split and zero coalesced drops", stats)
	}
}

// joinTexts 把一批模型可见消息的文本拼起来（断言「上下文里含有什么」用）。
func joinTexts(msgs []model.Message) string {
	var out []string
	for _, m := range msgs {
		for _, c := range m.Content {
			if c.Type == model.ContentTypeText {
				out = append(out, c.Text)
			}
		}
	}
	return strings.Join(out, "\n")
}