perf(tui): coalesce streamed text deltas to one frame

Every OnText delta was its own tea.Msg, so a fast provider (100+ tok/s)
drove 100+ full Update->View cycles per second, each re-parsing the
growing markdown — visible stutter over SSH and wasted CPU.

Batch agentTextMsg deltas at the runtime sink over a ~16ms frame and
forward them as a single message, decoupling render rate from token rate.
Any non-text message flushes pending text first so ordering with
tool-call / reasoning / row messages is preserved; the final
agentResponseMsg (a tea.Cmd return, not a sink message) is safe because
the model already drops deltas for an inactive runID.

Author: anandh8x

internal/tui/coalesce.go

@@ -0,0 +1,93 @@
+package tui
+
+import (
+	"sync"
+	"time"
+
+	tea "charm.land/bubbletea/v2"
+)
+
+// streamCoalesceInterval is roughly one 60fps frame. Assistant-text deltas that
+// arrive within this window are merged into a single agentTextMsg, so the render
+// rate decouples from the token rate: a fast local model (100+ tok/s) no longer
+// forces 100+ full Update→View cycles (each re-parsing the growing markdown) per
+// second. Rendering stays smooth regardless of provider speed.
+const streamCoalesceInterval = 16 * time.Millisecond
+
+// textCoalescer batches agentTextMsg deltas before forwarding them to the Bubble
+// Tea program. Any OTHER message flushes the pending text first, so ordering
+// between streamed prose and tool-call / reasoning / row / usage messages is
+// preserved. The turn's final agentResponseMsg does not pass through here (it is
+// a tea.Cmd return, not a sink message), but the model drops deltas whose runID
+// is no longer active, so a flush that races just past end-of-turn is harmless.
+//
+// Sink messages originate from the single agent goroutine and so arrive
+// serially; the only concurrent caller is the flush timer. The mutex guards the
+// buffer/timer against that one race.
+type textCoalescer struct {
+	forward func(tea.Msg) // downstream sink (external sink + program.Send)
+
+	mu    sync.Mutex
+	buf   []byte
+	runID int
+	timer *time.Timer
+}
+
+func newTextCoalescer(forward func(tea.Msg)) *textCoalescer {
+	return &textCoalescer{forward: forward}
+}
+
+// send is the coalescing entry point installed as the RuntimeMessageSink.
+func (c *textCoalescer) send(msg tea.Msg) {
+	text, ok := msg.(agentTextMsg)
+	if !ok {
+		// Non-text message: flush buffered text first (preserving order), then
+		// forward it unchanged.
+		c.flush()
+		c.forward(msg)
+		return
+	}
+
+	c.mu.Lock()
+	// A delta for a different run than the one buffered: flush the old run's text
+	// before buffering the new run's. In practice runs are sequential (the prior
+	// run's end already flushed via a non-text message), so this is belt-and-braces.
+	if len(c.buf) > 0 && text.runID != c.runID {
+		pending := c.drainLocked()
+		c.mu.Unlock()
+		c.forward(pending)
+		c.mu.Lock()
+	}
+	c.runID = text.runID
+	c.buf = append(c.buf, text.delta...)
+	if c.timer == nil {
+		c.timer = time.AfterFunc(streamCoalesceInterval, c.flush)
+	}
+	c.mu.Unlock()
+}
+
+// flush forwards any buffered text as one agentTextMsg. Safe to call from the
+// timer goroutine and inline; a no-op when nothing is buffered.
+func (c *textCoalescer) flush() {
+	c.mu.Lock()
+	if len(c.buf) == 0 {
+		c.mu.Unlock()
+		return
+	}
+	msg := c.drainLocked()
+	c.mu.Unlock()
+	c.forward(msg)
+}
+
+// drainLocked packages the buffer into an agentTextMsg and stops the timer. The
+// caller holds c.mu. string(c.buf) copies, so reusing the backing array via [:0]
+// is safe.
+func (c *textCoalescer) drainLocked() agentTextMsg {
+	if c.timer != nil {
+		c.timer.Stop()
+		c.timer = nil
+	}
+	msg := agentTextMsg{runID: c.runID, delta: string(c.buf)}
+	c.buf = c.buf[:0]
+	return msg
+}

internal/tui/coalesce_test.go

