fix(runtime): deliver StreamStopped with a bounded blocking send

pkg/a2a/adapter.go

@@ -135,6 +135,21 @@ func runDockerAgent(ctx agent.InvocationContext, t *team.Team, agentName string,
 		// Track accumulated content for chunked responses
 		var contentBuilder strings.Builder
 
+		// finalEvent builds the turn-complete ADK event from whatever content
+		// was accumulated so far. Shared by the StreamStoppedEvent case and the
+		// post-loop fallback below, so both paths build an identical event.
+		finalEvent := func() *adksession.Event {
+			return &adksession.Event{
+				Author: agentName,
+				LLMResponse: model.LLMResponse{
+					Content:      genai.NewContentFromParts([]*genai.Part{{Text: contentBuilder.String()}}, genai.RoleModel),
+					Partial:      false,
+					TurnComplete: true,
+					FinishReason: genai.FinishReasonStop,
+				},
+			}
+		}
+
 		// Convert docker agent events to ADK events and yield them
 
 		for event := range eventsChan {
@@ -171,20 +186,19 @@ func runDockerAgent(ctx agent.InvocationContext, t *team.Team, agentName string,
 			case *runtime.StreamStoppedEvent:
 				// Send final complete event with all accumulated content
 				if contentBuilder.Len() > 0 {
-					finalEvent := &adksession.Event{
-						Author: agentName,
-						LLMResponse: model.LLMResponse{
-							Content:      genai.NewContentFromParts([]*genai.Part{{Text: contentBuilder.String()}}, genai.RoleModel),
-							Partial:      false,
-							TurnComplete: true,
-							FinishReason: genai.FinishReasonStop,
-						},
-					}
-					yield(finalEvent, nil)
+					yield(finalEvent(), nil)
 					return
 				}
 			}
 		}
+
+		// The channel closed without a StreamStoppedEvent: the runtime bounds
+		// how long it waits to deliver that event (#4136), but a consumer that
+		// abandoned the channel or an unexpected close should still complete
+		// the ADK turn rather than leave it hanging.
+		if contentBuilder.Len() > 0 {
+			yield(finalEvent(), nil)
+		}
 	}
 }
 

pkg/runtime/defaults.go

@@ -46,6 +46,15 @@ const defaultMaxOverflowCompactions = 1
 // allowed to wedge the caller indefinitely.
 const toolsChangedTimeout = 5 * time.Second
 
+// defaultStreamStoppedDeliveryTimeout bounds how long finalizeEventChannel
+// waits to deliver StreamStopped when the event buffer is full. 128 buffered
+// events at roughly a millisecond each for a synchronous per-delta SQLite
+// write (the common back-pressure source, see #4136) drain in ~130ms; 5s is
+// comfortably above that for any consumer still reading, while still bounding
+// teardown once a consumer has abandoned the channel (the #3070 deadlock this
+// replaces). Tests shrink it via the streamStoppedDeliveryTimeout field.
+const defaultStreamStoppedDeliveryTimeout = 5 * time.Second
+
 // defaultToolListTimeout bounds how long EmitStartupInfo waits for a single
 // toolset to enumerate its tools while populating the sidebar. A toolset
 // whose Tools() blocks indefinitely — e.g. an MCP stdio subprocess that

pkg/runtime/elicitation_test.go

@@ -209,35 +209,47 @@ func TestLocalRuntime_FinalizeEventChannelEmitsStreamStoppedOnce(t *testing.T) {
 	assert.Equal(t, 1, stopped, "StreamStopped should be emitted exactly once")
 }
 
