Changeset 0.11.0 (#62)

server/events/bus.go

@@ -3,6 +3,7 @@ package events
 import (
 	"strings"
 	"sync"
+	"time"
 )
 
 // Bus is a labeled publish/subscribe event bus.
@@ -114,3 +115,28 @@ func match(label, pattern string) bool {
 	prefix := pattern[:len(pattern)-1] // "chat.message." from "chat.message.*"
 	return strings.HasPrefix(label, prefix)
 }
+
+// WaitFor subscribes to an exact label and blocks until an event arrives
+// or the timeout elapses. Returns the event and true, or a zero Event and
+// false on timeout. The subscription is automatically removed after.
+//
+// Used by the browser tool bridge: the server publishes tool.call.{id}
+// to the client and then calls WaitFor("tool.result.{id}", 30s) to block
+// until the browser returns the result.
+func (b *Bus) WaitFor(label string, timeout time.Duration) (Event, bool) {
+	ch := make(chan Event, 1)
+	unsub := b.Subscribe(label, func(e Event) {
+		select {
+		case ch <- e:
+		default:
+		}
+	})
+	defer unsub()
+
+	select {
+	case e := <-ch:
+		return e, true
+	case <-time.After(timeout):
+		return Event{}, false
+	}
+}

server/events/bus_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"sync/atomic"
 	"testing"
+	"time"
 )
 
 func TestMatch(t *testing.T) {
@@ -112,6 +113,12 @@ func TestRouteFor(t *testing.T) {
 		{"unknown.event", DirLocal}, // default
 		{"ping", DirFromClient},
 		{"pong", DirToClient},
+		// Tool bridge routes
+		{"tool.call.abc123", DirToClient},
+		{"tool.result.abc123", DirFromClient},
+		// Extension lifecycle
+		{"extension.loaded", DirLocal},
+		{"extension.error", DirLocal},
 	}
 
 	for _, tt := range tests {
@@ -121,3 +128,84 @@ func TestRouteFor(t *testing.T) {
 		}
 	}
 }
+
+func TestWaitFor_Success(t *testing.T) {
+	bus := NewBus()
+
+	// Publish after a short delay
+	go func() {
+		time.Sleep(10 * time.Millisecond)
+		bus.Publish(Event{
+			Label:   "tool.result.abc123",
+			Payload: json.RawMessage(`{"result":"hello"}`),
+		})
+	}()
+
+	event, ok := bus.WaitFor("tool.result.abc123", 1*time.Second)
+	if !ok {
+		t.Fatal("WaitFor timed out, expected success")
+	}
+	if event.Label != "tool.result.abc123" {
+		t.Errorf("expected label tool.result.abc123, got %s", event.Label)
+	}
+	if string(event.Payload) != `{"result":"hello"}` {
+		t.Errorf("unexpected payload: %s", string(event.Payload))
+	}
+}
+
+func TestWaitFor_Timeout(t *testing.T) {
+	bus := NewBus()
+
+	_, ok := bus.WaitFor("tool.result.never", 50*time.Millisecond)
+	if ok {
+		t.Error("WaitFor should have timed out")
+	}
+}
+
+func TestWaitFor_IgnoresOtherLabels(t *testing.T) {
+	bus := NewBus()
+
+	go func() {
+		time.Sleep(5 * time.Millisecond)
+		bus.Publish(Event{Label: "tool.result.other", Payload: json.RawMessage(`{}`)})
+		time.Sleep(5 * time.Millisecond)
+		bus.Publish(Event{Label: "tool.result.target", Payload: json.RawMessage(`{"ok":true}`)})
+	}()
+
+	event, ok := bus.WaitFor("tool.result.target", 1*time.Second)
+	if !ok {
+		t.Fatal("WaitFor timed out")
+	}
+	if event.Label != "tool.result.target" {
+		t.Errorf("got wrong label: %s", event.Label)
+	}
+}
+
+func TestWaitFor_CleansUpSubscription(t *testing.T) {
+	bus := NewBus()
+
+	// WaitFor with immediate timeout
+	bus.WaitFor("tool.result.cleanup", 1*time.Millisecond)
+	time.Sleep(5 * time.Millisecond)
+
+	// Verify no panic when publishing to the label after cleanup
+	bus.Publish(Event{Label: "tool.result.cleanup", Payload: json.RawMessage(`{}`)})
+}
+
+func TestToolCallRouteToClient(t *testing.T) {
+	if !ShouldSendToClient("tool.call.abc123") {
+		t.Error("tool.call.* should be sent to client")
+	}
+	if ShouldAcceptFromClient("tool.call.abc123") {
+		t.Error("tool.call.* should NOT be accepted from client")
+	}
+}
+
+func TestToolResultRouteFromClient(t *testing.T) {
+	if !ShouldAcceptFromClient("tool.result.abc123") {
+		t.Error("tool.result.* should be accepted from client")
+	}
+	if ShouldSendToClient("tool.result.abc123") {
+		t.Error("tool.result.* should NOT be sent to client")
+	}
+}

server/events/types.go

@@ -58,6 +58,14 @@ var routeTable = map[string]Direction{
 	"internal.":         DirLocal,
 	"db.":               DirLocal,
 
+	// Tool execution (browser tools bridge)
+	"tool.call.":    DirToClient,    // Server → specific client (browser tool invocation)
+	"tool.result.":  DirFromClient,  // Client → server (browser tool result)
+
+	// Extension lifecycle
+	"extension.loaded":  DirLocal,   // Client-only
+	"extension.error":   DirLocal,
+
 	// Heartbeat
 	"ping":  DirFromClient,
 	"pong":  DirToClient,

server/events/ws.go

@@ -118,6 +118,42 @@ func (h *Hub) ConnsByUser(userID string) []*Conn {
 	return result
 }
 
+// IsConnected returns true if the user has at least one active WebSocket.
+func (h *Hub) IsConnected(userID string) bool {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	for _, c := range h.conns {
+		if c.userID == userID {
+			return true
+		}
+	}
+	return false
+}
+
+// SendToUser sends an event directly to all WebSocket connections for a user,
+// bypassing room filtering. Used for targeted tool.call delivery.
+func (h *Hub) SendToUser(userID string, event Event) {
+	data, err := json.Marshal(event)
+	if err != nil {
+		return
+	}
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	for _, c := range h.conns {
+		if c.userID == userID {
+			select {
+			case c.send <- data:
+			default:
+			}
+		}
+	}
+}
+
+// GetBus returns the underlying event bus.
+func (h *Hub) GetBus() *Bus {
+	return h.bus
+}
+
 // removeConn cleans up a connection.
 func (h *Hub) removeConn(conn *Conn) {
 	h.mu.Lock()