AskaEth
5494c31474
- 原状：interrupt{sessionId} 取消在飞回合**并清空 pending** ⇒ 客户端「打断模式」连发会被静默丢弃
- 修法：新增 scope 三态（与项目既有「省略=保留现状」口径一致）
· 省略 或 "all" = 现状语义（取消当前 + 清 pending）
· "current" = **只取消当前回合、保留 pending**（pending 继续按编排器规则续跑）
· 非法值 → **定向回错误**（只回发送者，不广播），显式报错而非静默当默认
- 下行 session_interrupted 的 dropped 如实反映**真的被丢弃的条数**（scope=current 时通常为 0）
- 复核：gateway 包 build/vet/测试绿（新增 interrupt_scope_* 两组用例）；后端 :10942 全程未停
342 lines
13 KiB
Go
342 lines
13 KiB
Go
package gateway
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"net"
|
||
"slices"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"github.com/gorilla/websocket"
|
||
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||
)
|
||
|
||
// 本文件锁定协同项 A2 的 `interrupt {sessionId, scope}` 口径（上游清单：
|
||
// `Cyrene-For-Android/devdocs/02-前后端协同事项.md`）。客户端有「追加队列」与
|
||
// 「打断」两种模式，而后端过去只有一种口径（取消在途回合 + 清空待处理队列），
|
||
// 于是用户选「打断」时连发的多条消息会被静默丢弃，只回一个 dropped 数字。
|
||
//
|
||
// 落地口径（省略 = 维持现状）：
|
||
// - scope 省略 / "all" ⇒ 现状语义：取消在途回合 + 清空待处理队列；
|
||
// - scope "current" ⇒ 只取消在途回合，**保留**待处理队列（它们按编排器
|
||
// 的续跑规则继续执行），dropped 恒为 0；
|
||
// - 其余取值 ⇒ 只回发送者 {type:"error", data:{code:"bad_request"}}，
|
||
// 不广播、不打断任何会话（显式报错，绝不静默当默认执行）。
|
||
//
|
||
// 连接、令牌、编排器与 WS 路由都是真的（httptest + 真 JWT + 真 orchestrator），
|
||
// 只有模型适配器是脚本化的，因此「消息到底有没有被驱动」是被观测到的事实，
|
||
// 而不是从下行帧反推的推测。
|
||
const (
|
||
// a2ReadWait 界定一次预期会产出帧的读取。
|
||
a2ReadWait = 3 * time.Second
|
||
// a2Settle 让一个「本不该发生」的副作用有足够时间暴露自己，再断言它不存在。
|
||
a2Settle = 200 * time.Millisecond
|
||
// a2MainSession 是本组用例使用的会话 id。
|
||
a2MainSession = "web"
|
||
)
|
||
|
||
// a2Runner 是 WS 用例使用的假 Chatter：它记录每次 Run 的末条用户文本后阻塞，
|
||
// 直到 release 被关闭或自己的 ctx 被取消（打断路径）。会话因此稳定处于 busy，
|
||
// 测试能确定地断言忙时连发的消息最终有没有被驱动。
|
||
type a2Runner struct {
|
||
mu sync.Mutex
|
||
calls []string
|
||
release chan struct{}
|
||
}
|
||
|
||
func newA2Runner() *a2Runner { return &a2Runner{release: make(chan struct{})} }
|
||
|
||
func (r *a2Runner) Run(ctx context.Context, _ string, msgs []model.Message) (*loop.Result, error) {
|
||
text := ""
|
||
if n := len(msgs); n > 0 {
|
||
text = messageText(msgs[n-1])
|
||
}
|
||
r.mu.Lock()
|
||
r.calls = append(r.calls, text)
|
||
r.mu.Unlock()
|
||
select {
|
||
case <-r.release:
|
||
case <-ctx.Done():
|
||
return nil, ctx.Err()
|
||
}
|
||
return &loop.Result{Text: "reply-" + text, Steps: 1, StopReason: "stop"}, nil
|
||
}
|
||
|
||
// count 返回已被驱动的回合数。
|
||
func (r *a2Runner) count() int {
|
||
r.mu.Lock()
|
||
defer r.mu.Unlock()
|
||
return len(r.calls)
|
||
}
|
||
|
||
// texts 返回已驱动回合的末条用户文本快照。
|
||
func (r *a2Runner) texts() []string {
|
||
r.mu.Lock()
|
||
defer r.mu.Unlock()
|
||
return append([]string(nil), r.calls...)
|
||
}
|
||
|
||
// a2Runner 与 a2Event/a2Sink 等夹具、以及单元级用例见
|
||
// interrupt_scope_unit_test.go（拆开以守住单文件 ≤500 行）。
|
||
|
||
// interruptFrameWire 是解码后用于断言的 session_interrupted 帧（§3.2 形状：
|
||
// {sessionId, mode, dropped}）。
|
||
type interruptFrameWire struct {
|
||
Type string `json:"type"`
|
||
SessionID string `json:"sessionId"`
|
||
Data struct {
|
||
SessionID string `json:"sessionId"`
|
||
Mode string `json:"mode"`
|
||
Dropped int `json:"dropped"`
|
||
} `json:"data"`
|
||
}
|
||
|
||
// a2Env 为一个 A2 用例接线网关：脚本化的 runner + 真编排器 + 真 JWT + 真 WS。
|
||
type a2Env struct {
|
||
srv *Server
|
||
url string
|
||
jwt *auth.JWTService
|
||
run *a2Runner
|
||
o *Orchestrator
|
||
}
|
||
|
||
func newA2Env(t *testing.T, run *a2Runner) *a2Env {
|
||
t.Helper()
|
||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: &fakeHistory{}})
|
||
return &a2Env{srv: srv, url: ts.URL, jwt: jwt, run: run, o: srv.hub.orchestrator}
|
||
}
|
||
|
||
// owner 拨一条 owner 连接的 WS。
|
||
func (e *a2Env) owner(t *testing.T) *websocket.Conn {
|
||
t.Helper()
|
||
return dialWS(t, e.url, signAny(t, e.jwt, "acct", auth.RoleOwner))
|
||
}
|
||
|
||
// chatAndQueue 发出一条在飞消息与一条待处理消息，并等到两条都真的落地
|
||
// （在飞 + 队列深度 1），使后续断言面对一个确定的忙态而不是靠 sleep 赌时序。
|
||
func (e *a2Env) chatAndQueue(t *testing.T, conn *websocket.Conn, inFlight, queued string) {
|
||
t.Helper()
|
||
if err := conn.WriteJSON(wsInbound{Type: "chat", Data: a2ChatData(inFlight)}); err != nil {
|
||
t.Fatalf("write chat %s: %v", inFlight, err)
|
||
}
|
||
waitUntil(t, inFlight+" 回合在飞", func() bool { return e.run.count() == 1 })
|
||
if err := conn.WriteJSON(wsInbound{Type: "chat", Data: a2ChatData(queued)}); err != nil {
|
||
t.Fatalf("write chat %s: %v", queued, err)
|
||
}
|
||
waitUntil(t, queued+" 进入 pending", func() bool { return e.o.QueueDepth(a2MainSession) == 1 })
|
||
}
|
||
|
||
// a2ChatData 构造 chat 上行帧的 data。
|
||
func a2ChatData(text string) json.RawMessage {
|
||
b, _ := json.Marshal(map[string]string{"sessionId": a2MainSession, "text": text})
|
||
return b
|
||
}
|
||
|
||
// readFrameOfType 读帧直到出现 want 类型（握手回放、持久化增量等一并跳过），
|
||
// 返回那条帧的原始字节；超期未出现即失败。
|
||
func readFrameOfType(t *testing.T, conn *websocket.Conn, want, what string) []byte {
|
||
t.Helper()
|
||
deadline := time.Now().Add(a2ReadWait)
|
||
for time.Now().Before(deadline) {
|
||
_ = conn.SetReadDeadline(deadline)
|
||
_, msg, err := conn.ReadMessage()
|
||
if err != nil {
|
||
t.Fatalf("%s: 等待 %q 帧时读取失败: %v", what, want, err)
|
||
}
|
||
var shell struct {
|
||
Type string `json:"type"`
|
||
}
|
||
if err := json.Unmarshal(msg, &shell); err != nil {
|
||
t.Fatalf("%s: 解帧 %s: %v", what, msg, err)
|
||
}
|
||
if shell.Type == want {
|
||
return msg
|
||
}
|
||
}
|
||
t.Fatalf("%s: 截止时间前没有收到 %q 帧", what, want)
|
||
return nil
|
||
}
|
||
|
||
// a2DrainTypes 读取帧直到 window 内再无帧到达，返回窗口内收到的帧类型。
|
||
// 它给「本不该出现的帧」一个真实的时间窗口，并按类型断言（例如「不再有
|
||
// session_interrupted」），比单个 assertNoFrame 更能表达意图。它是该连接的
|
||
// 最后一次读取 —— 读截止时间到期后 gorilla 会留下一个已损坏的连接。
|
||
func a2DrainTypes(t *testing.T, conn *websocket.Conn, window time.Duration) []string {
|
||
t.Helper()
|
||
var types []string
|
||
// 每次读取单独续期，但总窗口有上限，使迟到的帧仍能被看见而测试不会挂住。
|
||
stop := time.Now().Add(4 * window)
|
||
for {
|
||
readBy := time.Now().Add(window)
|
||
if readBy.After(stop) {
|
||
readBy = stop
|
||
}
|
||
_ = conn.SetReadDeadline(readBy)
|
||
_, msg, err := conn.ReadMessage()
|
||
if err != nil {
|
||
var netErr net.Error
|
||
if errors.As(err, &netErr) && netErr.Timeout() {
|
||
return types
|
||
}
|
||
t.Fatalf("读取帧失败: %v", err)
|
||
}
|
||
var shell struct {
|
||
Type string `json:"type"`
|
||
}
|
||
if err := json.Unmarshal(msg, &shell); err != nil {
|
||
t.Fatalf("解帧 %s: %v", msg, err)
|
||
}
|
||
types = append(types, shell.Type)
|
||
}
|
||
}
|
||
|
||
// TestInterruptScopeOmittedAndAllDropPending 锁定 A2 的「维持现状」侧：scope
|
||
// 省略（老客户端只发 {sessionId}）与显式 "all" 必须逐字同解 —— 取消在途回合 +
|
||
// 清空待处理队列，session_interrupted 的 dropped 等于真的被丢弃的条数，且被
|
||
// 丢弃的那条绝不被驱动。
|
||
func TestInterruptScopeOmittedAndAllDropPending(t *testing.T) {
|
||
t.Parallel()
|
||
cases := []struct {
|
||
name string
|
||
data string
|
||
}{
|
||
{"scope 省略", `{"sessionId":"web"}`},
|
||
{"scope=all", `{"sessionId":"web","scope":"all"}`},
|
||
}
|
||
for _, tc := range cases {
|
||
tc := tc
|
||
t.Run(tc.name, func(t *testing.T) {
|
||
t.Parallel()
|
||
run := newA2Runner()
|
||
env := newA2Env(t, run)
|
||
conn := env.owner(t)
|
||
defer conn.Close()
|
||
waitConnCount(t, env.srv, 1)
|
||
env.chatAndQueue(t, conn, "A", "B")
|
||
|
||
if err := conn.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(tc.data)}); err != nil {
|
||
t.Fatalf("write interrupt: %v", err)
|
||
}
|
||
raw := readFrameOfType(t, conn, "session_interrupted", tc.name)
|
||
var frame interruptFrameWire
|
||
if err := json.Unmarshal(raw, &frame); err != nil {
|
||
t.Fatalf("%s: 解 session_interrupted 帧 %s: %v", tc.name, raw, err)
|
||
}
|
||
if frame.SessionID != a2MainSession || frame.Data.SessionID != a2MainSession {
|
||
t.Fatalf("%s: 帧的会话作用域 = %q/%q, want %q", tc.name, frame.SessionID, frame.Data.SessionID, a2MainSession)
|
||
}
|
||
if frame.Data.Dropped != 1 {
|
||
t.Fatalf("%s: dropped = %d, want 1（队列里那一条真的被丢弃）frame=%s", tc.name, frame.Data.Dropped, raw)
|
||
}
|
||
if frame.Data.Mode != string(ModeIdle) {
|
||
t.Fatalf("%s: mode = %q, want %q（队列清空后会话真的空闲）", tc.name, frame.Data.Mode, ModeIdle)
|
||
}
|
||
waitUntil(t, "队列清空且会话空闲", func() bool {
|
||
return !env.o.Busy(a2MainSession) && env.o.QueueDepth(a2MainSession) == 0
|
||
})
|
||
|
||
// 打断后 A 的收尾会发出一条 stream_end（被取消的回合如实收口），
|
||
// 因此这里按类型断言窗口内容，而不是断言「再无任何帧」：既不允许
|
||
// 第二条 session_interrupted，也不允许被丢弃的 B 被驱动。
|
||
close(run.release)
|
||
types := a2DrainTypes(t, conn, a2Settle)
|
||
if slices.Contains(types, "session_interrupted") {
|
||
t.Fatalf("%s: 一次 interrupt 只应上报一次 session_interrupted，实际帧序列 %v", tc.name, types)
|
||
}
|
||
if got := run.texts(); len(got) != 1 || got[0] != "A" {
|
||
t.Fatalf("%s: driven = %v, want [A]（被丢弃的 B 绝不能执行）", tc.name, got)
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// TestInterruptScopeCurrentOverWSResumesPending 是 A2 在真实 WS 路由上的端到端
|
||
// 落地：scope:"current" 的下行帧必须如实报告 dropped=0 且会话未空闲，而下一条
|
||
// 已被排队的消息必须真的被驱动（这就是「不再静默丢弃」的证据）。
|
||
func TestInterruptScopeCurrentOverWSResumesPending(t *testing.T) {
|
||
t.Parallel()
|
||
run := newA2Runner()
|
||
env := newA2Env(t, run)
|
||
conn := env.owner(t)
|
||
defer conn.Close()
|
||
waitConnCount(t, env.srv, 1)
|
||
env.chatAndQueue(t, conn, "A", "B")
|
||
|
||
if err := conn.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web","scope":"current"}`)}); err != nil {
|
||
t.Fatalf("write interrupt: %v", err)
|
||
}
|
||
raw := readFrameOfType(t, conn, "session_interrupted", "scope=current")
|
||
var frame interruptFrameWire
|
||
if err := json.Unmarshal(raw, &frame); err != nil {
|
||
t.Fatalf("解 session_interrupted 帧 %s: %v", raw, err)
|
||
}
|
||
if frame.Data.Dropped != 0 {
|
||
t.Fatalf("dropped = %d, want 0（current 口径不丢弃任何待处理消息）frame=%s", frame.Data.Dropped, raw)
|
||
}
|
||
if frame.Data.Mode == string(ModeIdle) {
|
||
t.Fatalf("mode = %q, want 非 idle（队列仍在续跑，会话没有被报成空闲）frame=%s", frame.Data.Mode, raw)
|
||
}
|
||
|
||
waitUntil(t, "被保留的 B 续跑", func() bool { return run.count() == 2 })
|
||
if got := run.texts(); len(got) != 2 || got[0] != "A" || got[1] != "B" {
|
||
t.Fatalf("driven = %v, want [A B]（pending 必须续跑）", got)
|
||
}
|
||
close(run.release)
|
||
waitUntil(t, "会话安定", func() bool {
|
||
return !env.o.Busy(a2MainSession) && env.o.QueueDepth(a2MainSession) == 0
|
||
})
|
||
}
|
||
|
||
// TestInterruptScopeInvalidIsDirectedErrorOnly 锁定 A2 的报错侧：非法 scope
|
||
// 只回发送者一条 {code:"bad_request"}，不广播（第二条连接收不到）、不打断任何
|
||
// 会话（在飞回合继续跑、待处理队列原样保留）。
|
||
func TestInterruptScopeInvalidIsDirectedErrorOnly(t *testing.T) {
|
||
t.Parallel()
|
||
run := newA2Runner()
|
||
env := newA2Env(t, run)
|
||
sender := env.owner(t)
|
||
defer sender.Close()
|
||
observer := env.owner(t)
|
||
defer observer.Close()
|
||
waitConnCount(t, env.srv, 2)
|
||
env.chatAndQueue(t, sender, "A", "B")
|
||
if err := sender.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web","scope":"pending"}`)}); err != nil {
|
||
t.Fatalf("write interrupt: %v", err)
|
||
}
|
||
raw := readFrameOfType(t, sender, "error", "非法 scope")
|
||
var ef errorFrameWire
|
||
if err := json.Unmarshal(raw, &ef); err != nil {
|
||
t.Fatalf("解错误帧 %s: %v", raw, err)
|
||
}
|
||
if ef.Data.Code != wsBadRequestCode {
|
||
t.Fatalf("error code = %q, want %q（非法 scope 必须显式报错，不得静默当默认）frame=%s", ef.Data.Code, wsBadRequestCode, raw)
|
||
}
|
||
|
||
// 只到发送者：观察者（它先收到了 chat 带来的状态帧与握手回放，这些被跳过）
|
||
// 在窗口里既收不到 error、也收不到 session_interrupted —— 拒绝不是广播。
|
||
seen := a2DrainTypes(t, observer, a2Settle)
|
||
if slices.Contains(seen, "error") || slices.Contains(seen, "session_interrupted") {
|
||
t.Fatalf("非法 scope 被广播了：观察者收到 %v（必须只定向回发送者）", seen)
|
||
}
|
||
// 发送者自己也只多了那一条错误帧：没有 session_interrupted。
|
||
sendSeen := a2DrainTypes(t, sender, a2Settle)
|
||
if slices.Contains(sendSeen, "session_interrupted") {
|
||
t.Fatalf("非法 scope 仍然打断了会话：发送者收到 %v", sendSeen)
|
||
}
|
||
|
||
// 会话完全没有被触动：回合 A 仍在飞（a2Runner 没被取消），队列仍是 1。
|
||
if got := run.texts(); len(got) != 1 || got[0] != "A" {
|
||
t.Fatalf("driven = %v, want [A]（非法 scope 不得打断或续跑任何东西）", got)
|
||
}
|
||
if env.o.QueueDepth(a2MainSession) != 1 || !env.o.Busy(a2MainSession) {
|
||
t.Fatalf("会话被非法 scope 触动了：busy=%v depth=%d", env.o.Busy(a2MainSession), env.o.QueueDepth(a2MainSession))
|
||
}
|
||
close(run.release)
|
||
}
|