diff --git a/checkpoint.go b/checkpoint.go index febb316..a3ae4d8 100644 --- a/checkpoint.go +++ b/checkpoint.go @@ -15,50 +15,88 @@ type Checkpointer[S any] interface { Save(ctx context.Context, threadID string, thread *Thread[S]) error // Load 读取之前保存的线程快照,用于恢复执行。 Load(ctx context.Context, threadID string) (*Thread[S], error) + // Delete 删除指定线程的快照文件,用于工作流完结后的清理。 + Delete(ctx context.Context, threadID string) error } // FileCheckpointer 实现了基于本地 JSON 文件的状态持久化。 type FileCheckpointer[S any] struct { - dir string // 存放状态文件的文件夹路径 + dir string // 存放状态文件的文件夹路径 + pretty bool // 是否使用缩进格式(方便人工阅读,但体积更大) +} + +// FileCheckpointerOption 是 FileCheckpointer 的可选配置项。 +type FileCheckpointerOption func(*fileCheckpointerConfig) + +type fileCheckpointerConfig struct { + pretty bool +} + +// WithPrettyJSON 开启 JSON 缩进格式(方便人工阅读,适合开发调试阶段)。 +// 默认关闭,使用紧凑格式以减少磁盘 I/O 和文件体积。 +func WithPrettyJSON() FileCheckpointerOption { + return func(c *fileCheckpointerConfig) { + c.pretty = true + } } // NewFileCheckpointer 创建一个文件检查点管理器。 -func NewFileCheckpointer[S any](dir string) (*FileCheckpointer[S], error) { +// 支持通过 opts 传入可选配置(如 WithPrettyJSON())。 +func NewFileCheckpointer[S any](dir string, opts ...FileCheckpointerOption) (*FileCheckpointer[S], error) { // 确保文件夹存在,不存在则创建 if err := os.MkdirAll(dir, 0755); err != nil { return nil, fmt.Errorf("failed to create checkpoint dir: %w", err) } - return &FileCheckpointer[S]{dir: dir}, nil + cfg := &fileCheckpointerConfig{} + for _, opt := range opts { + opt(cfg) + } + + return &FileCheckpointer[S]{dir: dir, pretty: cfg.pretty}, nil } // Save 将 Thread 序列化为 JSON 并写入本地文件。 -// 采用“先写临时文件再原子重命名”的方式,防止写入中途程序崩溃导致文件损坏。 +// 采用"先写临时文件再原子重命名"的方式,防止写入中途程序崩溃导致文件损坏。 +// 若存在上次崩溃残留的临时文件,会在写入前自动清理。 func (fc *FileCheckpointer[S]) Save(ctx context.Context, threadID string, thread *Thread[S]) error { if err := validateThreadID(threadID); err != nil { return err } path := filepath.Join(fc.dir, threadID+".json") + tmpPath := path + ".tmp" - // 使用 MarshalIndent 方便人工阅读生成的 JSON 文件 - data, err := json.MarshalIndent(thread, "", " ") + // 序列化:生产环境默认使用紧凑格式以减少磁盘 I/O;开发阶段可用 WithPrettyJSON() 开启缩进 + var data []byte + var err error + if fc.pretty { + data, err = json.MarshalIndent(thread, "", " ") + } else { + data, err = json.Marshal(thread) + } if err != nil { return fmt.Errorf("failed to marshal thread: %w", err) } + // 清理上次崩溃可能残留的临时文件(忽略不存在的错误) + _ = os.Remove(tmpPath) + // 原子写入:先写临时文件,再 Rename 替换,确保断电/崩溃安全 - tmpPath := path + ".tmp" if err := os.WriteFile(tmpPath, data, 0644); err != nil { return fmt.Errorf("failed to write temp file: %w", err) } if err := os.Rename(tmpPath, path); err != nil { + // Rename 失败时,尝试清理临时文件,避免残留 + _ = os.Remove(tmpPath) return fmt.Errorf("failed to rename temp file: %w", err) } return nil } // Load 从本地文件中读取 JSON 并反序列化为 Thread。 +// 若加载到一个已完结(IsFinished == true)的线程,会包装 ErrAlreadyFinished 作为额外信息返回, +// 方便调用方做幂等处理(注意:此时 thread 依然有效,err 可通过 errors.Is 判断)。 func (fc *FileCheckpointer[S]) Load(ctx context.Context, threadID string) (*Thread[S], error) { if err := validateThreadID(threadID); err != nil { return nil, err @@ -76,9 +114,29 @@ func (fc *FileCheckpointer[S]) Load(ctx context.Context, threadID string) (*Thre return nil, fmt.Errorf("failed to unmarshal thread: %w", err) } + // 如果加载到已完结的线程,附带哨兵错误提示,方便调用方感知 + if thread.IsFinished { + return &thread, fmt.Errorf("thread %q has already finished: %w", threadID, ErrAlreadyFinished) + } + return &thread, nil } +// Delete 删除指定 threadID 对应的快照文件。 +// 通常在工作流完结后调用,防止快照文件无限积累占用磁盘空间。 +// 若文件不存在,静默返回 nil(幂等操作)。 +func (fc *FileCheckpointer[S]) Delete(ctx context.Context, threadID string) error { + if err := validateThreadID(threadID); err != nil { + return err + } + + path := filepath.Join(fc.dir, threadID+".json") + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("failed to delete checkpoint file: %w", err) + } + return nil +} + // validateThreadID 校验 threadID 的合法性,防止路径穿越(path traversal)攻击。 // threadID 中不允许出现目录分隔符或 ".." 序列。 func validateThreadID(id string) error { diff --git a/doc.go b/doc.go index 51b5351..b60e43f 100644 --- a/doc.go +++ b/doc.go @@ -11,7 +11,7 @@ // - [Engine] 增强包装器:通过 [NewEngine] 创建,在 CompiledGraph 基础上叠加 // StateCloner(深拷贝)、MaxSteps(死循环熔断)、PreNodeHook/PostNodeHook(生命周期钩子)。 // - [Thread] 执行快照:承载当前状态、下一节点、暂停/结束标记,可序列化用于断点恢复。 -// - [Checkpointer] 持久化接口:内置 [FileCheckpointer] 实现基于 JSON 文件的状态存储。 +// - [Checkpointer] 持久化接口:提供 Save/Load/Delete 用于工作流进度存储和清理,内置 [FileCheckpointer] 实现基于 JSON 文件的状态存储。 // // # 零外部依赖 // diff --git a/engine.go b/engine.go index f92deef..45c92bd 100644 --- a/engine.go +++ b/engine.go @@ -27,6 +27,11 @@ type CompiledGraph[S any] struct { // Compile 校验图结构的合法性,并生成可运行的 CompiledGraph。 func (g *Graph[S]) Compile() (*CompiledGraph[S], error) { + // 首先展示所有注册阶段积累的错误(如重复注册节点或边) + if err := errorsJoin(g.errs); err != nil { + return nil, fmt.Errorf("graph registration errors:\n%w", err) + } + if len(g.nodes) == 0 { return nil, errors.New("cannot compile graph: graph contains no nodes") } @@ -133,6 +138,18 @@ var ErrNotPaused = errors.New("cannot resume: thread is not paused") // ErrAlreadyFinished 在调用 Resume 时线程已结束时返回。可用于 errors.Is 判断。 var ErrAlreadyFinished = errors.New("cannot resume: thread is already finished") +// validateResume 验证线程是否满足恢复执行的前提条件。 +// CompiledGraph.Resume 和 Engine.Resume 共用此逻辑,避免重复实现。 +func validateResume[S any](thread *Thread[S]) error { + if !thread.IsPaused { + return ErrNotPaused + } + if thread.IsFinished { + return ErrAlreadyFinished + } + return nil +} + // Start 从指定的起始节点开始运行图,并持续流转,直到遇到中断或运行结束 func (cg *CompiledGraph[S]) Start(ctx context.Context, startNode string, initialState S) (*Thread[S], error) { thread := &Thread[S]{ @@ -145,11 +162,8 @@ func (cg *CompiledGraph[S]) Start(ctx context.Context, startNode string, initial // Resume 恢复执行一个被暂停的线程,并允许注入外部修改后的状态数据(例如人工审批修改后的结果) func (cg *CompiledGraph[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState S) (*Thread[S], error) { - if !thread.IsPaused { - return nil, ErrNotPaused - } - if thread.IsFinished { - return nil, ErrAlreadyFinished + if err := validateResume(thread); err != nil { + return nil, err } // 注入人工修改后的状态,并解除暂停标记 @@ -172,17 +186,6 @@ func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S], opts run stepCount := 0 for { - // 【熔断 1】检查步数上限,防御死循环(infinite loop protection) - if opts.maxSteps > 0 { - if stepCount >= opts.maxSteps { - return thread, fmt.Errorf( - "engine halt: max steps (%d) exceeded, possible infinite loop in graph", - opts.maxSteps, - ) - } - stepCount++ - } - // 【熔断 2】检查 Context,支持外部超时控制或取消 select { case <-ctx.Done(): @@ -197,6 +200,17 @@ func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S], opts run return thread, nil } + // 【熔断 1】检查步数上限,防御死循环(infinite loop protection) + if opts.maxSteps > 0 { + if stepCount >= opts.maxSteps { + return thread, fmt.Errorf( + "engine halt: max steps (%d) exceeded, possible infinite loop in graph", + opts.maxSteps, + ) + } + stepCount++ + } + // 查找当前节点 nodeFn, exists := cg.nodes[currentNodeName] if !exists { @@ -242,6 +256,12 @@ func (cg *CompiledGraph[S]) run(ctx context.Context, thread *Thread[S], opts run if err != nil { return thread, fmt.Errorf("router for node %q execution error: %w", currentNodeName, err) } + // 立即校验返回的节点是否存在,提供更清晰的报错 + if next != "" { + if _, exists := cg.nodes[next]; !exists { + return thread, fmt.Errorf("router for node %q returned non-existent node %q (not found)", currentNodeName, next) + } + } nextNode = next } else { // 否则使用静态连线边 @@ -322,18 +342,26 @@ func (cg *CompiledGraph[S]) runParallelBranches( wg.Wait() close(resultCh) - // 按索引顺序收集结果,保证合并函数收到的 branches 顺序与 targets 一致 + // 【P0 修复】先收集所有结果,再判断是否有真实错误。 + // 原先在循环中遇到非 context.Canceled 错误就立即 return,但由于 channel 中的写入顺序是不确定的, + // 存在以下竞态:分支 A 先把 context.Canceled 写入 channel,紧接着分支 B 才把真实错误写入。 + // 若此时早返回,真实错误永远不会被读取,调用方可能收到 nil error 而实际上有分支失败了。 + // 解决方案:等全部结果到齐后,再过滤出第一个非取消的真实错误。 branches := make([]S, len(targets)) + var firstErr error for r := range resultCh { if r.err != nil { - // 过滤掉因取消产生的次生错误,只返回第一个真实错误 - if !errors.Is(r.err, context.Canceled) { - return nil, r.err + // 过滤掉因取消产生的次生错误,只记录第一个真实错误 + if !errors.Is(r.err, context.Canceled) && firstErr == nil { + firstErr = r.err } } else { branches[r.idx] = r.state } } + if firstErr != nil { + return nil, firstErr + } return branches, nil } diff --git a/engine_options.go b/engine_options.go index 329d822..b1df31e 100644 --- a/engine_options.go +++ b/engine_options.go @@ -1,6 +1,9 @@ package GopherGraph -import "context" +import ( + "context" + "sync" +) // Engine 是 CompiledGraph 的增强包装器,在完全兼容原有 API 的基础上, // 提供以下四项生产级能力: @@ -12,6 +15,12 @@ import "context" // // 所有选项均为可选,通过链式调用注册。未注册的选项将退回到 CompiledGraph 的默认行为。 // +// # 线程安全说明 +// +// Engine 使用 sync.RWMutex 保护配置字段,WithXxx 配置方法和 opts() 读取方法均加锁。 +// 因此,允许在 Start/Resume 运行期间并发调用 WithXxx 修改配置(但不推荐这样做)。 +// 常见的安全模式:在服务启动时一次性完成链式配置,之后多个 Goroutine 并发调用 Start/Resume。 +// // 用法示例: // // engine := GopherGraph.NewEngine(compiledGraph). @@ -30,6 +39,7 @@ import "context" // // thread, err := engine.Start(ctx, "start", initialState) type Engine[S any] struct { + mu sync.RWMutex graph *CompiledGraph[S] stateCloner func(S) S maxSteps int @@ -50,6 +60,8 @@ func NewEngine[S any](graph *CompiledGraph[S]) *Engine[S] { // 若不注册,引擎退化为 Go 默认的值拷贝语义(浅拷贝)。 // 对于状态中含有引用类型的场景,强烈建议注册此函数。 func (e *Engine[S]) WithStateCloner(fn func(S) S) *Engine[S] { + e.mu.Lock() + defer e.mu.Unlock() e.stateCloner = fn return e } @@ -62,6 +74,8 @@ func (e *Engine[S]) WithStateCloner(fn func(S) S) *Engine[S] { // 建议根据工作流的预期最大深度合理设置(如 100~1000)。 // 传入 0 或负数表示不限制(与不调用此方法等效)。 func (e *Engine[S]) WithMaxSteps(n int) *Engine[S] { + e.mu.Lock() + defer e.mu.Unlock() e.maxSteps = n return e } @@ -73,9 +87,12 @@ func (e *Engine[S]) WithMaxSteps(n int) *Engine[S] { // - 记录 OpenTelemetry Span 的开始时间 // - 向前端推送"正在执行 xxx 节点"的流式事件 // +// 支持多次调用:每次调用将新钩子追加到执行链末尾,按注册顺序依次执行,不覆盖已有钩子。 // 钩子函数接收到的 state 是节点执行【前】的状态快照,请勿修改(只读语义)。 // 钩子中的 panic 会向上传播,建议在内部做好 recover。 func (e *Engine[S]) WithPreNodeHook(fn HookFn[S]) *Engine[S] { + e.mu.Lock() + defer e.mu.Unlock() if e.preNodeHook == nil { e.preNodeHook = fn } else { @@ -95,8 +112,11 @@ func (e *Engine[S]) WithPreNodeHook(fn HookFn[S]) *Engine[S] { // - 上报节点耗时指标(Prometheus/DataDog) // - 向前端推送节点执行完毕的流式事件 // +// 支持多次调用:每次调用将新钩子追加到执行链末尾,按注册顺序依次执行,不覆盖已有钩子。 // 钩子函数接收到的 state 是节点执行【后】更新过的状态,同为只读语义。 func (e *Engine[S]) WithPostNodeHook(fn HookFn[S]) *Engine[S] { + e.mu.Lock() + defer e.mu.Unlock() if e.postNodeHook == nil { e.postNodeHook = fn } else { @@ -109,8 +129,11 @@ func (e *Engine[S]) WithPostNodeHook(fn HookFn[S]) *Engine[S] { return e } -// opts 将 Engine 的配置打包成 runOptions,传入底层 run() 调度器。 +// opts 将 Engine 的配置快照打包成 runOptions,传入底层 run() 调度器。 +// 使用 RLock 保证读取配置时的并发安全。 func (e *Engine[S]) opts() runOptions[S] { + e.mu.RLock() + defer e.mu.RUnlock() return runOptions[S]{ maxSteps: e.maxSteps, stateCloner: e.stateCloner, @@ -131,12 +154,10 @@ func (e *Engine[S]) Start(ctx context.Context, startNode string, initialState S) // Resume 恢复执行一个被暂停的线程,行为与 CompiledGraph.Resume 完全一致, // 但会附加所有已注册的 Engine 增强选项(Hooks、MaxSteps、StateCloner)。 +// 复用 validateResume 进行前提条件校验,与 CompiledGraph.Resume 保持一致。 func (e *Engine[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState S) (*Thread[S], error) { - if !thread.IsPaused { - return nil, ErrNotPaused - } - if thread.IsFinished { - return nil, ErrAlreadyFinished + if err := validateResume(thread); err != nil { + return nil, err } thread.State = modifiedState @@ -144,5 +165,3 @@ func (e *Engine[S]) Resume(ctx context.Context, thread *Thread[S], modifiedState return e.graph.run(ctx, thread, e.opts()) } - - diff --git a/examples/hooks/main.go b/examples/hooks/main.go index 9bfacdf..a1131ba 100644 --- a/examples/hooks/main.go +++ b/examples/hooks/main.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "sync" "time" GopherGraph "github.com/unclesam-ly/GopherGraph" @@ -82,7 +83,10 @@ func main() { // ============================================================ // 使用 Engine 包装器,挂载所有增强能力 // ============================================================ - var nodeTimings = make(map[string]time.Time) + var ( + nodeTimings = make(map[string]time.Time) + nodeTimingsMu sync.Mutex + ) engine := GopherGraph.NewEngine(cg). // 1. 注册深拷贝函数,消除并发分支对 Messages 切片的数据竞争 @@ -91,13 +95,17 @@ func main() { WithMaxSteps(50). // 3. Pre Hook:记录节点开始时间 + 向控制台推送"流式进度" WithPreNodeHook(func(ctx context.Context, name string, s AgentState) { + nodeTimingsMu.Lock() nodeTimings[name] = time.Now() + nodeTimingsMu.Unlock() fmt.Printf(" ⏳ [PRE ] %-15s | 消息数: %d\n", name, len(s.Messages)) }). // 4. Post Hook:计算并打印节点耗时 WithPostNodeHook(func(ctx context.Context, name string, s AgentState) { + nodeTimingsMu.Lock() elapsed := time.Since(nodeTimings[name]) - fmt.Printf(" ✅ [POST] %-15s | 耗时: %v | 消息数: %d\n", name, elapsed.Round(time.Millisecond), len(s.Messages)) + nodeTimingsMu.Unlock() + fmt.Printf(" [POST] %-15s | 耗时: %v | 消息数: %d\n", name, elapsed.Round(time.Millisecond), len(s.Messages)) }) fmt.Println("🚀 启动工作流...") diff --git a/examples/translation/main.go b/examples/translation/main.go index 6ef91fa..29e65a6 100644 --- a/examples/translation/main.go +++ b/examples/translation/main.go @@ -6,6 +6,7 @@ import ( "fmt" "os" "strings" + "time" GopherGraph "github.com/unclesam-ly/GopherGraph" ) @@ -122,7 +123,8 @@ func main() { // 包含敏感词“秘密”,会触发人工审核中断: inputText := "这是一个秘密你好" fmt.Printf("🚀 启动工作流,输入原文: %q\n", inputText) - ctx := context.Background() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() thread, err := cg.Start(ctx, "translator", TranslationState{InputText: inputText}) if err != nil { fmt.Printf("运行失败: %v\n", err) diff --git a/graph.go b/graph.go index 98d67ad..da8354e 100644 --- a/graph.go +++ b/graph.go @@ -1,6 +1,10 @@ package GopherGraph -import "context" +import ( + "context" + "errors" + "fmt" +) // NodeFn 代表图中的一个执行节点(Agent)。 // 它接收当前的上下文和状态 S,处理后返回更新后的状态 S,或者返回错误。 @@ -27,6 +31,7 @@ type Graph[S any] struct { conditional map[string]RouterFn[S] // 条件路由边:源节点 -> 路由函数 interrupts map[string]bool // 标记需要中断的节点:执行完该节点后挂起 parallels map[string]parallelStep[S] // 并发分支:源节点 -> 并发步骤配置 + errs []error // 注册阶段的错误积累(重复注册等),在 Compile 时统一报告 } // NewGraph 创建并初始化一个全新强类型的图构建器 @@ -36,27 +41,43 @@ func NewGraph[S any]() *Graph[S] { edges: make(map[string]string), conditional: make(map[string]RouterFn[S]), interrupts: make(map[string]bool), - parallels: make(map[string]parallelStep[S]), // 初始化 parallels + parallels: make(map[string]parallelStep[S]), } } -// AddNode 向图中注册一个节点(Agent) +// AddNode 向图中注册一个节点(Agent)。 +// 若同名节点已被注册,不会覆盖原有节点,错误将在 Compile() 时统一报告。 func (g *Graph[S]) AddNode(name string, fn NodeFn[S]) { + if _, exists := g.nodes[name]; exists { + g.errs = append(g.errs, fmt.Errorf("AddNode: node %q is already registered", name)) + return + } g.nodes[name] = fn } -// AddEdge 建立一条从 from 节点到 to 节点的静态连接线 +// AddEdge 建立一条从 from 节点到 to 节点的静态连接线。 +// 若 from 节点已有静态出边,不会覆盖,错误将在 Compile() 时统一报告。 func (g *Graph[S]) AddEdge(from, to string) { + if _, exists := g.edges[from]; exists { + g.errs = append(g.errs, fmt.Errorf("AddEdge: node %q already has a static edge registered", from)) + return + } g.edges[from] = to } // AddConditionalEdges 建立一条从 from 节点出发的条件路由 -// 到底去哪,由 router 函数在运行时根据当前 State 动态决定 +// 到底去哪,由 router 函数在运行时根据当前 State 动态决定。 +// 若 from 节点已有条件出边,不会覆盖,错误将在 Compile() 时统一报告。 func (g *Graph[S]) AddConditionalEdges(from string, router RouterFn[S]) { + if _, exists := g.conditional[from]; exists { + g.errs = append(g.errs, fmt.Errorf("AddConditionalEdges: node %q already has a conditional edge registered", from)) + return + } g.conditional[from] = router } -// AddInterrupt 标记在执行 nodeName 节点“之前”进行中断挂起,等待人工确认 +// AddInterrupt 标记在执行 nodeName 节点"之前"进行中断挂起,等待人工确认。 +// 重复标记同一节点为幂等操作(设置 true 再设置 true,无副作用)。 func (g *Graph[S]) AddInterrupt(nodeName string) { g.interrupts[nodeName] = true } @@ -66,7 +87,12 @@ func (g *Graph[S]) AddInterrupt(nodeName string) { // targets: 需要并发执行的多个目标节点名称(如 "translate_en", "translate_fr") // next: 所有并发节点运行完并合并后,下一步要去的节点(如 "publisher") // merger: 用于合并各个并发分支状态的函数 +// 若 from 节点已有并发出边,不会覆盖,错误将在 Compile() 时统一报告。 func (g *Graph[S]) AddParallelEdges(from string, targets []string, next string, merger ParallelMergeFn[S]) { + if _, exists := g.parallels[from]; exists { + g.errs = append(g.errs, fmt.Errorf("AddParallelEdges: node %q already has parallel edges registered", from)) + return + } g.parallels[from] = parallelStep[S]{ targets: targets, next: next, @@ -78,3 +104,12 @@ func (g *Graph[S]) AddParallelEdges(from string, targets []string, next string, // nodeName 为当前执行的节点名称,state 为该节点执行时的状态快照(只读语义,请勿修改)。 // 可通过此钩子接入日志、链路追踪(OpenTelemetry)、监控指标等能力,而无需修改任何 NodeFn。 type HookFn[S any] func(ctx context.Context, nodeName string, state S) + +// errorsJoin 兼容多个错误合并为一个,避免 errors.Join 输出格式不够友好。 +// 仅供内部 graph.errs 汇总使用。 +func errorsJoin(errs []error) error { + if len(errs) == 0 { + return nil + } + return errors.Join(errs...) +} diff --git a/graph_test.go b/graph_test.go index d82492e..0c4904b 100644 --- a/graph_test.go +++ b/graph_test.go @@ -3,6 +3,7 @@ package GopherGraph import ( "context" "errors" + "os" "strings" "testing" "time" @@ -256,7 +257,7 @@ func TestParallelExecution(t *testing.T) { func TestFileCheckpointer(t *testing.T) { // 创建测试临时文件夹(测试结束后 Go 会自动清理) tmpDir := t.TempDir() - fc, err := NewFileCheckpointer[TestState](tmpDir) + fc, err := NewFileCheckpointer[TestState](tmpDir, WithPrettyJSON()) if err != nil { t.Fatalf("创建文件检查点管理器失败: %v", err) } @@ -316,6 +317,28 @@ func TestFileCheckpointer(t *testing.T) { if !thread.IsFinished || thread.State.Value != 700 { t.Errorf("执行结果错误,最终值: %d, 是否结束: %t", thread.State.Value, thread.IsFinished) } + + // 5. 保存已完结的线程,Load 后应附带 ErrAlreadyFinished 提示 + if err := fc.Save(ctx, sessionID, thread); err != nil { + t.Fatalf("保存已完结线程失败: %v", err) + } + _, loadErr := fc.Load(ctx, sessionID) + if !errors.Is(loadErr, ErrAlreadyFinished) { + t.Errorf("Load 已完结线程应附带 ErrAlreadyFinished,实际: %v", loadErr) + } + + // 6. Delete 后文件应不存在 + if err := fc.Delete(ctx, sessionID); err != nil { + t.Fatalf("删除检查点失败: %v", err) + } + _, err = fc.Load(ctx, sessionID) + if err == nil { + t.Error("删除后 Load 应报错(文件不存在)") + } + // 幂等:重复 Delete 不应报错 + if err := fc.Delete(ctx, sessionID); err != nil { + t.Errorf("重复 Delete 应幂等不报错,实际: %v", err) + } } // TestMaxStepsCircuitBreaker 验证 Engine.WithMaxSteps 能在步数超限时熔断,防止死循环 @@ -512,6 +535,35 @@ func TestCompileValidationErrors(t *testing.T) { if err == nil || !strings.Contains(err.Error(), "interrupt") { t.Errorf("不存在的中断节点应报错: %v", err) } + + // 4. 重复注册节点和边 + g4 := NewGraph[TestState]() + g4.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { return s, nil }) + g4.AddNode("A", func(ctx context.Context, s TestState) (TestState, error) { return s, nil }) // 重复节点 + g4.AddEdge("A", "B") + g4.AddEdge("A", "C") // 重复静态边 + g4.AddConditionalEdges("A", func(ctx context.Context, s TestState) (string, error) { return "B", nil }) + g4.AddConditionalEdges("A", func(ctx context.Context, s TestState) (string, error) { return "C", nil }) // 重复条件边 + g4.AddParallelEdges("A", []string{"B"}, "", func(ctx context.Context, parent TestState, branches []TestState) (TestState, error) { return parent, nil }) + g4.AddParallelEdges("A", []string{"C"}, "", func(ctx context.Context, parent TestState, branches []TestState) (TestState, error) { return parent, nil }) // 重复并发边 + + _, err = g4.Compile() + if err == nil { + t.Fatal("重复注册应报错") + } + errMsg := err.Error() + if !strings.Contains(errMsg, "already registered") { + t.Errorf("应包含节点重复错误: %v", err) + } + if !strings.Contains(errMsg, "static edge registered") { + t.Errorf("应包含静态边重复错误: %v", err) + } + if !strings.Contains(errMsg, "conditional edge registered") { + t.Errorf("应包含条件边重复错误: %v", err) + } + if !strings.Contains(errMsg, "parallel edges registered") { + t.Errorf("应包含并发边重复错误: %v", err) + } } // TestCompileEdgeConflict 验证同一节点不能同时存在于多种出边类型中 @@ -673,5 +725,36 @@ func TestCheckpointerThreadIDValidation(t *testing.T) { if err := fc.Save(ctx, "valid-id_123", thread); err != nil { t.Errorf("合法 threadID 不应报错: %v", err) } + + // Delete 校验 + if err := fc.Delete(ctx, "../escape"); err == nil { + t.Error("Delete 包含 .. 的 threadID 应报错") + } } +// TestFileCheckpointerCompactJSON 验证默认格式为紧凑 JSON(不带缩进) +func TestFileCheckpointerCompactJSON(t *testing.T) { + tmpDir := t.TempDir() + + // 默认紧凑格式 + fc, _ := NewFileCheckpointer[TestState](tmpDir) + thread := &Thread[TestState]{State: TestState{Value: 42}, NextNode: "A"} + _ = fc.Save(context.Background(), "compact", thread) + + data, _ := os.ReadFile(tmpDir + "/compact.json") + if len(data) == 0 { + t.Fatal("文件为空") + } + // 紧凑格式不应包含换行符 + if strings.Contains(string(data), "\n") { + t.Error("默认模式应为紧凑 JSON,不应包含换行符") + } + + // WithPrettyJSON 模式应包含换行符 + fcPretty, _ := NewFileCheckpointer[TestState](tmpDir, WithPrettyJSON()) + _ = fcPretty.Save(context.Background(), "pretty", thread) + dataPretty, _ := os.ReadFile(tmpDir + "/pretty.json") + if !strings.Contains(string(dataPretty), "\n") { + t.Error("WithPrettyJSON 模式应包含换行符") + } +} diff --git a/scratch/testmod/go.mod b/scratch/testmod/go.mod new file mode 100644 index 0000000..3551eb6 --- /dev/null +++ b/scratch/testmod/go.mod @@ -0,0 +1,5 @@ +module testmod + +go 1.26.2 + +require github.com/unclesam-ly/GopherGraph v1.1.1 // indirect diff --git a/scratch/testmod/go.sum b/scratch/testmod/go.sum new file mode 100644 index 0000000..bd9b30a --- /dev/null +++ b/scratch/testmod/go.sum @@ -0,0 +1,2 @@ +github.com/unclesam-ly/GopherGraph v1.1.1 h1:1jou9QFtsom/eK69Evx+J3HLkZTjKjnuEuRvkir8qi4= +github.com/unclesam-ly/GopherGraph v1.1.1/go.mod h1:+99WeqLyTIYfOs2OSob/SQ3Pnw+R7pxW8+o6D7S4m6g=