fix(loop): checkpoint 恢复加自洽性判据（纵深防御，堵「旧窗口吞新消息」的最后一层）
- 背景：`loadCheckpoint` 恢复时会**整段替换**调用方带来的消息 ⇒ 任何「收尾落定前发起下一回合」的路径都会吞掉 用户刚发的消息。已有两层防护（收尾 defer 只让 tool_calls 可恢复 + 网关回合围栏），本次补 loop 侧判据 - 判据（保守版）：**只有在能证明「窗口已被取代」时才改判**——快照窗口的最后一条用户消息**确实出现在本回合 带来的消息里**（说明同一对话谱系），且本回合的最后一条用户消息 ≠ 它（说明带来了更新的输入）⇒ 不恢复、全新回合。 判不出来一律维持既有崩溃语义（fail-open 落在「保留崩溃恢复」一侧） - 比对键＝文本与图片引用按序拼接（纯图片消息也可区分，避免「刚发的一张图」被判成空键而漏判） - 复核：`internal/runtime/loop` 的 `-count=1`/`-race`/`-shuffle=on` 全绿；**未动任何既有崩溃用例**； 新增 4 条用例/9 条子用例（一致→仍恢复 / 有取代证据→不恢复且不含旧窗口 / 与网关围栏两条路径结论一致 / 边界表含重复播种·纯图片·无交集）；**变异两点**（判据恒假 → 4 处红；恒真 → 10 处红，精确打红崩溃用例） - 待主人拍板（已登记）：字面版判据（无交集也改判）会与「崩溃 + 用户随后发不相关新消息」的既有语义冲突； 更彻底的选项是「**恢复旧窗口并追加新消息**（union 而非替换）」——需改那条用例与设计口径
This commit is contained in:
AskaEth committed
1 parent
3d8822c922
commit
9b01efd796
2 files changed
+346
-2
No files matched your search
@@ -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()
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user