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）⇒ 两边共用同一份规整文本 ⇒ **历史一次性变好、两个客户端零改动**、渲染口径不分叉 （测试与变异详见切片报告）
249 lines
10 KiB
Go
249 lines
10 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"testing"
|
||
|
||
"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 出口的契约断言在 history_turnid_test.go —— 那一半走的是直接调用写入函数的
|
||
// 窄路径；这一半全部走**真编排器 + 真存储**，挡的正是「列与 DTO 都在、却没人
|
||
// 在正确的时刻写」这类缺陷。
|
||
//
|
||
// 四个写入点：
|
||
// - A 新起的一路（单路 / 并行路，publishLegStart 把 turnID 挂进 ctx ⇒ 随 INSERT 写）；
|
||
// - B 队首续跑路（assignTurnID 补写被弹出的那条排队消息 —— 它入队时还没有路）；
|
||
// - C 被 steer 吃进的消息（认领成功后 assignTurnID 补写）；
|
||
// - D 助手回复（persistTurnReply 由 leg.turnID 透传 ⇒ 随 INSERT 写）。
|
||
|
||
// turnIDRig 是真存储 + 真编排器 + 可逐路放行 runner 的三件套，使每个场景都能确定地
|
||
// 断言「哪一路落了哪两条」而不是靠睡眠赌时序。
|
||
type turnIDRig struct {
|
||
store *history.Store
|
||
run *gatedLegRunner
|
||
o *Orchestrator
|
||
}
|
||
|
||
// newTurnIDRig 装配一套夹具（并行额度 2，因此显式声明的第二路会被照准）。
|
||
func newTurnIDRig(t *testing.T) *turnIDRig {
|
||
t.Helper()
|
||
st := newHistoryStoreForGateway(t)
|
||
run := newGatedLegRunner()
|
||
o := NewOrchestrator(OrchestratorOptions{
|
||
Run: run.Run,
|
||
Log: gwTestLogger(),
|
||
History: st,
|
||
MaxParallelTurns: 2,
|
||
})
|
||
return &turnIDRig{store: st, run: run, o: o}
|
||
}
|
||
|
||
// turn 返回该内容那条消息当前的 turn_id；第二个返回值报告它是否已经落库。区分
|
||
// 「没落库」与「落了但列是空的」是刻意的：前者说明消息丢了，后者说明接线漏了。
|
||
func (r *turnIDRig) turn(t *testing.T, content string) (string, bool) {
|
||
t.Helper()
|
||
rows, err := r.store.List(context.Background(), history.MainSessionID, 0, 50, "")
|
||
if err != nil {
|
||
t.Fatalf("list: %v", err)
|
||
}
|
||
for _, m := range rows {
|
||
if m.Content == content {
|
||
return m.TurnID, true
|
||
}
|
||
}
|
||
return "", false
|
||
}
|
||
|
||
// assertTurns 按内容断言一组消息的 turn_id；缺行即失败（少了那一行说明这条消息
|
||
// 根本没落库，那比 turn_id 错了更严重）。
|
||
func (r *turnIDRig) assertTurns(t *testing.T, want map[string]string) {
|
||
t.Helper()
|
||
for content, wantTurn := range want {
|
||
got, ok := r.turn(t, content)
|
||
if !ok {
|
||
t.Fatalf("消息 %q 没有落库（want turn_id %q）", content, wantTurn)
|
||
}
|
||
if got != wantTurn {
|
||
t.Fatalf("消息 %q 的 turn_id = %q, want %q", content, got, wantTurn)
|
||
}
|
||
}
|
||
}
|
||
|
||
// waitPersisted 等到某条内容的消息真的落库（列表读取是异步之后的读，因此先等它
|
||
// 出现，再谈它的 turn_id）。
|
||
func (r *turnIDRig) waitPersisted(t *testing.T, content string) {
|
||
t.Helper()
|
||
waitUntil(t, "消息 "+content+" 落库", func() bool {
|
||
_, ok := r.turn(t, content)
|
||
return ok
|
||
})
|
||
}
|
||
|
||
// turnIDTextRunner 是「被取消后仍返回一段真实正文」的 runner：真实 loop 的回合在
|
||
// 被打断前可能已经产出可见正文，而「内容是真的」这条口径要求它照旧落库（design/15
|
||
// §3）—— 本用例正是要证明这种收尾不会把 turn_id 丢掉。
|
||
type turnIDTextRunner struct {
|
||
started chan struct{}
|
||
release chan struct{}
|
||
}
|
||
|
||
func newTurnIDTextRunner() *turnIDTextRunner {
|
||
return &turnIDTextRunner{started: make(chan struct{}, 1), release: make(chan struct{})}
|
||
}
|
||
|
||
func (r *turnIDTextRunner) Run(_ context.Context, _ string, _ []model.Message, _ loop.SessionEnv) (*loop.Result, error) {
|
||
r.started <- struct{}{}
|
||
<-r.release
|
||
return &loop.Result{Text: "取消前已经产出的正文", Steps: 1, StopReason: "stop"}, nil
|
||
}
|
||
|
||
// TestAssistantTurnIDSurvivesInterruptAll 钉住 D 的**取值来源**：assistant 行的
|
||
// turn_id 来自 `leg.turnID` 的透传，而不是按世代号回查 `turnIDOfLeg`。判据是这条
|
||
// 真实存在的窄口 —— InterruptScope(all) 会把 st.legs 整个清掉，而被打断那一路已经
|
||
// 产出的正文仍要落库；那一刻按 gen 回查只会拿到空串（配对就此丢掉），透传则不受影响。
|
||
func TestAssistantTurnIDSurvivesInterruptAll(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
st := newHistoryStoreForGateway(t)
|
||
runner := newTurnIDTextRunner()
|
||
o := NewOrchestrator(OrchestratorOptions{Run: runner.Run, Log: gwTestLogger(), History: st})
|
||
rig := &turnIDRig{store: st, o: o}
|
||
|
||
sr := o.Submit(ctx, history.MainSessionID, []model.Message{userMsg("被中断的问题")}, loop.OwnerEnv())
|
||
if !sr.Immediate || sr.TurnID == "" {
|
||
t.Fatalf("Submit = %+v, want Immediate 且带 turnId", sr)
|
||
}
|
||
<-runner.started
|
||
// scope=all：清空在途路并把会话报成空闲 —— 这条路从此按 gen 再也查不到。
|
||
if dropped := o.InterruptScope(history.MainSessionID, InterruptScopeAll); dropped != 0 {
|
||
t.Fatalf("dropped = %d, want 0（队列本来是空的）", dropped)
|
||
}
|
||
close(runner.release)
|
||
waitUntil(t, "被中断路的正文落库", func() bool {
|
||
_, ok := rig.turn(t, storedParagraph("取消前已经产出的正文"))
|
||
return ok
|
||
})
|
||
rig.assertTurns(t, map[string]string{
|
||
"被中断的问题": sr.TurnID,
|
||
storedParagraph("取消前已经产出的正文"): sr.TurnID,
|
||
})
|
||
}
|
||
|
||
// TestOrchestratorWiresTurnIDForParallelLegs 钉住②：并行两路各自的两条（用户消息 +
|
||
// 助手回复）**同值**，两条路之间**不同值**。它是「回复 ↔ 提问」精确配对的前提 ——
|
||
// 按顺序渲染会把两条交错的回答配错。
|
||
func TestOrchestratorWiresTurnIDForParallelLegs(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
rig := newTurnIDRig(t)
|
||
|
||
first := rig.o.Submit(ctx, history.MainSessionID, []model.Message{userMsg("第一路的问题")}, loop.OwnerEnv())
|
||
if !first.Immediate || first.TurnID == "" {
|
||
t.Fatalf("第一路 = %+v, want Immediate 且带 turnId", first)
|
||
}
|
||
rig.run.waitStarted(1)
|
||
second := rig.o.Submit(parallelCtx(), history.MainSessionID, []model.Message{userMsg("第二路的问题")}, loop.OwnerEnv())
|
||
if !second.Parallel || second.TurnID == "" {
|
||
t.Fatalf("第二路 = %+v, want 并行且带 turnId", second)
|
||
}
|
||
rig.run.waitStarted(2)
|
||
if first.TurnID == second.TurnID {
|
||
t.Fatalf("两路 turnId 相同（%q）：并行两路必须各自不同", first.TurnID)
|
||
}
|
||
|
||
rig.run.release(1)
|
||
rig.run.release(0)
|
||
waitActiveTurns(t, rig.o, history.MainSessionID, 0)
|
||
rig.assertTurns(t, map[string]string{
|
||
"第一路的问题": first.TurnID,
|
||
storedParagraph("reply-1"): first.TurnID,
|
||
"第二路的问题": second.TurnID,
|
||
storedParagraph("reply-2"): second.TurnID,
|
||
})
|
||
}
|
||
|
||
// TestOrchestratorWiresTurnIDForResumedLeg 钉住③（写入点 B）：忙时入队的那条消息在
|
||
// 落库那一刻**还没有路**（列里必须先是空的），轮到它续跑时由那条新路补写 —— 它与
|
||
// 该路的助手回复同值，且与上一路不同值。
|
||
func TestOrchestratorWiresTurnIDForResumedLeg(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
rig := newTurnIDRig(t)
|
||
|
||
first := rig.o.Submit(ctx, history.MainSessionID, []model.Message{userMsg("先来的")}, loop.OwnerEnv())
|
||
if !first.Immediate {
|
||
t.Fatalf("第一条 = %+v, want Immediate", first)
|
||
}
|
||
rig.run.waitStarted(1)
|
||
if sr := rig.o.Submit(ctx, history.MainSessionID, []model.Message{userMsg("排队的那条")}, loop.OwnerEnv()); !sr.Pending {
|
||
t.Fatalf("忙时第二条 = %+v, want Pending", sr)
|
||
}
|
||
rig.waitPersisted(t, "排队的那条")
|
||
if got, _ := rig.turn(t, "排队的那条"); got != "" {
|
||
t.Fatalf("排队消息在续跑之前就有 turn_id %q，want 空（那一刻它还没有路，因此本用例才谈得上「补写」）", got)
|
||
}
|
||
|
||
rig.run.release(0)
|
||
calls := rig.run.waitStarted(2)
|
||
resumeTurn := loop.TurnIDFrom(calls[1].ctx)
|
||
if resumeTurn == "" || resumeTurn == first.TurnID {
|
||
t.Fatalf("续跑路 turnId = %q（第一路 %q）：必须是非空的新路", resumeTurn, first.TurnID)
|
||
}
|
||
rig.run.release(1)
|
||
waitActiveTurns(t, rig.o, history.MainSessionID, 0)
|
||
|
||
rig.assertTurns(t, map[string]string{
|
||
"先来的": first.TurnID,
|
||
storedParagraph("reply-1"): first.TurnID,
|
||
"排队的那条": resumeTurn,
|
||
storedParagraph("reply-2"): resumeTurn,
|
||
})
|
||
}
|
||
|
||
// TestOrchestratorWiresTurnIDForSteeredRow 钉住④（写入点 C）：被转向吃进的消息同样
|
||
// 在落库时没有归属（它还在 pending），认领成功之后立刻补写吃进它的那一路的标识 ——
|
||
// 灰态气泡因此能与**正确那一路**的回复配到一起。
|
||
func TestOrchestratorWiresTurnIDForSteeredRow(t *testing.T) {
|
||
t.Parallel()
|
||
ctx := context.Background()
|
||
rig := newTurnIDRig(t)
|
||
|
||
first := rig.o.Submit(ctx, history.MainSessionID, []model.Message{userMsg("在飞的问题")}, loop.OwnerEnv())
|
||
if !first.Immediate || first.TurnID == "" {
|
||
t.Fatalf("第一条 = %+v, want Immediate 且带 turnId", first)
|
||
}
|
||
calls := rig.run.waitStarted(1)
|
||
if sr := rig.o.Submit(ctx, history.MainSessionID, []model.Message{userMsg("被吃进的那条")}, loop.OwnerEnv()); !sr.Pending {
|
||
t.Fatalf("忙时第二条 = %+v, want Pending", sr)
|
||
}
|
||
rig.waitPersisted(t, "被吃进的那条")
|
||
if got, _ := rig.turn(t, "被吃进的那条"); got != "" {
|
||
t.Fatalf("被吃进的消息在补写之前就有 turn_id %q，want 空（那时它还在 pending）", got)
|
||
}
|
||
|
||
// 用**该路自己的转向源**驱动一次回合内边界（与 loop 在边界上调用的是同一个对象）。
|
||
src := loop.SteerSourceFrom(calls[0].ctx)
|
||
if src == nil {
|
||
t.Fatal("本路的 ctx 上没有挂转向源")
|
||
}
|
||
if got := src.Steer(calls[0].ctx, loop.SteerRequest{SessionID: history.MainSessionID, Step: 2}); len(got) != 1 {
|
||
t.Fatalf("注入 = %d 条, want 1（被吃进的那条）", len(got))
|
||
}
|
||
// 补写就在认领之后同步发生，因此这一刻它已经带上本路的标识。
|
||
if got, _ := rig.turn(t, "被吃进的那条"); got != first.TurnID {
|
||
t.Fatalf("被吃进的消息 turn_id = %q, want %q（吃进它的那一路）", got, first.TurnID)
|
||
}
|
||
|
||
rig.run.release(0)
|
||
waitActiveTurns(t, rig.o, history.MainSessionID, 0)
|
||
rig.assertTurns(t, map[string]string{
|
||
"在飞的问题": first.TurnID,
|
||
storedParagraph("reply-1"): first.TurnID,
|
||
"被吃进的那条": first.TurnID,
|
||
})
|
||
}
|