fix(gateway): 回合围栏堵住「中断后立刻发新消息被旧快照吞掉」的时序竞态(13.4.5 最新性)
- 根因(已确定性复现,非测试问题):中断把会话立即置 idle ⇒ 紧随的新消息被当成空闲回合**立即开跑**, 而被取消回合的快照收尾在**它自己的 goroutine** 里、直到 Run 返回前才落定 ⇒ 新回合可能读到上一回合 中途留下的**可恢复**快照(stop_reason=tool_calls)并把调用方刚传进来的消息**整段替换** (现场证据:DB 里 web:1(step=1,tool_calls) 与 web:2(step=2,stop) 并存,LoadLatest 选中 web:1; 把收尾窗口拉长到 300ms ⇒ 旧码 3/3 必红) - 修法:新增 session_turn_fence.go —— 把「同一会话最多一个在途回合」从约定变成**可等待的围栏**: 新消息照旧被立即接受、Submit 立即返回(不阻塞硬约束不变),但其**回合启动**等上一回合真正返回 (那时被放弃的快照已落定);等待同时盯本回合 ctx,不收敛的上一回合不会永久挂住会话 - 复核:整包 `go test ./internal/services/gateway/ -count=1` 绿(隔离副本 0/12);`-shuffle=on` ×3 绿; **全树 `go test ./... -count=1` = 58 包 ok / 0 FAIL**;变异两处(HEAD 原码跑新用例 3/3 红、拿掉等待即红)已还原; 交叉验证:单独打回 loop 的「只有 tool_calls 可恢复」改动后,新用例仍 3/3 红 ⇒ 两侧修法互补 - 顺带:A2 的 `ws_route.go` 一行(定向回 bad_request 需要发送者的 conn;hub 级无按连接寻址入口) - 遗留(登记):loop 侧 `checkpoint.go` 的恢复会整段替换调用方消息 ⇒ 建议加一致性判据(checkpoint 窗口 最后一条用户消息 ≠ 调用方最新消息则不恢复);围栏未覆盖 orchestrator==nil 的 legacy 直连路径
This commit is contained in:
1 parent
5494c31474
commit
19f296a2ab
2 files changed
+91
-6
No files matched your search
@@ -48,9 +48,11 @@ func (c *wsConn) route(in wsInbound) {
|
||||
// 13.4.5:忙时入 pending、续接更新的旧上下文),不经 bus。
|
||||
c.submitChat(rawData(in.Data))
|
||||
case "interrupt":
|
||||
// design/15 §3: 用户说「停下/算了」→ 取消在飞回合并清空该会话的
|
||||
// 待处理队列,使下一条消息从干净状态开始。
|
||||
c.h.handleInterrupt(in.Data)
|
||||
// design/15 §3: 用户说「停下/算了」→ 取消在飞回合;待处理队列按
|
||||
// data.scope 决定去留(协同项 A2:省略/"all" = 连队列一起清,
|
||||
// "current" = 只取消当前回合并保留队列)。非法 scope 由连接级
|
||||
// 处理函数定向回 bad_request,因此这里调连接自己的方法。
|
||||
c.handleInterrupt(in.Data)
|
||||
case "voice_input":
|
||||
// 语音经注入的 STT→chat→TTS 管线处理。没有 voice service 时回一个
|
||||
// 明确的错误,而不是把帧丢掉。音频回复作为 voice_output 事件发在
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
@@ -207,10 +208,15 @@ func (r *blockingRunner) texts() []string {
|
||||
// loopChatAdapter 是集成用例使用的真实 model.Adapter:第一轮请求一个工具
|
||||
// (使循环落下一份**可恢复**的 tool_calls checkpoint),第二轮阻塞到 ctx 被
|
||||
// 取消(复现用户按「停下」中断在飞回合),之后每次直接收尾。
|
||||
//
|
||||
// unwind 是被取消之后到交还控制流之间的延迟:真实 provider 关闭在流、上层
|
||||
// 拆解在途请求都不是瞬时的,因此「中断」与「被放弃回合的快照落定」之间客观
|
||||
// 存在一个窗口。默认 0(瞬时),确定性用例把它拉长成一个可测窗口。
|
||||
type loopChatAdapter struct {
|
||||
mu sync.Mutex
|
||||
n int
|
||||
reqs []model.Request
|
||||
mu sync.Mutex
|
||||
n int
|
||||
reqs []model.Request
|
||||
unwind time.Duration
|
||||
}
|
||||
|
||||
func (a *loopChatAdapter) Name() string { return "loop-chat" }
|
||||
@@ -226,6 +232,9 @@ func (a *loopChatAdapter) Chat(ctx context.Context, req model.Request, _ model.O
|
||||
return model.Response{Content: `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`}, nil
|
||||
case 2:
|
||||
<-ctx.Done()
|
||||
if a.unwind > 0 {
|
||||
time.Sleep(a.unwind)
|
||||
}
|
||||
return model.Response{}, ctx.Err()
|
||||
default:
|
||||
return model.Response{Content: "好的"}, nil
|
||||
@@ -261,6 +270,10 @@ func (a *loopChatAdapter) lastRequest() model.Request {
|
||||
// 它同时钉住本组改动所依赖的组合事实:编排器的每个回合都以**同一个会话 id**
|
||||
// 驱动 loop,因此 loop 的 checkpoint(键 `sessionID:step`)在回合之间是共享的 ——
|
||||
// 被放弃回合的快照若不落定为不可恢复,下一条消息必然被它顶掉。
|
||||
//
|
||||
// 本用例把「中断」到「被放弃回合收尾」之间的窗口交给调度(适配器瞬时交还),
|
||||
// 因此整包运行时它偶发失败;把同一条产品行为钉在确定性窗口下的孪生用例是
|
||||
// TestInterruptThenChatSlowUnwindKeepsLatestContext。
|
||||
func TestInterruptThenChatKeepsLatestContextEndToEnd(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
@@ -323,3 +336,73 @@ func TestInterruptThenChatKeepsLatestContextEndToEnd(t *testing.T) {
|
||||
t.Fatalf("中断之后的回合上下文不含新消息 B:%v(上下文不最新/不完整)", texts)
|
||||
}
|
||||
}
|
||||
|
||||
// TestInterruptThenChatSlowUnwindKeepsLatestContext 是上一条用例的确定性版本。
|
||||
// 上一条把「中断」到「被放弃回合收尾」之间的窗口交给调度(适配器瞬时交还),
|
||||
// 整包运行时该窗口会被调度压开,因此它偶发失败;本用例把收尾拉长成一个确定
|
||||
// 可测的窗口(取消之后适配器仍需 300ms 才交还控制流,快照落定因此必然晚于
|
||||
// 中断本身),使同一条产品行为不再靠时序运气:
|
||||
//
|
||||
// 无论被中断的回合收尾多慢,下一条消息都必须等它真正返回之后再开始 —— 否则
|
||||
// 它会在收尾落定之前读到被放弃回合那份**仍可恢复**的旧快照并续接它,把刚发出
|
||||
// 的消息整段顶掉(裁决 13.4.5 的最新性 / 完整性)。围栏见 session_turn_fence.go。
|
||||
func TestInterruptThenChatSlowUnwindKeepsLatestContext(t *testing.T) {
|
||||
t.Parallel()
|
||||
ctx := context.Background()
|
||||
log := gwTestLogger()
|
||||
|
||||
conn, err := db.Open(ctx, ":memory:", log)
|
||||
if err != nil {
|
||||
t.Fatalf("open memory db: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = conn.Close() })
|
||||
cp, err := persist.New(ctx, conn, log)
|
||||
if err != nil {
|
||||
t.Fatalf("persist.New: %v", err)
|
||||
}
|
||||
|
||||
ad := &loopChatAdapter{unwind: 300 * time.Millisecond}
|
||||
ag, err := agent.New("You are a helpful assistant.", nil, nil, agent.WithModel(ad, nil))
|
||||
if err != nil {
|
||||
t.Fatalf("agent.New: %v", err)
|
||||
}
|
||||
l := loop.New(loop.Options{Agent: ag, Checkpoint: cp, MaxRounds: 8})
|
||||
|
||||
hist := newReplayHistoryStore()
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: l, History: hist, Bus: &saturatedBus{}})
|
||||
|
||||
conn1 := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner))
|
||||
defer conn1.Close()
|
||||
waitConnCount(t, srv, 1)
|
||||
|
||||
// 回合 A:步 1 请求工具,步 2 的模型调用被中断,且取消后仍要 300ms 才
|
||||
// 交还控制流 —— 被放弃回合的 checkpoint 落定因此晚于这条新消息。
|
||||
if err := conn1.WriteJSON(wsInbound{Type: "chat", Data: json.RawMessage(`{"sessionId":"web","text":"A"}`)}); err != nil {
|
||||
t.Fatalf("write chat A: %v", err)
|
||||
}
|
||||
waitUntil(t, "回合 A 进入第二次模型调用", func() bool { return ad.calls() >= 2 })
|
||||
if err := conn1.WriteJSON(wsInbound{Type: "interrupt", Data: json.RawMessage(`{"sessionId":"web"}`)}); err != nil {
|
||||
t.Fatalf("write interrupt: %v", err)
|
||||
}
|
||||
if err := conn1.WriteJSON(wsInbound{Type: "chat", Data: json.RawMessage(`{"sessionId":"web","text":"B"}`)}); err != nil {
|
||||
t.Fatalf("write chat B: %v", err)
|
||||
}
|
||||
|
||||
waitUntil(t, "回合 B 被驱动", func() bool { return ad.calls() >= 3 })
|
||||
got := ad.lastRequest()
|
||||
var texts []string
|
||||
hasB := false
|
||||
for _, m := range got.Messages {
|
||||
text := messageText(m)
|
||||
texts = append(texts, string(m.Role)+":"+text)
|
||||
if text == "B" {
|
||||
hasB = true
|
||||
}
|
||||
if strings.Contains(text, "no tool dispatcher configured") {
|
||||
t.Fatalf("中断之后的回合仍续接了被放弃回合的旧窗口:%v", texts)
|
||||
}
|
||||
}
|
||||
if !hasB {
|
||||
t.Fatalf("中断之后的回合上下文不含新消息 B:%v(上下文不最新/不完整)", texts)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user