feat(gateway): 接线 chat_messages.turn_id 写入(A–D 四点齐)
A 新起一路:publishLegStart 在 persistUserMessages 之前 withTurnID(ctx, leg.turnID)
(该行随 ef3dcb6 已进 HEAD)。
B 队首续跑路:completeTurn 在 o.mu.Unlock() 之后 assignTurnID(sessionID, runLeg.turnID, next.persisted)。
C 被 steer 吃进:consumePending 认领成功后 assignTurnID(sessionID, s.turnID, steerRows(candidates))。
D 助手回复:persistTurnReply 增 turnID 形参,completeTurn 传 leg.turnID —— 刻意**不用** turnIDOfLeg:
interrupt scope=all 会清空 st.legs,而被打断那一路已产出的正文仍要照原样落库(内容是真的),
按 gen 回查只能拿到空串 ⇒ 配对丢失。
删掉 history_turnid_test.go 的 t.Skip:TestSubmitWiresTurnIDForBothRows 真跑并通过。
新增 session_turnid_wiring_test.go(真 SQLite 内存存储 + 真编排器 + 可逐路放行 runner):
单路两条同值 / 并行两路各自同值且两两不同 / 续跑补写(先钉「补写前为空」)/ steer 补写(用该路自己的转向源)/
TestAssistantTurnIDSurvivesInterruptAll(钉住 D 的取值来源)。
变异 5 条(A/B/C/D 各自 + D′ 改回 turnIDOfLeg —— 只有那条守门用例红,证明它守的正是这个取舍)。
诚实边界:turnIDOfLeg 现无生产调用方(保留为兜底);先到先得竞态未构造对抗用例;
IM 渠道桥不走本编排器 ⇒ 其行 turn_id 留空属预期(契约里键缺席)。
This commit is contained in:
1 parent
3589f4b14f
commit
d804c4e675
5 files changed
+281
-10
No files matched your search
@@ -302,13 +302,11 @@ func TestTurnIDOfLegResolvesInFlightLeg(t *testing.T) {
|
||||
// 同值且非空」的**端到端**验收:真存储 + 真编排器,走 Submit → 起路 → 落库 → 结算
|
||||
// 的完整链路,断言提交回报的 TurnID 与落库那两条的 turn_id 完全一致。
|
||||
//
|
||||
// ⏳ 它今天被 Skip 挡住:写入点必须在四个「本子代理硬约束禁止改动」的文件里接线
|
||||
// (session_parallel.go 的 publishLegStart、session_orchestrator.go 的续跑分支、
|
||||
// session_steer.go 的 Steer、session_turn_reply.go 的 persistTurnReply)。接线完成后
|
||||
// 删掉下面这行 Skip,本用例即成为该契约的端到端验收。
|
||||
// 它需要四个写入点全部接线才可能通过(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) {
|
||||
t.Skip("待接线:编排器的四个写入点尚未把 TurnID 交给落库(见交接报告),本用例是接线后的端到端验收")
|
||||
|
||||
ctx := context.Background()
|
||||
_, _, _, st := newHistoryEnv(t)
|
||||
run := newGatedLegRunner()
|
||||
|
||||
@@ -388,7 +388,7 @@ func (o *Orchestrator) goTurn(sessionID string, leg *turnLeg, messages []model.M
|
||||
// 额度);只有在途路全空且队列也空时才报空闲 —— 单路时该条件恒等价于现状。
|
||||
func (o *Orchestrator) completeTurn(sessionID string, leg *turnLeg, res *loop.Result, runErr error, deviceID, mode, personaID string) error {
|
||||
gen := leg.gen
|
||||
res = o.persistTurnReply(sessionID, gen, res, runErr, deviceID, mode, personaID)
|
||||
res = o.persistTurnReply(sessionID, gen, res, runErr, deviceID, mode, personaID, leg.turnID)
|
||||
o.persistDocumentStatus(sessionID, gen, res)
|
||||
o.dispatchVoiceReply(sessionID, gen, res)
|
||||
o.mu.Lock()
|
||||
@@ -428,6 +428,10 @@ func (o *Orchestrator) completeTurn(sessionID string, leg *turnLeg, res *loop.Re
|
||||
// 归属消息就是刚从队首弹出的那条(它在入队时就已落库,id 取自
|
||||
// next.persisted)——没有这一帧,客户端就不知道为什么气泡突然开始回答了。
|
||||
o.emitTurnBegin(sessionID, runLeg, userMessageIDs(next.persisted), activeTurns)
|
||||
// 续跑路刚起,就把刚从队首弹出的那几条消息认领到它名下(design/15 §2.5.9
|
||||
// 的两段式第二段):它们入队那一刻就落了库,而那时还没有路,此刻才谈得上
|
||||
// 归属 —— 补写之后它们与该路的助手回复同值。必须在锁外(它写库)。
|
||||
o.assignTurnID(sessionID, runLeg.turnID, next.persisted)
|
||||
go o.goTurn(sessionID, runLeg, resume, env, runCtx, cancel, next.deviceID, next.mode)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -113,6 +113,10 @@ func (s *steerTurn) Steer(ctx context.Context, req loop.SteerRequest) []model.Me
|
||||
if !o.consumePending(sessionID, s.gen, candidateSeqs(candidates)) {
|
||||
return nil
|
||||
}
|
||||
// 被吃进的消息在落库那一刻还不知道属于哪一路(那时它还在排队),因此认领成功
|
||||
// 之后立刻按本路补写 turn_id(两段式第二段,design/15 §2.5.9):它与本路的
|
||||
// 助手回复同值,客户端才能把灰态气泡与正确的回复配到一起。
|
||||
o.assignTurnID(sessionID, s.turnID, steerRows(candidates))
|
||||
out := make([]model.Message, 0, len(candidates))
|
||||
for _, c := range candidates {
|
||||
s.consumed[c.row.ID] = true
|
||||
@@ -195,6 +199,17 @@ func candidateSeqs(cs []steerCandidate) map[uint64]bool {
|
||||
return seqs
|
||||
}
|
||||
|
||||
// steerRows 把候选消息收成它们**已落库的转录行**,供按路补写 turn_id 用:这些行被
|
||||
// 吃进之前都是以「还不知道属于哪一路」的状态落库的(那时列里是空的),因此这一刻
|
||||
// 才谈得上给它们一个归属。
|
||||
func steerRows(cs []steerCandidate) []history.Message {
|
||||
rows := make([]history.Message, 0, len(cs))
|
||||
for _, c := range cs {
|
||||
rows = append(rows, c.row)
|
||||
}
|
||||
return rows
|
||||
}
|
||||
|
||||
// pendingSnapshot 在状态锁下取该会话待处理队列的一份副本(世代不符时返回 nil:
|
||||
// 那条路已失效,例如被打断、被取代或会话已清空)。取副本而不是
|
||||
// 持锁做转录读取,是为了不让有界存储读取堵住 Submit。
|
||||
|
||||
@@ -62,14 +62,20 @@ func (o *Orchestrator) turnIsCurrent(sessionID string, gen uint64) bool {
|
||||
// 带着这条说明 —— 队列里的下一条消息不会看到一个「什么都没说」的上下文。res 为
|
||||
// nil(回合报错)时它新建一个只带文案的 Result,调用方的 nil 语义不变(异步路径
|
||||
// 丢弃返回值,同步路径在此之前已按错误返回)。
|
||||
func (o *Orchestrator) persistTurnReply(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID, mode, personaID string) *loop.Result {
|
||||
//
|
||||
// turnID 是本路的权威回合标识(design/15 §2.5.9),由调用方**透传**下来而不是在这里
|
||||
// 按世代号回查:InterruptScope(all) 会把 st.legs 整个清掉,而被打断那一路的助手文本
|
||||
// 仍要照原样落库(内容是真的),那时回查只能拿到空串、把这一行的配对丢掉。调用方
|
||||
// 手里就有 `*turnLeg`,它的 turnID 出生后不可变,因此透传是唯一不会漏标的取法。
|
||||
// 空串等价于加入该列之前的行为(列里留空)。
|
||||
func (o *Orchestrator) persistTurnReply(sessionID string, gen uint64, res *loop.Result, runErr error, deviceID, mode, personaID, turnID string) *loop.Result {
|
||||
if o.history == nil {
|
||||
return res
|
||||
}
|
||||
text := turnreply.Text(res)
|
||||
if res != nil && res.Text == text {
|
||||
// 真实文本:逐字落库,行为与加入本保证之前完全一致。
|
||||
o.persistAssistantReply(sessionID, res, deviceID, mode, personaID)
|
||||
o.persistAssistantReply(sessionID, res, deviceID, mode, personaID, turnID)
|
||||
return res
|
||||
}
|
||||
if errors.Is(runErr, context.Canceled) || !o.turnIsCurrent(sessionID, gen) {
|
||||
@@ -79,6 +85,6 @@ func (o *Orchestrator) persistTurnReply(sessionID string, gen uint64, res *loop.
|
||||
res = &loop.Result{}
|
||||
}
|
||||
res.Text = text
|
||||
o.persistAssistantReply(sessionID, res, deviceID, mode, personaID)
|
||||
o.persistAssistantReply(sessionID, res, deviceID, mode, personaID, turnID)
|
||||
return res
|
||||
}
|
||||
@@ -0,0 +1,248 @@
|
||||
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, "取消前已经产出的正文")
|
||||
return ok
|
||||
})
|
||||
rig.assertTurns(t, map[string]string{
|
||||
"被中断的问题": sr.TurnID,
|
||||
"取消前已经产出的正文": 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,
|
||||
"reply-1": first.TurnID,
|
||||
"第二路的问题": second.TurnID,
|
||||
"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,
|
||||
"reply-1": first.TurnID,
|
||||
"排队的那条": resumeTurn,
|
||||
"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,
|
||||
"reply-1": first.TurnID,
|
||||
"被吃进的那条": first.TurnID,
|
||||
})
|
||||
}
|
||||
Reference in new issue
Block a user