From 9b01efd7969dbffd9dc90f2d57176be96b489ce3 Mon Sep 17 00:00:00 2001 From: AskaEth Date: Sun, 4 Oct 2026 16:21:20 +0800 Subject: [PATCH] =?UTF-8?q?fix(loop):=20checkpoint=20=E6=81=A2=E5=A4=8D?= =?UTF-8?q?=E5=8A=A0=E8=87=AA=E6=B4=BD=E6=80=A7=E5=88=A4=E6=8D=AE=EF=BC=88?= =?UTF-8?q?=E7=BA=B5=E6=B7=B1=E9=98=B2=E5=BE=A1=EF=BC=8C=E5=A0=B5=E3=80=8C?= =?UTF-8?q?=E6=97=A7=E7=AA=97=E5=8F=A3=E5=90=9E=E6=96=B0=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E3=80=8D=E7=9A=84=E6=9C=80=E5=90=8E=E4=B8=80=E5=B1=82=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 背景：`loadCheckpoint` 恢复时会**整段替换**调用方带来的消息 ⇒ 任何「收尾落定前发起下一回合」的路径都会吞掉 用户刚发的消息。已有两层防护（收尾 defer 只让 tool_calls 可恢复 + 网关回合围栏），本次补 loop 侧判据 - 判据（保守版）：**只有在能证明「窗口已被取代」时才改判**——快照窗口的最后一条用户消息**确实出现在本回合 带来的消息里**（说明同一对话谱系），且本回合的最后一条用户消息 ≠ 它（说明带来了更新的输入）⇒ 不恢复、全新回合。 判不出来一律维持既有崩溃语义（fail-open 落在「保留崩溃恢复」一侧） - 比对键＝文本与图片引用按序拼接（纯图片消息也可区分，避免「刚发的一张图」被判成空键而漏判） - 复核：`internal/runtime/loop` 的 `-count=1`/`-race`/`-shuffle=on` 全绿；**未动任何既有崩溃用例**； 新增 4 条用例/9 条子用例（一致→仍恢复 / 有取代证据→不恢复且不含旧窗口 / 与网关围栏两条路径结论一致 / 边界表含重复播种·纯图片·无交集）；**变异两点**（判据恒假 → 4 处红；恒真 → 10 处红，精确打红崩溃用例） - 待主人拍板（已登记）：字面版判据（无交集也改判）会与「崩溃 + 用户随后发不相关新消息」的既有语义冲突； 更彻底的选项是「**恢复旧窗口并追加新消息**（union 而非替换）」——需改那条用例与设计口径 --- internal/runtime/loop/checkpoint.go | 77 ++++- .../loop/checkpoint_consistency_test.go | 271 ++++++++++++++++++ 2 files changed, 346 insertions(+), 2 deletions(-) create mode 100644 internal/runtime/loop/checkpoint_consistency_test.go diff --git a/internal/runtime/loop/checkpoint.go b/internal/runtime/loop/checkpoint.go index bcec9ea..db28ebd 100644 --- a/internal/runtime/loop/checkpoint.go +++ b/internal/runtime/loop/checkpoint.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "strings" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" @@ -64,8 +65,9 @@ func (l *Loop) saveCheckpoint(ctx context.Context, sessionID string, step int, m // 窗口、下一个要运行的循环步（checkpoint 步 + 1），以及 // 是否找到了可恢复的 checkpoint。不存在 checkpoint，或最新的一个 // 反映的是已完成/被阻塞的 run、或任何已落定的未完成退出（取消 / 报错 / -// 预算耗尽 / 护栏拦截）时，它返回调用方的消息并置步为 1，使循环从新开始 -// （主人 2026-10-04 裁决：只有崩溃才允许恢复；见 resumableStopReason）。 +// 预算耗尽 / 护栏拦截）、或它的窗口已被调用方本次带来的消息取代 +// （见 supersededCheckpointWindow）时，它返回调用方的消息并置步为 1，使循环 +// 从新开始（主人 2026-10-04 裁决：只有崩溃才允许恢复；见 resumableStopReason）。 // // 恢复是幂等的：checkpoint 精确保存了上一个已完成步结束时 // 的消息窗口（assistant 工具调用及其结果已融合在内）， @@ -93,6 +95,15 @@ func (l *Loop) loadCheckpoint(ctx context.Context, sessionID string, messages [] l.log.Debug("checkpoint not resumable", "session", sessionID, "step", cp.Step, "stop_reason", st.StopReason) return messages, 1, false } + // 纵深防御第三层（前两层：driveTurn 的收尾 defer 把未落定退出落定成不可恢复； + // 网关的 session_turn_fence.go 让新回合等上一回合真正返回）：快照的 stop reason + // 说明它是崩溃遗留的在途 run，还要它的窗口与调用方本次带来的消息指向**同一条 + // 最新用户输入**才续接，否则当作全新回合。 + if supersededCheckpointWindow(st.Messages, messages) { + l.log.Info("checkpoint window superseded", "session", sessionID, "step", cp.Step, + "next_step", 1, "messages", len(messages)) + return messages, 1, false + } l.log.Info("checkpoint resumed", "session", sessionID, "step", cp.Step, "next_step", cp.Step+1, "messages", len(st.Messages)) return st.Messages, cp.Step + 1, true } @@ -115,3 +126,65 @@ func (l *Loop) loadCheckpoint(ctx context.Context, sessionID string, messages [] func resumableStopReason(stopReason string) bool { return stopReason == agent.StopReasonToolCalls } + +// supersededCheckpointWindow 报告快照窗口的最后一条用户消息是否已被调用方 +// 本次带来的**更新**用户消息取代——取代了就不恢复（纵深防御第三层）。 +// +// 它挡的是「上一回合收尾落定之前就发起下一回合」这一类路径的最后一道口子。 +// 网关播种的新回合窗口是「该会话最近的持久化消息 + 刚发出的那条」 +// （gateway.startTurnMessages），被放弃/被中断回合的最后一条用户消息因此 +// **出现在**新窗口里，只是不再是最新的一条；此时续接旧快照就等于把用户刚发的 +// 消息整段吞掉（裁决 13.4.5 的最新性 / 完整性）。 +// +// 判据刻意保守：只有**正面证据**——快照窗口的最后一条用户消息确实出现在本回合 +// 的消息里、而本回合又带来了更新的用户消息——才判为「已被取代」。判不出来的一律 +// 维持既有崩溃语义：快照窗口与调用方消息完全无交集（进程重启后用户直接发了条 +// 全新消息、历史不可得）是主人 2026-10-04 裁决明确保留的崩溃恢复代价 +// （resume_test.go 的 TestRun_Resume_FromCrashSnapshot_LoadLatestCalled）， +// 本判据不据它改判。 +// +// 「同一条消息被重复播种」不会误判：判据比对的是两侧**最后一条**用户消息， +// 重复的同一条消息让两侧的键相等，判为同源，照常恢复。 +func supersededCheckpointWindow(window, turn []model.Message) bool { + windowLast := lastUserMessageKey(window) + if windowLast == "" { + return false + } + turnLast := lastUserMessageKey(turn) + if turnLast == "" || turnLast == windowLast { + return false + } + for _, m := range turn { + if m.Role == model.RoleUser && userMessageKey(m) == windowLast { + return true + } + } + return false +} + +// lastUserMessageKey 返回 messages 中最后一条用户消息的比对键； +// 没有用户消息（或该条没有任何可比内容）时返回 ""。 +func lastUserMessageKey(messages []model.Message) string { + for i := len(messages) - 1; i >= 0; i-- { + if messages[i].Role == model.RoleUser { + return userMessageKey(messages[i]) + } + } + return "" +} + +// userMessageKey 返回一条用户消息的比对键：文本与图片引用按出现顺序拼接。 +// 纯图片消息（无文本）因此也能区分——只比文本会把「用户刚发的一张图」判成 +// 空键而漏判。角色与说话人不进键：这里只问「是不是同一条用户输入」。 +func userMessageKey(m model.Message) string { + var b strings.Builder + for _, part := range m.Content { + if part.Type == model.ContentTypeImage { + b.WriteString("\x00img:") + b.WriteString(part.ImageURL) + continue + } + b.WriteString(part.Text) + } + return b.String() +} diff --git a/internal/runtime/loop/checkpoint_consistency_test.go b/internal/runtime/loop/checkpoint_consistency_test.go new file mode 100644 index 0000000..0338770 --- /dev/null +++ b/internal/runtime/loop/checkpoint_consistency_test.go @@ -0,0 +1,271 @@ +package loop + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/agent" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/model" + "git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/persist" +) + +// 本文件钉住 loadCheckpoint 的**自洽性判据**（纵深防御第三层，见 checkpoint.go +// 的 supersededCheckpointWindow）：可恢复快照（stop reason = tool_calls）只在它 +// 记的窗口与调用方本次带来的消息指向**同一条最新用户输入**时才续接，否则当作 +// 全新回合——绝不把用户刚发的消息整段吞掉（裁决 13.4.5 的最新性 / 完整性）。 +// +// 前两层：driveTurn 的收尾 defer 把未落定退出落定成不可恢复（resume_abandon_test.go +// / run_cancel_test.go），网关的回合围栏让新回合等上一回合真正返回 +// （gateway/session_turn_fence.go，产品行为的用例 = TestInterruptThenChatSlowUnwindKeepsLatestContext）。 +// 崩溃仍可恢复见 resume_test.go。 + +// requireContextHas 断言请求里有一条携带该精确文本的消息。 +func requireContextHas(t *testing.T, req model.Request, want string) { + t.Helper() + if !requestHasText(req, want) { + t.Fatalf("请求不含 %q：%v", want, requestTexts(req)) + } +} + +// requireContextLacks 断言请求里没有任何消息带该子串（用于证明旧窗口没被续接）。 +func requireContextLacks(t *testing.T, req model.Request, bad string) { + t.Helper() + for _, m := range req.Messages { + if strings.Contains(messageText(m), bad) { + t.Fatalf("请求仍带着旧窗口的内容 %q：%v", bad, requestTexts(req)) + } + } +} + +// requireLatestResumable 断言该会话最新的快照**确实可恢复**（stop reason 是 +// 崩溃遗留的 tool_calls）。它使「不恢复」的断言归因到自洽性判据本身，而不是 +// 恰好撞上 resumableStopReason 那道门。 +func requireLatestResumable(t *testing.T, ctx context.Context, cp *persist.Store, sessionID string) { + t.Helper() + latest, err := cp.LoadLatest(ctx, sessionID) + if err != nil { + t.Fatalf("LoadLatest(%s): %v", sessionID, err) + } + var st checkpointState + if err := json.Unmarshal(latest.State, &st); err != nil { + t.Fatalf("unmarshal checkpoint %s: %v", latest.ID, err) + } + if !resumableStopReason(st.StopReason) { + t.Fatalf("夹具快照 %s 的 stop_reason = %q，本应可恢复（否则本用例证明不了自洽性判据）", latest.ID, st.StopReason) + } +} + +// crashWindow 返回一份**崩溃遗留**快照会有的窗口：本回合的用户消息，后跟 +// 已完成工具批次的回灌结果（saveCheckpoint 只在工具批次之后写，故崩溃窗口 +// 必然带着工具结果）。 +func crashWindow() []model.Message { + return []model.Message{ + model.TextMessage(model.RoleUser, "A"), + model.TextMessage(model.RoleTool, "tool echo result: hi"), + } +} + +// TestRun_CheckpointWindowConsistentStillResumes 是判据的**正面**：快照窗口以 +// 同一条用户消息收尾（把崩溃前那个回合的输入原样再播一次），续接照旧——崩溃 +// 恢复语义（design/04 §1 durable execution）不被这条判据削弱，已完成的工具步 +// 不重跑。 +func TestRun_CheckpointWindowConsistentStillResumes(t *testing.T) { + t.Parallel() + ctx := context.Background() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final"}}}} + cp := newTestCheckpointStore(t) + l := New(Options{Agent: newTestAgent(t, ad, nil), Checkpoint: cp, MaxRounds: 8}) + + const sess = "sess-consistent" + l.saveCheckpoint(ctx, sess, 2, crashWindow(), agent.StopReasonToolCalls) + requireLatestResumable(t, ctx, cp, sess) + + // 调用方本次带来的窗口同样以用户消息「A」收尾（网关的播种会让它既出现在 + // 窗口开头、也出现在末尾——同一条消息被重复播种不误判）。 + res, err := l.Run(ctx, sess, []model.Message{model.TextMessage(model.RoleUser, "A")}) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Steps != 3 { + t.Fatalf("Steps = %d, want 3（一致 → 仍从被中断的步 3 续接）", res.Steps) + } + if res.ToolCallsExecuted != 0 { + t.Fatalf("ToolCallsExecuted = %d, want 0（已完成的工具步不得重跑）", res.ToolCallsExecuted) + } + if len(ad.reqs) != 1 { + t.Fatalf("模型调用 %d 次, want 1（只跑续接的那一步）", len(ad.reqs)) + } + requireContextHas(t, ad.reqs[0], "tool echo result: hi") +} + +// TestRun_CheckpointWindowSupersededNotResumed 是判据的**反面**：快照仍然可恢复 +// （被放弃回合的收尾还没落定），但新回合的窗口是该会话已播种的转录（含 A） +// 后跟刚发出的 B——快照窗口的最后一条用户消息被 B 取代，因此不得续接，B 必须 +// 在自己的上下文里被处理，旧窗口的工具结果一个都不许带进来。 +// +// 这正是「上一回合收尾落定之前就发起下一回合」时新回合拿到的窗口形态 +// （gateway.startTurnMessages 的播种）。 +func TestRun_CheckpointWindowSupersededNotResumed(t *testing.T) { + t.Parallel() + ctx := context.Background() + ad := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final"}}}} + cp := newTestCheckpointStore(t) + l := New(Options{Agent: newTestAgent(t, ad, nil), Checkpoint: cp, MaxRounds: 8}) + + const sess = "sess-superseded" + l.saveCheckpoint(ctx, sess, 2, crashWindow(), agent.StopReasonToolCalls) + requireLatestResumable(t, ctx, cp, sess) + + res, err := l.Run(ctx, sess, []model.Message{ + model.TextMessage(model.RoleUser, "A"), // 播种的转录 + model.TextMessage(model.RoleAssistant, "好的"), // 播种的转录 + model.TextMessage(model.RoleUser, "B"), // 刚发出 + }) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.Steps != 1 { + t.Fatalf("Steps = %d, want 1（窗口已被取代 → 全新回合）", res.Steps) + } + got := ad.reqs[0] + requireContextHas(t, got, "B") + requireContextLacks(t, got, "tool echo result") +} + +// TestRun_CheckpointSupersedeAndAbandonLayersAgree 断言纵深防御的两层**不冲突**： +// 同一份夹具、同一条新消息，无论挡住它的是网关围栏等来的收尾落定 +// （abandonCheckpoint 落下 canceled，既有 resumableStopReason 拒绝恢复），还是 +// 本判据（快照仍可恢复、但窗口已被取代），结果都必须是同一个——带 B 的全新回合。 +// +// 网关侧同一产品行为的用例：TestInterruptThenChatSlowUnwindKeepsLatestContext。 +func TestRun_CheckpointSupersedeAndAbandonLayersAgree(t *testing.T) { + t.Parallel() + ctx := context.Background() + + // turn 是本回合带来的窗口（播种转录 + 刚发出的 B）。 + turn := []model.Message{ + model.TextMessage(model.RoleUser, "A"), + model.TextMessage(model.RoleAssistant, "好的"), + model.TextMessage(model.RoleUser, "B"), + } + + // 路径①：收尾尚未落定 —— 快照仍是可恢复的 tool_calls，由本判据挡住。 + ad1 := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final"}}}} + cp1 := newTestCheckpointStore(t) + l1 := New(Options{Agent: newTestAgent(t, ad1, nil), Checkpoint: cp1, MaxRounds: 8}) + l1.saveCheckpoint(ctx, "sess-layers-resumable", 2, crashWindow(), agent.StopReasonToolCalls) + requireLatestResumable(t, ctx, cp1, "sess-layers-resumable") + res1, err := l1.Run(ctx, "sess-layers-resumable", turn) + if err != nil { + t.Fatalf("Run（收尾未落定）: %v", err) + } + + // 路径②：围栏等到了收尾 —— 快照已被落定为不可恢复，由 resumableStopReason 挡住。 + ad2 := &mockAdapter{chat: []mgResp{{resp: model.Response{Content: "final"}}}} + cp2 := newTestCheckpointStore(t) + l2 := New(Options{Agent: newTestAgent(t, ad2, nil), Checkpoint: cp2, MaxRounds: 8}) + l2.abandonCheckpoint(ctx, "sess-layers-abandoned", 2, crashWindow(), StopReasonCanceled) + res2, err := l2.Run(ctx, "sess-layers-abandoned", turn) + if err != nil { + t.Fatalf("Run（收尾已落定）: %v", err) + } + + for _, tc := range []struct { + name string + res *Result + ad *mockAdapter + }{ + {name: "收尾未落定（本判据挡住）", res: res1, ad: ad1}, + {name: "收尾已落定（resumableStopReason 挡住）", res: res2, ad: ad2}, + } { + if tc.res.Steps != 1 { + t.Fatalf("%s：Steps = %d, want 1（两层都必须落到全新回合）", tc.name, tc.res.Steps) + } + got := tc.ad.reqs[0] + requireContextHas(t, got, "B") + requireContextLacks(t, got, "tool echo result") + } +} + +// TestSupersededCheckpointWindow 逐条钉住判据的边界口径（含两个易误判点： +// 同一条消息被重复播种、纯图片消息）。 +func TestSupersededCheckpointWindow(t *testing.T) { + t.Parallel() + user := func(text string) model.Message { return model.TextMessage(model.RoleUser, text) } + image := func(url string) model.Message { return model.ImageMessage(model.RoleUser, "", url) } + tool := func(text string) model.Message { return model.TextMessage(model.RoleTool, text) } + assistant := func(text string) model.Message { return model.TextMessage(model.RoleAssistant, text) } + + cases := []struct { + name string + window []model.Message + turn []model.Message + want bool + }{ + { + name: "一致：最后一条用户消息相同 → 恢复", + window: []model.Message{user("A"), tool("r")}, + turn: []model.Message{user("A")}, + want: false, + }, + { + name: "一致：同一条消息被重复播种 → 恢复", + window: []model.Message{user("A"), tool("r")}, + turn: []model.Message{user("A"), user("A")}, + want: false, + }, + { + name: "被取代：本回合又带来了更新的用户消息 → 不恢复", + window: []model.Message{user("A"), tool("r")}, + turn: []model.Message{user("A"), assistant("好的"), user("B")}, + want: true, + }, + { + name: "无交集：判据不据它改判（崩溃语义保留）", + window: []model.Message{user("keep going")}, + turn: []model.Message{user("new")}, + want: false, + }, + { + name: "无交集：本回合有更早的用户消息也不改判", + window: []model.Message{user("X"), tool("r")}, + turn: []model.Message{user("M"), user("N")}, + want: false, + }, + { + name: "窗口没有用户消息 → 无从判定，不改判", + window: []model.Message{assistant("x")}, + turn: []model.Message{user("B")}, + want: false, + }, + { + name: "本回合没有用户消息 → 没有被吞掉的输入，不改判", + window: []model.Message{user("A")}, + turn: []model.Message{assistant("x")}, + want: false, + }, + { + name: "纯图片：窗口最后一张图出现在本回合、其后又有新图 → 不恢复", + window: []model.Message{image("data:a.png"), tool("r")}, + turn: []model.Message{image("data:a.png"), image("data:b.png")}, + want: true, + }, + { + name: "纯图片：同一张图 → 恢复", + window: []model.Message{image("data:a.png"), tool("r")}, + turn: []model.Message{image("data:a.png")}, + want: false, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := supersededCheckpointWindow(tc.window, tc.turn); got != tc.want { + t.Fatalf("supersededCheckpointWindow = %v, want %v", got, tc.want) + } + }) + } +}