diff --git a/internal/runtime/history/recent_order_test.go b/internal/runtime/history/recent_order_test.go new file mode 100644 index 0000000..56b534d --- /dev/null +++ b/internal/runtime/history/recent_order_test.go @@ -0,0 +1,96 @@ +package history + +import ( + "context" + "fmt" + "testing" +) + +// TestRecentKeepsNewestOnTimestampTies 钉住「同一毫秒内的多条消息」时 Recent +// 仍返回**最新**的 n 条且顺序等于真实插入顺序（裁决 13.4.5：上下文续接窗口 +// 必须最新、完整、一致）。 +// +// 为什么同毫秒是常态而不是边角：Windows 上 time.Now() 的步长约 0.44ms +// （工作小结「平台坑位」），用户连发多条消息时 created_at 大量并列； +// design/15 §2.1 的「忙时连发多条」场景本身就会连续写入。若 Recent 的并列 +// 顺序取「插入更早的优先」，续接窗口就会**丢掉刚发出的最新消息**，而 +// 保留更旧的——上下文既不最新也不完整。 +func TestRecentKeepsNewestOnTimestampTies(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + const session = "sess-ties" + const ties = 25 + const window = 20 + + // 全部 25 条共用同一个 created_at：只有插入顺序（rowid）能区分先后。 + for i := 0; i < ties; i++ { + m := Message{ + SessionID: session, + Role: RoleUser, + Content: fmt.Sprintf("m%02d", i), + CreatedAt: 1_700_000_000_000, + } + if err := st.Append(ctx, m); err != nil { + t.Fatalf("append %d: %v", i, err) + } + } + + got, err := st.Recent(ctx, session, window) + if err != nil { + t.Fatalf("recent: %v", err) + } + if len(got) != window { + t.Fatalf("len = %d, want %d", len(got), window) + } + // 最新优先被保住的窗口 = 最后插入的 window 条，按插入顺序（最旧在前）。 + for i, m := range got { + want := fmt.Sprintf("m%02d", ties-window+i) + if m.Content != want { + t.Fatalf("recent[%d] = %q, want %q (窗口必须取最新插入的那几条)", i, m.Content, want) + } + } + if last := got[len(got)-1].Content; last != fmt.Sprintf("m%02d", ties-1) { + t.Fatalf("recent tail = %q, want the newest inserted message", last) + } +} + +// TestRecentOrdersMixedTiesChronologically 证明并列与非并列混在一起时， +// 窗口整体仍是「时间升序 + 同毫秒按插入顺序」，即上下文时序与真实对话一致。 +func TestRecentOrdersMixedTiesChronologically(t *testing.T) { + t.Parallel() + st := newStore(t) + ctx := context.Background() + const session = "sess-mixed" + + type row struct { + content string + at int64 + } + rows := []row{ + {"a0", 100}, + {"a1", 100}, + {"a2", 100}, + {"b0", 200}, + {"b1", 200}, + {"c0", 300}, + } + for _, r := range rows { + if err := st.Append(ctx, Message{SessionID: session, Role: RoleUser, Content: r.content, CreatedAt: r.at}); err != nil { + t.Fatalf("append %s: %v", r.content, err) + } + } + + got, err := st.Recent(ctx, session, len(rows)) + if err != nil { + t.Fatalf("recent: %v", err) + } + if len(got) != len(rows) { + t.Fatalf("len = %d, want %d", len(got), len(rows)) + } + for i, want := range rows { + if got[i].Content != want.content { + t.Fatalf("recent[%d] = %q, want %q (同毫秒必须按插入顺序)", i, got[i].Content, want.content) + } + } +} diff --git a/internal/runtime/history/store.go b/internal/runtime/history/store.go index 6a14014..878d41c 100644 --- a/internal/runtime/history/store.go +++ b/internal/runtime/history/store.go @@ -444,12 +444,19 @@ func (s *Store) ListAllLog(ctx context.Context, before int64, limit int) ([]Mess // Recent 按最旧优先返回 sessionID 最近的 n 条消息（最靠尾部的那段窗口），供 // WS 回放 / 上下文尾部使用。返回切片非 nil。 +// +// 并列打破（裁决 13.4.5 的上下文「最新性/一致性」）：同一毫秒写入的多条消息 +// 在 created_at 上并列，而 Windows 上 time.Now() 的步长本就约 0.44ms —— 用户 +// 连发多条（design/15 §2.1 的忙碌续接场景）时并列是常态。`rowid DESC` 让 LIMIT +// 取到的是**最后插入**的那几条（否则窗口会留下更旧的、丢掉刚发出的），而反转后 +// 同毫秒内即回到插入顺序，使续接窗口的时序与真实对话一致。该口径与 ListAfter +// 的「rowid 打破 created_at 的并列」相同。 func (s *Store) Recent(ctx context.Context, sessionID string, n int) ([]Message, error) { if n < 1 { n = defaultPageLimit } rows, err := s.db.QueryContext(ctx, `SELECT id, session_id, role, content, mode, mood, image, thinking, platform, device_id, sender_name, adapter_id, persona_id, created_at - FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC LIMIT ?`, sessionID, n) + FROM chat_messages WHERE session_id = ? ORDER BY created_at DESC, rowid DESC LIMIT ?`, sessionID, n) if err != nil { return nil, fmt.Errorf("history: recent %s: %w", sessionID, err) } diff --git a/internal/runtime/loop/run.go b/internal/runtime/loop/run.go index e8668af..9649b0d 100644 --- a/internal/runtime/loop/run.go +++ b/internal/runtime/loop/run.go @@ -82,7 +82,21 @@ func (l *Loop) Run(ctx context.Context, sessionID string, messages []model.Messa // 已 settle 的 Result；step 或工具批次的错误以 %w 包装返回。 func (l *Loop) driveTurn(ctx context.Context, sessionID string, msgs []model.Message, startStep int, res *Result) (*Result, error) { toolState := &toolLoopState{prevFP: make(map[string]bool)} - for step := startStep; step <= l.maxRounds; step++ { + step := startStep + // 被用户中断（design/15 §3）的回合不留下「可恢复的在途态」：design/15 + // 要求中断后**立即**按最新上下文处理下一条消息，而一个被放弃的回合若把 + // 可恢复 checkpoint 留在该会话上，下一条消息会被它顶掉——继承旧窗口、 + // 丢掉刚发出的那条，上下文既不最新也不完整（裁决 13.4.5）。因此取消后 + // 把该会话的快照落定为不可恢复态（快照本身保留，仍可查询）。 + // + // 只在回合**未落定**时标记（落定路径已写入终态快照，其 stop reason 本就 + // 不可恢复），因此这里不会改写一个已交付答案的回合。 + defer func() { + if ctx.Err() != nil && res.StopReason == "" { + l.abandonCanceledCheckpoint(ctx, sessionID, step, msgs) + } + }() + for ; step <= l.maxRounds; step++ { if err := ctx.Err(); err != nil { return nil, fmt.Errorf("loop: run: step %d: %w", step, err) } @@ -143,3 +157,22 @@ func (l *Loop) settleAndReturn(ctx context.Context, sessionID string, msgs []mod l.settleTurn(ctx, sessionID, msgs, res) return res, nil } + +// StopReasonCanceled 标记一个因用户中断（design/15 §3「停下/算了/换个话题」） +// 而放弃的回合。它不是可恢复的在途态：design/15 要求中断后立即按**最新** +// 上下文处理下一条消息，因此这种快照绝不能被下一条消息续接 +// （见 driveTurn 与 abandonCanceledCheckpoint）。 +const StopReasonCanceled = "canceled" + +// abandonCanceledCheckpoint 把一个被取消回合的 checkpoint 落定为不可恢复态 +// （stop reason = StopReasonCanceled），使该会话的下一条消息从它自己的最新 +// 上下文开始，而不是被放弃回合的旧窗口顶掉（裁决 13.4.5：上下文一致性 / +// 完整性 / 最新性）。快照本身保留在 store 里，仍可查询。 +// +// 写入必须脱离已取消的 ctx（否则 ExecContext 必然立即失败），因此这里用 +// WithoutCancel 派生背景上下文，保留原 ctx 的值而丢弃取消信号。失败只记日志， +// 绝不让这个收尾动作把中断路径变成错误。 +func (l *Loop) abandonCanceledCheckpoint(ctx context.Context, sessionID string, step int, msgs []model.Message) { + l.log.Info("checkpoint abandoned", "session", sessionID, "step", step, "stop_reason", StopReasonCanceled) + l.saveCheckpoint(context.WithoutCancel(ctx), sessionID, step, msgs, StopReasonCanceled) +} diff --git a/internal/runtime/loop/run_cancel_test.go b/internal/runtime/loop/run_cancel_test.go new file mode 100644 index 0000000..adf11eb --- /dev/null +++ b/internal/runtime/loop/run_cancel_test.go @@ -0,0 +1,145 @@ +package loop + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" +) + +// 本文件锁定裁决 13.4.5 在「回合被用户中断之后」的上下文口径（design/15 §3）： +// 中断后**立即处理下一条消息**，且它必须带着**最新**上下文——被放弃的回合 +// 绝不能把它的可恢复 checkpoint 留给下一条消息续接（否则新消息会被旧窗口顶掉， +// 上下文既不最新也不完整）。 + +// cancelingAdapter 复现「用户中断正在跑的回合」：第一次 Chat 交回一个工具调用， +// 第二次 Chat 阻塞到 ctx 被取消并把 ctx 的错误交回，之后每次都直接收尾。 +// 它使一次回合能在**已写入可恢复 checkpoint**（工具步之后）之后才被中断。 +type cancelingAdapter struct { + toolCall string + + mu sync.Mutex + n int + reqs []model.Request +} + +func (a *cancelingAdapter) Name() string { return "canceling" } + +func (a *cancelingAdapter) 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: a.toolCall}, nil + case 2: + <-ctx.Done() + return model.Response{}, ctx.Err() + default: + return model.Response{Content: "final"}, nil + } +} + +func (a *cancelingAdapter) ChatStream(ctx context.Context, req model.Request, _ func(model.StreamEvent)) (model.Response, error) { + return a.Chat(ctx, req, model.Options{}) +} + +// requests 返回被记录模型请求的快照。 +func (a *cancelingAdapter) requests() []model.Request { + a.mu.Lock() + defer a.mu.Unlock() + return append([]model.Request(nil), a.reqs...) +} + +// TestRun_CanceledTurnDoesNotHijackNextTurn 是 13.4.5 的中断续接用例： +// 回合 A 在工具循环中途被中断（design/15 §3），同一会话的下一条消息 B 必须 +// **以 B 自己的最新上下文**开始，不得继承 A 的旧窗口。 +// +// A 的窗口里带着工具结果、且**不含 B**，因此「续接 A」就等于把 B 丢掉 —— +// 这正是本用例要挡住的上下文不完整/不最新。用例跑在真实 persist.Store 上， +// 因此 upsert 与 LoadLatest 的取新语义都是真的。 +func TestRun_CanceledTurnDoesNotHijackNextTurn(t *testing.T) { + t.Parallel() + ctx := context.Background() + ad := &cancelingAdapter{toolCall: `{"tool_calls":[{"name":"echo","args":{"q":"x"}}]}`} + td := dispatcherWith(t, &mockTool{name: "echo", result: "hi"}) + a := newTestAgent(t, ad, td) + cp := newTestCheckpointStore(t) + l := New(Options{Agent: a, Checkpoint: cp, MaxRounds: 8}) + + const sess = "sess-cancel" + + // 回合 A：步 1 请求工具（落下一个 resumable 的 tool_calls checkpoint）， + // 步 2 的模型调用期间被中断。 + runCtx, cancel := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { + defer close(done) + if _, err := l.Run(runCtx, sess, []model.Message{model.TextMessage(model.RoleUser, "A")}); err == nil { + t.Errorf("canceled turn returned no error, want the cancellation error") + } + }() + waitAdapterCall(t, ad, 2) + cancel() + <-done + + // 下一条消息 B：必须看到 B 自己，而不是 A 的旧窗口。 + res, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "B")}) + if err != nil { + t.Fatalf("Run(B): %v", err) + } + if res.Text != "final" { + t.Fatalf("Text = %q, want final", res.Text) + } + reqs := ad.requests() + last := reqs[len(reqs)-1] + if !requestHasText(last, "B") { + t.Fatalf("被中断回合之后的请求不含新消息 B：%v（上下文不最新/不完整）", requestTexts(last)) + } + for _, m := range last.Messages { + if strings.Contains(messageText(m), "tool echo result") { + t.Fatalf("被中断回合之后的请求仍带着旧窗口的工具结果：%v", requestTexts(last)) + } + } +} + +// requestHasText 报告请求中是否有一条消息携带该精确文本。 +func requestHasText(req model.Request, want string) bool { + for _, m := range req.Messages { + if messageText(m) == want { + return true + } + } + return false +} + +// requestTexts 汇总请求里每条消息的文本，供失败信息可读。 +func requestTexts(req model.Request) []string { + out := make([]string, 0, len(req.Messages)) + for _, m := range req.Messages { + out = append(out, string(m.Role)+":"+messageText(m)) + } + return out +} + +// waitAdapterCall 阻塞到适配器已被调用 n 次（使中断精确发生在第二次模型调用 +// 的阻塞期内），超时即失败而不是静默继续。 +func waitAdapterCall(t *testing.T, ad *cancelingAdapter, n int) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + ad.mu.Lock() + got := ad.n + ad.mu.Unlock() + if got >= n { + return + } + time.Sleep(time.Millisecond) + } + t.Fatalf("adapter calls = %d, want >= %d", n, n) +} diff --git a/internal/services/gateway/ws_route.go b/internal/services/gateway/ws_route.go index d6680c6..c5ee738 100644 --- a/internal/services/gateway/ws_route.go +++ b/internal/services/gateway/ws_route.go @@ -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) +} diff --git a/internal/services/gateway/ws_route_chat_test.go b/internal/services/gateway/ws_route_chat_test.go new file mode 100644 index 0000000..4a9b733 --- /dev/null +++ b/internal/services/gateway/ws_route_chat_test.go @@ -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) + } +}