Cyrene_v2/internal/services/channelbridge/coalesce_continuation_test.go

package channelbridge

// coalesce_continuation_test.go 是 M23 的**回合中续跑**（⑤）与两个边界（陈旧度叠加、
// 停机诚实）的断言集，对应蓝图 §12 的 #11/#12/#13/#14/#17。
//
// ⑤ 的实现要点（蓝图 §5）：**不存在 "turn in flight" 标志** —— 车道的单消费者性质
// 就是那个容器。回合在跑时 worker 不在循环头，新消息落进同一个队列；回合结束回到
// 循环头，takeAll 把这段期间攒下的整片取出 ⇒ **一轮续跑**（不是 N 轮、也不是 0 条）。

import (
	"context"
	"fmt"
	"log/slog"
	"strings"
	"testing"
	"time"
)

// TestCoalesceMultiGroupIsolation（#11）断言作用域：两个群各 3 条交错到达 ⇒ 两批、
// 各自 3 条、互不串窗；同一平台的**两个账号**对同一个群则共用一条车道（与会话键同源，
// 是一个会话而不是两个 —— 见 lane.go 的 laneKey 取舍说明）。
func TestCoalesceMultiGroupIsolation(t *testing.T) {
	t.Parallel()

	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	for i := 0; i < 3; i++ {
		f.send(coalesceGroupMsg("telegram", "-1", "9", "张三", fmt.Sprintf("a%d", i)))
		f.send(coalesceGroupMsg("telegram", "-2", "9", "张三", fmt.Sprintf("b%d", i)))
	}
	f.pump(t, func() bool { return chat.runCount() >= 2 })
	f.waitIdle(t)

	if got := chat.runCount(); got != 2 {
		t.Fatalf("turns started = %d, want 2 (one per group)", got)
	}
	seen := map[string]int{}
	for i := 0; i < chat.runCount(); i++ {
		texts := userTextsIn(chat.runAt(i).messages)
		if len(texts) != 3 {
			t.Fatalf("batch %d has %d messages, want 3 (%v)", i, len(texts), texts)
		}
		prefix := texts[0][:1]
		for _, text := range texts {
			if text[:1] != prefix {
				t.Fatalf("batch %d mixes groups: %v", i, texts)
			}
			seen[prefix]++
		}
	}
	if seen["a"] != 3 || seen["b"] != 3 {
		t.Fatalf("messages per group = %v, want a:3 b:3", seen)
	}

	// 多账号：同一个群的两个账号（同 platform / 同 channel）是**同一个会话** ⇒ 一条车道、
	// 一批；回复走最后一条消息自己的适配器（车道本身不持有账号级状态）。
	other := &laneAdapter{name: "telegram:conf1"}
	f.b.onMessage(context.Background(), coalesceGroupMsg("telegram", "-3", "9", "张三", "c0"), f.ad)
	f.b.onMessage(context.Background(), coalesceGroupMsg("telegram", "-3", "10", "李四", "c1"), other)
	before := chat.runCount()
	f.pump(t, func() bool { return chat.runCount() > before })
	f.waitIdle(t)
	if got := chat.runCount(); got != before+1 {
		t.Fatalf("turns started = %d, want %d (two accounts of one group = one lane, one batch)", got, before+1)
	}
	if got := userTextsIn(chat.runAt(before).messages); len(got) != 2 {
		t.Fatalf("multi-account batch = %v, want both messages in one turn", got)
	}
}

