store: unify SessionFileStore and BatchJobStore via RunFileStore
Introduce FileSpec and RunFileStore as a shared, table-driven implementation of RunnableStore. Both SessionFileStore and BatchJobStore are now type aliases for RunFileStore, constructed from a list of FileSpecs that define per-file read/write/wait behavior via function pointers. This eliminates ~94 lines of duplicated store plumbing (stat/list/ open/entryFor methods) while preserving the existing filesystem layout and all public API signatures. No behavioral changes. Tests pass unchanged.
This commit is contained in:
Levi Neely
parent
81fedb862c
commit
8f856ea186
|
|
@ -469,40 +469,8 @@ func ParseBatchSpec(input string) (batchSpec, error) {
|
|||
|
||||
// --- BatchJobStore ---
|
||||
|
||||
var batchJobFiles = []struct {
|
||||
name string
|
||||
mode os.FileMode
|
||||
}{
|
||||
{"spec", 0444},
|
||||
{"state", 0444},
|
||||
{"statewait", 0444},
|
||||
{"result", 0444},
|
||||
{"usage", 0444},
|
||||
{"ctxsz", 0444},
|
||||
{"log", 0444},
|
||||
}
|
||||
|
||||
// BatchJobStore provides Stat/List/Get for the files within a single batch job
|
||||
// directory (/b/{id}/*). The file set is fixed.
|
||||
type BatchJobStore struct {
|
||||
*storeConfig
|
||||
Runnable
|
||||
job *batchJob
|
||||
}
|
||||
|
||||
func NewBatchJobStore(job *batchJob) *BatchJobStore {
|
||||
js := &BatchJobStore{Runnable: job, job: job}
|
||||
notSupported := func(string) error { return fmt.Errorf("not supported") }
|
||||
js.storeConfig = &storeConfig{
|
||||
StatFn: js.stat,
|
||||
ListFn: js.list,
|
||||
OpenFn: js.open,
|
||||
DeleteFn: notSupported,
|
||||
CreateFn: notSupported,
|
||||
RenameFn: func(string, string) error { return fmt.Errorf("not supported") },
|
||||
}
|
||||
return js
|
||||
}
|
||||
// BatchJobStore is a RunFileStore for a batch job directory.
|
||||
type BatchJobStore = RunFileStore
|
||||
|
||||
// OpenStore returns a RunnableStore for the given batch job ID.
|
||||
func (s *BatchStore) OpenStore(id string) (RunnableStore, error) {
|
||||
|
|
@ -510,100 +478,66 @@ func (s *BatchStore) OpenStore(id string) (RunnableStore, error) {
|
|||
if job == nil {
|
||||
return nil, fmt.Errorf("batch job not found: %s", id)
|
||||
}
|
||||
return NewBatchJobStore(job), nil
|
||||
return newBatchJobFileStore(job), nil
|
||||
}
|
||||
|
||||
func (js *BatchJobStore) list() ([]os.DirEntry, error) {
|
||||
entries := make([]os.DirEntry, len(batchJobFiles))
|
||||
for i, f := range batchJobFiles {
|
||||
entries[i] = FileEntry(f.name, f.mode)
|
||||
}
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
func (js *BatchJobStore) stat(name string) (os.FileInfo, error) {
|
||||
for _, f := range batchJobFiles {
|
||||
if f.name == name {
|
||||
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_: f.mode, Size_: size}, nil
|
||||
func newBatchJobFileStore(job *batchJob) *RunFileStore {
|
||||
content := func(name string) string {
|
||||
job.mu.RLock()
|
||||
defer job.mu.RUnlock()
|
||||
switch name {
|
||||
case "spec":
|
||||
return job.spec
|
||||
case "state":
|
||||
return job.state + "\n"
|
||||
case "result":
|
||||
return job.result
|
||||
case "usage":
|
||||
return job.usage + "\n"
|
||||
case "ctxsz":
|
||||
return job.ctxsz + "\n"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
readContent := func(name string) func() ([]byte, error) {
|
||||
return func() ([]byte, error) { return []byte(content(name)), nil }
|
||||
}
|
||||
logRead := func() ([]byte, error) {
|
||||
job.mu.RLock()
|
||||
data := make([]byte, len(job.log))
|
||||
copy(data, job.log)
|
||||
job.mu.RUnlock()
|
||||
return data, nil
|
||||
}
|
||||
logSize := func() int64 {
|
||||
job.mu.RLock()
|
||||
n := len(job.log)
|
||||
job.mu.RUnlock()
|
||||
return int64(n)
|
||||
}
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
func (js *BatchJobStore) open(name string) (StoreEntry, error) {
|
||||
for _, f := range batchJobFiles {
|
||||
if f.name == name {
|
||||
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
|
||||
specs := []FileSpec{
|
||||
{Name: "spec", Mode: 0444, Read: readContent("spec")},
|
||||
{Name: "state", Mode: 0444, Read: readContent("state")},
|
||||
{Name: "statewait", Mode: 0444, Read: readContent("state"),
|
||||
Wait: func(ctx context.Context, base string) ([]byte, error) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, nil
|
||||
case <-job.done:
|
||||
}
|
||||
job.mu.RLock()
|
||||
state := job.state
|
||||
job.mu.RUnlock()
|
||||
return []byte(state + "\n"), nil
|
||||
},
|
||||
},
|
||||
{Name: "result", Mode: 0444, Read: readContent("result")},
|
||||
{Name: "usage", Mode: 0444, Read: readContent("usage")},
|
||||
{Name: "ctxsz", Mode: 0444, Read: readContent("ctxsz")},
|
||||
{Name: "log", Mode: 0444, Read: logRead, Size: logSize},
|
||||
}
|
||||
}
|
||||
|
||||
func (js *BatchJobStore) content(name string) string {
|
||||
js.job.mu.RLock()
|
||||
defer js.job.mu.RUnlock()
|
||||
switch name {
|
||||
case "spec":
|
||||
return js.job.spec
|
||||
case "state":
|
||||
return js.job.state + "\n"
|
||||
case "result":
|
||||
return js.job.result
|
||||
case "usage":
|
||||
return js.job.usage + "\n"
|
||||
case "ctxsz":
|
||||
return js.job.ctxsz + "\n"
|
||||
}
|
||||
return ""
|
||||
return NewRunFileStore(job, specs)
|
||||
}
|
||||
|
||||
// LoadAgentConfig resolves and loads the config for a named agent.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,102 @@
|
|||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
)
|
||||
|
||||
// FileSpec describes a single synthetic file exposed by a RunFileStore.
|
||||
type FileSpec struct {
|
||||
Name string
|
||||
Mode os.FileMode
|
||||
Read func() ([]byte, error)
|
||||
Write func([]byte) error // nil = read-only
|
||||
Wait func(ctx context.Context, base string) ([]byte, error) // nil = not waitable
|
||||
Size func() int64 // optional; if nil, len(Read())
|
||||
}
|
||||
|
||||
// RunFileStore implements RunnableStore for any Runnable using a table of FileSpecs.
|
||||
type RunFileStore struct {
|
||||
*storeConfig
|
||||
Runnable
|
||||
specs []FileSpec
|
||||
index map[string]int // name -> index into specs
|
||||
}
|
||||
|
||||
// NewRunFileStore creates a RunnableStore backed by the given Runnable and file table.
|
||||
func NewRunFileStore(r Runnable, specs []FileSpec) *RunFileStore {
|
||||
rs := &RunFileStore{
|
||||
Runnable: r,
|
||||
specs: specs,
|
||||
index: make(map[string]int, len(specs)),
|
||||
}
|
||||
for i, s := range specs {
|
||||
rs.index[s.Name] = i
|
||||
}
|
||||
notSupported := func(string) error { return fmt.Errorf("not supported") }
|
||||
rs.storeConfig = &storeConfig{
|
||||
StatFn: rs.stat,
|
||||
ListFn: rs.list,
|
||||
OpenFn: rs.open,
|
||||
DeleteFn: notSupported,
|
||||
CreateFn: notSupported,
|
||||
RenameFn: func(string, string) error { return fmt.Errorf("not supported") },
|
||||
}
|
||||
return rs
|
||||
}
|
||||
|
||||
func (rs *RunFileStore) lookup(name string) (*FileSpec, bool) {
|
||||
i, ok := rs.index[name]
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
return &rs.specs[i], true
|
||||
}
|
||||
|
||||
func (rs *RunFileStore) stat(name string) (os.FileInfo, error) {
|
||||
spec, ok := rs.lookup(name)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
var size int64
|
||||
if spec.Size != nil {
|
||||
size = spec.Size()
|
||||
} else if spec.Wait == nil {
|
||||
if data, err := spec.Read(); err == nil {
|
||||
size = int64(len(data))
|
||||
}
|
||||
}
|
||||
return &SyntheticFileInfo{Name_: name, Mode_: spec.Mode, Size_: size}, nil
|
||||
}
|
||||
|
||||
func (rs *RunFileStore) list() ([]os.DirEntry, error) {
|
||||
entries := make([]os.DirEntry, len(rs.specs))
|
||||
for i, s := range rs.specs {
|
||||
entries[i] = FileEntry(s.Name, s.Mode)
|
||||
}
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
func (rs *RunFileStore) open(name string) (StoreEntry, error) {
|
||||
spec, ok := rs.lookup(name)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
writeFn := func(data []byte) error { return fmt.Errorf("%s: read-only", name) }
|
||||
if spec.Write != nil {
|
||||
writeFn = spec.Write
|
||||
}
|
||||
waitFn := func(context.Context, string) ([]byte, error) {
|
||||
return nil, fmt.Errorf("%s: not a wait file", name)
|
||||
}
|
||||
if spec.Wait != nil {
|
||||
waitFn = spec.Wait
|
||||
}
|
||||
return &EntryConfig{
|
||||
StatFn: func() (os.FileInfo, error) { return rs.stat(name) },
|
||||
ReadFn: spec.Read,
|
||||
WriteFn: writeFn,
|
||||
BlockingReadFn: waitFn,
|
||||
}, nil
|
||||
}
|
||||
|
|
@ -40,11 +40,20 @@ var SessionFileList = []struct {
|
|||
{"tail", 0555},
|
||||
}
|
||||
|
||||
// SessionFileStore implements Store for the files within a single
|
||||
// session directory. The file set is fixed.
|
||||
type SessionFileStore struct {
|
||||
*storeConfig
|
||||
Runnable
|
||||
// SessionFileStore is a RunFileStore for a session directory.
|
||||
type SessionFileStore = RunFileStore
|
||||
|
||||
func NewSessionFileStore(sess *Session, log *olog.Logger, kill func(), rename func(newID string) error, saveTranscript func([]byte) error) *SessionFileStore {
|
||||
h := &sessionHelper{sess: sess, log: log, kill: kill, rename: rename, saveTranscript: saveTranscript}
|
||||
specs := make([]FileSpec, len(SessionFileList))
|
||||
for i, f := range SessionFileList {
|
||||
specs[i] = h.fileSpec(f.Name, f.Mode)
|
||||
}
|
||||
return NewRunFileStore(sess, specs)
|
||||
}
|
||||
|
||||
// sessionHelper holds the dependencies needed to build session FileSpecs.
|
||||
type sessionHelper struct {
|
||||
sess *Session
|
||||
log *olog.Logger
|
||||
kill func()
|
||||
|
|
@ -52,182 +61,253 @@ type SessionFileStore struct {
|
|||
saveTranscript func([]byte) error
|
||||
}
|
||||
|
||||
func NewSessionFileStore(sess *Session, log *olog.Logger, kill func(), rename func(newID string) error, saveTranscript func([]byte) error) *SessionFileStore {
|
||||
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,
|
||||
OpenFn: sf.open,
|
||||
DeleteFn: notSupported,
|
||||
CreateFn: notSupported,
|
||||
RenameFn: func(string, string) error { return fmt.Errorf("not supported") },
|
||||
}
|
||||
return sf
|
||||
}
|
||||
func (h *sessionHelper) fileSpec(name string, mode os.FileMode) FileSpec {
|
||||
fs := FileSpec{Name: name, Mode: mode}
|
||||
|
||||
func (s *SessionFileStore) list() ([]os.DirEntry, error) {
|
||||
entries := make([]os.DirEntry, len(SessionFileList))
|
||||
for i, f := range SessionFileList {
|
||||
entries[i] = FileEntry(f.Name, f.Mode)
|
||||
}
|
||||
return entries, nil
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) stat(name string) (os.FileInfo, error) {
|
||||
for _, f := range SessionFileList {
|
||||
if f.Name == name {
|
||||
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":
|
||||
default:
|
||||
size = int64(len(s.content(name)))
|
||||
}
|
||||
return &SyntheticFileInfo{Name_: name, Mode_: f.Mode, Size_: size}, nil
|
||||
}
|
||||
}
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
}
|
||||
|
||||
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) {
|
||||
// Read
|
||||
switch name {
|
||||
case "chat":
|
||||
s.sess.mu.RLock()
|
||||
data := make([]byte, len(s.sess.log))
|
||||
copy(data, s.sess.log)
|
||||
s.sess.mu.RUnlock()
|
||||
return data, nil
|
||||
case "offset":
|
||||
return []byte(s.content("offset")), nil
|
||||
fs.Read = func() ([]byte, error) {
|
||||
h.sess.mu.RLock()
|
||||
data := make([]byte, len(h.sess.log))
|
||||
copy(data, h.sess.log)
|
||||
h.sess.mu.RUnlock()
|
||||
return data, nil
|
||||
}
|
||||
fs.Size = func() int64 {
|
||||
h.sess.mu.RLock()
|
||||
n := len(h.sess.log)
|
||||
h.sess.mu.RUnlock()
|
||||
return int64(n)
|
||||
}
|
||||
case "fifo.out":
|
||||
item, ok := s.sess.Core.PopQueue()
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return []byte(item), nil
|
||||
default:
|
||||
for _, f := range SessionFileList {
|
||||
if f.Name == name {
|
||||
return []byte(s.content(name)), nil
|
||||
fs.Read = func() ([]byte, error) {
|
||||
item, ok := h.sess.Core.PopQueue()
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return []byte(item), nil
|
||||
}
|
||||
return nil, fmt.Errorf("%s: not found", name)
|
||||
default:
|
||||
fs.Read = func() ([]byte, error) { return []byte(h.content(name)), nil }
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) writeFile(name string, data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
// Write
|
||||
switch name {
|
||||
case "chat":
|
||||
return s.saveTranscript([]byte(input))
|
||||
fs.Write = func(data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
return h.saveTranscript([]byte(input))
|
||||
}
|
||||
case "prompt":
|
||||
pub := s.makePublish()
|
||||
s.sess.Core.Submit(s.sess.Ctx, input, pub)
|
||||
s.sess.EnsureTrailingNewline()
|
||||
fs.Write = func(data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
pub := h.makePublish()
|
||||
h.sess.Core.Submit(h.sess.Ctx, input, pub)
|
||||
h.sess.EnsureTrailingNewline()
|
||||
return nil
|
||||
}
|
||||
case "fifo.in":
|
||||
s.sess.Core.Queue(input)
|
||||
fs.Write = func(data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
h.sess.Core.Queue(input)
|
||||
return nil
|
||||
}
|
||||
case "ctl":
|
||||
return s.handleCtl(input)
|
||||
case "backend":
|
||||
if s.sess.Core.IsRunning() {
|
||||
return fmt.Errorf("cannot switch backend while agent is running")
|
||||
fs.Write = func(data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
return h.handleCtl(input)
|
||||
}
|
||||
s.sess.Core.Submit(s.sess.Ctx, "/backend "+input, s.makePublish())
|
||||
case "agent":
|
||||
if s.sess.Core.IsRunning() {
|
||||
return fmt.Errorf("cannot switch agent while agent is running")
|
||||
case "backend", "model", "agent":
|
||||
fs.Write = func(data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
if h.sess.Core.IsRunning() {
|
||||
return fmt.Errorf("cannot switch %s while agent is running", name)
|
||||
}
|
||||
h.sess.Core.Submit(h.sess.Ctx, "/"+name+" "+input, h.makePublish())
|
||||
return nil
|
||||
}
|
||||
s.sess.Core.Submit(s.sess.Ctx, "/agent "+input, s.makePublish())
|
||||
case "model":
|
||||
if s.sess.Core.IsRunning() {
|
||||
return fmt.Errorf("cannot switch model while agent is running")
|
||||
}
|
||||
s.sess.Core.Submit(s.sess.Ctx, "/model "+input, s.makePublish())
|
||||
case "cwd":
|
||||
if err := s.sess.Core.SetCWD(input); err != nil {
|
||||
return err
|
||||
fs.Write = func(data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
return h.sess.Core.SetCWD(input)
|
||||
}
|
||||
case "params":
|
||||
if s.sess.Core.IsRunning() {
|
||||
return fmt.Errorf("cannot change params while agent is running")
|
||||
fs.Write = func(data []byte) error {
|
||||
input := strings.TrimSpace(string(data))
|
||||
if input == "" {
|
||||
return nil
|
||||
}
|
||||
if h.sess.Core.IsRunning() {
|
||||
return fmt.Errorf("cannot change params while agent is running")
|
||||
}
|
||||
params, err := ParseParams(input, h.sess.Core.GenerationParams())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return h.sess.Core.SetGenerationParams(params)
|
||||
}
|
||||
params, err := ParseParams(input, s.sess.Core.GenerationParams())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Wait
|
||||
switch name {
|
||||
case "statewait", "usagewait", "ctxszwait", "cwdwait":
|
||||
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
|
||||
}
|
||||
return s.sess.Core.SetGenerationParams(params)
|
||||
fs.Wait = func(connCtx context.Context, base string) ([]byte, error) {
|
||||
ctx, cancel := context.WithCancel(connCtx)
|
||||
defer cancel()
|
||||
context.AfterFunc(h.sess.Ctx, cancel)
|
||||
if base == "" {
|
||||
base = h.currentWaitValue(name)
|
||||
}
|
||||
v, ok := h.sess.Core.WaitChange(ctx, field, base)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
return []byte(v + "\n"), nil
|
||||
}
|
||||
}
|
||||
|
||||
return fs
|
||||
}
|
||||
|
||||
func (h *sessionHelper) content(name string) string {
|
||||
h.sess.mu.RLock()
|
||||
defer h.sess.mu.RUnlock()
|
||||
switch name {
|
||||
case "backend":
|
||||
return h.sess.Core.BackendName() + "\n"
|
||||
case "agent":
|
||||
return h.sess.Core.AgentName() + "\n"
|
||||
case "model":
|
||||
return h.sess.Core.ModelName() + "\n"
|
||||
case "state":
|
||||
return h.sess.Core.State() + "\n"
|
||||
case "cwd":
|
||||
return h.sess.Core.CWD() + "\n"
|
||||
case "usage":
|
||||
return h.sess.Core.Usage() + "\n"
|
||||
case "ctxsz":
|
||||
return h.sess.Core.CtxSz() + "\n"
|
||||
case "models":
|
||||
return h.sess.Core.ListModels() + "\n"
|
||||
case "systemprompt":
|
||||
return h.sess.Core.SystemPrompt()
|
||||
case "offset":
|
||||
return fmt.Sprintf("%d\n", h.sess.ChatOffset)
|
||||
case "params":
|
||||
return FormatParams(h.sess.Core.GenerationParams())
|
||||
case "tail":
|
||||
return "#!/bin/sh\nexec tail -f \"$(dirname \"$0\")/chat\"\n"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (h *sessionHelper) currentWaitValue(name string) string {
|
||||
switch name {
|
||||
case "statewait":
|
||||
return h.sess.Core.State()
|
||||
case "usagewait":
|
||||
return h.sess.Core.Usage()
|
||||
case "ctxszwait":
|
||||
return h.sess.Core.CtxSz()
|
||||
case "cwdwait":
|
||||
return h.sess.Core.CWD()
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (h *sessionHelper) makePublish() func(agent.Event) {
|
||||
assistantStarted := false
|
||||
return func(ev agent.Event) {
|
||||
if ev.Role == "user" {
|
||||
if assistantStarted {
|
||||
h.sess.AppendLog([]byte("\n"))
|
||||
assistantStarted = false
|
||||
}
|
||||
h.sess.mu.Lock()
|
||||
h.sess.ChatOffset = len(h.sess.log)
|
||||
h.sess.mu.Unlock()
|
||||
} else {
|
||||
switch ev.Role {
|
||||
case "call", "tool":
|
||||
if assistantStarted {
|
||||
h.sess.AppendLog([]byte("\n"))
|
||||
assistantStarted = false
|
||||
}
|
||||
case "assistant":
|
||||
if !assistantStarted {
|
||||
h.sess.AppendLog([]byte("assistant: "))
|
||||
assistantStarted = true
|
||||
}
|
||||
}
|
||||
}
|
||||
h.sess.AppendLog(FormatEvent(ev))
|
||||
}
|
||||
}
|
||||
|
||||
func (h *sessionHelper) handleCtl(input string) error {
|
||||
cmd := strings.Fields(input)
|
||||
if len(cmd) == 0 {
|
||||
return fmt.Errorf("empty ctl command")
|
||||
}
|
||||
switch cmd[0] {
|
||||
case "stop":
|
||||
h.sess.Core.Interrupt(agent.ErrInterrupted)
|
||||
case "kill":
|
||||
h.kill()
|
||||
case "rn":
|
||||
if name := strings.TrimSpace(input[3:]); name != "" {
|
||||
if err := h.rename(name); err != nil {
|
||||
h.log.Error("rename: %v", err)
|
||||
}
|
||||
}
|
||||
case "save":
|
||||
h.sess.mu.RLock()
|
||||
data := make([]byte, len(h.sess.log))
|
||||
copy(data, h.sess.log)
|
||||
h.sess.mu.RUnlock()
|
||||
return h.saveTranscript(data)
|
||||
case "compact", "clear", "backend", "model", "models",
|
||||
"agents", "agent", "sessions", "cwd", "skills",
|
||||
"tools", "context", "usage", "history",
|
||||
"irw", "help":
|
||||
h.sess.Core.Submit(h.sess.Ctx, "/"+input, h.makePublish())
|
||||
default:
|
||||
return fmt.Errorf("unknown ctl command: %s", cmd[0])
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
// FormatParams formats generation parameters as key=value lines.
|
||||
func FormatParams(p backend.GenerationParams) string {
|
||||
var sb strings.Builder
|
||||
fmt.Fprintf(&sb, "maxTokens=%d\n", p.MaxTokens)
|
||||
|
|
@ -249,6 +329,7 @@ func FormatParams(p backend.GenerationParams) string {
|
|||
return sb.String()
|
||||
}
|
||||
|
||||
// ParseParams parses key=value lines into generation parameters.
|
||||
func ParseParams(input string, current backend.GenerationParams) (backend.GenerationParams, error) {
|
||||
p := current
|
||||
for _, line := range strings.Split(input, "\n") {
|
||||
|
|
@ -303,112 +384,3 @@ func ParseParams(input string, current backend.GenerationParams) (backend.Genera
|
|||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) content(name string) string {
|
||||
s.sess.mu.RLock()
|
||||
defer s.sess.mu.RUnlock()
|
||||
switch name {
|
||||
case "backend":
|
||||
return s.sess.Core.BackendName() + "\n"
|
||||
case "agent":
|
||||
return s.sess.Core.AgentName() + "\n"
|
||||
case "model":
|
||||
return s.sess.Core.ModelName() + "\n"
|
||||
case "state":
|
||||
return s.sess.Core.State() + "\n"
|
||||
case "cwd":
|
||||
return s.sess.Core.CWD() + "\n"
|
||||
case "usage":
|
||||
return s.sess.Core.Usage() + "\n"
|
||||
case "ctxsz":
|
||||
return s.sess.Core.CtxSz() + "\n"
|
||||
case "models":
|
||||
return s.sess.Core.ListModels() + "\n"
|
||||
case "systemprompt":
|
||||
return s.sess.Core.SystemPrompt()
|
||||
case "offset":
|
||||
return fmt.Sprintf("%d\n", s.sess.ChatOffset)
|
||||
case "params":
|
||||
return FormatParams(s.sess.Core.GenerationParams())
|
||||
case "tail":
|
||||
return "#!/bin/sh\nexec tail -f \"$(dirname \"$0\")/chat\"\n"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// 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()
|
||||
case "usagewait":
|
||||
return s.sess.Core.Usage()
|
||||
case "ctxszwait":
|
||||
return s.sess.Core.CtxSz()
|
||||
case "cwdwait":
|
||||
return s.sess.Core.CWD()
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) makePublish() func(agent.Event) {
|
||||
assistantStarted := false
|
||||
return func(ev agent.Event) {
|
||||
if ev.Role == "user" {
|
||||
if assistantStarted {
|
||||
s.sess.AppendLog([]byte("\n"))
|
||||
assistantStarted = false
|
||||
}
|
||||
s.sess.mu.Lock()
|
||||
s.sess.ChatOffset = len(s.sess.log)
|
||||
s.sess.mu.Unlock()
|
||||
} else {
|
||||
switch ev.Role {
|
||||
case "call", "tool":
|
||||
if assistantStarted {
|
||||
s.sess.AppendLog([]byte("\n"))
|
||||
assistantStarted = false
|
||||
}
|
||||
case "assistant":
|
||||
if !assistantStarted {
|
||||
s.sess.AppendLog([]byte("assistant: "))
|
||||
assistantStarted = true
|
||||
}
|
||||
}
|
||||
}
|
||||
s.sess.AppendLog(FormatEvent(ev))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SessionFileStore) handleCtl(input string) error {
|
||||
cmd := strings.Fields(input)
|
||||
if len(cmd) == 0 {
|
||||
return fmt.Errorf("empty ctl command")
|
||||
}
|
||||
switch cmd[0] {
|
||||
case "stop":
|
||||
s.sess.Core.Interrupt(agent.ErrInterrupted)
|
||||
case "kill":
|
||||
s.kill()
|
||||
case "rn":
|
||||
if name := strings.TrimSpace(input[3:]); name != "" {
|
||||
if err := s.rename(name); err != nil {
|
||||
s.log.Error("rename: %v", err)
|
||||
}
|
||||
}
|
||||
case "save":
|
||||
s.sess.mu.RLock()
|
||||
data := make([]byte, len(s.sess.log))
|
||||
copy(data, s.sess.log)
|
||||
s.sess.mu.RUnlock()
|
||||
return s.saveTranscript(data)
|
||||
case "compact", "clear", "backend", "model", "models",
|
||||
"agents", "agent", "sessions", "cwd", "skills",
|
||||
"tools", "context", "usage", "history",
|
||||
"irw", "help":
|
||||
s.sess.Core.Submit(s.sess.Ctx, "/"+input, s.makePublish())
|
||||
default:
|
||||
return fmt.Errorf("unknown ctl command: %s", cmd[0])
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
|
|||
Reference in New Issue