bf4f2c5 refactor(output): deepen attachment output runtime (#218) · vev

R internal/usecase/daemon/output_state_stream.go → internal/usecase/daemon/attachment_output.go +125 -25 renamed, 62% similar


@@ -2,6 +2,7 @@ package daemon
2	2
3	3		import (
4	4		"errors"
	5	+	"strconv"
5	6		"sync"
6	7		"sync/atomic"
7	8

@@ -10,16 +11,16 @@ import (
10	11		"github.com/bnema/vev/internal/ports"
11	12		)
12	13
13		-	// outputStateStream owns the terminal-output dependency chain for one
14		-	// attached client. Each incremental output is rendered from the preceding
15		-	// emitted frame and therefore remains pipelineable without waiting for an
	14	+	// attachmentOutput owns the terminal-output dependency chain and emitted
	15	+	// render state for one attached client. Each incremental output is rendered
	16	+	// from the preceding emitted frame and remains pipelineable without waiting for an
16	17		// ACK. Cumulative ACKs advance the bounded paint window.
17	18		// A reset frame starts a new epoch, clears the dependency chain, and advertises
18	19		// state 0 as its base.
19	20		//
20	21		// Callers serialize access with attachedClient.sendMu. A prepared transaction
21	22		// must be sent or committed as a no-op before the next transaction is prepared.
22		-	type outputStateStream struct {
	23	+	type attachmentOutput struct {
23	24		renderer *renderer.Renderer
24	25		epoch uint64
25	26		next uint64

@@ -30,6 +31,8 @@ type outputStateStream struct {
30	31		maxOutstandingAtomic atomic.Uint64
31	32		forceSnapshot bool
32	33		initialized bool
	34	+	lastCursor cursorOut
	35	+	lastRoutePosition ports.RoutePosition
33	36		attachment *attachedClient
34	37		stateMu sync.Mutex
35	38		}

@@ -38,7 +41,7 @@ type outputStateStream struct {
38	41		// state transition. It deliberately does not take sendMu: callers that hold
39	42		// architecture locks can publish a view without reversing the paint path's
40	43		// sendMu -> session lock order.
41		-	func (s *outputStateStream) lockView() {
	44	+	func (s *attachmentOutput) lockView() {
42	45		if s == nil {
43	46		return
44	47		}

@@ -49,7 +52,7 @@ func (s *outputStateStream) lockView() {
49	52		s.attachment.viewMu.Lock()
50	53		}
51	54
52		-	func (s *outputStateStream) unlockView() {
	55	+	func (s *attachmentOutput) unlockView() {
53	56		if s == nil {
54	57		return
55	58		}

@@ -60,11 +63,11 @@ func (s *outputStateStream) unlockView() {
60	63		s.attachment.viewMu.Unlock()
61	64		}
62	65
63		-	func newOutputStateStream(windowSize ...uint8) *outputStateStream {
	66	+	func newOutputStateStream(windowSize ...uint8) *attachmentOutput {
64	67		return newOutputStateStreamForProfile(renderer.ColorProfileTrueColor, windowSize...)
65	68		}
66	69
67		-	func newOutputStateStreamForCapabilities(capabilities ports.TerminalCapabilities, windowSize ...uint8) *outputStateStream {
	70	+	func newOutputStateStreamForCapabilities(capabilities ports.TerminalCapabilities, windowSize ...uint8) *attachmentOutput {
68	71		profile := renderer.ColorProfileANSI256
69	72		if capabilities.TrueColor() {
70	73		profile = renderer.ColorProfileTrueColor

@@ -72,12 +75,12 @@ func newOutputStateStreamForCapabilities(capabilities ports.TerminalCapabilities
72	75		return newOutputStateStreamForProfile(profile, windowSize...)
73	76		}
74	77
75		-	func newOutputStateStreamForProfile(profile renderer.ColorProfile, windowSize ...uint8) *outputStateStream {
	78	+	func newOutputStateStreamForProfile(profile renderer.ColorProfile, windowSize ...uint8) *attachmentOutput {
76	79		window := uint8(maxUnackedOutputStates)
77	80		if len(windowSize) > 0 {
78	81		window = normalizeOutputWindow(windowSize[0])
79	82		}
80		-	stream := &outputStateStream{
	83	+	stream := &attachmentOutput{
81	84		renderer: renderer.NewWithColorProfile(renderer.Capabilities{}, profile),
82	85		epoch: 1,
83	86		maxOutstanding: uint64(window),

@@ -86,8 +89,87 @@ func newOutputStateStreamForProfile(profile renderer.ColorProfile, windowSize ..
86	89		return stream
87	90		}
88	91
	92	+	type cursorCandidate struct {
	93	+	data []byte
	94	+	next cursorOut
	95	+	}
	96	+
	97	+	// preparedAttachmentOutput is one speculative terminal-output transaction.
	98	+	// ANSI renderer state and cursor state advance together only after the output
	99	+	// frame is accepted by the attachment's state stream.
	100	+	type preparedAttachmentOutput struct {
	101	+	output *attachmentOutput
	102	+	ansi *preparedOutput
	103	+	cursor cursorCandidate
	104	+	data []byte
	105	+	}
	106	+
	107	+	func (o attachmentOutput) prepareFrame(frame renderer.Frame, damage []renderer.Damage, reset bool, desired cursorOut) (preparedAttachmentOutput, error) {
	108	+	ansi, err := o.prepare(frame, damage, reset)
	109	+	if err != nil {
	110	+	return nil, err
	111	+	}
	112	+	cursor := o.prepareCursorTail(desired, len(ansi.data) > 0)
	113	+	data := append([]byte(nil), ansi.data...)
	114	+	data = append(data, cursor.data...)
	115	+	return &preparedAttachmentOutput{output: o, ansi: ansi, cursor: cursor, data: data}, nil
	116	+	}
	117	+
	118	+	func (p *preparedAttachmentOutput) send(echoAck uint64, send func(ports.Frame) error) error {
	119	+	if p == nil \|\| p.ansi == nil {
	120	+	return nil
	121	+	}
	122	+	if err := p.ansi.send(p.data, echoAck, send); err != nil {
	123	+	return err
	124	+	}
	125	+	if p.ansi.sent && p.output != nil {
	126	+	p.output.lastCursor = p.cursor.next
	127	+	}
	128	+	return nil
	129	+	}
	130	+
	131	+	func (p *preparedAttachmentOutput) commitNoSend() {
	132	+	if p == nil \|\| p.ansi == nil {
	133	+	return
	134	+	}
	135	+	p.ansi.commitNoSend()
	136	+	if p.ansi.sent && p.output != nil {
	137	+	p.output.lastCursor = p.cursor.next
	138	+	}
	139	+	}
	140	+
	141	+	func (p *preparedAttachmentOutput) sent() bool {
	142	+	return p != nil && p.ansi != nil && p.ansi.sent
	143	+	}
	144	+
	145	+	func (o *attachmentOutput) prepareCursorTail(desired cursorOut, force bool) cursorCandidate {
	146	+	candidate := cursorCandidate{next: desired}
	147	+	candidate.next.valid = true
	148	+	prev := o.lastCursor
	149	+	changed := force \|\| !prev.valid \|\| prev.hidden != desired.hidden \|\| prev.row != desired.row \|\| prev.col != desired.col \|\| prev.style != desired.style \|\| prev.hasStyle != desired.hasStyle
	150	+	if !changed {
	151	+	return candidate
	152	+	}
	153	+	if desired.hidden {
	154	+	candidate.data = []byte("\x1b[?25l")
	155	+	return candidate
	156	+	}
	157	+	candidate.data = append(candidate.data, "\x1b["...)
	158	+	candidate.data = strconv.AppendInt(candidate.data, int64(desired.row+1), 10)
	159	+	candidate.data = append(candidate.data, ';')
	160	+	candidate.data = strconv.AppendInt(candidate.data, int64(desired.col+1), 10)
	161	+	candidate.data = append(candidate.data, 'H')
	162	+	if !prev.valid \|\| prev.hidden \|\| prev.style != desired.style \|\| prev.hasStyle != desired.hasStyle {
	163	+	candidate.data = append(candidate.data, "\x1b["...)
	164	+	candidate.data = strconv.AppendInt(candidate.data, int64(desired.style), 10)
	165	+	candidate.data = append(candidate.data, " q"...)
	166	+	}
	167	+	candidate.data = append(candidate.data, "\x1b[?25h"...)
	168	+	return candidate
	169	+	}
	170	+
89	171		type preparedOutput struct {
90		-	stream *outputStateStream
	172	+	stream *attachmentOutput
91	173		draw renderer.PreparedDraw
92	174		base uint64
93	175		next uint64

@@ -102,7 +184,7 @@ type preparedOutput struct {
102	184		sent bool
103	185		}
104	186
105		-	func (s outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (preparedOutput, error) {
	187	+	func (s attachmentOutput) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (preparedOutput, error) {
106	188		s.lockView()
107	189		defer s.unlockView()
108	190		if s.epoch == 0 {

@@ -190,20 +272,20 @@ func (p *preparedOutput) commit() {
190	272		p.stream.forceSnapshot = false
191	273		}
192	274
193		-	func (s *outputStateStream) currentEpoch() uint64 {
	275	+	func (s *attachmentOutput) currentEpoch() uint64 {
194	276		s.lockView()
195	277		defer s.unlockView()
196	278		return s.currentEpochLocked()
197	279		}
198	280
199		-	func (s *outputStateStream) currentEpochLocked() uint64 {
	281	+	func (s *attachmentOutput) currentEpochLocked() uint64 {
200	282		if s.epoch == 0 {
201	283		s.epoch = 1
202	284		}
203	285		return s.epoch
204	286		}
205	287
206		-	func (s *outputStateStream) fenceLocked() (uint64, uint64, uint64) {
	288	+	func (s *attachmentOutput) fenceLocked() (uint64, uint64, uint64) {
207	289		if s == nil {
208	290		return 0, 1, 0
209	291		}

@@ -215,7 +297,7 @@ func (s *outputStateStream) fenceLocked() (uint64, uint64, uint64) {
215	297		return generation, s.currentEpochLocked(), revision
216	298		}
217	299
218		-	func (s *outputStateStream) preparedCurrentLocked(generation, epoch, viewRevision, base uint64) bool {
	300	+	func (s *attachmentOutput) preparedCurrentLocked(generation, epoch, viewRevision, base uint64) bool {
219	301		if s == nil \|\| s.currentEpochLocked() != epoch \|\| s.generation != generation \|\| s.next != base {
220	302		return false
221	303		}

@@ -244,7 +326,7 @@ func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision ui
244	326		// sideEffect builds an output frame without advancing the state stream. When
245	327		// an attachment is present, callers must hold its sendMu; attachment-bound
246	328		// sends use sideEffectLocked after validating their transport incarnation.
247		-	func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) (ports.Frame, error) {
	329	+	func (s *attachmentOutput) sideEffect(data []byte, echoAck uint64) (ports.Frame, error) {
248	330		s.lockView()
249	331		defer s.unlockView()
250	332		return s.sideEffectLocked(data, echoAck)

@@ -253,7 +335,7 @@ func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) (ports.Frame
253	335		// sideEffectLocked requires the attachment view lock for both frame creation
254	336		// and transport emission. Keeping that lock across the send prevents a view
255	337		// publication from overtaking a frame that was fenced just before it.
256		-	func (s *outputStateStream) sideEffectLocked(data []byte, echoAck uint64) (ports.Frame, error) {
	338	+	func (s *attachmentOutput) sideEffectLocked(data []byte, echoAck uint64) (ports.Frame, error) {
257	339		_, epoch, viewRevision := s.fenceLocked()
258	340		size := domain.Size{Cols: 1, Rows: 1}
259	341		if s != nil && s.attachment != nil {

@@ -262,7 +344,7 @@ func (s *outputStateStream) sideEffectLocked(data []byte, echoAck uint64) (ports
262	344		return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false)
263	345		}
264	346
265		-	func (s *outputStateStream) ack(epoch, state uint64) bool {
	347	+	func (s *attachmentOutput) ack(epoch, state uint64) bool {
266	348		if s == nil {
267	349		return false
268	350		}

@@ -276,7 +358,7 @@ func (s *outputStateStream) ack(epoch, state uint64) bool {
276	358		return true
277	359		}
278	360
279		-	func (s *outputStateStream) rebase() {
	361	+	func (s *attachmentOutput) rebase() {
280	362		if s == nil {
281	363		return
282	364		}

@@ -285,9 +367,19 @@ func (s *outputStateStream) rebase() {
285	367		s.rebaseLocked()
286	368		}
287	369
	370	+	// rebaseAttachment retires all output state coupled to the attachment's
	371	+	// current session or transport while preserving independent client state.
	372	+	func (s *attachmentOutput) rebaseAttachment() {
	373	+	if s == nil {
	374	+	return
	375	+	}
	376	+	s.rebase()
	377	+	s.lastRoutePosition = ports.RoutePosition{}
	378	+	}
	379	+
288	380		// rebaseLocked is called while the attachment view lock is held. Publishing
289	381		// the next view revision takes this path before making that revision visible.
290		-	func (s *outputStateStream) rebaseLocked() {
	382	+	func (s *attachmentOutput) rebaseLocked() {
291	383		if s == nil {
292	384		return
293	385		}

@@ -306,7 +398,7 @@ func (s *outputStateStream) rebaseLocked() {
306	398		s.initialized = false
307	399		}
308	400
309		-	func (s *outputStateStream) publishOutstanding() {
	401	+	func (s *attachmentOutput) publishOutstanding() {
310	402		if s == nil {
311	403		return
312	404		}

@@ -315,7 +407,7 @@ func (s *outputStateStream) publishOutstanding() {
315	407
316	408		// syncCapacityLocked refreshes the lock-free readiness snapshot after a
317	409		// caller mutates the legacy fields under attachedClient.sendMu.
318		-	func (s *outputStateStream) syncCapacityLocked() {
	410	+	func (s *attachmentOutput) syncCapacityLocked() {
319	411		if s == nil {
320	412		return
321	413		}

@@ -325,7 +417,15 @@ func (s *outputStateStream) syncCapacityLocked() {
325	417		s.publishOutstanding()
326	418		}
327	419
328		-	func (s *outputStateStream) atCapacity() bool {
	420	+	func (s *attachmentOutput) setWindow(window uint8) {
	421	+	if s == nil {
	422	+	return
	423	+	}
	424	+	s.maxOutstanding = uint64(normalizeOutputWindow(window))
	425	+	s.maxOutstandingAtomic.Store(s.maxOutstanding)
	426	+	}
	427	+
	428	+	func (s *attachmentOutput) atCapacity() bool {
329	429		if s == nil {
330	430		return false
331	431		}

@@ -336,7 +436,7 @@ func (s *outputStateStream) atCapacity() bool {
336	436		return s.outstandingAtomic.Load() >= maxOutstanding
337	437		}
338	438
339		-	func (s *outputStateStream) outstanding() uint64 {
	439	+	func (s *attachmentOutput) outstanding() uint64 {
340	440		if s == nil {
341	441		return 0
342	442		}

 import (
 	"errors"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"github.com/bnema/vev/internal/ports"
 )
-// outputStateStream owns the terminal-output dependency chain for one
+// attachmentOutput owns the terminal-output dependency chain and emitted
-// attached client. Each incremental output is rendered from the preceding
+// render state for one attached client. Each incremental output is rendered
-// emitted frame and therefore remains pipelineable without waiting for an
+// from the preceding emitted frame and remains pipelineable without waiting for an
 // ACK. Cumulative ACKs advance the bounded paint window.
 // A reset frame starts a new epoch, clears the dependency chain, and advertises
 // state 0 as its base.
 //
 // Callers serialize access with attachedClient.sendMu. A prepared transaction
 // must be sent or committed as a no-op before the next transaction is prepared.
-type outputStateStream struct {
+type attachmentOutput struct {
 	renderer             *renderer.Renderer
 	epoch                uint64
 	next                 uint64
 	maxOutstandingAtomic atomic.Uint64
 	forceSnapshot        bool
 	initialized          bool
+	lastCursor           cursorOut
+	lastRoutePosition    ports.RoutePosition
 	attachment           *attachedClient
 	stateMu              sync.Mutex
 }
 // state transition. It deliberately does not take sendMu: callers that hold
 // architecture locks can publish a view without reversing the paint path's
 // sendMu -> session lock order.
-func (s *outputStateStream) lockView() {
+func (s *attachmentOutput) lockView() {
 	if s == nil {
 		return
 	}
 	s.attachment.viewMu.Lock()
 }
-func (s *outputStateStream) unlockView() {
+func (s *attachmentOutput) unlockView() {
 	if s == nil {
 		return
 	}
 	s.attachment.viewMu.Unlock()
 }
-func newOutputStateStream(windowSize ...uint8) *outputStateStream {
+func newOutputStateStream(windowSize ...uint8) *attachmentOutput {
 	return newOutputStateStreamForProfile(renderer.ColorProfileTrueColor, windowSize...)
 }
-func newOutputStateStreamForCapabilities(capabilities ports.TerminalCapabilities, windowSize ...uint8) *outputStateStream {
+func newOutputStateStreamForCapabilities(capabilities ports.TerminalCapabilities, windowSize ...uint8) *attachmentOutput {
 	profile := renderer.ColorProfileANSI256
 	if capabilities.TrueColor() {
 		profile = renderer.ColorProfileTrueColor
 	return newOutputStateStreamForProfile(profile, windowSize...)
 }
-func newOutputStateStreamForProfile(profile renderer.ColorProfile, windowSize ...uint8) *outputStateStream {
+func newOutputStateStreamForProfile(profile renderer.ColorProfile, windowSize ...uint8) *attachmentOutput {
 	window := uint8(maxUnackedOutputStates)
 	if len(windowSize) > 0 {
 		window = normalizeOutputWindow(windowSize[0])
 	}
-	stream := &outputStateStream{
+	stream := &attachmentOutput{
 		renderer:       renderer.NewWithColorProfile(renderer.Capabilities{}, profile),
 		epoch:          1,
 		maxOutstanding: uint64(window),
 	return stream
 }
+type cursorCandidate struct {
+	data []byte
+	next cursorOut
+}
+// preparedAttachmentOutput is one speculative terminal-output transaction.
+// ANSI renderer state and cursor state advance together only after the output
+// frame is accepted by the attachment's state stream.
+type preparedAttachmentOutput struct {
+	output *attachmentOutput
+	ansi   *preparedOutput
+	cursor cursorCandidate
+	data   []byte
+}
+func (o *attachmentOutput) prepareFrame(frame renderer.Frame, damage []renderer.Damage, reset bool, desired cursorOut) (*preparedAttachmentOutput, error) {
+	ansi, err := o.prepare(frame, damage, reset)
+	if err != nil {
+		return nil, err
+	}
+	cursor := o.prepareCursorTail(desired, len(ansi.data) > 0)
+	data := append([]byte(nil), ansi.data...)
+	data = append(data, cursor.data...)
+	return &preparedAttachmentOutput{output: o, ansi: ansi, cursor: cursor, data: data}, nil
+}
+func (p *preparedAttachmentOutput) send(echoAck uint64, send func(ports.Frame) error) error {
+	if p == nil || p.ansi == nil {
+		return nil
+	}
+	if err := p.ansi.send(p.data, echoAck, send); err != nil {
+		return err
+	}
+	if p.ansi.sent && p.output != nil {
+		p.output.lastCursor = p.cursor.next
+	}
+	return nil
+}
+func (p *preparedAttachmentOutput) commitNoSend() {
+	if p == nil || p.ansi == nil {
+		return
+	}
+	p.ansi.commitNoSend()
+	if p.ansi.sent && p.output != nil {
+		p.output.lastCursor = p.cursor.next
+	}
+}
+func (p *preparedAttachmentOutput) sent() bool {
+	return p != nil && p.ansi != nil && p.ansi.sent
+}
+func (o *attachmentOutput) prepareCursorTail(desired cursorOut, force bool) cursorCandidate {
+	candidate := cursorCandidate{next: desired}
+	candidate.next.valid = true
+	prev := o.lastCursor
+	changed := force || !prev.valid || prev.hidden != desired.hidden || prev.row != desired.row || prev.col != desired.col || prev.style != desired.style || prev.hasStyle != desired.hasStyle
+	if !changed {
+		return candidate
+	}
+	if desired.hidden {
+		candidate.data = []byte("\x1b[?25l")
+		return candidate
+	}
+	candidate.data = append(candidate.data, "\x1b["...)
+	candidate.data = strconv.AppendInt(candidate.data, int64(desired.row+1), 10)
+	candidate.data = append(candidate.data, ';')
+	candidate.data = strconv.AppendInt(candidate.data, int64(desired.col+1), 10)
+	candidate.data = append(candidate.data, 'H')
+	if !prev.valid || prev.hidden || prev.style != desired.style || prev.hasStyle != desired.hasStyle {
+		candidate.data = append(candidate.data, "\x1b["...)
+		candidate.data = strconv.AppendInt(candidate.data, int64(desired.style), 10)
+		candidate.data = append(candidate.data, " q"...)
+	}
+	candidate.data = append(candidate.data, "\x1b[?25h"...)
+	return candidate
+}
 type preparedOutput struct {
-	stream               *outputStateStream
+	stream               *attachmentOutput
 	draw                 renderer.PreparedDraw
 	base                 uint64
 	next                 uint64
 	sent                 bool
 }
-func (s *outputStateStream) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (*preparedOutput, error) {
+func (s *attachmentOutput) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (*preparedOutput, error) {
 	s.lockView()
 	defer s.unlockView()
 	if s.epoch == 0 {
 	p.stream.forceSnapshot = false
 }
-func (s *outputStateStream) currentEpoch() uint64 {
+func (s *attachmentOutput) currentEpoch() uint64 {
 	s.lockView()
 	defer s.unlockView()
 	return s.currentEpochLocked()
 }
-func (s *outputStateStream) currentEpochLocked() uint64 {
+func (s *attachmentOutput) currentEpochLocked() uint64 {
 	if s.epoch == 0 {
 		s.epoch = 1
 	}
 	return s.epoch
 }
-func (s *outputStateStream) fenceLocked() (uint64, uint64, uint64) {
+func (s *attachmentOutput) fenceLocked() (uint64, uint64, uint64) {
 	if s == nil {
 		return 0, 1, 0
 	}
 	return generation, s.currentEpochLocked(), revision
 }
-func (s *outputStateStream) preparedCurrentLocked(generation, epoch, viewRevision, base uint64) bool {
+func (s *attachmentOutput) preparedCurrentLocked(generation, epoch, viewRevision, base uint64) bool {
 	if s == nil || s.currentEpochLocked() != epoch || s.generation != generation || s.next != base {
 		return false
 	}
 // sideEffect builds an output frame without advancing the state stream. When
 // an attachment is present, callers must hold its sendMu; attachment-bound
 // sends use sideEffectLocked after validating their transport incarnation.
-func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) (ports.Frame, error) {
+func (s *attachmentOutput) sideEffect(data []byte, echoAck uint64) (ports.Frame, error) {
 	s.lockView()
 	defer s.unlockView()
 	return s.sideEffectLocked(data, echoAck)
 // sideEffectLocked requires the attachment view lock for both frame creation
 // and transport emission. Keeping that lock across the send prevents a view
 // publication from overtaking a frame that was fenced just before it.
-func (s *outputStateStream) sideEffectLocked(data []byte, echoAck uint64) (ports.Frame, error) {
+func (s *attachmentOutput) sideEffectLocked(data []byte, echoAck uint64) (ports.Frame, error) {
 	_, epoch, viewRevision := s.fenceLocked()
 	size := domain.Size{Cols: 1, Rows: 1}
 	if s != nil && s.attachment != nil {
 	return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false)
 }
-func (s *outputStateStream) ack(epoch, state uint64) bool {
+func (s *attachmentOutput) ack(epoch, state uint64) bool {
 	if s == nil {
 		return false
 	}
 	return true
 }
-func (s *outputStateStream) rebase() {
+func (s *attachmentOutput) rebase() {
 	if s == nil {
 		return
 	}
 	s.rebaseLocked()
 }
+// rebaseAttachment retires all output state coupled to the attachment's
+// current session or transport while preserving independent client state.
+func (s *attachmentOutput) rebaseAttachment() {
+	if s == nil {
+		return
+	}
+	s.rebase()
+	s.lastRoutePosition = ports.RoutePosition{}
+}
 // rebaseLocked is called while the attachment view lock is held. Publishing
 // the next view revision takes this path before making that revision visible.
-func (s *outputStateStream) rebaseLocked() {
+func (s *attachmentOutput) rebaseLocked() {
 	if s == nil {
 		return
 	}
 	s.initialized = false
 }
-func (s *outputStateStream) publishOutstanding() {
+func (s *attachmentOutput) publishOutstanding() {
 	if s == nil {
 		return
 	}
 // syncCapacityLocked refreshes the lock-free readiness snapshot after a
 // caller mutates the legacy fields under attachedClient.sendMu.
-func (s *outputStateStream) syncCapacityLocked() {
+func (s *attachmentOutput) syncCapacityLocked() {
 	if s == nil {
 		return
 	}
 	s.publishOutstanding()
 }
-func (s *outputStateStream) atCapacity() bool {
+func (s *attachmentOutput) setWindow(window uint8) {
+	if s == nil {
+		return
+	}
+	s.maxOutstanding = uint64(normalizeOutputWindow(window))
+	s.maxOutstandingAtomic.Store(s.maxOutstanding)
+}
+func (s *attachmentOutput) atCapacity() bool {
 	if s == nil {
 		return false
 	}
 	return s.outstandingAtomic.Load() >= maxOutstanding
 }
-func (s *outputStateStream) outstanding() uint64 {
+func (s *attachmentOutput) outstanding() uint64 {
 	if s == nil {
 		return 0
 	}

R internal/usecase/daemon/output_state_stream_test.go → internal/usecase/daemon/attachment_output_test.go +49 -17 renamed, 86% similar


@@ -12,7 +12,7 @@ import (
12	12		"github.com/stretchr/testify/require"
13	13		)
14	14
15		-	func TestOutputStateStreamBuildsPipelinedDependencyChain(t *testing.T) {
	15	+	func TestAttachmentOutputBuildsPipelinedDependencyChain(t *testing.T) {
16	16		stream := newOutputStateStream()
17	17		firstFrame := renderer.NewFrame(3, 1)
18	18		fillOutputStateRows(firstFrame, []string{"abc"})

@@ -37,7 +37,7 @@ func TestOutputStateStreamBuildsPipelinedDependencyChain(t *testing.T) {
37	37		require.Equal(t, uint64(2), stream.outstanding())
38	38		}
39	39
40		-	func TestOutputStateStreamUsesTerminalColorProfile(t *testing.T) {
	40	+	func TestAttachmentOutputUsesTerminalColorProfile(t *testing.T) {
41	41		frame := renderer.NewFrame(1, 1)
42	42		frame.Set(0, 0, renderer.Cell{Rune: 'X', Style: renderer.Style{
43	43		Foreground: -1,

@@ -77,7 +77,7 @@ func TestOutputStateStreamUsesTerminalColorProfile(t *testing.T) {
77	77		}
78	78		}
79	79
80		-	func TestOutputStateStreamCapacityProbeDoesNotRaceWithSend(t *testing.T) {
	80	+	func TestAttachmentOutputCapacityProbeDoesNotRaceWithSend(t *testing.T) {
81	81		stream := newOutputStateStream(1)
82	82		frame := renderer.NewFrame(3, 1)
83	83		fillOutputStateRows(frame, []string{"abc"})

@@ -115,10 +115,10 @@ func TestOutputStateStreamCapacityProbeDoesNotRaceWithSend(t *testing.T) {
115	115		require.False(t, stream.atCapacity())
116	116		}
117	117
118		-	func TestOutputStateStreamEpochsAreAttachmentLocalAndPreparedFramesAreFenced(t *testing.T) {
	118	+	func TestAttachmentOutputEpochsAreAttachmentLocalAndPreparedFramesAreFenced(t *testing.T) {
119	119		frame := renderer.NewFrame(3, 1)
120	120		fillOutputStateRows(frame, []string{"abc"})
121		-	newAttachment := func() *outputStateStream {
	121	+	newAttachment := func() *attachmentOutput {
122	122		stream := newOutputStateStream()
123	123		ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
124	124		stream.attachment = ac

@@ -251,7 +251,7 @@ func TestOutputStateSideEffectsDoNotAdvanceEpochStateOrACK(t *testing.T) {
251	251		require.Equal(t, beforeAck, stream.acked)
252	252		}
253	253
254		-	func TestOutputStateStreamFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {
	254	+	func TestAttachmentOutputFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {
255	255		stream := newOutputStateStream()
256	256		initial := renderer.NewFrame(3, 1)
257	257		fillOutputStateRows(initial, []string{"abc"})

@@ -298,7 +298,7 @@ func TestOutputStateStreamFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T
298	298		require.Empty(t, probe.Bytes(), "successful retry commits the candidate shadow")
299	299		}
300	300
301		-	func TestOutputStateStreamNoByteCommitAdvancesShadowWithoutState(t *testing.T) {
	301	+	func TestAttachmentOutputNoByteCommitAdvancesShadowWithoutState(t *testing.T) {
302	302		stream := newOutputStateStream()
303	303		frame := renderer.NewFrame(3, 1)
304	304		fillOutputStateRows(frame, []string{"abc"})

@@ -318,7 +318,7 @@ func TestOutputStateStreamNoByteCommitAdvancesShadowWithoutState(t *testing.T) {
318	318		require.Empty(t, probe.Bytes(), "no-byte commit retains the prepared visual candidate")
319	319		}
320	320
321		-	func TestOutputStateStreamRepeatedUnackedScrollRemainsClientCorrect(t *testing.T) {
	321	+	func TestAttachmentOutputRepeatedUnackedScrollRemainsClientCorrect(t *testing.T) {
322	322		stream := newOutputStateStream()
323	323		client := vt.NewScreen(4, 3)
324	324		initial := renderer.NewFrame(4, 3)

@@ -352,7 +352,7 @@ func TestOutputStateStreamRepeatedUnackedScrollRemainsClientCorrect(t *testing.T
352	352		}
353	353		}
354	354
355		-	func (s *outputStateStream) render(frame renderer.Frame, damage []renderer.Damage, reset bool) ([]byte, error) {
	355	+	func (s *attachmentOutput) render(frame renderer.Frame, damage []renderer.Damage, reset bool) ([]byte, error) {
356	356		prepared, err := s.prepare(frame, damage, reset)
357	357		if err != nil {
358	358		return nil, err

@@ -368,7 +368,7 @@ func (s *outputStateStream) render(frame renderer.Frame, damage []renderer.Damag
368	368		return prepared.data, nil
369	369		}
370	370
371		-	func outputStateFrame(stream *outputStateStream, data []byte, reset bool, echoAck uint64) ports.Frame {
	371	+	func outputStateFrame(stream *attachmentOutput, data []byte, reset bool, echoAck uint64) ports.Frame {
372	372		stream.next++
373	373		base := stream.next - 1
374	374		if reset {

@@ -381,7 +381,7 @@ func outputStateFrame(stream *outputStateStream, data []byte, reset bool, echoAc
381	381		return frame
382	382		}
383	383
384		-	func drawOutputState(t testing.T, stream outputStateStream, frame renderer.Frame, damage []renderer.Damage, reset bool, echoAck uint64) (ports.Frame, bool, error) {
	384	+	func drawOutputState(t testing.T, stream attachmentOutput, frame renderer.Frame, damage []renderer.Damage, reset bool, echoAck uint64) (ports.Frame, bool, error) {
385	385		t.Helper()
386	386		prepared, err := stream.prepare(frame, damage, reset)
387	387		if err != nil \|\| len(prepared.data) == 0 {

@@ -403,7 +403,7 @@ func drawOutputState(t testing.T, stream outputStateStream, frame renderer.Fra
403	403
404	404		// A resize is a state reset, but it must not turn subsequent idle or damaged
405	405		// renders into full frames.
406		-	func TestOutputStateStreamResizeFrameThenNoopAndDamageAreDifferential(t *testing.T) {
	406	+	func TestAttachmentOutputResizeFrameThenNoopAndDamageAreDifferential(t *testing.T) {
407	407		screen := vt.NewScreen(4, 2)
408	408		screen.Write([]byte("abcd"))
409	409		screen.ClearDamage()

@@ -431,11 +431,43 @@ func TestOutputStateStreamResizeFrameThenNoopAndDamageAreDifferential(t *testing
431	431		require.Equal(t, resizeOutput.New, damageOutput.Base, "later damage must remain incremental")
432	432		}
433	433
434		-	func TestOutputStateStreamDefaultsAndNormalizesWindow(t *testing.T) {
435		-	require.Equal(t, uint64(8), newOutputStateStream().maxOutstanding)
436		-	require.Equal(t, uint64(8), newOutputStateStream(0).maxOutstanding)
437		-	require.Equal(t, uint64(8), newOutputStateStream(9).maxOutstanding)
438		-	require.Equal(t, uint64(1), newOutputStateStream(1).maxOutstanding)
	434	+	func TestAttachmentOutputRebaseRetiresAttachmentState(t *testing.T) {
	435	+	output := newOutputStateStream()
	436	+	output.next = 3
	437	+	output.acked = 2
	438	+	output.lastCursor = cursorOut{valid: true, row: 2, col: 3}
	439	+	output.lastRoutePosition = ports.RoutePosition{ActiveTabID: "tab"}
	440	+
	441	+	output.rebaseAttachment()
	442	+
	443	+	require.Equal(t, uint64(2), output.epoch)
	444	+	require.Zero(t, output.next)
	445	+	require.Zero(t, output.acked)
	446	+	require.True(t, output.forceSnapshot)
	447	+	require.Equal(t, cursorOut{valid: true, row: 2, col: 3}, output.lastCursor, "the forced frame will reassert cursor state")
	448	+	require.Equal(t, ports.RoutePosition{}, output.lastRoutePosition)
	449	+	}
	450	+
	451	+	func TestAttachmentOutputDefaultsAndNormalizesWindow(t *testing.T) {
	452	+	for _, tt := range []struct {
	453	+	name string
	454	+	window []uint8
	455	+	want uint64
	456	+	}{
	457	+	{name: "omitted", want: 8},
	458	+	{name: "zero", window: []uint8{0}, want: 8},
	459	+	{name: "oversized", window: []uint8{9}, want: 8},
	460	+	{name: "exact window", window: []uint8{1}, want: 1},
	461	+	} {
	462	+	t.Run(tt.name, func(t *testing.T) {
	463	+	require.Equal(t, tt.want, newOutputStateStream(tt.window...).maxOutstanding)
	464	+	})
	465	+	}
	466	+
	467	+	output := newOutputStateStream()
	468	+	output.setWindow(2)
	469	+	require.Equal(t, uint64(2), output.maxOutstanding)
	470	+	require.Equal(t, uint64(2), output.maxOutstandingAtomic.Load())
439	471		}
440	472
441	473		func TestNormalizeOutputWindow(t *testing.T) {

 	"github.com/stretchr/testify/require"
 )
-func TestOutputStateStreamBuildsPipelinedDependencyChain(t *testing.T) {
+func TestAttachmentOutputBuildsPipelinedDependencyChain(t *testing.T) {
 	stream := newOutputStateStream()
 	firstFrame := renderer.NewFrame(3, 1)
 	fillOutputStateRows(firstFrame, []string{"abc"})
 	require.Equal(t, uint64(2), stream.outstanding())
 }
-func TestOutputStateStreamUsesTerminalColorProfile(t *testing.T) {
+func TestAttachmentOutputUsesTerminalColorProfile(t *testing.T) {
 	frame := renderer.NewFrame(1, 1)
 	frame.Set(0, 0, renderer.Cell{Rune: 'X', Style: renderer.Style{
 		Foreground:           -1,
 	}
 }
-func TestOutputStateStreamCapacityProbeDoesNotRaceWithSend(t *testing.T) {
+func TestAttachmentOutputCapacityProbeDoesNotRaceWithSend(t *testing.T) {
 	stream := newOutputStateStream(1)
 	frame := renderer.NewFrame(3, 1)
 	fillOutputStateRows(frame, []string{"abc"})
 	require.False(t, stream.atCapacity())
 }
-func TestOutputStateStreamEpochsAreAttachmentLocalAndPreparedFramesAreFenced(t *testing.T) {
+func TestAttachmentOutputEpochsAreAttachmentLocalAndPreparedFramesAreFenced(t *testing.T) {
 	frame := renderer.NewFrame(3, 1)
 	fillOutputStateRows(frame, []string{"abc"})
-	newAttachment := func() *outputStateStream {
+	newAttachment := func() *attachmentOutput {
 		stream := newOutputStateStream()
 		ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
 		stream.attachment = ac
 	require.Equal(t, beforeAck, stream.acked)
 }
-func TestOutputStateStreamFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {
+func TestAttachmentOutputFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {
 	stream := newOutputStateStream()
 	initial := renderer.NewFrame(3, 1)
 	fillOutputStateRows(initial, []string{"abc"})
 	require.Empty(t, probe.Bytes(), "successful retry commits the candidate shadow")
 }
-func TestOutputStateStreamNoByteCommitAdvancesShadowWithoutState(t *testing.T) {
+func TestAttachmentOutputNoByteCommitAdvancesShadowWithoutState(t *testing.T) {
 	stream := newOutputStateStream()
 	frame := renderer.NewFrame(3, 1)
 	fillOutputStateRows(frame, []string{"abc"})
 	require.Empty(t, probe.Bytes(), "no-byte commit retains the prepared visual candidate")
 }
-func TestOutputStateStreamRepeatedUnackedScrollRemainsClientCorrect(t *testing.T) {
+func TestAttachmentOutputRepeatedUnackedScrollRemainsClientCorrect(t *testing.T) {
 	stream := newOutputStateStream()
 	client := vt.NewScreen(4, 3)
 	initial := renderer.NewFrame(4, 3)
 	}
 }
-func (s *outputStateStream) render(frame renderer.Frame, damage []renderer.Damage, reset bool) ([]byte, error) {
+func (s *attachmentOutput) render(frame renderer.Frame, damage []renderer.Damage, reset bool) ([]byte, error) {
 	prepared, err := s.prepare(frame, damage, reset)
 	if err != nil {
 		return nil, err
 	return prepared.data, nil
 }
-func outputStateFrame(stream *outputStateStream, data []byte, reset bool, echoAck uint64) ports.Frame {
+func outputStateFrame(stream *attachmentOutput, data []byte, reset bool, echoAck uint64) ports.Frame {
 	stream.next++
 	base := stream.next - 1
 	if reset {
 	return frame
 }
-func drawOutputState(t *testing.T, stream *outputStateStream, frame renderer.Frame, damage []renderer.Damage, reset bool, echoAck uint64) (ports.Frame, bool, error) {
+func drawOutputState(t *testing.T, stream *attachmentOutput, frame renderer.Frame, damage []renderer.Damage, reset bool, echoAck uint64) (ports.Frame, bool, error) {
 	t.Helper()
 	prepared, err := stream.prepare(frame, damage, reset)
 	if err != nil || len(prepared.data) == 0 {
 // A resize is a state reset, but it must not turn subsequent idle or damaged
 // renders into full frames.
-func TestOutputStateStreamResizeFrameThenNoopAndDamageAreDifferential(t *testing.T) {
+func TestAttachmentOutputResizeFrameThenNoopAndDamageAreDifferential(t *testing.T) {
 	screen := vt.NewScreen(4, 2)
 	screen.Write([]byte("abcd"))
 	screen.ClearDamage()
 	require.Equal(t, resizeOutput.New, damageOutput.Base, "later damage must remain incremental")
 }
-func TestOutputStateStreamDefaultsAndNormalizesWindow(t *testing.T) {
+func TestAttachmentOutputRebaseRetiresAttachmentState(t *testing.T) {
-	require.Equal(t, uint64(8), newOutputStateStream().maxOutstanding)
+	output := newOutputStateStream()
-	require.Equal(t, uint64(8), newOutputStateStream(0).maxOutstanding)
+	output.next = 3
-	require.Equal(t, uint64(8), newOutputStateStream(9).maxOutstanding)
+	output.acked = 2
-	require.Equal(t, uint64(1), newOutputStateStream(1).maxOutstanding)
+	output.lastCursor = cursorOut{valid: true, row: 2, col: 3}
+	output.lastRoutePosition = ports.RoutePosition{ActiveTabID: "tab"}
+	output.rebaseAttachment()
+	require.Equal(t, uint64(2), output.epoch)
+	require.Zero(t, output.next)
+	require.Zero(t, output.acked)
+	require.True(t, output.forceSnapshot)
+	require.Equal(t, cursorOut{valid: true, row: 2, col: 3}, output.lastCursor, "the forced frame will reassert cursor state")
+	require.Equal(t, ports.RoutePosition{}, output.lastRoutePosition)
+}
+func TestAttachmentOutputDefaultsAndNormalizesWindow(t *testing.T) {
+	for _, tt := range []struct {
+		name   string
+		window []uint8
+		want   uint64
+	}{
+		{name: "omitted", want: 8},
+		{name: "zero", window: []uint8{0}, want: 8},
+		{name: "oversized", window: []uint8{9}, want: 8},
+		{name: "exact window", window: []uint8{1}, want: 1},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, newOutputStateStream(tt.window...).maxOutstanding)
+		})
+	}
+	output := newOutputStateStream()
+	output.setWindow(2)
+	require.Equal(t, uint64(2), output.maxOutstanding)
+	require.Equal(t, uint64(2), output.maxOutstandingAtomic.Load())
 }
 func TestNormalizeOutputWindow(t *testing.T) {

M internal/usecase/daemon/client.go +4 -9


@@ -35,7 +35,7 @@ type appliedTheme struct {
35	35		type attachedClient struct {
36	36		tr ports.Transport
37	37		transportIncarnation uint64
38		-	output *outputStateStream
	38	+	output *attachmentOutput
39	39		overlays *overlayRuntime
40	40		overlayOnce sync.Once
41	41		clientID [16]byte

@@ -90,8 +90,6 @@ type attachedClient struct {
90	90		themeMu sync.Mutex
91	91		clientTheme themeui.Theme
92	92		appliedTheme appliedTheme
93		-	lastCursor cursorOut
94		-	lastRoutePosition ports.RoutePosition
95	93		renderStages renderStageHooks // optional render and handoff observability hooks
96	94		linkMu sync.Mutex
97	95		sendMu sync.Mutex

@@ -421,13 +419,10 @@ func (ac *attachedClient) ackOutputState(epoch, state uint64) {
421	419		// rebaseOutput resets only this attachment's output representations. Callers
422	420		// hold sendMu (or the activation barrier).
423	421		func (ac *attachedClient) rebaseOutput() {
424		-	if ac == nil {
	422	+	if ac == nil \|\| ac.output == nil {
425	423		return
426	424		}
427		-	if ac.output != nil {
428		-	ac.output.rebase()
429		-	}
430		-	ac.lastRoutePosition = ports.RoutePosition{}
	425	+	ac.output.rebaseAttachment()
431	426		}
432	427
433	428		var errTransportReplaced = errors.New("client transport was replaced")

@@ -791,7 +786,7 @@ func (d Daemon) ensureAttachmentRenderCoordinatorPrelocked(entry session) *ren
791	786		}
792	787		},
793	788		ackReadyFor: func(attached *attachedClient) bool {
794		-	// outputStateStream publishes capacity atomically. Do not take
	789	+	// attachmentOutput publishes capacity atomically. Do not take
795	790		// attached.sendMu here: a slow transport may be holding it for an
796	791		// in-flight Send, and that peer must not gate healthy attachments.
797	792		return attached == nil \|\|

 type attachedClient struct {
 	tr                     ports.Transport
 	transportIncarnation   uint64
-	output                 *outputStateStream
+	output                 *attachmentOutput
 	overlays               *overlayRuntime
 	overlayOnce            sync.Once
 	clientID               [16]byte
 	themeMu                    sync.Mutex
 	clientTheme                themeui.Theme
 	appliedTheme               appliedTheme
-	lastCursor                 cursorOut
-	lastRoutePosition          ports.RoutePosition
 	renderStages               renderStageHooks // optional render and handoff observability hooks
 	linkMu                     sync.Mutex
 	sendMu                     sync.Mutex
 // rebaseOutput resets only this attachment's output representations. Callers
 // hold sendMu (or the activation barrier).
 func (ac *attachedClient) rebaseOutput() {
-	if ac == nil {
+	if ac == nil || ac.output == nil {
 		return
 	}
-	if ac.output != nil {
+	ac.output.rebaseAttachment()
-		ac.output.rebase()
-	}
-	ac.lastRoutePosition = ports.RoutePosition{}
 }
 var errTransportReplaced = errors.New("client transport was replaced")
 			}
 		},
 		ackReadyFor: func(attached *attachedClient) bool {
-			// outputStateStream publishes capacity atomically. Do not take
+			// attachmentOutput publishes capacity atomically. Do not take
 			// attached.sendMu here: a slow transport may be holding it for an
 			// in-flight Send, and that peer must not gate healthy attachments.
 			return attached == nil ||

M internal/usecase/daemon/render_coordinator_types.go +1 -1


@@ -103,7 +103,7 @@ type renderCoordinatorOptions struct {
103	103		// wake composes a coalesced render request.
104	104		wake func(renderWake)
105	105		// ackReady reports whether the attachment may compose another output
106		-	// state (the outputStateStream window has capacity). It is retained for
	106	+	// state (the attachmentOutput window has capacity). It is retained for
107	107		// coordinator-only tests and headless callers; production uses ackReadyFor
108	108		// so one attachment cannot gate another.
109	109		ackReady func() bool


@@ -103,7 +103,7 @@ type renderCoordinatorOptions struct {
103		// wake composes a coalesced render request.	103		// wake composes a coalesced render request.
104		wake func(renderWake)	104		wake func(renderWake)
105		// ackReady reports whether the attachment may compose another output	105		// ackReady reports whether the attachment may compose another output
106	-	// state (the outputStateStream window has capacity). It is retained for	106	+	// state (the attachmentOutput window has capacity). It is retained for
107		// coordinator-only tests and headless callers; production uses ackReadyFor	107		// coordinator-only tests and headless callers; production uses ackReadyFor
108		// so one attachment cannot gate another.	108		// so one attachment cannot gate another.
109		ackReady func() bool	109		ackReady func() bool

M internal/usecase/daemon/render_pipeline.go +10 -50


@@ -2,7 +2,6 @@ package daemon
2	2
3	3		import (
4	4		"errors"
5		-	"strconv"
6	5
7	6		renderer "github.com/bnema/vev-vt"
8	7		"github.com/bnema/vev/internal/domain"

@@ -379,37 +378,6 @@ func noticeStylesFrom(styles themeui.Styles) ui.NoticeStyles {
379	378		}
380	379		}
381	380
382		-	type cursorCandidate struct {
383		-	data []byte
384		-	next cursorOut
385		-	}
386		-
387		-	func (ac *attachedClient) prepareCursorTail(desired cursorOut, force bool) cursorCandidate {
388		-	candidate := cursorCandidate{next: desired}
389		-	candidate.next.valid = true
390		-	prev := ac.lastCursor
391		-	changed := force \|\| !prev.valid \|\| prev.hidden != desired.hidden \|\| prev.row != desired.row \|\| prev.col != desired.col \|\| prev.style != desired.style \|\| prev.hasStyle != desired.hasStyle
392		-	if !changed {
393		-	return candidate
394		-	}
395		-	if desired.hidden {
396		-	candidate.data = []byte("\x1b[?25l")
397		-	return candidate
398		-	}
399		-	candidate.data = append(candidate.data, "\x1b["...)
400		-	candidate.data = strconv.AppendInt(candidate.data, int64(desired.row+1), 10)
401		-	candidate.data = append(candidate.data, ';')
402		-	candidate.data = strconv.AppendInt(candidate.data, int64(desired.col+1), 10)
403		-	candidate.data = append(candidate.data, 'H')
404		-	if !prev.valid \|\| prev.hidden \|\| prev.style != desired.style \|\| prev.hasStyle != desired.hasStyle {
405		-	candidate.data = append(candidate.data, "\x1b["...)
406		-	candidate.data = strconv.AppendInt(candidate.data, int64(desired.style), 10)
407		-	candidate.data = append(candidate.data, " q"...)
408		-	}
409		-	candidate.data = append(candidate.data, "\x1b[?25h"...)
410		-	return candidate
411		-	}
412		-
413	381		// commitDamageReceipts acknowledges only snapshots that reached the client.
414	382		// sendMu serializes attachment transactions; panes are locked individually so
415	383		// no pane lock is held while another is acquired. A stale generation is

@@ -468,17 +436,10 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
468	436		}
469	437		}
470	438		endDiff := marks.span(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0)
471		-	var (
472		-	preparedANSI *preparedOutput
473		-	err error
474		-	cursor cursorCandidate
475		-	data []byte
476		-	)
477		-	preparedANSI, err = ac.output.prepare(composed.frame, composed.damage, composed.reset)
478		-	if err == nil {
479		-	cursor = ac.prepareCursorTail(composed.cursor, len(preparedANSI.data) > 0)
480		-	data = append([]byte(nil), preparedANSI.data...)
481		-	data = append(data, cursor.data...)
	439	+	prepared, err := ac.output.prepareFrame(composed.frame, composed.damage, composed.reset, composed.cursor)
	440	+	var data []byte
	441	+	if prepared != nil {
	442	+	data = prepared.data
482	443		}
483	444		endDiff(0, err == nil)
484	445		if err != nil {

@@ -526,7 +487,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
526	487		}
527	488		}
528	489		if sendErr == nil {
529		-	sendErr = preparedANSI.send(data, ac.echoAck.Load(), send)
	490	+	sendErr = prepared.send(ac.echoAck.Load(), send)
530	491		}
531	492		if marks.attachmentEffect != nil && interruptible {
532	493		if sendErr != nil {

@@ -535,7 +496,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
535	496		marks.attachmentEffect.endTransportSend()
536	497		}
537	498		}
538		-	emitted := sendErr == nil && preparedANSI.sent
	499	+	emitted := sendErr == nil && prepared.sent()
539	500		endEmit(uint64(len(data)), emitted)
540	501		if sendErr == nil && !emitted {
541	502		ac.sendMu.Unlock()

@@ -544,8 +505,8 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
544	505		}
545	506		if sendErr == nil {
546	507		if len(data) == 0 {
547		-	preparedANSI.commitNoSend()
548		-	if !preparedANSI.sent {
	508	+	prepared.commitNoSend()
	509	+	if !prepared.sent() {
549	510		ac.sendMu.Unlock()
550	511		return true
551	512		}

@@ -554,7 +515,6 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
554	515		// succeed. A cross-session transition may publish concurrently, but its
555	516		// mandatory first-paint rebase waits for sendMu and therefore follows this
556	517		// completed output transaction.
557		-	ac.lastCursor = cursor.next
558	518		ac.pipelineScratch = ac.pipelineCache
559	519		ac.pipelineCache = composed.cache
560	520

@@ -566,7 +526,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
566	526		ac.renderStages.emit()
567	527		}
568	528
569		-	if position.ActiveTabID != "" && position != ac.lastRoutePosition {
	529	+	if position.ActiveTabID != "" && position != ac.output.lastRoutePosition {
570	530		payload, positionMarshalErr := ports.MarshalRoutePosition(position)
571	531		if positionMarshalErr != nil {
572	532		d.log.Error("marshal route position", "err", positionMarshalErr)

@@ -598,7 +558,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
598	558		marks.attachmentEffect.endTransportSend()
599	559		}
600	560		if sendErr == nil {
601		-	ac.lastRoutePosition = position
	561	+	ac.output.lastRoutePosition = position
602	562		}
603	563		}
604	564		}

 import (
 	"errors"
-	"strconv"
 	renderer "github.com/bnema/vev-vt"
 	"github.com/bnema/vev/internal/domain"
 	}
 }
-type cursorCandidate struct {
-	data []byte
-	next cursorOut
-}
-func (ac *attachedClient) prepareCursorTail(desired cursorOut, force bool) cursorCandidate {
-	candidate := cursorCandidate{next: desired}
-	candidate.next.valid = true
-	prev := ac.lastCursor
-	changed := force || !prev.valid || prev.hidden != desired.hidden || prev.row != desired.row || prev.col != desired.col || prev.style != desired.style || prev.hasStyle != desired.hasStyle
-	if !changed {
-		return candidate
-	}
-	if desired.hidden {
-		candidate.data = []byte("\x1b[?25l")
-		return candidate
-	}
-	candidate.data = append(candidate.data, "\x1b["...)
-	candidate.data = strconv.AppendInt(candidate.data, int64(desired.row+1), 10)
-	candidate.data = append(candidate.data, ';')
-	candidate.data = strconv.AppendInt(candidate.data, int64(desired.col+1), 10)
-	candidate.data = append(candidate.data, 'H')
-	if !prev.valid || prev.hidden || prev.style != desired.style || prev.hasStyle != desired.hasStyle {
-		candidate.data = append(candidate.data, "\x1b["...)
-		candidate.data = strconv.AppendInt(candidate.data, int64(desired.style), 10)
-		candidate.data = append(candidate.data, " q"...)
-	}
-	candidate.data = append(candidate.data, "\x1b[?25h"...)
-	return candidate
-}
 // commitDamageReceipts acknowledges only snapshots that reached the client.
 // sendMu serializes attachment transactions; panes are locked individually so
 // no pane lock is held while another is acquired. A stale generation is
 		}
 	}
 	endDiff := marks.span(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0)
-	var (
+	prepared, err := ac.output.prepareFrame(composed.frame, composed.damage, composed.reset, composed.cursor)
-		preparedANSI *preparedOutput
+	var data []byte
-		err          error
+	if prepared != nil {
-		cursor       cursorCandidate
+		data = prepared.data
-		data         []byte
-	)
-	preparedANSI, err = ac.output.prepare(composed.frame, composed.damage, composed.reset)
-	if err == nil {
-		cursor = ac.prepareCursorTail(composed.cursor, len(preparedANSI.data) > 0)
-		data = append([]byte(nil), preparedANSI.data...)
-		data = append(data, cursor.data...)
 	}
 	endDiff(0, err == nil)
 	if err != nil {
 				}
 			}
 			if sendErr == nil {
-				sendErr = preparedANSI.send(data, ac.echoAck.Load(), send)
+				sendErr = prepared.send(ac.echoAck.Load(), send)
 			}
 			if marks.attachmentEffect != nil && interruptible {
 				if sendErr != nil {
 				marks.attachmentEffect.endTransportSend()
 			}
 		}
-		emitted := sendErr == nil && preparedANSI.sent
+		emitted := sendErr == nil && prepared.sent()
 		endEmit(uint64(len(data)), emitted)
 		if sendErr == nil && !emitted {
 			ac.sendMu.Unlock()
 	}
 	if sendErr == nil {
 		if len(data) == 0 {
-			preparedANSI.commitNoSend()
+			prepared.commitNoSend()
-			if !preparedANSI.sent {
+			if !prepared.sent() {
 				ac.sendMu.Unlock()
 				return true
 			}
 		// succeed. A cross-session transition may publish concurrently, but its
 		// mandatory first-paint rebase waits for sendMu and therefore follows this
 		// completed output transaction.
-		ac.lastCursor = cursor.next
 		ac.pipelineScratch = ac.pipelineCache
 		ac.pipelineCache = composed.cache
 			ac.renderStages.emit()
 		}
-		if position.ActiveTabID != "" && position != ac.lastRoutePosition {
+		if position.ActiveTabID != "" && position != ac.output.lastRoutePosition {
 			payload, positionMarshalErr := ports.MarshalRoutePosition(position)
 			if positionMarshalErr != nil {
 				d.log.Error("marshal route position", "err", positionMarshalErr)
 					marks.attachmentEffect.endTransportSend()
 				}
 				if sendErr == nil {
-					ac.lastRoutePosition = position
+					ac.output.lastRoutePosition = position
 				}
 			}
 		}

M internal/usecase/daemon/render_pipeline_test.go +17 -12


@@ -213,12 +213,12 @@ func TestProductionRenderUsesOnlyTypedComposeEntryPoint(t *testing.T) {
213	213		}
214	214
215	215		func TestCursorCandidateDoesNotPublishDuringPreparation(t *testing.T) {
216		-	ac := &attachedClient{lastCursor: cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}}
	216	+	output := &attachmentOutput{lastCursor: cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}}
217	217		desired := cursorOut{row: 3, col: 4, style: 2, hasStyle: true}
218	218
219		-	candidate := ac.prepareCursorTail(desired, false)
	219	+	candidate := output.prepareCursorTail(desired, false)
220	220
221		-	require.Equal(t, cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}, ac.lastCursor)
	221	+	require.Equal(t, cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}, output.lastCursor)
222	222		require.NotEmpty(t, candidate.data)
223	223		require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, candidate.next)
224	224		}

@@ -227,7 +227,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
227	227		d, sess, ac, sends := newManualSessionWithPTYs(t)
228	228		healthy := ac.transport()
229	229		beforeCursor := cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}
230		-	ac.lastCursor = beforeCursor
	230	+	ac.output.lastCursor = beforeCursor
231	231		ac.replaceTransport(cacheFailTransport{})
232	232		state := cacheState("failed", 1)
233	233		state.attachment = ac

@@ -237,7 +237,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
237	237		ac.sendMu.Lock()
238	238		require.True(t, d.emitFrame(sess, ac, &state, composed))
239	239
240		-	require.Equal(t, beforeCursor, ac.lastCursor)
	240	+	require.Equal(t, beforeCursor, ac.output.lastCursor)
241	241		require.Zero(t, ac.output.next)
242	242		probe, err := ac.output.renderer.Prepare(composed.frame, nil, false)
243	243		require.NoError(t, err)

@@ -251,7 +251,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
251	251		state.attachment = ac
252	252		ac.sendMu.Lock()
253	253		require.True(t, d.emitFrame(sess, ac, &state, composed))
254		-	require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, ac.lastCursor)
	254	+	require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, ac.output.lastCursor)
255	255		require.Equal(t, uint64(1), ac.output.next)
256	256		out, err := ports.UnmarshalOutput((<-sends).Payload)
257	257		require.NoError(t, err)

@@ -276,7 +276,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
276	276		p.mu.Unlock()
277	277
278	278		beforeNext := ac.output.next
279		-	beforeCursor := ac.lastCursor
	279	+	beforeCursor := ac.output.lastCursor
280	280		state.receipts = []damageReceipt{{pane: p, generation: capture.Generation}}
281	281		noByte := composedRenderFrame{
282	282		frame: initial.frame,

@@ -288,7 +288,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
288	288		require.True(t, d.emitFrame(sess, ac, &state, noByte))
289	289
290	290		require.Equal(t, beforeNext, ac.output.next)
291		-	require.Equal(t, beforeCursor, ac.lastCursor)
	291	+	require.Equal(t, beforeCursor, ac.output.lastCursor)
292	292		require.Equal(t, "no-byte-committed", ac.pipelineCache.layoutFingerprint)
293	293		p.mu.Lock()
294	294		require.Empty(t, p.screen.Damage())

@@ -330,15 +330,20 @@ func TestComposeEmitExactReplayTiledFloatingBarsOverlayAndCursor(t *testing.T) {
330	330		require.True(t, composed.cursor.hidden, "overlay owns cursor visibility")
331	331
332	332		stream := newOutputStateStream()
333		-	prepared, err := stream.prepare(composed.frame, composed.damage, composed.reset)
	333	+	prepared, err := stream.prepareFrame(composed.frame, composed.damage, composed.reset, composed.cursor)
334	334		require.NoError(t, err)
335		-	terminalBytes := append(append([]byte(nil), prepared.data...), (&attachedClient{}).prepareCursorTail(composed.cursor, true).data...)
	335	+	var outputFrame ports.Frame
	336	+	require.NoError(t, prepared.send(0, func(frame ports.Frame) error {
	337	+	outputFrame = frame
	338	+	return nil
	339	+	}))
	340	+	output, err := ports.UnmarshalOutput(outputFrame.Payload)
	341	+	require.NoError(t, err)
	342	+	terminalBytes := output.Data
336	343		require.Equal(t, "\x1b[1;1H\x1b[0;7m tab \x1b[0m R\x1b[2;1HAAAAAAAAAAAA\x1b[3;1HBBB┌─fl─┐BBB\x1b[4;1H───Prompt───\x1b[5;1HPROMPT\x1b[K\x1b[B\x1b[2K\x1b[7;1H\x1b[0;7m sess \x1b[0m B\x1b[0m\x1b[?25l", string(terminalBytes))
337	344		client := vt.NewScreen(composed.frame.Width, composed.frame.Height)
338	345		client.Write(terminalBytes)
339	346		require.Equal(t, frameRows(composed.frame), frameRows(client.Frame))
340		-	prepared.draw.Commit()
341		-	stream.forceSnapshot = false
342	347		again, err := stream.renderer.Draw(composed.frame, nil)
343	348		require.NoError(t, err)
344	349		require.Empty(t, again, "renderer shadow must exactly equal the composed frame")

 }
 func TestCursorCandidateDoesNotPublishDuringPreparation(t *testing.T) {
-	ac := &attachedClient{lastCursor: cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}}
+	output := &attachmentOutput{lastCursor: cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}}
 	desired := cursorOut{row: 3, col: 4, style: 2, hasStyle: true}
-	candidate := ac.prepareCursorTail(desired, false)
+	candidate := output.prepareCursorTail(desired, false)
-	require.Equal(t, cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}, ac.lastCursor)
+	require.Equal(t, cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}, output.lastCursor)
 	require.NotEmpty(t, candidate.data)
 	require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, candidate.next)
 }
 	d, sess, ac, sends := newManualSessionWithPTYs(t)
 	healthy := ac.transport()
 	beforeCursor := cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}
-	ac.lastCursor = beforeCursor
+	ac.output.lastCursor = beforeCursor
 	ac.replaceTransport(cacheFailTransport{})
 	state := cacheState("failed", 1)
 	state.attachment = ac
 	ac.sendMu.Lock()
 	require.True(t, d.emitFrame(sess, ac, &state, composed))
-	require.Equal(t, beforeCursor, ac.lastCursor)
+	require.Equal(t, beforeCursor, ac.output.lastCursor)
 	require.Zero(t, ac.output.next)
 	probe, err := ac.output.renderer.Prepare(composed.frame, nil, false)
 	require.NoError(t, err)
 	state.attachment = ac
 	ac.sendMu.Lock()
 	require.True(t, d.emitFrame(sess, ac, &state, composed))
-	require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, ac.lastCursor)
+	require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, ac.output.lastCursor)
 	require.Equal(t, uint64(1), ac.output.next)
 	out, err := ports.UnmarshalOutput((<-sends).Payload)
 	require.NoError(t, err)
 	p.mu.Unlock()
 	beforeNext := ac.output.next
-	beforeCursor := ac.lastCursor
+	beforeCursor := ac.output.lastCursor
 	state.receipts = []damageReceipt{{pane: p, generation: capture.Generation}}
 	noByte := composedRenderFrame{
 		frame:  initial.frame,
 	require.True(t, d.emitFrame(sess, ac, &state, noByte))
 	require.Equal(t, beforeNext, ac.output.next)
-	require.Equal(t, beforeCursor, ac.lastCursor)
+	require.Equal(t, beforeCursor, ac.output.lastCursor)
 	require.Equal(t, "no-byte-committed", ac.pipelineCache.layoutFingerprint)
 	p.mu.Lock()
 	require.Empty(t, p.screen.Damage())
 	require.True(t, composed.cursor.hidden, "overlay owns cursor visibility")
 	stream := newOutputStateStream()
-	prepared, err := stream.prepare(composed.frame, composed.damage, composed.reset)
+	prepared, err := stream.prepareFrame(composed.frame, composed.damage, composed.reset, composed.cursor)
 	require.NoError(t, err)
-	terminalBytes := append(append([]byte(nil), prepared.data...), (&attachedClient{}).prepareCursorTail(composed.cursor, true).data...)
+	var outputFrame ports.Frame
+	require.NoError(t, prepared.send(0, func(frame ports.Frame) error {
+		outputFrame = frame
+		return nil
+	}))
+	output, err := ports.UnmarshalOutput(outputFrame.Payload)
+	require.NoError(t, err)
+	terminalBytes := output.Data
 	require.Equal(t, "\x1b[1;1H\x1b[0;7m tab \x1b[0m      R\x1b[2;1HAAAAAAAAAAAA\x1b[3;1HBBB┌─fl─┐BBB\x1b[4;1H───Prompt───\x1b[5;1HPROMPT\x1b[K\x1b[B\x1b[2K\x1b[7;1H\x1b[0;7m sess \x1b[0m     B\x1b[0m\x1b[?25l", string(terminalBytes))
 	client := vt.NewScreen(composed.frame.Width, composed.frame.Height)
 	client.Write(terminalBytes)
 	require.Equal(t, frameRows(composed.frame), frameRows(client.Frame))
-	prepared.draw.Commit()
-	stream.forceSnapshot = false
 	again, err := stream.renderer.Draw(composed.frame, nil)
 	require.NoError(t, err)
 	require.Empty(t, again, "renderer shadow must exactly equal the composed frame")

M internal/usecase/daemon/render_test.go +2 -2


@@ -1085,8 +1085,8 @@ func TestPaintAlignsFloatingCursorWithCommittedGeometry(t *testing.T) {
1085	1085		want := committed.translate(contentArea.X, contentArea.Y)
1086	1086		require.Contains(t, string(data), cursorCSI(want.Inner.Y+1, 1))
1087	1087		require.Contains(t, string(data), "F")
1088		-	require.Equal(t, want.Inner.Y, ac.lastCursor.row)
1089		-	require.Equal(t, want.Inner.X, ac.lastCursor.col)
	1088	+	require.Equal(t, want.Inner.Y, ac.output.lastCursor.row)
	1089	+	require.Equal(t, want.Inner.X, ac.output.lastCursor.col)
1090	1090		require.Contains(t, string(data), cursorCSI(want.Inner.Y+1, want.Inner.X+1))
1091	1091		}
1092	1092


@@ -1085,8 +1085,8 @@ func TestPaintAlignsFloatingCursorWithCommittedGeometry(t *testing.T) {
1085		want := committed.translate(contentArea.X, contentArea.Y)	1085		want := committed.translate(contentArea.X, contentArea.Y)
1086		require.Contains(t, string(data), cursorCSI(want.Inner.Y+1, 1))	1086		require.Contains(t, string(data), cursorCSI(want.Inner.Y+1, 1))
1087		require.Contains(t, string(data), "F")	1087		require.Contains(t, string(data), "F")
1088	-	require.Equal(t, want.Inner.Y, ac.lastCursor.row)	1088	+	require.Equal(t, want.Inner.Y, ac.output.lastCursor.row)
1089	-	require.Equal(t, want.Inner.X, ac.lastCursor.col)	1089	+	require.Equal(t, want.Inner.X, ac.output.lastCursor.col)
1090		require.Contains(t, string(data), cursorCSI(want.Inner.Y+1, want.Inner.X+1))	1090		require.Contains(t, string(data), cursorCSI(want.Inner.Y+1, want.Inner.X+1))
1091		}	1091		}
1092			1092

M internal/usecase/daemon/resume.go +1 -2


@@ -638,8 +638,7 @@ func (d *Daemon) resumeParkedLocked(h ports.Hello, tr ports.Transport, sz domain
638	638		// abandoned transport's output chain before binding the replacement so the
639	639		// mandatory first paint cannot be blocked by ACKs that died with the link.
640	640		ac.rebaseOutput()
641		-	ac.output.maxOutstanding = uint64(normalizeOutputWindow(h.MaxOutputInFlight))
642		-	ac.output.maxOutstandingAtomic.Store(ac.output.maxOutstanding)
	641	+	ac.output.setWindow(h.MaxOutputInFlight)
643	642		ac.replaceTransport(tr)
644	643		ac.setSize(sz)
645	644		ac.resumeToken = d.nextResumeTokenLocked()

 	// abandoned transport's output chain before binding the replacement so the
 	// mandatory first paint cannot be blocked by ACKs that died with the link.
 	ac.rebaseOutput()
-	ac.output.maxOutstanding = uint64(normalizeOutputWindow(h.MaxOutputInFlight))
+	ac.output.setWindow(h.MaxOutputInFlight)
-	ac.output.maxOutstandingAtomic.Store(ac.output.maxOutstanding)
 	ac.replaceTransport(tr)
 	ac.setSize(sz)
 	ac.resumeToken = d.nextResumeTokenLocked()

M internal/usecase/daemon/toast_damage_test.go +1 -1


@@ -124,7 +124,7 @@ func toastDamageState(notices []domain.Notification, damage []renderer.Damage, r
124	124		}
125	125		}
126	126
127		-	func replayToastFrame(t testing.T, stream outputStateStream, terminal *vt.Screen, composed composedRenderFrame) {
	127	+	func replayToastFrame(t testing.T, stream attachmentOutput, terminal *vt.Screen, composed composedRenderFrame) {
128	128		t.Helper()
129	129		data, err := stream.render(composed.frame, composed.damage, composed.reset)
130	130		require.NoError(t, err)


@@ -124,7 +124,7 @@ func toastDamageState(notices []domain.Notification, damage []renderer.Damage, r
124		}	124		}
125		}	125		}
126			126
127	-	func replayToastFrame(t testing.T, stream outputStateStream, terminal *vt.Screen, composed composedRenderFrame) {	127	+	func replayToastFrame(t testing.T, stream attachmentOutput, terminal *vt.Screen, composed composedRenderFrame) {
128		t.Helper()	128		t.Helper()
129		data, err := stream.render(composed.frame, composed.damage, composed.reset)	129		data, err := stream.render(composed.frame, composed.damage, composed.reset)
130		require.NoError(t, err)	130		require.NoError(t, err)