// TestContinuationAfterTurnEnds（#12）断言⑤：回合进行中到达 3 条 ⇒ 回合结束后**一轮**
// 续跑吃掉 3 条（不是 3 轮、不是 0 条）。
func TestContinuationAfterTurnEnds(t *testing.T) {
	t.Parallel()

	gate := make(chan struct{})
	chat := &laneChatter{}
	chat.onRun = func(user string) {
		if user == "first" {
			<-gate
		}
	}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "first"))
	f.advanceAndFireUntil(t)           // 第一批到点发车
	chat.waitRuns(t, 1, 3*time.Second) // 第一轮在跑（卡住）
	for i := 0; i < 3; i++ {
		f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("c%d", i)))
	}
	// 模拟「第一轮花了 100ms」：因此这 3 条的到达时刻**早于**上一轮结束时刻，
	// 它们才是「回合中到达」的续跑（真实时钟下这一步自然成立）。
	f.clock.advance(100 * time.Millisecond)
	close(gate)

	f.pump(t, func() bool { return chat.runCount() >= 2 })
	f.waitIdle(t)

	if got := chat.runCount(); got != 2 {
		t.Fatalf("turns started = %d, want 2 (the first one + exactly one continuation)", got)
	}
	if got := userTextsIn(chat.runAt(1).messages); strings.Join(got, ",") != "c0,c1,c2" {
		t.Fatalf("continuation batch = %v, want [c0 c1 c2]", got)
	}
	if stats := f.b.CoalesceStats(); stats.ContinuationsTotal != 1 {
		t.Fatalf("continuations_total = %d, want 1", stats.ContinuationsTotal)
	}
}

// TestContinuationNoDuplicateAndNoLoss（#13）断言⑤的「不重不丢」：每条消息的落库行恰好
// 1 条、每个 turn_id 恰好 1 条助手回复、`continuations_total==1`。
func TestContinuationNoDuplicateAndNoLoss(t *testing.T) {
	t.Parallel()

	gate := make(chan struct{})
	chat := &laneChatter{}
	chat.onRun = func(user string) {
		if user == "first" {
			<-gate
		}
	}
	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.advanceAndFireUntil(t) // 第一批到点发车
	chat.waitRuns(t, 1, 3*time.Second)
	for i := 0; i < 3; i++ {
		f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", fmt.Sprintf("c%d", i)))
	}
	f.clock.advance(100 * time.Millisecond)
	close(gate)
	f.pump(t, func() bool { return chat.runCount() >= 2 })
	f.waitIdle(t)

	userRows := map[string]int{}
	assistantPerTurn := map[string]int{}
	for _, row := range hist.snapshot("sess--20") {
		switch row.Role {
		case "user":
			userRows[row.Content]++
		case "assistant":
			assistantPerTurn[row.TurnID]++
		}
	}
	for _, text := range []string{"first", "c0", "c1", "c2"} {
		if userRows[text] != 1 {
			t.Fatalf("user row %q persisted %d times, want exactly 1", text, userRows[text])
		}
	}
	if len(userRows) != 4 {
		t.Fatalf("distinct user rows = %d, want 4: %v", len(userRows), userRows)
	}
	for turnID, n := range assistantPerTurn {
		if n != 1 {
			t.Fatalf("turn %s has %d assistant rows, want exactly 1", turnID, n)
		}
	}
	if stats := f.b.CoalesceStats(); stats.ContinuationsTotal != 1 || stats.CoalescedDroppedTotal != 0 {
		t.Fatalf("stats = %+v, want one continuation and zero coalesced drops", stats)
	}
}

