Cyrene_v2/internal/services/channelbridge/coalesce_batch.go

package channelbridge

// coalesce_batch.go 承载「**一条批 = 一次回合**」的落地：一个 turnID、一次 loop.Run、
// 一条助手回复，但批内 N 条仍是**各自独立的 user 行**（design/15 §2.1 明文禁止把多条
// 合并成一条：open-webui 曾把多条拼成一条，被当 bug 修过）。
//
// 形状（对 M22 的 runInboundTurn 逐字保持，只是把「一条」换成「一对多」）：
//
//	[来源/边界系统行(首条)] …转录… [user₁, user₂, …, userₙ]   ← 模型可见
//	chat_messages:  user×N（同一 turnID、到达顺序）+ assistant×1（同 turnID）
//
// 三条口径：
//  1. **转录播种在落库之前**（与 M22 同序）：因此本批自己的行绝不会同时出现在转录与
//     当次消息里；批内每条都只出现一次（不重）。
//  2. **回复路由取批内最后一条**（`sendReply` 只认 Platform/ChannelID/GroupID/SenderID）：
//     同一车道的消息属于同一个对话，用最新那条的路由字段回话是唯一不会回错群的口径。
//  3. **附属内容各自挂各自的消息**：每条消息的附件/图片/语音只挂它自己那条 user 行，
//     绝不合并成一条附件清单（用户内容只进用户角色消息，项目红线）。

import (
	"context"

	"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"
	voiceref "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/voice"
)

// batchMessage 是一条入站消息在本批里**已装配好、尚未落库**的产物。
type batchMessage struct {
	prepared    *preparedInbound
	senderName  string
	attachments inboundContent
	images      inboundContent
	voiceRef    voiceref.Ref
	document    string
	image       string
	msgs        []model.Message
}

// runBatch 跑一批：**一个 turnID、一次 Run、一条助手回复**。批内 N 条消息各自落一条
// user 行（共享 turnID、顺序 = 到达顺序），助手回复与它们同 turnID。
//
// 它同时是「一条消息」的路径（batch 只有一个成员时逐字等于 M22 的 runInboundTurn）：
// 主人私聊、以及 `process()` 的同步便捷路径都走这里。
func (b *Bridge) runBatch(ctx context.Context, batch []preparedInbound) {
	if len(batch) == 0 {
		return
	}
	last := &batch[len(batch)-1]
	// 陌生人私聊的「同一发送者同时最多 1 个在跑回合」槽：一批 = 一个回合 = 一个槽。
	if release := b.acquireStrangerTurn(last.item.msg, last.sess, last.item.a); release != nil {
		defer release()
	}
	target := inboundTargetFor(last.sess, last.item.msg.Platform)
	turnID := newTurnID()
	runEnv := turnEnv(batch[0].sess, target,
		b.resolvePersona(ctx, batch[0].sess, target, adapterName(batch[0].item.a)))

	parts, msgs := b.buildBatchContext(ctx, batch, runEnv)
	// 转录在落库**之前**读（M22 的既有保证）：本批自己的行因此不会在上下文里出现两次。
	msgs = b.seedTurnTranscript(ctx, target, msgs)

	rows := b.persistBatchUsers(ctx, parts, target, turnID)
	res, err := b.run.Run(loop.WithTurnID(loop.WithSessionEnv(ctx, runEnv), turnID), target.sessionID, msgs)
	if err != nil {
		b.log.Warn("agent run failed", "platform", last.item.msg.Platform,
			"session", target.sessionID, "messages", len(batch), "error", err)
	}
	// 抽取结论（status/chars/truncated/reason）只能在回合跑起来之后才有，这里按
	// 服务端存储名把结论补回刚落库的那一行（design/24 §6.1 的两段式）。
	for _, row := range rows {
		b.persistTurnDocumentStatus(row, res)
	}
	// 出错也交给 deliverReply：回合不能因为「模型那边出了问题」就对用户静默
	// （空回复 / 轮次用尽 / 内部报错都落一条诚实的可见回复；被取消的回合除外）。
	b.deliverReply(ctx, last.item.msg, last.item.a, last.sess, target, res, err, turnID)
}

// runBatchSegments 按 `solo` 把一批切成若干段：solo（系统通知升级成的真会话）**单独
// 一轮**，其余相邻消息合并成一轮。切分保持到达顺序，绝不为了多合并而把通知并进聊天。
//
// 罕见路径：solo 只在「传入前审查判为高价值事件并升级」时出现，通常一批里没有。
func (b *Bridge) runBatchSegments(ctx context.Context, kept []preparedInbound) {
	start := 0
	for i := range kept {
		if !kept[i].solo {
			continue
		}
		if i > start {
			b.runBatch(ctx, kept[start:i])
		}
		b.runBatch(ctx, kept[i:i+1])
		start = i + 1
	}
	if start < len(kept) {
		b.runBatch(ctx, kept[start:])
	}
}

