diff --git a/cmd/olliesrv/main.go b/cmd/olliesrv/main.go index de0b64f..ecbfcb0 100644 --- a/cmd/olliesrv/main.go +++ b/cmd/olliesrv/main.go @@ -377,11 +377,12 @@ func NewRootStore(reg *toolsrv.Registry) *fs.Tree { } return "", fmt.Errorf("%s: not found", name) }), - fs.WithLister(func(_ []string) ([]os.DirEntry, error) { +fs.WithLister(func(_ []string) ([]os.DirEntry, error) { return []os.DirEntry{ fs.FileEntry("backends", fs.Perms[fs.PathRoot].FileMode), fs.FileEntry("ctl", 0666), fs.DirEntry("elevate", 0755), + fs.FileEntry("eventwait", fs.Perms[fs.PathRoot].FileMode), fs.FileEntry("generate", 0666), fs.FileEntry("models", fs.Perms[fs.PathRoot].FileMode), fs.FileEntry("help", fs.Perms[fs.PathRoot].FileMode), @@ -389,6 +390,19 @@ func NewRootStore(reg *toolsrv.Registry) *fs.Tree { }, nil }), fs.WithOpener(func(_ []string, name string) (fs.File, error) { + if name == "eventwait" { + return &fs.FileConfig{ + StatFn: func() (os.FileInfo, error) { + return &fs.SyntheticFileInfo{Name_: "eventwait", Mode_: fs.Perms[fs.PathRoot].Files["eventwait"]}, nil + }, + ReadFn: func() ([]byte, error) { + return nil, fmt.Errorf("eventwait: use blocking read") + }, + WriteFn: readOnly, + BlockingReadFn: fs.WaitEvent, + IsBlocking_: true, + }, nil + } if name == "complete" { return &fs.FileConfig{ StatFn: func() (os.FileInfo, error) { return &fs.SyntheticFileInfo{Name_: "complete", Mode_: 0666}, nil }, diff --git a/fs/session/create.go b/fs/session/create.go index 064cd29..a620f08 100644 --- a/fs/session/create.go +++ b/fs/session/create.go @@ -59,7 +59,7 @@ func Create(rs *rootState, args []string) (string, error) { rs.mu.Unlock() rs.cfg.Log.Info("created empty session %s (id=%s)", name, sessID) - broadcastEvents() + PostEvent("S new session/" + name) return name, nil } @@ -323,7 +323,7 @@ func CreateAgent(rs *rootState, sessName string, args []string) error { rs.cfg.Log.Info("created agent in session %s (backend=%s model=%s agent=%s)", sessName, ag.BackendName(), ag.ModelName(), ag.Name()) - broadcastEvents() + PostEvent("A new session/" + sessName + "/agent/" + ag.Name()) return nil } diff --git a/fs/session/perm.go b/fs/session/perm.go index a7f26c7..6f00c32 100644 --- a/fs/session/perm.go +++ b/fs/session/perm.go @@ -41,8 +41,11 @@ var Perms = map[string]Perm{ DirMode: 0755, FileMode: 0444, Files: map[string]os.FileMode{ - "backends": 0444, - "help": 0444, + "backends": 0444, + "help": 0444, + "models": 0444, + "agents": 0444, + "eventwait": 0444, }, }, PathElevate: { @@ -57,9 +60,8 @@ var Perms = map[string]Perm{ DirMode: 0755, FileMode: 0444, Files: map[string]os.FileMode{ - "new": 0666, - "idx": 0444, - "eventwait": 0444, + "new": 0666, + "idx": 0444, }, }, PathSessionDir: {DirMode: 0755}, diff --git a/fs/session/persist.go b/fs/session/persist.go index 3c2d7ca..7464824 100644 --- a/fs/session/persist.go +++ b/fs/session/persist.go @@ -23,17 +23,25 @@ func Kill(rs *rootState, id string) { delete(rs.sessions, id) rs.mu.Unlock() if sess != nil { - sess.Cancel() + // Capture agent names before cleanup for events. + var agentNames []string if sess.Core != nil { + for _, ag := range sess.Core.Agents() { + agentNames = append(agentNames, ag.Name()) + } sess.Core.Close() } + sess.Cancel() if sess.proc != nil { sess.proc.Close() } removePersistedAgent(rs, sess.id) rs.cfg.Log.Info("killed session %s", id) + for _, name := range agentNames { + PostEvent("A kill session/" + id + "/agent/" + name) + } + PostEvent("S kill session/" + id) } - broadcastEvents() } // KillFromRoot is the public API for callers with *Tree. @@ -65,7 +73,7 @@ func Rename(rs *rootState, old, new string) error { rs.mu.Unlock() rs.cfg.Log.Info("renamed session %s -> %s", old, new) - broadcastEvents() + PostEvent("S rename session/" + old + " session/" + new) return nil } diff --git a/fs/session/root.go b/fs/session/root.go index 3d3564f..9fa8d53 100644 --- a/fs/session/root.go +++ b/fs/session/root.go @@ -17,45 +17,81 @@ import ( "ollie/toolsrv" ) -// eventBroadcaster manages notification of session index changes. -// The GUI can tail session/eventwait to detect creates, deletes, and renames. -var ( - evMu sync.Mutex - evChans []chan struct{} -) +// Event ring buffer — global lifecycle events for session/agent changes. +// Consumers read from /eventwait (blocking, offset-based). +const eventRingCap = 100 -func broadcastEvents() { - evMu.Lock() - for _, ch := range evChans { - select { - case ch <- struct{}{}: - default: - } +type eventRing struct { + mu sync.Mutex + cond sync.Cond + buf []string // circular: buf[start%cap]..buf[(start+count-1)%cap] + cap int + start int + count int + total int64 // monotonic counter; doubles as stable offset +} + +var evRing = func() eventRing { + e := eventRing{cap: eventRingCap, buf: make([]string, eventRingCap)} + e.cond.L = &e.mu + return e +}() + +// PostEvent appends a structured event line and wakes all waiters. +// Event format: " " e.g. "S new session/foo" +func PostEvent(line string) { + evRing.mu.Lock() + idx := (evRing.start + evRing.count) % evRing.cap + evRing.buf[idx] = line + if evRing.count < evRing.cap { + evRing.count++ + } else { + evRing.start = (evRing.start + 1) % evRing.cap } - evMu.Unlock() + evRing.total++ + evRing.mu.Unlock() + evRing.cond.Broadcast() } -// eventRegister returns a channel that fires on the next session index change. -func eventRegister() chan struct{} { - ch := make(chan struct{}, 1) - evMu.Lock() - evChans = append(evChans, ch) - evMu.Unlock() - return ch -} +// WaitEvent blocks until events are available since the given offset (string-encoded int64). +// Returns new event lines and the updated offset for the next call. +func WaitEvent(ctx context.Context, offsetStr string) ([]byte, string, error) { + evRing.mu.Lock() + defer evRing.mu.Unlock() -func eventRelease(ch chan struct{}) { - evMu.Lock() - for i, c := range evChans { - if c == ch { - evChans = append(evChans[:i], evChans[i+1:]...) - break - } + var offset int64 + if offsetStr != "" { + fmt.Sscanf(offsetStr, "%d", &offset) + } else { + offset = evRing.total } - evMu.Unlock() + + // Block until new events or context cancelled. + for evRing.total <= offset { + if ctx.Err() != nil { + return nil, "", ctx.Err() + } + evRing.cond.Wait() + } + + // Calculate the index range for events after `offset`. + // offset is the total count at last read. We need events + // from position (evRing.total - evRing.count) to evRing.total. + base := evRing.total - int64(evRing.count) + if offset < base { + offset = base // consumer fell behind — give them everything we have + } + skip := int(offset - base) + var sb strings.Builder + for i := 0; i < evRing.count-skip; i++ { + idx := (evRing.start + skip + i) % evRing.cap + sb.WriteString(evRing.buf[idx]) + sb.WriteByte('\n') + } + return []byte(sb.String()), fmt.Sprintf("%d", evRing.total), nil } -var rootFileOrder = []string{"new", "idx", "eventwait"} +var rootFileOrder = []string{"new", "idx"} // FileMode returns the mode for a fixed session file, // or 0 and false if the name is not a fixed file. @@ -157,9 +193,9 @@ func NewRoot(cfg Config) *Tree { } else { sess.Core.Agent().SetName(new) } - rs.cfg.Log.Info("renamed agent in session %s -> %s", sessName, new) - broadcastEvents() - return nil + rs.cfg.Log.Info("renamed agent in session %s -> %s", sessName, new) + PostEvent("A rename session/" + sessName + "/agent/" + aid + " session/" + sessName + "/agent/" + new) + return nil } return Rename(rs, old, new) }), @@ -426,30 +462,9 @@ func rootOpen(rs *rootState, name string) (File, error) { }, nil } - switch name { - case "eventwait": - return &FileConfig{ - StatFn: func() (os.FileInfo, error) { - return &SyntheticFileInfo{Name_: "eventwait", Mode_: Perms[PathSessions].Files["eventwait"]}, nil - }, - ReadFn: func() ([]byte, error) { - return []byte(buildIndex(rs)), nil - }, - WriteFn: func([]byte) error { return fmt.Errorf("eventwait: read-only") }, - BlockingReadFn: func(ctx context.Context, _ string) ([]byte, string, error) { - ch := eventRegister() - defer eventRelease(ch) - select { - case <-ch: - return []byte(buildIndex(rs)), "", nil - case <-ctx.Done(): - return nil, "", ctx.Err() - } - }, - IsBlocking_: true, - }, nil - case "new": - return &FileConfig{ + switch name { + case "new": + return &FileConfig{ StatFn: func() (os.FileInfo, error) { return &SyntheticFileInfo{Name_: "new", Mode_: Perms[PathSessions].Files["new"]}, nil }, diff --git a/kde b/kde index cd7bb9e..bfac41b 160000 --- a/kde +++ b/kde @@ -1 +1 @@ -Subproject commit cd7bb9e272abe857d78fd984c84d90f475b17b1f +Subproject commit bfac41b2aec18abe8808fd710587c93d58ed309b