diff --git a/main.go b/main.go index e1f80cf..7a86f53 100644 --- a/main.go +++ b/main.go @@ -7,6 +7,7 @@ import ( "hash/crc32" "os" "os/signal" + "path/filepath" "strings" "sync" "syscall" @@ -524,6 +525,226 @@ func (m *SessionManager) sendNotification(title, body string) { ) } +// --- Session Persistence --- + +// activeSessionsDir returns the directory for persisting active sessions across restarts. +func activeSessionsDir() string { + return filepath.Join(paths.DataDir(), "active-sessions") +} + +// saveAllSessions persists every active session to disk so they can be restored on restart. +func (m *SessionManager) saveAllSessions() { + dir := activeSessionsDir() + os.MkdirAll(dir, 0700) + + // Remove stale files from previous run + entries, _ := os.ReadDir(dir) + for _, e := range entries { + os.Remove(filepath.Join(dir, e.Name())) + } + + m.mu.RLock() + defer m.mu.RUnlock() + + for id, sess := range m.sessions { + // Skip copilot sessions — they are ephemeral + if strings.HasSuffix(id, "-copilot") { + continue + } + path := filepath.Join(dir, id+".json") + if err := sess.core.SaveSession(path); err != nil { + fmt.Fprintf(os.Stderr, "persist session %s: %v\n", id, err) + } + } +} + +// restoreAllSessions loads previously-persisted sessions and recreates them. +func (m *SessionManager) restoreAllSessions() { + dir := activeSessionsDir() + entries, err := os.ReadDir(dir) + if err != nil { + return // no saved sessions + } + + for _, e := range entries { + if !strings.HasSuffix(e.Name(), ".json") { + continue + } + path := filepath.Join(dir, e.Name()) + ps, err := agent.LoadPersistedSession(path) + if err != nil { + fmt.Fprintf(os.Stderr, "restore session %s: %v\n", e.Name(), err) + continue + } + if err := m.restoreSession(ps); err != nil { + fmt.Fprintf(os.Stderr, "restore session %s: %v\n", ps.ID, err) + } + } + + // Clean up persisted files after successful restore + for _, e := range entries { + os.Remove(filepath.Join(dir, e.Name())) + } +} + +// restoreSession recreates a single session from persisted state. +func (m *SessionManager) restoreSession(ps *agent.PersistedSession) error { + cwd := ps.CWD + if cwd == "" { + cwd, _ = os.Getwd() + } + + agentName := ps.Agent + if agentName == "" { + agentName = "default" + } + + sessID := ps.ID + + // Load agent config + agentsDir := paths.CfgDir() + "/agents" + var cfg *config.Config + cfgPath := agent.AgentConfigPath(agentsDir, agentName) + if f, err := os.Open(cfgPath); err == nil { + cfg, _ = config.Load(f) + f.Close() + } + + // Resolve backend + backendName := ps.Backend + if backendName == "" && cfg != nil && cfg.Backend != "" { + backendName = cfg.Backend + } + be, err := backend.NewWithName(backendName) + if err != nil { + return fmt.Errorf("backend: %w", err) + } + + // Resolve model + modelName := ps.Model + if modelName == "" && cfg != nil && cfg.Model != "" { + modelName = cfg.Model + } + if modelName == "" { + modelName = os.Getenv("OLLIE_MODEL") + } + if modelName != "" { + be.SetModel(modelName) + } + + // Sessions dir + sessionsDir := paths.DataDir() + "/sessions" + os.MkdirAll(sessionsDir, 0700) + + // Build dispatcher + runtime + newDisp := tools.NewDispatcherFunc(map[string]func() tools.Server{ + "execute": execute.Decl(cwd), + }) + rt := agent.BuildRuntime(cfg, newDisp(), cwd, []string{"OLLIE_SESSION_ID=" + sessID}) + + // Restore session from persisted messages + restoredSession := agent.RestoreSession(ps.Messages) + if ps.TaskState != nil { + restoredSession.TaskState = ps.TaskState + } + + // Create the agent core with restored session + core := agent.NewAgentCore(agent.AgentCoreConfig{ + Backend: be, + AgentName: agentName, + AgentsDir: agentsDir, + SessionsDir: sessionsDir, + SessionID: sessID, + CWD: cwd, + Session: restoredSession, + Runtime: rt, + NewDispatcher: newDisp, + }) + + ctx, cancel := context.WithCancel(context.Background()) + sess := &managedSession{ + core: core, + ctx: ctx, + cancel: cancel, + id: sessID, + agent: agentName, + } + + // Subscribe to events for the chat log + signal dispatch + streamingRole := "" + core.Bus().Subscribe("event", func(ev agent.Event) { + var text string + switch ev.Role { + case "assistant", "reasoning": + if streamingRole != ev.Role { + if streamingRole != "" { + text += "\n\n" + } + text += "[" + ev.Role + "]\n" + streamingRole = ev.Role + } + text += ev.Content + default: + if ev.Role == "usage" { + return + } + if streamingRole != "" { + text += "\n\n" + streamingRole = "" + } + switch ev.Role { + case "user": + text += "[user]\n" + ev.Content + "\n\n" + case "call": + text += "[call:" + ev.Name + "]\n" + ev.Content + "\n\n" + case "tool": + text += "[tool:" + ev.Name + "]\n" + ev.Content + "\n\n" + case "error": + text += "[error]\n" + ev.Content + "\n\n" + case "info": + text += "[info] " + ev.Content + "\n\n" + default: + text += "[" + ev.Role + "]\n" + ev.Content + "\n\n" + } + } + if text == "" { + return + } + + sess.logMu.Lock() + offset := int64(len(sess.log)) + sess.log = append(sess.log, []byte(text)...) + sess.logMu.Unlock() + + m.conn.Emit(busPath, busIface+".ChatUpdated", sess.id, offset, text) + }) + + // Watch state changes + go func() { + current := "idle" + for { + next, ok := core.WaitChange(ctx, agent.WatchState, current) + if !ok { + return + } + current = next + m.conn.Emit(busPath, busIface+".StateChanged", sess.id, current) + + if current == "idle" { + m.sendNotification("Ollie", "Session "+sess.id+" finished") + } + } + }() + + m.mu.Lock() + m.sessions[sessID] = sess + m.mu.Unlock() + + m.conn.Emit(busPath, busIface+".SessionCreated", sessID) + fmt.Printf(" restored session: %s [%s/%s]\n", sessID, agentName, modelName) + return nil +} + // --- Shutdown --- // --- Completion --- @@ -727,6 +948,8 @@ func crc32Str(s string) uint32 { } func (m *SessionManager) Shutdown() { + m.saveAllSessions() + m.mu.Lock() for id, sess := range m.sessions { sess.core.Interrupt(agent.ErrInterrupted) @@ -877,6 +1100,7 @@ func main() { } mgr := NewSessionManager(conn) + mgr.restoreAllSessions() conn.Export(mgr, busPath, busIface) conn.Export(introspect.Introspectable(introspectXML), busPath,