AskaEth
5de708a560
裁定（已发安卓）：**能修**，走**写入侧规整** —— 不指望模型输出、不动客户端。 - 新增 internal/runtime/turnreply/paragraph.go 的 NormalizeParagraphs： 把未被 <p> 包裹的段落补上 <p>（按空行分段；<action>/<narration> 段也包进 <p>， 与正确形态 <p><action>…</action>…</p> 一致） - **幂等**守门断言：已完全包裹的文本逐字不变（防二次加工） - 落点覆盖 Web 与 IM 两条落库路径（gateway/session_turn_reply.go、channelbridge/turnreply.go、 proactive.go）⇒ 两边共用同一份规整文本 ⇒ **历史一次性变好、两个客户端零改动**、渲染口径不分叉 （测试与变异详见切片报告）
336 lines
14 KiB
Go
336 lines
14 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"net/http"
|
||
"strings"
|
||
"testing"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||
"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"
|
||
)
|
||
|
||
// 本文件锁定 chat_messages.turn_id 的四条出口与两条写入路径（design/15 §2.5.9）：
|
||
//
|
||
// - 出口：历史 DTO（REST `GET /sessions/{id}/messages`）、WS 握手 `history` 回放、
|
||
// 以及经 toHistoryMessage 构造的 `message_persisted` / `message_edited` 帧；
|
||
// - 写入：落库时已知所属路（新起的一路 + 它的助手回复，随 INSERT）、以及落库时
|
||
// 还不知道所属路（忙时入队 / 被转向吃进，由实际处理它的那一路用 assignTurnID
|
||
// 补写）。
|
||
//
|
||
// 语义口径全部是**可选**的：空 ⇒ 该键从 JSON 里消失，与同族可选字段
|
||
// （personaId / clientMsgId / documents）逐字同形。
|
||
|
||
// turnsOf 把一页消息按内容映射到它们的 turn_id，便于断言配对。
|
||
func turnsOf(msgs []history.Message) map[string]string {
|
||
out := make(map[string]string, len(msgs))
|
||
for _, m := range msgs {
|
||
out[m.Content] = m.TurnID
|
||
}
|
||
return out
|
||
}
|
||
|
||
// TestHistoryMessageTurnIDCarriedAndOmitted 钉住 DTO 形状：有标识就带出来，空则
|
||
// 整个键消失（客户端据「键在不在」区分「属于某一路」与「不属于任何已知回合」）。
|
||
func TestHistoryMessageTurnIDCarriedAndOmitted(t *testing.T) {
|
||
t.Parallel()
|
||
|
||
withTurn := toHistoryMessage(history.Message{ID: "m1", SessionID: "web", Role: history.RoleUser, Content: "问", TurnID: "turn-a"})
|
||
b, err := json.Marshal(withTurn)
|
||
if err != nil {
|
||
t.Fatalf("marshal: %v", err)
|
||
}
|
||
var decoded map[string]any
|
||
if err := json.Unmarshal(b, &decoded); err != nil {
|
||
t.Fatalf("unmarshal: %v", err)
|
||
}
|
||
if decoded["turnId"] != "turn-a" {
|
||
t.Fatalf("turnId = %v, want turn-a (%s)", decoded["turnId"], b)
|
||
}
|
||
|
||
// 空标识（旧行、IM 镜像、落库时还不知道所属路的行）⇒ 键缺席。
|
||
legacy := toHistoryMessage(history.Message{ID: "m2", SessionID: "web", Role: history.RoleUser, Content: "问"})
|
||
b, err = json.Marshal(legacy)
|
||
if err != nil {
|
||
t.Fatalf("marshal legacy: %v", err)
|
||
}
|
||
if strings.Contains(string(b), `"turnId"`) {
|
||
t.Fatalf("空 turn_id 应当省略该键，得到 %s", b)
|
||
}
|
||
}
|
||
|
||
// TestHistoryMessagesEndpointCarriesTurnID 端到端演练读接口：同一路的两条消息
|
||
// 带同一个 turnId，旧行（空）连键都不出现 —— 客户端因此能在重连后按 turnId 把
|
||
// 「回复 ↔ 提问」配对起来，而不是靠顺序猜。
|
||
func TestHistoryMessagesEndpointCarriesTurnID(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
_, ts, jwt, st := newHistoryEnv(t)
|
||
for _, m := range []history.Message{
|
||
{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "这条路的问题", TurnID: "turn-leg1", CreatedAt: 1000},
|
||
{SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "这条路的回答", TurnID: "turn-leg1", CreatedAt: 1500},
|
||
{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "另一路的问题", TurnID: "turn-leg2", CreatedAt: 2000},
|
||
{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "本列之前的老行", CreatedAt: 2500},
|
||
} {
|
||
if err := st.Append(ctx, m); err != nil {
|
||
t.Fatalf("seed %q: %v", m.Content, err)
|
||
}
|
||
}
|
||
tok := signAny(t, jwt, "acct", auth.RoleOwner)
|
||
|
||
// 先按原始 JSON 断言「键存在与否」，再按 DTO 断言取值。
|
||
resp, body := doReq(t, http.MethodGet, ts.URL+"/api/v1/sessions/"+history.MainSessionID+"/messages", nil, tok)
|
||
if resp.StatusCode != http.StatusOK {
|
||
t.Fatalf("messages status = %d, want 200 (body=%s)", resp.StatusCode, body)
|
||
}
|
||
var raw []map[string]any
|
||
decodeDataSlice(t, body, &raw)
|
||
seen := make(map[string]string, len(raw))
|
||
for _, m := range raw {
|
||
content, _ := m["content"].(string)
|
||
id, ok := m["turnId"]
|
||
if !ok {
|
||
seen[content] = "<absent>"
|
||
continue
|
||
}
|
||
seen[content], _ = id.(string)
|
||
}
|
||
for content, want := range map[string]string{
|
||
"这条路的问题": "turn-leg1",
|
||
"这条路的回答": "turn-leg1",
|
||
"另一路的问题": "turn-leg2",
|
||
"本列之前的老行": "<absent>",
|
||
} {
|
||
if got := seen[content]; got != want {
|
||
t.Fatalf("%q 的 turnId = %q, want %q（整页=%s）", content, got, want, body)
|
||
}
|
||
}
|
||
|
||
var page []historyMessage
|
||
decodeDataSlice(t, body, &page)
|
||
turns := turnsOfWire(page)
|
||
if turns["这条路的问题"] != "turn-leg1" || turns["这条路的回答"] != "turn-leg1" {
|
||
t.Fatalf("同一路两条 = %q / %q，want 都是 turn-leg1", turns["这条路的问题"], turns["这条路的回答"])
|
||
}
|
||
}
|
||
|
||
// turnsOfWire 是 historyMessage 版的内容 → turnId 映射。
|
||
func turnsOfWire(msgs []historyMessage) map[string]string {
|
||
out := make(map[string]string, len(msgs))
|
||
for _, m := range msgs {
|
||
out[m.Content] = m.TurnID
|
||
}
|
||
return out
|
||
}
|
||
|
||
// TestWSHandshakeReplayCarriesTurnID 钉住 WS 握手 `history` 回放这一面：它与读
|
||
// 接口共用同一份映射，因此客户端刚连上就能按 turnId 恢复回复侧的分组小标。
|
||
func TestWSHandshakeReplayCarriesTurnID(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
_, ts, jwt, st := newHistoryEnv(t)
|
||
if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleUser, Content: "回放里的提问", TurnID: "turn-replay", CreatedAt: 1000}); err != nil {
|
||
t.Fatalf("seed user: %v", err)
|
||
}
|
||
if err := st.Append(ctx, history.Message{SessionID: history.MainSessionID, Role: history.RoleAssistant, Content: "回放里的回答", TurnID: "turn-replay", CreatedAt: 1500}); err != nil {
|
||
t.Fatalf("seed reply: %v", err)
|
||
}
|
||
|
||
conn := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner))
|
||
defer conn.Close()
|
||
_, raw, err := conn.ReadMessage()
|
||
if err != nil {
|
||
t.Fatalf("read handshake frame: %v", err)
|
||
}
|
||
var frame struct {
|
||
Type string `json:"type"`
|
||
Data struct {
|
||
Messages []map[string]any `json:"messages"`
|
||
} `json:"data"`
|
||
}
|
||
if err := json.Unmarshal(raw, &frame); err != nil {
|
||
t.Fatalf("decode frame %s: %v", raw, err)
|
||
}
|
||
if frame.Type != "history" {
|
||
t.Fatalf("first frame = %q, want history (%s)", frame.Type, raw)
|
||
}
|
||
if len(frame.Data.Messages) != 2 {
|
||
t.Fatalf("replayed %d messages, want 2 (%s)", len(frame.Data.Messages), raw)
|
||
}
|
||
for _, m := range frame.Data.Messages {
|
||
if m["turnId"] != "turn-replay" {
|
||
t.Fatalf("回放消息 = %v, want turnId=turn-replay", m)
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestPersistTurnCarriesTurnIDOnBothRows 锁定第一段写入：新起的一路的用户消息与
|
||
// 它的助手回复带**同一个** turn_id，且 `message_persisted` 那一帧（由同一条已存
|
||
// 消息构造）也带着它 —— 客户端因此不需要等一次历史拉取就能配对。
|
||
func TestPersistTurnCarriesTurnIDOnBothRows(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
_, _, _, st := newHistoryEnv(t)
|
||
var broadcast []history.Message
|
||
o := &Orchestrator{history: st, log: testLogger(), onPersisted: func(m history.Message) { broadcast = append(broadcast, m) }}
|
||
|
||
legCtx := withTurnID(ctx, "turn-solo")
|
||
o.persistUserMessages(legCtx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "单路的问题")}, "web:test", "chat", "cyrene")
|
||
o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "单路的回答"}, "web:test", "chat", "cyrene", "turn-solo")
|
||
|
||
msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
turns := turnsOf(msgs)
|
||
if turns["单路的问题"] != "turn-solo" || turns["单路的回答"] != "turn-solo" {
|
||
t.Fatalf("两条 = %q / %q，want 都是 turn-solo（同一路两条同值）", turns["单路的问题"], turns["单路的回答"])
|
||
}
|
||
if len(broadcast) != 2 || broadcast[0].TurnID != "turn-solo" || broadcast[1].TurnID != "turn-solo" {
|
||
t.Fatalf("message_persisted 载荷 = %+v, want 两条都带 turn-solo", broadcast)
|
||
}
|
||
}
|
||
|
||
// TestPersistParallelLegsCarryDistinctTurnIDs 锁定并行两路：每一路的两条消息
|
||
// 同值，两路之间不同值 —— 这正是「回复 ↔ 提问」精确配对的前提（按顺序渲染会
|
||
// 把两条交错的回答配错）。
|
||
func TestPersistParallelLegsCarryDistinctTurnIDs(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
_, _, _, st := newHistoryEnv(t)
|
||
o := &Orchestrator{history: st, log: testLogger()}
|
||
|
||
o.persistUserMessages(withTurnID(ctx, "turn-1"), history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "第一路的问题")}, "web:test", "chat", "cyrene")
|
||
o.persistUserMessages(withTurnID(ctx, "turn-2"), history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "第二路的问题")}, "web:test", "chat", "cyrene")
|
||
o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "第一路的回答"}, "web:test", "chat", "cyrene", "turn-1")
|
||
o.persistAssistantReply(history.MainSessionID, &loop.Result{Text: "第二路的回答"}, "web:test", "chat", "cyrene", "turn-2")
|
||
|
||
msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
turns := turnsOf(msgs)
|
||
for content, want := range map[string]string{
|
||
"第一路的问题": "turn-1",
|
||
"第一路的回答": "turn-1",
|
||
"第二路的问题": "turn-2",
|
||
"第二路的回答": "turn-2",
|
||
} {
|
||
if got := turns[content]; got != want {
|
||
t.Fatalf("%q 的 turn_id = %q, want %q（两路各自两条同值、两路不同值）", content, got, want)
|
||
}
|
||
}
|
||
}
|
||
|
||
// TestAssignTurnIDStampsQueuedAndSteeredRows 锁定第二段写入与它的归属口径：忙时
|
||
// 入队（以及被转向吃进）的消息在落库那一刻还不知道自己属于哪一路，由**实际处理
|
||
// 它的那一路**补写；已经带上标识的行不被后来者改写（先到先得，一行只属于一路）。
|
||
func TestAssignTurnIDStampsQueuedAndSteeredRows(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
_, _, _, st := newHistoryEnv(t)
|
||
o := &Orchestrator{history: st, log: testLogger()}
|
||
|
||
// 忙时入队：落库时 ctx 上没有回合标识 ⇒ 列里留空。
|
||
queued := o.persistUserMessages(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "排队的一句")}, "web:test", "chat", "cyrene")
|
||
// 被转向吃进的一句：同样在它自己的那一轮之前就落了库。
|
||
steered := o.persistUserMessages(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "被转向吃进的一句")}, "web:test", "chat", "cyrene")
|
||
if len(queued) != 1 || len(steered) != 1 {
|
||
t.Fatalf("落库条数 = %d / %d, want 各 1", len(queued), len(steered))
|
||
}
|
||
before, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
for _, m := range before {
|
||
if m.TurnID != "" {
|
||
t.Fatalf("消息 %q 在补写之前就有 turn_id %q，want 空", m.Content, m.TurnID)
|
||
}
|
||
}
|
||
|
||
// 这两条最终由同一路回答（转向把它并进了在飞的那一路）⇒ 补写同一个标识。
|
||
o.assignTurnID(history.MainSessionID, "turn-inflight", append(queued, steered...))
|
||
after, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list after assign: %v", err)
|
||
}
|
||
for _, m := range after {
|
||
if m.TurnID != "turn-inflight" {
|
||
t.Fatalf("消息 %q 的 turn_id = %q, want turn-inflight（按实际处理它的那一路写）", m.Content, m.TurnID)
|
||
}
|
||
}
|
||
|
||
// 另一路（例如续跑）试图认领同一行：不得改写既有的归属。
|
||
o.assignTurnID(history.MainSessionID, "turn-later", queued)
|
||
again, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list after re-assign: %v", err)
|
||
}
|
||
for _, m := range again {
|
||
if m.TurnID != "turn-inflight" {
|
||
t.Fatalf("消息 %q 的 turn_id 被改写成 %q, want 仍是 turn-inflight（先到先得）", m.Content, m.TurnID)
|
||
}
|
||
}
|
||
|
||
// 空标识 / 空列表是空操作（未接线的旧路径与「这一路没有落库任何消息」）。
|
||
o.assignTurnID(history.MainSessionID, "", queued)
|
||
o.assignTurnID(history.MainSessionID, "turn-x", nil)
|
||
}
|
||
|
||
// TestTurnIDOfLegResolvesInFlightLeg 钉住补写路径要用的那个查询：在途路的 gen →
|
||
// 该路的权威 turnId；不在途（被打断 / 已被取代 / 会话里没有它）⇒ 空串。
|
||
func TestTurnIDOfLegResolvesInFlightLeg(t *testing.T) {
|
||
t.Parallel()
|
||
o := &Orchestrator{log: testLogger(), states: map[string]*sessionState{
|
||
history.MainSessionID: {legs: map[uint64]*turnLeg{7: {turnID: "turn-7"}, 8: {turnID: "turn-8"}}},
|
||
}}
|
||
if got := o.turnIDOfLeg(history.MainSessionID, 8); got != "turn-8" {
|
||
t.Fatalf("turnIDOfLeg(8) = %q, want turn-8", got)
|
||
}
|
||
if got := o.turnIDOfLeg(history.MainSessionID, 9); got != "" {
|
||
t.Fatalf("turnIDOfLeg(未知 gen) = %q, want 空串", got)
|
||
}
|
||
if got := o.turnIDOfLeg("web:unknown", 7); got != "" {
|
||
t.Fatalf("turnIDOfLeg(未知会话) = %q, want 空串", got)
|
||
}
|
||
}
|
||
|
||
// TestSubmitWiresTurnIDForBothRows 是「单路回合 ⇒ 用户消息与助手回复两条 turn_id
|
||
// 同值且非空」的**端到端**验收：真存储 + 真编排器，走 Submit → 起路 → 落库 → 结算
|
||
// 的完整链路，断言提交回报的 TurnID 与落库那两条的 turn_id 完全一致。
|
||
//
|
||
// 它需要四个写入点全部接线才可能通过（session_parallel.go 的 publishLegStart 走
|
||
// ctx 通道、session_turn_reply.go 的 persistTurnReply 由 leg.turnID 透传、
|
||
// session_orchestrator.go 的续跑分支与 session_steer.go 的 Steer 走 assignTurnID
|
||
// 补写），因此它就是那一批接线的端到端验收。
|
||
func TestSubmitWiresTurnIDForBothRows(t *testing.T) {
|
||
ctx := context.Background()
|
||
_, _, _, st := newHistoryEnv(t)
|
||
run := newGatedLegRunner()
|
||
o := NewOrchestrator(OrchestratorOptions{Run: run.Run, Log: testLogger(), History: st})
|
||
|
||
sr := o.Submit(ctx, history.MainSessionID, []model.Message{model.TextMessage(model.RoleUser, "端到端的问题")}, loop.OwnerEnv())
|
||
if !sr.Immediate || sr.TurnID == "" {
|
||
t.Fatalf("Submit = %+v, want 立刻起一路并回报非空 TurnID", sr)
|
||
}
|
||
run.waitStarted(1)
|
||
run.release(0)
|
||
waitActiveTurns(t, o, history.MainSessionID, 0)
|
||
|
||
msgs, err := st.List(ctx, history.MainSessionID, 0, 10, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
if len(msgs) != 2 {
|
||
t.Fatalf("落库 %d 条, want 2（用户消息 + 助手回复）: %+v", len(msgs), msgs)
|
||
}
|
||
turns := turnsOf(msgs)
|
||
if turns["端到端的问题"] != sr.TurnID || turns[storedParagraph("reply-1")] != sr.TurnID {
|
||
t.Fatalf("两条 turn_id = %q / %q, want 都是提交回报的 %q",
|
||
turns["端到端的问题"], turns[storedParagraph("reply-1")], sr.TurnID)
|
||
}
|
||
}
|