// buildBatchContext 装配本批的模型可见消息：逐条摄取附件/语音/图片、构建用户消息、
// 只保留首条的系统框架行，并把每条消息的附属内容挂到**它自己**那条 user 行上。
func (b *Bridge) buildBatchContext(ctx context.Context, batch []preparedInbound, runEnv loop.SessionEnv) ([]batchMessage, []model.Message) {
	parts := make([]batchMessage, 0, len(batch))
	var msgs []model.Message
	for i := range batch {
		p := &batch[i]
		msg := p.item.msg
		// 入站附件摄取留在**闸门之后**（走到这里意味着这条已被放行）：被黑白名单或
		// 最小打扰拦下的消息绝不在盘上留下文件。
		part := batchMessage{
			prepared:    p,
			senderName:  thirdPartySenderName(msg, p.item.actor),
			attachments: b.ingestInboundDocuments(ctx, msg),
			images:      b.ingestInboundImages(msg),
			voiceRef:    b.ingestInboundVoiceCopy(ctx, msg, p.voiceIn),
		}
		part.msgs = b.buildLoopMessages(ctx, p.text, runEnv, p.item.actor, part.senderName, msg)
		if i > 0 {
			// 系统框架（来源行 + 第三方身份边界）每轮只出现一次：后续消息只贡献自己的
			// user 行，否则上下文里会重复 N 遍同一段边界。
			part.msgs = dropLeadingSystem(part.msgs)
		}
		attachInboundContent(part.msgs, part.attachments, part.images)
		part.document = b.inboundDocumentColumn(msg, part.attachments)
		part.image = inboundImageColumn(part.images)
		parts = append(parts, part)
		msgs = append(msgs, part.msgs...)
	}
	return parts, msgs
}

// dropLeadingSystem 去掉消息列表最前面的系统行（批内第 2..N 条用）。
func dropLeadingSystem(msgs []model.Message) []model.Message {
	i := 0
	for i < len(msgs) && msgs[i].Role == model.RoleSystem {
		i++
	}
	return msgs[i:]
}

// persistBatchUsers 按**到达顺序**逐条落 user 行（共享 turnID），并把语音事实与用户
// 消息同源落库（design/24 §10.2 第 5 步）。返回刚落库的行，供回合结束后补抽取结论。
//
// 🔴 顺序（M44，主人 2026-10-26 指令）：通知 = **先写 `voice` 列、再广播**。此前是
// 「先广播、后写列」，于是这条语音消息的**第一帧**（`message_persisted` /
// `channel_persisted`）里读不到 voice，客户端得靠重连或下一次 REST 拉取才升级 ——
// 而「实时帧一到就出播放器」正是本切片的诉求。换序后帧发出时列已写好。
//
// 之所以不能把写列真正挪到落库**之前**：`persistMessageVoice` 是按 id 的 UPDATE，
// 行还不存在时它必然 ErrMessageNotFound（`voice` 列也没有随 INSERT 写入的通道，
// 见 design/24 §10.5 的两种口径）。因此可行的最小形态就是「INSERT → UPDATE 列 →
// notify」——语句数不变，只把 notify 从 INSERT 内部挪到最后。
func (b *Bridge) persistBatchUsers(ctx context.Context, parts []batchMessage, target inboundTarget, turnID string) []history.Message {
	rows := make([]history.Message, 0, len(parts))
	for i := range parts {
		p := &parts[i]
		msg := p.prepared.item.msg
		a := p.prepared.item.a
		// 非语音消息（p.voiceRef 零值）时这个包装与 notifyPersisted 逐字等价。
		notify := b.persistVoiceFirstNotify(p.voiceRef)
		var row history.Message
		if target.private {
			row = b.persistMainWith(ctx, mainRow(history.RoleUser, p.prepared.text, "", p.document, p.image,
				target.platform, deviceIDFor(msg), adapterName(a), turnID), notify)
		} else {
			row = b.persistIsolatedWith(ctx, isolatedRow(history.RoleUser, p.prepared.text, p.document, p.image,
				deviceIDFor(msg), p.prepared.sess, msg.Platform, p.senderName, adapterName(a), turnID), notify)
		}
		rows = append(rows, row)
	}
	return rows
}