From 130bf607083d26daac6100fb92b30253931d2d82 Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Tue, 21 Apr 2026 20:07:27 +0200 Subject: [PATCH] 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 --- pkg/store/batch.go | 116 +++++++++++---------- pkg/store/session.go | 49 +++++---- pkg/store/sessionfile.go | 110 ++++++++++++-------- pkg/store/skill.go | 71 +++++++------ pkg/store/store.go | 102 ++++++++++--------- pkg/store/store_test.go | 212 +++++++++++++++++++++------------------ 6 files changed, 376 insertions(+), 284 deletions(-) diff --git a/pkg/store/batch.go b/pkg/store/batch.go index 2b1ac4d..7353f63 100644 --- a/pkg/store/batch.go +++ b/pkg/store/batch.go @@ -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. diff --git a/pkg/store/session.go b/pkg/store/session.go index 94f2b0a..d33b828 100644 --- a/pkg/store/session.go +++ b/pkg/store/session.go @@ -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) diff --git a/pkg/store/sessionfile.go b/pkg/store/sessionfile.go index 9a2e67d..e1562b0 100644 --- a/pkg/store/sessionfile.go +++ b/pkg/store/sessionfile.go @@ -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) { diff --git a/pkg/store/skill.go b/pkg/store/skill.go index 6862330..991943c 100644 --- a/pkg/store/skill.go +++ b/pkg/store/skill.go @@ -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 { diff --git a/pkg/store/store.go b/pkg/store/store.go index 5fc3c5f..e2e0e91 100644 --- a/pkg/store/store.go +++ b/pkg/store/store.go @@ -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)) }, } } diff --git a/pkg/store/store_test.go b/pkg/store/store_test.go index 22dc1a0..fdbeb8d 100644 --- a/pkg/store/store_test.go +++ b/pkg/store/store_test.go @@ -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") } }