claude: a reply that fills its segment before the request reads it doesn't hold the run's lock (CI Test 37434913069 on 8fbd39f8: TestClaudeLastWordsAreRead hung 20m on macOS -race) — emit sent each event to the segment, a 64-event channel, with r.mu held, and the request calls run.heard, which takes r.mu, before it starts reading the segment. A Claude Code that wrote more than 64 events first (the test's 300 deltas, or a quick long reply) left emit blocked on the send holding the lock and heard waiting on it: the run hung for good, and abort, stderr and every other r.mu user with it. emit now takes the segment and updates shown under r.mu, lets it go, and sends under a lock of its own (sending); the two places that close a segment (endSegment, finish) take that lock first, so a send never meets a closed channel. Events keep their order, emit being serialised by sending. TestEmitOnFullSegmentLeavesTheRunFree (200 events emitted before anyone reads, then heard) fails on the old code (heard waits) and passes; it, TestClaudeLastWordsAreRead, TestClientGoneRecordedAsCanceled and Keepalive pass -race -count=20; the whole gateway package passes -race; go vet, linux(nogui)/windows builds pass. The other sends in the file (answered, the MCP waiter) already send after unlocking.

internal/gateway/claude_emit_test.go

@@ -0,0 +1,37 @@
+package gateway
+
+import (
+	"testing"
+	"time"
+)
+
+// A reply that fills the segment before the request reads it doesn't hold
+// the run: the request's own heard, and so its reading, still go on (CI on
+// 8fbd39f8: TestClaudeLastWordsAreRead hung 20m on macOS -race, emit
+// blocked on a full segment with the run's lock held, heard waiting for it).
+func TestEmitOnFullSegmentLeavesTheRunFree(t *testing.T) {
+	r := &subscriptionRun{}
+	ch := r.attach()
+	const n = 200
+	go func() {
+		for i := 0; i < n; i++ {
+			r.emit(Event{Kind: KText, Text: "x"})
+		}
+		r.endSegment()
+	}()
+	time.Sleep(50 * time.Millisecond) // the segment is full by now
+	heard := make(chan struct{})
+	go func() { r.heard(nil); close(heard) }()
+	select {
+	case <-heard:
+	case <-time.After(5 * time.Second):
+		t.Fatal("heard waited on a send to a segment no one reads yet")
+	}
+	got := 0
+	for range ch {
+		got++
+	}
+	if got != n {
+		t.Fatalf("read %d events of %d", got, n)
+	}
+}

internal/gateway/claude_subscription.go

@@ -133,7 +133,11 @@ type subscriptionRun struct {
 	// the tools its agent was told of, as it started and since
 	tools map[string]bool
 
-	mu       sync.Mutex
+	mu sync.Mutex
+	// sending is held across a send to the segment and taken before it is
+	// closed, so a send blocked on a full segment no one reads yet holds
+	// this, not mu: the request's heard, stderr and abort go on meanwhile
+	sending  sync.Mutex
 	segment  chan Event
 	pending  map[string]chan mcpToolResult
 	closed   bool
@@ -1717,9 +1721,11 @@ func (r *subscriptionRun) attach() chan Event {
 }
 
 func (r *subscriptionRun) emit(ev Event) {
+	r.sending.Lock()
+	defer r.sending.Unlock()
 	r.mu.Lock()
-	defer r.mu.Unlock()
-	if r.segment != nil {
+	ch := r.segment
+	if ch != nil {
 		switch ev.Kind {
 		case KStart: // a reply begins
 			r.shown = nil
@@ -1730,8 +1736,19 @@ func (r *subscriptionRun) emit(ev Event) {
 				r.shown[n-1].args += ev.Text
 			}
 		}
-		r.segment <- ev
 	}
+	r.mu.Unlock()
+	if ch != nil {
+		ch <- ev
+	}
+}
+
+// closeSegment closes a segment taken off the run, once no send to it is
+// under way.
+func (r *subscriptionRun) closeSegment(ch chan Event) {
+	r.sending.Lock()
+	close(ch)
+	r.sending.Unlock()
 }
 
 func (r *subscriptionRun) endSegment() {
@@ -1743,7 +1760,7 @@ func (r *subscriptionRun) endSegment() {
 	}
 	r.mu.Unlock()
 	if ch != nil {
-		close(ch)
+		r.closeSegment(ch)
 	}
 }
 
@@ -2796,7 +2813,7 @@ func (r *subscriptionRun) finish() {
 		close(waiter)
 	}
 	if ch != nil {
-		close(ch)
+		r.closeSegment(ch)
 	}
 	if r.bridge != nil { // a run made in a test may have none
 		r.bridge.removeRun(r)