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

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

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)

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 <id> <action> [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 <id> <action> [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
 }
 

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
 }