ollie/session/session.go

358 lines
7.7 KiB
Go

package session
import (
"context"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
"sync"
"time"
"ollie/agent"
olog "ollie/log"
"ollie/paths"
"ollie/toolsrv"
)
// Session owns all session-level state: identity, agents, tool server lifecycle.
type Session struct {
// Identity (immutable after creation)
ID string // UUID
Uname string // user principal (numeric UID)
// Context lifecycle
Ctx context.Context
Cancel context.CancelFunc
// Tool server lifecycle (set during setup/teardown, not raced)
Proc *toolsrv.Process
Keeper *toolsrv.ProcessKeeper
Remote string
// Agent state
agents []*agent.Agent
log *olog.Logger
// Mutable state protected by mu
mu sync.RWMutex
name string // friendly name
toolsConn *toolsrv.Conn // multiplexed RPC connection
paused bool
// Autosave
saveMu sync.Mutex
saveDirty bool
saveTimer *time.Timer
}
var sweepTmpOnce sync.Once
func ollieTmpDir() string {
return filepath.Join(paths.DataDir(), "tmp")
}
func sweepStaleTmpDirs() {
sweepTmpOnce.Do(func() {
base := ollieTmpDir()
os.RemoveAll(base) //nolint:errcheck
os.MkdirAll(base, 0700) //nolint:errcheck
})
}
// --- Mutable field accessors (only for fields that race) ---
// Name returns the mutable friendly name.
func (s *Session) Name() string {
s.mu.RLock()
defer s.mu.RUnlock()
return s.name
}
// SetName sets the friendly name.
func (s *Session) SetName(name string) {
s.mu.Lock()
defer s.mu.Unlock()
s.name = name
}
// ToolsConn returns the tool server connection.
func (s *Session) ToolsConn() *toolsrv.Conn {
s.mu.RLock()
defer s.mu.RUnlock()
return s.toolsConn
}
// SetToolsConn sets the tool server connection.
func (s *Session) SetToolsConn(conn *toolsrv.Conn) {
s.mu.Lock()
defer s.mu.Unlock()
s.toolsConn = conn
}
// IsPaused returns true if the session is paused.
func (s *Session) IsPaused() bool {
s.mu.RLock()
defer s.mu.RUnlock()
return s.paused
}
// FindAgent returns the agent matching the given name or ID, or nil.
func (s *Session) FindAgent(nameOrID string) *agent.Agent {
for _, ag := range s.agents {
if ag.Name() == nameOrID || ag.ID() == nameOrID {
return ag
}
}
return nil
}
// Agents returns the full agent slice.
func (s *Session) Agents() []*agent.Agent { return s.agents }
// AddAgent appends an agent to the session.
func (s *Session) AddAgent(ag *agent.Agent) {
s.agents = append(s.agents, ag)
ag.SetSessionEnv(s.ID)
}
// RemoveAgent removes an agent by ID and closes it.
func (s *Session) RemoveAgent(id string) bool {
idx := -1
for i, ag := range s.agents {
if ag.ID() == id {
idx = i
break
}
}
if idx < 0 {
return false
}
s.agents[idx].Close()
s.agents = append(s.agents[:idx], s.agents[idx+1:]...)
return true
}
// Close releases resources for this session.
func (s *Session) Close() {
s.log.Debug("Close() session=%q", s.ID)
s.flushSave()
for _, ag := range s.agents {
ag.Close()
}
if s.ID != "" {
os.RemoveAll(filepath.Join(ollieTmpDir(), s.ID)) //nolint:errcheck
}
}
// SetEnv stores a session-scoped variable and propagates to all agents.
func (s *Session) SetEnv(key, value string) {
for _, ag := range s.agents {
ag.SetEnv(key, value)
}
}
// SetSessionID renames the session.
func (s *Session) SetSessionID(newID string) error {
oldID := s.ID
if oldID == newID {
return nil
}
if pkgSessionsDir != "" && oldID != "" {
for _, suffix := range []string{".json", ".compaction.jsonl"} {
oldPath := s.activeSessionPath(oldID, suffix)
if _, err := os.Stat(oldPath); err == nil {
if err := os.Rename(oldPath, s.activeSessionPath(newID, suffix)); err != nil {
return fmt.Errorf("rename %s: %w", suffix, err)
}
}
}
}
s.ID = newID
oldTemp := filepath.Join(ollieTmpDir(), oldID)
newTemp := filepath.Join(ollieTmpDir(), newID)
if _, err := os.Stat(oldTemp); err == nil {
os.Rename(oldTemp, newTemp) //nolint:errcheck
}
for _, ag := range s.agents {
ag.SetSessionEnv(newID)
}
return nil
}
// IsConnected returns true if the tool server connection is alive.
func (s *Session) IsConnected() bool {
s.mu.RLock()
paused := s.paused
conn := s.toolsConn
s.mu.RUnlock()
if paused || conn == nil {
return false
}
return conn.Ping() == nil
}
// Pause stops the tool server to save resources.
func (s *Session) Pause() error {
s.mu.Lock()
defer s.mu.Unlock()
if s.paused {
return fmt.Errorf("session already paused")
}
if s.Cancel != nil {
s.Cancel()
}
if s.Keeper != nil {
s.Keeper.Close()
} else if s.Proc != nil {
s.Proc.Close()
}
if s.toolsConn != nil {
s.toolsConn.Close()
s.toolsConn = nil
}
s.paused = true
go s.saveSession()
return nil
}
// Resume restarts the tool server after a Pause.
func (s *Session) Resume() error {
s.mu.Lock()
defer s.mu.Unlock()
if !s.paused {
return fmt.Errorf("session not paused")
}
ctx, cancel := context.WithCancel(serverCtx)
s.Ctx = ctx
s.Cancel = cancel
if s.Keeper != nil {
s.Keeper.SetContext(ctx)
conn, err := s.Keeper.Dial()
if err != nil {
return fmt.Errorf("resume failed: %w", err)
}
s.toolsConn = conn
} else {
// Session was restored paused without infra — set up per-agent.
for i, ag := range s.agents {
cwd := ag.Cwd()
if cwd == "" {
cwd, _ = os.Getwd()
}
var reuseFrom *InfraConfig
if s.Proc != nil {
reuseFrom = &InfraConfig{
Proc: s.Proc,
Keeper: s.Keeper,
ToolsConn: s.toolsConn,
}
}
infra, err := SetupToolServer(ToolServerConfig{
Ctx: ctx,
CWD: cwd,
RemoteTarget: s.Remote,
Yolo: pkgYolo,
ReuseFrom: reuseFrom,
})
if err != nil {
return fmt.Errorf("resume setup for agent %s failed: %w", ag.ID(), err)
}
if i == 0 {
s.Proc = infra.Proc
s.Keeper = infra.Keeper
s.toolsConn = infra.ToolsConn
}
ag.SetToolServer(infra.NewToolServer, infra.ToolsConn)
}
}
s.paused = false
go s.saveSession()
return nil
}
// --- Tool loading ---
// LoadTool loads a tool into the tool server.
func (s *Session) LoadTool(name string, ag *agent.Agent) error {
name = strings.TrimSpace(name)
if name == "" {
return nil
}
return LoadToolOnConn(s.ToolsConn(), name)
}
// LoadToolOnConn loads a tool on the given connection.
func LoadToolOnConn(conn *toolsrv.Conn, name string) error {
if conn == nil {
return nil
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
args, _ := json.Marshal(map[string]string{"name": name})
if _, err := conn.CallTool(ctx, "tool_load", json.RawMessage(args)); err != nil {
return fmt.Errorf("remote load: %w", err)
}
return nil
}
// --- Autosave ---
func (s *Session) activeSessionPath(id, suffix string) string {
return filepath.Join(pkgSessionsDir, "active", id+suffix)
}
func (s *Session) saveSession() {
s.saveMu.Lock()
s.saveDirty = true
if s.saveTimer == nil {
s.saveTimer = time.AfterFunc(2*time.Second, s.flushSave)
}
s.saveMu.Unlock()
}
func (s *Session) flushSave() {
s.saveMu.Lock()
dirty := s.saveDirty
s.saveDirty = false
if s.saveTimer != nil {
s.saveTimer.Stop()
s.saveTimer = nil
}
s.saveMu.Unlock()
if !dirty || s.ID == "" || pkgSessionsDir == "" {
return
}
if err := PersistSession(s.Name()); err != nil {
s.log.Error("session save: %v", err)
}
}
// Save writes the current session state via PersistSession.
func (s *Session) Save() error {
return PersistSession(s.Name())
}
// --- Constructor for empty session ---
// NewEmpty creates an empty session without an agent.
func NewEmpty(id string, ctx context.Context, cancel context.CancelFunc) *Session {
name := id
if idx := strings.IndexByte(id, '-'); idx > 0 {
name = id[:idx]
}
log := pkgSink.NewLogger("session")
return &Session{
ID: id,
name: name,
Ctx: ctx,
Cancel: cancel,
agents: make([]*agent.Agent, 0),
log: log,
}
}