From 4e817f5de9059a3b18019b0dd713dfa8703e3eb0 Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Mon, 3 Aug 2026 18:48:30 +0200 Subject: [PATCH] replace event ring buffer with pubsub - Use github.com/simonfxr/pubsub for topic-based event routing - Topic format: session.{id}.{action}, agent.{id}.{action} - Wildcard subscriptions: *, session.*, agent.* - PublishEvent routes to exact topic + wildcards - WaitEvent maintains backwards compatibility for eventwait - Add agent.SetOnStateChange callback for state change events - Extend session/idx format with paused, connected, agent list Events now published: - session.{id}.new/kill/pause/resume/rename - agent.{id}.new/kill/state This enables efficient selective subscriptions - GUI can subscribe to specific sessions/agents instead of receiving all events. --- agent/agent.go | 9 +++ fs/agentfiles.go | 2 +- fs/lifecycle.go | 193 +++++++++++++++++++++++++++++------------------ fs/types.go | 2 + 4 files changed, 133 insertions(+), 73 deletions(-) diff --git a/agent/agent.go b/agent/agent.go index b499851..6fa9b1a 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -57,6 +57,7 @@ type Agent struct { saveSession func() // trigger debounced persistence flushSave func() // immediately flush persistence turnError func(ctx context.Context, errType, errMsg string) HookResult // overridable for tests + onStateChange func(agentID, state string) // optional callback for state changes } // Backend returns the active backend from the runtime. @@ -110,6 +111,9 @@ func (ag *Agent) SetState(state string) { ag.state = state ag.stateMu.Unlock() ag.notifyChange() + if ag.onStateChange != nil { + ag.onStateChange(ag.id, state) + } } // Reply returns the agent's last assistant response. @@ -120,6 +124,11 @@ func (ag *Agent) Reply() string { return r } +// SetOnStateChange sets a callback invoked whenever agent state changes. +func (ag *Agent) SetOnStateChange(fn func(agentID, state string)) { + ag.onStateChange = fn +} + // SetReply sets the agent's last response. func (ag *Agent) SetReply(reply string) { ag.stateMu.Lock() diff --git a/fs/agentfiles.go b/fs/agentfiles.go index e3d7e63..5e22847 100644 --- a/fs/agentfiles.go +++ b/fs/agentfiles.go @@ -415,7 +415,7 @@ func handleAgentCtl(a *agent.Agent, s *Session, data []byte) error { case "kill": if s.Core != nil { s.Core.RemoveAgent(a.ID()) - PostEvent("A kill session/" + s.Name() + "/agent/" + a.ID()) + PublishEvent("agent."+a.ID()+".kill", "") } case "stop": a.Interrupt(agent.ErrInterrupted) diff --git a/fs/lifecycle.go b/fs/lifecycle.go index ea7fe46..5dc4c72 100644 --- a/fs/lifecycle.go +++ b/fs/lifecycle.go @@ -10,6 +10,8 @@ import ( "sync" "time" + "github.com/simonfxr/pubsub" + "ollie/agent" "ollie/backend" "ollie/paths" @@ -18,7 +20,7 @@ import ( "ollie/toolsrv" ) -// ── Event ring ────────────────────────────────────────────────── +// ── Event Bus (pubsub) ───────────────────────────────────────── const eventRingCap = 100 @@ -38,74 +40,103 @@ func state(root *Tree) *rootState { return root.Data.(*rootState) } -type eventRing struct { - mu sync.Mutex - cond *sync.Cond - buf []string - cap int - start int - count int - total int64 +var eventBus = pubsub.NewBus() + +// Event represents a published event with topic and payload. +type Event struct { + Topic string + Payload string } -var evRing = func() *eventRing { - e := &eventRing{cap: eventRingCap, buf: make([]string, eventRingCap)} - e.cond = sync.NewCond(&e.mu) - return e -}() - -// PostEvent appends a structured event line and wakes all waiters. -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 +// PublishEvent publishes an event to a specific topic. +// Topic format: "session.{id}.{action}" or "agent.{id}.{action}" +func PublishEvent(topic, payload string) { + ev := Event{Topic: topic, Payload: payload} + eventBus.Publish(topic, ev) + // Also publish to wildcard topics for subscribers who want everything + parts := strings.Split(topic, ".") + if len(parts) >= 1 { + eventBus.Publish(parts[0]+".*", ev) } - evRing.total++ - evRing.mu.Unlock() - evRing.cond.Broadcast() + eventBus.Publish("*", ev) } -// WaitEvent blocks until events are available since the given offset. -func WaitEvent(ctx context.Context, offsetStr string) ([]byte, string, error) { - ctx, cancel := context.WithCancel(ctx) - defer cancel() - context.AfterFunc(ctx, func() { evRing.cond.Broadcast() }) +// SubscribeEvents returns a channel that receives all events matching the topic. +// Use "*" for all events, "session.*" for all session events, etc. +func SubscribeEvents(ctx context.Context, topic string) <-chan Event { + ch := make(chan Event, 64) + sub := eventBus.SubscribeChan(topic, ch, pubsub.KeepOpen) + go func() { + <-ctx.Done() + eventBus.Unsubscribe(sub) + close(ch) + }() + return ch +} - evRing.mu.Lock() - var offset int64 - if offsetStr != "" { - fmt.Sscanf(offsetStr, "%d", &offset) - } else { - offset = evRing.total - } - for evRing.total <= offset { - if ctx.Err() != nil { - evRing.mu.Unlock() - return nil, offsetStr, nil +// WaitEvent blocks until an event is available. Returns the event as "topic payload\n". +// For backwards compatibility with the old eventwait interface. +func WaitEvent(ctx context.Context, _ string) ([]byte, string, error) { + ch := SubscribeEvents(ctx, "*") + select { + case ev, ok := <-ch: + if !ok { + return nil, "", nil } - evRing.cond.Wait() + line := ev.Topic + if ev.Payload != "" { + line += " " + ev.Payload + } + return []byte(line + "\n"), "", nil + case <-ctx.Done(): + return nil, "", nil } - base := evRing.total - int64(evRing.count) - if offset < base { - offset = base +} + +// PostEvent publishes an event. Converts old format to new topic format for compatibility. +// Old: "A agentId state thinking" -> topic: "agent.agentId.state", payload: "thinking" +// Old: "S sessionId pause" -> topic: "session.sessionId.pause", payload: "" +func PostEvent(line string) { + parts := strings.Fields(line) + if len(parts) < 2 { + return } - 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') + switch parts[0] { + case "A": + // Agent event: A [payload...] + if len(parts) >= 3 { + topic := "agent." + parts[1] + "." + parts[2] + payload := "" + if len(parts) > 3 { + payload = strings.Join(parts[3:], " ") + } + PublishEvent(topic, payload) + } + case "S": + // Session event: S [payload...] + if len(parts) >= 3 { + topic := "session." + parts[1] + "." + parts[2] + payload := "" + if len(parts) > 3 { + payload = strings.Join(parts[3:], " ") + } + PublishEvent(topic, payload) + } + default: + // Unknown format, publish as-is to catch-all + PublishEvent("legacy", line) } - next := fmt.Sprintf("%d", evRing.total) - evRing.mu.Unlock() - return []byte(sb.String()), next, nil } // ── buildIndex — session/idx content ──────────────────────────── +// +// Format: one line per session (not per agent) +// Fields (tab-separated): +// name, id, paused, connected, cwd, backend, model, agents +// Where agents is a semicolon-separated list of: agentName:agentId:state +// +// Example: +// myproject\tf60604bf-...\tfalse\ttrue\t/home/user/proj\tkiro\tclaude-opus-4.5\tdev:7c36c9e4:idle;test:abc123:thinking func buildIndex(rs *rootState) []byte { var sb strings.Builder @@ -120,19 +151,33 @@ func buildIndex(rs *rootState) []byte { sess.mu.RLock() name := sess.name sid := sess.id - if sess.Core != nil && sess.Core.AgentCount() > 0 { - for _, ag := range sess.Core.Agents() { - st := ag.State() - cwd := sess.Core.CWD() - be := ag.BackendName() - model := ag.ModelName() - agName := ag.Name() - fmt.Fprintf(&sb, "%s\t%s\t%s\t%s\t%s\t%s\t%s\n", name, st, cwd, be, model, agName, sid) + paused := sess.paused + + // Check connected status + connected := false + if conn := sess.toolsConn; conn != nil { + if pinger, ok := conn.(interface{ Ping() error }); ok { + connected = pinger.Ping() == nil + } else { + connected = true } - } else { - // Session exists but has no agents (or Core is nil) - fmt.Fprintf(&sb, "%s\t\t\t\t\t\t%s\n", name, sid) } + + var cwd, be, model string + var agentList []string + if sess.Core != nil { + cwd = sess.Core.CWD() + for _, ag := range sess.Core.Agents() { + if be == "" { + be = ag.BackendName() + model = ag.ModelName() + } + // Format: name:id:state + agentList = append(agentList, fmt.Sprintf("%s:%s:%s", ag.Name(), ag.ID(), ag.State())) + } + } + agents := strings.Join(agentList, ";") + fmt.Fprintf(&sb, "%s\t%s\t%t\t%t\t%s\t%s\t%s\t%s\n", name, sid, paused, connected, cwd, be, model, agents) sess.mu.RUnlock() } rs.mu.RUnlock() @@ -169,9 +214,9 @@ func Kill(rs *rootState, id string) { removePersistedAgent(rs, sess.id) rs.cfg.Log.Info("killed session %s", id) for _, name := range agentNames { - PostEvent("A kill session/" + id + "/agent/" + name) + PublishEvent("agent."+name+".kill", "") } - PostEvent("S kill session/" + id) + PublishEvent("session."+id+".kill", "") } // Rename renames a session by updating its Name and the sessions map. @@ -202,7 +247,7 @@ func Rename(rs *rootState, old, new string) error { // agent's own autosave does not know the fs-level session display name. persistSession(rs, new) rs.cfg.Log.Info("renamed session %s -> %s", old, new) - PostEvent("S rename session/" + old + " session/" + new) + PublishEvent("session."+sess.id+".rename", old+" "+new) return nil } @@ -392,7 +437,7 @@ func Create(rs *rootState, args []string) (string, error) { rs.mu.Unlock() rs.cfg.Log.Info("created empty session %s (id=%s)", name, sessID) - PostEvent("S new session/" + name) + PublishEvent("session."+sessID+".new", name) return name, nil } @@ -681,6 +726,10 @@ func CreateAgent(rs *rootState, sessName string, args []string) error { FlushSave: flushFn, }) sess.Core.AddAgent(ag) + // Wire up state change events + ag.SetOnStateChange(func(agentID, state string) { + PublishEvent("agent."+agentID+".state", state) + }) sess.mu.Lock() sess.setAgentLogLocked(uname, al) sess.mu.Unlock() @@ -700,7 +749,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()) - PostEvent("A new session/" + sessName + "/agent/" + ag.ID()) + PublishEvent("agent."+ag.ID()+".new", sessName) return nil } diff --git a/fs/types.go b/fs/types.go index 87d5e4b..c3ea580 100644 --- a/fs/types.go +++ b/fs/types.go @@ -273,6 +273,7 @@ func (sess *Session) Pause() error { sess.toolsConn = nil } sess.paused = true + PublishEvent("session."+sess.id+".pause", "") return nil } @@ -294,6 +295,7 @@ func (sess *Session) Resume() error { } sess.toolsConn = conn sess.paused = false + PublishEvent("session."+sess.id+".resume", "") return nil }