-func TestLocalRuntime_FinalizeEventChannelDoesNotDeadlockWhenBufferFullAndConsumerGone(t *testing.T) {
+// TestLocalRuntime_FinalizeEventChannelDropsStreamStoppedAfterBoundedTimeout
+// pins the #4136 fix's bound: the StreamStopped send waits for the
+// abandoned-consumer timeout (proving it is a bounded blocking send, not the
+// old plain non-blocking one) but still returns — and still drops the event
+// — rather than hanging forever when nothing ever drains the buffer.
+func TestLocalRuntime_FinalizeEventChannelDropsStreamStoppedAfterBoundedTimeout(t *testing.T) {
 	t.Parallel()
 
 	rt := newElicitationTestRuntime(t)
+	const timeout = 50 * time.Millisecond
+	rt.streamStoppedDeliveryTimeout = timeout
 	sess := session.New()
 	events := make(chan Event, 1)
 	parent := make(chan Event, 1)
 	events <- Error("buffer already full")
 	rt.elicitation.swap(events)
 
 	done := make(chan struct{})
+	start := time.Now()
 	go func() {
 		rt.finalizeEventChannel(t.Context(), sess, turnEndReasonNormal, parent, events)
 		close(done)
 	}()
 
 	select {
 	case <-done:
-	case <-time.After(time.Second):
+	case <-time.After(timeout + 2*time.Second):
 		t.Fatal("finalizeEventChannel deadlocked with a full buffer and no consumer")
 	}
+	elapsed := time.Since(start)
+
+	assert.GreaterOrEqual(t, elapsed, timeout, "the send should wait out the full deadline before giving up")
+	assert.Less(t, elapsed, timeout+2*time.Second, "finalizeEventChannel should return shortly after the deadline, not hang")
 
 	var stopped int
 	for ev := range events {
 		if _, ok := ev.(*StreamStoppedEvent); ok {
 			stopped++
 		}
 	}
-	assert.Zero(t, stopped, "StreamStopped should be dropped instead of blocking when the buffer is full")
+	assert.Zero(t, stopped, "StreamStopped should be dropped instead of blocking forever when the buffer is full and abandoned")
 }
 
 // TestLocalRuntime_FinalizeEventChannelStreamStoppedIsLastBeforeClose pins the

pkg/runtime/event.go

@@ -586,11 +586,13 @@ func (e *SessionCompactionEvent) GetSessionID() string { return e.SessionID }
 // the turnEndReason* classification (normal, error, canceled) so consumers can
 // tell successful completion apart from crashes and user-initiated stops.
 //
-// Delivery is best-effort: it is emitted non-blockingly during teardown and is
-// dropped if the events buffer is full and the consumer has gone away (see
-// finalizeEventChannel). Treat the channel close, not receipt of this event, as
-// the guaranteed terminal signal. Do not assume session-end hooks have finished
-// when this event arrives: it is emitted before they run.
+// Delivery is bounded-blocking: finalizeEventChannel waits (up to a deadline)
+// for the consumer to accept it, so any consumer still draining the channel
+// reliably receives it, and it is dropped only once the consumer has
+// abandoned the channel entirely (#4136). Still, treat the channel close, not
+// receipt of this event, as the guaranteed terminal signal — it remains the
+// only delivery that can never be dropped. Do not assume session-end hooks
+// have finished when this event arrives: it is emitted before they run.
 type StreamStoppedEvent struct {
 	AgentContext
 

pkg/runtime/event_sink.go

@@ -1,5 +1,10 @@
 package runtime
 
+import (
+	"log/slog"
+	"time"
+)
+
 // EventSink is the write side of the runtime's event stream. Methods
 // that produce events accept an EventSink instead of a raw chan Event,
 // decoupling event producers from the channel implementation.
@@ -67,6 +72,41 @@ func (s nonBlockingChannelSink) Emit(e Event) {
 	}
 }
 
+// boundedChannelSink wraps an event channel with a bounded-blocking send:
+// it waits up to wait for the consumer to accept the event, then drops it.
+// Use this for turn-boundary events that a live, still-draining consumer
+// must reliably receive, but that must not hang teardown indefinitely when
+// the consumer has genuinely gone away (#3070, #4136). Regular runtime code
+// should use the unbounded blocking [channelSink] instead, so ordinary
+// back-pressure is preserved.
+type boundedChannelSink struct {
+	ch   chan Event
+	wait time.Duration
+}
+
+// bounded returns a bounded-blocking sink for sink with the given deadline.
+// If sink wraps a channel directly, the result writes to that channel with a
+// timeout; otherwise, sink is returned unchanged because non-channel sinks
+// (notably [EventSinkFunc] used in tests) do not have an underlying buffer
+// that can fill up.
+func bounded(sink EventSink, wait time.Duration) EventSink {
+	if cs, ok := sink.(*channelSink); ok {
+		return boundedChannelSink{ch: cs.ch, wait: wait}
+	}
+	return sink
+}
+
+func (s boundedChannelSink) Emit(e Event) {
+	defer func() { recover() }() //nolint:errcheck // swallow send-on-closed-channel panic
+	t := time.NewTimer(s.wait)
+	defer t.Stop()
+	select {
+	case s.ch <- e:
+	case <-t.C:
+		slog.Warn("dropping event: consumer did not drain within the delivery deadline", "wait", s.wait)
+	}
+}
+
 // EventSinkFunc adapts a plain function into an [EventSink], following
 // the same adapter pattern as http.HandlerFunc. This is convenient for
 // tests and one-off callbacks that don't need a full struct.

pkg/runtime/loop.go

@@ -193,22 +193,27 @@ func (r *LocalRuntime) emitHookDrivenShutdown(
 // "all cleanup done" signal is the channel close (done last, in
 // restoreAndClose) that terminates a `for range`, not the StreamStopped event.
 //
-// Delivery: StreamStopped is best-effort. It is emitted non-blockingly and is
-// dropped when the buffer is full and the consumer has gone away, rather than
-// blocking teardown (a blocking send here is the deadlock #3070 fixed).
-// Consumers must rely on the channel close, not on receiving StreamStopped, as
-// the guaranteed terminal signal.
+// Delivery: StreamStopped is delivered with a bounded blocking send (see
+// [boundedChannelSink]) rather than a plain non-blocking one. A consumer
+// still draining the channel — including one that already got Esc or
+// otherwise cancelled its context, since the TUI keeps draining until close —
+// reliably receives it. It is dropped only if nothing accepts it within
+// streamStoppedTimeout, i.e. the consumer has genuinely abandoned the channel
+// (#4136 fixed the non-blocking drop that caused this; #3070 is why the send
+// is bounded rather than unbounded). Consumers must still rely on the channel
+// close, not on receiving StreamStopped, as the one guaranteed terminal
+// signal.
 func (r *LocalRuntime) finalizeEventChannel(ctx context.Context, sess *session.Session, reason string, prevElicitationCh, events chan Event) {
 	a := r.resolveSessionAgent(sess)
 
 	if ctx.Err() != nil && reason == "" {
 		reason = turnEndReasonCanceled
 	}
 
-	// Best-effort, non-blocking on purpose: a blocking send here reintroduces
-	// the #3070 teardown deadlock. See the doc comment for the ordering and
-	// delivery contract.
-	nonBlocking(&channelSink{ch: events}).Emit(StreamStopped(sess.ID, a.Name(), reason))
+	// Bounded, not unbounded: an abandoned consumer must not hang teardown
+	// forever (#3070), but a live one draining past cancellation must still
+	// get this event (#4136) — so this never selects on ctx.Done().
+	bounded(&channelSink{ch: events}, r.streamStoppedTimeout()).Emit(StreamStopped(sess.ID, a.Name(), reason))
 
 	// Execute session end hooks with a context that won't be cancelled so
 	// cleanup hooks run even when the stream was interrupted (e.g. Ctrl+C).
@@ -227,6 +232,17 @@ func (r *LocalRuntime) finalizeEventChannel(ctx context.Context, sess *session.S
 	r.elicitation.restoreAndClose(events, prevElicitationCh)
 }
 
+// streamStoppedTimeout returns the bounded-delivery deadline for the
+// StreamStopped emit in finalizeEventChannel, falling back to
+// defaultStreamStoppedDeliveryTimeout when unset (e.g. a *LocalRuntime built
+// directly as a struct literal in tests, bypassing NewLocalRuntime).
+func (r *LocalRuntime) streamStoppedTimeout() time.Duration {
+	if r.streamStoppedDeliveryTimeout > 0 {
+		return r.streamStoppedDeliveryTimeout
+	}
+	return defaultStreamStoppedDeliveryTimeout
+}
+
 // RunStream starts the agent's interaction loop and returns a channel of events.
 // The returned channel is closed when the loop terminates (success, error, or
 // context cancellation). Each iteration: sends messages to the model, streams

pkg/runtime/runtime.go

@@ -402,6 +402,14 @@ type LocalRuntime struct {
 	// Defaults to defaultToolStartTimeout; overridden via WithToolStartTimeout.
 	toolStartTimeout time.Duration
 
+	// streamStoppedDeliveryTimeout bounds how long finalizeEventChannel
+	// blocks trying to deliver StreamStopped when the event buffer is full.
+	// Defaults to defaultStreamStoppedDeliveryTimeout; tests shrink it to
+	// exercise the drop-when-abandoned path without a real-time wait. Zero
+	// falls back to the default (see streamStoppedTimeout), so runtimes built
+	// directly via a struct literal in tests still get bounded delivery.
+	streamStoppedDeliveryTimeout time.Duration
+
 	// pauseMu guards pauseCh.
 	pauseMu sync.Mutex
 	// pauseCh is non-nil and open while /pause has paused the run loop;
@@ -678,25 +686,26 @@ func NewLocalRuntime(ctx context.Context, agents *team.Team, opts ...Opt) (*Loca
 	}
 
 	r := &LocalRuntime{
-		ctx:                    func() context.Context { return context.WithoutCancel(ctx) },
-		toolMap:                make(map[string]ToolHandlerFunc),
-		liveSessions:           make(map[string]*liveSessionEntry),
-		team:                   agents,
-		agents:                 newAgentRouter(agents, defaultAgent.Name()),
-		resumeChan:             make(chan ResumeRequest),
-		steerQueue:             NewInMemoryMessageQueue(defaultSteerQueueCapacity),
-		followUpQueue:          NewInMemoryMessageQueue(defaultFollowUpQueueCapacity),
-		sessionCompaction:      true,
-		managedOAuth:           true,
-		sessionStore:           session.NewInMemorySessionStore(),
-		fallback:               newFallbackExecutor(),
-		now:                    time.Now,
-		telemetry:              defaultTelemetry{},
-		providerRegistry:       provider.DefaultRegistry(),
-		maxOverflowCompactions: defaultMaxOverflowCompactions,
-		toolListTimeout:        defaultToolListTimeout,
-		toolStartTimeout:       defaultToolStartTimeout,
-		dmrModelLister:         dmrmodels.ListModels,
+		ctx:                          func() context.Context { return context.WithoutCancel(ctx) },
+		toolMap:                      make(map[string]ToolHandlerFunc),
+		liveSessions:                 make(map[string]*liveSessionEntry),
+		team:                         agents,
+		agents:                       newAgentRouter(agents, defaultAgent.Name()),
+		resumeChan:                   make(chan ResumeRequest),
+		steerQueue:                   NewInMemoryMessageQueue(defaultSteerQueueCapacity),
+		followUpQueue:                NewInMemoryMessageQueue(defaultFollowUpQueueCapacity),
+		sessionCompaction:            true,
+		managedOAuth:                 true,
+		sessionStore:                 session.NewInMemorySessionStore(),
+		fallback:                     newFallbackExecutor(),
+		now:                          time.Now,
+		telemetry:                    defaultTelemetry{},
+		providerRegistry:             provider.DefaultRegistry(),
+		maxOverflowCompactions:       defaultMaxOverflowCompactions,
+		toolListTimeout:              defaultToolListTimeout,
+		toolStartTimeout:             defaultToolStartTimeout,
+		streamStoppedDeliveryTimeout: defaultStreamStoppedDeliveryTimeout,
+		dmrModelLister:               dmrmodels.ListModels,
 	}
 	r.bgAgents = agenttool.NewHandler(r)