lkn
/
ollie
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: eliminate Manager type, use package-level session registry 

- session/registry.go: package-level vars + functions replace Manager struct
- session.Init() initializes package config
- session.Sessions(), Lookup(), Register(), Kill(), Rename() etc are now top-level
- fs/ calls session.X() directly instead of rs.Manager.X()
- Same thread safety via package-level sync.RWMutex
This commit is contained in:
Levi Neely 2026-08-04 13:55:45 +02:00
parent 2aa7db8445
commit 7fc183c5cb
5 changed files with 125 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
fs/handlers.go
View File

@ -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)
}

16
fs/lifecycle.go
View File

@ -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)
}

8
fs/newroot.go
View File

@ -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
}
}

6
fs/persist.go
View File

@ -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)

199
session/manager.go → session/registry.go
View File

@ -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
}