565 lines
15 KiB
Go
565 lines
15 KiB
Go
package fs
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"fmt"
|
|
"os"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
|
|
"ollie/agent"
|
|
olog "ollie/log"
|
|
coresession "ollie/session"
|
|
"ollie/skills"
|
|
"ollie/toolsrv"
|
|
)
|
|
|
|
// eventBroadcaster manages notification of session index changes.
|
|
// The GUI can tail session/eventwait to detect creates, deletes, and renames.
|
|
var (
|
|
evMu sync.Mutex
|
|
evChans []chan struct{}
|
|
)
|
|
|
|
func broadcastEvents() {
|
|
evMu.Lock()
|
|
for _, ch := range evChans {
|
|
select {
|
|
case ch <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
evMu.Unlock()
|
|
}
|
|
|
|
// eventRegister returns a channel that fires on the next session index change.
|
|
func eventRegister() chan struct{} {
|
|
ch := make(chan struct{}, 1)
|
|
evMu.Lock()
|
|
evChans = append(evChans, ch)
|
|
evMu.Unlock()
|
|
return ch
|
|
}
|
|
|
|
func eventRelease(ch chan struct{}) {
|
|
evMu.Lock()
|
|
for i, c := range evChans {
|
|
if c == ch {
|
|
evChans = append(evChans[:i], evChans[i+1:]...)
|
|
break
|
|
}
|
|
}
|
|
evMu.Unlock()
|
|
}
|
|
|
|
var rootFileOrder = []string{"new", "idx", "eventwait"}
|
|
|
|
// FileMode returns the mode for a fixed session file,
|
|
// or 0 and false if the name is not a fixed file.
|
|
func FileMode(name string) (os.FileMode, bool) {
|
|
m, ok := Perms[PathSessions].Files[name]
|
|
return m, ok
|
|
}
|
|
|
|
// Config holds the dependencies for the session tree.
|
|
type Config struct {
|
|
Ctx context.Context // daemon context — sessions derive from this
|
|
AgentsDir string
|
|
SessionsDir string
|
|
|
|
Log *olog.Logger
|
|
Sink *olog.Sink
|
|
ReadFile func(string) ([]byte, error)
|
|
MkdirAll func(string, os.FileMode) error
|
|
// NewCore, if non-nil, replaces the default backend.New + agent.New
|
|
// path. It receives the session ID, agent name, and cwd, and returns a Session.
|
|
NewCore func(sessionID, agentName, cwd string) (*coresession.Session, error)
|
|
// Strict rejects inline code steps; only tool steps are allowed.
|
|
Strict bool
|
|
// Yolo skips the landrun sandbox.
|
|
Yolo bool
|
|
// NoMount disables the per-session FUSE mount (e.g. when listening on TCP).
|
|
NoMount bool
|
|
// Enable9P indicates the 9P listener is active (for operational model injection).
|
|
Enable9P bool
|
|
// InvalidateModels clears the model cache, forcing a refresh.
|
|
InvalidateModels func()
|
|
// ToolRegistry is the shared tool registry for lazy tool promotion.
|
|
ToolRegistry *toolsrv.Registry
|
|
// SkillsRegistry is the shared skills registry for skill loading.
|
|
SkillsRegistry *skills.Registry
|
|
}
|
|
|
|
// rootState is the package-level state stored in root.Data.
|
|
type rootState struct {
|
|
cfg Config
|
|
mu sync.RWMutex
|
|
sessions map[string]*Session
|
|
}
|
|
|
|
func (rs *rootState) nextUname() string {
|
|
b := make([]byte, 6)
|
|
rand.Read(b) //nolint:errcheck
|
|
return fmt.Sprintf("%06x%06x", b[:3], b[3:6])
|
|
}
|
|
|
|
// state extracts rootState from a tree's Data field.
|
|
func state(root *Tree) *rootState {
|
|
return root.Data.(*rootState)
|
|
}
|
|
|
|
// NewRoot creates the session tree with all callbacks wired up.
|
|
// The returned tree IS the session collection.
|
|
func NewRoot(cfg Config) *Tree {
|
|
if cfg.ReadFile == nil {
|
|
cfg.ReadFile = os.ReadFile
|
|
}
|
|
if cfg.MkdirAll == nil {
|
|
cfg.MkdirAll = os.MkdirAll
|
|
}
|
|
rs := &rootState{
|
|
cfg: cfg,
|
|
sessions: make(map[string]*Session),
|
|
}
|
|
restoreAllSessions(rs)
|
|
tree := NewTree(nil, 0,
|
|
WithStat(func(_ []string, name string) (os.FileInfo, error) { return rootStat(rs, name) }),
|
|
WithOpener(func(_ []string, name string) (File, error) { return rootOpen(rs, name) }),
|
|
WithLister(func(_ []string) ([]os.DirEntry, error) { return rootList(rs) }),
|
|
WithReaddir(func(_ []string, name string) ([]os.DirEntry, error) { return rootReaddir(rs, name) }),
|
|
WithDeleter(func(_ []string, name string) error { return rootDel(rs, name) }),
|
|
WithCreator(func(_ []string, name string, _ os.FileMode) error {
|
|
return fmt.Errorf("create not supported: %s", name)
|
|
}),
|
|
WithRenamer(func(_ []string, old, new string) error {
|
|
if strings.Contains(old, "/agent/") {
|
|
parts := strings.SplitN(old, "/", 3)
|
|
if len(parts) < 3 {
|
|
return fmt.Errorf("invalid path: %s", old)
|
|
}
|
|
sessName := parts[0]
|
|
rs.mu.RLock()
|
|
sess := rs.sessions[sessName]
|
|
rs.mu.RUnlock()
|
|
if sess == nil {
|
|
return fmt.Errorf("session not found: %s", sessName)
|
|
}
|
|
sess.Core.Agent().SetName(new)
|
|
rs.cfg.Log.Info("renamed agent in session %s -> %s", sessName, new)
|
|
broadcastEvents()
|
|
return nil
|
|
}
|
|
return Rename(rs, old, new)
|
|
}),
|
|
)
|
|
tree.Data = rs
|
|
return tree
|
|
}
|
|
|
|
// Lookup returns the session for the given ID, or nil.
|
|
func Lookup(root *Tree, id string) *Session {
|
|
rs := state(root)
|
|
rs.mu.RLock()
|
|
defer rs.mu.RUnlock()
|
|
return rs.sessions[id]
|
|
}
|
|
|
|
// AddSession inserts a pre-built session (used by tests).
|
|
func AddSession(root *Tree, sess *Session) {
|
|
rs := state(root)
|
|
rs.mu.Lock()
|
|
rs.sessions[sess.RunnableID()] = sess
|
|
rs.mu.Unlock()
|
|
}
|
|
|
|
// LookupByUname returns the session with the given uname (principal), or nil.
|
|
func LookupByUname(root *Tree, uname string) *Session {
|
|
rs := state(root)
|
|
rs.mu.RLock()
|
|
defer rs.mu.RUnlock()
|
|
for _, sess := range rs.sessions {
|
|
if sess.uname == uname {
|
|
return sess
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// InterruptAll interrupts every active session.
|
|
func InterruptAll(root *Tree) {
|
|
rs := state(root)
|
|
rs.mu.RLock()
|
|
defer rs.mu.RUnlock()
|
|
for _, sess := range rs.sessions {
|
|
sess.Core.Agent().Interrupt(agent.ErrInterrupted)
|
|
}
|
|
}
|
|
|
|
// OpenSessionTree returns a Tree for the given session ID.
|
|
func OpenSessionTree(root *Tree, id string) (*Tree, error) {
|
|
rs := state(root)
|
|
sess := lookupLocked(rs, id)
|
|
if sess == nil {
|
|
return nil, fmt.Errorf("session not found: %s", id)
|
|
}
|
|
return openSessionTree(rs, sess), nil
|
|
}
|
|
|
|
func lookupLocked(rs *rootState, id string) *Session {
|
|
rs.mu.RLock()
|
|
defer rs.mu.RUnlock()
|
|
return rs.sessions[id]
|
|
}
|
|
|
|
func openSessionTree(rs *rootState, sess *Session) *Tree {
|
|
return NewSessionTree(
|
|
sess,
|
|
rs.cfg.Log,
|
|
func() { Kill(rs, sess.Name()) },
|
|
func(newID string) error { return Rename(rs, sess.Name(), newID) },
|
|
nil,
|
|
rs.cfg.InvalidateModels,
|
|
rs.cfg.ToolRegistry,
|
|
)
|
|
}
|
|
|
|
func openAgentTree(rs *rootState, sess *Session) *Tree {
|
|
return NewAgentTree(
|
|
sess,
|
|
sess.AgentLog,
|
|
rs.cfg.Log,
|
|
nil,
|
|
rs.cfg.InvalidateModels,
|
|
rs.cfg.ToolRegistry,
|
|
)
|
|
}
|
|
|
|
// --- Tree callbacks ---
|
|
|
|
func rootList(rs *rootState) ([]os.DirEntry, error) {
|
|
entries := make([]os.DirEntry, 0, len(rootFileOrder))
|
|
for _, name := range rootFileOrder {
|
|
entries = append(entries, FileEntry(name, Perms[PathSessions].Files[name]))
|
|
}
|
|
rs.mu.RLock()
|
|
for id := range rs.sessions {
|
|
entries = append(entries, DirEntry(id, Perms[PathSessionDir].DirMode))
|
|
}
|
|
rs.mu.RUnlock()
|
|
return entries, nil
|
|
}
|
|
|
|
func rootReaddir(rs *rootState, name string) ([]os.DirEntry, error) {
|
|
parts := strings.SplitN(name, "/", 4)
|
|
sessID := parts[0]
|
|
rs.mu.RLock()
|
|
sess := rs.sessions[sessID]
|
|
rs.mu.RUnlock()
|
|
if sess == nil {
|
|
return nil, fmt.Errorf("session not found: %s", sessID)
|
|
}
|
|
if len(parts) == 1 {
|
|
entries := []os.DirEntry{
|
|
FileEntry("plan", 0666),
|
|
FileEntry("env", 0444),
|
|
DirEntry("agent", 0755),
|
|
}
|
|
return entries, nil
|
|
}
|
|
if len(parts) >= 2 && parts[1] == "agent" {
|
|
if len(parts) == 2 {
|
|
entries := []os.DirEntry{FileEntry("new", 0666)}
|
|
if sess.Core != nil && sess.Core.Agent() != nil {
|
|
agentName := sess.Core.Agent().Name()
|
|
aid := sess.Core.Agent().ID()
|
|
dirName := aid
|
|
if agentName != "" {
|
|
dirName = agentName
|
|
}
|
|
entries = append(entries, DirEntry(dirName, 0755))
|
|
}
|
|
return entries, nil
|
|
}
|
|
if len(parts) >= 3 && parts[2] == "new" {
|
|
if len(parts) == 3 {
|
|
return nil, fmt.Errorf("%s: is a directory", name)
|
|
}
|
|
return nil, fmt.Errorf("%s: not a directory", name)
|
|
}
|
|
if sess.Core == nil || sess.Core.Agent() == nil {
|
|
return nil, fmt.Errorf("no agent in session %s", sessID)
|
|
}
|
|
if len(parts) >= 3 {
|
|
agentName := sess.Core.Agent().Name()
|
|
if agentName == "" || parts[2] != agentName {
|
|
return nil, fmt.Errorf("agent %s not found in session %s", parts[2], sessID)
|
|
}
|
|
}
|
|
if len(parts) == 3 {
|
|
afs := openAgentTree(rs, sess)
|
|
entries, err := afs.List()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
entries = append(entries, DirEntry("proc", 0755))
|
|
return entries, nil
|
|
}
|
|
if len(parts) == 4 && parts[3] == "proc" {
|
|
procs := sess.Core.Agent().ListDetached()
|
|
entries := make([]os.DirEntry, len(procs))
|
|
for i, p := range procs {
|
|
entries[i] = FileEntry(fmt.Sprintf("%d", p.PID), 0666)
|
|
}
|
|
return entries, nil
|
|
}
|
|
}
|
|
return nil, fmt.Errorf("%s: not a directory", name)
|
|
}
|
|
|
|
func rootStat(rs *rootState, name string) (os.FileInfo, error) {
|
|
if mode, ok := Perms[PathSessions].Files[name]; ok {
|
|
return &SyntheticFileInfo{Name_: name, Mode_: mode}, nil
|
|
}
|
|
parts := strings.SplitN(name, "/", 4)
|
|
sessID := parts[0]
|
|
rs.mu.RLock()
|
|
sess, ok := rs.sessions[sessID]
|
|
rs.mu.RUnlock()
|
|
if !ok {
|
|
return nil, fmt.Errorf("%s: not found", name)
|
|
}
|
|
if len(parts) == 1 {
|
|
return &SyntheticFileInfo{Name_: sessID, Mode_: Perms[PathSessionDir].DirMode, IsDir_: true}, nil
|
|
}
|
|
if len(parts) >= 2 && parts[1] == "agent" {
|
|
if len(parts) == 2 {
|
|
return &SyntheticFileInfo{Name_: "agent", Mode_: 0755, IsDir_: true}, nil
|
|
}
|
|
if len(parts) >= 3 && parts[2] == "new" {
|
|
return &SyntheticFileInfo{Name_: "new", Mode_: Perms[PathSessionFile].Files["new"]}, nil
|
|
}
|
|
if len(parts) >= 3 {
|
|
if sess.Core == nil || sess.Core.Agent() == nil {
|
|
return nil, fmt.Errorf("no agent in session %s", sessID)
|
|
}
|
|
agentName := sess.Core.Agent().Name()
|
|
if agentName == "" || parts[2] != agentName {
|
|
return nil, fmt.Errorf("agent %s not found in session %s", parts[2], sessID)
|
|
}
|
|
}
|
|
if len(parts) == 3 {
|
|
return &SyntheticFileInfo{Name_: parts[2], Mode_: 0755, IsDir_: true}, nil
|
|
}
|
|
if len(parts) >= 4 && parts[3] == "proc" {
|
|
return &SyntheticFileInfo{Name_: "proc", Mode_: 0755, IsDir_: true}, nil
|
|
}
|
|
afs := openAgentTree(rs, sess)
|
|
return afs.Stat(parts[3])
|
|
}
|
|
sfs := openSessionTree(rs, sess)
|
|
return sfs.Stat(parts[1])
|
|
}
|
|
|
|
func rootOpen(rs *rootState, name string) (File, error) {
|
|
notBlocking := func(context.Context, string) ([]byte, string, error) {
|
|
return nil, "", fmt.Errorf("blocking read not supported")
|
|
}
|
|
|
|
parts := strings.SplitN(name, "/", 3)
|
|
|
|
// Proc file: {id}/proc/{pid}
|
|
if len(parts) == 3 && parts[1] == "proc" {
|
|
sessID := parts[0]
|
|
rs.mu.RLock()
|
|
sess := rs.sessions[sessID]
|
|
rs.mu.RUnlock()
|
|
if sess == nil {
|
|
return nil, fmt.Errorf("%s: not found", name)
|
|
}
|
|
pidStr := parts[2]
|
|
pid, err := strconv.Atoi(pidStr)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("invalid pid: %s", pidStr)
|
|
}
|
|
return &FileConfig{
|
|
StatFn: func() (os.FileInfo, error) {
|
|
return &SyntheticFileInfo{Name_: pidStr, Mode_: 0444}, nil
|
|
},
|
|
ReadFn: func() ([]byte, error) {
|
|
output, err := sess.Core.Agent().GetDetachedOutput(pid)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return []byte(output), nil
|
|
},
|
|
WriteFn: func([]byte) error { return fmt.Errorf("read-only") },
|
|
BlockingReadFn: notBlocking,
|
|
}, nil
|
|
}
|
|
|
|
switch name {
|
|
case "eventwait":
|
|
return &FileConfig{
|
|
StatFn: func() (os.FileInfo, error) {
|
|
return &SyntheticFileInfo{Name_: "eventwait", Mode_: Perms[PathSessions].Files["eventwait"]}, nil
|
|
},
|
|
ReadFn: func() ([]byte, error) {
|
|
return []byte(buildIndex(rs)), nil
|
|
},
|
|
WriteFn: func([]byte) error { return fmt.Errorf("eventwait: read-only") },
|
|
BlockingReadFn: func(ctx context.Context, _ string) ([]byte, string, error) {
|
|
ch := eventRegister()
|
|
defer eventRelease(ch)
|
|
select {
|
|
case <-ch:
|
|
return []byte(buildIndex(rs)), "", nil
|
|
case <-ctx.Done():
|
|
return nil, "", ctx.Err()
|
|
}
|
|
},
|
|
IsBlocking_: true,
|
|
}, nil
|
|
case "new":
|
|
return &FileConfig{
|
|
StatFn: func() (os.FileInfo, error) {
|
|
return &SyntheticFileInfo{Name_: "new", Mode_: Perms[PathSessions].Files["new"]}, nil
|
|
},
|
|
ReadFn: func() ([]byte, error) {
|
|
return []byte("name=\n"), nil
|
|
},
|
|
WriteFn: func(data []byte) error {
|
|
return fmt.Errorf("use ollie-9p rdwr session/new")
|
|
},
|
|
BlockingReadFn: notBlocking,
|
|
}, nil
|
|
case "idx":
|
|
return &FileConfig{
|
|
StatFn: func() (os.FileInfo, error) {
|
|
return &SyntheticFileInfo{Name_: "idx", Mode_: Perms[PathSessions].Files["idx"]}, nil
|
|
},
|
|
ReadFn: func() ([]byte, error) { return buildIndex(rs), nil },
|
|
WriteFn: func([]byte) error { return fmt.Errorf("idx: read-only") },
|
|
BlockingReadFn: notBlocking,
|
|
}, nil
|
|
}
|
|
|
|
// Hierarchical paths: {id}/{file} or {id}/agent/{aid}/{file}
|
|
parts = strings.SplitN(name, "/", 4)
|
|
sessID := parts[0]
|
|
rs.mu.RLock()
|
|
sess := rs.sessions[sessID]
|
|
rs.mu.RUnlock()
|
|
if sess == nil {
|
|
return nil, fmt.Errorf("%s: not found", name)
|
|
}
|
|
if len(parts) == 1 {
|
|
return nil, fmt.Errorf("%s: is a directory", name)
|
|
}
|
|
|
|
if len(parts) >= 2 && parts[1] == "agent" {
|
|
if len(parts) == 2 {
|
|
return nil, fmt.Errorf("%s: is a directory", name)
|
|
}
|
|
if len(parts) >= 3 && parts[2] == "new" {
|
|
if len(parts) == 3 {
|
|
// agent/new — request-response file to create an agent
|
|
var agentResult string
|
|
return &FileConfig{
|
|
StatFn: func() (os.FileInfo, error) {
|
|
return &SyntheticFileInfo{Name_: "new", Mode_: Perms[PathSessionFile].Files["new"]}, nil
|
|
},
|
|
ReadFn: func() ([]byte, error) {
|
|
if agentResult != "" {
|
|
return []byte(agentResult + "\n"), nil
|
|
}
|
|
return []byte("cwd=\nbackend=\nmodel=\nagent=\nagent_alias=\nremote=\nsystem_prompt=\n"), nil
|
|
},
|
|
WriteFn: func(data []byte) error {
|
|
if err := CreateAgent(rs, sessID, strings.Fields(strings.TrimSpace(string(data)))); err != nil {
|
|
return err
|
|
}
|
|
agentResult = "ok"
|
|
return nil
|
|
},
|
|
BlockingReadFn: notBlocking,
|
|
}, nil
|
|
}
|
|
return nil, fmt.Errorf("%s: not a file", name)
|
|
}
|
|
if len(parts) < 4 {
|
|
return nil, fmt.Errorf("%s: is a directory", name)
|
|
}
|
|
if sess.Core == nil || sess.Core.Agent() == nil {
|
|
return nil, fmt.Errorf("no agent in session %s", sessID)
|
|
}
|
|
agentName := sess.Core.Agent().Name()
|
|
if agentName == "" || parts[2] != agentName {
|
|
return nil, fmt.Errorf("agent %s not found in session %s", parts[2], sessID)
|
|
}
|
|
afs := openAgentTree(rs, sess)
|
|
return afs.Open(parts[3])
|
|
}
|
|
|
|
sfs := openSessionTree(rs, sess)
|
|
return sfs.Open(parts[1])
|
|
}
|
|
|
|
func rootDel(rs *rootState, name string) error {
|
|
parts := strings.SplitN(name, "/", 3)
|
|
sessID := parts[0]
|
|
if len(parts) == 1 {
|
|
rs.mu.RLock()
|
|
_, ok := rs.sessions[sessID]
|
|
rs.mu.RUnlock()
|
|
if ok {
|
|
Kill(rs, sessID)
|
|
return nil
|
|
}
|
|
return fmt.Errorf("session not found: %s", sessID)
|
|
}
|
|
if len(parts) == 3 && parts[1] == "proc" {
|
|
rs.mu.RLock()
|
|
sess := rs.sessions[sessID]
|
|
rs.mu.RUnlock()
|
|
if sess == nil {
|
|
return fmt.Errorf("session not found: %s", sessID)
|
|
}
|
|
pid, err := strconv.Atoi(parts[2])
|
|
if err != nil {
|
|
return fmt.Errorf("invalid pid: %s", parts[2])
|
|
}
|
|
if !sess.Core.Agent().DismissDetached(pid) {
|
|
return fmt.Errorf("process %d not found or still running", pid)
|
|
}
|
|
return nil
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func buildIndex(rs *rootState) []byte {
|
|
var sb strings.Builder
|
|
rs.mu.RLock()
|
|
ids := make([]string, 0, len(rs.sessions))
|
|
for id := range rs.sessions {
|
|
ids = append(ids, id)
|
|
}
|
|
sort.Strings(ids)
|
|
for _, id := range ids {
|
|
sess := rs.sessions[id]
|
|
sess.mu.RLock()
|
|
st := sess.Core.Agent().State()
|
|
cwd := sess.Core.CWD()
|
|
be := sess.Core.Agent().BackendName()
|
|
model := sess.Core.Agent().ModelName()
|
|
ag := sess.Core.Agent().Name()
|
|
name := sess.name
|
|
sess.mu.RUnlock()
|
|
fmt.Fprintf(&sb, "%s\t%s\t%s\t%s\t%s\t%s\n", name, st, cwd, be, model, ag)
|
|
}
|
|
rs.mu.RUnlock()
|
|
return []byte(sb.String())
|
|
}
|