diff --git a/fs/handlers.go b/fs/handlers.go index 9432bc1..3360789 100644 --- a/fs/handlers.go +++ b/fs/handlers.go @@ -152,9 +152,8 @@ func requestRoute(_ HandlerCtx, data []byte) ([]byte, error) { } func readAliases(_ HandlerCtx) ([]byte, error) { - rs := globalRootState var sb strings.Builder - for name, sess := range rs.Manager.Sessions() { + for name, sess := range session.Sessions() { sid := sess.ID() fmt.Fprintf(&sb, "%s\tsession/%s\n", sid, name) for _, ag := range sess.Agents() { @@ -231,7 +230,7 @@ func writeElevateRequest(ctx HandlerCtx, data []byte) error { } func readSessionIdx(_ HandlerCtx) ([]byte, error) { - return globalRootState.Manager.BuildIndex(), nil + return session.BuildIndex(), nil } func requestSessionNew(_ HandlerCtx, data []byte) ([]byte, error) { @@ -242,7 +241,7 @@ func requestSessionNew(_ HandlerCtx, data []byte) ([]byte, error) { name = v } } - sess, err := globalRootState.Manager.CreateEmpty(name) + sess, err := session.CreateEmpty(name) if err != nil { return nil, err } @@ -253,7 +252,7 @@ func requestSessionNew(_ HandlerCtx, data []byte) ([]byte, error) { // The Blueprint pattern clones the structure and binds ctx.Session. func sessionBindings(_ HandlerCtx) ([]Binding, error) { rs := globalRootState - sessions := rs.Manager.Sessions() + sessions := session.Sessions() out := make([]Binding, 0, len(sessions)) for name, sess := range sessions { @@ -268,11 +267,11 @@ func sessionBindings(_ HandlerCtx) ([]Binding, error) { Applier: func(c HandlerCtx) HandlerCtx { c.Session = node c.Remove = func() { - rs.Manager.Kill(n) + session.Kill(n) rs.RemoveNode(n) } c.Rename = func(newName string) error { - if err := rs.Manager.Rename(n, newName); err != nil { + if err := session.Rename(n, newName); err != nil { return err } rs.RenameNode(n, newName) @@ -281,11 +280,11 @@ func sessionBindings(_ HandlerCtx) ([]Binding, error) { return c }, Remove: func() { - rs.Manager.Kill(n) + session.Kill(n) rs.RemoveNode(n) }, Rename: func(newName string) error { - if err := rs.Manager.Rename(n, newName); err != nil { + if err := session.Rename(n, newName); err != nil { return err } rs.RenameNode(n, newName) @@ -390,7 +389,7 @@ func writeSessionElevate(ctx HandlerCtx, data []byte) error { func requestAgentNew(ctx HandlerCtx, data []byte) ([]byte, error) { args := strings.Fields(string(data)) - ag, err := globalRootState.Manager.CreateAgent(ctx.Session.Name(), args) + ag, err := session.CreateAgent(ctx.Session.Name(), args) if err != nil { return nil, fmt.Errorf("create agent: %w", err) } diff --git a/fs/lifecycle.go b/fs/lifecycle.go index f324826..b73003b 100644 --- a/fs/lifecycle.go +++ b/fs/lifecycle.go @@ -7,16 +7,15 @@ import ( ) // globalRootState is set by NewRoot so handler functions can access -// the session manager without going through ctx.Root. +// root state without going through ctx.Root. var globalRootState *RootState // RootState holds the root-level state for the 9P filesystem. type RootState struct { - Cfg Config - Manager *session.Manager + Cfg Config // SessionNodes caches the fs-layer wrappers for sessions. - // Keyed by session name (mutable), must be kept in sync with Manager. + // Keyed by session name (mutable), must be kept in sync with session registry. nodesMu sync.RWMutex SessionNodes map[string]*SessionNode } @@ -64,18 +63,15 @@ func state(root *Tree) *RootState { // Shutdown interrupts all sessions, waits for idle, persists, and closes. func Shutdown(root *Tree) { - rs := state(root) - rs.Manager.Shutdown() + session.Shutdown() } // InterruptAll interrupts every active session. func InterruptAll(root *Tree) { - rs := state(root) - rs.Manager.InterruptAll() + session.InterruptAll() } // Lookup returns the session for the given ID, or nil. func Lookup(root *Tree, id string) *session.Session { - rs := state(root) - return rs.Manager.Lookup(id) + return session.Lookup(id) } diff --git a/fs/newroot.go b/fs/newroot.go index 11b4306..f6ee05a 100644 --- a/fs/newroot.go +++ b/fs/newroot.go @@ -43,14 +43,13 @@ func NewRoot(cfg Config) *Tree { treeSpec.UID = treeSpecUID // root node: owner is system user treeSpec.GID = treeSpecGID // root node: group is "agent" - // Create session manager - mgr := session.NewManager(session.ManagerConfig{ + // Initialize session package + session.Init(session.InitConfig{ Ctx: cfg.Ctx, Log: cfg.Log, Sink: cfg.Sink, AgentsDir: cfg.AgentsDir, SessionsDir: cfg.SessionsDir, - MkdirAll: cfg.MkdirAll, Yolo: cfg.Yolo, }) @@ -72,7 +71,6 @@ func NewRoot(cfg Config) *Tree { // Initialize root state. rs := &RootState{ Cfg: cfg, - Manager: mgr, SessionNodes: make(map[string]*SessionNode), } tree.Data = rs @@ -82,4 +80,4 @@ func NewRoot(cfg Config) *Tree { restoreAllSessions(rs) return tree -} \ No newline at end of file +} diff --git a/fs/persist.go b/fs/persist.go index 8a5c6f2..1e0505b 100644 --- a/fs/persist.go +++ b/fs/persist.go @@ -19,7 +19,7 @@ func activeSessionsDir(rs *RootState) string { } func persistSession(rs *RootState, name string) { - sess := rs.Manager.Lookup(name) + sess := session.Lookup(name) if sess == nil { return } @@ -41,7 +41,7 @@ func removePersistedSession(rs *RootState, id string) { } func saveAllSessions(rs *RootState) { - for name := range rs.Manager.Sessions() { + for name := range session.Sessions() { persistSession(rs, name) } } @@ -194,7 +194,7 @@ func restoreSession(rs *RootState, ps *agent.PersistedAgent) error { replayMessagesToLog(al, ps.Messages) // Register session and create SessionNode - rs.Manager.Register(sess.Name(), sess) + session.Register(sess.Name(), sess) node := rs.GetOrCreateNode(sess) node.SetAgentLog(uname, al) diff --git a/session/manager.go b/session/registry.go similarity index 66% rename from session/manager.go rename to session/registry.go index f5c6284..3128b32 100644 --- a/session/manager.go +++ b/session/registry.go @@ -16,51 +16,60 @@ import ( "ollie/toolsrv" ) -// ManagerConfig configures the session manager. -type ManagerConfig struct { +// Package-level session registry with thread-safe access. +var ( + mu sync.RWMutex + sessions = make(map[string]*Session) + + // Package config, set via Init. + pkgCtx context.Context + pkgLog *olog.Logger + pkgSink *olog.Sink + pkgAgentsDir string + pkgSessionsDir string + pkgYolo bool +) + +// InitConfig configures the session package. +type InitConfig struct { Ctx context.Context Log *olog.Logger Sink *olog.Sink AgentsDir string SessionsDir string - MkdirAll func(string, os.FileMode) error Yolo bool } -// Manager manages the lifecycle of all sessions. -type Manager struct { - cfg ManagerConfig - mu sync.RWMutex - sessions map[string]*Session -} - -// NewManager creates a session manager. -func NewManager(cfg ManagerConfig) *Manager { - return &Manager{ - cfg: cfg, - sessions: make(map[string]*Session), - } +// Init initializes the session package with the given configuration. +// Must be called before any other session functions. +func Init(cfg InitConfig) { + pkgCtx = cfg.Ctx + pkgLog = cfg.Log + pkgSink = cfg.Sink + pkgAgentsDir = cfg.AgentsDir + pkgSessionsDir = cfg.SessionsDir + pkgYolo = cfg.Yolo } // Sessions returns a snapshot of all sessions. -func (m *Manager) Sessions() map[string]*Session { - m.mu.RLock() - defer m.mu.RUnlock() - result := make(map[string]*Session, len(m.sessions)) - for k, v := range m.sessions { +func Sessions() map[string]*Session { + mu.RLock() + defer mu.RUnlock() + result := make(map[string]*Session, len(sessions)) + for k, v := range sessions { result[k] = v } return result } // Lookup returns the session by name or ID, or nil. -func (m *Manager) Lookup(nameOrID string) *Session { - m.mu.RLock() - defer m.mu.RUnlock() - if s := m.sessions[nameOrID]; s != nil { +func Lookup(nameOrID string) *Session { + mu.RLock() + defer mu.RUnlock() + if s := sessions[nameOrID]; s != nil { return s } - for _, sess := range m.sessions { + for _, sess := range sessions { if sess.ID() == nameOrID { return sess } @@ -68,23 +77,23 @@ func (m *Manager) Lookup(nameOrID string) *Session { return nil } -// Register adds a session to the manager. -func (m *Manager) Register(name string, sess *Session) { - m.mu.Lock() - defer m.mu.Unlock() - m.sessions[name] = sess +// Register adds a session to the registry. +func Register(name string, sess *Session) { + mu.Lock() + sessions[name] = sess + mu.Unlock() } -// Kill terminates a session and removes it from the manager. -func (m *Manager) Kill(name string) error { - m.mu.Lock() - sess := m.sessions[name] +// Kill terminates a session and removes it from the registry. +func Kill(name string) error { + mu.Lock() + sess := sessions[name] if sess == nil { - m.mu.Unlock() + mu.Unlock() return fmt.Errorf("session not found: %s", name) } - delete(m.sessions, name) - m.mu.Unlock() + delete(sessions, name) + mu.Unlock() var agentIDs []string for _, ag := range sess.Agents() { @@ -98,7 +107,7 @@ func (m *Manager) Kill(name string) error { sess.Proc().Close() } - m.cfg.Log.Info("killed session %s", name) + pkgLog.Info("killed session %s", name) for _, agentID := range agentIDs { PublishEvent("session."+sess.ID()+".agent."+agentID+".kill", "") } @@ -107,50 +116,50 @@ func (m *Manager) Kill(name string) error { } // Rename renames a session. -func (m *Manager) Rename(oldName, newName string) error { - m.mu.Lock() - sess, ok := m.sessions[oldName] +func Rename(oldName, newName string) error { + mu.Lock() + sess, ok := sessions[oldName] if !ok { - m.mu.Unlock() + mu.Unlock() return fmt.Errorf("session not found: %s", oldName) } - if _, exists := m.sessions[newName]; exists { - m.mu.Unlock() + if _, exists := sessions[newName]; exists { + mu.Unlock() return fmt.Errorf("session already exists: %s", newName) } for _, ag := range sess.Agents() { if ag.IsRunning() { - m.mu.Unlock() + mu.Unlock() return fmt.Errorf("cannot rename while agent is running") } } sess.SetName(newName) - m.sessions[newName] = sess - delete(m.sessions, oldName) - m.mu.Unlock() + sessions[newName] = sess + delete(sessions, oldName) + mu.Unlock() - m.cfg.Log.Info("renamed session %s -> %s", oldName, newName) + pkgLog.Info("renamed session %s -> %s", oldName, newName) PublishEvent("session."+sess.ID()+".rename", oldName+" "+newName) return nil } // Shutdown interrupts all sessions, waits for idle, and closes them. -func (m *Manager) Shutdown() { - m.InterruptAll() - m.waitIdle(100*time.Millisecond, 5*time.Second) +func Shutdown() { + InterruptAll() + waitIdle(100*time.Millisecond, 5*time.Second) - m.mu.Lock() - names := make([]string, 0, len(m.sessions)) - for name := range m.sessions { + mu.Lock() + names := make([]string, 0, len(sessions)) + for name := range sessions { names = append(names, name) } - m.mu.Unlock() + mu.Unlock() for _, name := range names { - m.mu.Lock() - sess := m.sessions[name] - delete(m.sessions, name) - m.mu.Unlock() + mu.Lock() + sess := sessions[name] + delete(sessions, name) + mu.Unlock() if sess != nil { sess.Cancel() sess.Close() @@ -159,28 +168,28 @@ func (m *Manager) Shutdown() { } else if sess.Proc() != nil { sess.Proc().Close() } - m.cfg.Log.Info("shutdown session %s", name) + pkgLog.Info("shutdown session %s", name) } } } // InterruptAll interrupts every active session. -func (m *Manager) InterruptAll() { - m.mu.RLock() - defer m.mu.RUnlock() - for _, sess := range m.sessions { +func InterruptAll() { + mu.RLock() + defer mu.RUnlock() + for _, sess := range sessions { for _, ag := range sess.Agents() { ag.Interrupt(agent.ErrInterrupted) } } } -func (m *Manager) waitIdle(poll, timeout time.Duration) { +func waitIdle(poll, timeout time.Duration) { deadline := time.Now().Add(timeout) for time.Now().Before(deadline) { allIdle := true - m.mu.RLock() - for _, sess := range m.sessions { + mu.RLock() + for _, sess := range sessions { for _, ag := range sess.Agents() { if ag.State() != "idle" { allIdle = false @@ -191,26 +200,26 @@ func (m *Manager) waitIdle(poll, timeout time.Duration) { break } } - m.mu.RUnlock() + mu.RUnlock() if allIdle { return } time.Sleep(poll) } - m.cfg.Log.Warn("shutdown: timed out waiting for idle") + pkgLog.Warn("shutdown: timed out waiting for idle") } // BuildIndex builds the session index content. -func (m *Manager) BuildIndex() []byte { +func BuildIndex() []byte { var sb strings.Builder - m.mu.RLock() - names := make([]string, 0, len(m.sessions)) - for name := range m.sessions { + mu.RLock() + names := make([]string, 0, len(sessions)) + for name := range sessions { names = append(names, name) } sort.Strings(names) for _, name := range names { - sess := m.sessions[name] + sess := sessions[name] paused := sess.IsPaused() connected := sess.IsConnected() cwd := sess.CWD() @@ -227,12 +236,12 @@ func (m *Manager) BuildIndex() []byte { fmt.Fprintf(&sb, "%s\t%s\t%t\t%t\t%s\t%s\t%s\t%s\n", name, sess.ID(), paused, connected, cwd, be, model, agents) } - m.mu.RUnlock() + mu.RUnlock() return []byte(sb.String()) } // CreateEmpty creates an empty session (no agent). -func (m *Manager) CreateEmpty(name string) (*Session, error) { +func CreateEmpty(name string) (*Session, error) { sessID := NewSessionID() if name == "" { name = sessID @@ -241,20 +250,20 @@ func (m *Manager) CreateEmpty(name string) (*Session, error) { } } - m.mu.RLock() - _, exists := m.sessions[name] - m.mu.RUnlock() + mu.RLock() + _, exists := sessions[name] + mu.RUnlock() if exists { return nil, fmt.Errorf("session already exists: %s", name) } - ctx, cancel := context.WithCancel(m.cfg.Ctx) + ctx, cancel := context.WithCancel(pkgCtx) sess := NewEmpty(sessID, ctx, cancel) // Spawn tool server cwd, _ := os.Getwd() var dialOpts []toolsrv.Option - if m.cfg.Yolo { + if pkgYolo { dialOpts = append(dialOpts, toolsrv.WithYolo()) } proc, err := toolsrv.Spawn(ctx, cwd, dialOpts...) @@ -274,11 +283,11 @@ func (m *Manager) CreateEmpty(name string) (*Session, error) { } sess.SetToolsConn(conn) - m.mu.Lock() - m.sessions[name] = sess - m.mu.Unlock() + mu.Lock() + sessions[name] = sess + mu.Unlock() - m.cfg.Log.Info("created empty session %s (id=%s)", name, sessID) + pkgLog.Info("created empty session %s (id=%s)", name, sessID) PublishEvent("session."+sessID+".new", name) return sess, nil } @@ -292,8 +301,8 @@ func NextUname() string { // CreateAgent creates a new agent within an existing session. // Args: [profile] [key=value...] where keys include backend, model, remote. -func (m *Manager) CreateAgent(sessName string, args []string) (*agent.Agent, error) { - sess := m.Lookup(sessName) +func CreateAgent(sessName string, args []string) (*agent.Agent, error) { + sess := Lookup(sessName) if sess == nil { return nil, fmt.Errorf("session not found: %s", sessName) } @@ -316,7 +325,7 @@ func (m *Manager) CreateAgent(sessName string, args []string) (*agent.Agent, err } } - cfg := LoadAgentConfig(m.cfg.AgentsDir, profile, nil) + cfg := LoadAgentConfig(pkgAgentsDir, profile, nil) uname := NextUname() sessID := sess.ID() @@ -327,7 +336,7 @@ func (m *Manager) CreateAgent(sessName string, args []string) (*agent.Agent, err Ctx: sess.Ctx(), CWD: cwd, RemoteTarget: remoteTarget, - Yolo: m.cfg.Yolo, + Yolo: pkgYolo, ReuseFrom: &InfraConfig{ Proc: sess.Proc(), Keeper: sess.Keeper(), @@ -343,7 +352,7 @@ func (m *Manager) CreateAgent(sessName string, args []string) (*agent.Agent, err // Load tools LoadAutoLoadTools(cfg, infra.ToolsConn, sessID, uname, func(f string, a ...any) { - m.cfg.Log.Error("session %s agent %s: "+f, append([]any{sessID, uname}, a...)...) + pkgLog.Error("session %s agent %s: "+f, append([]any{sessID, uname}, a...)...) }) // Build runtime @@ -366,18 +375,18 @@ func (m *Manager) CreateAgent(sessName string, args []string) (*agent.Agent, err History: nil, // new agent starts with empty history Runtime: rt, Profile: profile, - AgentsDir: m.cfg.AgentsDir, + AgentsDir: pkgAgentsDir, AgentID: uname, CWD: cwd, BaseLayers: layers.BaseLayers(), PromptEnvExtra: infra.PromptEnv, NewToolServer: infra.NewToolServer, NewBackend: backend.NewWithName, - Log: m.cfg.Sink.NewLogger("agent"), + Log: pkgSink.NewLogger("agent"), SessionID: sessID, }) sess.AddAgent(ag) - m.cfg.Log.Info("created agent %s in session %s (profile=%s)", uname, sessName, profile) + pkgLog.Info("created agent %s in session %s (profile=%s)", uname, sessName, profile) return ag, nil }