store: replace Get/Put with Store+StoreEntry model
- Store: Stat, List, Open, Create, Delete, Rename - StoreEntry: Stat, Read, Write, BlockingRead - RunnableStore: Store + Runnable - Open(name) returns StoreEntry (file handle) - OpenStore(id) on SessionStore/BatchStore returns RunnableStore - Remove ReadableStore, WritableStore, ReadWriteStore, BlobStore, BlockingReadStore - Export EntryConfig for external store implementations - Wait/LogInfo no longer require type assertions
This commit is contained in:
Levi Neely
parent
9b0bf8aca8
commit
130bf60708
|
|
@ -103,9 +103,7 @@ func NewBatchStore(cfg BatchStoreConfig) *BatchStore {
|
|||
bs.storeConfig = &storeConfig{
|
||||
StatFn: bs.stat,
|
||||
ListFn: bs.list,
|
||||
GetFn: bs.get,
|
||||
OpenFn: bs.open,
|
||||
PutFn: bs.put,
|
||||
OpenFn: bs.openEntry,
|
||||
DeleteFn: bs.del,
|
||||
CreateFn: func(string) error { return fmt.Errorf("create not supported for batch jobs") },
|
||||
RenameFn: func(string, string) error { return fmt.Errorf("rename not supported for batch jobs") },
|
||||
|
|
@ -158,23 +156,29 @@ func (s *BatchStore) stat(name string) (os.FileInfo, error) {
|
|||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (s *BatchStore) get(name string) ([]byte, error) {
|
||||
func (s *BatchStore) openEntry(name string) (StoreEntry, error) {
|
||||
notBlocking := func(context.Context, string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("blocking read not supported")
|
||||
}
|
||||
switch name {
|
||||
case "new":
|
||||
return []byte(batchNewTemplate), nil
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return &SyntheticFileInfo{Name_: "new", Mode_: 0666, Size_: int64(len(batchNewTemplate))}, nil },
|
||||
ReadFn: func() ([]byte, error) { return []byte(batchNewTemplate), nil },
|
||||
WriteFn: func(data []byte) error { return s.handleNewBatch(strings.TrimSpace(string(data))) },
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
case "idx":
|
||||
return s.index(), nil
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return &SyntheticFileInfo{Name_: "idx", Mode_: 0444, Size_: int64(len(s.index()))}, nil },
|
||||
ReadFn: func() ([]byte, error) { return s.index(), nil },
|
||||
WriteFn: func([]byte) error { return fmt.Errorf("idx: read-only") },
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("%s: not a readable file", name)
|
||||
}
|
||||
|
||||
func (s *BatchStore) put(name string, data []byte) error {
|
||||
if name != "new" {
|
||||
return fmt.Errorf("%s: not writable", name)
|
||||
}
|
||||
return s.handleNewBatch(strings.TrimSpace(string(data)))
|
||||
}
|
||||
|
||||
func (s *BatchStore) del(name string) error {
|
||||
s.mu.Lock()
|
||||
job, ok := s.jobs[name]
|
||||
|
|
@ -470,19 +474,17 @@ var batchJobFiles = []struct {
|
|||
// directory (/b/{id}/*). The file set is fixed.
|
||||
type BatchJobStore struct {
|
||||
*storeConfig
|
||||
Runnable
|
||||
job *batchJob
|
||||
}
|
||||
|
||||
func NewBatchJobStore(job *batchJob) *BatchJobStore {
|
||||
js := &BatchJobStore{job: job}
|
||||
notDir := func(string) (Store, error) { return nil, fmt.Errorf("not a directory") }
|
||||
js := &BatchJobStore{Runnable: job, job: job}
|
||||
notSupported := func(string) error { return fmt.Errorf("not supported") }
|
||||
js.storeConfig = &storeConfig{
|
||||
StatFn: js.stat,
|
||||
ListFn: js.list,
|
||||
GetFn: js.get,
|
||||
OpenFn: notDir,
|
||||
PutFn: func(string, []byte) error { return fmt.Errorf("not supported") },
|
||||
OpenFn: js.open,
|
||||
DeleteFn: notSupported,
|
||||
CreateFn: notSupported,
|
||||
RenameFn: func(string, string) error { return fmt.Errorf("not supported") },
|
||||
|
|
@ -490,7 +492,8 @@ func NewBatchJobStore(job *batchJob) *BatchJobStore {
|
|||
return js
|
||||
}
|
||||
|
||||
func (s *BatchStore) open(id string) (Store, error) {
|
||||
// OpenStore returns a RunnableStore for the given batch job ID.
|
||||
func (s *BatchStore) OpenStore(id string) (RunnableStore, error) {
|
||||
job := s.Job(id)
|
||||
if job == nil {
|
||||
return nil, fmt.Errorf("batch job not found: %s", id)
|
||||
|
|
@ -523,22 +526,56 @@ func (js *BatchJobStore) stat(name string) (os.FileInfo, error) {
|
|||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (js *BatchJobStore) get(name string) ([]byte, error) {
|
||||
if name == "log" {
|
||||
js.job.mu.RLock()
|
||||
data := make([]byte, len(js.job.log))
|
||||
copy(data, js.job.log)
|
||||
js.job.mu.RUnlock()
|
||||
return data, nil
|
||||
}
|
||||
func (js *BatchJobStore) open(name string) (StoreEntry, error) {
|
||||
for _, f := range batchJobFiles {
|
||||
if f.name == name {
|
||||
return []byte(js.content(name)), nil
|
||||
return js.entryFor(name, f.mode), nil
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (js *BatchJobStore) entryFor(name string, mode os.FileMode) StoreEntry {
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) {
|
||||
var size int64
|
||||
if name == "log" {
|
||||
js.job.mu.RLock()
|
||||
size = int64(len(js.job.log))
|
||||
js.job.mu.RUnlock()
|
||||
} else {
|
||||
size = int64(len(js.content(name)))
|
||||
}
|
||||
return &SyntheticFileInfo{Name_: name, Mode_: mode, Size_: size}, nil
|
||||
},
|
||||
ReadFn: func() ([]byte, error) {
|
||||
if name == "log" {
|
||||
js.job.mu.RLock()
|
||||
data := make([]byte, len(js.job.log))
|
||||
copy(data, js.job.log)
|
||||
js.job.mu.RUnlock()
|
||||
return data, nil
|
||||
}
|
||||
return []byte(js.content(name)), nil
|
||||
},
|
||||
WriteFn: func([]byte) error { return fmt.Errorf("%s: read-only", name) },
|
||||
BlockingReadFn: func(ctx context.Context, base string) ([]byte, error) {
|
||||
if name != "statewait" {
|
||||
return nil, fmt.Errorf("%s: not a wait file", name)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, nil
|
||||
case <-js.job.done:
|
||||
}
|
||||
js.job.mu.RLock()
|
||||
state := js.job.state
|
||||
js.job.mu.RUnlock()
|
||||
return []byte(state + "\n"), nil
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (js *BatchJobStore) content(name string) string {
|
||||
js.job.mu.RLock()
|
||||
defer js.job.mu.RUnlock()
|
||||
|
|
@ -557,29 +594,6 @@ func (js *BatchJobStore) content(name string) string {
|
|||
return ""
|
||||
}
|
||||
|
||||
// Wait blocks until the job reaches a terminal state, then returns it.
|
||||
func (js *BatchJobStore) Wait(ctx context.Context, name, _ string) ([]byte, error) {
|
||||
if name != "statewait" {
|
||||
return nil, fmt.Errorf("%s: not a wait file", name)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, nil
|
||||
case <-js.job.done:
|
||||
}
|
||||
js.job.mu.RLock()
|
||||
state := js.job.state
|
||||
js.job.mu.RUnlock()
|
||||
return []byte(state + "\n"), nil
|
||||
}
|
||||
|
||||
// LogInfo returns the current log length and version atomically.
|
||||
func (js *BatchJobStore) LogInfo() (length int, vers uint32) {
|
||||
js.job.mu.RLock()
|
||||
defer js.job.mu.RUnlock()
|
||||
return len(js.job.log), js.job.logVers
|
||||
}
|
||||
|
||||
// LoadAgentConfig resolves and loads the config for a named agent.
|
||||
// Returns nil (not an error) if the config file does not exist;
|
||||
// BuildAgentEnv handles nil configs. open defaults to os.Open if nil.
|
||||
|
|
|
|||
|
|
@ -110,9 +110,7 @@ func NewSessionStore(cfg SessionStoreConfig) *SessionStore {
|
|||
ss.storeConfig = &storeConfig{
|
||||
StatFn: ss.stat,
|
||||
ListFn: ss.list,
|
||||
GetFn: ss.get,
|
||||
OpenFn: ss.open,
|
||||
PutFn: ss.put,
|
||||
OpenFn: ss.openEntry,
|
||||
DeleteFn: ss.del,
|
||||
CreateFn: func(string) error { return fmt.Errorf("create not supported for sessions") },
|
||||
RenameFn: ss.renameSession,
|
||||
|
|
@ -153,26 +151,41 @@ func (s *SessionStore) stat(name string) (os.FileInfo, error) {
|
|||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (s *SessionStore) get(name string) ([]byte, error) {
|
||||
func (s *SessionStore) openEntry(name string) (StoreEntry, error) {
|
||||
notBlocking := func(context.Context, string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("blocking read not supported")
|
||||
}
|
||||
switch name {
|
||||
case "new":
|
||||
return []byte("name=\ncwd=\nbackend=\nmodel=\nagent=\n"), nil
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return &SyntheticFileInfo{Name_: "new", Mode_: 0666}, nil },
|
||||
ReadFn: func() ([]byte, error) { return []byte("name=\ncwd=\nbackend=\nmodel=\nagent=\n"), nil },
|
||||
WriteFn: func(data []byte) error {
|
||||
args := strings.Fields(strings.TrimSpace(string(data)))
|
||||
return s.createSession(args)
|
||||
},
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
case "idx":
|
||||
return s.index(), nil
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return &SyntheticFileInfo{Name_: "idx", Mode_: 0444}, nil },
|
||||
ReadFn: func() ([]byte, error) { return s.index(), nil },
|
||||
WriteFn: func([]byte) error { return fmt.Errorf("idx: read-only") },
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
default:
|
||||
if _, ok := sessionStoreFiles[name]; ok {
|
||||
return s.cfg.ReadFile(paths.CfgDir() + "/scripts/s/" + name)
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return &SyntheticFileInfo{Name_: name, Mode_: sessionStoreFiles[name]}, nil },
|
||||
ReadFn: func() ([]byte, error) {
|
||||
return s.cfg.ReadFile(paths.CfgDir() + "/scripts/s/" + name)
|
||||
},
|
||||
WriteFn: func([]byte) error { return fmt.Errorf("%s: not writable", name) },
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("%s: not a readable file", name)
|
||||
}
|
||||
|
||||
func (s *SessionStore) put(name string, data []byte) error {
|
||||
if name != "new" {
|
||||
return fmt.Errorf("%s: not writable", name)
|
||||
}
|
||||
args := strings.Fields(strings.TrimSpace(string(data)))
|
||||
return s.createSession(args)
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (s *SessionStore) del(name string) error {
|
||||
|
|
@ -193,8 +206,8 @@ func (s *SessionStore) Session(id string) *Session {
|
|||
return s.sessions[id]
|
||||
}
|
||||
|
||||
// SessionFileStore returns a SessionFileStore for the given session ID.
|
||||
func (s *SessionStore) open(id string) (Store, error) {
|
||||
// OpenStore returns a RunnableStore for the given session ID.
|
||||
func (s *SessionStore) OpenStore(id string) (RunnableStore, error) {
|
||||
sess := s.Session(id)
|
||||
if sess == nil {
|
||||
return nil, fmt.Errorf("session not found: %s", id)
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ var SessionFileList = []struct {
|
|||
// session directory. The file set is fixed.
|
||||
type SessionFileStore struct {
|
||||
*storeConfig
|
||||
Runnable
|
||||
sess *Session
|
||||
log *olog.Logger
|
||||
kill func()
|
||||
|
|
@ -51,15 +52,12 @@ type SessionFileStore struct {
|
|||
}
|
||||
|
||||
func NewSessionFileStore(sess *Session, log *olog.Logger, kill func(), rename func(newID string) error, saveTranscript func([]byte) error) *SessionFileStore {
|
||||
sf := &SessionFileStore{sess: sess, log: log, kill: kill, rename: rename, saveTranscript: saveTranscript}
|
||||
notDir := func(string) (Store, error) { return nil, fmt.Errorf("not a directory") }
|
||||
sf := &SessionFileStore{Runnable: sess, sess: sess, log: log, kill: kill, rename: rename, saveTranscript: saveTranscript}
|
||||
notSupported := func(string) error { return fmt.Errorf("not supported") }
|
||||
sf.storeConfig = &storeConfig{
|
||||
StatFn: sf.stat,
|
||||
ListFn: sf.list,
|
||||
GetFn: sf.get,
|
||||
OpenFn: notDir,
|
||||
PutFn: sf.put,
|
||||
OpenFn: sf.open,
|
||||
DeleteFn: notSupported,
|
||||
CreateFn: notSupported,
|
||||
RenameFn: func(string, string) error { return fmt.Errorf("not supported") },
|
||||
|
|
@ -85,7 +83,6 @@ func (s *SessionFileStore) stat(name string) (os.FileInfo, error) {
|
|||
size = int64(len(s.sess.log))
|
||||
s.sess.mu.RUnlock()
|
||||
case "statewait", "usagewait", "ctxszwait", "cwdwait":
|
||||
// Blocking reads; size is unknown until resolved.
|
||||
default:
|
||||
size = int64(len(s.content(name)))
|
||||
}
|
||||
|
|
@ -95,7 +92,38 @@ func (s *SessionFileStore) stat(name string) (os.FileInfo, error) {
|
|||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) get(name string) ([]byte, error) {
|
||||
func (s *SessionFileStore) open(name string) (StoreEntry, error) {
|
||||
for _, f := range SessionFileList {
|
||||
if f.Name == name {
|
||||
return s.entryFor(name, f.Mode), nil
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) entryFor(name string, mode os.FileMode) StoreEntry {
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) {
|
||||
var size int64
|
||||
switch name {
|
||||
case "chat":
|
||||
s.sess.mu.RLock()
|
||||
size = int64(len(s.sess.log))
|
||||
s.sess.mu.RUnlock()
|
||||
case "statewait", "usagewait", "ctxszwait", "cwdwait":
|
||||
// Blocking reads; size is unknown until resolved.
|
||||
default:
|
||||
size = int64(len(s.content(name)))
|
||||
}
|
||||
return &SyntheticFileInfo{Name_: name, Mode_: mode, Size_: size}, nil
|
||||
},
|
||||
ReadFn: func() ([]byte, error) { return s.readFile(name) },
|
||||
WriteFn: func(data []byte) error { return s.writeFile(name, data) },
|
||||
BlockingReadFn: func(ctx context.Context, base string) ([]byte, error) { return s.blockingRead(ctx, name, base) },
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) readFile(name string) ([]byte, error) {
|
||||
switch name {
|
||||
case "chat":
|
||||
s.sess.mu.RLock()
|
||||
|
|
@ -121,7 +149,7 @@ func (s *SessionFileStore) get(name string) ([]byte, error) {
|
|||
}
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) put(name string, data []byte) error {
|
||||
func (s *SessionFileStore) writeFile(name string, data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
|
|
@ -167,6 +195,36 @@ func (s *SessionFileStore) put(name string, data []byte) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) blockingRead(connCtx context.Context, name, base string) ([]byte, error) {
|
||||
ctx, cancel := context.WithCancel(connCtx)
|
||||
defer cancel()
|
||||
context.AfterFunc(s.sess.Ctx, cancel)
|
||||
|
||||
var field string
|
||||
switch name {
|
||||
case "statewait":
|
||||
field = agent.WatchState
|
||||
case "usagewait":
|
||||
field = agent.WatchUsage
|
||||
case "ctxszwait":
|
||||
field = agent.WatchCtxSz
|
||||
case "cwdwait":
|
||||
field = agent.WatchCWD
|
||||
default:
|
||||
return nil, fmt.Errorf("%s: not a wait file", name)
|
||||
}
|
||||
|
||||
if base == "" {
|
||||
base = s.currentWaitValue(name)
|
||||
}
|
||||
|
||||
v, ok := s.sess.Core.WaitChange(ctx, field, base)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return []byte(v + "\n"), nil
|
||||
}
|
||||
|
||||
func FormatParams(p backend.GenerationParams) string {
|
||||
var sb strings.Builder
|
||||
fmt.Fprintf(&sb, "maxTokens=%d\n", p.MaxTokens)
|
||||
|
|
@ -273,8 +331,8 @@ func (s *SessionFileStore) content(name string) string {
|
|||
return ""
|
||||
}
|
||||
|
||||
// CurrentWaitValue returns the current value for the named *wait file.
|
||||
func (s *SessionFileStore) CurrentWaitValue(name string) string {
|
||||
// currentWaitValue returns the current value for the named *wait file.
|
||||
func (s *SessionFileStore) currentWaitValue(name string) string {
|
||||
switch name {
|
||||
case "statewait":
|
||||
return s.sess.Core.State()
|
||||
|
|
@ -288,38 +346,6 @@ func (s *SessionFileStore) CurrentWaitValue(name string) string {
|
|||
return ""
|
||||
}
|
||||
|
||||
// Wait blocks until the named *wait file's underlying value changes from base,
|
||||
// then returns the new value.
|
||||
func (s *SessionFileStore) Wait(connCtx context.Context, name, base string) ([]byte, error) {
|
||||
ctx, cancel := context.WithCancel(connCtx)
|
||||
defer cancel()
|
||||
context.AfterFunc(s.sess.Ctx, cancel)
|
||||
|
||||
var field string
|
||||
switch name {
|
||||
case "statewait":
|
||||
field = agent.WatchState
|
||||
case "usagewait":
|
||||
field = agent.WatchUsage
|
||||
case "ctxszwait":
|
||||
field = agent.WatchCtxSz
|
||||
case "cwdwait":
|
||||
field = agent.WatchCWD
|
||||
default:
|
||||
return nil, fmt.Errorf("%s: not a wait file", name)
|
||||
}
|
||||
|
||||
if base == "" {
|
||||
base = s.CurrentWaitValue(name)
|
||||
}
|
||||
|
||||
v, ok := s.sess.Core.WaitChange(ctx, field, base)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return []byte(v + "\n"), nil
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) makePublish() func(agent.Event) {
|
||||
assistantStarted := false
|
||||
return func(ev agent.Event) {
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
|
@ -27,7 +28,7 @@ type SkillStoreConfig struct {
|
|||
type skillState struct {
|
||||
dirs []string
|
||||
readDir func(string) ([]os.DirEntry, error)
|
||||
open func(string) (*os.File, error)
|
||||
openFile func(string) (*os.File, error)
|
||||
readFile func(string) ([]byte, error)
|
||||
writeFile func(string, []byte, os.FileMode) error
|
||||
mkdirAll func(string, os.FileMode) error
|
||||
|
|
@ -35,11 +36,11 @@ type skillState struct {
|
|||
rename func(string, string) error
|
||||
}
|
||||
|
||||
func NewSkillStore() BlobStore {
|
||||
func NewSkillStore() Store {
|
||||
return NewSkillStoreWith(SkillStoreConfig{})
|
||||
}
|
||||
|
||||
func NewSkillStoreWith(cfg SkillStoreConfig) BlobStore {
|
||||
func NewSkillStoreWith(cfg SkillStoreConfig) Store {
|
||||
if cfg.Dirs == nil {
|
||||
cfg.Dirs = skillDirs()
|
||||
}
|
||||
|
|
@ -68,7 +69,7 @@ func NewSkillStoreWith(cfg SkillStoreConfig) BlobStore {
|
|||
ss := &skillState{
|
||||
dirs: cfg.Dirs,
|
||||
readDir: cfg.ReadDir,
|
||||
open: cfg.Open,
|
||||
openFile: cfg.Open,
|
||||
readFile: cfg.ReadFile,
|
||||
writeFile: cfg.WriteFile,
|
||||
mkdirAll: cfg.MkdirAll,
|
||||
|
|
@ -79,8 +80,7 @@ func NewSkillStoreWith(cfg SkillStoreConfig) BlobStore {
|
|||
return &storeConfig{
|
||||
StatFn: ss.stat,
|
||||
ListFn: ss.list,
|
||||
GetFn: ss.get,
|
||||
PutFn: ss.put,
|
||||
OpenFn: ss.open,
|
||||
DeleteFn: ss.del,
|
||||
CreateFn: ss.create,
|
||||
RenameFn: ss.ren,
|
||||
|
|
@ -107,7 +107,7 @@ func (ss *skillState) listSkills() []skills.Meta {
|
|||
continue
|
||||
}
|
||||
skillDir := filepath.Join(dir, e.Name())
|
||||
f, err := ss.open(filepath.Join(skillDir, "SKILL.md"))
|
||||
f, err := ss.openFile(filepath.Join(skillDir, "SKILL.md"))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
|
@ -152,30 +152,45 @@ func (ss *skillState) list() ([]os.DirEntry, error) {
|
|||
return result, nil
|
||||
}
|
||||
|
||||
func (ss *skillState) get(name string) ([]byte, error) {
|
||||
func (ss *skillState) open(name string) (StoreEntry, error) {
|
||||
notBlocking := func(context.Context, string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("blocking read not supported")
|
||||
}
|
||||
if name == "idx" {
|
||||
return ss.index()
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return &SyntheticFileInfo{Name_: "idx", Mode_: 0444}, nil },
|
||||
ReadFn: func() ([]byte, error) { return ss.index() },
|
||||
WriteFn: func([]byte) error { return fmt.Errorf("idx: read-only") },
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
}
|
||||
skillName := strings.TrimSuffix(name, ".md")
|
||||
return ss.readSkill(skillName)
|
||||
}
|
||||
|
||||
func (ss *skillState) put(name string, data []byte) error {
|
||||
skillName := strings.TrimSuffix(name, ".md")
|
||||
dir := ""
|
||||
for _, m := range ss.listSkills() {
|
||||
if m.Name == skillName {
|
||||
dir = m.Dir
|
||||
break
|
||||
}
|
||||
}
|
||||
if dir == "" {
|
||||
dir = filepath.Join(ss.dirs[0], skillName)
|
||||
}
|
||||
if err := ss.mkdirAll(dir, 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
return ss.writeFile(filepath.Join(dir, "SKILL.md"), data, 0644)
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) {
|
||||
if _, err := ss.readSkill(skillName); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &SyntheticFileInfo{Name_: name, Mode_: 0666}, nil
|
||||
},
|
||||
ReadFn: func() ([]byte, error) { return ss.readSkill(skillName) },
|
||||
WriteFn: func(data []byte) error {
|
||||
dir := ""
|
||||
for _, m := range ss.listSkills() {
|
||||
if m.Name == skillName {
|
||||
dir = m.Dir
|
||||
break
|
||||
}
|
||||
}
|
||||
if dir == "" {
|
||||
dir = filepath.Join(ss.dirs[0], skillName)
|
||||
}
|
||||
if err := ss.mkdirAll(dir, 0755); err != nil {
|
||||
return err
|
||||
}
|
||||
return ss.writeFile(filepath.Join(dir, "SKILL.md"), data, 0644)
|
||||
},
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (ss *skillState) del(name string) error {
|
||||
|
|
|
|||
|
|
@ -3,19 +3,29 @@
|
|||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"time"
|
||||
)
|
||||
|
||||
// StoreEntry is a directory entry within a Store.
|
||||
type StoreEntry = os.DirEntry
|
||||
// StoreEntry is an open handle to a single named blob.
|
||||
type StoreEntry interface {
|
||||
Stat() (os.FileInfo, error)
|
||||
Read() ([]byte, error)
|
||||
Write(data []byte) error
|
||||
BlockingRead(ctx context.Context, base string) ([]byte, error)
|
||||
}
|
||||
|
||||
// ReadableStore is a named collection of byte blobs supporting only read operations.
|
||||
type ReadableStore interface {
|
||||
// Store is a named collection of entries.
|
||||
type Store interface {
|
||||
Stat(name string) (os.FileInfo, error)
|
||||
List() ([]StoreEntry, error)
|
||||
Get(name string) ([]byte, error)
|
||||
List() ([]os.DirEntry, error)
|
||||
Open(name string) (StoreEntry, error)
|
||||
Create(name string) error
|
||||
Delete(name string) error
|
||||
Rename(oldName, newName string) error
|
||||
}
|
||||
|
||||
// Runnable is a running agent that can be observed and controlled.
|
||||
|
|
@ -27,74 +37,76 @@ type Runnable interface {
|
|||
LogInfo() (length int, vers uint32)
|
||||
}
|
||||
|
||||
// WritableStore supports writing blobs by name.
|
||||
type WritableStore interface {
|
||||
Put(name string, data []byte) error
|
||||
// RunnableStore is a Store backed by a running agent.
|
||||
type RunnableStore interface {
|
||||
Store
|
||||
Runnable
|
||||
}
|
||||
|
||||
// ReadWriteStore supports reading and writing but not structural mutations.
|
||||
type ReadWriteStore interface {
|
||||
ReadableStore
|
||||
WritableStore
|
||||
// EntryConfig implements StoreEntry via function pointers.
|
||||
type EntryConfig struct {
|
||||
StatFn func() (os.FileInfo, error)
|
||||
ReadFn func() ([]byte, error)
|
||||
WriteFn func([]byte) error
|
||||
BlockingReadFn func(context.Context, string) ([]byte, error)
|
||||
}
|
||||
|
||||
// BlobStore extends ReadWriteStore with creation, deletion, and renaming.
|
||||
type BlobStore interface {
|
||||
ReadWriteStore
|
||||
Delete(name string) error
|
||||
Create(name string) error
|
||||
Rename(oldName, newName string) error
|
||||
func (e *EntryConfig) Stat() (os.FileInfo, error) { return e.StatFn() }
|
||||
func (e *EntryConfig) Read() ([]byte, error) { return e.ReadFn() }
|
||||
func (e *EntryConfig) Write(data []byte) error { return e.WriteFn(data) }
|
||||
func (e *EntryConfig) BlockingRead(ctx context.Context, base string) ([]byte, error) {
|
||||
return e.BlockingReadFn(ctx, base)
|
||||
}
|
||||
|
||||
// Store is a hierarchical store that supports subdirectories.
|
||||
type Store interface {
|
||||
BlobStore
|
||||
Open(name string) (Store, error)
|
||||
}
|
||||
|
||||
// storeConfig implements the store interfaces via function pointers.
|
||||
// storeConfig implements Store via function pointers.
|
||||
type storeConfig struct {
|
||||
StatFn func(string) (os.FileInfo, error)
|
||||
ListFn func() ([]StoreEntry, error)
|
||||
GetFn func(string) ([]byte, error)
|
||||
OpenFn func(string) (Store, error)
|
||||
PutFn func(string, []byte) error
|
||||
DeleteFn func(string) error
|
||||
ListFn func() ([]os.DirEntry, error)
|
||||
OpenFn func(string) (StoreEntry, error)
|
||||
CreateFn func(string) error
|
||||
DeleteFn func(string) error
|
||||
RenameFn func(string, string) error
|
||||
}
|
||||
|
||||
func (s *storeConfig) Stat(name string) (os.FileInfo, error) { return s.StatFn(name) }
|
||||
func (s *storeConfig) List() ([]StoreEntry, error) { return s.ListFn() }
|
||||
func (s *storeConfig) Get(name string) ([]byte, error) { return s.GetFn(name) }
|
||||
func (s *storeConfig) Open(name string) (Store, error) { return s.OpenFn(name) }
|
||||
func (s *storeConfig) Put(name string, data []byte) error { return s.PutFn(name, data) }
|
||||
func (s *storeConfig) Delete(name string) error { return s.DeleteFn(name) }
|
||||
func (s *storeConfig) List() ([]os.DirEntry, error) { return s.ListFn() }
|
||||
func (s *storeConfig) Open(name string) (StoreEntry, error) { return s.OpenFn(name) }
|
||||
func (s *storeConfig) Create(name string) error { return s.CreateFn(name) }
|
||||
func (s *storeConfig) Delete(name string) error { return s.DeleteFn(name) }
|
||||
func (s *storeConfig) Rename(old, new string) error { return s.RenameFn(old, new) }
|
||||
|
||||
// NewFlatDir returns a BlobStore backed by a directory on the local filesystem.
|
||||
func NewFlatDir(dir string, perm os.FileMode) BlobStore {
|
||||
// NewFlatDir returns a Store backed by a directory on the local filesystem.
|
||||
func NewFlatDir(dir string, perm os.FileMode) Store {
|
||||
join := func(name string) string { return filepath.Join(dir, name) }
|
||||
ensureDir := func() error { return os.MkdirAll(dir, 0755) }
|
||||
notBlocking := func(context.Context, string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("blocking read not supported")
|
||||
}
|
||||
|
||||
return &storeConfig{
|
||||
StatFn: func(name string) (os.FileInfo, error) { return os.Stat(join(name)) },
|
||||
ListFn: func() ([]os.DirEntry, error) { return os.ReadDir(dir) },
|
||||
GetFn: func(name string) ([]byte, error) { return os.ReadFile(join(name)) },
|
||||
PutFn: func(name string, data []byte) error {
|
||||
if err := ensureDir(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(join(name), data, perm)
|
||||
OpenFn: func(name string) (StoreEntry, error) {
|
||||
path := join(name)
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return os.Stat(path) },
|
||||
ReadFn: func() ([]byte, error) { return os.ReadFile(path) },
|
||||
WriteFn: func(data []byte) error {
|
||||
if err := ensureDir(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(path, data, perm)
|
||||
},
|
||||
BlockingReadFn: notBlocking,
|
||||
}, nil
|
||||
},
|
||||
DeleteFn: func(name string) error { return os.Remove(join(name)) },
|
||||
CreateFn: func(name string) error {
|
||||
if err := ensureDir(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.WriteFile(join(name), nil, perm)
|
||||
},
|
||||
DeleteFn: func(name string) error { return os.Remove(join(name)) },
|
||||
RenameFn: func(old, new string) error { return os.Rename(join(old), join(new)) },
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -95,9 +95,35 @@ func newBatchStore(t *testing.T) *store.BatchStore {
|
|||
})
|
||||
}
|
||||
|
||||
// storeRead is a test helper: Open + Read.
|
||||
func storeRead(t *testing.T, s store.Store, name string) []byte {
|
||||
t.Helper()
|
||||
e, err := s.Open(name)
|
||||
if err != nil {
|
||||
t.Fatalf("Open(%q): %v", name, err)
|
||||
}
|
||||
data, err := e.Read()
|
||||
if err != nil {
|
||||
t.Fatalf("Read(%q): %v", name, err)
|
||||
}
|
||||
return data
|
||||
}
|
||||
|
||||
// storeWrite is a test helper: Open + Write.
|
||||
func storeWrite(t *testing.T, s store.Store, name string, data []byte) {
|
||||
t.Helper()
|
||||
e, err := s.Open(name)
|
||||
if err != nil {
|
||||
t.Fatalf("Open(%q): %v", name, err)
|
||||
}
|
||||
if err := e.Write(data); err != nil {
|
||||
t.Fatalf("Write(%q): %v", name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// --- contract checks ---
|
||||
|
||||
func checkReadableContract(t *testing.T, s store.ReadableStore, name string) {
|
||||
func checkReadableContract(t *testing.T, s store.Store, name string) {
|
||||
t.Helper()
|
||||
fi, err := s.Stat(name)
|
||||
if err != nil {
|
||||
|
|
@ -122,42 +148,47 @@ func checkReadableContract(t *testing.T, s store.ReadableStore, name string) {
|
|||
if !found {
|
||||
t.Errorf("List() missing %q", name)
|
||||
}
|
||||
if _, err := s.Get(name); err != nil {
|
||||
t.Fatalf("Get(%q): %v", name, err)
|
||||
e, err := s.Open(name)
|
||||
if err != nil {
|
||||
t.Fatalf("Open(%q): %v", name, err)
|
||||
}
|
||||
if _, err := s.Get("__nonexistent__"); err == nil {
|
||||
t.Error("Get(nonexistent) should error")
|
||||
if _, err := e.Read(); err != nil {
|
||||
t.Fatalf("Read(%q): %v", name, err)
|
||||
}
|
||||
}
|
||||
|
||||
func checkReadWriteContract(t *testing.T, s store.ReadWriteStore, name string) {
|
||||
func checkReadWriteContract(t *testing.T, s store.Store, name string) {
|
||||
t.Helper()
|
||||
want := []byte("hello")
|
||||
if err := s.Put(name, want); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
got, err := s.Get(name)
|
||||
e, err := s.Open(name)
|
||||
if err != nil {
|
||||
t.Fatalf("Get after Put: %v", err)
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
if err := e.Write(want); err != nil {
|
||||
t.Fatalf("Write: %v", err)
|
||||
}
|
||||
got, err := e.Read()
|
||||
if err != nil {
|
||||
t.Fatalf("Read after Write: %v", err)
|
||||
}
|
||||
if string(got) != string(want) {
|
||||
t.Errorf("Get = %q; want %q", got, want)
|
||||
t.Errorf("Read = %q; want %q", got, want)
|
||||
}
|
||||
want2 := []byte("world")
|
||||
if err := s.Put(name, want2); err != nil {
|
||||
t.Fatalf("Put overwrite: %v", err)
|
||||
if err := e.Write(want2); err != nil {
|
||||
t.Fatalf("Write overwrite: %v", err)
|
||||
}
|
||||
got2, err := s.Get(name)
|
||||
got2, err := e.Read()
|
||||
if err != nil {
|
||||
t.Fatalf("Get after overwrite: %v", err)
|
||||
t.Fatalf("Read after overwrite: %v", err)
|
||||
}
|
||||
if string(got2) != string(want2) {
|
||||
t.Errorf("Get after overwrite = %q; want %q", got2, want2)
|
||||
t.Errorf("Read after overwrite = %q; want %q", got2, want2)
|
||||
}
|
||||
checkReadableContract(t, s, name)
|
||||
}
|
||||
|
||||
func checkBlobStoreContract(t *testing.T, s store.BlobStore, name string) {
|
||||
func checkStoreContract(t *testing.T, s store.Store, name string) {
|
||||
t.Helper()
|
||||
if err := s.Create(name); err != nil {
|
||||
t.Fatalf("Create: %v", err)
|
||||
|
|
@ -191,15 +222,10 @@ func checkBlobStoreContract(t *testing.T, s store.BlobStore, name string) {
|
|||
checkReadWriteContract(t, s, name)
|
||||
}
|
||||
|
||||
func checkStoreContract(t *testing.T, s store.Store, name string) {
|
||||
t.Helper()
|
||||
checkBlobStoreContract(t, s, name)
|
||||
}
|
||||
|
||||
// ===== FlatDir =====
|
||||
|
||||
func TestFlatDirContract(t *testing.T) {
|
||||
checkBlobStoreContract(t, store.NewFlatDir(t.TempDir(), 0644), "test-file")
|
||||
checkStoreContract(t, store.NewFlatDir(t.TempDir(), 0644), "test-file")
|
||||
}
|
||||
|
||||
func TestFlatDirCreateMkdirError(t *testing.T) {
|
||||
|
|
@ -211,8 +237,12 @@ func TestFlatDirCreateMkdirError(t *testing.T) {
|
|||
|
||||
func TestFlatDirPutMkdirError(t *testing.T) {
|
||||
fd := store.NewFlatDir("/nonexistent/path", 0644)
|
||||
if err := fd.Put("f", []byte("x")); err == nil {
|
||||
t.Error("Put should fail when dir doesn't exist")
|
||||
e, err := fd.Open("f")
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
if err := e.Write([]byte("x")); err == nil {
|
||||
t.Error("Write should fail when dir doesn't exist")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -250,27 +280,22 @@ func TestSkillStoreContract(t *testing.T) {
|
|||
t.Error("Delete(nonexistent) should error")
|
||||
}
|
||||
|
||||
// Put/Get round-trip (needs valid front matter)
|
||||
// Write/Read round-trip (needs valid front matter)
|
||||
content := []byte("---\ndescription: rw test\n---\nbody\n")
|
||||
if err := s.Put("rw-skill.md", content); err != nil {
|
||||
t.Fatalf("Put: %v", err)
|
||||
}
|
||||
got, err := s.Get("rw-skill.md")
|
||||
if err != nil {
|
||||
t.Fatalf("Get after Put: %v", err)
|
||||
if err := s.Create("rw-skill.md"); err != nil {
|
||||
t.Fatalf("Create: %v", err)
|
||||
}
|
||||
storeWrite(t, s, "rw-skill.md", content)
|
||||
got := storeRead(t, s, "rw-skill.md")
|
||||
if string(got) != string(content) {
|
||||
t.Errorf("Get = %q; want %q", got, content)
|
||||
t.Errorf("Read = %q; want %q", got, content)
|
||||
}
|
||||
|
||||
// Readable
|
||||
checkReadableContract(t, s, "test-skill.md")
|
||||
|
||||
// idx
|
||||
idx, err := s.Get("idx")
|
||||
if err != nil {
|
||||
t.Fatalf("Get(idx): %v", err)
|
||||
}
|
||||
idx := storeRead(t, s, "idx")
|
||||
if len(idx) == 0 {
|
||||
t.Error("idx should be non-empty with seeded skills")
|
||||
}
|
||||
|
|
@ -347,10 +372,7 @@ func TestSessionStoreGetIdx(t *testing.T) {
|
|||
defer sess.Cancel()
|
||||
s.AddSession(sess)
|
||||
|
||||
data, err := s.Get("idx")
|
||||
if err != nil {
|
||||
t.Fatalf("Get(idx): %v", err)
|
||||
}
|
||||
data := storeRead(t, s, "idx")
|
||||
if !strings.Contains(string(data), "abc") {
|
||||
t.Errorf("idx = %q; want to contain abc", data)
|
||||
}
|
||||
|
|
@ -358,12 +380,9 @@ func TestSessionStoreGetIdx(t *testing.T) {
|
|||
|
||||
func TestSessionStoreGetScript(t *testing.T) {
|
||||
s := newSessionStore(t)
|
||||
data, err := s.Get("ls")
|
||||
if err != nil {
|
||||
t.Fatalf("Get(ls): %v", err)
|
||||
}
|
||||
data := storeRead(t, s, "ls")
|
||||
if string(data) != "#!/bin/sh\n" {
|
||||
t.Errorf("Get(ls) = %q", data)
|
||||
t.Errorf("Read(ls) = %q", data)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -400,10 +419,14 @@ func TestSessionStoreListIncludesSessions(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestSessionStorePutNotWritable(t *testing.T) {
|
||||
func TestSessionStoreWriteNotWritable(t *testing.T) {
|
||||
s := newSessionStore(t)
|
||||
if err := s.Put("idx", nil); err == nil {
|
||||
t.Error("Put(idx) should error")
|
||||
e, err := s.Open("idx")
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
if err := e.Write(nil); err == nil {
|
||||
t.Error("Write(idx) should error")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -558,12 +581,9 @@ func TestSessionFileStoreGetChat(t *testing.T) {
|
|||
sf := store.NewSessionFileStore(sess, sink.NewLogger("test"),
|
||||
func() {}, func(string) error { return nil }, func([]byte) error { return nil })
|
||||
|
||||
data, err := sf.Get("chat")
|
||||
if err != nil {
|
||||
t.Fatalf("Get(chat): %v", err)
|
||||
}
|
||||
data := storeRead(t, sf, "chat")
|
||||
if string(data) != "hello" {
|
||||
t.Errorf("Get(chat) = %q; want hello", data)
|
||||
t.Errorf("Read(chat) = %q; want hello", data)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -575,7 +595,7 @@ func TestSessionFileStoreGetContent(t *testing.T) {
|
|||
func() {}, func(string) error { return nil }, func([]byte) error { return nil })
|
||||
|
||||
for _, name := range []string{"backend", "agent", "model", "state", "cwd", "offset", "params"} {
|
||||
if _, err := sf.Get(name); err != nil {
|
||||
if _, err := sf.Open(name); err != nil {
|
||||
t.Errorf("Get(%q): %v", name, err)
|
||||
}
|
||||
}
|
||||
|
|
@ -588,9 +608,7 @@ func TestSessionFileStorePutCwd(t *testing.T) {
|
|||
sf := store.NewSessionFileStore(sess, sink.NewLogger("test"),
|
||||
func() {}, func(string) error { return nil }, func([]byte) error { return nil })
|
||||
|
||||
if err := sf.Put("cwd", []byte("/new/path")); err != nil {
|
||||
t.Fatalf("Put(cwd): %v", err)
|
||||
}
|
||||
storeWrite(t, sf, "cwd", []byte("/new/path"))
|
||||
core := sess.Core.(*stubCore)
|
||||
if core.cwd != "/new/path" {
|
||||
t.Errorf("cwd = %q; want /new/path", core.cwd)
|
||||
|
|
@ -605,23 +623,12 @@ func TestSessionFileStorePutEmpty(t *testing.T) {
|
|||
func() {}, func(string) error { return nil }, func([]byte) error { return nil })
|
||||
|
||||
// Empty write is a no-op
|
||||
if err := sf.Put("cwd", []byte("")); err != nil {
|
||||
t.Fatalf("Put(cwd, empty): %v", err)
|
||||
e, err := sf.Open("cwd")
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionFileStoreCurrentWaitValue(t *testing.T) {
|
||||
sess := testSession("s1")
|
||||
defer sess.Cancel()
|
||||
sink := testSink()
|
||||
sf := store.NewSessionFileStore(sess, sink.NewLogger("test"),
|
||||
func() {}, func(string) error { return nil }, func([]byte) error { return nil })
|
||||
|
||||
if v := sf.CurrentWaitValue("statewait"); v != "idle" {
|
||||
t.Errorf("CurrentWaitValue(statewait) = %q; want idle", v)
|
||||
}
|
||||
if v := sf.CurrentWaitValue("cwdwait"); v != "/tmp" {
|
||||
t.Errorf("CurrentWaitValue(cwdwait) = %q; want /tmp", v)
|
||||
if err := e.Write([]byte("")); err != nil {
|
||||
t.Fatalf("Write(cwd, empty): %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -709,19 +716,20 @@ func TestBatchStoreGetIdx(t *testing.T) {
|
|||
s := newBatchStore(t)
|
||||
s.AddJob("j1", "done", "result", "spec")
|
||||
|
||||
data, err := s.Get("idx")
|
||||
if err != nil {
|
||||
t.Fatalf("Get(idx): %v", err)
|
||||
}
|
||||
data := storeRead(t, s, "idx")
|
||||
if !strings.Contains(string(data), "j1") {
|
||||
t.Errorf("idx = %q; want to contain j1", data)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchStorePutNotWritable(t *testing.T) {
|
||||
func TestBatchStoreWriteNotWritable(t *testing.T) {
|
||||
s := newBatchStore(t)
|
||||
if err := s.Put("idx", nil); err == nil {
|
||||
t.Error("Put(idx) should error")
|
||||
e, err := s.Open("idx")
|
||||
if err != nil {
|
||||
t.Fatalf("Open: %v", err)
|
||||
}
|
||||
if err := e.Write(nil); err == nil {
|
||||
t.Error("Write(idx) should error")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -779,7 +787,7 @@ func TestBatchStoreShutdown(t *testing.T) {
|
|||
func TestBatchJobStoreReadableContract(t *testing.T) {
|
||||
s := newBatchStore(t)
|
||||
s.AddJob("j1", "done", "the result", "the spec")
|
||||
js, err := s.Open("j1")
|
||||
js, err := s.OpenStore("j1")
|
||||
if err != nil {
|
||||
t.Fatal("JobStore(j1) not found")
|
||||
}
|
||||
|
|
@ -789,7 +797,7 @@ func TestBatchJobStoreReadableContract(t *testing.T) {
|
|||
func TestBatchJobStoreContent(t *testing.T) {
|
||||
s := newBatchStore(t)
|
||||
s.AddJob("j1", "done", "the result", "the spec")
|
||||
js, _ := s.Open("j1")
|
||||
js, _ := s.OpenStore("j1")
|
||||
|
||||
for _, tc := range []struct {
|
||||
name, want string
|
||||
|
|
@ -798,13 +806,9 @@ func TestBatchJobStoreContent(t *testing.T) {
|
|||
{"state", "done\n"},
|
||||
{"result", "the result"},
|
||||
} {
|
||||
data, err := js.Get(tc.name)
|
||||
if err != nil {
|
||||
t.Errorf("Get(%q): %v", tc.name, err)
|
||||
continue
|
||||
}
|
||||
data := storeRead(t, js, tc.name)
|
||||
if string(data) != tc.want {
|
||||
t.Errorf("Get(%q) = %q; want %q", tc.name, data, tc.want)
|
||||
t.Errorf("Read(%q) = %q; want %q", tc.name, data, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -812,29 +816,37 @@ func TestBatchJobStoreContent(t *testing.T) {
|
|||
func TestBatchJobStoreWait(t *testing.T) {
|
||||
s := newBatchStore(t)
|
||||
s.AddJob("j1", "done", "", "")
|
||||
js, _ := s.Open("j1")
|
||||
js, _ := s.OpenStore("j1")
|
||||
|
||||
// Job is already done, so Wait returns immediately
|
||||
data, err := js.(*store.BatchJobStore).Wait(context.Background(), "statewait", "")
|
||||
// Job is already done, so BlockingRead returns immediately
|
||||
e, err := js.Open("statewait")
|
||||
if err != nil {
|
||||
t.Fatalf("Wait: %v", err)
|
||||
t.Fatalf("Open(statewait): %v", err)
|
||||
}
|
||||
data, err := e.BlockingRead(context.Background(), "")
|
||||
if err != nil {
|
||||
t.Fatalf("BlockingRead: %v", err)
|
||||
}
|
||||
if !strings.Contains(string(data), "done") {
|
||||
t.Errorf("Wait = %q; want to contain done", data)
|
||||
t.Errorf("BlockingRead = %q; want to contain done", data)
|
||||
}
|
||||
|
||||
// Non-wait file
|
||||
if _, err := js.(*store.BatchJobStore).Wait(context.Background(), "spec", ""); err == nil {
|
||||
t.Error("Wait(spec) should error")
|
||||
e2, err := js.Open("spec")
|
||||
if err != nil {
|
||||
t.Fatalf("Open(spec): %v", err)
|
||||
}
|
||||
if _, err := e2.BlockingRead(context.Background(), ""); err == nil {
|
||||
t.Error("BlockingRead(spec) should error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchJobStoreLogInfo(t *testing.T) {
|
||||
s := newBatchStore(t)
|
||||
s.AddJob("j1", "done", "", "")
|
||||
js, _ := s.Open("j1")
|
||||
js, _ := s.OpenStore("j1")
|
||||
|
||||
l, v := js.(*store.BatchJobStore).LogInfo()
|
||||
l, v := js.LogInfo()
|
||||
if l != 0 || v != 0 {
|
||||
t.Errorf("LogInfo = (%d, %d); want (0, 0)", l, v)
|
||||
}
|
||||
|
|
@ -842,7 +854,7 @@ func TestBatchJobStoreLogInfo(t *testing.T) {
|
|||
|
||||
func TestBatchJobStoreNotFound(t *testing.T) {
|
||||
s := newBatchStore(t)
|
||||
if _, err := s.Open("nope"); err == nil {
|
||||
if _, err := s.OpenStore("nope"); err == nil {
|
||||
t.Error("JobStore(nonexistent) should be false")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Reference in New Issue