AskaEth
88d7dd0026
- 原状：只有「被取消」的回合把 checkpoint 落定为不可恢复；**模型/工具报错、预算耗尽**等未落定退出 同样留下可恢复快照 ⇒ 下一条消息被旧窗口整段顶掉（用户看到自己刚发的消息像丢了） - 修法：收尾 defer 改为「**回合未落定就废弃快照**」（不再只看 ctx 是否取消），新增 StopReasonAborted； checkpoint.resumableStopReason 收紧为「**只有 `tool_calls` 可恢复**」（空 stop reason 不再算）； 终态快照改用 context.WithoutCancel 写（修「取消擦肩最终答案」时终态写失败、中途快照留任最新） - 语义边界（写进用例注释）：**只有崩溃**（进程猝死、函数不返回）才允许恢复；控制流一旦回到调用方 就不得留可恢复态。原 resume_test 的用例改名为**崩溃恢复**用例并保留断言（含「新消息不被顶掉」的反面断言在新用例里） - 复核：build/vet 绿 + loop 包测试与 `-race` 全绿；新增 6 条（模型报错/工具失败+模型报错/预算耗尽/ 空 stop reason/取消擦肩最终答案）+ 端到端 TestInterruptThenChatKeepsLatestContextEndToEnd 绿； **4 条变异反证**（含把废弃快照写回 tool_calls → 精确复现真实劫持现场）均已还原 - 附：修 session_turn_fence.go 里引用旧函数名的陈旧注释
118 lines
5.1 KiB
Go
118 lines
5.1 KiB
Go
package loop
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"fmt"
|
||
|
||
"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"
|
||
)
|
||
|
||
// checkpointStore 是循环持久化持久快照并恢复被中断
|
||
// run 所需的最小表面（design §1 durable execution）。
|
||
// 生产环境由 *persist.Store 满足；测试替换为假实现，以在
|
||
// 没有真实数据库的情况下断言恢复路径。
|
||
type checkpointStore interface {
|
||
// Save 插入或更新一个 checkpoint。
|
||
Save(ctx context.Context, cp *persist.Checkpoint) error
|
||
// LoadLatest 返回 agentID 最近更新的 checkpoint；
|
||
// 该 agent 没有时返回 ErrNotFound。
|
||
LoadLatest(ctx context.Context, agentID string) (*persist.Checkpoint, error)
|
||
}
|
||
|
||
// checkpointState 是每步持久化的持久快照。携带消息
|
||
// 窗口与步索引，使长任务能从最新 checkpoint 恢复而不
|
||
// 丢失上下文（design §1 durable execution）。
|
||
type checkpointState struct {
|
||
Step int `json:"step"`
|
||
Messages []model.Message `json:"messages"`
|
||
StopReason string `json:"stop_reason,omitempty"`
|
||
}
|
||
|
||
// saveCheckpoint 在 step 处持久化一个快照。它绝不让 run 失败：
|
||
// store 为 nil 时为空操作，marshal/save 出错则记录日志。
|
||
func (l *Loop) saveCheckpoint(ctx context.Context, sessionID string, step int, messages []model.Message, stopReason string) {
|
||
if l.checkpoint == nil {
|
||
return
|
||
}
|
||
state := checkpointState{
|
||
Step: step,
|
||
Messages: messages,
|
||
StopReason: stopReason,
|
||
}
|
||
buf, err := json.Marshal(state)
|
||
if err != nil {
|
||
l.log.Warn("checkpoint marshal", "step", step, "error", err)
|
||
return
|
||
}
|
||
cp := &persist.Checkpoint{
|
||
ID: fmt.Sprintf("%s:%d", sessionID, step),
|
||
AgentID: sessionID,
|
||
Step: step,
|
||
State: buf,
|
||
}
|
||
if err := l.checkpoint.Save(ctx, cp); err != nil {
|
||
l.log.Warn("checkpoint save", "step", step, "error", err)
|
||
}
|
||
}
|
||
|
||
// loadCheckpoint 从该会话最新的持久 checkpoint 恢复一个**崩溃遗留**的
|
||
// run（design/04 §1 durable execution）。它返回恢复后的消息
|
||
// 窗口、下一个要运行的循环步（checkpoint 步 + 1），以及
|
||
// 是否找到了可恢复的 checkpoint。不存在 checkpoint，或最新的一个
|
||
// 反映的是已完成/被阻塞的 run、或任何已落定的未完成退出（取消 / 报错 /
|
||
// 预算耗尽 / 护栏拦截）时，它返回调用方的消息并置步为 1，使循环从新开始
|
||
// （主人 2026-10-04 裁决：只有崩溃才允许恢复；见 resumableStopReason）。
|
||
//
|
||
// 恢复是幂等的：checkpoint 精确保存了上一个已完成步结束时
|
||
// 的消息窗口（assistant 工具调用及其结果已融合在内），
|
||
// 因此从下一步恢复会由历史重建模型
|
||
// 上下文，绝不重跑已完成步的工具调用，也不
|
||
// 重新生成已完成的答案。加载是只读的 —— 恢复绝不删除
|
||
// checkpoint，因此每一步的快照始终可查询。
|
||
func (l *Loop) loadCheckpoint(ctx context.Context, sessionID string, messages []model.Message) ([]model.Message, int, bool) {
|
||
if l.checkpoint == nil {
|
||
return messages, 1, false
|
||
}
|
||
cp, err := l.checkpoint.LoadLatest(ctx, sessionID)
|
||
if err != nil {
|
||
if !errors.Is(err, persist.ErrNotFound) {
|
||
l.log.Warn("checkpoint load", "session", sessionID, "error", err)
|
||
}
|
||
return messages, 1, false
|
||
}
|
||
st := checkpointState{}
|
||
if err := json.Unmarshal(cp.State, &st); err != nil {
|
||
l.log.Warn("checkpoint unmarshal", "session", sessionID, "step", cp.Step, "error", err)
|
||
return messages, 1, false
|
||
}
|
||
if !resumableStopReason(st.StopReason) {
|
||
l.log.Debug("checkpoint not resumable", "session", sessionID, "step", cp.Step, "stop_reason", st.StopReason)
|
||
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
|
||
}
|
||
|
||
// resumableStopReason 报告最后一个步以上述 stop reason 结束的
|
||
// checkpoint 是否代表一个**崩溃遗留**的在途 run。只有落在工具循环中途的步
|
||
// 算「崩溃遗留」：循环已把工具结果并入窗口并即将继续，进程却猝死了——
|
||
// 此时收尾 defer 根本没执行，快照是进程最后留下的证据，从下一步接上会由
|
||
// 历史重建上下文而不重复工作（design/04 §1 durable execution）。
|
||
//
|
||
// 其余一律落到全新开始（主人 2026-10-04 裁决：只有崩溃才允许恢复）：
|
||
// - 携带最终答案、护栏拒绝、预算耗尽或安全网终止的 checkpoint 是
|
||
// 已完成/被阻塞的 run——恢复它会重新生成已产出的答案（重复副作用）；
|
||
// - canceled / aborted 是**已落定**的未完成退出，下一回合必须自己组
|
||
// 上下文（裁决 13.4.5），不得被旧窗口顶掉；
|
||
// - 空的 stop reason 无法证明是崩溃遗留（崩溃留下的快照必带
|
||
// tool_calls），保守起见同样不恢复。
|
||
//
|
||
// 打开真正新会话的调用方使用不同的 sessionID，本就错过 LoadLatest。
|
||
func resumableStopReason(stopReason string) bool {
|
||
return stopReason == agent.StopReasonToolCalls
|
||
}
|