Simplify Go event dispatch: synchronous user handlers, goroutine broadcast

Revert from channel-based approach to simpler model that preserves the
existing synchronous event delivery contract:

- User event handlers are invoked synchronously on the calling goroutine,
  maintaining the implicit contract that events registered during
  CreateSession are available immediately after it returns.
- Broadcast handlers (tool calls, permission requests) are fired in
  separate goroutines via 'go s.handleBroadcastEvent(event)' to avoid
  deadlocking the JSON-RPC readLoop.
- Panics in user handlers are recovered per-handler so one handler's
  failure doesn't prevent others from receiving the event.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: stephentoub

go/session.go

@@ -12,8 +12,6 @@ import (
 	"github.com/github/copilot-sdk/go/rpc"
 )
 
-const eventChannelSize = 1024
-
 type sessionHandler struct {
 	id uint64
 	fn SessionEventHandler
@@ -67,12 +65,6 @@ type Session struct {
 	hooks             *SessionHooks
 	hooksMux          sync.RWMutex
 
-	// eventCh serializes event dispatch. DispatchEvent enqueues; a single
-	// background goroutine (processEvents) dequeues and invokes handlers
-	// one at a time, preserving arrival order and keeping the JSON-RPC
-	// read loop unblocked.
-	eventCh chan SessionEvent
-
 	// RPC provides typed session-scoped RPC methods.
 	RPC *rpc.SessionRpc
 }
@@ -86,17 +78,14 @@ func (s *Session) WorkspacePath() string {
 
 // newSession creates a new session wrapper with the given session ID and client.
 func newSession(sessionID string, client *jsonrpc2.Client, workspacePath string) *Session {
-	s := &Session{
+	return &Session{
 		SessionID:     sessionID,
 		workspacePath: workspacePath,
 		client:        client,
 		handlers:      make([]sessionHandler, 0),
 		toolHandlers:  make(map[string]ToolHandler),
-		eventCh:       make(chan SessionEvent, eventChannelSize),
 		RPC:           rpc.NewSessionRpc(client, sessionID),
 	}
-	go s.processEvents()
-	return s
 }
 
 // Send sends a message to this session and waits for the response.
@@ -446,49 +435,33 @@ func (s *Session) handleHooksInvoke(hookType string, rawInput json.RawMessage) (
 	}
 }
 
-// dispatchEvent enqueues an event for serial dispatch to all registered handlers.
+// dispatchEvent dispatches an event to all registered handlers.
 //
-// This method is non-blocking. The event is placed into an in-memory channel and
-// processed by a single background goroutine (processEvents), which guarantees
-// handlers see events one at a time, in order, without blocking the JSON-RPC
-// read loop.
+// Broadcast work (tool calls, permission requests) is fired in a separate
+// goroutine so it does not block the JSON-RPC read loop (which would deadlock
+// when the handler issues an RPC request). User event handlers are invoked
+// synchronously on the calling goroutine, preserving the existing delivery
+// contract. Panics in user handlers are recovered.
 func (s *Session) dispatchEvent(event SessionEvent) {
-	select {
-	case s.eventCh <- event:
-	default:
-		fmt.Printf("Warning: event %s dropped; channel full or session shutting down\n", event.Type)
-	}
-}
+	// Fire broadcast handlers in a goroutine to avoid deadlocking the readLoop.
+	go s.handleBroadcastEvent(event)
 
-// processEvents is the single-goroutine consumer loop that processes events
-// from the channel. Ensures user event handlers are invoked serially and in
-// FIFO order. Broadcast work (tool calls, permission requests) is fired
-// concurrently so that a stalled handler does not block event delivery.
-func (s *Session) processEvents() {
-	for event := range s.eventCh {
-		// Fire broadcast handlers in a separate goroutine so they don't block
-		// event delivery. This preserves pre-existing fire-and-forget semantics
-		// (important when a secondary client's handler intentionally never
-		// completes), while running off the readLoop to avoid deadlocks.
-		go s.handleBroadcastEvent(event)
-
-		s.handlerMutex.RLock()
-		handlers := make([]SessionEventHandler, 0, len(s.handlers))
-		for _, h := range s.handlers {
-			handlers = append(handlers, h.fn)
-		}
-		s.handlerMutex.RUnlock()
-
-		for _, handler := range handlers {
-			func() {
-				defer func() {
-					if r := recover(); r != nil {
-						fmt.Printf("Error in session event handler: %v\n", r)
-					}
-				}()
-				handler(event)
+	s.handlerMutex.RLock()
+	handlers := make([]SessionEventHandler, 0, len(s.handlers))
+	for _, h := range s.handlers {
+		handlers = append(handlers, h.fn)
+	}
+	s.handlerMutex.RUnlock()
+
+	for _, handler := range handlers {
+		func() {
+			defer func() {
+				if r := recover(); r != nil {
+					fmt.Printf("Error in session event handler: %v\n", r)
+				}
 			}()
-		}
+			handler(event)
+		}()
 	}
 }
 
@@ -669,10 +642,6 @@ func (s *Session) Disconnect() error {
 		return fmt.Errorf("failed to disconnect session: %w", err)
 	}
 
-	// Stop accepting new events. The consumer goroutine will exit naturally
-	// after draining remaining events.
-	close(s.eventCh)
-
 	// Clear handlers
 	s.handlerMutex.Lock()
 	s.handlers = nil

go/session_test.go

@@ -11,20 +11,14 @@ func TestSession_On(t *testing.T) {
 	t.Run("multiple handlers all receive events", func(t *testing.T) {
 		session := &Session{
 			handlers: make([]sessionHandler, 0),
-			eventCh:  make(chan SessionEvent, eventChannelSize),
 		}
-		go session.processEvents()
 
-		var wg sync.WaitGroup
-		wg.Add(3)
 		var received1, received2, received3 bool
-		session.On(func(event SessionEvent) { received1 = true; wg.Done() })
-		session.On(func(event SessionEvent) { received2 = true; wg.Done() })
-		session.On(func(event SessionEvent) { received3 = true; wg.Done() })
+		session.On(func(event SessionEvent) { received1 = true })
+		session.On(func(event SessionEvent) { received2 = true })
+		session.On(func(event SessionEvent) { received3 = true })
 
 		session.dispatchEvent(SessionEvent{Type: "test"})
-		wg.Wait()
-		close(session.eventCh)
 
 		if !received1 || !received2 || !received3 {
 			t.Errorf("Expected all handlers to receive event, got received1=%v, received2=%v, received3=%v",
@@ -35,94 +29,66 @@ func TestSession_On(t *testing.T) {
 	t.Run("unsubscribing one handler does not affect others", func(t *testing.T) {
 		session := &Session{
 			handlers: make([]sessionHandler, 0),
-			eventCh:  make(chan SessionEvent, eventChannelSize),
 		}
-		go session.processEvents()
-
-		var count1, count2, count3 atomic.Int32
-		var wg sync.WaitGroup
 
-		wg.Add(3)
-		session.On(func(event SessionEvent) { count1.Add(1); wg.Done() })
-		unsub2 := session.On(func(event SessionEvent) { count2.Add(1); wg.Done() })
-		session.On(func(event SessionEvent) { count3.Add(1); wg.Done() })
+		var count1, count2, count3 int
+		session.On(func(event SessionEvent) { count1++ })
+		unsub2 := session.On(func(event SessionEvent) { count2++ })
+		session.On(func(event SessionEvent) { count3++ })
 
 		// First event - all handlers receive it
 		session.dispatchEvent(SessionEvent{Type: "test"})
-		wg.Wait()
 
 		// Unsubscribe handler 2
 		unsub2()
 
 		// Second event - only handlers 1 and 3 should receive it
-		wg.Add(2)
 		session.dispatchEvent(SessionEvent{Type: "test"})
-		wg.Wait()
-		close(session.eventCh)
 
-		if count1.Load() != 2 {
-			t.Errorf("Expected handler 1 to receive 2 events, got %d", count1.Load())
+		if count1 != 2 {
+			t.Errorf("Expected handler 1 to receive 2 events, got %d", count1)
 		}
-		if count2.Load() != 1 {
-			t.Errorf("Expected handler 2 to receive 1 event (before unsubscribe), got %d", count2.Load())
+		if count2 != 1 {
+			t.Errorf("Expected handler 2 to receive 1 event (before unsubscribe), got %d", count2)
 		}
-		if count3.Load() != 2 {
-			t.Errorf("Expected handler 3 to receive 2 events, got %d", count3.Load())
+		if count3 != 2 {
+			t.Errorf("Expected handler 3 to receive 2 events, got %d", count3)
 		}
 	})
 
 	t.Run("calling unsubscribe multiple times is safe", func(t *testing.T) {
 		session := &Session{
 			handlers: make([]sessionHandler, 0),
-			eventCh:  make(chan SessionEvent, eventChannelSize),
 		}
-		go session.processEvents()
 
-		var count atomic.Int32
-		var wg sync.WaitGroup
-
-		wg.Add(1)
-		unsub := session.On(func(event SessionEvent) { count.Add(1); wg.Done() })
+		var count int
+		unsub := session.On(func(event SessionEvent) { count++ })
 
 		session.dispatchEvent(SessionEvent{Type: "test"})
-		wg.Wait()
 
 		// Call unsubscribe multiple times - should not panic
 		unsub()
 		unsub()
 		unsub()
 
-		wg.Add(1)
-		// Dispatch to drain; handler won't run since it's unsubscribed.
-		// Use a sentinel handler to signal completion.
-		session.On(func(event SessionEvent) { wg.Done() })
 		session.dispatchEvent(SessionEvent{Type: "test"})
-		wg.Wait()
-		close(session.eventCh)
 
-		if count.Load() != 1 {
-			t.Errorf("Expected handler to receive 1 event, got %d", count.Load())
+		if count != 1 {
+			t.Errorf("Expected handler to receive 1 event, got %d", count)
 		}
 	})
 
 	t.Run("handlers are called in registration order", func(t *testing.T) {
 		session := &Session{
 			handlers: make([]sessionHandler, 0),
-			eventCh:  make(chan SessionEvent, eventChannelSize),
 		}
-		go session.processEvents()
 
-		var mu sync.Mutex
 		var order []int
-		var wg sync.WaitGroup
-		wg.Add(3)
-		session.On(func(event SessionEvent) { mu.Lock(); order = append(order, 1); mu.Unlock(); wg.Done() })
-		session.On(func(event SessionEvent) { mu.Lock(); order = append(order, 2); mu.Unlock(); wg.Done() })
-		session.On(func(event SessionEvent) { mu.Lock(); order = append(order, 3); mu.Unlock(); wg.Done() })
+		session.On(func(event SessionEvent) { order = append(order, 1) })
+		session.On(func(event SessionEvent) { order = append(order, 2) })
+		session.On(func(event SessionEvent) { order = append(order, 3) })
 
 		session.dispatchEvent(SessionEvent{Type: "test"})
-		wg.Wait()
-		close(session.eventCh)
 
 		if len(order) != 3 || order[0] != 1 || order[1] != 2 || order[2] != 3 {
 			t.Errorf("Expected handlers to be called in order [1,2,3], got %v", order)
@@ -158,9 +124,7 @@ func TestSession_On(t *testing.T) {
 	t.Run("events are dispatched serially", func(t *testing.T) {
 		session := &Session{
 			handlers: make([]sessionHandler, 0),
-			eventCh:  make(chan SessionEvent, eventChannelSize),
 		}
-		go session.processEvents()
 
 		var concurrentCount atomic.Int32
 		var maxConcurrent atomic.Int32
@@ -180,20 +144,14 @@ func TestSession_On(t *testing.T) {
 			done.Done()
 		})
 
-		// Dispatch from multiple goroutines to test that the channel
-		// serializes concurrent arrivals.
-		var wg sync.WaitGroup
-		wg.Add(totalEvents)
+		// dispatchEvent invokes user handlers synchronously on the calling
+		// goroutine, so dispatching from a single goroutine guarantees serial
+		// delivery (handlers cannot interleave).
 		for i := 0; i < totalEvents; i++ {
-			go func() {
-				defer wg.Done()
-				session.dispatchEvent(SessionEvent{Type: "test"})
-			}()
+			session.dispatchEvent(SessionEvent{Type: "test"})
 		}
-		wg.Wait()
 
 		done.Wait()
-		close(session.eventCh)
 
 		if max := maxConcurrent.Load(); max != 1 {
 			t.Errorf("Expected max concurrent count of 1, got %d", max)
@@ -203,30 +161,21 @@ func TestSession_On(t *testing.T) {
 	t.Run("handler panic does not halt delivery", func(t *testing.T) {
 		session := &Session{
 			handlers: make([]sessionHandler, 0),
-			eventCh:  make(chan SessionEvent, eventChannelSize),
 		}
-		go session.processEvents()
-
-		var eventCount atomic.Int32
-		var done sync.WaitGroup
-		done.Add(2)
 
+		var eventCount int
 		session.On(func(event SessionEvent) {
-			count := eventCount.Add(1)
-			defer done.Done()
-			if count == 1 {
+			eventCount++
+			if eventCount == 1 {
 				panic("boom")
 			}
 		})
 
 		session.dispatchEvent(SessionEvent{Type: "test"})
 		session.dispatchEvent(SessionEvent{Type: "test"})
 
-		done.Wait()
-		close(session.eventCh)
-
-		if eventCount.Load() != 2 {
-			t.Errorf("Expected 2 events dispatched, got %d", eventCount.Load())
+		if eventCount != 2 {
+			t.Errorf("Expected 2 events dispatched, got %d", eventCount)
 		}
 	})
 }