lkn
/
ollie-9p
Archived
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

kill mgr/ package: replace Manager with session/ package functions 

The session tree IS the session collection. No Manager struct —
rootState is unexported, stored in tree.Data. All CRUD exposed
as package functions (NewRoot, Lookup, Create, Kill, Rename, etc).

- mgr/ → session/ (package rename + Manager elimination)
- server.Config.SessionMgr → SessionTree (*fs.Tree)
- dbus adapter takes *fs.Tree directly
- Tests pass, build + vet clean
This commit is contained in:
Levi Neely 2026-07-30 00:02:36 +02:00
parent 525da67b31
commit fee8d1bf36
14 changed files with 805 additions and 831 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
9p
View File

Binary file not shown.

101
dbus.go
View File

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

5
fs/tree.go
View File

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

12
main.go
View File

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

210
main_test.go
View File

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

348
mgr/persist.go
View File

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

24
server/server.go
View File

@ -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":

149
mgr/create.go → session/create.go
View File

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

2
mgr/sessionfile.go → session/files.go
View File

@ -1,4 +1,4 @@
package mgr
package session
import (
"context"

2
mgr/format.go → session/format.go
View File

@ -1,4 +1,4 @@
package mgr
package session
import (
"ollie/agent"

360
session/persist.go Normal file
View File

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

372
mgr/manager.go → session/root.go
View File

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

31
session/session.go Normal file
View File

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

20
mgr/session.go → session/types.go
View File

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