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