Blizzard
f3bf42c432
HITL 持久化中断/恢复的地基：compose checkpoint 需要一个可持久化、抗重启的 存储后端。dispatcher 是"只说 NATS、无 DB"的纯净设计，故复用既有 JetStream （bus.js）开 KV 桶，不引 Redis、不破坏架构原则。 - shared/bus: 新增 KVHandle + Bus.Checkpoints(bucket, ttl)。薄封装把 NATS 细节（ErrKeyNotFound→ok=false、Delete 幂等）挡在 bus 内，对外是朴素 Get/Put/Delete；bus 无需反向依赖 eino。File 存储 + 桶级 TTL 兜底清理。 - dispatcher/eino: checkpointStore 把 CheckpointKV 适配成 compose.CheckPointStore （Get/Set + 可选 Delete）。CheckpointKV 是最小接口，bus.KVHandle 结构化满足。 - 测试: 内存桩往返（Set→Get→Delete→miss）+ 编译期契约断言 `var _ CheckpointKV = (*bus.KVHandle)(nil)` 钉死 bus↔eino 隐式契约。 零爆炸半径（纯新增）。go test ./... 全绿。 下一步增量2：审批节点改 compose.Interrupt + Orchestrator 识别中断置 waiting。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
45 lines
2.1 KiB
Go
45 lines
2.1 KiB
Go
package eino
|
||
|
||
import (
|
||
"context"
|
||
|
||
"github.com/cloudwego/eino/compose"
|
||
)
|
||
|
||
// CheckpointKV 是 compose checkpoint 的最小持久化后端：朴素的键值 + 软缺失语义。
|
||
// 生产由 bus 的 JetStream KV（*bus.KVHandle）实现（结构化满足，无需反向依赖 bus）；
|
||
// 单测用内存桩。键 = checkPointID（dispatcher 取 task_id），值 = compose 序列化的图状态。
|
||
type CheckpointKV interface {
|
||
Get(ctx context.Context, key string) (val []byte, ok bool, err error)
|
||
Put(ctx context.Context, key string, val []byte) error
|
||
Delete(ctx context.Context, key string) error
|
||
}
|
||
|
||
// checkpointStore 把 CheckpointKV 适配成 Eino compose.CheckPointStore（Get/Set），
|
||
// 并实现可选的 CheckPointDeleter（Delete）——审批恢复 / 终态后显式删 checkpoint，
|
||
// 避免 KV 堆积；桶级 TTL 再兜底清理被遗弃的中断（审批人始终不处理）。
|
||
//
|
||
// 中断模型：审批节点 compose.Interrupt 时，compose 把整图状态写进本 store（key=task_id）；
|
||
// dispatcher 据此置 waiting 并返回（释放 goroutine）。决定到达后以同 key 重入图 → 从断点恢复。
|
||
type checkpointStore struct{ kv CheckpointKV }
|
||
|
||
func newCheckpointStore(kv CheckpointKV) *checkpointStore { return &checkpointStore{kv: kv} }
|
||
|
||
// Get 读 checkpoint；不存在返回 (nil,false,nil)，由 compose 视作"无断点、全新执行"。
|
||
func (s *checkpointStore) Get(ctx context.Context, checkPointID string) ([]byte, bool, error) {
|
||
return s.kv.Get(ctx, checkPointID)
|
||
}
|
||
|
||
// Set 持久化一次中断的图状态。
|
||
func (s *checkpointStore) Set(ctx context.Context, checkPointID string, checkPoint []byte) error {
|
||
return s.kv.Put(ctx, checkPointID, checkPoint)
|
||
}
|
||
|
||
// Delete 显式清理 checkpoint（实现 compose 的可选 CheckPointDeleter，结构化匹配）。
|
||
func (s *checkpointStore) Delete(ctx context.Context, checkPointID string) error {
|
||
return s.kv.Delete(ctx, checkPointID)
|
||
}
|
||
|
||
// 编译期确认实现了 compose.CheckPointStore（Get/Set）。Delete 经结构化断言由运行时识别。
|
||
var _ compose.CheckPointStore = (*checkpointStore)(nil)
|