fix(history,loop,gateway): 上下文续接的最新性/一致性三条（裁决 13.4.5 的面内部分）
本组只落地「在允许文件面内、且能用 design/15 既有规格推出」的部分；13.4.5 的「真并行 + 动态旧上下文」 与 design/15 §2.1/§2.2 的既有串行续接规格存在读法冲突，按规程**停下上报**（未自创语义）。 - history.Recent（续接窗口）并列打破：created_at DESC, rowid DESC —— Windows time.Now() 步长 ~0.44ms， 用户连发必然同毫秒并列；修复前并列时取**更早**的 ⇒ 丢掉刚发出的最新 n 条且同毫秒内反插入序 - 上行 chat 不再经**可丢弃**的 bus：直达会话编排器的非阻塞 Submit（design/15 §2.1「会话消息不能被 drop」 + §4「不阻塞是硬约束」）；orchestrator 为 nil 时回退 bus，旧断言仍绿 - 被**取消**（中断）的回合把该会话 checkpoint 落定为不可恢复（用 context.WithoutCancel 写，快照保留可查询， 仅在回合未落定时触发）⇒ 不再顶掉下一条消息的上下文 - 复核：build/vet 绿 + loop/gateway/history 三包测试与 -race 全绿；新增 5 条用例（含一条真 WS+真编排器+ 真 loop+真 persist 的集成用例）；**3 条变异反证**精确复现「新消息被丢弃只见旧窗口」「窗口取早不取新」 「bus 饱和时会话消息静默消失」，均已还原 - 待裁决（已上报主人）：① 「同一会话多回合真并行」需改 checkpoint 键（sessionID:step 并发互相覆盖）、 flow 图按轮键控、前端单槽 busy —— 均在本组文件面外；② 「动态吸收新到消息」在 design/15 **无规格**， 属新语义，须先写设计；③ 非取消的错误退出同样留可恢复快照（同机制，未扩范围，因 resume_test 正钉住旧行为）
This commit is contained in:
AskaEth committed
1 parent
a1083a322f
commit
74b12d87ca
6 files changed
+633
-3
No files matched your search
@@ -42,7 +42,11 @@ func (c *wsConn) route(in wsInbound) {
|
||||
case "mode_switch":
|
||||
c.publishPublic("ws.mode_switch", rawData(in.Data))
|
||||
case "chat":
|
||||
c.publishPublic(wsChatSubject, rawData(in.Data))
|
||||
// design/15 §2.1：会话消息「不能被 drop」——bus 的有界订阅缓冲满时
|
||||
// 会丢弃事件（那是给事件/通知用的通道），而用户消息被丢就是上下文
|
||||
// 不完整。因此上行 chat 直接进入会话编排器的非阻塞 Submit（裁决
|
||||
// 13.4.5：忙时入 pending、续接更新的旧上下文），不经 bus。
|
||||
c.submitChat(rawData(in.Data))
|
||||
case "interrupt":
|
||||
// design/15 §3: 用户说「停下/算了」→ 取消在飞回合并清空该会话的
|
||||
// 待处理队列，使下一条消息从干净状态开始。
|
||||
@@ -145,3 +149,23 @@ func claimRole(claims *auth.Claims) string {
|
||||
}
|
||||
return string(claims.Role)
|
||||
}
|
||||
|
||||
// submitChat 把一条 chat 上行帧直接交给会话编排器（design/15 §2.1/§2.2），
|
||||
// 而不是经 bus 转发：会话消息不得依赖可丢弃的通道（见 route 的 chat 分支）。
|
||||
// 解码、角色解析与用户消息落库都在该 goroutine 上完成，因此读泵绝不等待
|
||||
// 回合（design/15 §4「不阻塞是硬约束」）。
|
||||
//
|
||||
// 未接编排器时（单元测试构造的裸 hub、或未接编排器的构建）回退到既有的
|
||||
// bus 触发器路径，使 hub 的 ws.chat 契约逐字节不变。
|
||||
func (c *wsConn) submitChat(payload any) {
|
||||
if c.h == nil {
|
||||
return
|
||||
}
|
||||
if c.h.orchestrator == nil {
|
||||
c.publishPublic(wsChatSubject, payload)
|
||||
return
|
||||
}
|
||||
// 回合自身的取消由编排器持有（Interrupt/InterruptAll 取消它），
|
||||
// 因此这里的基 ctx 只需是存活到回合结束的背景上下文。
|
||||
go c.h.streamChat(context.Background(), payload)
|
||||
}
|
||||
@@ -0,0 +1,325 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/auth"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/bus"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/db"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/loop"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model"
|
||||
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist"
|
||||
)
|
||||
|
||||
// 本文件锁定裁决 13.4.5 在 WS 上行 chat 路径上的落地口径（design/15 §2.1）：
|
||||
// **会话消息不能被 drop** ——「bus 的 drop 是给事件/通知用的」，用户消息丢了
|
||||
// 就是上下文不完整。因此上行 chat 直接进入会话编排器的非阻塞 Submit，
|
||||
// 绝不经可丢弃的 bus 通道去驱动回合。
|
||||
|
||||
// saturatedBus 模拟 bus 文档明示的代价：订阅的有界缓冲满时 Publish 丢弃事件。
|
||||
// 它记录每一次发布而不投递，用于证明会话消息不再依赖那条通道 ——
|
||||
// 若 chat 仍经 bus，用户消息会在「bus 已饱和」时静默消失（既不落库也不驱动回合）。
|
||||
type saturatedBus struct {
|
||||
mu sync.Mutex
|
||||
published []string
|
||||
}
|
||||
|
||||
func (b *saturatedBus) Publish(_ context.Context, subject string, _ any) error {
|
||||
b.mu.Lock()
|
||||
b.published = append(b.published, subject)
|
||||
b.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (b *saturatedBus) Subscribe(context.Context, string, bus.Handler) (func(), error) {
|
||||
return func() {}, nil
|
||||
}
|
||||
|
||||
func (b *saturatedBus) Close() error { return nil }
|
||||
|
||||
// subjects 返回迄今被发布的主题快照。
|
||||
func (b *saturatedBus) subjects() []string {
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return append([]string(nil), b.published...)
|
||||
}
|
||||
|
||||
// recordingRunner 是一个把每次被驱动的回合记下来的 Chatter 接缝：断言
|
||||
// 「用户消息真的到达了 agent」需要观测 runner 而非从广播反推。
|
||||
type recordingRunner struct {
|
||||
mu sync.Mutex
|
||||
calls []string
|
||||
}
|
||||
|
||||
func (r *recordingRunner) Run(_ context.Context, _ string, msgs []model.Message) (*loop.Result, error) {
|
||||
var text string
|
||||
if n := len(msgs); n > 0 {
|
||||
text = messageText(msgs[n-1])
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.calls = append(r.calls, text)
|
||||
r.mu.Unlock()
|
||||
return &loop.Result{Text: "收到", Steps: 1, StopReason: "stop"}, nil
|
||||
}
|
||||
|
||||
// count 返回已驱动的回合数。
|
||||
func (r *recordingRunner) count() int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return len(r.calls)
|
||||
}
|
||||
|
||||
// texts 返回被驱动的回合的用户文本快照。
|
||||
func (r *recordingRunner) texts() []string {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return append([]string(nil), r.calls...)
|
||||
}
|
||||
|
||||
// TestWSChatFrameSurvivesSaturatedBus 证明一条 WS 上行 chat 帧在 bus 完全
|
||||
// 不投递（饱和丢弃）时仍然完整处理：驱动回合 + 用户消息与助手回复都落库。
|
||||
// 修复前该帧只被 publishPublic 丢进 bus，于是既不驱动回合也不落库 ——
|
||||
// 用户消息在「忙时连发」场景下静默消失（13.4.5 的完整性）。
|
||||
func TestWSChatFrameSurvivesSaturatedBus(t *testing.T) {
|
||||
t.Parallel()
|
||||
run := &recordingRunner{}
|
||||
hist := newReplayHistoryStore()
|
||||
sb := &saturatedBus{}
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: sb})
|
||||
|
||||
conn := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner))
|
||||
defer conn.Close()
|
||||
waitConnCount(t, srv, 1)
|
||||
|
||||
if err := conn.WriteJSON(wsInbound{
|
||||
Type: "chat",
|
||||
Data: json.RawMessage(`{"sessionId":"web","text":"第一条","deviceId":"web:test:1"}`),
|
||||
}); err != nil {
|
||||
t.Fatalf("write chat: %v", err)
|
||||
}
|
||||
|
||||
// 回合被驱动，且携带的正是用户刚发的那条文本。
|
||||
waitUntil(t, "chat 帧驱动回合", func() bool { return run.count() == 1 })
|
||||
if got := run.texts(); len(got) != 1 || got[0] != "第一条" {
|
||||
t.Fatalf("driven turns = %v, want [\"第一条\"]", got)
|
||||
}
|
||||
|
||||
// 用户消息与助手回复都落库：消息不丢、会话上下文完整。
|
||||
waitUntil(t, "用户+助手消息落库", func() bool { return len(hist.records()) == 2 })
|
||||
recs := hist.records()
|
||||
if recs[0].Role != "user" || recs[0].Content != "第一条" {
|
||||
t.Fatalf("user record = %+v, want the sent text persisted", recs[0])
|
||||
}
|
||||
if recs[1].Role != "assistant" || recs[1].Content != "收到" {
|
||||
t.Fatalf("assistant record = %+v, want the reply persisted", recs[1])
|
||||
}
|
||||
|
||||
// 会话消息绝不经可丢弃通道：chat 主题上不应有任何发布。
|
||||
for _, subject := range sb.subjects() {
|
||||
if subject == wsChatSubject {
|
||||
t.Fatalf("chat 帧仍被发布到可丢弃的 bus 主题 %q（design/15 §2.1 禁止会话消息走 drop 通道）", subject)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestWSChatFrameBusyStaysPendingNotDropped 锁定 design/15 §2.1 的忙碌分支：
|
||||
// 上一条回合未结束时又发一条，第二条不阻塞、不被丢，而是挂进该会话的
|
||||
// pending 队列（裁决 13.4.5「真并发、不阻塞主会话」的消息层落地）。
|
||||
func TestWSChatFrameBusyStaysPendingNotDropped(t *testing.T) {
|
||||
t.Parallel()
|
||||
release := make(chan struct{})
|
||||
run := &blockingRunner{release: release}
|
||||
hist := newReplayHistoryStore()
|
||||
sb := &saturatedBus{}
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: run, History: hist, Bus: sb})
|
||||
|
||||
conn := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner))
|
||||
defer conn.Close()
|
||||
waitConnCount(t, srv, 1)
|
||||
|
||||
if err := conn.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 run.started() == 1 })
|
||||
|
||||
if err := conn.WriteJSON(wsInbound{Type: "chat", Data: json.RawMessage(`{"sessionId":"web","text":"B"}`)}); err != nil {
|
||||
t.Fatalf("write chat B: %v", err)
|
||||
}
|
||||
// B 到达时 A 仍在飞：它必须进 pending（深度 1），且立即落库。
|
||||
waitUntil(t, "B 进入 pending", func() bool { return srv.hub.orchestrator.QueueDepth("web") == 1 })
|
||||
waitUntil(t, "A 与 B 的消息都落库", func() bool { return len(hist.records()) == 2 })
|
||||
if got := hist.records(); got[0].Content != "A" || got[1].Content != "B" {
|
||||
t.Fatalf("persisted = %q,%q, want A,B (排队消息也必须落库，不得丢)", got[0].Content, got[1].Content)
|
||||
}
|
||||
|
||||
close(release)
|
||||
waitUntil(t, "会话排空", func() bool {
|
||||
return !srv.hub.orchestrator.Busy("web") && srv.hub.orchestrator.QueueDepth("web") == 0
|
||||
})
|
||||
if got := run.texts(); len(got) != 2 || got[0] != "A" || got[1] != "B" {
|
||||
t.Fatalf("driven turns = %v, want A then B (有序续接，不丢)", got)
|
||||
}
|
||||
}
|
||||
|
||||
// blockingRunner 是一个停留在 Run 里直到 release 被关闭的 Chatter：它让会话
|
||||
// 稳定处于 busy，使测试能确定地断言「忙时到达」的分支。
|
||||
type blockingRunner struct {
|
||||
release chan struct{}
|
||||
mu sync.Mutex
|
||||
calls []string
|
||||
seen int
|
||||
}
|
||||
|
||||
func (r *blockingRunner) Run(ctx context.Context, _ string, msgs []model.Message) (*loop.Result, error) {
|
||||
var text string
|
||||
if n := len(msgs); n > 0 {
|
||||
text = messageText(msgs[n-1])
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.calls = append(r.calls, text)
|
||||
r.seen++
|
||||
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
|
||||
}
|
||||
|
||||
func (r *blockingRunner) started() int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return r.seen
|
||||
}
|
||||
|
||||
func (r *blockingRunner) texts() []string {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return append([]string(nil), r.calls...)
|
||||
}
|
||||
|
||||
// loopChatAdapter 是集成用例使用的真实 model.Adapter：第一轮请求一个工具
|
||||
// （使循环落下一份**可恢复**的 tool_calls checkpoint），第二轮阻塞到 ctx 被
|
||||
// 取消（复现用户按「停下」中断在飞回合），之后每次直接收尾。
|
||||
type loopChatAdapter struct {
|
||||
mu sync.Mutex
|
||||
n int
|
||||
reqs []model.Request
|
||||
}
|
||||
|
||||
func (a *loopChatAdapter) Name() string { return "loop-chat" }
|
||||
|
||||
func (a *loopChatAdapter) Chat(ctx context.Context, req model.Request, _ model.Options) (model.Response, error) {
|
||||
a.mu.Lock()
|
||||
a.n++
|
||||
n := a.n
|
||||
a.reqs = append(a.reqs, req)
|
||||
a.mu.Unlock()
|
||||
switch n {
|
||||
case 1:
|
||||
return model.Response{Content: `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`}, nil
|
||||
case 2:
|
||||
<-ctx.Done()
|
||||
return model.Response{}, ctx.Err()
|
||||
default:
|
||||
return model.Response{Content: "好的"}, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (a *loopChatAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) {
|
||||
return a.Chat(ctx, req, model.Options{})
|
||||
}
|
||||
|
||||
// calls 返回已发生的模型调用次数。
|
||||
func (a *loopChatAdapter) calls() int {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
return a.n
|
||||
}
|
||||
|
||||
// lastRequest 返回最近一次模型请求。
|
||||
func (a *loopChatAdapter) lastRequest() model.Request {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
if len(a.reqs) == 0 {
|
||||
return model.Request{}
|
||||
}
|
||||
return a.reqs[len(a.reqs)-1]
|
||||
}
|
||||
|
||||
// TestInterruptThenChatKeepsLatestContextEndToEnd 是 13.4.5 的集成用例
|
||||
// （真 WS + 真编排器 + **真 loop** + 真 persist.Store，只有模型适配器是脚本化的）：
|
||||
// 回合 A 在工具循环中途被 `interrupt` 帧中断后，下一条消息 B 必须带着 B 自己的
|
||||
// 最新上下文到达模型，而不是继承 A 留下的旧窗口。
|
||||
//
|
||||
// 它同时钉住本组改动所依赖的组合事实：编排器的每个回合都以**同一个会话 id**
|
||||
// 驱动 loop，因此 loop 的 checkpoint（键 `sessionID:step`）在回合之间是共享的 ——
|
||||
// 被放弃回合的快照若不落定为不可恢复，下一条消息必然被它顶掉。
|
||||
func TestInterruptThenChatKeepsLatestContextEndToEnd(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{}
|
||||
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()
|
||||
sb := &saturatedBus{}
|
||||
srv, ts, jwt := buildEnv(t, cfgDefault(), Deps{Loop: l, History: hist, Bus: sb})
|
||||
|
||||
conn1 := dialWS(t, ts.URL, signAny(t, jwt, "acct", auth.RoleOwner))
|
||||
defer conn1.Close()
|
||||
waitConnCount(t, srv, 1)
|
||||
|
||||
// 回合 A：步 1 请求工具（写下可恢复快照），步 2 的模型调用被中断。
|
||||
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)
|
||||
}
|
||||
|
||||
// 下一条消息 B：必须带着 B 自己到达模型（不得继承 A 的旧窗口）。
|
||||
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, "tool echo result") {
|
||||
t.Fatalf("中断之后的回合仍带着被放弃回合的旧窗口：%v", texts)
|
||||
}
|
||||
}
|
||||
if !hasB {
|
||||
t.Fatalf("中断之后的回合上下文不含新消息 B：%v（上下文不最新/不完整）", texts)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user