feat(channels): IM 车道生成并写入真实 turn_id（user/assistant 成对 + 同源 WithTurnID；群聊隔离与静默语义不变）

internal/runtime/history/turn_id.go

@@ -22,7 +22,10 @@ import (
 //     SetMessagesTurnID。
 //
 // 空串（列的默认值）意味着「这一行不属于任何已知的回合」：加入该列之前的历史行、
-// 未经编排器的写入路径（IM 镜像）都是这种情形，读起来与以往逐字相同。
+// 以及**本就不属于任何回合**的写入路径（被静默登记的非主人消息、无入站回合的主动
+// 推送）都是这种情形，读起来与以往逐字相同。IM 车道不在其列 —— 它没有编排器，但
+// 渠道桥自己就是那条车道的回合权威，在落库前生成并写入这一列
+// （internal/services/channelbridge/turnid.go）。
 
 // SetMessagesTurnID 把 ids 所指的消息行的 turn_id 落定为 turnID —— 两段式写入的
 // 第二段（起跑 / 被 steer 消费时补写）。

internal/runtime/history/types.go

@@ -63,13 +63,15 @@ type Message struct {
 	// 逐字相同。
 	ClientMsgID string
 	// TurnID 是这条消息所属的那一**路**回合的权威标识（design/15 §2.5.9）：
-	// 编排器生成，与该路全部流帧 / `turn_begin` 帧壳上的 turnId **同值**。因此
-	// 同一路的 user 行与 assistant 行带同一个值 —— 客户端据此精确配对「回复 ↔
-	// 提问」，并在重连 / 回放时从历史行恢复回复侧的分组小标（真并行下同一会话
-	// 有两个交错的回答，只按顺序渲染会配错）。
+	// 由驱动该回合的那一方生成（Web 由编排器、IM 由渠道桥，见
+	// internal/services/channelbridge/turnid.go），与该路全部流帧 / `turn_begin`
+	// 帧壳上的 turnId **同值**。因此同一路的 user 行与 assistant 行带同一个值 ——
+	// 客户端据此精确配对「回复 ↔ 提问」，并在重连 / 回放时从历史行恢复回复侧的
+	// 分组小标（真并行下同一会话有两个交错的回答，只按顺序渲染会配错）。
 	//
 	// 它可选：空串意味着「这一行不属于任何已知的回合」—— 加入该列之前的行、以及
-	// 未经编排器的写入路径（IM 镜像等）都是这种情形。
+	// **本就不属于任何回合**的写入路径（被静默登记的非主人消息、主动推送）都是
+	// 这种情形。
 	//
 	// 两个写入时点（与 Document 同一手法，见 turn_id.go）：① 消息落库时它所属的
 	// 路已经存在（新起的一路、以及助手回复）⇒ 随 INSERT 写入；② 排队 / 被转向

internal/services/channelbridge/channelbridge_test.go

@@ -62,13 +62,15 @@ func (f *fakeAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, erro
 type stubChatter struct {
 	sessionID string
 	messages  []model.Message
+	runCtx    context.Context
 	res       *loop.Result
 	err       error
 }
 
-func (s *stubChatter) Run(_ context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) {
+func (s *stubChatter) Run(ctx context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) {
 	s.sessionID = sessionID
 	s.messages = msgs
+	s.runCtx = ctx
 	return s.res, s.err
 }
 

internal/services/channelbridge/inbound.go

@@ -263,6 +263,13 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage,
 	// （同一条用户消息上两条都能挂）。它没有 ctx —— 这里没有任何 I/O（字节由适配器取回）。
 	images := b.ingestInboundImages(msg)
 
+	// 本回合的**权威 turn id**：IM 车道没有编排器，桥接器自己就是这条车道的回合
+	// 权威（turnid.go）。它在落库**之前**生成，因此同一次入站回合的用户消息、助手
+	// 回复带上的是同一个值，前端按 (sessionId, turnId) 两级分桶时才分得进同一个回合
+	// —— 这正是此前缺口所在：IM 的落库行没有 turn id，任何消费方都无法判定「这条
+	// 回复属于哪个回合」。它同时挂到 loop ctx 上，使循环发出的流帧与历史行同源。
+	turnID := newTurnID()
+
 	// 第三方边界只在受限（群聊/陌生人）上下文中生效：
 	// 记忆只读（MemWritable=false），并携带发送者名，
 	// 使 agent 能看到是谁在说话（design/16 §4.6 §1-2）。
@@ -295,12 +302,12 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage,
 	// 无论有没有文本都要落库：附件/图片就是纯媒体消息的内容（design/24 §4.4）。
 	var userMsg history.Message
 	if target.private {
-		userMsg = b.persistToMain(ctx, history.RoleUser, text, "", document, image, target.platform, deviceIDFor(msg), adapterName(a))
+		userMsg = b.persistToMain(ctx, history.RoleUser, text, "", document, image, target.platform, deviceIDFor(msg), adapterName(a), turnID)
 	} else {
-		userMsg = b.persistIsolated(ctx, history.RoleUser, text, document, image, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a))
+		userMsg = b.persistIsolated(ctx, history.RoleUser, text, document, image, deviceIDFor(msg), sess, msg.Platform, senderName, adapterName(a), turnID)
 	}
 
-	res, err := b.run.Run(loop.WithSessionEnv(ctx, runEnv), target.sessionID, loopMsgs)
+	res, err := b.run.Run(loop.WithTurnID(loop.WithSessionEnv(ctx, runEnv), turnID), target.sessionID, loopMsgs)
 	if err != nil {
 		b.log.Warn("agent run failed", "platform", msg.Platform,
 			"session", target.sessionID, "error", err)
@@ -312,7 +319,7 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage,
 	// 出错也交给 deliverReply：回合不能因为「模型那边出了问题」就对用户静默
 	// （空回复 / 轮次用尽 / 内部报错都落一条诚实的可见回复；被取消的回合除外，
 	// 见 turnreply.go）。
-	b.deliverReply(ctx, msg, a, sess, target, res, err)
+	b.deliverReply(ctx, msg, a, sess, target, res, err, turnID)
 }
 
 // deliverReply 解析本回合的可见回复、持久化并经接收该消息的适配器
@@ -321,8 +328,11 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage,
 // IM 回合因此绝不会「没声了」；被取消的回合刻意不交付。不支持出站文本的适配器
 // （adapter.ErrUnsupported）只按 debug 级跳过，不算失败：回复
 // 已在历史存储中落地。
+//
+// turnID 是这次入站回合的权威标识（见 runInboundTurn）：助手回复与刚落的用户消息
+// 传同一个值，落库后两条行同属一路，客户端据此配对「回复 ↔ 提问」。
 func (b *Bridge) deliverReply(ctx context.Context, msg adapter.UnifiedMessage, a adapter.PlatformAdapter,
-	sess session.Session, target inboundTarget, res *loop.Result, runErr error) {
+	sess session.Session, target inboundTarget, res *loop.Result, runErr error, turnID string) {
 	reply, ok := turnVisibleReply(ctx, res, runErr)
 	if !ok {
 		b.log.Debug("turn reply canceled, nothing delivered",
@@ -339,9 +349,9 @@ func (b *Bridge) deliverReply(ctx context.Context, msg adapter.UnifiedMessage, a
 	// 广播；受限 → 隔离日志 + channel_persisted 广播），随后
 	// 送回渠道。
 	if target.private {
-		b.persistToMain(ctx, history.RoleAssistant, reply, mood, "", "", target.platform, deviceIDFor(msg), adapterName(a))
+		b.persistToMain(ctx, history.RoleAssistant, reply, mood, "", "", target.platform, deviceIDFor(msg), adapterName(a), turnID)
 	} else {
-		b.persistIsolated(ctx, history.RoleAssistant, reply, "", "", "", sess, msg.Platform, "", adapterName(a))
+		b.persistIsolated(ctx, history.RoleAssistant, reply, "", "", "", sess, msg.Platform, "", adapterName(a), turnID)
 	}
 	if err := b.sendReply(ctx, msg, a, reply); err != nil {
 		if errors.Is(err, adapter.ErrUnsupported) {

internal/services/channelbridge/light.go

@@ -238,7 +238,9 @@ func (b *Bridge) mentionsTriggerWord(text string) bool {
 func (b *Bridge) registerSilenced(ctx context.Context, msg adapter.UnifiedMessage, actor adapter.Actor, sess session.Session, a adapter.PlatformAdapter) {
 	// 被静默的消息不摄取附件（摄取在闸门之后），因此这两列恒为空 ——
 	// 绝不写一条指向不存在文件的引用，也绝不写一张没进过上下文的图片。
-	b.persistIsolated(ctx, history.RoleUser, msg.Content, "", "", deviceIDFor(msg), sess, msg.Platform, senderDisplayName(msg, actor), adapterName(a))
+	// turn_id 同样传 ""：这条消息**根本不会被回答**，不存在「它属于哪一路回合」，
+	// 给它编一个标识就是把不属于任何回合的行说成有回合（见 turnid.go 的口径）。
+	b.persistIsolated(ctx, history.RoleUser, msg.Content, "", "", deviceIDFor(msg), sess, msg.Platform, senderDisplayName(msg, actor), adapterName(a), "")
 	b.log.Debug("non-owner message silenced, registered only",
 		"platform", msg.Platform, "channel", msg.ChannelID, "session", sess.SessionID)
 }

internal/services/channelbridge/persist.go

@@ -67,17 +67,21 @@ func (b *Bridge) notifyPersisted(msg history.Message) {
 // image 是这条消息携带的图片（data: URI 或 http(s) URL），写进 DTO 的 `image`
 // 字段（design/16 §4.4）；没有图片时传 ""。它**不是** document 列的一部分：
 // 图片走视觉链路，文档走抽取，两者是两条路（design/24 §4.2 末的明确分野）。
+// turnID 是这条消息所属的那**一路** IM 回合的权威标识（design/15 §2.5.9），由
+// 驱动这次回合的桥接器生成（turnid.go）：同一次入站回合的用户消息与助手回复传
+// **同一个**值，前端才能按 (sessionId, turnId) 两级分桶。不属于任何回合的消息
+// （主动推送等）传 ""，那一列因此留空，行为与加入该列之前逐字相同。
 //
 // 它返回刚落库的完整消息（未装配历史存储、或写库失败时是零值），
 // 使调用方能在回合结束后按 id 把文档抽取结论补进同一列
 // （inbound_documents_persist.go 的两段式）。未装配历史存储时为空操作，
 // 因此桥接器在没有存储的情况下行为不变。
-func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, document, image, platform, deviceID, adapterID string) history.Message {
+func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, document, image, platform, deviceID, adapterID, turnID string) history.Message {
 	if b.history == nil {
 		return history.Message{}
 	}
 	full, err := b.history.AppendFull(ctx, history.Message{
-		SessionID: history.MainSessionID, Role: role, Content: content, Mood: mood, Document: document, Image: image, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat,
+		SessionID: history.MainSessionID, Role: role, Content: content, Mood: mood, Document: document, Image: image, Platform: platform, DeviceID: deviceID, AdapterID: adapterID, Mode: modeChat, TurnID: turnID,
 	})
 	if err != nil {
 		b.log.Warn("persist main session", "err", err, "role", role, "platform", platform, "device_id", deviceID, "adapter", adapterID)
@@ -98,18 +102,19 @@ func (b *Bridge) persistToMain(ctx context.Context, role, content, mood, documen
 // 即可追加。document 是该消息携带的文档引用数组（design/24 §2.4），
 // 与 persistToMain 同一列、同一形状；没有附件时传 ""。image 是该消息携带的图片
 // （data: URI 或 http(s) URL），同样与 persistToMain 同一列（DTO 的 `image` 字段）；
-// 没有图片时传 ""。
+// 没有图片时传 ""。turnID 与 persistToMain 同义：这次入站回合的权威标识，
+// 不属于任何回合（被静默登记的消息）时传 ""。
 //
 // 它返回刚落库的完整消息（未装配历史存储、或写库失败时是零值），
 // 供调用方在回合结束后补回文档抽取结论。未装配历史存储时为空操作，
 // 因此桥接器在没有存储的情况下行为不变。
-func (b *Bridge) persistIsolated(ctx context.Context, role, content, document, image, deviceID string, sess session.Session, platform, senderName, adapterID string) history.Message {
+func (b *Bridge) persistIsolated(ctx context.Context, role, content, document, image, deviceID string, sess session.Session, platform, senderName, adapterID, turnID string) history.Message {
 	if b.history == nil {
 		return history.Message{}
 	}
 	full, err := b.history.AppendFull(ctx, history.Message{
 		SessionID: sess.SessionID, Role: role, Content: content, Document: document, Image: image, Platform: platform,
-		DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, Mode: modeChat,
+		DeviceID: deviceID, SenderName: senderName, AdapterID: adapterID, Mode: modeChat, TurnID: turnID,
 	})
 	if err != nil {
 		b.log.Warn("persist isolated session", "err", err, "role", role,

internal/services/channelbridge/proactive.go

@@ -79,9 +79,11 @@ func (b *Bridge) SendProactive(ctx context.Context, msg adapter.UnifiedMessage)
 	// 主人后续回复就会显得不连贯（该消息「丢失」了历史）。
 	// 它与主人私聊 IM 回复走同一条镜像路径，作用域为 MainSessionID，
 	// 落库的同样是规整后的那一份（Web 与 IM 看到的必须是同一份）。
+	// turn_id 传 ""：主动推送是**她单方面开口**，没有对应的入站回合，也就没有
+	// 「它属于哪一路回合」可归属 —— 空值诚实地表示这一点（见 turnid.go 的口径）。
 	b.persistToMain(ctx, history.RoleAssistant, out.Content, "", "", "", target.platform,
 		fmt.Sprintf("%s:%s:%s", target.platform, target.accountID, target.accountID),
-		target.platform)
+		target.platform, "")
 	return nil
 }
 

internal/services/channelbridge/turnid.go

@@ -0,0 +1,31 @@
+package channelbridge
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"time"
+)
+
+// 本文件承载 IM 车道自己的**回合标识**生成点（design/15 §2.5.9）。
+//
+// IM 与 Web 是两条独立的驱动路径：Web 由 gateway 编排器起路、由它生成权威 turnId
+// （gateway/ws_hub_turn.go 的 newTurnID），而 IM 由渠道桥自己驱动 loop
+// （inbound.go 的 runInboundTurn → b.run.Run）—— 桥接器就是**这条车道的回合权威**，
+// 因此这里生成的是真实归属：一条入站消息 → 一次 run → 一条回复，标识只在这一段
+// 生命周期内有效。绝不拿时间戳或会话 id 冒充 turn id —— 那是伪造归属，比留空更坏
+// （留空诚实地表示「不属于任何已知回合」，冒充则会让两个不同回合被分进同一桶）。
+//
+// 格式与 gateway 那份**逐字一致**（`turn-` + 16 位十六进制），使两条车道产出的
+// turnId 对任何消费方（前端按 (sessionId, turnId) 两级分桶）都是同一种东西。它成为
+// 第二处副本，是因为 gateway 那份未导出、且该包已冻结，无法共享 —— 不是有意另造。
+
+// newTurnID 返回一个新的随机十六进制回合标识，用于把一次 IM 入站回合的落库行
+// （用户消息 + 助手回复）关联到同一路。它消耗 crypto/rand 源，源不可用时回退到
+// 纳秒时间戳，因此总能产出 id —— 空的 turnId 会静默破坏客户端的两级分桶。
+func newTurnID() string {
+	var b [8]byte
+	if _, err := rand.Read(b[:]); err == nil {
+		return "turn-" + hex.EncodeToString(b[:])
+	}
+	return "turn-" + time.Now().UTC().Format("20060102150405.000000000")
+}

internal/services/channelbridge/turnid_wiring_test.go

@@ -0,0 +1,137 @@
+package channelbridge
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"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/runtime/history"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
+	"git.yeij.top/AskaEth/Cyrene_v2/internal/user"
+)
+
+// 本文件断言 IM 入站车道的**回合标识接线**（design/15 §2.5.9 的缺口形状）：
+// IM 没有编排器，落库行过去一律是空 turn_id，于是前端按 (sessionId, turnId)
+// 两级分桶时分不进任何回合，只能按时间平铺 —— 面板看着交错只是症状，数据缺
+// 归属才是缺口。渠道桥自己就是这条车道的回合权威（turnid.go），因此一次入站
+// 回合的两条行（用户消息 + 助手回复）必须带**同一个非空**的 turn id，且它与
+// 挂到 loop ctx 上的那个值同源（流帧与历史行同值）。
+//
+// 主线（主人私聊镜像进主会话）与隔离线（群聊/陌生人留在 sess-xxx）都断言，
+// 同时确认隔离语义没被破坏：受限消息仍绝不镜像进主会话。
+
+// assertTurnIDsPaired 断言两条落库行带的 turn id 非空且同值。
+func assertTurnIDsPaired(t *testing.T, userRow, assistantRow history.Message) {
+	t.Helper()
+	if userRow.TurnID == "" {
+		t.Fatalf("user row turn_id is empty; want a real turn id (record=%+v)", userRow)
+	}
+	if assistantRow.TurnID == "" {
+		t.Fatalf("assistant row turn_id is empty; want the same turn id as the user row")
+	}
+	if userRow.TurnID != assistantRow.TurnID {
+		t.Fatalf("turn ids differ: user=%q assistant=%q; want one turn id per turn",
+			userRow.TurnID, assistantRow.TurnID)
+	}
+	if !strings.HasPrefix(userRow.TurnID, "turn-") {
+		t.Fatalf("turn id = %q, want the gateway-compatible %q prefix", userRow.TurnID, "turn-")
+	}
+}
+
+// TestOwnerPrivateTurnIDPersistedAndPaired 覆盖主人私聊（镜像进 Web 主会话）：
+// 用户消息与助手回复落在主会话，且带同一个非空 turn id。
+func TestOwnerPrivateTurnIDPersistedAndPaired(t *testing.T) {
+	t.Parallel()
+
+	chat := &stubChatter{res: &loop.Result{Text: "微信回复"}}
+	sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}}
+	hist := &captureHistory{}
+	b, err := New(Options{
+		Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "100"), History: hist,
+	})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	b.Add(&fakeAdapter{name: "telegram"})
+
+	b.process(context.Background(), directMessage())
+
+	recs := hist.records()
+	if len(recs) != 2 {
+		t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs))
+	}
+	for _, r := range recs {
+		if r.SessionID != history.MainSessionID {
+			t.Fatalf("record session = %q, want %q (owner private mirrors to main)", r.SessionID, history.MainSessionID)
+		}
+	}
+	assertTurnIDsPaired(t, recs[0], recs[1])
+
+	// 流帧与历史行**同源**：loop ctx 上挂的 turn id 必须就是落库的那个值，
+	// 否则前端按帧分桶与按行分桶会得到两个答案。
+	if got := loop.TurnIDFrom(chat.runCtx); got != recs[0].TurnID {
+		t.Fatalf("loop ctx turn id = %q, want %q (same as persisted rows)", got, recs[0].TurnID)
+	}
+}
+
+// TestIsolatedGroupTurnIDPersistedAndPaired 覆盖受限（群聊）车道：被放行的非主人
+// 群消息仍留在其隔离会话（绝不镜像），同样带上成对的非空 turn id。
+func TestIsolatedGroupTurnIDPersistedAndPaired(t *testing.T) {
+	t.Parallel()
+
+	chat := &stubChatter{res: &loop.Result{Text: "group ack"}}
+	sess := &stubSessions{sess: nonOwnerSession()}
+	hist := &captureHistory{}
+	b, err := New(Options{
+		Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist,
+	})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+
+	// 被点名 ⇒ 结构上被放行、驱动真回合（design/12 §6 策略二）。
+	b.process(context.Background(), nonOwnerGroupMsg("hello all", adapter.Mention{UserID: "9"}))
+
+	recs := hist.records()
+	if len(recs) != 2 {
+		t.Fatalf("appends = %d, want 2 (user + assistant)", len(recs))
+	}
+	for _, r := range recs {
+		if r.SessionID != "sess-g" {
+			t.Fatalf("record session = %q, want sess-g (restricted lanes never mirror to main)", r.SessionID)
+		}
+	}
+	assertTurnIDsPaired(t, recs[0], recs[1])
+}
+
+// TestSilencedNonOwnerStillHasNoTurnID 断言隔离行为不变的那一侧：被静默登记、
+// **根本不会被回答**的非主人消息不存在回合，turn id 必须仍为空 —— 给它编一个
+// 标识就是把不属于任何回合的行说成有回合。它同时确认这类消息仍只落隔离会话。
+func TestSilencedNonOwnerStillHasNoTurnID(t *testing.T) {
+	t.Parallel()
+
+	chat := &stubChatter{res: &loop.Result{Text: "不该回"}}
+	sess := &stubSessions{sess: nonOwnerSession()}
+	hist := &captureHistory{}
+	b, err := New(Options{
+		Loop: chat, Sessions: sess, Owner: &user.OwnerProfile{}, History: hist,
+	})
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+
+	b.process(context.Background(), nonOwnerGroupMsg("hello all"))
+
+	if chat.sessionID != "" {
+		t.Fatalf("loop ran for a silenced message: session=%q", chat.sessionID)
+	}
+	recs := hist.records()
+	if len(recs) != 1 {
+		t.Fatalf("appends = %d, want 1 (registered only, no reply)", len(recs))
+	}
+	if recs[0].SessionID != "sess-g" || recs[0].TurnID != "" {
+		t.Fatalf("registered record = %+v, want isolated sess-g with empty turn_id", recs[0])
+	}
+}