AskaEth
c0dc435770
新下行帧 turn_begin：编排器每启动一路恰好一帧（单路 kind:"turn" / 并行路 "parallel"），
data{kind, messageIds, leg, activeTurns}；角色门 fail-closed 仅 owner+admin，未进 frameReachesAll。
steer_ack 增帧壳 turnId（与该路 turn_begin 同源）—— 并行下没有它配不上正确的 turn.end。
上行 interrupt.turnId（仅 scope:"current" 生效）：精确停某一路；目标不存在 ⇒ 定向
{type:"error",data:{code:"bad_request"}}，不广播、不动任何会话；省略 turnId = v0.65 原口径（全部在途路）。
帧序按实现：session_busy / pending_drain → message_persisted(user) → turn_begin
（规范草案 §2.9 的时序图需按实现更正：messageIds 必须是已落库的权威 id，故只能在落库之后）。
测试：编排器侧 + 真 WS/JWT 端到端；变异 5 条（每路一帧 / 帧壳 turnId / 目标化打断 / 取消筛选）。
Docs 待升 v0.66（由监工登记）。
233 lines
9.2 KiB
Go
233 lines
9.2 KiB
Go
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)
|
||
}
|
||
}
|