diff --git a/fs/types.go b/fs/agent_log.go similarity index 63% rename from fs/types.go rename to fs/agent_log.go index 77893cf..2dcaf8f 100644 --- a/fs/types.go +++ b/fs/agent_log.go @@ -1,135 +1,13 @@ package fs import ( - "context" - "strings" "sync" - "time" "ollie/agent" "ollie/backend" "ollie/format" - "ollie/session" - "ollie/toolsrv" ) -// SessionNode wraps session.Session for 9P handlers. -// It adds AgentLogs (9P chat streaming state) and models cache. -type SessionNode struct { - Core *session.Session - - // AgentLogs for chat streaming, keyed by agent ID. - mu sync.RWMutex - AgentLogs map[string]*AgentLog - - // Cached model list (expensive API call; refreshed every 24h). - modelsMu sync.Mutex - modelsCache string - modelsCacheAt time.Time -} - -// NewSessionNode creates a SessionNode wrapping a session. -func NewSessionNode(core *session.Session) *SessionNode { - return &SessionNode{Core: core, AgentLogs: make(map[string]*AgentLog)} -} - -// --- Delegation to Core --- - -func (s *SessionNode) ID() string { return s.Core.ID() } -func (s *SessionNode) Name() string { return s.Core.Name() } -func (s *SessionNode) SetName(n string) { s.Core.SetName(n) } -func (s *SessionNode) Uname() string { return s.Core.Uname() } -func (s *SessionNode) Cancel() { s.Core.Cancel() } -func (s *SessionNode) IsPaused() bool { return s.Core.IsPaused() } -func (s *SessionNode) IsConnected() bool { return s.Core.IsConnected() } - -func (s *SessionNode) Pause() error { - err := s.Core.Pause() - if err == nil { - session.PublishEvent("session."+s.Core.ID()+".pause", "") - } - return err -} - -func (s *SessionNode) Resume() error { - err := s.Core.Resume() - if err == nil { - session.PublishEvent("session."+s.Core.ID()+".resume", "") - } - return err -} - -func (s *SessionNode) LoadTool(name string, ag *agent.Agent) error { - return s.Core.LoadTool(name, ag) -} - -func (s *SessionNode) Ctx() context.Context { - return s.Core.Ctx() -} - -func (s *SessionNode) ToolsConn() toolsrv.Runner { - return s.Core.ToolsConn() -} - -// --- AgentLog management --- - -func (s *SessionNode) AgentLog() *AgentLog { - s.mu.RLock() - defer s.mu.RUnlock() - for _, al := range s.AgentLogs { - return al - } - return nil -} - -func (s *SessionNode) AgentLogFor(agentID string) *AgentLog { - s.mu.RLock() - defer s.mu.RUnlock() - return s.AgentLogs[agentID] -} - -func (s *SessionNode) SetAgentLog(agentID string, al *AgentLog) { - s.mu.Lock() - defer s.mu.Unlock() - if s.AgentLogs == nil { - s.AgentLogs = make(map[string]*AgentLog) - } - s.AgentLogs[agentID] = al -} - -// --- Models cache --- - -const modelsCacheTTL = 24 * time.Hour - -func (s *SessionNode) CachedListModels() string { - s.modelsMu.Lock() - if s.modelsCache != "" && time.Since(s.modelsCacheAt) < modelsCacheTTL { - result := s.modelsCache - s.modelsMu.Unlock() - return result - } - s.modelsMu.Unlock() - - agents := s.Core.Agents() - if len(agents) == 0 { - return "" - } - models := agents[0].ListModels() - result := strings.Join(models, "\n") - s.modelsMu.Lock() - s.modelsCache = result - s.modelsCacheAt = time.Now() - s.modelsMu.Unlock() - return result -} - -func (s *SessionNode) InvalidateModelsCache() { - s.modelsMu.Lock() - s.modelsCache = "" - s.modelsCacheAt = time.Time{} - s.modelsMu.Unlock() -} - // AgentLog holds per-agent 9P streaming state. type AgentLog struct { mu sync.RWMutex @@ -174,12 +52,12 @@ func (as *AgentLog) LogInfo() (length int, vers uint32) { return len(as.log), as.logVers } -func (as *AgentLog) Log() []byte { return as.log } -func (as *AgentLog) Plan() []byte { return as.plan } -func (as *AgentLog) SetPlan(p []byte) { as.plan = p } +func (as *AgentLog) Log() []byte { return as.log } +func (as *AgentLog) Plan() []byte { return as.plan } +func (as *AgentLog) SetPlan(p []byte) { as.plan = p } func (as *AgentLog) PrevPrompt() []byte { return as.prevPrompt } -func (as *AgentLog) Mu() *sync.RWMutex { return &as.mu } -func (as *AgentLog) Remote() string { return as.remote } +func (as *AgentLog) Mu() *sync.RWMutex { return &as.mu } +func (as *AgentLog) Remote() string { return as.remote } // NewEventHandler returns an event handler that writes to an AgentLog. func NewEventHandler(as *AgentLog) agent.EventHandler { diff --git a/fs/session_node.go b/fs/session_node.go new file mode 100644 index 0000000..809cc4a --- /dev/null +++ b/fs/session_node.go @@ -0,0 +1,129 @@ +package fs + +import ( + "context" + "strings" + "sync" + "time" + + "ollie/agent" + "ollie/session" + "ollie/toolsrv" +) + +// SessionNode wraps session.Session for 9P handlers. +// It adds AgentLogs (9P chat streaming state) and models cache. +type SessionNode struct { + Core *session.Session + + // AgentLogs for chat streaming, keyed by agent ID. + mu sync.RWMutex + AgentLogs map[string]*AgentLog + + // Cached model list (expensive API call; refreshed every 24h). + modelsMu sync.Mutex + modelsCache string + modelsCacheAt time.Time +} + +// NewSessionNode creates a SessionNode wrapping a session. +func NewSessionNode(core *session.Session) *SessionNode { + return &SessionNode{Core: core, AgentLogs: make(map[string]*AgentLog)} +} + +// --- Delegation to Core --- + +func (s *SessionNode) ID() string { return s.Core.ID() } +func (s *SessionNode) Name() string { return s.Core.Name() } +func (s *SessionNode) SetName(n string) { s.Core.SetName(n) } +func (s *SessionNode) Uname() string { return s.Core.Uname() } +func (s *SessionNode) Cancel() { s.Core.Cancel() } +func (s *SessionNode) IsPaused() bool { return s.Core.IsPaused() } +func (s *SessionNode) IsConnected() bool { return s.Core.IsConnected() } + +func (s *SessionNode) Pause() error { + err := s.Core.Pause() + if err == nil { + session.PublishEvent("session."+s.Core.ID()+".pause", "") + } + return err +} + +func (s *SessionNode) Resume() error { + err := s.Core.Resume() + if err == nil { + session.PublishEvent("session."+s.Core.ID()+".resume", "") + } + return err +} + +func (s *SessionNode) LoadTool(name string, ag *agent.Agent) error { + return s.Core.LoadTool(name, ag) +} + +func (s *SessionNode) Ctx() context.Context { + return s.Core.Ctx() +} + +func (s *SessionNode) ToolsConn() toolsrv.Runner { + return s.Core.ToolsConn() +} + +// --- AgentLog management --- + +func (s *SessionNode) AgentLog() *AgentLog { + s.mu.RLock() + defer s.mu.RUnlock() + for _, al := range s.AgentLogs { + return al + } + return nil +} + +func (s *SessionNode) AgentLogFor(agentID string) *AgentLog { + s.mu.RLock() + defer s.mu.RUnlock() + return s.AgentLogs[agentID] +} + +func (s *SessionNode) SetAgentLog(agentID string, al *AgentLog) { + s.mu.Lock() + defer s.mu.Unlock() + if s.AgentLogs == nil { + s.AgentLogs = make(map[string]*AgentLog) + } + s.AgentLogs[agentID] = al +} + +// --- Models cache --- + +const modelsCacheTTL = 24 * time.Hour + +func (s *SessionNode) CachedListModels() string { + s.modelsMu.Lock() + if s.modelsCache != "" && time.Since(s.modelsCacheAt) < modelsCacheTTL { + result := s.modelsCache + s.modelsMu.Unlock() + return result + } + s.modelsMu.Unlock() + + agents := s.Core.Agents() + if len(agents) == 0 { + return "" + } + models := agents[0].ListModels() + result := strings.Join(models, "\n") + s.modelsMu.Lock() + s.modelsCache = result + s.modelsCacheAt = time.Now() + s.modelsMu.Unlock() + return result +} + +func (s *SessionNode) InvalidateModelsCache() { + s.modelsMu.Lock() + s.modelsCache = "" + s.modelsCacheAt = time.Time{} + s.modelsMu.Unlock() +}