diff --git a/9p b/9p deleted file mode 100755 index 14881ab..0000000 Binary files a/9p and /dev/null differ diff --git a/dbus.go b/dbus.go index 23dd764..c7b0b46 100644 --- a/dbus.go +++ b/dbus.go @@ -15,7 +15,8 @@ import ( "ollie/agent" "ollie/backend" - "olliesrv/mgr" + "olliesrv/fs" + "olliesrv/session" ) const ( @@ -24,10 +25,10 @@ const ( busIface = "org.ollie.SessionManager" ) -// DBusAdapter exposes a mgr.Manager over D-Bus. +// DBusAdapter exposes sessions over D-Bus. type DBusAdapter struct { conn *dbus.Conn - mgr *mgr.Manager + tree *fs.Tree mu sync.RWMutex watchers map[string]context.CancelFunc // sessionID -> cancel for state watcher @@ -35,7 +36,7 @@ type DBusAdapter struct { // startDBus connects to the session bus, claims the well-known name, and // exports the adapter. Returns nil (no-op) if the bus is unavailable. -func startDBus(mgr *mgr.Manager) *DBusAdapter { +func startDBus(tree *fs.Tree) *DBusAdapter { conn, err := dbus.ConnectSessionBus() if err != nil { fmt.Fprintf(os.Stderr, "dbus: session bus unavailable: %v\n", err) @@ -51,7 +52,7 @@ func startDBus(mgr *mgr.Manager) *DBusAdapter { a := &DBusAdapter{ conn: conn, - mgr: mgr, + tree: tree, watchers: make(map[string]context.CancelFunc), } @@ -68,16 +69,8 @@ func (a *DBusAdapter) WatchExisting() { if a == nil { return } - entries, _ := a.mgr.List() - for _, e := range entries { - if !e.IsDir() { - continue - } - id := e.Name() - sess := a.mgr.Session(id) - if sess == nil { - continue - } + for _, sess := range session.All(a.tree) { + id := sess.RunnableID() a.startWatcher(id, sess) // Emit SessionCreated so GUI clients that connected before restore // (via serviceRegistered) learn about pre-existing sessions. @@ -97,9 +90,9 @@ func (a *DBusAdapter) Close() { a.conn.Close() } -// --- Lifecycle callbacks (called by mgr.Manager) --- +// --- Lifecycle callbacks --- -func (a *DBusAdapter) OnSessionCreated(id string, sess *mgr.Session) { +func (a *DBusAdapter) OnSessionCreated(id string, sess *session.Session) { if a == nil { return } @@ -121,7 +114,7 @@ func (a *DBusAdapter) OnSessionRenamed(oldID, newID string) { } a.stopWatcher(oldID) // The session object is the same, just re-lookup - sess := a.mgr.Session(newID) + sess := session.Lookup(a.tree, newID) if sess != nil { a.startWatcher(newID, sess) } @@ -129,7 +122,7 @@ func (a *DBusAdapter) OnSessionRenamed(oldID, newID string) { } // startWatcher launches goroutines for StateChanged and ChatUpdated signals. -func (a *DBusAdapter) startWatcher(id string, sess *mgr.Session) { +func (a *DBusAdapter) startWatcher(id string, sess *session.Session) { ctx, cancel := context.WithCancel(context.Background()) a.mu.Lock() a.watchers[id] = cancel @@ -255,7 +248,7 @@ func (a *DBusAdapter) CreateSession(cwd, backendName, modelName, agentName, syst dir, _ := os.Getwd() args = append(args, "cwd="+dir) } - id, err := a.mgr.CreateSession(args) + id, err := session.CreateFromRoot(a.tree, args) if err != nil { return "", dbus.MakeFailedError(err) } @@ -263,17 +256,9 @@ func (a *DBusAdapter) CreateSession(cwd, backendName, modelName, agentName, syst } func (a *DBusAdapter) ListSessions() ([]string, *dbus.Error) { - entries, _ := a.mgr.List() var result []string - for _, e := range entries { - if !e.IsDir() { - continue - } - id := e.Name() - sess := a.mgr.Session(id) - if sess == nil { - continue - } + for _, sess := range session.All(a.tree) { + id := sess.RunnableID() result = append(result, fmt.Sprintf("%s\t%s\t%s\t%s", id, sess.Core.Agent().State(), sess.Core.Agent().ModelName(), sess.Core.Agent().Name())) } @@ -281,23 +266,23 @@ func (a *DBusAdapter) ListSessions() ([]string, *dbus.Error) { } func (a *DBusAdapter) KillSession(sessionID string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } - a.mgr.KillSession(sessionID) + session.KillFromRoot(a.tree, sessionID) return true, nil } func (a *DBusAdapter) RenameSession(sessionID, newName string) (bool, *dbus.Error) { - if err := a.mgr.RenameSession(sessionID, newName); err != nil { + if err := session.RenameFromRoot(a.tree, sessionID, newName); err != nil { return false, dbus.MakeFailedError(err) } return true, nil } func (a *DBusAdapter) Submit(sessionID, prompt string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } @@ -310,7 +295,7 @@ func (a *DBusAdapter) React(sessionID, emoji string) (bool, *dbus.Error) { } func (a *DBusAdapter) ReactTo(sessionID, responseID, emoji string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } @@ -321,7 +306,7 @@ func (a *DBusAdapter) ReactTo(sessionID, responseID, emoji string) (bool, *dbus. } func (a *DBusAdapter) Interrupt(sessionID string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } @@ -330,7 +315,7 @@ func (a *DBusAdapter) Interrupt(sessionID string) (bool, *dbus.Error) { } func (a *DBusAdapter) GetState(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -338,7 +323,7 @@ func (a *DBusAdapter) GetState(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) GetChat(sessionID string, offset int64) (string, int64, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", offset, nil } @@ -357,7 +342,7 @@ func (a *DBusAdapter) GetChat(sessionID string, offset int64) (string, int64, *d } func (a *DBusAdapter) GetUsage(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -365,7 +350,7 @@ func (a *DBusAdapter) GetUsage(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) GetCost(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -373,7 +358,7 @@ func (a *DBusAdapter) GetCost(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) GetConfig(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -387,7 +372,7 @@ func (a *DBusAdapter) GetConfig(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) SetConfig(sessionID, key, value string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } @@ -405,7 +390,7 @@ func (a *DBusAdapter) SetConfig(sessionID, key, value string) (bool, *dbus.Error } func (a *DBusAdapter) GetReactions(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "{}", nil } @@ -417,7 +402,7 @@ func (a *DBusAdapter) GetReactions(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) GetContext(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -446,7 +431,7 @@ func (a *DBusAdapter) ListModelsByBackend(backendName string) ([]string, *dbus.E } func (a *DBusAdapter) ListModels(sessionID string) ([]string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return nil, nil } @@ -480,7 +465,7 @@ func (a *DBusAdapter) ListAgents() ([]string, *dbus.Error) { // --- Detached processes --- func (a *DBusAdapter) DetachProcess(sessionID string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } @@ -492,7 +477,7 @@ func (a *DBusAdapter) DetachProcess(sessionID string) (bool, *dbus.Error) { } func (a *DBusAdapter) ListDetached(sessionID string) ([]string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return nil, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } @@ -509,7 +494,7 @@ func (a *DBusAdapter) ListDetached(sessionID string) ([]string, *dbus.Error) { } func (a *DBusAdapter) SignalDetached(sessionID string, pid int32, signal string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } @@ -529,7 +514,7 @@ func (a *DBusAdapter) SignalDetached(sessionID string, pid int32, signal string) } func (a *DBusAdapter) GetDetachedOutput(sessionID string, pid int32) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } @@ -541,7 +526,7 @@ func (a *DBusAdapter) GetDetachedOutput(sessionID string, pid int32) (string, *d } func (a *DBusAdapter) DismissDetached(sessionID string, pid int32) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, dbus.MakeFailedError(fmt.Errorf("session not found: %s", sessionID)) } @@ -583,7 +568,7 @@ func (a *DBusAdapter) Complete(cwd, filePath, prefix, suffix, extraContext strin } func (a *DBusAdapter) GetEnv(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -599,7 +584,7 @@ func (a *DBusAdapter) GetEnv(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) GetPlan(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -610,7 +595,7 @@ func (a *DBusAdapter) GetPlan(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) SetPlan(sessionID, plan string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } @@ -621,7 +606,7 @@ func (a *DBusAdapter) SetPlan(sessionID, plan string) (bool, *dbus.Error) { } func (a *DBusAdapter) GetSystemPrompt(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -629,7 +614,7 @@ func (a *DBusAdapter) GetSystemPrompt(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) GetContextSize(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -637,7 +622,7 @@ func (a *DBusAdapter) GetContextSize(sessionID string) (string, *dbus.Error) { } func (a *DBusAdapter) GetPreviousPrompt(sessionID string) (string, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return "", nil } @@ -648,7 +633,7 @@ func (a *DBusAdapter) GetPreviousPrompt(sessionID string) (string, *dbus.Error) } func (a *DBusAdapter) Compact(sessionID string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } @@ -657,7 +642,7 @@ func (a *DBusAdapter) Compact(sessionID string) (bool, *dbus.Error) { } func (a *DBusAdapter) ClearContext(sessionID string) (bool, *dbus.Error) { - sess := a.mgr.Session(sessionID) + sess := session.Lookup(a.tree, sessionID) if sess == nil { return false, nil } diff --git a/fs/tree.go b/fs/tree.go index 2671c46..4a928ca 100644 --- a/fs/tree.go +++ b/fs/tree.go @@ -111,6 +111,11 @@ func (d *Tree) Child(name string) *Tree { return d.children[name] } + +// Children returns all mounted children. +func (d *Tree) Children() map[string]*Tree { + return d.children +} // split checks if name routes to a child tree. Returns (child, rest, true) // if a child handles this path, or (nil, "", false) otherwise. func (d *Tree) split(name string) (*Tree, string, bool) { diff --git a/main.go b/main.go index 7401227..4a4ed85 100644 --- a/main.go +++ b/main.go @@ -25,7 +25,7 @@ import ( fs "olliesrv/fs" "olliesrv/mount" "olliesrv/server" - "olliesrv/mgr" + "olliesrv/session" ) const serviceName = "ollie" @@ -181,7 +181,7 @@ func runServer(sockPath string) { daemonCtx, daemonCancel := context.WithCancel(context.Background()) defer daemonCancel() - mgr := mgr.NewManager(mgr.ManagerConfig{ + sessionTree := session.NewRoot(session.Config{ Ctx: daemonCtx, ToolRegistry: toolRegistry, SkillsRegistry: skillsRegistry, @@ -200,7 +200,7 @@ func runServer(sockPath string) { elevateBroker.ResetTurn(sessionID) } }, - OnSessionCreated: func(id string, sess *mgr.Session) { + OnSessionCreated: func(id string, sess *session.Session) { if dbusAdapter != nil { dbusAdapter.OnSessionCreated(id, sess) } @@ -224,7 +224,7 @@ func runServer(sockPath string) { // -tcp implies -nodbus (remote/headless deployments have no session bus). // -no9p forces D-Bus on (it's the only interface). if !*nodbus && (*no9p || *tcpAddr == "") { - dbusAdapter = startDBus(mgr) + dbusAdapter = startDBus(sessionTree) if dbusAdapter != nil { dbusAdapter.WatchExisting() } @@ -257,7 +257,7 @@ func runServer(sockPath string) { PolicyPath: policyPath, Notify: notifyFn, Logf: func(f string, a ...any) { fmt.Printf(f+"\n", a...) }, - SessionValid: func(id string) bool { return mgr.Session(id) != nil }, + SessionValid: func(id string) bool { return session.Lookup(sessionTree, id) != nil }, }) if err != nil { fmt.Fprintf(os.Stderr, "warning: %v\n", err) @@ -274,7 +274,7 @@ func runServer(sockPath string) { if !*no9p { srv = server.New(server.Config{ Sink: sink, - SessionMgr: mgr, + SessionTree: sessionTree, RootStore: NewRootStore(toolRegistry), ElevateBroker: elevateBroker, InvalidateModels: modelCache.Invalidate, diff --git a/main_test.go b/main_test.go index bfaa431..b4fe979 100644 --- a/main_test.go +++ b/main_test.go @@ -11,23 +11,23 @@ import ( "testing" "time" - "ollie/session" + coresession "ollie/session" "ollie/agent" "ollie/toolsrv" "ollie/backend" olog "ollie/log" "olliesrv/fs" - "olliesrv/mgr" + "olliesrv/session" ) // --- test helpers --- func testSink() *olog.Sink { return olog.NewSink(io.Discard, io.Discard, olog.LevelError) } -// newNoopCore creates a real *session.Session with a noop backend. -func newNoopCore(id string) *session.Session { +// newNoopCore creates a real *coresession.Session with a noop backend. +func newNoopCore(id string) *coresession.Session { be := backend.NewNoop("stub", "m") - return session.New(session.Config{ + return coresession.New(coresession.Config{ Backend: be, AgentName: "default", CWD: "/tmp", @@ -39,8 +39,8 @@ func newNoopCore(id string) *session.Session { }) } -// newBlockingCore creates a real *session.Session whose backend blocks until ctx is cancelled. -func newBlockingCore(id string) *session.Session { +// newBlockingCore creates a real *coresession.Session whose backend blocks until ctx is cancelled. +func newBlockingCore(id string) *coresession.Session { be := backend.NewNoop("stub", "m") be.ChatStreamFunc = func(ctx context.Context, _ []backend.Message, _ []backend.Tool, _ backend.GenerationParams) (<-chan backend.StreamEvent, error) { ch := make(chan backend.StreamEvent, 1) @@ -51,7 +51,7 @@ func newBlockingCore(id string) *session.Session { }() return ch, nil } - return session.New(session.Config{ + return coresession.New(coresession.Config{ Backend: be, AgentName: "default", CWD: "/tmp", @@ -63,8 +63,8 @@ func newBlockingCore(id string) *session.Session { }) } -// newContentCore creates a real *session.Session whose backend emits specific content. -func newContentCore(id, content string) *session.Session { +// newContentCore creates a real *coresession.Session whose backend emits specific content. +func newContentCore(id, content string) *coresession.Session { be := backend.NewNoop("stub", "m") be.ChatStreamFunc = func(ctx context.Context, _ []backend.Message, _ []backend.Tool, _ backend.GenerationParams) (<-chan backend.StreamEvent, error) { ch := make(chan backend.StreamEvent, 2) @@ -73,7 +73,7 @@ func newContentCore(id, content string) *session.Session { close(ch) return ch, nil } - return session.New(session.Config{ + return coresession.New(coresession.Config{ Backend: be, AgentName: "default", CWD: "/tmp", @@ -85,22 +85,22 @@ func newContentCore(id, content string) *session.Session { }) } -func testSession(id string) *mgr.Session { +func testSession(id string) *session.Session { ctx, cancel := context.WithCancel(context.Background()) core := newNoopCore(id) - return mgr.NewSession(id, core, ctx, cancel) + return session.NewSession(id, core, ctx, cancel) } -func testBlockingSession(id string) *mgr.Session { +func testBlockingSession(id string) *session.Session { ctx, cancel := context.WithCancel(context.Background()) core := newBlockingCore(id) - return mgr.NewSession(id, core, ctx, cancel) + return session.NewSession(id, core, ctx, cancel) } -func newTestSessionManager(t *testing.T) *mgr.Manager { +func newTestSessionManager(t *testing.T) *fs.Tree { t.Helper() sink := testSink() - return mgr.NewManager(mgr.ManagerConfig{ + return session.NewRoot(session.Config{ Ctx: context.Background(), Log: sink.NewLogger("test"), Sink: sink, @@ -109,18 +109,18 @@ func newTestSessionManager(t *testing.T) *mgr.Manager { }) } -func newTestSessionManagerWithCore(t *testing.T) *mgr.Manager { +func newTestSessionManagerWithCore(t *testing.T) *fs.Tree { t.Helper() sink := testSink() - return mgr.NewManager(mgr.ManagerConfig{ + return session.NewRoot(session.Config{ Ctx: context.Background(), Log: sink.NewLogger("test"), Sink: sink, ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil }, MkdirAll: func(string, os.FileMode) error { return nil }, - NewCore: func(sessionID, agentName, cwd string) (*session.Session, error) { + NewCore: func(sessionID, agentName, cwd string) (*coresession.Session, error) { be := backend.NewNoop("stub", "m") - return session.New(session.Config{ + return coresession.New(coresession.Config{ Backend: be, AgentName: agentName, CWD: cwd, @@ -165,11 +165,11 @@ var _ = os.Remove // suppress unused import var _ = filepath.Join // suppress unused import var _ sync.Mutex // suppress unused import -func newTestSessionFileStore(t *testing.T, sess *mgr.Session) *fs.Tree { +func newTestSessionFileStore(t *testing.T, sess *session.Session) *fs.Tree { t.Helper() sink := testSink() // Use agent tree for agent-level file tests - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func(data []byte) error { return nil }, func() {}, nil, @@ -178,14 +178,14 @@ func newTestSessionFileStore(t *testing.T, sess *mgr.Session) *fs.Tree { return sf } -func newTestSessionFileStoreWith(t *testing.T, sess *mgr.Session, kill func(), rename func(string) error, save func([]byte) error) *fs.Tree { +func newTestSessionFileStoreWith(t *testing.T, sess *session.Session, kill func(), rename func(string) error, save func([]byte) error) *fs.Tree { t.Helper() sink := testSink() // Session tree for kill/rename/save operations - return mgr.NewSessionTree(sess, sink.NewLogger("test"), kill, rename, save, func() {}, nil, nil) + return session.NewSessionTree(sess, sink.NewLogger("test"), kill, rename, save, func() {}, nil, nil) } -// ===== mgr.Session ===== +// ===== session.Session ===== func TestSessionAppendLog(t *testing.T) { sess := testSession("s1") @@ -205,15 +205,15 @@ func TestSessionAppendLog(t *testing.T) { } func TestSessionManagerFileMode(t *testing.T) { - if m, ok := mgr.FileMode("new"); !ok || m != 0666 { - t.Errorf("mgr.FileMode(new) = %o, %v", m, ok) + if m, ok := session.FileMode("new"); !ok || m != 0666 { + t.Errorf("session.FileMode(new) = %o, %v", m, ok) } - if _, ok := mgr.FileMode("bogus"); ok { - t.Error("mgr.FileMode(bogus) should be false") + if _, ok := session.FileMode("bogus"); ok { + t.Error("session.FileMode(bogus) should be false") } } -// ===== mgr.Manager ===== +// ===== session tree ===== func TestSessionManagerReadableContract(t *testing.T) { } @@ -222,9 +222,9 @@ func TestSessionManagerGetIdx(t *testing.T) { s := newTestSessionManager(t) sess := testSession("abc") defer sess.Cancel() - s.AddSession(sess) + session.AddSession(s, sess) - data := testStoreRead(t, s.Tree(), "idx") + data := testStoreRead(t, s, "idx") if !strings.Contains(string(data), "abc") { t.Errorf("idx = %q; want to contain abc", data) } @@ -232,7 +232,7 @@ func TestSessionManagerGetIdx(t *testing.T) { func TestSessionManagerGetScript(t *testing.T) { s := newTestSessionManager(t) - data := testStoreRead(t, s.Tree(), "ls") + data := testStoreRead(t, s, "ls") if string(data) != "#!/bin/sh\n" { t.Errorf("Read(ls) = %q", data) } @@ -242,9 +242,9 @@ func TestSessionManagerStatSession(t *testing.T) { s := newTestSessionManager(t) sess := testSession("s1") defer sess.Cancel() - s.AddSession(sess) + session.AddSession(s, sess) - fi, err := s.Tree().Stat("s1") + fi, err := s.Stat("s1") if err != nil { t.Fatalf("Stat(s1): %v", err) } @@ -257,9 +257,9 @@ func TestSessionManagerListIncludesSessions(t *testing.T) { s := newTestSessionManager(t) sess := testSession("s1") defer sess.Cancel() - s.AddSession(sess) + session.AddSession(s, sess) - entries, _ := s.Tree().List() + entries, _ := s.List() found := false for _, e := range entries { if e.Name() == "s1" { @@ -273,7 +273,7 @@ func TestSessionManagerListIncludesSessions(t *testing.T) { func TestSessionManagerWriteNotWritable(t *testing.T) { s := newTestSessionManager(t) - e, err := s.Tree().Open("idx") + e, err := s.Open("idx") if err != nil { t.Fatalf("Open: %v", err) } @@ -284,7 +284,7 @@ func TestSessionManagerWriteNotWritable(t *testing.T) { func TestSessionManagerCreateErrors(t *testing.T) { s := newTestSessionManager(t) - if err := s.Tree().Create("x"); err == nil { + if err := s.Create("x"); err == nil { t.Error("Create should always error") } } @@ -292,33 +292,33 @@ func TestSessionManagerCreateErrors(t *testing.T) { func TestSessionManagerDeleteAndKill(t *testing.T) { s := newTestSessionManager(t) sess := testSession("s1") - s.AddSession(sess) + session.AddSession(s, sess) - if err := s.Tree().Delete("s1"); err != nil { + if err := s.Delete("s1"); err != nil { t.Fatalf("Delete: %v", err) } - if s.Session("s1") != nil { + if session.Lookup(s, "s1") != nil { t.Error("session should be gone after Delete") } // Verify core was closed by checking ctx is cancelled if sess.SessionCtx.Err() == nil { t.Error("session ctx should be cancelled after Delete") } - if err := s.Tree().Delete("nope"); err == nil { + if err := s.Delete("nope"); err == nil { t.Error("Delete(nonexistent) should error") } } func TestSessionManagerSession(t *testing.T) { s := newTestSessionManager(t) - if s.Session("nope") != nil { - t.Error("mgr.Session(nonexistent) should be nil") + if session.Lookup(s, "nope") != nil { + t.Error("session.FileMode(nonexistent) should be nil") } sess := testSession("s1") defer sess.Cancel() - s.AddSession(sess) - if s.Session("s1") == nil { - t.Error("mgr.Session(s1) should not be nil") + session.AddSession(s, sess) + if session.Lookup(s, "s1") == nil { + t.Error("session.FileMode(s1) should not be nil") } } @@ -330,25 +330,25 @@ func TestSessionManagerInterruptAll(t *testing.T) { // Submit in background to make it "running" go sess.Core.Agent().Submit(sess.SessionCtx, "hello") time.Sleep(50 * time.Millisecond) // let it enter thinking state - s.AddSession(sess) + session.AddSession(s, sess) - s.InterruptAll() // should not panic + session.InterruptAll(s) // should not panic } func TestSessionManagerShutdown(t *testing.T) { s := newTestSessionManager(t) sess := testSession("s1") - s.AddSession(sess) + session.AddSession(s, sess) - s.Shutdown() - if s.Session("s1") != nil { + session.Shutdown(s) + if session.Lookup(s, "s1") != nil { t.Error("session should be gone after Shutdown") } } func TestSessionManagerRename(t *testing.T) { sink := testSink() - s := mgr.NewManager(mgr.ManagerConfig{ + s := session.NewRoot(session.Config{ Ctx: context.Background(), Log: sink.NewLogger("test"), Sink: sink, @@ -357,15 +357,15 @@ func TestSessionManagerRename(t *testing.T) { }) sess := testSession("old") defer sess.Cancel() - s.AddSession(sess) + session.AddSession(s, sess) - if err := s.Tree().Rename("old", "new"); err != nil { + if err := s.Rename("old", "new"); err != nil { t.Fatalf("Rename: %v", err) } - if s.Session("old") != nil { + if session.Lookup(s, "old") != nil { t.Error("old session should be gone") } - if s.Session("new") == nil { + if session.Lookup(s, "new") == nil { t.Error("new session should exist") } } @@ -373,21 +373,21 @@ func TestSessionManagerRename(t *testing.T) { func TestSessionManagerRenameErrors(t *testing.T) { s := newTestSessionManager(t) // nonexistent - if err := s.Tree().Rename("nope", "x"); err == nil { + if err := s.Rename("nope", "x"); err == nil { t.Error("Rename(nonexistent) should error") } // duplicate - s.AddSession(testSession("a")) - s.AddSession(testSession("b")) - if err := s.Tree().Rename("a", "b"); err == nil { + session.AddSession(s, testSession("a")) + session.AddSession(s, testSession("b")) + if err := s.Rename("a", "b"); err == nil { t.Error("Rename to existing should error") } // running sess := testBlockingSession("r") go sess.Core.Agent().Submit(sess.SessionCtx, "hello") time.Sleep(50 * time.Millisecond) - s.AddSession(sess) - if err := s.Tree().Rename("r", "r2"); err == nil { + session.AddSession(s, sess) + if err := s.Rename("r", "r2"); err == nil { t.Error("Rename while running should error") } sess.Cancel() // cleanup @@ -399,7 +399,7 @@ func TestSessionFileStoreReadableContract(t *testing.T) { sess := testSession("s1") defer sess.Cancel() sink := testSink() - _ = mgr.NewAgentTree(sess, sink.NewLogger("test"), + _ = session.NewAgentTree(sess, sink.NewLogger("test"), func([]byte) error { return nil }, func() {}, nil, nil) } @@ -407,15 +407,15 @@ func TestSessionFileStoreList(t *testing.T) { sess := testSession("s1") defer sess.Cancel() sink := testSink() - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func([]byte) error { return nil }, func() {}, nil, nil) entries, err := sf.List() if err != nil { t.Fatalf("List: %v", err) } - if len(entries) != len(mgr.AgentFileList) { - t.Errorf("List() returned %d entries; want %d", len(entries), len(mgr.AgentFileList)) + if len(entries) != len(session.AgentFileList) { + t.Errorf("List() returned %d entries; want %d", len(entries), len(session.AgentFileList)) } } @@ -424,7 +424,7 @@ func TestSessionFileStoreStatChat(t *testing.T) { defer sess.Cancel() sess.AppendLog([]byte("hello")) sink := testSink() - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func([]byte) error { return nil }, func() {}, nil, nil) fi, err := sf.Stat("chat") @@ -441,7 +441,7 @@ func TestSessionFileStoreGetChat(t *testing.T) { defer sess.Cancel() sess.AppendLog([]byte("hello")) sink := testSink() - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func([]byte) error { return nil }, func() {}, nil, nil) data := testStoreRead(t, sf, "chat") @@ -454,7 +454,7 @@ func TestSessionFileStoreGetContent(t *testing.T) { sess := testSession("s1") defer sess.Cancel() sink := testSink() - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func([]byte) error { return nil }, func() {}, nil, nil) for _, name := range []string{"cfg", "offset", "usage", "ctxsz", "models", "systemprompt"} { @@ -468,7 +468,7 @@ func TestSessionFileStorePutCwd(t *testing.T) { sess := testSession("s1") defer sess.Cancel() sink := testSink() - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func([]byte) error { return nil }, func() {}, nil, nil) // Create the directory first so SetCWD validates it @@ -484,7 +484,7 @@ func TestSessionFileStorePutEmpty(t *testing.T) { sess := testSession("s1") defer sess.Cancel() sink := testSink() - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func([]byte) error { return nil }, func() {}, nil, nil) // Empty write is a no-op @@ -591,7 +591,7 @@ func TestSessionFileStoreWriteChat(t *testing.T) { defer sess.Cancel() var saved []byte sink := testSink() - sf := mgr.NewAgentTree(sess, sink.NewLogger("test"), + sf := session.NewAgentTree(sess, sink.NewLogger("test"), func(data []byte) error { saved = data; return nil }, func() {}, nil, nil) testStoreWrite(t, sf, "chat", []byte("transcript data")) @@ -833,7 +833,7 @@ func TestSessionFileStoreMakePublish(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() core := newContentCore("s1", "hello world") - sess := mgr.NewSession("s1", core, ctx, cancel) + sess := session.NewSession("s1", core, ctx, cancel) sf := newTestSessionFileStore(t, sess) // Submit triggers the content backend @@ -859,7 +859,7 @@ func TestSessionFileStoreMakePublishMultipleEvents(t *testing.T) { close(ch) return ch, nil } - core := session.New(session.Config{ + core := coresession.New(coresession.Config{ Backend: be, AgentName: "default", CWD: "/tmp", @@ -868,7 +868,7 @@ func TestSessionFileStoreMakePublishMultipleEvents(t *testing.T) { }) ctx, cancel := context.WithCancel(context.Background()) defer cancel() - sess := mgr.NewSession("s1", core, ctx, cancel) + sess := session.NewSession("s1", core, ctx, cancel) sf := newTestSessionFileStore(t, sess) testStoreWrite(t, sf, "prompt", []byte("multi")) @@ -923,10 +923,10 @@ func TestSessionManagerOpenStore(t *testing.T) { s := newTestSessionManager(t) sess := testSession("s1") defer sess.Cancel() - s.AddSession(sess) + session.AddSession(s, sess) // Access a session-level file through the manager - e, err := s.Tree().Open("s1/plan") + e, err := s.Open("s1/plan") if err != nil { t.Fatalf("Open(s1/plan): %v", err) } @@ -941,7 +941,7 @@ func TestLoadAgentConfig(t *testing.T) { data := []byte(`{"prompt":"test prompt","maxTokens":1024}`) os.WriteFile(filepath.Join(dir, "test.json"), data, 0644) - cfg := mgr.LoadAgentConfig(dir, "test", nil) + cfg := session.LoadAgentConfig(dir, "test", nil) if cfg == nil { t.Fatal("LoadAgentConfig returned nil") } @@ -962,7 +962,7 @@ func TestFormatEvent(t *testing.T) { {agent.Event{Role: "info", Content: "msg\n"}, "[info]\nmsg\n"}, } for _, tc := range tests { - got := string(mgr.FormatEvent(tc.ev)) + got := string(session.FormatEvent(tc.ev)) if got != tc.want { t.Errorf("FormatEvent(%v) = %q; want %q", tc.ev.Role, got, tc.want) } @@ -1007,7 +1007,7 @@ func TestFormatParamsRoundTrip(t *testing.T) { Temperature: &temp, TopP: &topP, } - text := mgr.FormatParams(p) + text := session.FormatParams(p) if !strings.Contains(text, "maxTokens=4096") { t.Errorf("FormatParams missing maxTokens; got: %s", text) } @@ -1015,7 +1015,7 @@ func TestFormatParamsRoundTrip(t *testing.T) { t.Errorf("FormatParams missing temperature; got: %s", text) } - parsed, err := mgr.ParseParams(text, backend.GenerationParams{}) + parsed, err := session.ParseParams(text, backend.GenerationParams{}) if err != nil { t.Fatalf("ParseParams: %v", err) } @@ -1029,7 +1029,7 @@ func TestFormatParamsRoundTrip(t *testing.T) { func TestFormatParamsNilOptionals(t *testing.T) { p := backend.GenerationParams{MaxTokens: 100} - text := mgr.FormatParams(p) + text := session.FormatParams(p) // Real FormatParams lists all fields; nil optionals appear as empty values if !strings.Contains(text, "maxTokens=100") { t.Errorf("FormatParams missing maxTokens=100; got: %s", text) @@ -1043,7 +1043,7 @@ func TestFormatParamsNilOptionals(t *testing.T) { func TestParseParamsClearWithEmpty(t *testing.T) { temp := 0.7 base := backend.GenerationParams{Temperature: &temp, MaxTokens: 100} - parsed, err := mgr.ParseParams("temperature=\nmaxTokens=200", base) + parsed, err := session.ParseParams("temperature=\nmaxTokens=200", base) if err != nil { t.Fatalf("ParseParams: %v", err) } @@ -1056,11 +1056,11 @@ func TestParseParamsClearWithEmpty(t *testing.T) { } func TestParseParamsErrors(t *testing.T) { - _, err := mgr.ParseParams("maxTokens=notanumber", backend.GenerationParams{}) + _, err := session.ParseParams("maxTokens=notanumber", backend.GenerationParams{}) if err == nil { t.Error("non-numeric maxTokens should error") } - _, err = mgr.ParseParams("temperature=notafloat", backend.GenerationParams{}) + _, err = session.ParseParams("temperature=notafloat", backend.GenerationParams{}) if err == nil { t.Error("non-numeric temperature should error") } @@ -1082,7 +1082,7 @@ func TestSessionInterrupt(t *testing.T) { func TestSessionManagerCreateSessionViaWrite(t *testing.T) { s := newTestSessionManagerWithCore(t) - e, err := s.Tree().Open("new") + e, err := s.Open("new") if err != nil { t.Fatalf("Open(new): %v", err) } @@ -1090,7 +1090,7 @@ func TestSessionManagerCreateSessionViaWrite(t *testing.T) { t.Fatalf("Write(new): %v", err) } // Should have created a session - entries, _ := s.Tree().List() + entries, _ := s.List() found := false for _, ent := range entries { if ent.IsDir() { @@ -1105,7 +1105,7 @@ func TestSessionManagerCreateSessionViaWrite(t *testing.T) { func TestSessionManagerCreateSessionNoCwd(t *testing.T) { s := newTestSessionManagerWithCore(t) - e, err := s.Tree().Open("new") + e, err := s.Open("new") if err != nil { t.Fatalf("Open(new): %v", err) } @@ -1119,9 +1119,9 @@ func TestSessionManagerCreateSessionDuplicate(t *testing.T) { s := newTestSessionManagerWithCore(t) sess := testSession("dup") defer sess.Cancel() - s.AddSession(sess) + session.AddSession(s, sess) - e, _ := s.Tree().Open("new") + e, _ := s.Open("new") if err := e.Write([]byte("name=dup\ncwd=/tmp")); err == nil { t.Error("creating duplicate session should error") } @@ -1129,7 +1129,7 @@ func TestSessionManagerCreateSessionDuplicate(t *testing.T) { func TestSessionManagerCreateSessionBadOption(t *testing.T) { s := newTestSessionManagerWithCore(t) - e, _ := s.Tree().Open("new") + e, _ := s.Open("new") // Malformed line (no =) if err := e.Write([]byte("badline\ncwd=/tmp")); err == nil { t.Error("malformed option should error") @@ -1138,7 +1138,7 @@ func TestSessionManagerCreateSessionBadOption(t *testing.T) { func TestSessionManagerCreateSessionUnknownKey(t *testing.T) { s := newTestSessionManagerWithCore(t) - e, _ := s.Tree().Open("new") + e, _ := s.Open("new") if err := e.Write([]byte("unknownkey=val\ncwd=/tmp")); err == nil { t.Error("unknown key should error") } @@ -1147,15 +1147,15 @@ func TestSessionManagerCreateSessionUnknownKey(t *testing.T) { func TestSessionManagerCreateSessionEnvExpansion(t *testing.T) { t.Setenv("TEST_CWD", "/expanded/path") sink := testSink() - s := mgr.NewManager(mgr.ManagerConfig{ + s := session.NewRoot(session.Config{ Ctx: context.Background(), Log: sink.NewLogger("test"), Sink: sink, ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil }, MkdirAll: func(string, os.FileMode) error { return nil }, - NewCore: func(sessionID, agentName, cwd string) (*session.Session, error) { + NewCore: func(sessionID, agentName, cwd string) (*coresession.Session, error) { be := backend.NewNoop("stub", "m") - return session.New(session.Config{ + return coresession.New(coresession.Config{ Backend: be, AgentName: agentName, CWD: cwd, @@ -1164,7 +1164,7 @@ func TestSessionManagerCreateSessionEnvExpansion(t *testing.T) { }), nil }, }) - e, _ := s.Tree().Open("new") + e, _ := s.Open("new") if err := e.Write([]byte("cwd=$TEST_CWD")); err != nil { t.Fatalf("Write: %v", err) } @@ -1172,15 +1172,15 @@ func TestSessionManagerCreateSessionEnvExpansion(t *testing.T) { func TestSessionManagerCreateSessionTildeExpansion(t *testing.T) { sink := testSink() - s := mgr.NewManager(mgr.ManagerConfig{ + s := session.NewRoot(session.Config{ Ctx: context.Background(), Log: sink.NewLogger("test"), Sink: sink, ReadFile: func(string) ([]byte, error) { return []byte("#!/bin/sh\n"), nil }, MkdirAll: func(string, os.FileMode) error { return nil }, - NewCore: func(sessionID, agentName, cwd string) (*session.Session, error) { + NewCore: func(sessionID, agentName, cwd string) (*coresession.Session, error) { be := backend.NewNoop("stub", "m") - return session.New(session.Config{ + return coresession.New(coresession.Config{ Backend: be, AgentName: agentName, CWD: cwd, @@ -1189,7 +1189,7 @@ func TestSessionManagerCreateSessionTildeExpansion(t *testing.T) { }), nil }, }) - e, _ := s.Tree().Open("new") + e, _ := s.Open("new") if err := e.Write([]byte("cwd=~/projects")); err != nil { t.Fatalf("Write: %v", err) } @@ -1197,14 +1197,14 @@ func TestSessionManagerCreateSessionTildeExpansion(t *testing.T) { func TestSessionManagerOpenEntryNotFound(t *testing.T) { s := newTestSessionManager(t) - if _, err := s.Tree().Open("nonexistent/cfg"); err == nil { + if _, err := s.Open("nonexistent/cfg"); err == nil { t.Error("Open(nonexistent/cfg) should error") } } func TestSessionManagerOpenStoreNotFound(t *testing.T) { s := newTestSessionManager(t) - if _, err := s.Tree().Open("nosess/chat"); err == nil { + if _, err := s.Open("nosess/chat"); err == nil { t.Error("Open for nonexistent session should error") } } diff --git a/mgr/persist.go b/mgr/persist.go deleted file mode 100644 index d0a1667..0000000 --- a/mgr/persist.go +++ /dev/null @@ -1,348 +0,0 @@ -package mgr - -import ( - "ollie/session" - "ollie/agent" - "context" - "fmt" - "ollie/backend" - "ollie/toolsrv" - "ollie/tools" - "olliesrv/prompts" - "os" - "path/filepath" - "sort" - "strings" - "sync" - "time" -) - -// --- Session Persistence --- - -func (s *Manager) activeSessionsDir() string { - return filepath.Join(s.cfg.SessionsDir, "active") -} - -func (s *Manager) persistSession(id string) { - s.mu.RLock() - sess, ok := s.sessions[id] - s.mu.RUnlock() - if !ok { - return - } - dir := s.activeSessionsDir() - os.MkdirAll(dir, 0700) - path := filepath.Join(dir, id+".json") - if err := sess.Core.SaveSession(path); err != nil { - s.cfg.Log.Error("persist session %s: %v", id, err) - } -} - -func (s *Manager) removePersistedAgent(id string) { - path := filepath.Join(s.activeSessionsDir(), id+".json") - os.Remove(path) -} - -func (s *Manager) saveAllSessions() { - s.mu.RLock() - ids := make([]string, 0, len(s.sessions)) - for id := range s.sessions { - ids = append(ids, id) - } - s.mu.RUnlock() - for _, id := range ids { - s.persistSession(id) - } -} - -func (s *Manager) restoreAllSessions() { - dir := s.activeSessionsDir() - entries, err := os.ReadDir(dir) - if err != nil { - return - } - - // Load all persisted session JSONs (fast, sequential disk reads) - type loadedSession struct { - ps *agent.PersistedAgent - name string - } - var loaded []loadedSession - for _, e := range entries { - if !strings.HasSuffix(e.Name(), ".json") { - continue - } - path := filepath.Join(dir, e.Name()) - ps, err := agent.LoadPersistedAgent(path) - if err != nil { - s.cfg.Log.Error("restore session %s: %v", e.Name(), err) - continue - } - loaded = append(loaded, loadedSession{ps: ps, name: e.Name()}) - } - - if len(loaded) == 0 { - return - } - - // Restore sessions in parallel - var wg sync.WaitGroup - for _, ls := range loaded { - wg.Add(1) - go func(ls loadedSession) { - defer wg.Done() - if err := s.restoreSession(ls.ps); err != nil { - s.cfg.Log.Error("restore session %s: %v", ls.ps.ID, err) - } - }(ls) - } - wg.Wait() -} - -func (s *Manager) restoreSession(ps *agent.PersistedAgent) error { - cwd := ps.CWD - if cwd == "" { - cwd, _ = os.Getwd() - } - agentName := ps.Agent - if agentName == "" { - agentName = "default" - } - sessID := ps.ID - - cfg := LoadAgentConfig(s.cfg.AgentsDir, agentName, nil) - - 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) - } - modelName := ps.Model - if modelName == "" && cfg != nil && cfg.Model != "" { - modelName = cfg.Model - } - if modelName == "" { - modelName = os.Getenv("OLLIE_MODEL") - } - if modelName != "" { - be.SetModel(modelName) - } - - uname := s.nextUname() - var newToolServer func() toolsrv.Runner - var promptEnv []string - remoteTarget := ps.Remote - - if remoteTarget != "" { - rsrv, dialErr := toolsrv.RemoteDial(s.cfg.Ctx, toolsrv.RemoteConfig{ - SSHTarget: remoteTarget, - CWD: cwd, - }) - if dialErr != nil { - return fmt.Errorf("remote dial: %w", dialErr) - } - newToolServer = func() toolsrv.Runner { return rsrv } - promptEnv = []string{ - "PRIME_CWD=" + cwd, - "PRIME_PLATFORM=" + rsrv.Info.Platform, - "PRIME_IS_GIT_REPO=" + fmt.Sprintf("%v", rsrv.Info.IsGitRepo), - } - } else { - var execOpts []toolsrv.Option - if !s.cfg.NoMount { - } - if s.cfg.Strict { - execOpts = append(execOpts, toolsrv.WithStrict()) - } - if s.cfg.Yolo { - execOpts = append(execOpts, toolsrv.WithYolo()) - } - if s.cfg.ToolRegistry != nil { - execOpts = append(execOpts, toolsrv.WithToolRegistry(s.cfg.ToolRegistry, sessID)) - } - if s.cfg.SkillsRegistry != nil { - execOpts = append(execOpts, toolsrv.WithSkillsRegistry(s.cfg.SkillsRegistry)) - execOpts = append(execOpts, toolsrv.WithBuiltins(tools.Builtins())) - } - newToolServer = func() toolsrv.Runner { - srv, _ := toolsrv.Dial(s.cfg.Ctx, toolsrv.LocalAddr(cwd), execOpts...) - return srv - } - promptEnv = agent.PromptEnv(cwd) - } - - env := []string{"OLLIE_SESSION_ID=" + sessID, "OLLIE_UNAME=" + uname} - env = append(env, promptEnv...) - - // Compute base layers (same as new-session path). - var spOverride string - if cfg != nil { - spOverride = cfg.SystemPrompt - } - sysPrompt := prompts.ResolveSystemPrompt(spOverride) - envMap := make(map[string]string) - for _, e := range env { - if k, v, ok := strings.Cut(e, "="); ok { - envMap[k] = v - } - } - opModel := prompts.OperationalModel(s.cfg.Enable9P, s.cfg.EnableDBus, envMap) - platform := "linux" - isGitRepo := false - for _, e := range promptEnv { - if k, v, ok := strings.Cut(e, "="); ok { - switch k { - case "PRIME_PLATFORM": - platform = v - case "PRIME_IS_GIT_REPO": - isGitRepo = v == "true" - } - } - } - envBlock := prompts.Environment(cwd, platform, isGitRepo, "") - - toolSrv := newToolServer() - rt := agent.BuildRuntime(cfg, toolSrv, cwd, env, sysPrompt, opModel, envBlock) - - restoredSession := agent.RestoreHistory(ps) - - var sessPtr *Session - core := session.New(session.Config{ - Backend: be, - AgentName: agentName, - AgentsDir: s.cfg.AgentsDir, - SessionsDir: s.cfg.SessionsDir, - SessionID: sessID, - AgentID: uname, - CWD: cwd, - Remote: remoteTarget, - History: restoredSession, - Runtime: rt, - NewToolServer: newToolServer, - PromptEnvExtra: promptEnv, - BaseLayers: []string{sysPrompt, opModel, envBlock}, - Log: s.cfg.Sink.NewLogger("core"), - ReadPlanStep: func() string { - if sessPtr == nil { - return "" - } - sessPtr.mu.RLock() - data := make([]byte, len(sessPtr.plan)) - copy(data, sessPtr.plan) - sessPtr.mu.RUnlock() - return session.NextUncheckedStep(data) - }, - }) - - sessionCtx, sessionCancel := context.WithCancel(s.cfg.Ctx) - sess := NewSession(sessID, core, sessionCtx, sessionCancel) - sessPtr = sess - sess.uname = uname - sess.remote = remoteTarget - - // Replay tail of persisted messages into the chat log so the GUI - // and `chat` file show recent history on restore. - replayMessagesToLog(sess, ps.Messages) - - s.mu.Lock() - s.sessions[sessID] = sess - s.mu.Unlock() - - s.cfg.Log.Info("restored session %s (backend=%s model=%s agent=%s)", sessID, backendName, modelName, agentName) - if s.cfg.OnSessionCreated != nil { - s.cfg.OnSessionCreated(sessID, sess) - } - return nil -} - -func (s *Manager) Shutdown() { - // Interrupt all in-progress turns and wait for them to finish - // before persisting state, so we capture the latest messages. - s.InterruptAll() - s.waitIdle(100*time.Millisecond, 5*time.Second) - s.saveAllSessions() - s.mu.Lock() - ids := make([]string, 0, len(s.sessions)) - for id := range s.sessions { - ids = append(ids, id) - } - s.mu.Unlock() - for _, id := range ids { - s.mu.Lock() - sess := s.sessions[id] - delete(s.sessions, id) - s.mu.Unlock() - if sess != nil { - sess.Cancel() - sess.Core.Close() - - s.cfg.Log.Info("shutdown session %s", id) - } - } -} - -// waitIdle polls all sessions until they are idle or the timeout expires. -func (s *Manager) waitIdle(poll, timeout time.Duration) { - deadline := time.Now().Add(timeout) - for time.Now().Before(deadline) { - allIdle := true - s.mu.RLock() - for _, sess := range s.sessions { - if sess.Core.Agent().State() != "idle" { - allIdle = false - break - } - } - s.mu.RUnlock() - if allIdle { - return - } - time.Sleep(poll) - } - s.cfg.Log.Warn("shutdown: timed out waiting for sessions to become idle, saving anyway") -} - -func (s *Manager) KillSession(id string) { - s.mu.Lock() - sess := s.sessions[id] - delete(s.sessions, id) - s.mu.Unlock() - if sess != nil { - sess.Cancel() // signal: context cancellation propagates to all agents - sess.Core.Close() - s.removePersistedAgent(id) - s.cfg.Log.Info("killed session %s", id) - if s.cfg.OnSessionKilled != nil { - s.cfg.OnSessionKilled(id) - } - } -} - -func (s *Manager) index() []byte { - var sb strings.Builder - s.mu.RLock() - ids := make([]string, 0, len(s.sessions)) - for id := range s.sessions { - ids = append(ids, id) - } - sort.Strings(ids) - for _, id := range ids { - sess := s.sessions[id] - sess.mu.RLock() - state := sess.Core.Agent().State() - cwd := sess.Core.CWD() - be := sess.Core.Agent().BackendName() - model := sess.Core.Agent().ModelName() - agent := sess.Core.Agent().Name() - sess.mu.RUnlock() - fmt.Fprintf(&sb, "%s\t%s\t%s\t%s\t%s\t%s\n", id, state, cwd, be, model, agent) - } - s.mu.RUnlock() - return []byte(sb.String()) -} - -// CreateSession creates a new agent session from key=value args. -// Returns the session ID on success. diff --git a/server/server.go b/server/server.go index e443b63..a3dd7d5 100644 --- a/server/server.go +++ b/server/server.go @@ -21,7 +21,7 @@ import ( olog "ollie/log" "ollie/paths" "olliesrv/fs" - "olliesrv/mgr" + "olliesrv/session" "ollie/agent" "9fans.net/go/plan9" @@ -69,7 +69,7 @@ type Server struct { conns []*connState log *olog.Logger sink *olog.Sink - sessionMgr *mgr.Manager + sessionTree *fs.Tree rootStore *fs.Tree elevateTree *elevateTree groups map[string]map[string]bool // group → set of members @@ -81,7 +81,7 @@ type Server struct { // Config holds the pre-built trees and manager for the server. type Config struct { Sink *olog.Sink - SessionMgr *mgr.Manager + SessionTree *fs.Tree RootStore *fs.Tree ElevateBroker *elevate.Broker InvalidateModels func() @@ -98,7 +98,7 @@ func New(cfg Config) *Server { s := &Server{ log: cfg.Sink.Logger("9p", olog.LevelDebug), sink: cfg.Sink, - sessionMgr: cfg.SessionMgr, + sessionTree: cfg.SessionTree, rootStore: cfg.RootStore, groups: make(map[string]map[string]bool), invalidateModels: cfg.InvalidateModels, @@ -122,7 +122,7 @@ type fsRoute struct { // Directory permissions come from the fs.Perms registry. func (s *Server) routes() []fsRoute { routes := []fsRoute{ - {fs.PathSessions, fs.Perms[fs.PathSessions].DirMode, func() FileTree { return s.sessionMgr.Tree() }}, + {fs.PathSessions, fs.Perms[fs.PathSessions].DirMode, func() FileTree { return s.sessionTree }}, } if s.elevateTree != nil { routes = append(routes, fsRoute{fs.PathElevate, fs.Perms[fs.PathElevate].DirMode, func() FileTree { return s.elevateTree }}) @@ -261,7 +261,7 @@ func (s *Server) fileOwnerGroup(path string) (uid, gid string) { if strings.HasPrefix(path, fs.PathSessions) { parts := strings.SplitN(strings.TrimPrefix(path, fs.PathSessions), "/", 2) if len(parts) >= 1 && parts[0] != "new" && !isSessionFile(path) { - if sess := s.sessionMgr.Session(parts[0]); sess != nil { + if sess := session.Lookup(s.sessionTree, parts[0]); sess != nil { return sess.Uname(), "agent" } } @@ -489,7 +489,7 @@ func isSessionFile(path string) bool { if !ok || strings.Contains(name, "/") { return false } - _, ok = mgr.FileMode(name) + _, ok = session.FileMode(name) return ok } @@ -536,7 +536,7 @@ func (s *Server) pathType(path string) string { } // /session/{id}/proc/{pid} — verify PID exists in detached list if len(parts) == 3 && parts[1] == "proc" { - sess := s.sessionMgr.Session(parts[0]) + sess := session.Lookup(s.sessionTree, parts[0]) if sess == nil { return "" } @@ -1100,12 +1100,12 @@ func (s *Server) handleWrite(path, input, uname string) error { // Shutdown kills all active sessions and batch jobs. func (s *Server) Shutdown() { - s.sessionMgr.Shutdown() + session.Shutdown(s.sessionTree) } // InterruptAll cancels any in-progress agent turn on every active session. func (s *Server) InterruptAll() { - s.sessionMgr.InterruptAll() + session.InterruptAll(s.sessionTree) } // buildDirData serializes all directory entries for the given path into a byte slice. @@ -1218,7 +1218,7 @@ func (s *Server) makeStat(path string) plan9.Dir { if strings.HasPrefix(path, fs.PathSessions) { parts := strings.SplitN(strings.TrimPrefix(path, "/"), "/", 3) if len(parts) == 3 && parts[0] == "session" { - if sess := s.sessionMgr.Session(parts[1]); sess != nil { + if sess := session.Lookup(s.sessionTree, parts[1]); sess != nil { if base == "chat" { length, vers := sess.LogInfo() dir.Length = uint64(length) @@ -1233,7 +1233,7 @@ func (s *Server) makeStat(path string) plan9.Dir { if dir.Length == 0 && !isDir { switch { case isSessionFile(path): - if content, err := readFile(s.sessionMgr.Tree(), base); err == nil { + if content, err := readFile(s.sessionTree, base); err == nil { dir.Length = uint64(len(content)) } case path == "/backends": diff --git a/mgr/create.go b/session/create.go similarity index 56% rename from mgr/create.go rename to session/create.go index 7609738..0465184 100644 --- a/mgr/create.go +++ b/session/create.go @@ -1,21 +1,23 @@ -package mgr +package session import ( - "ollie/session" - "ollie/agent" "context" "fmt" + "os" + "strings" + + "ollie/agent" "ollie/backend" + coresession "ollie/session" "ollie/toolsrv" "ollie/tools" "ollie/paths" "olliesrv/fs" "olliesrv/prompts" - "os" - "strings" ) -func (s *Manager) CreateSession(args []string) (string, error) { +// Create creates a new session from key=value args. Returns the session ID. +func Create(rs *rootState, args []string) (string, error) { name := "" backendOverride := "" modelOverride := "" @@ -65,27 +67,27 @@ func (s *Manager) CreateSession(args []string) (string, error) { sessID := name if sessID == "" { - sessID = session.NewSessionID() + sessID = coresession.NewSessionID() } - s.mu.RLock() - _, exists := s.sessions[sessID] - s.mu.RUnlock() + rs.mu.RLock() + _, exists := rs.sessions[sessID] + rs.mu.RUnlock() if exists { return "", fmt.Errorf("session already exists: %s", sessID) } - var core *session.Session + var core *coresession.Session var sessPtr *Session - uname := s.nextUname() - if s.cfg.NewCore != nil { + uname := rs.nextUname() + if rs.cfg.NewCore != nil { var err error - core, err = s.cfg.NewCore(sessID, agentName, cwd) + core, err = rs.cfg.NewCore(sessID, agentName, cwd) if err != nil { return "", err } } else { - cfg := LoadAgentConfig(s.cfg.AgentsDir, agentName, nil) + cfg := LoadAgentConfig(rs.cfg.AgentsDir, agentName, nil) if cfg != nil { if backendOverride == "" && cfg.Backend != "" { backendOverride = cfg.Backend @@ -107,38 +109,32 @@ func (s *Manager) CreateSession(args []string) (string, error) { be.SetModel(modelOverride) } - if err := s.cfg.MkdirAll(s.cfg.SessionsDir, fs.PermMkdirPrivate); err != nil { + if err := rs.cfg.MkdirAll(rs.cfg.SessionsDir, fs.PermMkdirPrivate); err != nil { return "", fmt.Errorf("sessions dir: %w", err) } var execOpts []toolsrv.Option - if !s.cfg.NoMount { - } - if s.cfg.Strict { + if rs.cfg.Strict { execOpts = append(execOpts, toolsrv.WithStrict()) } - if s.cfg.Yolo { + if rs.cfg.Yolo { execOpts = append(execOpts, toolsrv.WithYolo()) } if cfg != nil && len(cfg.AllowTools) > 0 { execOpts = append(execOpts, toolsrv.WithAllowTools(cfg.AllowTools)) } - if s.cfg.ToolRegistry != nil { - execOpts = append(execOpts, toolsrv.WithToolRegistry(s.cfg.ToolRegistry, sessID)) + if rs.cfg.ToolRegistry != nil { + execOpts = append(execOpts, toolsrv.WithToolRegistry(rs.cfg.ToolRegistry, sessID)) } - if s.cfg.SkillsRegistry != nil { - execOpts = append(execOpts, toolsrv.WithSkillsRegistry(s.cfg.SkillsRegistry)) - execOpts = append(execOpts, toolsrv.WithBuiltins(tools.Builtins())) + if rs.cfg.SkillsRegistry != nil { + execOpts = append(execOpts, toolsrv.WithSkillsRegistry(rs.cfg.SkillsRegistry)) + execOpts = append(execOpts, toolsrv.WithBuiltins(tools.Builtins())) } var newToolServer func() toolsrv.Runner - var remoteEnv []string // extra env vars from remote host info + var remoteEnv []string if remoteTarget != "" { - // Dial eagerly so we get HostInfo for prompt resolution. - // FIXME: Eager dial means session creation blocks on SSH connect. - // Eventually make this async or move host info into the prompt - // scripts themselves (running remotely via the RPC). - rsrv, dialErr := toolsrv.RemoteDial(s.cfg.Ctx, toolsrv.RemoteConfig{ + rsrv, dialErr := toolsrv.RemoteDial(rs.cfg.Ctx, toolsrv.RemoteConfig{ SSHTarget: remoteTarget, CWD: cwd, }) @@ -153,16 +149,9 @@ func (s *Manager) CreateSession(args []string) (string, error) { } } else { newToolServer = func() toolsrv.Runner { - srv, _ := toolsrv.Dial(s.cfg.Ctx, toolsrv.LocalAddr(cwd), execOpts...) + srv, _ := toolsrv.Dial(rs.cfg.Ctx, toolsrv.LocalAddr(cwd), execOpts...) return srv } - - - - - - - } var promptEnv []string @@ -174,7 +163,6 @@ func (s *Manager) CreateSession(args []string) (string, error) { env := []string{"OLLIE_SESSION_ID=" + sessID, "OLLIE_UNAME=" + uname} env = append(env, promptEnv...) - // Compose base layers: system prompt + operational model + environment. spOverride := systemPromptOverride if spOverride == "" && cfg != nil { spOverride = cfg.SystemPrompt @@ -187,9 +175,8 @@ func (s *Manager) CreateSession(args []string) (string, error) { envMap[k] = v } } - opModel := prompts.OperationalModel(s.cfg.Enable9P, s.cfg.EnableDBus, envMap) + opModel := prompts.OperationalModel(rs.cfg.Enable9P, rs.cfg.EnableDBus, envMap) - // Determine environment values for the environment layer. platform := "linux" isGitRepo := false for _, e := range promptEnv { @@ -207,21 +194,20 @@ func (s *Manager) CreateSession(args []string) (string, error) { toolSrv := newToolServer() rt := agent.BuildRuntime(cfg, toolSrv, cwd, env, sysPrompt, opModel, envBlock) - // sessPtr is set after NewSession; the ReadPlanStep closure captures it. - core = session.New(session.Config{ + core = coresession.New(coresession.Config{ Backend: be, AgentName: agentName, - AgentsDir: s.cfg.AgentsDir, - SessionsDir: s.cfg.SessionsDir, + AgentsDir: rs.cfg.AgentsDir, + SessionsDir: rs.cfg.SessionsDir, SessionID: sessID, - AgentID: uname, + AgentID: uname, CWD: cwd, Remote: remoteTarget, Runtime: rt, NewToolServer: newToolServer, PromptEnvExtra: promptEnv, BaseLayers: []string{sysPrompt, opModel, envBlock}, - Log: s.cfg.Sink.NewLogger("core"), + Log: rs.cfg.Sink.NewLogger("core"), ReadPlanStep: func() string { if sessPtr == nil { return "" @@ -230,76 +216,33 @@ func (s *Manager) CreateSession(args []string) (string, error) { data := make([]byte, len(sessPtr.plan)) copy(data, sessPtr.plan) sessPtr.mu.RUnlock() - return session.NextUncheckedStep(data) + return coresession.NextUncheckedStep(data) }, }) } - sessionCtx, sessionCancel := context.WithCancel(s.cfg.Ctx) + sessionCtx, sessionCancel := context.WithCancel(rs.cfg.Ctx) sess := NewSession(sessID, core, sessionCtx, sessionCancel) sessPtr = sess sess.remote = remoteTarget - // For remote sessions, the mount isn't managed by a local toolsrv.Server - // (which doesn't exist). Start it directly so $OLLIE is available for - s.mu.Lock() + rs.mu.Lock() sess.uname = uname - s.sessions[sessID] = sess - s.mu.Unlock() + rs.sessions[sessID] = sess + rs.mu.Unlock() - // HACK: Load the agent now that the session mount is active. + // Load the agent now that the session is registered. core.Agent().Submit(sessionCtx, "/agent "+agentName) - s.cfg.Log.Info("new session %s (backend=%s model=%s agent=%s)", + rs.cfg.Log.Info("new session %s (backend=%s model=%s agent=%s)", sessID, core.Agent().BackendName(), core.Agent().ModelName(), core.Agent().Name()) - if s.cfg.OnSessionCreated != nil { - s.cfg.OnSessionCreated(sessID, sess) + if rs.cfg.OnSessionCreated != nil { + rs.cfg.OnSessionCreated(sessID, sess) } return sessID, nil } -// RenameSession renames a session (exported for D-Bus adapter). -func (s *Manager) RenameSession(old, new string) error { - return s.renameSession(old, new) +// CreateFromRoot is the public API for callers that have a *fs.Tree. +func CreateFromRoot(root *fs.Tree, args []string) (string, error) { + return Create(state(root), args) } - -func (s *Manager) renameSession(old, new string) error { - // Session rename: {oldID} -> {newID} - oldID := old - newID := new - s.mu.Lock() - - sess, ok := s.sessions[oldID] - if !ok { - s.mu.Unlock() - return fmt.Errorf("session not found: %s", oldID) - } - if _, exists := s.sessions[newID]; exists { - s.mu.Unlock() - return fmt.Errorf("session already exists: %s", newID) - } - if sess.Core.Agent().IsRunning() { - s.mu.Unlock() - return fmt.Errorf("cannot rename while agent is running") - } - - if err := sess.Core.SetSessionID(newID); err != nil { - s.mu.Unlock() - return err - } - - sess.id = newID - s.sessions[newID] = sess - delete(s.sessions, oldID) - s.mu.Unlock() - - sess.AppendLog([]byte(fmt.Sprintf(":: session renamed: %s -> %s\n", oldID, newID))) - s.cfg.Log.Info("renamed session %s -> %s", oldID, newID) - if s.cfg.OnSessionRenamed != nil { - s.cfg.OnSessionRenamed(oldID, newID) - } - return nil -} - -// LoadAgentConfig resolves and loads the config for a named agent. -// Returns nil if the config file does not exist; BuildRuntime handles nil configs. diff --git a/mgr/sessionfile.go b/session/files.go similarity index 99% rename from mgr/sessionfile.go rename to session/files.go index bc724a4..efcf4b2 100644 --- a/mgr/sessionfile.go +++ b/session/files.go @@ -1,4 +1,4 @@ -package mgr +package session import ( "context" diff --git a/mgr/format.go b/session/format.go similarity index 99% rename from mgr/format.go rename to session/format.go index 596f441..450dbce 100644 --- a/mgr/format.go +++ b/session/format.go @@ -1,4 +1,4 @@ -package mgr +package session import ( "ollie/agent" diff --git a/session/persist.go b/session/persist.go new file mode 100644 index 0000000..7b979f4 --- /dev/null +++ b/session/persist.go @@ -0,0 +1,360 @@ +package session + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "ollie/agent" + "ollie/backend" + coresession "ollie/session" + "ollie/toolsrv" + "ollie/tools" + "olliesrv/fs" + "olliesrv/prompts" +) + +// Kill terminates a session and removes it from the tree. +func Kill(rs *rootState, id string) { + rs.mu.Lock() + sess := rs.sessions[id] + delete(rs.sessions, id) + rs.mu.Unlock() + if sess != nil { + sess.Cancel() + sess.Core.Close() + removePersistedAgent(rs, id) + rs.cfg.Log.Info("killed session %s", id) + if rs.cfg.OnSessionKilled != nil { + rs.cfg.OnSessionKilled(id) + } + } +} + +// KillFromRoot is the public API for callers with *fs.Tree. +func KillFromRoot(root *fs.Tree, id string) { + Kill(state(root), id) +} + +// Rename renames a session. +func Rename(rs *rootState, old, new string) error { + rs.mu.Lock() + sess, ok := rs.sessions[old] + if !ok { + rs.mu.Unlock() + return fmt.Errorf("session not found: %s", old) + } + if _, exists := rs.sessions[new]; exists { + rs.mu.Unlock() + return fmt.Errorf("session already exists: %s", new) + } + if sess.Core.Agent().IsRunning() { + rs.mu.Unlock() + return fmt.Errorf("cannot rename while agent is running") + } + + if err := sess.Core.SetSessionID(new); err != nil { + rs.mu.Unlock() + return err + } + + sess.id = new + rs.sessions[new] = sess + delete(rs.sessions, old) + rs.mu.Unlock() + + sess.AppendLog([]byte(fmt.Sprintf(":: session renamed: %s -> %s\n", old, new))) + rs.cfg.Log.Info("renamed session %s -> %s", old, new) + if rs.cfg.OnSessionRenamed != nil { + rs.cfg.OnSessionRenamed(old, new) + } + return nil +} + +// RenameFromRoot is the public API for callers with *fs.Tree. +func RenameFromRoot(root *fs.Tree, old, new string) error { + return Rename(state(root), old, new) +} + +// Shutdown interrupts all sessions, waits for idle, persists, and closes. +func Shutdown(root *fs.Tree) { + rs := state(root) + InterruptAll(root) + waitIdle(rs, 100*time.Millisecond, 5*time.Second) + saveAllSessions(rs) + rs.mu.Lock() + ids := make([]string, 0, len(rs.sessions)) + for id := range rs.sessions { + ids = append(ids, id) + } + rs.mu.Unlock() + for _, id := range ids { + rs.mu.Lock() + sess := rs.sessions[id] + delete(rs.sessions, id) + rs.mu.Unlock() + if sess != nil { + sess.Cancel() + sess.Core.Close() + rs.cfg.Log.Info("shutdown session %s", id) + } + } +} + +func waitIdle(rs *rootState, poll, timeout time.Duration) { + deadline := time.Now().Add(timeout) + for time.Now().Before(deadline) { + allIdle := true + rs.mu.RLock() + for _, sess := range rs.sessions { + if sess.Core.Agent().State() != "idle" { + allIdle = false + break + } + } + rs.mu.RUnlock() + if allIdle { + return + } + time.Sleep(poll) + } + rs.cfg.Log.Warn("shutdown: timed out waiting for sessions to become idle, saving anyway") +} + +// --- Persistence --- + +func activeSessionsDir(rs *rootState) string { + return filepath.Join(rs.cfg.SessionsDir, "active") +} + +func persistSession(rs *rootState, id string) { + rs.mu.RLock() + sess, ok := rs.sessions[id] + rs.mu.RUnlock() + if !ok { + return + } + dir := activeSessionsDir(rs) + os.MkdirAll(dir, 0700) + path := filepath.Join(dir, id+".json") + if err := sess.Core.SaveSession(path); err != nil { + rs.cfg.Log.Error("persist session %s: %v", id, err) + } +} + +func removePersistedAgent(rs *rootState, id string) { + path := filepath.Join(activeSessionsDir(rs), id+".json") + os.Remove(path) +} + +func saveAllSessions(rs *rootState) { + rs.mu.RLock() + ids := make([]string, 0, len(rs.sessions)) + for id := range rs.sessions { + ids = append(ids, id) + } + rs.mu.RUnlock() + for _, id := range ids { + persistSession(rs, id) + } +} + +func restoreAllSessions(rs *rootState) { + dir := activeSessionsDir(rs) + entries, err := os.ReadDir(dir) + if err != nil { + return + } + + type loadedSession struct { + ps *agent.PersistedAgent + name string + } + var loaded []loadedSession + for _, e := range entries { + if !strings.HasSuffix(e.Name(), ".json") { + continue + } + path := filepath.Join(dir, e.Name()) + ps, err := agent.LoadPersistedAgent(path) + if err != nil { + rs.cfg.Log.Error("restore session %s: %v", e.Name(), err) + continue + } + loaded = append(loaded, loadedSession{ps: ps, name: e.Name()}) + } + + if len(loaded) == 0 { + return + } + + var wg sync.WaitGroup + for _, ls := range loaded { + wg.Add(1) + go func(ls loadedSession) { + defer wg.Done() + if err := restoreSession(rs, ls.ps); err != nil { + rs.cfg.Log.Error("restore session %s: %v", ls.ps.ID, err) + } + }(ls) + } + wg.Wait() +} + +func restoreSession(rs *rootState, ps *agent.PersistedAgent) error { + cwd := ps.CWD + if cwd == "" { + cwd, _ = os.Getwd() + } + agentName := ps.Agent + if agentName == "" { + agentName = "default" + } + sessID := ps.ID + + cfg := LoadAgentConfig(rs.cfg.AgentsDir, agentName, nil) + + 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) + } + modelName := ps.Model + if modelName == "" && cfg != nil && cfg.Model != "" { + modelName = cfg.Model + } + if modelName == "" { + modelName = os.Getenv("OLLIE_MODEL") + } + if modelName != "" { + be.SetModel(modelName) + } + + uname := rs.nextUname() + var newToolServer func() toolsrv.Runner + var promptEnv []string + remoteTarget := ps.Remote + + if remoteTarget != "" { + rsrv, dialErr := toolsrv.RemoteDial(rs.cfg.Ctx, toolsrv.RemoteConfig{ + SSHTarget: remoteTarget, + CWD: cwd, + }) + if dialErr != nil { + return fmt.Errorf("remote dial: %w", dialErr) + } + newToolServer = func() toolsrv.Runner { return rsrv } + promptEnv = []string{ + "PRIME_CWD=" + cwd, + "PRIME_PLATFORM=" + rsrv.Info.Platform, + "PRIME_IS_GIT_REPO=" + fmt.Sprintf("%v", rsrv.Info.IsGitRepo), + } + } else { + var execOpts []toolsrv.Option + if rs.cfg.Strict { + execOpts = append(execOpts, toolsrv.WithStrict()) + } + if rs.cfg.Yolo { + execOpts = append(execOpts, toolsrv.WithYolo()) + } + if rs.cfg.ToolRegistry != nil { + execOpts = append(execOpts, toolsrv.WithToolRegistry(rs.cfg.ToolRegistry, sessID)) + } + if rs.cfg.SkillsRegistry != nil { + execOpts = append(execOpts, toolsrv.WithSkillsRegistry(rs.cfg.SkillsRegistry)) + execOpts = append(execOpts, toolsrv.WithBuiltins(tools.Builtins())) + } + newToolServer = func() toolsrv.Runner { + srv, _ := toolsrv.Dial(rs.cfg.Ctx, toolsrv.LocalAddr(cwd), execOpts...) + return srv + } + promptEnv = agent.PromptEnv(cwd) + } + + env := []string{"OLLIE_SESSION_ID=" + sessID, "OLLIE_UNAME=" + uname} + env = append(env, promptEnv...) + + var spOverride string + if cfg != nil { + spOverride = cfg.SystemPrompt + } + sysPrompt := prompts.ResolveSystemPrompt(spOverride) + envMap := make(map[string]string) + for _, e := range env { + if k, v, ok := strings.Cut(e, "="); ok { + envMap[k] = v + } + } + opModel := prompts.OperationalModel(rs.cfg.Enable9P, rs.cfg.EnableDBus, envMap) + platform := "linux" + isGitRepo := false + for _, e := range promptEnv { + if k, v, ok := strings.Cut(e, "="); ok { + switch k { + case "PRIME_PLATFORM": + platform = v + case "PRIME_IS_GIT_REPO": + isGitRepo = v == "true" + } + } + } + envBlock := prompts.Environment(cwd, platform, isGitRepo, "") + + toolSrv := newToolServer() + rt := agent.BuildRuntime(cfg, toolSrv, cwd, env, sysPrompt, opModel, envBlock) + + restoredSession := agent.RestoreHistory(ps) + + var sessPtr *Session + core := coresession.New(coresession.Config{ + Backend: be, + AgentName: agentName, + AgentsDir: rs.cfg.AgentsDir, + SessionsDir: rs.cfg.SessionsDir, + SessionID: sessID, + AgentID: uname, + CWD: cwd, + Remote: remoteTarget, + History: restoredSession, + Runtime: rt, + NewToolServer: newToolServer, + PromptEnvExtra: promptEnv, + BaseLayers: []string{sysPrompt, opModel, envBlock}, + Log: rs.cfg.Sink.NewLogger("core"), + ReadPlanStep: func() string { + if sessPtr == nil { + return "" + } + sessPtr.mu.RLock() + data := make([]byte, len(sessPtr.plan)) + copy(data, sessPtr.plan) + sessPtr.mu.RUnlock() + return coresession.NextUncheckedStep(data) + }, + }) + + sessionCtx, sessionCancel := context.WithCancel(rs.cfg.Ctx) + sess := NewSession(sessID, core, sessionCtx, sessionCancel) + sessPtr = sess + sess.uname = uname + sess.remote = remoteTarget + + replayMessagesToLog(sess, ps.Messages) + + rs.mu.Lock() + rs.sessions[sessID] = sess + rs.mu.Unlock() + + rs.cfg.Log.Info("restored session %s (backend=%s model=%s agent=%s)", sessID, backendName, modelName, agentName) + if rs.cfg.OnSessionCreated != nil { + rs.cfg.OnSessionCreated(sessID, sess) + } + return nil +} diff --git a/mgr/manager.go b/session/root.go similarity index 60% rename from mgr/manager.go rename to session/root.go index 7ac683f..8b87455 100644 --- a/mgr/manager.go +++ b/session/root.go @@ -1,23 +1,25 @@ -package mgr +package session import ( - olog "ollie/log" "context" "fmt" - "ollie/paths" - "ollie/session" - "ollie/agent" - "ollie/skills" - "ollie/toolsrv" - "olliesrv/fs" "os" + "sort" "strconv" "strings" "sync" "sync/atomic" + + "ollie/agent" + olog "ollie/log" + "ollie/paths" + coresession "ollie/session" + "ollie/skills" + "ollie/toolsrv" + "olliesrv/fs" ) -var sessionFileOrder = []string{"new", "idx", "ls", "kill", "sh", "b", "bfg", "bbg", "cleanup"} +var rootFileOrder = []string{"new", "idx", "ls", "kill", "sh", "b", "bfg", "bbg", "cleanup"} // FileMode returns the mode for a fixed session file, // or 0 and false if the name is not a fixed file. @@ -26,8 +28,8 @@ func FileMode(name string) (os.FileMode, bool) { return m, ok } -// ManagerConfig holds the dependencies for a Manager. -type ManagerConfig struct { +// Config holds the dependencies for the session tree. +type Config struct { Ctx context.Context // daemon context — sessions derive from this AgentsDir string SessionsDir string @@ -38,7 +40,7 @@ type ManagerConfig struct { MkdirAll func(string, os.FileMode) error // NewCore, if non-nil, replaces the default backend.New + agent.New // path. It receives the session ID, agent name, and cwd, and returns a Session. - NewCore func(sessionID, agentName, cwd string) (*session.Session, error) + NewCore func(sessionID, agentName, cwd string) (*coresession.Session, error) // Strict rejects inline code steps; only tool steps are allowed. Strict bool // Yolo skips the landrun sandbox. @@ -57,91 +59,172 @@ type ManagerConfig struct { ToolRegistry *toolsrv.Registry // SkillsRegistry is the shared skills registry for skill loading. SkillsRegistry *skills.Registry - // OnSessionCreated is called after a new session is added to the manager. - // Receives the session ID and the Session pointer. + // OnSessionCreated is called after a new session is added. OnSessionCreated func(id string, sess *Session) - // OnSessionKilled is called after a session is removed from the manager. + // OnSessionKilled is called after a session is removed. OnSessionKilled func(id string) // OnSessionRenamed is called after a session is renamed. OnSessionRenamed func(oldID, newID string) } -// Manager manages session lifecycle and exposes sessions as a Tree. -type Manager struct { - tree *fs.Tree - cfg ManagerConfig +// rootState is the package-level state stored in root.Data. +type rootState struct { + cfg Config mu sync.RWMutex sessions map[string]*Session - nextUID atomic.Uint32 // incrementing principal counter + nextUID atomic.Uint32 } -// Tree returns the Tree view of the session namespace. -func (s *Manager) Tree() *fs.Tree { return s.tree } - -// nextUname generates the next uname atomically. -func (s *Manager) nextUname() string { - return fmt.Sprintf("%d", s.nextUID.Add(1)) +func (rs *rootState) nextUname() string { + return fmt.Sprintf("%d", rs.nextUID.Add(1)) } -func NewManager(cfg ManagerConfig) *Manager { +// state extracts rootState from a tree's Data field. +func state(root *fs.Tree) *rootState { + return root.Data.(*rootState) +} + +// NewRoot creates the session tree with all callbacks wired up. +// The returned tree IS the session collection. +func NewRoot(cfg Config) *fs.Tree { if cfg.ReadFile == nil { cfg.ReadFile = os.ReadFile } if cfg.MkdirAll == nil { cfg.MkdirAll = os.MkdirAll } - ss := &Manager{ + rs := &rootState{ cfg: cfg, sessions: make(map[string]*Session), } - ss.nextUID.Store(9999) - ss.restoreAllSessions() - ss.tree = fs.NewTree(nil, 0, - fs.WithStat(func(_ []string, name string) (os.FileInfo, error) { return ss.stat(name) }), - fs.WithOpener(func(_ []string, name string) (fs.File, error) { return ss.openEntry(name) }), - fs.WithLister(func(_ []string) ([]os.DirEntry, error) { return ss.list() }), - fs.WithReaddir(func(_ []string, name string) ([]os.DirEntry, error) { return ss.Readdir(name) }), - fs.WithDeleter(func(_ []string, name string) error { return ss.del(name) }), - fs.WithCreator(func(_ []string, name string, _ os.FileMode) error { return ss.create(name) }), - fs.WithRenamer(func(_ []string, old, new string) error { return ss.renameSession(old, new) }), + rs.nextUID.Store(9999) + restoreAllSessions(rs) + tree := fs.NewTree(nil, 0, + fs.WithStat(func(_ []string, name string) (os.FileInfo, error) { return rootStat(rs, name) }), + fs.WithOpener(func(_ []string, name string) (fs.File, error) { return rootOpen(rs, name) }), + fs.WithLister(func(_ []string) ([]os.DirEntry, error) { return rootList(rs) }), + fs.WithReaddir(func(_ []string, name string) ([]os.DirEntry, error) { return rootReaddir(rs, name) }), + fs.WithDeleter(func(_ []string, name string) error { return rootDel(rs, name) }), + fs.WithCreator(func(_ []string, name string, _ os.FileMode) error { return fmt.Errorf("create not supported: %s", name) }), + fs.WithRenamer(func(_ []string, old, new string) error { return Rename(rs, old, new) }), ) - return ss + tree.Data = rs + return tree } -// AddSession inserts a pre-built session into the fs. -func (s *Manager) AddSession(sess *Session) { - s.mu.Lock() - s.sessions[sess.RunnableID()] = sess - s.mu.Unlock() +// Lookup returns the session for the given ID, or nil. +func Lookup(root *fs.Tree, id string) *Session { + rs := state(root) + rs.mu.RLock() + defer rs.mu.RUnlock() + return rs.sessions[id] } -// List returns all root-level entries (fixed files + session directories). -func (s *Manager) List() ([]os.DirEntry, error) { - return s.list() +// AddSession inserts a pre-built session (used by tests). +func AddSession(root *fs.Tree, sess *Session) { + rs := state(root) + rs.mu.Lock() + rs.sessions[sess.RunnableID()] = sess + rs.mu.Unlock() } -func (s *Manager) list() ([]os.DirEntry, error) { - entries := make([]os.DirEntry, 0, len(sessionFileOrder)) - for _, name := range sessionFileOrder { +// LookupByUname returns the session with the given uname (principal), or nil. +func LookupByUname(root *fs.Tree, uname string) *Session { + rs := state(root) + rs.mu.RLock() + defer rs.mu.RUnlock() + for _, sess := range rs.sessions { + if sess.uname == uname { + return sess + } + } + return nil +} + +// InterruptAll interrupts every active session. +func InterruptAll(root *fs.Tree) { + rs := state(root) + rs.mu.RLock() + defer rs.mu.RUnlock() + for _, sess := range rs.sessions { + sess.Core.Agent().Interrupt(agent.ErrInterrupted) + } +} + +// OpenSessionTree returns a Tree for the given session ID. +func OpenSessionTree(root *fs.Tree, id string) (*fs.Tree, error) { + rs := state(root) + sess := lookupLocked(rs, id) + if sess == nil { + return nil, fmt.Errorf("session not found: %s", id) + } + return openSessionTree(rs, sess), nil +} + +func lookupLocked(rs *rootState, id string) *Session { + rs.mu.RLock() + defer rs.mu.RUnlock() + return rs.sessions[id] +} + +func openSessionTree(rs *rootState, sess *Session) *fs.Tree { + var resetElev func() + if rs.cfg.ResetElevation != nil { + id := sess.id + resetElev = func() { rs.cfg.ResetElevation(id) } + } + return NewSessionTree( + sess, + rs.cfg.Log, + func() { Kill(rs, sess.id) }, + func(newID string) error { return Rename(rs, sess.id, newID) }, + nil, + rs.cfg.InvalidateModels, + resetElev, + rs.cfg.ToolRegistry, + ) +} + +func openAgentTree(rs *rootState, sess *Session) *fs.Tree { + var resetElev func() + if rs.cfg.ResetElevation != nil { + id := sess.id + resetElev = func() { rs.cfg.ResetElevation(id) } + } + return NewAgentTree( + sess, + rs.cfg.Log, + nil, + rs.cfg.InvalidateModels, + resetElev, + rs.cfg.ToolRegistry, + ) +} + +// --- Tree callbacks --- + +func rootList(rs *rootState) ([]os.DirEntry, error) { + entries := make([]os.DirEntry, 0, len(rootFileOrder)) + for _, name := range rootFileOrder { entries = append(entries, fs.FileEntry(name, fs.Perms[fs.PathSessions].Files[name])) } - s.mu.RLock() - for id := range s.sessions { + rs.mu.RLock() + for id := range rs.sessions { entries = append(entries, fs.DirEntry(id, fs.Perms[fs.PathSessionDir].DirMode)) } - s.mu.RUnlock() + rs.mu.RUnlock() return entries, nil } -// Readdir lists entries in a subdirectory. -func (s *Manager) Readdir(name string) ([]os.DirEntry, error) { +func rootReaddir(rs *rootState, name string) ([]os.DirEntry, error) { parts := strings.SplitN(name, "/", 4) sessID := parts[0] - sess := s.Session(sessID) + rs.mu.RLock() + sess := rs.sessions[sessID] + rs.mu.RUnlock() if sess == nil { return nil, fmt.Errorf("session not found: %s", sessID) } - // {id} — list session-level files + agent/ if len(parts) == 1 { entries := []os.DirEntry{ fs.FileEntry("plan", 0666), @@ -150,23 +233,17 @@ func (s *Manager) Readdir(name string) ([]os.DirEntry, error) { } return entries, nil } - // {id}/agent — list agent IDs if parts[1] == "agent" { if len(parts) == 2 { - // Currently single-agent: use the agent's ID aid := sess.Core.Agent().ID() if aid == "" { - aid = "0" // fallback for sessions without explicit agent ID + aid = "0" } entries := []os.DirEntry{fs.DirEntry(aid, 0755)} return entries, nil } - // {id}/agent/{aid} — list agent files + proc/ if len(parts) == 3 { - afs, err := s.openAgentTree(sess) - if err != nil { - return nil, err - } + afs := openAgentTree(rs, sess) entries, err := afs.List() if err != nil { return nil, err @@ -174,7 +251,6 @@ func (s *Manager) Readdir(name string) ([]os.DirEntry, error) { entries = append(entries, fs.DirEntry("proc", 0755)) return entries, nil } - // {id}/agent/{aid}/proc — list detached process PIDs if len(parts) == 4 && parts[3] == "proc" { procs := sess.Core.Agent().ListDetached() entries := make([]os.DirEntry, len(procs)) @@ -187,61 +263,50 @@ func (s *Manager) Readdir(name string) ([]os.DirEntry, error) { return nil, fmt.Errorf("%s: not a directory", name) } - -func (s *Manager) stat(name string) (os.FileInfo, error) { - // Top-level fixed files (new, idx, sh, etc.) +func rootStat(rs *rootState, name string) (os.FileInfo, error) { if mode, ok := fs.Perms[fs.PathSessions].Files[name]; ok { return &fs.SyntheticFileInfo{Name_: name, Mode_: mode}, nil } parts := strings.SplitN(name, "/", 4) sessID := parts[0] - s.mu.RLock() - sess, ok := s.sessions[sessID] - s.mu.RUnlock() + rs.mu.RLock() + sess, ok := rs.sessions[sessID] + rs.mu.RUnlock() if !ok { return nil, fmt.Errorf("%s: not found", name) } - // Session directory: {id} if len(parts) == 1 { return &fs.SyntheticFileInfo{Name_: sessID, Mode_: fs.Perms[fs.PathSessionDir].DirMode, IsDir_: true}, nil } - // Agent directory: {id}/agent if parts[1] == "agent" { if len(parts) == 2 { return &fs.SyntheticFileInfo{Name_: "agent", Mode_: 0755, IsDir_: true}, nil } - // Agent instance: {id}/agent/{aid} if len(parts) == 3 { return &fs.SyntheticFileInfo{Name_: parts[2], Mode_: 0755, IsDir_: true}, nil } - // Agent file: {id}/agent/{aid}/{file} if parts[3] == "proc" { return &fs.SyntheticFileInfo{Name_: "proc", Mode_: 0755, IsDir_: true}, nil } - afs, err := s.openAgentTree(sess) - if err != nil { - return nil, err - } + afs := openAgentTree(rs, sess) return afs.Stat(parts[3]) } - // Session file: {id}/{file} (plan, env) - sfs, err := s.openSessionTree(sess) - if err != nil { - return nil, err - } + sfs := openSessionTree(rs, sess) return sfs.Stat(parts[1]) } -func (s *Manager) openEntry(name string) (fs.File, error) { +func rootOpen(rs *rootState, name string) (fs.File, error) { notBlocking := func(context.Context, string) ([]byte, string, error) { return nil, "", fmt.Errorf("blocking read not supported") } parts := strings.SplitN(name, "/", 3) - // Proc file: {id}/proc/{pid} — read returns ring buffer output + // Proc file: {id}/proc/{pid} if len(parts) == 3 && parts[1] == "proc" { sessID := parts[0] - sess := s.Session(sessID) + rs.mu.RLock() + sess := rs.sessions[sessID] + rs.mu.RUnlock() if sess == nil { return nil, fmt.Errorf("%s: not found", name) } @@ -276,7 +341,7 @@ func (s *Manager) openEntry(name string) (fs.File, error) { return []byte("name=\ncwd=\nremote=\nbackend=\nmodel=\nagent=\nmaxTokens=\nmaxCompletionTokens=\ntemperature=\ntopP=\ntopK=\nminP=\ntopA=\nfrequencyPenalty=\npresencePenalty=\nrepetitionPenalty=\nreasoning=\nreasoningEffort=\nincludeReasoning=\nresponseFormat=\nstop=\nverbosity=\n"), nil }, WriteFn: func(data []byte) error { - _, err := s.CreateSession(strings.Fields(strings.TrimSpace(string(data)))) + _, err := Create(rs, strings.Fields(strings.TrimSpace(string(data)))) return err }, BlockingReadFn: notBlocking, @@ -286,7 +351,7 @@ func (s *Manager) openEntry(name string) (fs.File, error) { StatFn: func() (os.FileInfo, error) { return &fs.SyntheticFileInfo{Name_: "idx", Mode_: fs.Perms[fs.PathSessions].Files["idx"]}, nil }, - ReadFn: func() ([]byte, error) { return s.index(), nil }, + ReadFn: func() ([]byte, error) { return buildIndex(rs), nil }, WriteFn: func([]byte) error { return fmt.Errorf("idx: read-only") }, BlockingReadFn: notBlocking, }, nil @@ -297,7 +362,7 @@ func (s *Manager) openEntry(name string) (fs.File, error) { return &fs.SyntheticFileInfo{Name_: name, Mode_: fs.Perms[fs.PathSessions].Files[name]}, nil }, ReadFn: func() ([]byte, error) { - return s.cfg.ReadFile(paths.CfgDir() + "/scripts/session/" + name) + return rs.cfg.ReadFile(paths.CfgDir() + "/scripts/session/" + name) }, WriteFn: func([]byte) error { return fmt.Errorf("%s: not writable", name) }, BlockingReadFn: notBlocking, @@ -307,7 +372,9 @@ func (s *Manager) openEntry(name string) (fs.File, error) { // Hierarchical paths: {id}/{file} or {id}/agent/{aid}/{file} parts = strings.SplitN(name, "/", 4) sessID := parts[0] - sess := s.Session(sessID) + rs.mu.RLock() + sess := rs.sessions[sessID] + rs.mu.RUnlock() if sess == nil { return nil, fmt.Errorf("%s: not found", name) } @@ -315,48 +382,35 @@ func (s *Manager) openEntry(name string) (fs.File, error) { return nil, fmt.Errorf("%s: is a directory", name) } - // Agent file: {id}/agent/{aid}/{file} if parts[1] == "agent" { if len(parts) < 4 { return nil, fmt.Errorf("%s: is a directory", name) } - // parts[2] = aid, parts[3] = file - afs, err := s.openAgentTree(sess) - if err != nil { - return nil, err - } + afs := openAgentTree(rs, sess) return afs.Open(parts[3]) } - // Session file: {id}/{file} - sfs, err := s.openSessionTree(sess) - if err != nil { - return nil, err - } + sfs := openSessionTree(rs, sess) return sfs.Open(parts[1]) } -func (s *Manager) create(name string) error { - return fmt.Errorf("create not supported: %s", name) -} - -func (s *Manager) del(name string) error { +func rootDel(rs *rootState, name string) error { parts := strings.SplitN(name, "/", 3) sessID := parts[0] - // Delete session directory itself. if len(parts) == 1 { - s.mu.RLock() - _, ok := s.sessions[sessID] - s.mu.RUnlock() + rs.mu.RLock() + _, ok := rs.sessions[sessID] + rs.mu.RUnlock() if ok { - s.KillSession(sessID) + Kill(rs, sessID) return nil } return fmt.Errorf("session not found: %s", sessID) } - // Proc dismiss: {id}/proc/{pid} if len(parts) == 3 && parts[1] == "proc" { - sess := s.Session(sessID) + rs.mu.RLock() + sess := rs.sessions[sessID] + rs.mu.RUnlock() if sess == nil { return fmt.Errorf("session not found: %s", sessID) } @@ -369,78 +423,28 @@ func (s *Manager) del(name string) error { } return nil } - // Session files are synthetic; allow rm -r to continue. return nil } -// Session returns the session for the given ID, or nil. -func (s *Manager) Session(id string) *Session { - s.mu.RLock() - defer s.mu.RUnlock() - return s.sessions[id] -} - -// SessionByUname returns the session with the given uname (principal), or nil. -func (s *Manager) SessionByUname(uname string) *Session { - s.mu.RLock() - defer s.mu.RUnlock() - for _, sess := range s.sessions { - if sess.uname == uname { - return sess - } +func buildIndex(rs *rootState) []byte { + var sb strings.Builder + rs.mu.RLock() + ids := make([]string, 0, len(rs.sessions)) + for id := range rs.sessions { + ids = append(ids, id) } - return nil -} - -// OpenSessionTree returns a Tree for the given session ID. -func (s *Manager) OpenSessionTree(id string) (*fs.Tree, error) { - if sess := s.Session(id); sess != nil { - return s.openSessionTree(sess) + sort.Strings(ids) + for _, id := range ids { + sess := rs.sessions[id] + sess.mu.RLock() + st := sess.Core.Agent().State() + cwd := sess.Core.CWD() + be := sess.Core.Agent().BackendName() + model := sess.Core.Agent().ModelName() + ag := sess.Core.Agent().Name() + sess.mu.RUnlock() + fmt.Fprintf(&sb, "%s\t%s\t%s\t%s\t%s\t%s\n", id, st, cwd, be, model, ag) } - return nil, fmt.Errorf("session not found: %s", id) + rs.mu.RUnlock() + return []byte(sb.String()) } - -func (s *Manager) openSessionTree(sess *Session) (*fs.Tree, error) { - var resetElev func() - if s.cfg.ResetElevation != nil { - id := sess.id - resetElev = func() { s.cfg.ResetElevation(id) } - } - return NewSessionTree( - sess, - s.cfg.Log, - func() { s.KillSession(sess.id) }, - func(newID string) error { return s.renameSession(sess.id, newID) }, - nil, // no transcript saving - s.cfg.InvalidateModels, - resetElev, - s.cfg.ToolRegistry, - ), nil -} - -func (s *Manager) openAgentTree(sess *Session) (*fs.Tree, error) { - var resetElev func() - if s.cfg.ResetElevation != nil { - id := sess.id - resetElev = func() { s.cfg.ResetElevation(id) } - } - return NewAgentTree( - sess, - s.cfg.Log, - nil, // no transcript saving - s.cfg.InvalidateModels, - resetElev, - s.cfg.ToolRegistry, - ), nil -} - -// InterruptAll interrupts every active session. -func (s *Manager) InterruptAll() { - s.mu.RLock() - defer s.mu.RUnlock() - for _, sess := range s.sessions { - sess.Core.Agent().Interrupt(agent.ErrInterrupted) - } -} - -// Shutdown kills all active sessions. diff --git a/session/session.go b/session/session.go new file mode 100644 index 0000000..eb99d96 --- /dev/null +++ b/session/session.go @@ -0,0 +1,31 @@ +// Package session builds the session/ file tree for the 9P namespace. +// The tree IS the session collection — sessions are mounted children. +// No manager, no store. Package functions for lifecycle. +package session + +import ( + "olliesrv/fs" +) + +// Get returns the Session mounted at id, or nil. +func Get(root *fs.Tree, id string) *Session { + child := root.Child(id) + if child == nil { + return nil + } + if sess, ok := child.Data.(*Session); ok { + return sess + } + return nil +} + +// All returns all mounted sessions. +func All(root *fs.Tree) []*Session { + var sessions []*Session + for _, child := range root.Children() { + if sess, ok := child.Data.(*Session); ok { + sessions = append(sessions, sess) + } + } + return sessions +} diff --git a/mgr/session.go b/session/types.go similarity index 90% rename from mgr/session.go rename to session/types.go index 4ca9345..7cf95c7 100644 --- a/mgr/session.go +++ b/session/types.go @@ -1,29 +1,29 @@ -package mgr +package session import ( "context" - "sync" "strings" + "sync" "time" - "ollie/session" "ollie/agent" + "ollie/session" ) -// Session holds all state for one agent session. +// Session holds all state for one agent session exposed via 9P. type Session struct { mu sync.RWMutex id string uname string // immutable user principal (numeric UID), set at creation Core *session.Session - SessionCtx context.Context + SessionCtx context.Context cancel context.CancelFunc log []byte logVers uint32 ChatOffset int plan []byte - prevPrompt []byte // last submitted prompt; overwritten on each new submission - remote string // SSH target for remote execution (empty = local) + prevPrompt []byte // last submitted prompt; overwritten on each new submission + remote string // SSH target for remote execution (empty = local) // Cached model list (expensive API call; refreshed every 24h). modelsMu sync.Mutex @@ -70,7 +70,6 @@ func (sess *Session) InvalidateModelsCache() { sess.modelsMu.Unlock() } - func (sess *Session) RunnableID() string { return sess.id } func (sess *Session) Uname() string { return sess.uname } @@ -132,11 +131,8 @@ func (sess *Session) startEventLog() { sess.AppendLog(FormatEvent(ev)) case "tool": - // Tool events: empty content = header-only (stream start), - // non-empty while streaming = chunk, non-empty without streaming = full result. if streamingRole == "tool" { if ev.Content == "" { - // End of stream. sess.AppendLog([]byte("\n")) streamingRole = "" } else { @@ -149,11 +145,9 @@ func (sess *Session) startEventLog() { streamingRole = "" } if ev.Content == "" { - // Stream start: write header, enter streaming mode. sess.AppendLog([]byte("[tool:" + ev.Name + "]\n")) streamingRole = "tool" } else { - // Non-streamed tool result (fast tool, no streaming happened). sess.AppendLog(FormatEvent(ev)) }