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 }