From 9b903a89bc2a25d2b9eb46acc7223bb3f6e121df Mon Sep 17 00:00:00 2001 From: AskaEth Date: Mon, 5 Oct 2026 11:36:18 +0800 Subject: [PATCH] =?UTF-8?q?feat(runtime):=20IM=20=E4=B8=BB=E4=BA=BA?= =?UTF-8?q?=E7=A7=81=E8=81=8A=E8=A1=A5=E5=8E=86=E5=8F=B2=E8=BD=AC=E5=BD=95?= =?UTF-8?q?=EF=BC=88=E4=B8=8E=20Web=20=E4=B8=BB=E4=BC=9A=E8=AF=9D=E5=90=8C?= =?UTF-8?q?=E6=BA=90=EF=BC=9A=E6=8A=BD=E5=85=B1=E4=BA=AB=20transcript=20?= =?UTF-8?q?=E5=8C=85=20+=20OwnerEnvFor=20=E5=8D=95=E5=85=A5=E5=8F=A3?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/runtime/loop/sessionenv.go | 15 ++ internal/runtime/transcript/documents.go | 51 ++++ internal/runtime/transcript/transcript.go | 98 ++++++++ .../runtime/transcript/transcript_test.go | 167 +++++++++++++ internal/services/channelbridge/inbound.go | 13 +- .../services/channelbridge/persona_test.go | 14 +- internal/services/channelbridge/transcript.go | 61 +++++ .../services/channelbridge/transcript_test.go | 222 ++++++++++++++++++ .../gateway/chat_documents_bounds_test.go | 9 +- internal/services/gateway/documents.go | 7 +- internal/services/gateway/persona_env.go | 8 +- .../gateway/session_document_status_test.go | 5 +- internal/services/gateway/session_steer.go | 7 +- .../services/gateway/session_transcript.go | 180 ++------------ 14 files changed, 676 insertions(+), 181 deletions(-) create mode 100644 internal/runtime/transcript/documents.go create mode 100644 internal/runtime/transcript/transcript.go create mode 100644 internal/runtime/transcript/transcript_test.go create mode 100644 internal/services/channelbridge/transcript.go create mode 100644 internal/services/channelbridge/transcript_test.go diff --git a/internal/runtime/loop/sessionenv.go b/internal/runtime/loop/sessionenv.go index 7d1cc48..cb9f300 100644 --- a/internal/runtime/loop/sessionenv.go +++ b/internal/runtime/loop/sessionenv.go @@ -46,6 +46,21 @@ func OwnerEnv() SessionEnv { return SessionEnv{ActorEnv: restrict.ActorOwner, MemWritable: true} } +// OwnerEnvFor 返回绑定了显式角色的主人会话环境。它是**主人车道的唯一构造入口**： +// Web 主会话回合（gateway 的 turnEnvResolver.owner）与 IM 主人私聊回合 +// （channelbridge 的私聊车道）都经它派生自己的 runEnv，两处因此逐字段同源 —— +// SessionEnv 将来新增字段时，两条车道会一起拿到它，而不会各自漂移出「同一个她、 +// 两套环境」。 +// +// personaID 为空表示「继承循环配置的默认角色」（与 OwnerEnv 同义，见 design/22 §4.2）。 +// 它只界定提示词/记忆命名空间/情绪机/思考预算的作用域，**不是身份或权限输入** +// （项目红线：身份与权限在 channel 层的 actor 元数据上）。 +func OwnerEnvFor(personaID string) SessionEnv { + env := OwnerEnv() + env.PersonaID = personaID + return env +} + // defaultSessionEnv 是调用方未挂载显式 env 时使用的 // owner 全权限环境：记忆全读写，即既有行为。 func defaultSessionEnv() SessionEnv { diff --git a/internal/runtime/transcript/documents.go b/internal/runtime/transcript/documents.go new file mode 100644 index 0000000..53c4484 --- /dev/null +++ b/internal/runtime/transcript/documents.go @@ -0,0 +1,51 @@ +package transcript + +import ( + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// DocumentPathPrefix 是文档读取的版本化路径前缀。转录还原出的文档部件必须与网关读侧 +// （`GET /api/v1/documents/{name}`）指向同一个 URL，否则续接的回合会把模型指到一个取不到的 +// 地址。它是这个契约字符串的**唯一字面量**：网关的 `documentFilePrefix` 直接取自这里。 +const DocumentPathPrefix = "/api/v1/documents/" + +// documentParts 把已存 document 列里的**全部**引用还原为该回合的文档部件（design/24 §9.5）。 +// 部件只交给 RoleUser 回合（§2.3-③）：文档在任何别的角色上都会让编码器 fail-loud，而一行 +// 坏数据不该把整个会话卡死。 +// +// 列里是引用数组（一个回合最多 5 个附件），因此这里逐条还原、保持部件顺序 —— 少还原一条， +// 那个附件就会在发起它的回合之后的所有回合里消失，与「清单恒在」的承诺自相矛盾。 +// +// 列被写坏（或附件落在非用户行上）时**不静默**：打一条带来源的 WARN 并省掉该部件，回合照常 +// 继续 —— 附件凭空消失正是这条清单要修的缺陷，因此必须留下痕迹。日志只带消息 id 与错误， +// 绝不带正文。 +func documentParts(role model.Role, h history.Message) []model.Content { + if h.Document == "" { + return nil + } + if role != model.RoleUser { + transcriptLog().Warn("transcript document on non-user row", "message", h.ID, "role", h.Role) + return nil + } + refs, err := history.DecodeDocuments(h.Document) + if err != nil { + transcriptLog().Warn("transcript document decode", "message", h.ID, "err", err) + return nil + } + parts := make([]model.Content, 0, len(refs)) + for _, ref := range refs { + if ref.Name == "" { + transcriptLog().Warn("transcript document without name", "message", h.ID) + continue + } + doc := model.Document{ + Name: ref.Name, + MIME: ref.MIME, + URL: DocumentPathPrefix + ref.Name, + Size: ref.Size, + } + parts = append(parts, model.DocumentMessage(model.RoleUser, "", doc).Content...) + } + return parts +} diff --git a/internal/runtime/transcript/transcript.go b/internal/runtime/transcript/transcript.go new file mode 100644 index 0000000..d0df2fa --- /dev/null +++ b/internal/runtime/transcript/transcript.go @@ -0,0 +1,98 @@ +// Package transcript 拥有「把持久化的聊天转录重建为一个回合的模型上下文」这一件事 +// （design/15 §2.2/§4 的上下文有界、design/02 §1 主会话承载完整上下文、design/22 §7-5）。 +// +// 两条编排路径都要在回合开始时带上该会话最近的若干条已落库消息：网关的会话编排器 +// （internal/services/gateway/session_transcript.go）与渠道桥的 IM 回合 +// （internal/services/channelbridge）。IM 与 Web 只有**呈现形态**不同（IM 只发 chat 文本）， +// 模型看到的对话历史必须是同一份 —— 两边各写一套映射，迟早会漂移成「消息在同一条会话流 +// 里，模型看到的历史却不同」，也就是同一个她记得的事不一样。 +// +// 本包只做纯映射：读窄接口 Recent，把 history.Message 翻成 model.Message。**该不该播种、 +// 播在哪条车道**由调用方按各自的会话与身份边界判定（网关按会话，渠道桥按 private/受限）， +// 因此本包不持有任何编排状态，也不做身份判断（项目红线：身份在 channel 层的 actor 元数据上）。 +package transcript + +import ( + "context" + "log/slog" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// ContextCount 限定一个新回合带入模型上下文的持久化消息数。它与编排器的续跑尾部上限 +// （网关的 defaultMaxTail）一致，使播种回合与续跑回合保持同样大小；超出之后仍由 loop +// 自身的上下文预算治理压缩。**两条车道都从这里取**：这是条数的唯一来源，不许各自另立常量。 +const ContextCount = 20 + +// Reader 是播种所需的窄历史视图。*history.Store 满足它；未满足的存储（测试替身，或 +// 未接持久化构建的调用方）只是得不到播种上下文，因此回合完全照旧继续运行。 +type Reader interface { + Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) +} + +// Read 返回 sessionID 最近的持久化消息映射成的模型消息（最旧优先）。store 为 nil、 +// 未实现 Reader 或读取出错都得到 nil，因此聊天绝不被一次元数据查找阻塞；没有历史的 +// 会话同样什么都不产出 —— 这些情形都退化为既有的「只有本次消息」上下文。 +func Read(ctx context.Context, store Reader, sessionID string) []model.Message { + if store == nil || sessionID == "" { + return nil + } + recent, err := store.Recent(ctx, sessionID, ContextCount) + if err != nil { + transcriptLog().Debug("turn transcript read", "err", err, "session", sessionID) + return nil + } + out := make([]model.Message, 0, len(recent)) + for _, h := range recent { + out = append(out, Message(h)) + } + return out +} + +// Message 把一条持久化聊天消息映射为回合延续所依据的模型消息。助手消息携带说话的角色 +// （该回合已存的 persona_id）作为其说话者标签，因此即使会话被切换过，上下文也能把它归属 +// 到那个角色；非助手消息保持其已存发送者标签（design/16 §4.6 的第三方显示名），绝不 +// 被重新标为说话角色。文档引用**全部**一并还原（design/24 §9.5）：少了它，附件会在发起 +// 它的那一回合之后的所有回合里消失。 +func Message(h history.Message) model.Message { + role := RoleOf(h.Role) + msg := model.Message{ + Role: role, + Content: []model.Content{{Type: model.ContentTypeText, Text: h.Content}}, + } + if h.Image != "" { + msg.Content = append(msg.Content, model.Content{Type: model.ContentTypeImage, ImageURL: h.Image}) + } + msg.Content = append(msg.Content, documentParts(role, h)...) + if role == model.RoleAssistant { + msg.SenderName = h.PersonaID + return msg + } + msg.SenderName = h.SenderName + return msg +} + +// RoleOf 把已存聊天角色映射为模型的角色词汇，未知角色默认归到用户通道。 +func RoleOf(role string) model.Role { + switch role { + case history.RoleAssistant: + return model.RoleAssistant + case history.RoleSystem: + return model.RoleSystem + default: + return model.RoleUser + } +} + +// logSource 是转录重建的日志来源标签。 +const logSource = "runtime/transcript" + +// transcriptLog 返回标注来源的子日志器。转录重建是纯函数（两条车道的调用点都只拿得到 +// history.Message，作用域里没有各自编排器的 logger），因此以进程默认 logger 为基底再盖上 +// 来源标签，使读取失败与坏数据的痕迹既可见又可追溯（与 internal/auth 的 apikey/jwt 同一手法）。 +// 它只在冷路径上打日志，不在每回合的正常路径上打任何日志，且绝不记消息正文。 +func transcriptLog() *slog.Logger { + return logger.WithSource(slog.Default(), logSource) +} diff --git a/internal/runtime/transcript/transcript_test.go b/internal/runtime/transcript/transcript_test.go new file mode 100644 index 0000000..e3a935c --- /dev/null +++ b/internal/runtime/transcript/transcript_test.go @@ -0,0 +1,167 @@ +package transcript + +import ( + "context" + "errors" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// 本文件钉住共享转录映射的行为（网关与渠道桥的 IM 主人私聊回合都经它读历史）： +// 顺序最旧优先、角色翻译、说话角色归属、文档引用全量还原、条数上限取自本包的 +// ContextCount，以及「读不出来就退化为无转录」的降级口径。 + +// fakeReader 是只实现 Recent 的窄历史视图，并记录每次请求的会话与条数上限。 +type fakeReader struct { + rows []history.Message + err error + session string + n int + callCount int +} + +func (f *fakeReader) Recent(_ context.Context, sessionID string, n int) ([]history.Message, error) { + f.session = sessionID + f.n = n + f.callCount++ + if f.err != nil { + return nil, f.err + } + out := make([]history.Message, 0, len(f.rows)) + for _, m := range f.rows { + if m.SessionID == sessionID { + out = append(out, m) + } + } + if n <= 0 { + return nil, nil + } + if len(out) > n { + out = out[len(out)-n:] + } + return out, nil +} + +// textOf 拼接一条模型消息里的文本部件。 +func textOf(m model.Message) string { + out := "" + for _, c := range m.Content { + if c.Type == model.ContentTypeText { + out += c.Text + } + } + return out +} + +func TestReadMapsRecentRowsOldestFirstWithSpeakerAttribution(t *testing.T) { + t.Parallel() + + rd := &fakeReader{rows: []history.Message{ + {ID: "m1", SessionID: "web", Role: history.RoleUser, Content: "喵"}, + {ID: "m2", SessionID: "web", Role: history.RoleAssistant, Content: "又用喵来敲门啦", PersonaID: "cyrene"}, + {ID: "m3", SessionID: "web", Role: history.RoleSystem, Content: "系统行"}, + {ID: "x1", SessionID: "sess-other", Role: history.RoleUser, Content: "别的会话"}, + }} + + got := Read(context.Background(), rd, "web") + if len(got) != 3 { + t.Fatalf("messages = %d, want 3 (only the requested session): %+v", len(got), got) + } + if got[0].Role != model.RoleUser || textOf(got[0]) != "喵" { + t.Fatalf("messages[0] = %+v, want the oldest user turn first", got[0]) + } + // 助手话语带上说出它的角色，非助手消息保持已存发送者标签。 + if got[1].Role != model.RoleAssistant || got[1].SenderName != "cyrene" { + t.Fatalf("messages[1] = %+v, want her reply attributed to cyrene", got[1]) + } + if got[2].Role != model.RoleSystem { + t.Fatalf("messages[2] = %+v, want the persisted system row", got[2]) + } +} + +func TestReadWindowIsTheSharedContextCount(t *testing.T) { + t.Parallel() + + rows := make([]history.Message, 0, ContextCount+5) + for i := 0; i < ContextCount+5; i++ { + rows = append(rows, history.Message{SessionID: "web", Role: history.RoleUser, Content: "第" + string(rune('a'+i)) + "句"}) + } + rd := &fakeReader{rows: rows} + + got := Read(context.Background(), rd, "web") + // 条数上限只有一个来源：本包的 ContextCount（网关与 IM 车道都从这里取，不许各自造数）。 + if rd.n != ContextCount { + t.Fatalf("requested window = %d, want the shared ContextCount (%d)", rd.n, ContextCount) + } + if len(got) != ContextCount { + t.Fatalf("messages = %d, want %d (the newest window)", len(got), ContextCount) + } +} + +func TestReadDegradesToNil(t *testing.T) { + t.Parallel() + + if got := Read(context.Background(), nil, "web"); got != nil { + t.Fatalf("nil store = %+v, want nil", got) + } + rd := &fakeReader{rows: []history.Message{{SessionID: "web", Role: history.RoleUser, Content: "喵"}}} + if got := Read(context.Background(), rd, ""); got != nil { + t.Fatalf("empty session = %+v, want nil", got) + } + if rd.callCount != 0 { + t.Fatalf("empty session read the store %d times, want 0", rd.callCount) + } + bad := &fakeReader{err: errors.New("boom")} + if got := Read(context.Background(), bad, "web"); got != nil { + t.Fatalf("read error = %+v, want nil (the turn must still run)", got) + } +} + +func TestMessageRestoresEveryDocumentPartOnUserRows(t *testing.T) { + t.Parallel() + + raw, err := history.EncodeDocuments([]history.DocumentRef{ + {Name: "0123456789abcdef0123456789abcdef.txt", MIME: "text/plain", Size: 12}, + {Name: "fedcba9876543210fedcba9876543210.pdf", MIME: "application/pdf", Size: 34}, + }) + if err != nil { + t.Fatalf("EncodeDocuments: %v", err) + } + + msg := Message(history.Message{ID: "m1", Role: history.RoleUser, Content: "读一下", Document: raw}) + if len(msg.Content) != 3 { + t.Fatalf("parts = %d, want text + 2 documents: %+v", len(msg.Content), msg.Content) + } + for i, want := range []history.DocumentRef{ + {Name: "0123456789abcdef0123456789abcdef.txt", MIME: "text/plain", Size: 12}, + {Name: "fedcba9876543210fedcba9876543210.pdf", MIME: "application/pdf", Size: 34}, + } { + part := msg.Content[i+1] + if part.DocumentURL != DocumentPathPrefix+want.Name || part.DocumentMIME != want.MIME { + t.Fatalf("document part[%d] = %+v, want %s%s", i, part, DocumentPathPrefix, want.Name) + } + } + // 非用户行上的文档部件会让编码器 fail-loud，因此刻意省掉（坏数据不该卡死会话）。 + other := Message(history.Message{ID: "m2", Role: history.RoleAssistant, Content: "好", Document: raw}) + for _, c := range other.Content { + if c.DocumentURL != "" { + t.Fatalf("assistant row carries a document part: %+v", c) + } + } + // 坏列不 panic、也不产出部件（调用方另行记录 WARN）。 + broken := Message(history.Message{ID: "m3", Role: history.RoleUser, Content: "坏列", Document: "{not json"}) + if len(broken.Content) != 1 { + t.Fatalf("broken document column produced %d parts, want just the text: %+v", len(broken.Content), broken.Content) + } +} + +func TestMessageRestoresImages(t *testing.T) { + t.Parallel() + + msg := Message(history.Message{ID: "m1", Role: history.RoleUser, Content: "看图", Image: "data:image/png;base64,AAAA"}) + if len(msg.Content) != 2 || msg.Content[1].ImageURL != "data:image/png;base64,AAAA" { + t.Fatalf("image part = %+v, want the stored data URI restored", msg.Content) + } +} diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index 81cf8dd..e3d380d 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -232,7 +232,8 @@ func inboundTargetFor(sess session.Session, platform string) inboundTarget { // runInboundTurn 为被放行的消息驱动 agent 轮次：解析 // 路由目标、把消息携带的文件类附件**落库并挂成文档部件** // （design/24 §3；切片 M1）、把图片媒体内联成 data: URI 部件挂上既有视觉链路 -// （切片 IM-IMG；design/24 §4.2 末）、构建模型可见的消息列表、在轮次之前持久化入站 +// （切片 IM-IMG；design/24 §4.2 末）、构建模型可见的消息列表（主人私聊车道上再前置主会话的 +// 历史转录，见 transcript.go），在轮次之前持久化入站 // 用户消息**连同它们的 `document` 引用列与 `image` 列**（因此即使渠道回复失败它也已落地， // 客户端也有附件芯片与图片可看），并在轮次之后把抽取结论补回同一列，运行 // 循环并把结果交给 deliverReply。 @@ -258,11 +259,13 @@ func (b *Bridge) runInboundTurn(ctx context.Context, msg adapter.UnifiedMessage, // 镜像进主会话时取主会话的绑定，否则取账号的 // 渠道级绑定，再否则取全局默认；群聊或 // 陌生人会话强制使用默认角色。 - runEnv := loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable} - runEnv.PersonaID = b.resolvePersona(ctx, sess, target, adapterName(a)) + runEnv := turnEnv(sess, target, b.resolvePersona(ctx, sess, target, adapterName(a))) senderName := senderDisplayName(msg, actor) loopMsgs := b.buildLoopMessages(ctx, text, runEnv, actor, senderName, msg) + // 附件/图片先挂到**本回合的**用户消息上（attachInboundContent 认列表里第一条 + // RoleUser），再前置历史转录 —— 顺序反过来会把附件挂到转录里的旧用户消息上。 attachInboundContent(loopMsgs, attachments, images) + loopMsgs = b.seedTurnTranscript(ctx, target, loopMsgs) // 附件引用与图片必须同批落进用户消息各自的列（design/24 §2.4 / design/16 §4.4）： // 只把部件挂在内存里的模型可见消息上，客户端读到的那条消息就是一片空白 —— 附件芯片 @@ -368,6 +371,10 @@ func deviceIDFor(msg adapter.UnifiedMessage) string { // 入站内容摄取的产物（已落库的文档部件、内联成 data: URI 的图片部件，以及两条路上 // 「读不到」的诚实提示行）由 attachInboundContent（inbound_content.go）挂到**同一条 // RoleUser 消息**上 —— 用户内容只进用户角色消息是项目红线。 +// +// 历史转录**不在本函数里**：它由 seedTurnTranscript（transcript.go）在本函数返回之后 +// 前置到列表最前面，与网关编排器的播种同序（转录在前、本次消息在后）。因此本函数只负责 +// **本回合自己的**消息（来源行 + 当次用户消息），受限车道的形状在有无转录时保持一致。 func (b *Bridge) buildLoopMessages(_ context.Context, text string, env loop.SessionEnv, actor adapter.Actor, senderName string, src adapter.UnifiedMessage) []model.Message { msg := userTurnMessage(text) thirdParty := env.Restricted() && !actor.IsOwner() && senderName != "" diff --git a/internal/services/channelbridge/persona_test.go b/internal/services/channelbridge/persona_test.go index 8dc70d9..40ec9a4 100644 --- a/internal/services/channelbridge/persona_test.go +++ b/internal/services/channelbridge/persona_test.go @@ -27,15 +27,19 @@ func (p *personaHistory) SessionPersona(_ context.Context, sessionID string) (st return p.personas[sessionID], nil } -// envChatter 是一个 Chatter，记录桥接器为该轮解析出的 SessionEnv， -// 角色正是经由它传递（循环从 ctx 读取它）。 +// envChatter 是一个 Chatter，记录桥接器为该轮解析出的 SessionEnv（角色正是经由它传递， +// 循环从 ctx 读取它）、目标会话 id 与模型可见的消息列表。 type envChatter struct { - env loop.SessionEnv - got bool + env loop.SessionEnv + got bool + sessionID string + msgs []model.Message } -func (e *envChatter) Run(ctx context.Context, _ string, _ []model.Message) (*loop.Result, error) { +func (e *envChatter) Run(ctx context.Context, sessionID string, msgs []model.Message) (*loop.Result, error) { e.env = loop.SessionEnvFrom(ctx) + e.sessionID = sessionID + e.msgs = msgs e.got = true return &loop.Result{Text: "在呢"}, nil } diff --git a/internal/services/channelbridge/transcript.go b/internal/services/channelbridge/transcript.go new file mode 100644 index 0000000..f3ff39f --- /dev/null +++ b/internal/services/channelbridge/transcript.go @@ -0,0 +1,61 @@ +package channelbridge + +import ( + "context" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/session" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" +) + +// 本文件让 IM 的**主人私聊**回合与 Web 主会话回合的环境逐项一致（主人 2026-10-05 要求： +// 处理其他平台的主人消息时和主会话环境完全一致）。 +// +// 三条环境在 loop.SessionEnv 上早已一致（ActorEnv=owner、MemWritable=true、PersonaID 取 +// 主会话绑定，design/22 §7-3），缺的是**历史转录**：Web 主会话回合由编排器播种该会话最近 +// N 条转录（gateway/session_transcript.go），而 IM 回合此前只有「来源系统行 + 当次消息」。 +// 于是消息明明落在同一条会话流里，模型看到的历史却不同 —— 她像是第一次见到主人。 +// +// 本文件补上这一半：主人私聊回合在装配完本回合自己的消息之后、交给循环之前，前置**主会话** +// 的最近 N 条转录。重建逻辑与条数上限都取自 internal/runtime/transcript —— 与网关同一份 +// 实现、同一个常量，绝不另写第二套（两份映射必然漂移）。受限车道（群聊/陌生人）**绝不注入**： +// 那是隔离会话，只看得到自己会话里的当次消息，主人的主会话历史不得外泄（design/16 §4.6）。 + +// turnEnv 返回一条入站消息本轮运行所用的 loop 会话环境。 +// +// - 主人私聊（inboundTarget.private）与 Web 主会话是**同一条车道**，因此经主人车道的 +// 唯一构造入口 loop.OwnerEnvFor 派生：两条车道逐字段同源，SessionEnv 将来新增字段时 +// 会一起拿到，而不是各自漂移出「同一个她、两套环境」。 +// - 受限会话（群聊/陌生人）由结构化解析出的隔离边界决定：记忆只读（MemWritable=false）、 +// 并强制默认角色（design/22 §7-4 —— resolvePersona 对受限会话恒返回 ""）。身份与权限 +// 始终来自 actor 元数据，绝不来自这里的任何字符串（项目红线）。 +func turnEnv(sess session.Session, target inboundTarget, personaID string) loop.SessionEnv { + if target.private { + return loop.OwnerEnvFor(personaID) + } + return loop.SessionEnv{ActorEnv: sess.Env(), MemWritable: sess.MemWritable, PersonaID: personaID} +} + +// seedTurnTranscript 在主人私聊车道上把**主会话**最近的转录置于本回合模型可见消息之前 +// （与网关编排器的 startTurnMessages 同序：转录在前、本次消息在后），使 IM 回合与 Web 主 +// 会话看到同一份对话历史。受限（群聊/陌生人）车道原样返回 —— 隔离上下文绝不注入主人的 +// 主会话历史。 +// +// 时刻与网关一致：调用点在本回合用户消息落库**之前**（runInboundTurn 里装配先于 +// persistToMain），因此当次消息绝不会在上下文里出现两次。 +// +// 转录读不出来（未接历史存储、读失败、空会话）时同样原样返回，这正是既有行为：主人私聊 +// 回合退化为「来源行 + 当次消息」，而不是让回合失败。 +func (b *Bridge) seedTurnTranscript(ctx context.Context, target inboundTarget, msgs []model.Message) []model.Message { + if !target.private { + return msgs + } + seed := transcript.Read(ctx, b.history, target.sessionID) + if len(seed) == 0 { + return msgs + } + out := make([]model.Message, 0, len(seed)+len(msgs)) + out = append(out, seed...) + return append(out, msgs...) +} diff --git a/internal/services/channelbridge/transcript_test.go b/internal/services/channelbridge/transcript_test.go new file mode 100644 index 0000000..62a6a1a --- /dev/null +++ b/internal/services/channelbridge/transcript_test.go @@ -0,0 +1,222 @@ +package channelbridge + +import ( + "context" + "strings" + "sync" + "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/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" +) + +// 本文件覆盖「IM 主人私聊的回合环境与 Web 主会话完全一致」这条要求里**历史转录**那一半 +// （见 transcript.go 的文件头、DevDocs/工作小结/2026-10-05_QQ判主失效与脱敏日志.md 附 2）： +// +// - 主人私聊车道的模型可见列表 = 主会话最近转录（同一个重建实现、同一条数上限）+ +// 渠道来源系统行 + 当次用户消息； +// - 群聊/陌生人/隔离会话**绝不**注入：不读主会话历史、主人的话一个字都不进模型可见消息； +// - 车道环境由 loop.OwnerEnvFor 同源派生，因此与 Web 主会话逐字段相同（不靠各自手搓）。 +// +// Web 侧的端到端对应断言在 gateway 的 TestChatTurnCarriesSpeakerAttributionFromTranscript +// （同一份 internal/runtime/transcript 产出的列表）。 + +// seededHistory 在 captureHistory（记录落库）之上把 Recent 换成「返回预置的转录行」， +// 使播种用例无需真库即可驱动历史。它记录每次请求的会话与条数上限 —— 前者断言 +// 「只读主会话」，后者断言「条数取自共享常量」。 +type seededHistory struct { + *captureHistory + rows []history.Message + + mu sync.Mutex + asks []transcriptAsk +} + +// transcriptAsk 是一次 Recent 调用的入参快照。 +type transcriptAsk struct { + session string + n int +} + +func (h *seededHistory) Recent(_ context.Context, sessionID string, n int) ([]history.Message, error) { + h.mu.Lock() + defer h.mu.Unlock() + h.asks = append(h.asks, transcriptAsk{session: sessionID, n: n}) + rows := make([]history.Message, 0, len(h.rows)) + for _, m := range h.rows { + if m.SessionID == sessionID { + rows = append(rows, m) + } + } + if n <= 0 { + return nil, nil + } + if len(rows) > n { + rows = rows[len(rows)-n:] + } + return rows, nil +} + +// asksFor 返回已经发生的 Recent 调用（按发生顺序）。 +func (h *seededHistory) asksFor() []transcriptAsk { + h.mu.Lock() + defer h.mu.Unlock() + return append([]transcriptAsk(nil), h.asks...) +} + +// turnMessageText 拼接一条模型消息里的文本部件。 +func turnMessageText(m model.Message) string { + out := "" + for _, c := range m.Content { + if c.Type == model.ContentTypeText { + out += c.Text + } + } + return out +} + +// groupMessage 是一条被点名的群消息（design/12 §6 策略二 ⇒ 会得到回复）， +// 因此群/陌生人用例能走到循环，从而验证「转录没有被注入」。 +func groupMessage(senderID string, sender adapter.Actor) adapter.UnifiedMessage { + return adapter.UnifiedMessage{ + Platform: "telegram", ChannelID: "-20", ChannelType: adapter.ChannelGroup, + GroupID: "-20", GroupName: "Gang", SenderID: senderID, SenderName: "张三", + Sender: sender, Mentions: []adapter.Mention{{UserID: senderID}}, + Content: "hello all", ContentType: adapter.ContentText, + } +} + +func TestOwnerPrivateTurnSeedsMainSessionTranscript(t *testing.T) { + t.Parallel() + + hist := &seededHistory{captureHistory: &captureHistory{}, rows: []history.Message{ + {ID: "m1", SessionID: history.MainSessionID, Role: history.RoleUser, Content: "喵"}, + {ID: "m2", SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "又用喵来敲门啦", PersonaID: "cyrene"}, + // 隔离会话的行绝不进主会话车道的上下文。 + {ID: "x1", SessionID: "sess-other", Role: history.RoleUser, Content: "群里的别人的话"}, + }} + chat := &envChatter{} + sess := &stubSessions{sess: session.Session{SessionID: "sess-1", Owner: true, MemWritable: true}} + 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()) + + // 与 Web 主会话是同一条车道：本轮跑在主会话上（design/16 §4.4）。 + if chat.sessionID != history.MainSessionID { + t.Fatalf("loop session id = %q, want %q", chat.sessionID, history.MainSessionID) + } + // 模型可见列表 = 主会话转录（最旧优先）+ 渠道来源系统行 + 当次用户消息。 + // 除来源行外，它与 Web 主会话回合的列表逐条一致 —— 这正是「同一个她」的前提。 + if len(chat.msgs) != 4 { + t.Fatalf("loop messages = %d, want 4 (transcript 2 + origin + user): %+v", len(chat.msgs), chat.msgs) + } + if got := chat.msgs[0]; got.Role != model.RoleUser || turnMessageText(got) != "喵" { + t.Fatalf("messages[0] = %+v, want the earlier user turn 喵", got) + } + if got := chat.msgs[1]; got.Role != model.RoleAssistant || turnMessageText(got) != "又用喵来敲门啦" || got.SenderName != "cyrene" { + t.Fatalf("messages[1] = %+v, want her earlier reply attributed to cyrene", got) + } + if got := chat.msgs[2]; got.Role != model.RoleSystem || !strings.Contains(turnMessageText(got), "来源：telegram · 私聊") { + t.Fatalf("messages[2] = %+v, want the telegram origin system row", got) + } + if got := chat.msgs[3]; got.Role != model.RoleUser || turnMessageText(got) != "hello" { + t.Fatalf("messages[3] = %+v, want this turn's user message", got) + } + // 当次消息只出现一次：转录是在本回合消息落库**之前**读的（与网关播种同一时刻）。 + seen := 0 + for _, m := range chat.msgs { + if m.Role == model.RoleUser && turnMessageText(m) == "hello" { + seen++ + } + } + if seen != 1 { + t.Fatalf("current user message appears %d times, want exactly 1", seen) + } + // 只读主会话，且条数上限取自共享常量（本包不许另造一个数）。 + asks := hist.asksFor() + if len(asks) != 1 || asks[0].session != history.MainSessionID { + t.Fatalf("transcript reads = %+v, want exactly one for %q", asks, history.MainSessionID) + } + if asks[0].n != transcript.ContextCount { + t.Fatalf("transcript window = %d, want the shared transcript.ContextCount (%d)", asks[0].n, transcript.ContextCount) + } + // 车道环境与 Web 主会话**同源派生**：本夹具主会话未绑定角色，因此逐字段等于 loop.OwnerEnv()。 + if chat.env != loop.OwnerEnv() { + t.Fatalf("run env = %+v, want the shared owner env %+v", chat.env, loop.OwnerEnv()) + } +} + +func TestGroupTurnNeverSeedsMainSessionTranscript(t *testing.T) { + t.Parallel() + + hist := &seededHistory{captureHistory: &captureHistory{}, rows: []history.Message{ + {ID: "m1", SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "主人的私事：明天九点体检", PersonaID: "cyrene"}, + }} + chat := &envChatter{} + sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: false, MemWritable: false}} + b, err := New(Options{Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "100"), History: hist}) + if err != nil { + t.Fatalf("New: %v", err) + } + + b.process(context.Background(), groupMessage("9", adapter.Actor{Role: adapter.RoleMember})) + + // 受限车道的形状与改动前逐字相同：第三方边界系统行 + 当次用户消息，没有转录。 + if len(chat.msgs) != 2 { + t.Fatalf("loop messages = %d, want 2 (boundary + user): %+v", len(chat.msgs), chat.msgs) + } + // 主人的主会话历史一个字都不许出现在群聊的模型可见消息里。 + for i, m := range chat.msgs { + if strings.Contains(turnMessageText(m), "主人的私事") { + t.Fatalf("messages[%d] leaked the owner's main-session transcript: %+v", i, m) + } + } + if asks := hist.asksFor(); len(asks) != 0 { + t.Fatalf("restricted lane read the transcript: %+v", asks) + } + if !chat.env.Restricted() || chat.env.MemWritable { + t.Fatalf("run env = %+v, want the restricted (read-only memory) env", chat.env) + } +} + +func TestOwnerGroupTurnKeepsIsolatedLaneWithoutTranscript(t *testing.T) { + t.Parallel() + + // 隔离会话自己有历史行也不播种：本切片只动主人私聊这条车道，群聊隔离行为不许改坏。 + hist := &seededHistory{captureHistory: &captureHistory{}, rows: []history.Message{ + {ID: "m1", SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "主人的私事", PersonaID: "cyrene"}, + {ID: "g1", SessionID: "sess-g", Role: history.RoleUser, Content: "群里的上一句"}, + }} + chat := &envChatter{} + sess := &stubSessions{sess: session.Session{SessionID: "sess-g", IsGroup: true, Owner: true, MemWritable: false}} + b, err := New(Options{Loop: chat, Sessions: sess, Owner: testOwnerProfile("telegram", "owner_uid"), History: hist}) + if err != nil { + t.Fatalf("New: %v", err) + } + + b.process(context.Background(), groupMessage("owner_uid", adapter.Actor{Role: adapter.RoleOwner})) + + if chat.sessionID != "sess-g" { + t.Fatalf("loop session id = %q, want the isolated sess-g", chat.sessionID) + } + if len(chat.msgs) != 2 { + t.Fatalf("loop messages = %d, want 2 (origin/boundary + user): %+v", len(chat.msgs), chat.msgs) + } + for i, m := range chat.msgs { + text := turnMessageText(m) + if strings.Contains(text, "主人的私事") || strings.Contains(text, "群里的上一句") { + t.Fatalf("messages[%d] = %q, want no injected transcript on an isolated lane", i, text) + } + } + if asks := hist.asksFor(); len(asks) != 0 { + t.Fatalf("isolated lane read the transcript: %+v", asks) + } +} diff --git a/internal/services/gateway/chat_documents_bounds_test.go b/internal/services/gateway/chat_documents_bounds_test.go index ec367fc..7123b7f 100644 --- a/internal/services/gateway/chat_documents_bounds_test.go +++ b/internal/services/gateway/chat_documents_bounds_test.go @@ -21,6 +21,7 @@ import ( "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" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" ) // TestChatDocumentsStreamPathCarriesDocuments 钉住 `stream:true` 这条转发路径： @@ -92,7 +93,7 @@ func TestTranscriptMessageKeepsDocumentReference(t *testing.T) { if err != nil { t.Fatalf("encode document: %v", err) } - msg := transcriptMessage(history.Message{ID: "m1", Role: history.RoleUser, Content: "读一下", Document: raw}) + msg := transcript.Message(history.Message{ID: "m1", Role: history.RoleUser, Content: "读一下", Document: raw}) docs := chatDocumentsOf([]model.Message{msg}) if len(docs) != 1 { t.Fatalf("文档引用没能重建回来：%+v", msg) @@ -103,17 +104,17 @@ func TestTranscriptMessageKeepsDocumentReference(t *testing.T) { } // 没有附件的消息照旧没有文档部件；坏列不 panic 也不产生部件（打 WARN）。 if got := chatDocumentsOf([]model.Message{ - transcriptMessage(history.Message{ID: "m2", Role: history.RoleUser, Content: "无附件"}), + transcript.Message(history.Message{ID: "m2", Role: history.RoleUser, Content: "无附件"}), }); len(got) != 0 { t.Fatalf("无附件消息产生了文档部件：%+v", got) } if got := chatDocumentsOf([]model.Message{ - transcriptMessage(history.Message{ID: "m3", Role: history.RoleUser, Content: "坏列", Document: "{not json"}), + transcript.Message(history.Message{ID: "m3", Role: history.RoleUser, Content: "坏列", Document: "{not json"}), }); len(got) != 0 { t.Fatalf("坏列产生了文档部件：%+v", got) } // 助手回合的说话者标注不受本次改动影响；附件不会落到非用户回合上。 - assistant := transcriptMessage(history.Message{ + assistant := transcript.Message(history.Message{ ID: "m4", Role: history.RoleAssistant, Content: "答复", PersonaID: "cyrene", Document: raw, }) if assistant.SenderName != "cyrene" { diff --git a/internal/services/gateway/documents.go b/internal/services/gateway/documents.go index afa62d4..b11e80e 100644 --- a/internal/services/gateway/documents.go +++ b/internal/services/gateway/documents.go @@ -31,6 +31,7 @@ import ( apperr "git.yeij.top/AskaEth/Cyrene_v2/internal/errors" "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/documents" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" ) const ( @@ -59,8 +60,10 @@ const ( documentFormField = "file" // documentFilePrefix 是文档读取的版本化路径前缀。它位于 `/api/v1/` 之下， - // 因此归网关 mux（而非 SPA 回退），并属于版本化客户端契约。 - documentFilePrefix = "/api/v1/documents/" + // 因此归网关 mux（而非 SPA 回退），并属于版本化客户端契约。字面量只有一处： + // 转录还原（internal/runtime/transcript）用的必须是同一个前缀，否则续接的 + // 回合会把模型指到一个取不到的地址。 + documentFilePrefix = transcript.DocumentPathPrefix // documentSubDir 是文档落盘子目录名：`/uploads/documents`。 // 放在图片同根之下，便于 S4 的 `file` 工具沙箱（roots = dataDir）覆盖。 diff --git a/internal/services/gateway/persona_env.go b/internal/services/gateway/persona_env.go index fd5423d..485b728 100644 --- a/internal/services/gateway/persona_env.go +++ b/internal/services/gateway/persona_env.go @@ -76,10 +76,12 @@ type turnEnvResolver struct { // 全访问 owner env（身份不变）加上该会话绑定的角色。 // 空白角色意味着「继承全局默认」；loop 执行那最后一次 // 回退，因此网关永远不必知道默认角色的 id。 +// +// 构造走 loop.OwnerEnvFor —— 与 IM 的主人私聊车道**同一个入口** +// （channelbridge/transcript.go 的 turnEnv）：两条车道的 runEnv 因此逐字段同源， +// SessionEnv 将来新增字段时会一起拿到，而不是各自漂移。 func (t turnEnvResolver) owner(ctx context.Context, sessionID string) loop.SessionEnv { - env := loop.OwnerEnv() - env.PersonaID = t.sessionPersona(ctx, sessionID) - return env + return loop.OwnerEnvFor(t.sessionPersona(ctx, sessionID)) } // sessionPersona 返回绑定到 sessionID 的角色，会话继承全局 diff --git a/internal/services/gateway/session_document_status_test.go b/internal/services/gateway/session_document_status_test.go index bfffe86..d8a9411 100644 --- a/internal/services/gateway/session_document_status_test.go +++ b/internal/services/gateway/session_document_status_test.go @@ -13,6 +13,7 @@ import ( "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" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" ) // 本文件是 S3.5 两个契约缺口的回归测试（design/24 §6.1、§9.7）： @@ -80,7 +81,7 @@ func TestTranscriptRestoresAllDocumentParts(t *testing.T) { if len(recent) != 1 { t.Fatalf("recent = %+v, want 1 条", recent) } - parts := chatDocumentsOf([]model.Message{transcriptMessage(recent[0])}) + parts := chatDocumentsOf([]model.Message{transcript.Message(recent[0])}) if len(parts) != 3 { t.Fatalf("还原出的文档部件 = %d 个, want 3（全部附件都要回到上下文）", len(parts)) } @@ -100,7 +101,7 @@ func TestTranscriptRestoresLegacySingleObjectColumn(t *testing.T) { t.Parallel() const legacy = `{"name":"aaaabbbbccccdddd.txt","mime":"text/plain","size":1234,"status":"extracted"}` parts := chatDocumentsOf([]model.Message{ - transcriptMessage(history.Message{ID: "m1", Role: history.RoleUser, Content: "旧行", Document: legacy}), + transcript.Message(history.Message{ID: "m1", Role: history.RoleUser, Content: "旧行", Document: legacy}), }) if len(parts) != 1 || parts[0].DocumentName != "aaaabbbbccccdddd.txt" { t.Fatalf("旧单对象列还原出 = %+v, want 那一条引用", parts) diff --git a/internal/services/gateway/session_steer.go b/internal/services/gateway/session_steer.go index 8a71475..d577d67 100644 --- a/internal/services/gateway/session_steer.go +++ b/internal/services/gateway/session_steer.go @@ -8,6 +8,7 @@ import ( "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" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" ) // 本文件实现「忙时新消息在回合内边界注入同一回合」的网关侧 @@ -89,7 +90,7 @@ func (s *steerTurn) Steer(ctx context.Context, req loop.SteerRequest) []model.Me sessionID = s.sessionID } o := s.orchestrator - reader, ok := o.history.(transcriptReader) + reader, ok := o.history.(transcript.Reader) if !ok { return nil } @@ -99,7 +100,7 @@ func (s *steerTurn) Steer(ctx context.Context, req loop.SteerRequest) []model.Me } // 每个边界都重读转录：设计要求的「最新性＝下一个边界即生效」正是靠这次重读， // 而不是靠回合开始时的那一次播种（design/15 §2.4）。 - recent, err := reader.Recent(ctx, sessionID, transcriptContextCount) + recent, err := reader.Recent(ctx, sessionID, transcript.ContextCount) if err != nil { steerLog(o).Debug("steer transcript read", "session", sessionID, "step", req.Step, "err", err) return nil @@ -127,7 +128,7 @@ func (s *steerTurn) Steer(ctx context.Context, req loop.SteerRequest) []model.Me // 帧壳必须带本路的 turnId（忙时并行规范草案 §2.5）：真并行下客户端要 // 靠它把这条消息配到**正确那一路**的 `turn.end`，否则灰态收不了口。 o.emitTurnState(sessionID, s.turnID, wsSteerAck, map[string]any{"messageId": c.row.ID, "atStep": req.Step}) - out = append(out, transcriptMessage(c.row)) + out = append(out, transcript.Message(c.row)) } steerLog(o).Info("pending steered", "session", sessionID, "step", req.Step, "messages", len(out)) return out diff --git a/internal/services/gateway/session_transcript.go b/internal/services/gateway/session_transcript.go index 3887eed..ef63c52 100644 --- a/internal/services/gateway/session_transcript.go +++ b/internal/services/gateway/session_transcript.go @@ -2,56 +2,34 @@ package gateway import ( "context" - "log/slog" - "git.yeij.top/AskaEth/Cyrene_v2/internal/logger" - "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/transcript" ) -// 本文件拥有从持久化转录重建一个回合对话上下文的工作 -// （design/15 §2.2/§4「上下文有界：只保留最近 N 条上下文」、design/02 -// §1 主会话承载完整上下文、design/22 §7-5）。一个回合此前只从刚 -// 发送的消息开始，使模型完全看不到它正在延续的 -// 对话；更早回合的助手话语也没有记录 -// 是谁说的，因此切换进会话的角色会把另一个 -// 角色的话读成自己的。 +// 本文件拥有「一个新回合的对话上下文从哪里来」的网关侧**决策与时刻** +// （design/15 §2.2/§4「上下文有界：只保留最近 N 条上下文」、design/02 §1 主会话 +// 承载完整上下文、design/22 §7-5）。重建本身（读最近 N 条 + 翻成模型消息）住在 +// internal/runtime/transcript —— 它与渠道桥的 IM 主人私聊回合是**同一份实现**， +// 两条车道的模型可见历史因此不会漂移（IM 与 Web 只在呈现形态上不同：IM 只发 chat 文本）。 // -// 播种修好了这两半：新回合携带该会话最近的 -// 持久化消息，且每条助手消息都被说话的角色标注 -// （model.Message.SenderName），使 loop 的上下文组装能归属它 -// （loop/attribution.go）。消息上只走标签 —— 没有任何消息体 -// 被改写。 +// 一个回合此前只从刚发送的消息开始，使模型完全看不到它正在延续的对话；更早回合的 +// 助手话语也没有记录是谁说的，因此切换进会话的角色会把另一个角色的话读成自己的。 // -// 范围说明：被播种的窗口就是该对话，因此 loop 的逐步 -// 后台记忆抽取现在读的是这些近期消息，而不是 -// 单条刚发送的消息。抽取出的条目仍归到该回合的角色 -// 名下（loop.persistMemories，design/22 §4.2），且同一内容的重复抽取 -// 会被抽取器去重，但一个在另一角色活跃时首次说出的事实， -// 若当时从未被抽取，现在就可能被当前角色拾取。 -// 把抽取输入收窄为当前回合的消息是一个 -// 独立的决定，留作后续跟进。 +// 播种修好了这两半：新回合携带该会话最近的持久化消息，且每条助手消息都被说话的角色 +// 标注（model.Message.SenderName），使 loop 的上下文组装能归属它 +// （loop/attribution.go）。消息上只走标签 —— 没有任何消息体被改写。 +// +// 范围说明：被播种的窗口就是该对话，因此 loop 的逐步后台记忆抽取现在读的是这些近期 +// 消息，而不是单条刚发送的消息。抽取出的条目仍归到该回合的角色名下（loop.persistMemories， +// design/22 §4.2），且同一内容的重复抽取会被抽取器去重，但一个在另一角色活跃时首次说出 +// 的事实，若当时从未被抽取，现在就可能被当前角色拾取。把抽取输入收窄为当前回合的消息是 +// 一个独立的决定，留作后续跟进。 -// transcriptContextCount 限定一个新回合带入模型上下文的持久化 -// 消息数。它与编排器的续跑尾部上限 -// （defaultMaxTail）一致，使播种回合与续跑回合保持同样大小； -// 超出之后仍由 loop 自身的上下文预算治理压缩。 -const transcriptContextCount = 20 - -// transcriptReader 是播种所需的窄历史视图。*history.Store -// 满足它；未满足的存储（测试替身，或未接持久化 -// 构建的网关）只是得不到播种上下文，因此编排器 -// 完全照旧继续运行。 -type transcriptReader interface { - Recent(ctx context.Context, sessionID string, n int) ([]history.Message, error) -} - -// startTurnMessages 返回新回合运行其上的对话：该会话 -// 已播种的转录，后跟客户端刚发送的消息。转录 -// 不可用时，调用方的消息原样返回，这正是 -// 既有行为。 +// startTurnMessages 返回新回合运行其上的对话：该会话已播种的转录，后跟客户端刚发送的 +// 消息。转录不可用时，调用方的消息原样返回，这正是既有行为。 func (o *Orchestrator) startTurnMessages(ctx context.Context, sessionID string, messages []model.Message) []model.Message { - seed := o.sessionTranscript(ctx, sessionID) + seed := transcript.Read(ctx, o.history, sessionID) if len(seed) == 0 { return messages } @@ -67,7 +45,7 @@ func (o *Orchestrator) startTurnMessages(ctx context.Context, sessionID string, // 播种的时刻是刻意的：它在 persistUserMessages 写入新消息**之前**，因此刚发送的 // 消息绝不在上下文中重复（design/15 §2.2/§4、design/22 §7-5）。没有历史（或空 // 会话）的存储让该回合只使用调用方给出的消息。它是在状态锁下的有界索引读取 -// （transcriptContextCount 行），而状态锁从不会被运行中的回合持有，因此它无法把 +// （transcript.ContextCount 行），而状态锁从不会被运行中的回合持有，因此它无法把 // Submit 堵在回合后面。 func (o *Orchestrator) seedTurnLocked(ctx context.Context, st *sessionState, sessionID string, messages []model.Message) []model.Message { turn := o.startTurnMessages(ctx, sessionID, messages) @@ -75,110 +53,6 @@ func (o *Orchestrator) seedTurnLocked(ctx context.Context, st *sessionState, ses return turn } -// sessionTranscript 读取该会话最近的持久化消息并把它们 -// 映射为模型消息（最旧优先）。nil/不可读的存储或读错误 -// 都得到 nil，因此聊天绝不被元数据查找阻塞；没有 -// 历史的会话也什么都不产出 —— 二者都退化为既有的 -// 「只有新消息」上下文。 -func (o *Orchestrator) sessionTranscript(ctx context.Context, sessionID string) []model.Message { - if o.history == nil || sessionID == "" { - return nil - } - reader, ok := o.history.(transcriptReader) - if !ok { - return nil - } - recent, err := reader.Recent(ctx, sessionID, transcriptContextCount) - if err != nil { - o.log.Debug("turn transcript read", "err", err, "session", sessionID) - return nil - } - out := make([]model.Message, 0, len(recent)) - for _, h := range recent { - out = append(out, transcriptMessage(h)) - } - return out -} - -// transcriptMessage 把一条持久化聊天消息映射为回合延续所依据的 -// 模型消息。助手消息携带说话的角色（该回合已存的 -// persona_id）作为其说话者标签，因此即使会话被切换过， -// 上下文也能把它归属到那个角色；非助手消息保持其 -// 已存发送者标签（design/16 §4.6 的第三方显示名），绝不 -// 被重新标为说话角色。文档引用**全部**一并还原（design/24 §9.5）： -// 少了它，附件会在发起它的那一回合之后的所有回合里消失。 -func transcriptMessage(h history.Message) model.Message { - role := historyRoleToModel(h.Role) - msg := model.Message{ - Role: role, - Content: []model.Content{{Type: model.ContentTypeText, Text: h.Content}}, - } - if h.Image != "" { - msg.Content = append(msg.Content, model.Content{Type: model.ContentTypeImage, ImageURL: h.Image}) - } - msg.Content = append(msg.Content, transcriptDocumentParts(role, h)...) - if role == model.RoleAssistant { - msg.SenderName = h.PersonaID - return msg - } - msg.SenderName = h.SenderName - return msg -} - -// transcriptDocumentParts 把已存 document 列里的**全部**引用还原为该回合的文档 -// 部件（design/24 §9.5）。部件只交给 RoleUser 回合（§2.3-③）：文档在任何别的 -// 角色上都会让编码器 fail-loud，而一行坏数据不该把整个会话卡死。 -// -// 列里是引用数组（一个回合最多 5 个附件，缺口①），因此这里逐条还原、保持部件 -// 顺序 —— 少还原一条，那个附件就会在发起它的回合之后的所有回合里消失，与 -// 「清单恒在」的承诺自相矛盾。 -// -// 列被写坏（或附件落在非用户行上）时**不静默**：打一条带来源的 WARN 并省掉 -// 该部件，回合照常继续 —— 附件凭空消失正是这条清单要修的缺陷，因此必须留下 -// 痕迹。日志只带消息 id 与错误，绝不带正文。 -func transcriptDocumentParts(role model.Role, h history.Message) []model.Content { - if h.Document == "" { - return nil - } - if role != model.RoleUser { - transcriptDocumentLog().Warn("transcript document on non-user row", "message", h.ID, "role", h.Role) - return nil - } - refs, err := history.DecodeDocuments(h.Document) - if err != nil { - transcriptDocumentLog().Warn("transcript document decode", "message", h.ID, "err", err) - return nil - } - parts := make([]model.Content, 0, len(refs)) - for _, ref := range refs { - if ref.Name == "" { - transcriptDocumentLog().Warn("transcript document without name", "message", h.ID) - continue - } - doc := model.Document{ - Name: ref.Name, - MIME: ref.MIME, - URL: documentFilePrefix + ref.Name, - Size: ref.Size, - } - parts = append(parts, model.DocumentMessage(model.RoleUser, "", doc).Content...) - } - return parts -} - -// historyRoleToModel 把已存聊天角色映射为模型的角色词汇， -// 未知角色默认归到用户通道。 -func historyRoleToModel(role string) model.Role { - switch role { - case history.RoleAssistant: - return model.RoleAssistant - case history.RoleSystem: - return model.RoleSystem - default: - return model.RoleUser - } -} - // assistantTailMessage 为一个已结算回合的回复构建续跑尾部 // 条目，并用说话的角色标注它（design/22 §7-5）：该尾部是 // 播种转录的内存孪生，因此之后切换进来的角色必须能 @@ -189,15 +63,3 @@ func assistantTailMessage(text, personaID string) model.Message { msg.SenderName = personaID return msg } - -// transcriptLogSource 是转录重建的日志来源标签。 -const transcriptLogSource = "gateway/transcript" - -// transcriptDocumentLog 返回标注来源的子日志器。转录重建是纯函数（编排器的 -// 播种与转向两条路径都只拿得到 history.Message，没有 logger 在作用域内）， -// 因此以进程默认 logger 为基底再盖上来源标签，使坏数据的 WARN 既可见又可 -// 追溯（与 internal/auth 的 apikey/jwt 同一手法）。它只用于「数据被写坏」 -// 这类冷路径，不在每回合的正常路径上打任何日志。 -func transcriptDocumentLog() *slog.Logger { - return logger.WithSource(slog.Default(), transcriptLogSource) -}