stream_presenter.go Raw History Folder

1package modes
2
3import (
4	"context"
5	"strings"
6	"time"
7)
8
9// paintPaceRate is the minimum number of runes the streaming pacer releases
10// per tick. With a 16ms tick this is ~375 runes/s, which reads like typing
11// when a provider sends a few fat chunks instead of a steady drip.
12const paintPaceRate = 6
13
14// paintCatchUpTicks sets how fast a backlog drains: each tick releases about
15// 1/paintCatchUpTicks of the unpainted text (at least paintPaceRate runes),
16// so a large lag shrinks geometrically instead of growing with a fast stream.
17const paintCatchUpTicks = 24
18
19// paintPaceInterval is the pacer tick. It matches the redraw throttle so the
20// pacer never produces frames faster than the terminal can show them.
21const paintPaceInterval = 16 * time.Millisecond
22
23type streamState uint8
24
25const (
26	// streamIdle: no assistant reply is on screen as live text.
27	streamIdle streamState = iota
28	// streamReceiving: the provider is still sending text deltas.
29	streamReceiving
30	// streamDraining: the provider finished; the pacer is painting the
31	// remaining buffered text before the transcript takes over.
32	streamDraining
33)
34
35// streamPresenter owns the paced, live display of one assistant reply.
36//
37// The agent appends the finished message to the transcript on its own
38// goroutine, while the pacer is often still typing that same text. Showing
39// both would paint the reply twice. The presenter prevents that with one
40// rule: while it is not idle, the transcript is shown only up to the length
41// it had when the stream started (the anchor). Everything the agent appended
42// since then — the assistant message and any fast tool results — becomes
43// visible in the same frame that the live text disappears.
44//
45// All methods must be called with Interactive.mu held.
46type streamPresenter struct {
47	state        streamState
48	painted      strings.Builder
49	paintedRunes int
50	pending      []rune
51
52	// anchor is the raw transcript length when the stream started, and
53	// anchorRev the transcript revision at that length. A negative anchor
54	// disables transcript clipping (no agent attached).
55	anchor    int
56	anchorRev uint64
57
58	// gates holds, per tool-call id, how many runes of live text must be
59	// painted before that tool block may appear, so a tool never shows up
60	// above the prose that preceded it. A missing entry means visible.
61	gates map[string]int
62}
63
64// Active reports whether live text is still owned by the presenter.
65func (s *streamPresenter) Active() bool { return s.state != streamIdle }
66
67// Text returns the live text painted so far.
68func (s *streamPresenter) Text() string { return s.painted.String() }
69
70// Start begins a new reply. Anything still buffered from a previous reply is
71// dropped: that reply is already complete in the transcript.
72func (s *streamPresenter) Start(anchor int, anchorRev uint64) {
73	s.Reset()
74	s.state = streamReceiving
75	s.anchor = anchor
76	s.anchorRev = anchorRev
77}
78
79// Push buffers a provider text delta for pacing.
80func (s *streamPresenter) Push(delta string) {
81	if s.state != streamReceiving || delta == "" {
82		return
83	}
84	s.pending = append(s.pending, []rune(delta)...)
85}
86
87// Finish marks the provider side of the reply as complete. The presenter
88// stays active until the pacer has painted every buffered rune.
89func (s *streamPresenter) Finish() {
90	if s.state == streamIdle {
91		return
92	}
93	if len(s.pending) == 0 {
94		s.Reset()
95		return
96	}
97	s.state = streamDraining
98}
99
100// Reset returns to idle immediately and reveals the whole transcript.
101func (s *streamPresenter) Reset() {
102	s.state = streamIdle
103	s.painted.Reset()
104	s.paintedRunes = 0
105	s.pending = s.pending[:0]
106	s.anchor = 0
107	s.anchorRev = 0
108	s.gates = nil
109}
110
111// needsTick reports whether the pacer has work to do.
112func (s *streamPresenter) needsTick() bool {
113	return len(s.pending) > 0 || s.state == streamDraining
114}
115
116// Tick paints the next batch of buffered runes. It reports whether the
117// visible state changed.
118func (s *streamPresenter) Tick() bool {
119	if len(s.pending) == 0 {
120		if s.state == streamDraining {
121			s.Reset()
122			return true
123		}
124		return false
125	}
126	n := max(paintPaceRate, (len(s.pending)+paintCatchUpTicks-1)/paintCatchUpTicks)
127	n = min(n, len(s.pending))
128	s.painted.WriteString(string(s.pending[:n]))
129	s.paintedRunes += n
130	s.pending = s.pending[n:]
131	if len(s.pending) == 0 && s.state == streamDraining {
132		// The last rune is painted: hand over to the transcript in the same
133		// frame instead of leaving a one-tick gap.
134		s.Reset()
135	}
136	return true
137}
138
139// Gate records where a tool block may appear. The first registration wins.
140func (s *streamPresenter) Gate(id string) {
141	if s.state == streamIdle {
142		return
143	}
144	if _, ok := s.gates[id]; ok {
145		return
146	}
147	if s.gates == nil {
148		s.gates = make(map[string]int)
149	}
150	s.gates[id] = s.paintedRunes + len(s.pending)
151}
152
153// GateOpen reports whether the tool block may render yet.
154func (s *streamPresenter) GateOpen(id string) bool {
155	gate, ok := s.gates[id]
156	return !ok || s.paintedRunes >= gate
157}
158
159// transcriptLimit returns how many raw transcript messages may be shown and
160// the revision describing that prefix. ok is false when no clipping applies.
161func (s *streamPresenter) transcriptLimit(total int) (limit int, rev uint64, ok bool) {
162	if s.state == streamIdle || s.anchor < 0 || s.anchor > total {
163		// A shorter transcript means it was replaced; never hide content of
164		// a transcript the anchor does not describe.
165		return 0, 0, false
166	}
167	return s.anchor, s.anchorRev, true
168}
169
170// startStreamLocked anchors a new reply at the current transcript length.
171func (i *Interactive) startStreamLocked() {
172	anchor, rev := -1, uint64(0)
173	if i.agent != nil {
174		// The agent goroutine delivers this event synchronously, so the
175		// transcript cannot grow between these two reads.
176		anchor, rev = i.agent.MessageCount(), i.agent.Revision()
177	}
178	i.stream.Start(anchor, rev)
179}
180
181// wakeStreamPacer nudges an idle pacer without blocking.
182func (i *Interactive) wakeStreamPacer() {
183	select {
184	case i.streamWake <- struct{}{}:
185	default:
186	}
187}
188
189// runStreamPacer paints buffered deltas a small batch per tick. It sleeps
190// while there is nothing to paint and exits when ctx is cancelled.
191//
192// Providers chunk text very differently: the Anthropic API-key path drips
193// many small deltas, while the OAuth path can coalesce a reply into a few
194// large chunks that would otherwise just appear. Pacing makes every path
195// look the same on screen.
196func (i *Interactive) runStreamPacer(ctx context.Context) {
197	ticker := time.NewTicker(paintPaceInterval)
198	ticker.Stop()
199	defer ticker.Stop()
200	running := false
201	for {
202		i.mu.Lock()
203		need := i.stream.needsTick()
204		i.mu.Unlock()
205		switch {
206		case need && !running:
207			ticker.Reset(paintPaceInterval)
208			running = true
209		case !need && running:
210			ticker.Stop()
211			running = false
212		}
213		var tick <-chan time.Time
214		if running {
215			tick = ticker.C
216		}
217		select {
218		case <-ctx.Done():
219			return
220		case <-i.streamWake:
221		case <-tick:
222			i.mu.Lock()
223			changed := i.stream.Tick()
224			i.mu.Unlock()
225			if changed {
226				i.invalidate()
227			}
228		}
229	}
230}
stream_presenter.go RawHistoryFolder

stream_presenter.go Raw History Folder