@@ -0,0 +1,124 @@
+package tui
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	tea "charm.land/bubbletea/v2"
+)
+
+// recorder is a thread-safe sink that captures forwarded messages in order.
+type recorder struct {
+	mu   sync.Mutex
+	msgs []tea.Msg
+}
+
+func (r *recorder) forward(msg tea.Msg) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.msgs = append(r.msgs, msg)
+}
+
+func (r *recorder) snapshot() []tea.Msg {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make([]tea.Msg, len(r.msgs))
+	copy(out, r.msgs)
+	return out
+}
+
+// Rapid deltas within one frame collapse into a single agentTextMsg carrying the
+// concatenated text.
+func TestCoalescerBatchesDeltas(t *testing.T) {
+	rec := &recorder{}
+	c := newTextCoalescer(rec.forward)
+
+	c.send(agentTextMsg{runID: 1, delta: "Hel"})
+	c.send(agentTextMsg{runID: 1, delta: "lo, "})
+	c.send(agentTextMsg{runID: 1, delta: "world"})
+
+	// Nothing forwarded yet — still buffered within the frame.
+	if got := rec.snapshot(); len(got) != 0 {
+		t.Fatalf("deltas should buffer, forwarded %d early: %#v", len(got), got)
+	}
+
+	c.flush()
+
+	got := rec.snapshot()
+	if len(got) != 1 {
+		t.Fatalf("expected 1 coalesced message, got %d: %#v", len(got), got)
+	}
+	text, ok := got[0].(agentTextMsg)
+	if !ok || text.delta != "Hello, world" || text.runID != 1 {
+		t.Fatalf("coalesced message = %#v, want agentTextMsg{1, \"Hello, world\"}", got[0])
+	}
+}
+
+// A non-text message flushes buffered text first, so ordering (text before the
+// tool call it precedes) is preserved.
+func TestCoalescerFlushesTextBeforeOtherMessages(t *testing.T) {
+	rec := &recorder{}
+	c := newTextCoalescer(rec.forward)
+
+	c.send(agentTextMsg{runID: 1, delta: "about to run"})
+	c.send(toolCallStreamStartMsg{runID: 1, id: "t1", name: "bash"})
+
+	got := rec.snapshot()
+	if len(got) != 2 {
+		t.Fatalf("expected text then tool-call, got %d: %#v", len(got), got)
+	}
+	if text, ok := got[0].(agentTextMsg); !ok || text.delta != "about to run" {
+		t.Fatalf("first forwarded message must be the flushed text, got %#v", got[0])
+	}
+	if _, ok := got[1].(toolCallStreamStartMsg); !ok {
+		t.Fatalf("second forwarded message must be the tool-call start, got %#v", got[1])
+	}
+}
+
+// A delta for a new run flushes the previous run's buffered text before buffering
+// the new run's, so text is never mis-attributed across runs.
+func TestCoalescerFlushesOnRunSwitch(t *testing.T) {
+	rec := &recorder{}
+	c := newTextCoalescer(rec.forward)
+
+	c.send(agentTextMsg{runID: 1, delta: "run one text"})
+	c.send(agentTextMsg{runID: 2, delta: "run two text"})
+	c.flush()
+
+	got := rec.snapshot()
+	if len(got) != 2 {
+		t.Fatalf("expected one message per run, got %d: %#v", len(got), got)
+	}
+	if first := got[0].(agentTextMsg); first.runID != 1 || first.delta != "run one text" {
+		t.Fatalf("first message = %#v, want run 1 text flushed on switch", got[0])
+	}
+	if second := got[1].(agentTextMsg); second.runID != 2 || second.delta != "run two text" {
+		t.Fatalf("second message = %#v, want run 2 text", got[1])
+	}
+}
+
+// The frame timer flushes buffered text on its own without an explicit flush or a
+// following message.
+func TestCoalescerTimerFlushes(t *testing.T) {
+	rec := &recorder{}
+	c := newTextCoalescer(rec.forward)
+
+	c.send(agentTextMsg{runID: 1, delta: "timer-driven"})
+
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) {
+		if len(rec.snapshot()) > 0 {
+			break
+		}
+		time.Sleep(streamCoalesceInterval)
+	}
+
+	got := rec.snapshot()
+	if len(got) != 1 {
+		t.Fatalf("timer should have flushed exactly one message, got %d: %#v", len(got), got)
+	}
+	if text := got[0].(agentTextMsg); text.delta != "timer-driven" {
+		t.Fatalf("timer-flushed message = %#v", got[0])
+	}
+}

internal/tui/run.go

@@ -25,14 +25,18 @@ func Run(ctx context.Context, options Options) int {
 
 	externalSink := options.RuntimeMessageSink
 	var program *tea.Program
-	options.RuntimeMessageSink = func(msg tea.Msg) {
+	forward := func(msg tea.Msg) {
 		if externalSink != nil {
 			externalSink(msg)
 		}
 		if program != nil {
 			program.Send(msg)
 		}
 	}
+	// Coalesce streamed assistant-text deltas to ~one frame each so a fast provider
+	// can't drive a full Update→View per token; every other message flushes pending
+	// text first, keeping order intact.
+	options.RuntimeMessageSink = newTextCoalescer(forward).send
 	options.AltScreen = useAltScreen(options)
 
 	programOpts := []tea.ProgramOption{