package eino import ( "context" "sync" "testing" "github.com/sundynix/sundynix-shared/bus" ) // memKV 是 CheckpointKV 的内存桩(并发安全),用于不依赖 NATS 的单测。 type memKV struct { mu sync.Mutex m map[string][]byte } func newMemKV() *memKV { return &memKV{m: map[string][]byte{}} } func (k *memKV) Get(_ context.Context, key string) ([]byte, bool, error) { k.mu.Lock() defer k.mu.Unlock() v, ok := k.m[key] return v, ok, nil } func (k *memKV) Put(_ context.Context, key string, val []byte) error { k.mu.Lock() defer k.mu.Unlock() k.m[key] = append([]byte(nil), val...) return nil } func (k *memKV) Delete(_ context.Context, key string) error { k.mu.Lock() defer k.mu.Unlock() delete(k.m, key) return nil } // bus.KVHandle(JetStream KV 后端)须结构化满足 CheckpointKV——钉死 bus 与 eino 的隐式契约, // 任一侧改了 Get/Put/Delete 签名都会在这里编译失败。 var _ CheckpointKV = (*bus.KVHandle)(nil) // TestCheckpointStoreRoundtrip 验证 compose.CheckPointStore 适配:Set→Get 命中、Delete 后 Get 落空。 func TestCheckpointStoreRoundtrip(t *testing.T) { s := newCheckpointStore(newMemKV()) ctx := context.Background() const id = "task_42" if _, ok, err := s.Get(ctx, id); err != nil || ok { t.Fatalf("空 store 应未命中: ok=%v err=%v", ok, err) } cp := []byte("compose-serialized-state") if err := s.Set(ctx, id, cp); err != nil { t.Fatalf("Set: %v", err) } got, ok, err := s.Get(ctx, id) if err != nil || !ok || string(got) != string(cp) { t.Fatalf("Get 应回放原状态: got=%q ok=%v err=%v", got, ok, err) } if err := s.Delete(ctx, id); err != nil { t.Fatalf("Delete: %v", err) } if _, ok, _ := s.Get(ctx, id); ok { t.Fatalf("Delete 后不应再命中") } }