// TestCoalesceWithStaleGate（#14）断言与陈旧度闸的叠加：批内 2 条超龄 + 1 条新鲜 ⇒
// 只跑新鲜的 1 条，超龄的逐条留痕；整批都超龄 ⇒ 不发车 + all_stale 计数。
//
// 顺序要求（蓝图 §5）：**先**陈旧度**后**组上下文，且绝不让「批里最新一条新鲜」把超龄
// 的旧条也塞进上下文（M22「太旧不补答」优先）。
func TestCoalesceWithStaleGate(t *testing.T) {
	t.Parallel()

	rec := &laneRecorder{}
	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:       chat,
		LaneDepth:  16,
		OnRejected: rec.record,
		Coalesce:   CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "old1"))
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "old2"))
	// 两条排着的时候时间越过陈旧度阈值，然后来了第三条新鲜的。
	f.clock.advance(defaultLaneStaleAfter + time.Second)
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "fresh"))
	// worker 可能已经把前面两条按「绝对上限已过」自行发车了（全超龄 ⇒ 不发车），
	// 也可能把三条并成一批 —— 两种时序下都要把这一批推到发车，因此按窗口触发（而不是
	// 假定它「自己会跑」，那是调度时序，不是被测语义）。
	f.pump(t, func() bool { return chat.runCount() >= 1 })
	f.waitIdle(t)

	if got := chat.runCount(); got != 1 {
		t.Fatalf("turns started = %d, want 1 (only the fresh message may run)", got)
	}
	if got := userTextsIn(chat.runAt(0).messages); strings.Join(got, ",") != "fresh" {
		t.Fatalf("batch user messages = %v, want [fresh] only", got)
	}
	if stats := f.b.LaneStats(); stats.DroppedStale != 2 {
		t.Fatalf("dropped_stale = %d, want 2 (each stale message traced individually)", stats.DroppedStale)
	}
	if reasons := rec.reasons(); len(reasons) != 2 {
		t.Fatalf("audit reasons = %v, want two lane_stale entries", reasons)
	}
	if stats := f.b.CoalesceStats(); stats.CoalescedDroppedTotal != 0 {
		t.Fatalf("coalesced_dropped_total = %d, want 0 (stale drops are M22's gate, not coalescing)", stats.CoalescedDroppedTotal)
	}

	// 整批都超龄：一条都不发车，但留一条可查的 all_stale 痕迹。
	allStale := newCoalesceBridge(t, Options{
		Loop:     &laneChatter{},
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	allStale.send(coalesceGroupMsg("telegram", "-30", "9", "张三", "gone"))
	// 时钟越过陈旧度阈值：循环头看见「绝对上限已过」⇒ 直接按 max_wait 发车，
	// 逐条判完发现整批都超龄 ⇒ 一条都不跑（这一步根本不建定时器）。
	allStale.clock.advance(defaultLaneStaleAfter + time.Second)
	allStale.waitIdle(t)
	if got := allStale.chat.runCount(); got != 0 {
		t.Fatalf("turns started = %d for an all-stale batch, want 0", got)
	}
	if stats := allStale.b.CoalesceStats(); stats.BatchesAllStaleTotal != 1 {
		t.Fatalf("batches_all_stale = %d, want 1 (stats=%+v)", stats.BatchesAllStaleTotal, stats)
	}
}

// TestCoalesceStopsHonestly（#17）断言停机诚实：桥在窗口内停止时，那一批**不会静默
// 消失** —— 有带 source 的 WARN（count=N）、`abandoned_at_stop_total` 增加。
func TestCoalesceStopsHonestly(t *testing.T) {
	t.Parallel()

	logs := &safeBuffer{}
	chat := &laneChatter{}
	f := newCoalesceBridge(t, Options{
		Loop:     chat,
		Log:      slog.New(slog.NewTextHandler(logs, &slog.HandlerOptions{Level: slog.LevelDebug})),
		Coalesce: CoalesceOptions{For: coalesceFor(coalesceOn(2000, 6000, 24))},
	})
	// 走真实的生命周期：Start 建 b.ctx，车道 worker 与投递都用它，Stop 才取消得到它。
	f.b.Start()
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m1"))
	f.send(coalesceGroupMsg("telegram", "-20", "9", "张三", "m2"))
	f.timers.waitRegistered(t, 3*time.Second) // 确认它真的在等窗口（定时器已注册）

	f.b.Stop(context.Background())

	if got := chat.runCount(); got != 0 {
		t.Fatalf("turns started = %d after stop, want 0", got)
	}
	if stats := f.b.CoalesceStats(); stats.AbandonedAtStopTotal != 2 {
		t.Fatalf("abandoned_at_stop_total = %d, want 2", stats.AbandonedAtStopTotal)
	}
	out := logs.String()
	if !strings.Contains(out, "coalesce batch abandoned at bridge stop") ||
		!strings.Contains(out, "source=services/channelbridge/coalesce") ||
		!strings.Contains(out, "count=2") {
		t.Fatalf("stop left no honest trace in the log:\n%s", out)
	}
}