diff --git a/cmd/ollie-9p/main.go b/cmd/ollie-9p/main.go index 40584a1..c25bd91 100644 --- a/cmd/ollie-9p/main.go +++ b/cmd/ollie-9p/main.go @@ -149,8 +149,20 @@ func dial() *client.Fsys { } // mountWithAname connects and attaches with a specific aname. +// Address format: "unix!/path" or "/path" for unix sockets, "tcp!host:port" or "host:port" for TCP. func mountWithAname(address, aname string) (*client.Fsys, error) { - conn, err := client.Dial("unix", address) + network := "unix" + addr := address + if strings.HasPrefix(address, "tcp!") { + network = "tcp" + addr = strings.TrimPrefix(address, "tcp!") + } else if strings.HasPrefix(address, "unix!") { + addr = strings.TrimPrefix(address, "unix!") + } else if strings.Contains(address, ":") && !strings.HasPrefix(address, "/") { + // Looks like host:port, assume TCP + network = "tcp" + } + conn, err := client.Dial(network, addr) if err != nil { return nil, err } diff --git a/fs/agentfiles.go b/fs/agentfiles.go index 5e22847..14c45d0 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()) - PublishEvent("agent."+a.ID()+".kill", "") + PublishEvent("session."+s.ID()+".agent."+a.ID()+".kill", "") } case "stop": a.Interrupt(agent.ErrInterrupted) diff --git a/fs/lifecycle.go b/fs/lifecycle.go index 5dc4c72..010c5fd 100644 --- a/fs/lifecycle.go +++ b/fs/lifecycle.go @@ -49,14 +49,26 @@ type Event struct { } // PublishEvent publishes an event to a specific topic. -// Topic format: "session.{id}.{action}" or "agent.{id}.{action}" +// Topic hierarchy: +// session.{sid}.new/kill/pause/resume/rename +// session.{sid}.agent.{aid}.new/kill/state +// +// Wildcard routing publishes to all ancestor wildcards: +// session.{sid}.agent.{aid}.state -> also publishes to: +// session.{sid}.agent.{aid}.* +// session.{sid}.agent.* +// session.{sid}.* +// session.* +// * 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 + + // Publish to hierarchical wildcards parts := strings.Split(topic, ".") - if len(parts) >= 1 { - eventBus.Publish(parts[0]+".*", ev) + for i := len(parts) - 1; i >= 1; i-- { + wildcard := strings.Join(parts[:i], ".") + ".*" + eventBus.Publish(wildcard, ev) } eventBus.Publish("*", ev) } @@ -213,10 +225,10 @@ func Kill(rs *rootState, id string) { } removePersistedAgent(rs, sess.id) rs.cfg.Log.Info("killed session %s", id) - for _, name := range agentNames { - PublishEvent("agent."+name+".kill", "") + for _, agentID := range agentNames { + PublishEvent("session."+sess.id+".agent."+agentID+".kill", "") } - PublishEvent("session."+id+".kill", "") + PublishEvent("session."+sess.id+".kill", "") } // Rename renames a session by updating its Name and the sessions map. @@ -726,9 +738,10 @@ func CreateAgent(rs *rootState, sessName string, args []string) error { FlushSave: flushFn, }) sess.Core.AddAgent(ag) - // Wire up state change events + // Wire up state change events - capture sessID for hierarchical topic + sid := sessID ag.SetOnStateChange(func(agentID, state string) { - PublishEvent("agent."+agentID+".state", state) + PublishEvent("session."+sid+".agent."+agentID+".state", state) }) sess.mu.Lock() sess.setAgentLogLocked(uname, al) @@ -749,7 +762,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()) - PublishEvent("agent."+ag.ID()+".new", sessName) + PublishEvent("session."+sessID+".agent."+ag.ID()+".new", ag.Name()) return nil } diff --git a/fs/types.go b/fs/types.go index c3ea580..b2f1c65 100644 --- a/fs/types.go +++ b/fs/types.go @@ -206,6 +206,7 @@ func (sess *Session) SetDisallowTools(names []string) { } func (sess *Session) RunnableID() string { return sess.id } +func (sess *Session) ID() string { return sess.id } func (sess *Session) Uname() string { return sess.uname } func (sess *Session) Name() string { sess.mu.RLock()