diff --git a/internal/services/gateway/history_turnid_test.go b/internal/services/gateway/history_turnid_test.go index eaaa899..077a33a 100644 --- a/internal/services/gateway/history_turnid_test.go +++ b/internal/services/gateway/history_turnid_test.go @@ -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() diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index ad59049..47792ac 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -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 } diff --git a/internal/services/gateway/session_steer.go b/internal/services/gateway/session_steer.go index 38b10bc..8a71475 100644 --- a/internal/services/gateway/session_steer.go +++ b/internal/services/gateway/session_steer.go @@ -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。 diff --git a/internal/services/gateway/session_turn_reply.go b/internal/services/gateway/session_turn_reply.go index 68d0d6d..05c9a67 100644 --- a/internal/services/gateway/session_turn_reply.go +++ b/internal/services/gateway/session_turn_reply.go @@ -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 } diff --git a/internal/services/gateway/session_turnid_wiring_test.go b/internal/services/gateway/session_turnid_wiring_test.go new file mode 100644 index 0000000..2ce18e1 --- /dev/null +++ b/internal/services/gateway/session_turnid_wiring_test.go @@ -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, + }) +}