diff --git a/internal/services/gateway/session_events.go b/internal/services/gateway/session_events.go index 3cf8b3c..465aa45 100644 --- a/internal/services/gateway/session_events.go +++ b/internal/services/gateway/session_events.go @@ -25,4 +25,16 @@ const ( // 迁移,因此不在 design/11 §7.1 的「状态信号」白名单里 —— 帧门默认只把它 // 投给 owner/admin(fail-closed)。 EventChatDuplicate = "chat_duplicate" + // EventTurnBegin 是「编排器刚起了这一路」的通知(design/15 §2.5.9、 + // 忙时并行规范草案 §2.4):帧壳带该路的权威 turnId,data 带 + // {kind, messageIds, leg, activeTurns},把「这一路 ↔ 它归属的用户消息 id」 + // 钉死。客户端靠它(而不是靠猜)把这些气泡置为「已受理·回答中」,并按 + // 帧壳 turnId 与会话 id 做**两级分桶** —— 真并行下两路的 message.delta + // 交错到达,只按 sessionId 分桶会把两条回复糊成一条。 + // + // 它**携带具体消息身份**,因此**不进** frameReachesAll 的「状态机信号」 + // 白名单,落 design/11 §7.1 的 fail-closed 默认档:只发 owner + admin。 + // 两条路径都发(单路 kind:"turn"、并行照准 kind:"parallel"),因为串行 + // 路径同样受益于「消息 ↔ 回合」绑定。 + EventTurnBegin = "turn_begin" ) diff --git a/internal/services/gateway/session_interrupt.go b/internal/services/gateway/session_interrupt.go index 1e1672e..911d7c1 100644 --- a/internal/services/gateway/session_interrupt.go +++ b/internal/services/gateway/session_interrupt.go @@ -56,6 +56,33 @@ func (o *Orchestrator) InterruptScope(sessionID string, scope InterruptScope) in return o.interruptLocked(sessionID, o.states[sessionID], scope) } +// InterruptTurn 只取消 `turnId` 指定的那**一路**(忙时并行规范草案 §2.3: +// `interrupt{scope:"current", turnId}`),其余在途路与待处理队列一概不受影响 —— +// 每路各有自己的取消链,因此停一路绝不牵动另一路已发起的工具(I3′)。 +// +// 返回 false 表示该会话此刻没有这条路(不存在 / 已收尾 / 已被取消)。调用方据此 +// 定向回一条显式错误帧,**绝不**静默按「全部在途路」处理:把「只停这一路」悄悄 +// 执行成「停掉所有路」,正是这类定向能力最不该有的失败方式。 +// +// 它与 InterruptScope(Current) 的差别只有「目标集」:没有目标就是每一路(现状), +// 有目标就只取消它。帧与队列口径完全一致(dropped 恒为 0、mode 如实报告当时的 +// 忙碌通道、被取消的路仍在 st.legs 里收尾到自然结算)。 +func (o *Orchestrator) InterruptTurn(sessionID, turnID string) bool { + if turnID == "" { + return false + } + sessionID = o.normalize(sessionID) + o.mu.Lock() + defer o.mu.Unlock() + st := o.states[sessionID] + if o.legByTurnIDLocked(st, turnID) == nil { + return false + } + canceled := o.cancelLegsLocked(st, turnID) + o.emitState(sessionID, EventSessionInterrupt, interruptPayload(sessionID, st.mode, 0, canceled, len(st.legs))) + return true +} + // InterruptAll 取消每一个在飞会话并清空所有待处理队列。它在网关关闭时使用, // 使没有后台回合比 hub 活得更久。它返回被丢弃的待处理回合总数。 func (o *Orchestrator) InterruptAll() int { @@ -73,15 +100,16 @@ func (o *Orchestrator) InterruptAll() int { // Current 只取消在途路,队列与路登记原样留下,使被取消回合的结算仍会续跑队列。 // 它返回真的被丢弃的待处理回合数(Current 恒为 0)。 // -// 多路口径(design/15 §2.5.4):无论哪个 scope,被取消的都是**当前全部在途路** -// —— 省略 turnId 的 scope 就是「作用于每一路」的意思(精确停某一路是后续切片的 -// `interrupt.turnId`,本切片不做)。解除在途登记的方式与旧口径的 `st.gen++` 等价: -// 把路从 st.legs 里摘掉,被取消路的迟到结算因此找不到自己的路、成为空操作。 +// 多路口径(design/15 §2.5.4):这里取消的是**当前全部在途路**(目标传空串) +// —— 省略 turnId 的 scope 就是「作用于每一路」的意思;「只停某一路」走 +// InterruptTurn(同一套取消链,只是带上目标)。解除在途登记的方式与旧口径的 +// `st.gen++` 等价:把路从 st.legs 里摘掉,被取消路的迟到结算因此找不到自己的 +// 路、成为空操作。 func (o *Orchestrator) interruptLocked(sessionID string, st *sessionState, scope InterruptScope) int { if st == nil || st.mode == ModeIdle { return 0 } - canceled := o.cancelLegsLocked(st) + canceled := o.cancelLegsLocked(st, "") if scope == InterruptScopeCurrent { // 只取消在途路:不碰 pending、不解除路的登记(它们仍在收尾)。mode 如实 // 报告为当时的忙碌通道(回合仍在收尾、队列还在续跑),而不是 idle。 diff --git a/internal/services/gateway/session_orchestrator.go b/internal/services/gateway/session_orchestrator.go index 112088a..ad59049 100644 --- a/internal/services/gateway/session_orchestrator.go +++ b/internal/services/gateway/session_orchestrator.go @@ -143,7 +143,7 @@ type sessionState struct { // 单路时它至多一条,且其行为与引入多路之前逐字相同;len(legs) 就是 // 忙态帧里的 activeTurns,也是「所有路都收尾了吗」的判据。 legs map[uint64]*turnLeg - // legSeq 是路序号(1 基)的分配源,供后续切片的 `turn_begin.data.leg` 使用。 + // legSeq 是路序号(1 基)的分配源,即 `turn_begin.data.leg`。 legSeq int pending []pendingMsg tail []model.Message // 用于上下文续跑的有界对话尾部 @@ -313,20 +313,21 @@ func (o *Orchestrator) submitInternal(ctx context.Context, sessionID string, mes turn := o.seedTurnLocked(ctx, st, sessionID, messages) leg, runCtx, cancel := o.beginLegLocked(ctx, st, newTurnID(), false, newTurnFence(st.fence.done)) st.fence = leg.fence + // 本次启动后的在途路数:`turn_begin.activeTurns` 与忙态帧**同源**(正在收尾的路 + // 也计入),因此它只能在锁内读一次 —— 解锁之后再读数会随另一路的收尾漂移。 + activeTurns := len(st.legs) o.mu.Unlock() // 本回合的转向源(design/15 §2.4):loop 在**回合内边界**回调它重读转录,把忙时 // 新到、已入 pending 的用户消息并入**同一回合**的下一次模型调用(I1:绝不新开 // 第二回合)。它挂在 runCtx 上,因此只作用于本回合;回合结算或被打断后随 ctx 消亡。 - runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen)) + // 它同时带上本路的 turnId,使 `steer_ack` 的帧壳能指明「被哪一路吃进」。 + runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen, leg.turnID)) - o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, ModeThinking, 1)) - // 只有此刻到达的消息会被持久化(已播种的转录已持久,重追加会重复整个对话), - // 且落库只写**引用** —— 抽取还发生在后面,结论由回合结算时补进同一列 - // (两段式见 session_document_status.go)。 - o.rememberTurnDocuments(sessionID, leg.gen, o.persistUserMessages(ctx, sessionID, messages, deviceID, msgMode, env.PersonaID)) - // 语音收尾回调同样按**路**配对(协同 C1,见 session_voice.go)。 - o.rememberVoiceTurn(sessionID, leg.gen, voiceReplySinkFrom(ctx)) + o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, ModeThinking, activeTurns)) + // 落库 → 按路配对 → 下发 `turn_begin`:与并行路的启动共用同一段,两条启动 + // 路径的帧序因此同形(见 publishLegStart)。 + o.publishLegStart(ctx, sessionID, leg, messages, deviceID, msgMode, env.PersonaID, activeTurns) if sync { return o.runImmediateSync(runCtx, sessionID, leg, turn, env, deviceID, msgMode, a.hint.MessageID) @@ -410,7 +411,7 @@ func (o *Orchestrator) completeTurn(sessionID string, leg *turnLeg, res *loop.Re // 任何在途路:围栏只负责放行它后面的那一个。 runLeg, runCtx, cancel := o.beginLegLocked(context.Background(), st, newTurnID(), false, newTurnFence(nil)) st.fence = runLeg.fence - runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, runLeg.gen)) + runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, runLeg.gen, runLeg.turnID)) // 续跑的这一轮同样有「回合前已落库的用户消息」,因此把它们的文档状态 // 记到**这一路**上(design/24 §6.1);语音回调同理(协同 C1:忙时排队 // 不许把音频丢掉)。 @@ -423,6 +424,10 @@ func (o *Orchestrator) completeTurn(sessionID string, leg *turnLeg, res *loop.Re o.mu.Unlock() o.emitState(sessionID, EventPendingDrain, map[string]any{"sessionId": sessionID, "mode": ModeThinking, "queueDepth": depth, "activeTurns": activeTurns}) + // 续跑也是一条**新路**,因此同样发一次 turn_begin(kind:"turn"):它的 + // 归属消息就是刚从队首弹出的那条(它在入队时就已落库,id 取自 + // next.persisted)——没有这一帧,客户端就不知道为什么气泡突然开始回答了。 + o.emitTurnBegin(sessionID, runLeg, userMessageIDs(next.persisted), activeTurns) go o.goTurn(sessionID, runLeg, resume, env, runCtx, cancel, next.deviceID, next.mode) return nil } diff --git a/internal/services/gateway/session_parallel.go b/internal/services/gateway/session_parallel.go index e7d526b..e16f31f 100644 --- a/internal/services/gateway/session_parallel.go +++ b/internal/services/gateway/session_parallel.go @@ -4,6 +4,7 @@ import ( "context" "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/loop" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" ) @@ -12,14 +13,17 @@ import ( // 2026-10-05 裁决 B)。它从 session_orchestrator.go 拆出,使编排器本体继续只留 // 状态机,并保持在「单文件 ≤500 行 / 函数 ≤80 行」的硬约束内。 // -// 它做四件事: +// 它做五件事: // - 解析上行的并行声明(ctx 通道,与 deviceID / mode / 幂等键同一种机制); // - 判定并发额度(默认 2,硬上限 3,<2 = 关闭真并行 ⇒ 完全回落到 steer-only 现状); // - 在状态锁下登记一条新路(自己的世代号 / turnId / 取消链 / checkpoint 归属键); -// - 忙态帧的在途路数(activeTurns)与排队原因(reason)。 +// - 忙态帧的在途路数(activeTurns)与排队原因(reason); +// - 下发新帧 `turn_begin`(`publishLegStart` / `emitTurnBegin`):单路 +// kind:"turn"、并行路 kind:"parallel",两条路径都发。`steer_ack` 的帧壳 +// turnId 由 session_steer.go 的转向游标带出,与这里**取值同源**(leg.turnID)。 // -// 它**不做**(留给后续切片):`turn_begin` 帧的发射、`steer_ack` 的帧壳 turnId、 -// 写类工具的文件级互斥、第二路的进行态 system 摘要。预留点见 turnLeg 的字段。 +// 它**不做**(留给后续切片):写类工具的文件级互斥、第二路的进行态 system 摘要。 +// 预留点见 turnLeg 的字段。 // 并发额度的取值口径(design/15 §2.5.7)。 const ( @@ -35,6 +39,11 @@ const ( pendingAckReasonParallelLimit = "parallel_limit" // parallelLogSource 是并发路相关日志的子功能来源标签。 parallelLogSource = "session-parallel" + // turn_begin 的 kind 取值(忙时并行规范草案 §2.4):服务端**实际**如何起这 + // 一路 —— 新回合 / 续跑 = "turn",显式声明并行且照准 = "parallel"。它**仅供 + // 展示**,客户端的状态机一律按「messageIds 是否含本气泡的 messageId」判定。 + turnBeginKindTurn = "turn" + turnBeginKindParallel = "parallel" ) // turnLeg 是会话内一条在途回合(design/15 §2.5 的「路」)。 @@ -47,16 +56,16 @@ const ( // goroutine 读取 turnID/gen(出生后不可变)。 type turnLeg struct { // turnID 是本路的权威回合标识(design/15 §2.5.9):编排器生成,挂进 runCtx - // 的 loop.WithTurnID,并由 loop 盖在该路全部流帧上。后续切片的 `turn_begin` - // 帧与 `steer_ack` 的帧壳 turnId 直接取它 —— 两处必须同值,否则客户端按 + // 的 loop.WithTurnID,并由 loop 盖在该路全部流帧上。`turn_begin` 帧与 + // `steer_ack` 的帧壳 turnId 都直接取它 —— 两处必须同值,否则客户端按 // turnId 两级分桶会立刻失配。 turnID string // gen 是本路的世代号:会话内单调、永不复用。它取代了旧的「会话当前代次」 // 概念 —— 陈旧判据从「st.gen == gen」变成「本路还在 st.legs 里」,因此启动 // 第二路绝不会让第一路变成陈旧。 gen uint64 - // leg 是本路的会话内序号(1 基)。它预留给后续切片的 `turn_begin.data.leg` - //(「并行路 2」的展示口径),本切片只在日志里用它。 + // leg 是本路的会话内序号(1 基),即 `turn_begin.data.leg`(「并行路 2」的 + // 展示口径)。除了发帧,日志也用它。 leg int // parallel 报告本路是否为**显式声明的并行路**:它决定 checkpoint 归属键 // (单路逐字等于 sessionID,并行路为 {sessionID}#{turnID})。 @@ -157,12 +166,13 @@ func (o *Orchestrator) startParallelLocked(ctx context.Context, st *sessionState activeTurns := len(st.legs) o.mu.Unlock() - runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen)) + runCtx = loop.WithSteerSource(runCtx, o.beginSteer(sessionID, leg.gen, leg.turnID)) o.emitState(sessionID, EventSessionBusy, busyPayload(sessionID, ModeThinking, activeTurns)) logger.Scoped(o.log, source, parallelLogSource).Info("parallel leg started", "session", sessionID, "turn", leg.turnID, "leg", leg.leg, "active_turns", activeTurns) - o.rememberTurnDocuments(sessionID, leg.gen, o.persistUserMessages(ctx, sessionID, messages, deviceID, mode, env.PersonaID)) - o.rememberVoiceTurn(sessionID, leg.gen, voiceReplySinkFrom(ctx)) + // 落库 → 按路配对 → `turn_begin{kind:"parallel"}`:messageIds 必须是刚落库的 + // 权威 id,因此这三步的顺序不能颠倒(与单路启动共用同一段)。 + o.publishLegStart(ctx, sessionID, leg, messages, deviceID, mode, env.PersonaID, activeTurns) go o.goTurn(sessionID, leg, turn, env, runCtx, cancel, deviceID, mode) return SubmitResult{Accepted: true, Immediate: true, Parallel: true, MessageID: hint.MessageID, TurnID: leg.turnID} } @@ -206,21 +216,45 @@ func (o *Orchestrator) legLocked(st *sessionState, gen uint64) *turnLeg { return st.legs[gen] } -// cancelLegsLocked 取消该会话当前**全部**在途路(design/15 §2.5.4:省略 turnId 的 -// scope 作用于每一路),返回真的取消了几路。每路各有自己的取消链,因此取消一路 -// 绝不影响另一路已发起的工具(I3′)。 +// legByTurnIDLocked 按权威 turnId 取该会话的在途路;不存在时返回 nil。turnId +// 是客户端手上的唯一目标标识(`interrupt.turnId`),因此「这条路还在吗」只能在 +// 状态锁下按它回答 —— 已收尾 / 已取消的路会被从 st.legs 摘掉,于是「不存在」 +// 与「已经停了」是同一个答案。 // // 必须在 o.mu 下调用。 -func (o *Orchestrator) cancelLegsLocked(st *sessionState) int { +func (o *Orchestrator) legByTurnIDLocked(st *sessionState, turnID string) *turnLeg { + if st == nil || turnID == "" { + return nil + } + for _, leg := range st.legs { + if leg.turnID == turnID { + return leg + } + } + return nil +} + +// cancelLegsLocked 取消该会话的在途路,返回真的取消了几路。turnID 为空 = 作用于 +// **每一路**(design/15 §2.5.4:省略 turnId 的 scope);非空 = 只作用于该 turnId +// 的那一路(忙时并行规范草案 §2.3 的目标化打断)。每路各有自己的取消链,因此取消 +// 一路绝不影响另一路已发起的工具(I3′)。 +// +// 必须在 o.mu 下调用。 +func (o *Orchestrator) cancelLegsLocked(st *sessionState, turnID string) int { if st == nil { return 0 } + canceled := 0 for _, leg := range st.legs { + if turnID != "" && leg.turnID != turnID { + continue + } if leg.cancel != nil { leg.cancel() } + canceled++ } - return len(st.legs) + return canceled } // busyPayload 组装忙态 / 空闲帧的载荷(design/15 §2.6):既有键(sessionId / mode) @@ -240,3 +274,68 @@ func pendingAckPayload(sessionID string, mode Mode, depth int, parallelLimit boo } return data } + +// legKind 返回这一路在 `turn_begin.data.kind` 里的取值(规范草案 §2.4):服务端 +// **实际**如何起它 —— 显式声明并行且照准 ⇒ "parallel",新回合 / 续跑 ⇒ "turn"。 +// 它只供展示,客户端的状态机不必依赖它。 +func legKind(leg *turnLeg) string { + if leg != nil && leg.parallel { + return turnBeginKindParallel + } + return turnBeginKindTurn +} + +// userMessageIDs 取出这一路已落库用户消息的**权威** id(客户端靠它把气泡钉到 +// 这一路)。没有落库行(未接历史存储、或该次写入失败 —— 那条失败路径已由 +// persistUserMessages 带来源记 Error)时返回**空数组**而不是 nil:契约里 +// `messageIds` 是数组,null 会让严格解码的客户端直接报错。 +func userMessageIDs(msgs []history.Message) []string { + ids := make([]string, 0, len(msgs)) + for _, m := range msgs { + if m.ID != "" { + ids = append(ids, m.ID) + } + } + return ids +} + +// publishLegStart 在**起了一路之后**统一做三件事(design/15 §2.5.9):把刚落库的 +// 用户消息按路配对(文档结论回写,design/24 §6.1)、挂本路的语音收尾回调(协同 +// C1)、并下发 `turn_begin`。顺序是刻意的:`turn_begin.data.messageIds` 必须是 +// **服务端权威的**消息 id(客户端靠它把这些气泡置为「已受理·回答中」并按 turnId +// 分桶),因此它只能在落库之后发。 +// +// 这里落库的只有**此刻到达**的那些消息(已播种的转录早就持久了,重追加会重复整个 +// 对话),且只写**引用** —— 抽取结论要等回合跑起来才有,由回合结算补进同一列 +// (两段式见 session_document_status.go)。 +// +// 单路与并行路共用这一段,于是两条启动路径的帧序完全同形 —— 客户端只需要实现一 +// 遍「turn_begin → 灰态」。 +func (o *Orchestrator) publishLegStart(ctx context.Context, sessionID string, leg *turnLeg, messages []model.Message, deviceID, mode, personaID string, activeTurns int) { + persisted := o.persistUserMessages(ctx, sessionID, messages, deviceID, mode, personaID) + o.rememberTurnDocuments(sessionID, leg.gen, persisted) + o.rememberVoiceTurn(sessionID, leg.gen, voiceReplySinkFrom(ctx)) + o.emitTurnBegin(sessionID, leg, userMessageIDs(persisted), activeTurns) +} + +// emitTurnBegin 下发一次 `turn_begin`(design/15 §2.5.9 / 规范草案 §2.4):帧壳 +// `turnId` 是该路的权威回合标识(与它全部流帧的 turnId 同值 ⇒ 客户端可两级分桶), +// data 带 {kind, messageIds, leg, activeTurns}。 +// +// `messageIds` 为空数组时**仍然发**:客户端的绑定是空操作,而「起了一路就恰好发 +// 一帧」是可数的契约(leg / activeTurns 仍有用),按有无消息 id 决定发不发反而会让 +// 帧数随存储接线而漂移。 +func (o *Orchestrator) emitTurnBegin(sessionID string, leg *turnLeg, messageIDs []string, activeTurns int) { + if leg == nil { + return + } + if messageIDs == nil { + messageIDs = []string{} + } + o.emitTurnState(sessionID, leg.turnID, EventTurnBegin, map[string]any{ + "kind": legKind(leg), + "messageIds": messageIDs, + "leg": leg.leg, + "activeTurns": activeTurns, + }) +} diff --git a/internal/services/gateway/session_parallel_frames_test.go b/internal/services/gateway/session_parallel_frames_test.go new file mode 100644 index 0000000..792c39d --- /dev/null +++ b/internal/services/gateway/session_parallel_frames_test.go @@ -0,0 +1,292 @@ +package gateway + +import ( + "context" + "log/slog" + "slices" + "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" +) + +// 本文件是 design/15 §2.5.9「帧层」的回归:`turn_begin` 的发射(每次启动一路 +// 恰好一帧,含续跑路)与 `steer_ack` 的帧壳 turnId。夹具复用 +// session_parallel_test.go(同包)的可逐路放行 runner 与记录型状态汇聚点,因此 +// 「哪一帧属于哪一路」是被观测到的事实,而不是靠睡眠赌出来的。 +// +// WS 层的对应断言(真连接上的 turn_begin 角色门、interrupt{turnId} 的端到端 +// 行为)在 session_parallel_turn_interrupt_test.go。 + +// transcriptHistory 是 fakeHistory 的转录可读变体:`Recent` 返回已落库的行。 +// 裸 fakeHistory 的 Recent 恒为空(那是 TestSteer_UnpersistedPendingStaysPending +// 刻意依赖的行为,不能改),而 steer 边界的匹配**必须**读到转录行 —— 因此本 +// 文件用嵌入的方式只覆盖这一个方法。 +type transcriptHistory struct{ *fakeHistory } + +// Recent 返回该会话最近 n 条已落库行(最旧优先),与 *history.Store 的口径一致。 +func (h *transcriptHistory) Recent(_ context.Context, sessionID string, n int) ([]history.Message, error) { + rows := h.records() + out := make([]history.Message, 0, len(rows)) + for _, m := range rows { + if m.SessionID == sessionID { + out = append(out, m) + } + } + if len(out) > n { + out = out[len(out)-n:] + } + return out, nil +} + +// frameMessageIDs 取出一帧 `turn_begin` 的 messageIds(缺席/形状不符时返回 nil)。 +func frameMessageIDs(rec turnStateRec) []string { + ids, _ := rec.data["messageIds"].([]string) + return ids +} + +// dataKeys 返回帧载荷的键(升序),用于「键集恰好是这几项」的断言 —— 多一个键 +// 就是契约变更,少一个键客户端就配不上。 +func dataKeys(data map[string]any) []string { + out := make([]string, 0, len(data)) + for k := range data { + out = append(out, k) + } + slices.Sort(out) + return out +} + +// assertTurnBeginFrame 逐项断言一帧 `turn_begin`:帧壳 turnId 是该路的权威回合 +// 标识,载荷键集与取值都按规范草案 §2.4 钉死。 +func assertTurnBeginFrame(t *testing.T, what string, rec turnStateRec, kind string, leg, activeTurns int, turnID string, messageIDs []string) { + t.Helper() + if rec.sessionID != "web" { + t.Fatalf("%s: 帧壳 sessionId = %q, want web", what, rec.sessionID) + } + if rec.typ != EventTurnBegin { + t.Fatalf("%s: 帧类型 = %q, want %q", what, rec.typ, EventTurnBegin) + } + if rec.turnID != turnID { + t.Fatalf("%s: 帧壳 turnId = %q, want %q(必须与该路的权威 turnId 同值,否则客户端两级分桶立即失配)", + what, rec.turnID, turnID) + } + wantKeys := []string{"activeTurns", "kind", "leg", "messageIds"} + if got := dataKeys(rec.data); !slices.Equal(got, wantKeys) { + t.Fatalf("%s: 载荷键集 = %v, want %v", what, got, wantKeys) + } + if rec.data["kind"] != kind { + t.Fatalf("%s: kind = %v, want %q", what, rec.data["kind"], kind) + } + if got := stateInt(rec, "leg"); got != leg { + t.Fatalf("%s: leg = %d, want %d", what, got, leg) + } + if got := stateInt(rec, "activeTurns"); got != activeTurns { + t.Fatalf("%s: activeTurns = %d, want %d", what, got, activeTurns) + } + if got := frameMessageIDs(rec); !slices.Equal(got, messageIDs) { + t.Fatalf("%s: messageIds = %v, want %v(客户端据此把这些气泡置为灰态)", what, got, messageIDs) + } +} + +// --------------------------------------------------------------------------- +// ① / 续跑:每次启动一路恰好一帧 turn_begin +// --------------------------------------------------------------------------- + +// TestTurnBeginEmittedOncePerLegStarted 钉住①与续跑路:空闲启动的第一路发 +// `turn_begin{kind:"turn", leg:1, activeTurns:1}` 且 messageIds 恰好是那条用户 +// 消息的**权威 id**;队首续跑同样发一帧(kind 仍是 "turn",但它是一条**新路**, +// turnId 与第一路不同)。两帧的帧壳 turnId 都必须等于交给 runner 的那个 id。 +func TestTurnBeginEmittedOncePerLegStarted(t *testing.T) { + t.Parallel() + f := newParallelFixture(5, 2) + + first := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv()) + if !first.Immediate || first.TurnID == "" { + t.Fatalf("第一条 = %+v, want Immediate 且带 turnId", first) + } + calls := f.run.waitStarted(1) + + frames := f.sink.ofType(EventTurnBegin) + if len(frames) != 1 { + t.Fatalf("第一路的 turn_begin 帧数 = %d, want 1(恰好一次)", len(frames)) + } + assertTurnBeginFrame(t, "第一路", frames[0], turnBeginKindTurn, 1, 1, first.TurnID, []string{"id-1"}) + if got := loop.TurnIDFrom(calls[0].ctx); got != frames[0].turnID { + t.Fatalf("交给 runner 的 turnId = %q,turn_begin 帧壳是 %q:两处必须同值", got, frames[0].turnID) + } + + // 忙时第二条(未声明并行)⇒ 入队;它的落库 id 就是续跑路的 messageIds。 + second := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv()) + if !second.Pending || second.QueueDepth != 1 { + t.Fatalf("第二条 = %+v, want Pending(1)", second) + } + f.run.release(0) + calls = f.run.waitStarted(2) + + frames = f.sink.ofType(EventTurnBegin) + if len(frames) != 2 { + t.Fatalf("turn_begin 帧数 = %d, want 2(续跑也是一条新路,必须有自己的帧)", len(frames)) + } + resumeTurnID := loop.TurnIDFrom(calls[1].ctx) + if resumeTurnID == first.TurnID { + t.Fatal("续跑路必须是新路:turnId 不能与上一路相同") + } + assertTurnBeginFrame(t, "续跑路", frames[1], turnBeginKindTurn, 2, 1, resumeTurnID, []string{"id-2"}) + + f.run.release(1) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } + if got := len(f.sink.ofType(EventTurnBegin)); got != 2 { + t.Fatalf("收尾后的 turn_begin 帧数 = %d, want 2(收尾绝不再补发一帧)", got) + } +} + +// --------------------------------------------------------------------------- +// ② 并行路:第二路有自己的 turn_begin(kind:"parallel") +// --------------------------------------------------------------------------- + +// TestTurnBeginSecondLegIsParallelKind 钉住②:显式声明并行且照准的第二路发 +// `turn_begin{kind:"parallel", leg:2, activeTurns:2}`,且它的 messageIds **只是** +// 它自己那条消息 —— 两帧各钉各的消息,客户端因此能把两条回复分到两个气泡。 +func TestTurnBeginSecondLegIsParallelKind(t *testing.T) { + t.Parallel() + f := newParallelFixture(5, 2) + + first := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv()) + f.run.waitStarted(1) + second := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv()) + if !second.Immediate || !second.Parallel { + t.Fatalf("显式并行提交 = %+v, want Immediate 且 Parallel", second) + } + calls := f.run.waitStarted(2) + waitActiveTurns(t, f.o, "web", 2) + + frames := f.sink.ofType(EventTurnBegin) + if len(frames) != 2 { + t.Fatalf("turn_begin 帧数 = %d, want 2(两条路各一帧)", len(frames)) + } + assertTurnBeginFrame(t, "第一路", frames[0], turnBeginKindTurn, 1, 1, first.TurnID, []string{"id-1"}) + assertTurnBeginFrame(t, "第二路", frames[1], turnBeginKindParallel, 2, 2, second.TurnID, []string{"id-2"}) + if got := loop.TurnIDFrom(calls[1].ctx); got != frames[1].turnID { + t.Fatalf("第二路交给 runner 的 turnId = %q,帧壳是 %q:两处必须同值", got, frames[1].turnID) + } + + f.run.release(1) + f.run.release(0) + if !waitIdle(f.o, "web") { + t.Fatal("会话没有回到空闲") + } +} + +// --------------------------------------------------------------------------- +// ③ steer_ack 的帧壳 turnId +// --------------------------------------------------------------------------- + +// TestSteerAckFrameCarriesLegTurnID 钉住③:被转向吃进的那条消息,其 +// `steer_ack` **帧壳**必须带吃进它的那一路的 turnId(真并行下客户端靠它把灰态 +// 配到正确那一路的 `turn.end`)。这里直接用该路挂在 ctx 上的转向源驱动边界 —— +// 与 loop 在回合内边界调用的是同一个对象。 +func TestSteerAckFrameCarriesLegTurnID(t *testing.T) { + t.Parallel() + run := newGatedLegRunner() + sink := &turnSink{} + o := NewOrchestrator(OrchestratorOptions{ + Run: run.Run, + Emit: sink.Emit, + EmitTurn: sink.EmitTurn, + Log: slog.New(slog.NewTextHandler(discardWriter{}, nil)), + MaxPending: 5, + MaxTail: 20, + History: &transcriptHistory{fakeHistory: &fakeHistory{}}, + MaxParallelTurns: 2, + }) + + first := o.Submit(context.Background(), "web", []model.Message{userMsg("第一条")}, loop.OwnerEnv()) + if !first.Immediate { + t.Fatalf("第一条 = %+v, want Immediate", first) + } + calls := run.waitStarted(1) + if sr := o.Submit(context.Background(), "web", []model.Message{userMsg("忙时新消息")}, loop.OwnerEnv()); !sr.Pending { + t.Fatalf("忙时第二条 = %+v, want Pending", sr) + } + + src := loop.SteerSourceFrom(calls[0].ctx) + if src == nil { + t.Fatal("本路的 ctx 上没有挂转向源") + } + steered := src.Steer(calls[0].ctx, loop.SteerRequest{SessionID: "web", Step: 2}) + if len(steered) != 1 || messageText(steered[0]) != "忙时新消息" { + t.Fatalf("注入 = %v, want [忙时新消息]", windowText(steered)) + } + + acks := sink.ofType(wsSteerAck) + if len(acks) != 1 { + t.Fatalf("steer_ack 帧数 = %d, want 1", len(acks)) + } + if acks[0].turnID != first.TurnID { + t.Fatalf("steer_ack 帧壳 turnId = %q, want %q(= 吃进它的那一路的 turnId)", acks[0].turnID, first.TurnID) + } + if acks[0].data["messageId"] != "id-2" || stateInt(acks[0], "atStep") != 2 { + t.Fatalf("steer_ack 载荷 = %v, want {messageId:id-2, atStep:2}", acks[0].data) + } + + run.release(0) + if !waitIdle(o, "web") { + t.Fatal("会话没有回到空闲") + } +} + +// --------------------------------------------------------------------------- +// ④ interrupt{turnId} 只停目标路 +// --------------------------------------------------------------------------- + +// TestInterruptTurnStopsOnlyTargetLeg 钉住④的编排器侧:`InterruptTurn` 只取消 +// 目标那一路 —— 另一路照常跑完并落库,被取消的那一路不补说明(现状语义), +// session_interrupted 如实报 cancelledTurns=1 / dropped=0,且在另一路仍在飞时 +// **不报空闲**。 +func TestInterruptTurnStopsOnlyTargetLeg(t *testing.T) { + t.Parallel() + f := newParallelFixture(5, 2) + + first := f.o.Submit(context.Background(), "web", []model.Message{userMsg("第一问")}, loop.OwnerEnv()) + if !first.Immediate || first.TurnID == "" { + t.Fatalf("第一路 = %+v, want Immediate 且带 turnId", first) + } + f.run.waitStarted(1) + second := f.o.Submit(parallelCtx(), "web", []model.Message{userMsg("第二问")}, loop.OwnerEnv()) + if !second.Parallel { + t.Fatalf("第二路 = %+v, want Parallel", second) + } + f.run.waitStarted(2) + waitActiveTurns(t, f.o, "web", 2) + + if !f.o.InterruptTurn("web", second.TurnID) { + t.Fatalf("InterruptTurn(%q) = false, want true(该路确实在途)", second.TurnID) + } + waitActiveTurns(t, f.o, "web", 1) + + intr := f.sink.ofType(EventSessionInterrupt) + if len(intr) != 1 { + t.Fatalf("session_interrupted 帧数 = %d, want 1", len(intr)) + } + if got := stateInt(intr[0], "cancelledTurns"); got != 1 { + t.Fatalf("cancelledTurns = %d, want 1(只停了一路)", got) + } + if got := stateInt(intr[0], "dropped"); got != 0 { + t.Fatalf("dropped = %d, want 0(目标化打断不碰待处理队列)", got) + } + if got := f.sink.ofType(EventSessionIdle); len(got) != 0 { + t.Fatalf("另一路仍在飞 ⇒ 不该报空闲,得到 %v", got) + } + + // 另一路(第一路)照常跑完:它的助手回复必须落库,而被取消的第二路不补说明。 + f.run.release(0) + if !waitIdle(f.o, "web") { + t.Fatal("另一路收尾后会话没有回到空闲") + } + if got := assistantContents(f.history.records()); !slices.Equal(got, []string{"reply-1"}) { + t.Fatalf("助手回复 = %v, want [reply-1](另一路继续作答,被取消路不补说明)", got) + } +} diff --git a/internal/services/gateway/session_parallel_turn_interrupt_test.go b/internal/services/gateway/session_parallel_turn_interrupt_test.go new file mode 100644 index 0000000..4ab7541 --- /dev/null +++ b/internal/services/gateway/session_parallel_turn_interrupt_test.go @@ -0,0 +1,232 @@ +package gateway + +import ( + "encoding/json" + "slices" + "testing" + + "github.com/gorilla/websocket" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" +) + +// 本文件是 design/15 §2.5.4「`interrupt` 增可选 `turnId`」在**真实 WS 路由**上的 +// 端到端回归(忙时并行规范草案 §2.3):真 chat 上行帧 → 真编排器(真 pending / +// 真额度判定)→ 真帧门 → 真 JWT 连接,只有模型适配器是脚本化的。 +// +// 两条断言: +// - 只停目标路:另一路照常收尾并落库,被取消的那一路不补说明; +// - 目标不存在:定向回 bad_request,且**一路都不动**(绝不静默降级成「停全部」)。 +// +// `turn_begin` 的帧层断言在 session_parallel_frames_test.go(编排器侧)。 + +// turnBeginWire 是 `turn_begin` 帧的线上视图(帧壳 + 载荷,与客户端解码同形)。 +type turnBeginWire struct { + Type string `json:"type"` + SessionID string `json:"sessionId"` + TurnID string `json:"turnId"` + Data struct { + Kind string `json:"kind"` + MessageIDs []string `json:"messageIds"` + Leg int `json:"leg"` + ActiveTurns int `json:"activeTurns"` + } `json:"data"` +} + +// interruptTurnWire 是 `session_interrupted` 帧的线上视图(含两个计数键)。 +type interruptTurnWire struct { + Type string `json:"type"` + SessionID string `json:"sessionId"` + Data struct { + SessionID string `json:"sessionId"` + Mode string `json:"mode"` + Dropped int `json:"dropped"` + CancelledTurns int `json:"cancelledTurns"` + ActiveTurns int `json:"activeTurns"` + } `json:"data"` +} + +// turnFrameFixture 是一次「忙时两路」场景的生产依赖:脚本化 runner(停在 Run 里, +// 直到被放行或被取消)+ 真编排器 + 真 WS 连接 + 记录型历史。 +type turnFrameFixture struct { + srv *Server + o *Orchestrator + run *a2Runner + hist *fakeHistory + conn *websocket.Conn +} + +// newTurnFrameFixture 拨一条 owner 连接并接线一个默认额度(2 路)的编排器。 +func newTurnFrameFixture(t *testing.T) *turnFrameFixture { + t.Helper() + run := newA2Runner() + hist := &fakeHistory{} + srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist}) + conn := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner)) + t.Cleanup(func() { _ = conn.Close() }) + waitConnCount(t, srv, 1) + return &turnFrameFixture{srv: srv, o: srv.hub.orchestrator, run: run, hist: hist, conn: conn} +} + +// startChat 写下一条 chat 上行帧(可带并行声明)。 +func (f *turnFrameFixture) startChat(t *testing.T, text string, parallel bool) { + t.Helper() + payload := map[string]any{"sessionId": "web", "text": text} + if parallel { + payload["parallel"] = true + } + b, err := json.Marshal(payload) + if err != nil { + t.Fatalf("编码 chat 帧: %v", err) + } + if err := f.conn.WriteJSON(wsInbound{Type: "chat", Data: b}); err != nil { + t.Fatalf("write chat %s: %v", text, err) + } +} + +// beginOneLeg 发一条 chat 帧并读回它的 `turn_begin` 帧(期间的状态帧与持久化帧 +// 一并跳过)。 +func (f *turnFrameFixture) beginOneLeg(t *testing.T, text string, parallel bool) turnBeginWire { + t.Helper() + f.startChat(t, text, parallel) + raw := readFrameOfType(t, f.conn, EventTurnBegin, "chat "+text) + var begin turnBeginWire + if err := json.Unmarshal(raw, &begin); err != nil { + t.Fatalf("解 turn_begin 帧 %s: %v", raw, err) + } + if begin.TurnID == "" { + t.Fatalf("turn_begin 帧壳没有 turnId:%s", raw) + } + return begin +} + +// interruptTurn 发一条 `interrupt{scope:"current", turnId}` 上行帧。 +func (f *turnFrameFixture) interruptTurn(t *testing.T, turnID string) { + t.Helper() + b, err := json.Marshal(map[string]string{ + "sessionId": "web", + "scope": string(InterruptScopeCurrent), + "turnId": turnID, + }) + if err != nil { + t.Fatalf("编码 interrupt 帧: %v", err) + } + if err := f.conn.WriteJSON(wsInbound{Type: "interrupt", Data: b}); err != nil { + t.Fatalf("write interrupt %s: %v", turnID, err) + } +} + +// twoLegsUp 起满两路(第一条普通、第二条显式并行),等到两路都进了 runner,并 +// 断言两帧 turn_begin 的 kind / leg / activeTurns(帧壳 turnId 由此取得)。 +func (f *turnFrameFixture) twoLegsUp(t *testing.T) (turnBeginWire, turnBeginWire) { + t.Helper() + first := f.beginOneLeg(t, "A", false) + waitUntil(t, "第一路进入模型调用", func() bool { return f.run.count() == 1 }) + second := f.beginOneLeg(t, "B", true) + waitUntil(t, "两路都在跑", func() bool { return f.run.count() == 2 && f.o.ActiveTurns("web") == 2 }) + if first.Data.Kind != turnBeginKindTurn || first.Data.Leg != 1 || first.Data.ActiveTurns != 1 { + t.Fatalf("第一路 turn_begin = %+v, want {kind:turn, leg:1, activeTurns:1}", first.Data) + } + if second.Data.Kind != turnBeginKindParallel || second.Data.Leg != 2 || second.Data.ActiveTurns != 2 { + t.Fatalf("第二路 turn_begin = %+v, want {kind:parallel, leg:2, activeTurns:2}", second.Data) + } + if first.Data.MessageIDs == nil || len(first.Data.MessageIDs) != 1 { + t.Fatalf("第一路 messageIds = %v, want 恰好一条用户消息 id", first.Data.MessageIDs) + } + if len(second.Data.MessageIDs) != 1 || second.Data.MessageIDs[0] == first.Data.MessageIDs[0] { + t.Fatalf("第二路 messageIds = %v,第一路 = %v:两路必须各钉各的消息", + second.Data.MessageIDs, first.Data.MessageIDs) + } + return first, second +} + +// TestInterruptTurnOverWSStopsOnlyTargetLeg 钉住④的端到端:`interrupt{scope: +// "current", turnId}` 只停那一路 —— 另一路仍在飞,且随后照常跑完并落库;被取消 +// 的第二路不补说明(现状语义);帧上 cancelledTurns=1、dropped=0,且本次打断 +// **不回错误帧**、只上报一次。 +func TestInterruptTurnOverWSStopsOnlyTargetLeg(t *testing.T) { + t.Parallel() + f := newTurnFrameFixture(t) + _, second := f.twoLegsUp(t) + + f.interruptTurn(t, second.TurnID) + raw := readFrameOfType(t, f.conn, EventSessionInterrupt, "interrupt turnId") + var frame interruptTurnWire + if err := json.Unmarshal(raw, &frame); err != nil { + t.Fatalf("解 session_interrupted 帧 %s: %v", raw, err) + } + if frame.Data.CancelledTurns != 1 { + t.Fatalf("cancelledTurns = %d, want 1(只停目标那一路)frame=%s", frame.Data.CancelledTurns, raw) + } + if frame.Data.Dropped != 0 { + t.Fatalf("dropped = %d, want 0(目标化打断不碰待处理队列)", frame.Data.Dropped) + } + if frame.Data.Mode == string(ModeIdle) { + t.Fatalf("mode = %q, want 非 idle(另一路仍在飞)", frame.Data.Mode) + } + waitUntil(t, "被取消的第二路收尾", func() bool { return f.o.ActiveTurns("web") == 1 }) + + // 另一路仍在跑:两路都进过 Run,而第一路此刻还停在它的 Run 里。 + if got := f.run.count(); got != 2 { + t.Fatalf("被驱动的回合数 = %d, want 2", got) + } + if got := f.run.texts(); !slices.Equal(got, []string{"A", "B"}) { + t.Fatalf("已驱动 = %v, want [A B]", got) + } + + close(f.run.release) + waitUntil(t, "会话排空", func() bool { return !f.o.Busy("web") && f.o.QueueDepth("web") == 0 }) + rows := assistantContents(f.hist.records()) + if !slices.Equal(rows, []string{"reply-A"}) { + t.Fatalf("助手回复 = %v, want [reply-A](另一路继续作答;被取消路不补说明)", rows) + } + + types := a2DrainTypes(t, f.conn, a2Settle) + if containsFrameType(types, "error") { + t.Fatalf("只停某一路不该回错误帧,实际帧序列 %v", types) + } + if containsFrameType(types, EventSessionInterrupt) { + t.Fatalf("一次 interrupt 只应上报一次 session_interrupted,实际帧序列 %v", types) + } +} + +// TestInterruptUnknownTurnIDIsDirectedErrorAndSparesAllLegs 钉住④的报错侧: +// `turnId` 指向不存在的路 ⇒ 定向回 `{code:"bad_request"}`,**且一路都不动** —— +// 两路随后都照常收尾并各自落一行回复(被误取消的那一路不会有回复),全程不发 +// session_interrupted。静默降级成「停全部」会让用户丢掉他并不想停的那一路, +// 这正是本用例要挡住的事。 +func TestInterruptUnknownTurnIDIsDirectedErrorAndSparesAllLegs(t *testing.T) { + t.Parallel() + f := newTurnFrameFixture(t) + f.twoLegsUp(t) + + f.interruptTurn(t, "turn-does-not-exist") + raw := readFrameOfType(t, f.conn, "error", "unknown turnId") + var errFrame struct { + Data struct { + Code string `json:"code"` + } `json:"data"` + } + if err := json.Unmarshal(raw, &errFrame); err != nil { + t.Fatalf("解 error 帧 %s: %v", raw, err) + } + if errFrame.Data.Code != wsBadRequestCode { + t.Fatalf("error.code = %q, want %q(显式报错,绝不静默当默认执行)", errFrame.Data.Code, wsBadRequestCode) + } + if got := f.o.ActiveTurns("web"); got != 2 { + t.Fatalf("目标不存在时在途路数 = %d, want 2(绝不误伤)", got) + } + + close(f.run.release) + waitUntil(t, "会话排空", func() bool { return !f.o.Busy("web") && f.o.QueueDepth("web") == 0 }) + rows := assistantContents(f.hist.records()) + slices.Sort(rows) + if want := []string{"reply-A", "reply-B"}; !slices.Equal(rows, want) { + t.Fatalf("助手回复 = %v, want %v(两路都必须照常作答)", rows, want) + } + + types := a2DrainTypes(t, f.conn, a2Settle) + if containsFrameType(types, EventSessionInterrupt) { + t.Fatalf("目标不存在的打断不得取消任何一路,实际帧序列 %v", types) + } +} diff --git a/internal/services/gateway/session_state.go b/internal/services/gateway/session_state.go index ea39f29..2ded023 100644 --- a/internal/services/gateway/session_state.go +++ b/internal/services/gateway/session_state.go @@ -12,8 +12,8 @@ import "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/history" type StateSink func(sessionID, typ string, data map[string]any) // TurnStateSink 是 StateSink 的**带 turnId 变体**(design/15 §2.5.9):帧壳上的 -// `turnId` 是客户端按回合两级分桶的依据,而少数帧(后续切片的 `turn_begin`、 -// 以及 `steer_ack`)必须让客户端知道**是哪一路**,只有 data 不足以表达。 +// `turnId` 是客户端按回合两级分桶的依据,而少数帧(`turn_begin`、`steer_ack`) +// 必须让客户端知道**是哪一路**,只有 data 不足以表达。 // // 它刻意是**另一个**回调而不是改 StateSink 的签名:既有接线点(server.go、 // 大量测试)的签名因此零改动,且空 turnID 时明确退化为现签名(帧壳上没有 diff --git a/internal/services/gateway/session_steer.go b/internal/services/gateway/session_steer.go index b9d8342..38b10bc 100644 --- a/internal/services/gateway/session_steer.go +++ b/internal/services/gateway/session_steer.go @@ -48,17 +48,24 @@ type steerTurn struct { orchestrator *Orchestrator sessionID string gen uint64 - consumed map[string]bool + // turnID 是本路(=本回合)的权威回合标识:`steer_ack` 的**帧壳**必须带上它 + // (忙时并行规范草案 §2.5)。真并行下同一会话有两条交错的路,只按 sessionId + // 分桶的话客户端不知道这条消息被**哪一路**吃进,灰态也就配不上正确的 + // `turn.end`。它在出生后不可变,因此这里存下来即可,无需在发帧时回查状态锁。 + turnID string + consumed map[string]bool } // beginSteer 为一个刚启动的回合建一份转向游标。编排器把它挂到**该回合自己的 ctx** // 上(loop.WithSteerSource),因此每个回合、每次续跑各有一份独立游标,上一个回合 -// 已吃进的消息绝不会被下一个回合重复注入或漏掉。 -func (o *Orchestrator) beginSteer(sessionID string, gen uint64) *steerTurn { +// 已吃进的消息绝不会被下一个回合重复注入或漏掉。turnID 是该路的权威回合标识, +// 与 `turn_begin` 帧壳同值(客户端靠它把 steer_ack 配到正确的 `turn.end`)。 +func (o *Orchestrator) beginSteer(sessionID string, gen uint64, turnID string) *steerTurn { return &steerTurn{ orchestrator: o, sessionID: sessionID, gen: gen, + turnID: turnID, consumed: make(map[string]bool), } } @@ -112,7 +119,10 @@ func (s *steerTurn) Steer(ctx context.Context, req loop.SteerRequest) []model.Me // steer_ack:该消息已被本回合吃进、尚未被正面回答(前端据此把它渲染成 // 灰态,design/15 §5)。它携带具体消息身份,因此不进 frameReachesAll 的 // 状态机信号白名单,落 fail-closed 默认档:只发 owner + admin。 - o.emitState(sessionID, wsSteerAck, map[string]any{"messageId": c.row.ID, "atStep": req.Step}) + // + // 帧壳必须带本路的 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)) } steerLog(o).Info("pending steered", "session", sessionID, "step", req.Step, "messages", len(out)) diff --git a/internal/services/gateway/session_steer_test.go b/internal/services/gateway/session_steer_test.go index 56ae72a..a622c98 100644 --- a/internal/services/gateway/session_steer_test.go +++ b/internal/services/gateway/session_steer_test.go @@ -286,6 +286,10 @@ func assertSteerExactlyOnce(t *testing.T, store *history.Store) { // assertSteerAckFrames 断言 design/15 §5 的帧壳/载荷键集/角色门:owner 与 admin 各 // 收到恰好一帧 steer_ack(messageId = 被吃进那条的权威 id、atStep = 2),viewer 收不到 // (负例以白名单哨兵确实到达它为前提,否则「什么都没收到」不可判定)。 +// +// 同时断言帧层的两件事(design/15 §2.5.9):这一路恰好一帧 `turn_begin`(把本路与 +// 第一条用户消息钉死),且 steer_ack 的**帧壳 turnId** 与它同值 —— 真并行下客户端 +// 就是靠这个值把灰态配到正确的 `turn.end`。 func assertSteerAckFrames(t *testing.T, f *steerFixture) { t.Helper() if err := f.bus.Publish(context.Background(), wsPrefix+scopeSentinel, map[string]any{"sessionId": "web"}); err != nil { @@ -298,15 +302,47 @@ func assertSteerAckFrames(t *testing.T, f *steerFixture) { for _, role := range []auth.Role{auth.RoleOwner, auth.RoleAdmin} { frames := drainFrames(t, f.conns[role]) assertFrameCount(t, string(role), frames, wsSteerAck, 1) - assertSteerAckFrame(t, string(role), findFrame(t, string(role), frames, wsSteerAck), users[1]) + ack := findFrame(t, string(role), frames, wsSteerAck) + assertSteerAckFrame(t, string(role), ack, users[1]) + assertTurnBeginBindsAck(t, string(role), frames, ack, users[0]) } viewerFrames := drainFrames(t, f.conns[auth.RoleViewer]) assertFrameCount(t, string(auth.RoleViewer), viewerFrames, wsSteerAck, 0) + // turn_begin 携带消息身份 ⇒ fail-closed 默认档,viewer 同样收不到。 + assertFrameCount(t, string(auth.RoleViewer), viewerFrames, EventTurnBegin, 0) if frameCountOf(viewerFrames, scopeSentinel) == 0 { t.Fatalf("viewer 未收到白名单哨兵 %q,负例不可判定(收到的帧:%v)", scopeSentinel, frameTypesOf(viewerFrames)) } } +// assertTurnBeginBindsAck 断言本路的 `turn_begin` 与其 `steer_ack` 在帧层对得上: +// 帧壳 turnId 同值、kind/leg/activeTurns 取规范草案 §2.4 的值、messageIds 是这一路 +// 归属的那条用户消息的权威 id。 +func assertTurnBeginBindsAck(t *testing.T, role string, frames []subagentFrame, ack subagentFrame, owner history.Message) { + t.Helper() + assertFrameCount(t, role, frames, EventTurnBegin, 1) + begin := findFrame(t, role, frames, EventTurnBegin) + if begin.TurnID == "" || begin.TurnID != ack.TurnID { + t.Fatalf("%s: turn_begin 帧壳 turnId = %q,steer_ack 的 = %q:两者必须同值(客户端靠它把灰态配到正确的 turn.end)", + role, begin.TurnID, ack.TurnID) + } + var data struct { + Kind string `json:"kind"` + MessageIDs []string `json:"messageIds"` + Leg int `json:"leg"` + ActiveTurns int `json:"activeTurns"` + } + if err := json.Unmarshal(begin.Data, &data); err != nil { + t.Fatalf("%s: turn_begin data = %s: %v", role, begin.Data, err) + } + if data.Kind != turnBeginKindTurn || data.Leg != 1 || data.ActiveTurns != 1 { + t.Fatalf("%s: turn_begin = %+v, want {kind:turn, leg:1, activeTurns:1}", role, data) + } + if len(data.MessageIDs) != 1 || data.MessageIDs[0] != owner.ID { + t.Fatalf("%s: turn_begin messageIds = %v, want [%s](本路归属的那条用户消息)", role, data.MessageIDs, owner.ID) + } +} + // frameCountOf 统计角色收到的某类型帧条数。 func frameCountOf(frames []subagentFrame, typ string) int { n := 0 diff --git a/internal/services/gateway/ws_bridge.go b/internal/services/gateway/ws_bridge.go index f39e730..12cd82f 100644 --- a/internal/services/gateway/ws_bridge.go +++ b/internal/services/gateway/ws_bridge.go @@ -22,9 +22,13 @@ const wsSubagentResult = "subagent_result" // (「已读、未答」,对齐 Claude Code 的 gray-until-responding);「正面回答」是随后 // 抵达的 message.* / turn.end 帧,不需要第二个确认帧。 // -// 帧壳:{type:"steer_ack", sessionId}。载荷 data:{messageId, atStep} —— messageId -// 是那条用户消息的权威 id(客户端据此把本地乐观气泡对齐到它,而不是靠文本相等这类 -// 脆弱匹配),atStep 是吃进它的模型步序号(从 1 开始)。 +// 帧壳:{type:"steer_ack", sessionId, turnId}。载荷 data:{messageId, atStep} —— +// messageId 是那条用户消息的权威 id(客户端据此把本地乐观气泡对齐到它,而不是靠 +// 文本相等这类脆弱匹配),atStep 是吃进它的模型步序号(从 1 开始)。 +// +// 帧壳的 turnId 是**吃进它的那一路**的权威回合标识(design/15 §2.5.9 / +// 忙时并行规范草案 §2.5):灰态要靠它与该路的 `turn.end` 配对才能收口,而真并行 +// 下同一会话有两条交错的路 —— 少了它,客户端只能猜这条消息属于哪一路。 // // 它携带具体消息身份(客户端能定位到某一条气泡,而不只是会话忙闲),因此**不登记** // 进 frameReachesAll 的「状态机信号」白名单,落 design/11 §7.1 的 fail-closed 默认 diff --git a/internal/services/gateway/ws_bridge_subagent_result_test.go b/internal/services/gateway/ws_bridge_subagent_result_test.go index 4c6212d..570e0d6 100644 --- a/internal/services/gateway/ws_bridge_subagent_result_test.go +++ b/internal/services/gateway/ws_bridge_subagent_result_test.go @@ -30,9 +30,12 @@ const subagentResultReadWindow = 1500 * time.Millisecond // subagentFrame 是测试侧的线上帧视图(等价于客户端解码):帧壳 type/sessionId 直接 // 解出,data 保留原始 JSON,使断言既能按形状解码、又能锁住键集。 type subagentFrame struct { - Type string `json:"type"` - SessionID string `json:"sessionId"` - Data json.RawMessage `json:"data"` + Type string `json:"type"` + SessionID string `json:"sessionId"` + // TurnID 是帧壳上的回合标识(design/15 §2.5.9):结构化流帧、`turn_begin` 与 + // 带帧壳 turnId 的 `steer_ack` 才有它,其余帧为空。 + TurnID string `json:"turnId"` + Data json.RawMessage `json:"data"` } // subagentFrameData 是 api-client §3.10 登记的 data 形状。 diff --git a/internal/services/gateway/ws_frame_scope_test.go b/internal/services/gateway/ws_frame_scope_test.go index 75ca09f..9290781 100644 --- a/internal/services/gateway/ws_frame_scope_test.go +++ b/internal/services/gateway/ws_frame_scope_test.go @@ -95,6 +95,10 @@ func frameScopeCases() []frameScopeCase { relayed("message_edited", false), relayed("message_deleted", false), relayed("channel_persisted", false), + // 新帧 turn_begin(design/15 §2.5.9):载荷带 messageIds ⇒ 携带具体消息 + // 身份,落 fail-closed 默认档(只发 owner + admin,与 steer_ack / + // chat_duplicate 同档),**不**进那 7 个状态机信号白名单。 + relayed(EventTurnBegin, false), // 系统内部 → 仅 owner/admin。 relayed("graph_update", false), diff --git a/internal/services/gateway/ws_hub_turn.go b/internal/services/gateway/ws_hub_turn.go index f75b3f3..be1d02f 100644 --- a/internal/services/gateway/ws_hub_turn.go +++ b/internal/services/gateway/ws_hub_turn.go @@ -103,7 +103,7 @@ func (h *hub) streamChat(ctx context.Context, payload any) SubmitResult { func (h *hub) runTurn(ctx context.Context, sessionID string, msgs []model.Message, env loop.SessionEnv) (*loop.Result, error) { // 该路的 turnId 优先取 ctx 上已有的那个(design/15 §2.5.9 的 ⚠️ 必改项): // 编排器在**启动该路之前**就生成了它,并把同一份 id 交给 runTurn —— 因为它 - // 发出(后续切片的)`turn_begin` 帧时必须报出这个 id,而客户端按 turnId + // 发出 `turn_begin` 帧时必须报出这个 id,而客户端按 turnId // 两级分桶。若这里再自己生成一个,两处 id 不同,分桶会立即失配。 // // 未挂载(子代理 / 后台思考 / 未接编排器的旧回退路径)时才新铸一个,因此 @@ -149,6 +149,14 @@ type interruptRequest struct { // 其余取值一律显式拒绝 —— 客户端以为队列保住了、实际被清空, // 就是静默丢消息,正是 A2 要修掉的那类缺陷。 Scope string `json:"scope,omitempty"` + // TurnID 是「只停这一路」的目标(忙时并行规范草案 §2.3):**只在 + // scope:"current" 时生效**;省略 scope 或 scope:"all" 时按契约**忽略**它 + // ——那两个取值的语义本就是「全部在途路 + 清 pending」。 + // + // 指向不存在的路(已被收尾 / 已被取消 / id 写错)⇒ 定向回 + // `{type:"error",data:{code:"bad_request"}}`,**绝不**静默降级成「停全部」: + // 那会把用户「只停这一路」的意图执行成「停掉所有路」。 + TurnID string `json:"turnId,omitempty"` } // parseInterruptScope 把上行 scope 解析为取值域内的口径。省略、空串与 @@ -178,6 +186,10 @@ func parseInterruptScope(raw string) (InterruptScope, error) { // 分发到此(design/11 §7.2 B)。 // // 缺 sessionId 时是空操作(既有行为,本次不改)。 +// +// 目标化打断(忙时并行规范草案 §2.3):`scope:"current"` + `turnId` ⇒ 只停那 +// 一路(InterruptTurn)。其它 scope 下 turnId 按契约**忽略**(它们的语义本就是 +// 「全部在途路 + 清队列」),这里只留一条 Debug 让「发了但没生效」可被追溯。 func (c *wsConn) handleInterrupt(payload json.RawMessage) { if c.h == nil { return @@ -193,9 +205,27 @@ func (c *wsConn) handleInterrupt(payload json.RawMessage) { if c.h.orchestrator == nil || req.SessionID == "" { return } + if req.TurnID != "" && scope != InterruptScopeCurrent { + c.h.log.Debug("ws interrupt turnId ignored", "session", req.SessionID, "scope", scope) + } + if req.TurnID != "" && scope == InterruptScopeCurrent { + c.interruptOneTurn(req.SessionID, req.TurnID) + return + } c.h.orchestrator.InterruptScope(req.SessionID, scope) } +// interruptOneTurn 执行一次「只停这一路」并如实回报结果:目标不存在时定向回 +// 发送者一条 {type:"error", data:{code:"bad_request"}}(只到发送者、不广播、也 +// 不动任何会话),**绝不**静默降级成「停掉全部在途路」。 +func (c *wsConn) interruptOneTurn(sessionID, turnID string) { + if c.h.orchestrator.InterruptTurn(sessionID, turnID) { + return + } + c.h.log.Debug("ws interrupt turn not found", "session", sessionID, "turn", turnID) + c.enqueueJSON(wsEvent{Type: "error", Data: map[string]any{"code": wsBadRequestCode}}) +} + // broadcastStream 把流式的 model.StreamEvent 映射为其下行 WS 事件类型, // 并经 hub 的帧门扇出(design/11 §7.1):每一个结构化流帧都携带模型 // 输出,因此只到达 owner + admin。它跑在 bus 订阅 worker 上;seq 原子