diff --git a/internal/services/gateway/ws_route.go b/internal/services/gateway/ws_route.go index c5ee738..9e8d5b5 100644 --- a/internal/services/gateway/ws_route.go +++ b/internal/services/gateway/ws_route.go @@ -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 事件发在 diff --git a/internal/services/gateway/ws_route_chat_test.go b/internal/services/gateway/ws_route_chat_test.go index 4a9b733..ede860b 100644 --- a/internal/services/gateway/ws_route_chat_test.go +++ b/internal/services/gateway/ws_route_chat_test.go @@ -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) + } +}