fix(client): harden terminal input handoff

commit: 68c86a5312bc825b0a7efcc79c565a8b17c2f0fb
author: brice · 3mo ago · 2026-07-18 12:55 UTC
parent: 3389f59
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

4 files · +228 -37

M internal/usecase/client/client.go +56 -24


@@ -1002,6 +1002,7 @@ type terminalInputPump struct {
1002	1002		delivering uint64
1003	1003		deliveringResidual bool
1004	1004		nextID uint64
	1005	+	activation uint64
1005	1006		closed bool
1006	1007		active bool
1007	1008		ready chan struct{}

@@ -1017,13 +1018,14 @@ func newTerminalInputPump(in io.Reader) *terminalInputPump {
1017	1018		space := make(chan struct{}, 1)
1018	1019		space <- struct{}{}
1019	1020		return &terminalInputPump{
1020		-	in: in,
1021		-	done: make(chan struct{}),
1022		-	active: true,
1023		-	ready: make(chan struct{}, 1),
1024		-	space: space,
1025		-	state: make(chan struct{}, 1),
1026		-	exited: make(chan struct{}),
	1021	+	in: in,
	1022	+	done: make(chan struct{}),
	1023	+	activation: 1,
	1024	+	active: true,
	1025	+	ready: make(chan struct{}, 1),
	1026	+	space: space,
	1027	+	state: make(chan struct{}, 1),
	1028	+	exited: make(chan struct{}),
1027	1029		}
1028	1030		}
1029	1031

@@ -1069,14 +1071,14 @@ func (p *terminalInputPump) signalReady() {
1069	1071
1070	1072		// enqueue retains at most one completed terminal read. The reader blocks
1071	1073		// before publishing another result, keeping handoff buffering bounded.
1072		-	func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
	1074	+	func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
1073	1075		select {
1074	1076		case <-p.done:
1075	1077		return false
1076	1078		case <-p.space:
1077	1079		}
1078	1080		p.mu.Lock()
1079		-	if p.closed \|\| !p.active {
	1081	+	if p.closed \|\| !p.active \|\| p.activation != activation {
1080	1082		p.mu.Unlock()
1081	1083		p.space <- struct{}{}
1082	1084		return false

@@ -1176,20 +1178,20 @@ func (p *terminalInputPump) signalState() {
1176	1178		}
1177	1179		}
1178	1180
1179		-	func (p *terminalInputPump) waitForActive() bool {
	1181	+	func (p *terminalInputPump) waitForActive() (uint64, bool) {
1180	1182		for {
1181	1183		p.mu.Lock()
1182		-	active, closed := p.active, p.closed
	1184	+	active, closed, activation := p.active, p.closed, p.activation
1183	1185		p.mu.Unlock()
1184	1186		if closed {
1185		-	return false
	1187	+	return 0, false
1186	1188		}
1187	1189		if active {
1188		-	return true
	1190	+	return activation, true
1189	1191		}
1190	1192		select {
1191	1193		case <-p.done:
1192		-	return false
	1194	+	return 0, false
1193	1195		case <-p.state:
1194	1196		}
1195	1197		}

@@ -1201,7 +1203,8 @@ func (p *terminalInputPump) start() {
1201	1203		defer close(p.exited)
1202	1204		buf := make([]byte, stdinBufSize)
1203	1205		for {
1204		-	if !p.waitForActive() {
	1206	+	activation, ok := p.waitForActive()
	1207	+	if !ok {
1205	1208		return
1206	1209		}
1207	1210		n, err := p.in.Read(buf)

@@ -1209,7 +1212,7 @@ func (p *terminalInputPump) start() {
1209	1212		if n > 0 {
1210	1213		result.data = append([]byte(nil), buf[:n]...)
1211	1214		}
1212		-	if !p.enqueue(result) {
	1215	+	if !p.enqueue(result, activation) {
1213	1216		if p.isClosed() {
1214	1217		return
1215	1218		}

@@ -1263,7 +1266,8 @@ func (p *terminalInputPump) suspend() {
1263	1266
1264	1267		func (p *terminalInputPump) resume() {
1265	1268		p.mu.Lock()
1266		-	if !p.closed {
	1269	+	if !p.closed && !p.active {
	1270	+	p.activation++
1267	1271		p.active = true
1268	1272		}
1269	1273		p.mu.Unlock()

@@ -1311,25 +1315,37 @@ func (p *stdinPump) run() {
1311	1315		var markers paletteMarkerScanner
1312	1316		var inputSeq uint64
1313	1317		var sendOK atomic.Bool
	1318	+	var undeliveredMu sync.Mutex
	1319	+	var undelivered []byte
1314	1320		sendOK.Store(true)
1315		-	send := func(frame ports.Frame) {
	1321	+	send := func(frame ports.Frame) bool {
	1322	+	if !sendOK.Load() {
	1323	+	return false
	1324	+	}
1316	1325		select {
1317	1326		case p.out <- frame:
1318	1327		if p.afterInputDelivered != nil {
1319	1328		p.afterInputDelivered()
1320	1329		}
	1330	+	return true
1321	1331		case <-p.ctx.Done():
1322	1332		sendOK.Store(false)
	1333	+	return false
1323	1334		}
1324	1335		}
1325		-	sendEvent := func(event paletteGenerationEvent) {
	1336	+	sendEvent := func(event paletteGenerationEvent) bool {
	1337	+	if !sendOK.Load() {
	1338	+	return false
	1339	+	}
1326	1340		select {
1327	1341		case p.paletteEvents <- event:
1328	1342		if p.afterPaletteEvent != nil {
1329	1343		p.afterPaletteEvent(event)
1330	1344		}
	1345	+	return true
1331	1346		case <-p.ctx.Done():
1332	1347		sendOK.Store(false)
	1348	+	return false
1333	1349		}
1334	1350		}
1335	1351		coalescer := newPasteCoalescer(p.clock, func(data []byte) {

@@ -1337,7 +1353,11 @@ func (p *stdinPump) run() {
1337	1353		return
1338	1354		}
1339	1355		inputSeq++
1340		-	send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: append([]byte(nil), data...)})})
	1356	+	if !send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: append([]byte(nil), data...)})}) {
	1357	+	undeliveredMu.Lock()
	1358	+	undelivered = append(undelivered, data...)
	1359	+	undeliveredMu.Unlock()
	1360	+	}
1341	1361		})
1342	1362		defer coalescer.Close()
1343	1363		sink := coalescer.Scan

@@ -1372,7 +1392,17 @@ func (p *stdinPump) run() {
1372	1392		// delivered and must not be replayed by the next attempt.
1373	1393		defer func() {
1374	1394		if p.ctx.Err() != nil {
1375		-	input.preserveResidual(consumer, markers.takePending())
	1395	+	// Closing flushes any bracketed-paste prefix or buffer through the
	1396	+	// coalescer's emit callback. Failed ordinary-input deliveries are
	1397	+	// retained ahead of a trailing undecided marker prefix, matching their
	1398	+	// original order within the raw read.
	1399	+	held := coalescer.CloseAndTakeHeld()
	1400	+	undeliveredMu.Lock()
	1401	+	residual := append([]byte(nil), undelivered...)
	1402	+	undeliveredMu.Unlock()
	1403	+	residual = append(residual, held...)
	1404	+	residual = append(residual, markers.takePending()...)
	1405	+	input.preserveResidual(consumer, residual)
1376	1406		}
1377	1407		}()
1378	1408

@@ -1465,6 +1495,11 @@ func (p *stdinPump) run() {
1465	1495		})
1466	1496		})
1467	1497		if !sendOK.Load() {
	1498	+	// Commit the source read so callbacks accepted before cancellation
	1499	+	// are never replayed. The deferred handoff retains only ordinary
	1500	+	// bytes whose callback could not be delivered, plus any undecided
	1501	+	// trailing marker prefix.
	1502	+	input.ack(consumer)
1468	1503		return
1469	1504		}
1470	1505		}

@@ -1481,9 +1516,6 @@ func (p *stdinPump) run() {
1481	1516		p.cancel()
1482	1517		return
1483	1518		}
1484		-	if !sendOK.Load() {
1485		-	return
1486		-	}
1487	1519		// Commit every callback from this raw read before observing a
1488	1520		// concurrent cancellation. Any remaining marker prefix is retained
1489	1521		// by the deferred handoff above, so reconnects never duplicate the

 	delivering         uint64
 	deliveringResidual bool
 	nextID             uint64
+	activation         uint64
 	closed             bool
 	active             bool
 	ready              chan struct{}
 	space := make(chan struct{}, 1)
 	space <- struct{}{}
 	return &terminalInputPump{
-		in:     in,
+		in:         in,
-		done:   make(chan struct{}),
+		done:       make(chan struct{}),
-		active: true,
+		activation: 1,
-		ready:  make(chan struct{}, 1),
+		active:     true,
-		space:  space,
+		ready:      make(chan struct{}, 1),
-		state:  make(chan struct{}, 1),
+		space:      space,
-		exited: make(chan struct{}),
+		state:      make(chan struct{}, 1),
+		exited:     make(chan struct{}),
 	}
 }
 // enqueue retains at most one completed terminal read. The reader blocks
 // before publishing another result, keeping handoff buffering bounded.
-func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
+func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
 	select {
 	case <-p.done:
 		return false
 	case <-p.space:
 	}
 	p.mu.Lock()
-	if p.closed || !p.active {
+	if p.closed || !p.active || p.activation != activation {
 		p.mu.Unlock()
 		p.space <- struct{}{}
 		return false
 	}
 }
-func (p *terminalInputPump) waitForActive() bool {
+func (p *terminalInputPump) waitForActive() (uint64, bool) {
 	for {
 		p.mu.Lock()
-		active, closed := p.active, p.closed
+		active, closed, activation := p.active, p.closed, p.activation
 		p.mu.Unlock()
 		if closed {
-			return false
+			return 0, false
 		}
 		if active {
-			return true
+			return activation, true
 		}
 		select {
 		case <-p.done:
-			return false
+			return 0, false
 		case <-p.state:
 		}
 	}
 			defer close(p.exited)
 			buf := make([]byte, stdinBufSize)
 			for {
-				if !p.waitForActive() {
+				activation, ok := p.waitForActive()
+				if !ok {
 					return
 				}
 				n, err := p.in.Read(buf)
 				if n > 0 {
 					result.data = append([]byte(nil), buf[:n]...)
 				}
-				if !p.enqueue(result) {
+				if !p.enqueue(result, activation) {
 					if p.isClosed() {
 						return
 					}
 func (p *terminalInputPump) resume() {
 	p.mu.Lock()
-	if !p.closed {
+	if !p.closed && !p.active {
+		p.activation++
 		p.active = true
 	}
 	p.mu.Unlock()
 	var markers paletteMarkerScanner
 	var inputSeq uint64
 	var sendOK atomic.Bool
+	var undeliveredMu sync.Mutex
+	var undelivered []byte
 	sendOK.Store(true)
-	send := func(frame ports.Frame) {
+	send := func(frame ports.Frame) bool {
+		if !sendOK.Load() {
+			return false
+		}
 		select {
 		case p.out <- frame:
 			if p.afterInputDelivered != nil {
 				p.afterInputDelivered()
 			}
+			return true
 		case <-p.ctx.Done():
 			sendOK.Store(false)
+			return false
 		}
 	}
-	sendEvent := func(event paletteGenerationEvent) {
+	sendEvent := func(event paletteGenerationEvent) bool {
+		if !sendOK.Load() {
+			return false
+		}
 		select {
 		case p.paletteEvents <- event:
 			if p.afterPaletteEvent != nil {
 				p.afterPaletteEvent(event)
 			}
+			return true
 		case <-p.ctx.Done():
 			sendOK.Store(false)
+			return false
 		}
 	}
 	coalescer := newPasteCoalescer(p.clock, func(data []byte) {
 			return
 		}
 		inputSeq++
-		send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: append([]byte(nil), data...)})})
+		if !send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: append([]byte(nil), data...)})}) {
+			undeliveredMu.Lock()
+			undelivered = append(undelivered, data...)
+			undeliveredMu.Unlock()
+		}
 	})
 	defer coalescer.Close()
 	sink := coalescer.Scan
 	// delivered and must not be replayed by the next attempt.
 	defer func() {
 		if p.ctx.Err() != nil {
-			input.preserveResidual(consumer, markers.takePending())
+			// Closing flushes any bracketed-paste prefix or buffer through the
+			// coalescer's emit callback. Failed ordinary-input deliveries are
+			// retained ahead of a trailing undecided marker prefix, matching their
+			// original order within the raw read.
+			held := coalescer.CloseAndTakeHeld()
+			undeliveredMu.Lock()
+			residual := append([]byte(nil), undelivered...)
+			undeliveredMu.Unlock()
+			residual = append(residual, held...)
+			residual = append(residual, markers.takePending()...)
+			input.preserveResidual(consumer, residual)
 		}
 	}()
 					})
 				})
 				if !sendOK.Load() {
+					// Commit the source read so callbacks accepted before cancellation
+					// are never replayed. The deferred handoff retains only ordinary
+					// bytes whose callback could not be delivered, plus any undecided
+					// trailing marker prefix.
+					input.ack(consumer)
 					return
 				}
 			}
 				p.cancel()
 				return
 			}
-			if !sendOK.Load() {
-				return
-			}
 			// Commit every callback from this raw read before observing a
 			// concurrent cancellation. Any remaining marker prefix is retained
 			// by the deferred handoff above, so reconnects never duplicate the

M internal/usecase/client/client_test.go +3 -6


@@ -230,8 +230,6 @@ func TestRunBoundsPreWelcomeOperations(t *testing.T) {
230	230		closed := make(chan struct{})
231	231		operationExited := make(chan struct{})
232	232		tr := portsmocks.NewMockTransport(t)
233		-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
234		-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
235	233		tr.EXPECT().Close().Run(func() { close(closed) }).Return(nil).Once()
236	234
237	235		block := func() error {

@@ -294,6 +292,7 @@ func TestRunBoundsPreWelcomeOperations(t *testing.T) {
294	292		case <-time.After(time.Second):
295	293		t.Fatal("pre-Welcome operation leaked after Run returned")
296	294		}
	295	+	tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
297	296		})
298	297		}
299	298		}

@@ -427,8 +426,6 @@ func TestAttachVersionMismatch(t *testing.T) {
427	426		// EnterRaw must NOT be called on the error-before-welcome path.
428	427
429	428		tr := portsmocks.NewMockTransport(t)
430		-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
431		-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
432	429		tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
433	430		tr.EXPECT().Recv().Return(
434	431		frameOf(ports.MsgError, ports.MarshalErrorMsg(ports.ErrorMsg{Code: ports.ErrVersionMismatch, Text: "version mismatch"})),

@@ -441,6 +438,7 @@ func TestAttachVersionMismatch(t *testing.T) {
441	438		var pe *client.ProtocolError
442	439		require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
443	440		require.Equal(t, ports.ErrVersionMismatch, pe.Code)
	441	+	tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
444	442		}
445	443
446	444		func TestAttachRestoredOnRecvErrorMidStream(t *testing.T) {

@@ -711,8 +709,6 @@ func TestRunDoesNotEnterRawBeforePreWelcomeError(t *testing.T) {
711	709		// EnterRaw must NOT be called when the daemon rejects Hello before Welcome.
712	710
713	711		tr := portsmocks.NewMockTransport(t)
714		-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
715		-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
716	712		tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
717	713		tr.EXPECT().Recv().Return(
718	714		frameOf(ports.MsgError, ports.MarshalErrorMsg(ports.ErrorMsg{Code: ports.ErrVersionMismatch, Text: "version mismatch"})),

@@ -726,6 +722,7 @@ func TestRunDoesNotEnterRawBeforePreWelcomeError(t *testing.T) {
726	722		require.Error(t, err)
727	723		var pe *client.ProtocolError
728	724		require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
	725	+	tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
729	726		}
730	727
731	728		func TestRunPhaseASingleAttempt(t *testing.T) {

 			closed := make(chan struct{})
 			operationExited := make(chan struct{})
 			tr := portsmocks.NewMockTransport(t)
-			tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
-			tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
 			tr.EXPECT().Close().Run(func() { close(closed) }).Return(nil).Once()
 			block := func() error {
 			case <-time.After(time.Second):
 				t.Fatal("pre-Welcome operation leaked after Run returned")
 			}
+			tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
 		})
 	}
 }
 	// EnterRaw must NOT be called on the error-before-welcome path.
 	tr := portsmocks.NewMockTransport(t)
-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
 	tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
 	tr.EXPECT().Recv().Return(
 		frameOf(ports.MsgError, ports.MarshalErrorMsg(ports.ErrorMsg{Code: ports.ErrVersionMismatch, Text: "version mismatch"})),
 	var pe *client.ProtocolError
 	require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
 	require.Equal(t, ports.ErrVersionMismatch, pe.Code)
+	tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
 }
 func TestAttachRestoredOnRecvErrorMidStream(t *testing.T) {
 	// EnterRaw must NOT be called when the daemon rejects Hello before Welcome.
 	tr := portsmocks.NewMockTransport(t)
-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
-	tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
 	tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
 	tr.EXPECT().Recv().Return(
 		frameOf(ports.MsgError, ports.MarshalErrorMsg(ports.ErrorMsg{Code: ports.ErrVersionMismatch, Text: "version mismatch"})),
 	require.Error(t, err)
 	var pe *client.ProtocolError
 	require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
+	tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
 }
 func TestRunPhaseASingleAttempt(t *testing.T) {

M internal/usecase/client/paste_coalescer.go +15 -3


@@ -158,15 +158,27 @@ func (c *pasteCoalescer) Buffering() bool {
158	158		return c.buffering \|\| len(c.pending) > 0
159	159		}
160	160
161		-	// Close stops any pending timers and their goroutines. Held bytes are dropped:
162		-	// Close runs only as the stdin pump unwinds on detach, when there is nowhere
163		-	// left to deliver them.
	161	+	// Close stops any pending timers and their goroutines. Callers which hand input
	162	+	// to a replacement must first use CloseAndTakeHeld so undecided bytes survive.
164	163		func (c *pasteCoalescer) Close() {
	164	+	_ = c.CloseAndTakeHeld()
	165	+	}
	166	+
	167	+	// CloseAndTakeHeld stops the coalescer and returns ordinary bytes withheld while
	168	+	// deciding whether an opening marker or complete bracketed paste would arrive.
	169	+	// The returned bytes retain their original order and are removed from c.
	170	+	func (c *pasteCoalescer) CloseAndTakeHeld() []byte {
165	171		c.mu.Lock()
166	172		defer c.mu.Unlock()
167	173		c.closed = true
168	174		c.stopFlushTimer()
169	175		c.stopSafetyTimer()
	176	+	held := append([]byte(nil), c.pending...)
	177	+	held = append(held, c.buf...)
	178	+	c.pending = nil
	179	+	c.buf = nil
	180	+	c.buffering = false
	181	+	return held
170	182		}
171	183
172	184		func (c *pasteCoalescer) startFlushTimer() {

 	return c.buffering || len(c.pending) > 0
 }
-// Close stops any pending timers and their goroutines. Held bytes are dropped:
+// Close stops any pending timers and their goroutines. Callers which hand input
-// Close runs only as the stdin pump unwinds on detach, when there is nowhere
+// to a replacement must first use CloseAndTakeHeld so undecided bytes survive.
-// left to deliver them.
 func (c *pasteCoalescer) Close() {
+	_ = c.CloseAndTakeHeld()
+}
+// CloseAndTakeHeld stops the coalescer and returns ordinary bytes withheld while
+// deciding whether an opening marker or complete bracketed paste would arrive.
+// The returned bytes retain their original order and are removed from c.
+func (c *pasteCoalescer) CloseAndTakeHeld() []byte {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	c.closed = true
 	c.stopFlushTimer()
 	c.stopSafetyTimer()
+	held := append([]byte(nil), c.pending...)
+	held = append(held, c.buf...)
+	c.pending = nil
+	c.buf = nil
+	c.buffering = false
+	return held
 }
 func (c *pasteCoalescer) startFlushTimer() {

M internal/usecase/client/terminal_input_pump_test.go +154 -4


@@ -7,6 +7,7 @@ import (
7	7		"sync"
8	8		"sync/atomic"
9	9		"testing"
	10	+	"time"
10	11
11	12		"github.com/stretchr/testify/require"
12	13

@@ -71,6 +72,37 @@ func (r *lateStopReader) Read(p []byte) (int, error) {
71	72		return copy(p, result.data), result.err
72	73		}
73	74
	75	+	func requireSignal(t *testing.T, ch <-chan struct{}, message string) {
	76	+	t.Helper()
	77	+	select {
	78	+	case <-ch:
	79	+	case <-time.After(time.Second):
	80	+	t.Fatal(message)
	81	+	}
	82	+	}
	83	+
	84	+	func requireFrame(t *testing.T, ch <-chan ports.Frame, message string) ports.Frame {
	85	+	t.Helper()
	86	+	select {
	87	+	case frame := <-ch:
	88	+	return frame
	89	+	case <-time.After(time.Second):
	90	+	t.Fatal(message)
	91	+	return ports.Frame{}
	92	+	}
	93	+	}
	94	+
	95	+	func requireTimer(t testing.T, ch <-chan attachPaletteTimer, message string) *attachPaletteTimer {
	96	+	t.Helper()
	97	+	select {
	98	+	case timer := <-ch:
	99	+	return timer
	100	+	case <-time.After(time.Second):
	101	+	t.Fatal(message)
	102	+	return nil
	103	+	}
	104	+	}
	105	+
74	106		func TestPaletteSlotRejectsOutOfRangeIntBeforeNarrowing(t *testing.T) {
75	107		for _, tc := range []struct {
76	108		slot int

@@ -140,7 +172,7 @@ func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T)
140	172		// Both cancellation and a terminal read are ready before the old scanner
141	173		// tries to dequeue. It must surrender the read to its replacement.
142	174		want := []byte("first\x00second\xff")
143		-	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})
	175	+	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)}, 1)
144	176
145	177		var activeGeneration atomic.Uint64
146	178		cancelledCtx, cancelCancelled := context.WithCancel(context.Background())

@@ -187,7 +219,7 @@ func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T)
187	219		func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
188	220		input := newTerminalInputPump(nil)
189	221		want := []byte("first\x00second\xff")
190		-	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})
	222	+	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)}, 1)
191	223
192	224		ctx, cancel := context.WithCancel(context.Background())
193	225		consumer := input.claim()

@@ -223,7 +255,7 @@ func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
223	255
224	256		func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t *testing.T) {
225	257		input := newTerminalInputPump(nil)
226		-	input.enqueue(terminalReadResult{data: []byte("\x1b")})
	258	+	input.enqueue(terminalReadResult{data: []byte("\x1b")}, 1)
227	259		var activeGeneration atomic.Uint64
228	260
229	261		oldCtx, cancelOld := context.WithCancel(context.Background())

@@ -267,9 +299,127 @@ func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t
267	299		<-newDone
268	300		}
269	301
	302	+	func TestTerminalInputPumpCancellationAfterInputBeforeOSCPreservesOnlyUndeliveredSuffix(t *testing.T) {
	303	+	input := newTerminalInputPump(nil)
	304	+	input.enqueue(terminalReadResult{data: []byte("x\x1b]10;rgb:ffff/0000/0000\x07y")}, 1)
	305	+	var activeGeneration atomic.Uint64
	306	+	ctx, cancel := context.WithCancel(context.Background())
	307	+	delivered := make(chan struct{})
	308	+	done := make(chan struct{})
	309	+	out := make(chan ports.Frame, 1)
	310	+	go func() {
	311	+	(&stdinPump{
	312	+	ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
	313	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	314	+	activeGeneration: &activeGeneration,
	315	+	afterInputDelivered: func() {
	316	+	cancel()
	317	+	close(delivered)
	318	+	},
	319	+	}).run()
	320	+	close(done)
	321	+	}()
	322	+	frame := requireFrame(t, out, "timed out waiting for delivered input prefix")
	323	+	got, err := ports.UnmarshalInput(frame.Payload)
	324	+	require.NoError(t, err)
	325	+	require.Equal(t, []byte("x"), got.Data)
	326	+	requireSignal(t, delivered, "timed out waiting for cancellation after input delivery")
	327	+	requireSignal(t, done, "timed out waiting for cancelled stdin pump")
	328	+
	329	+	replacementCtx, cancelReplacement := context.WithCancel(context.Background())
	330	+	replacementOut := make(chan ports.Frame, 1)
	331	+	replacementDone := make(chan struct{})
	332	+	go func() {
	333	+	(&stdinPump{
	334	+	ctx: replacementCtx, cancel: cancelReplacement, input: input,
	335	+	out: replacementOut, clock: newAttachPaletteClock(),
	336	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	337	+	activeGeneration: &activeGeneration,
	338	+	}).run()
	339	+	close(replacementDone)
	340	+	}()
	341	+	frame = requireFrame(t, replacementOut, "timed out waiting for undelivered input suffix")
	342	+	got, err = ports.UnmarshalInput(frame.Payload)
	343	+	require.NoError(t, err)
	344	+	require.Equal(t, []byte("y"), got.Data, "replacement must receive only the undelivered ordinary suffix")
	345	+
	346	+	cancelReplacement()
	347	+	requireSignal(t, replacementDone, "timed out waiting for replacement stdin pump shutdown")
	348	+	}
	349	+
	350	+	func TestTerminalInputPumpCancellationPreservesCoalescerHeldSuffix(t *testing.T) {
	351	+	input := newTerminalInputPump(nil)
	352	+	held := ports.BracketedPasteOpenMarker[:3]
	353	+	raw := append([]byte("x\x1b]10;rgb:ffff/0000/0000\x07"), held...)
	354	+	input.enqueue(terminalReadResult{data: raw}, 1)
	355	+	var activeGeneration atomic.Uint64
	356	+	ctx, cancel := context.WithCancel(context.Background())
	357	+	out := make(chan ports.Frame, 1)
	358	+	done := make(chan struct{})
	359	+	go func() {
	360	+	(&stdinPump{
	361	+	ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
	362	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	363	+	activeGeneration: &activeGeneration,
	364	+	afterInputDelivered: cancel,
	365	+	}).run()
	366	+	close(done)
	367	+	}()
	368	+	frame := requireFrame(t, out, "timed out waiting for delivered input prefix")
	369	+	got, err := ports.UnmarshalInput(frame.Payload)
	370	+	require.NoError(t, err)
	371	+	require.Equal(t, []byte("x"), got.Data)
	372	+	requireSignal(t, done, "timed out waiting for cancelled stdin pump")
	373	+
	374	+	replacementCtx, cancelReplacement := context.WithCancel(context.Background())
	375	+	replacementClock := newAttachPaletteClock()
	376	+	replacementOut := make(chan ports.Frame, 1)
	377	+	replacementDone := make(chan struct{})
	378	+	go func() {
	379	+	(&stdinPump{
	380	+	ctx: replacementCtx, cancel: cancelReplacement, input: input,
	381	+	out: replacementOut, clock: replacementClock,
	382	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	383	+	activeGeneration: &activeGeneration,
	384	+	}).run()
	385	+	close(replacementDone)
	386	+	}()
	387	+	requireTimer(t, replacementClock.timers, "timed out waiting for paste-prefix ambiguity timer").fire()
	388	+	frame = requireFrame(t, replacementOut, "timed out waiting for coalescer-held suffix")
	389	+	got, err = ports.UnmarshalInput(frame.Payload)
	390	+	require.NoError(t, err)
	391	+	require.Equal(t, held, got.Data, "replacement must receive the coalescer-held suffix exactly once")
	392	+
	393	+	cancelReplacement()
	394	+	requireSignal(t, replacementDone, "timed out waiting for replacement stdin pump shutdown")
	395	+	}
	396	+
	397	+	func TestTerminalInputPumpRejectsReadFromPriorActivation(t *testing.T) {
	398	+	reader := newLateStopReader()
	399	+	input := newTerminalInputPump(reader)
	400	+	input.start()
	401	+	requireSignal(t, reader.started, "timed out waiting for initial activation Read")
	402	+
	403	+	input.suspend()
	404	+	input.resume()
	405	+	reader.release <- terminalReadResult{data: []byte("stale")}
	406	+
	407	+	// The old read must be discarded after resume, then the reader may begin a
	408	+	// new Read for the current activation.
	409	+	requireSignal(t, reader.started, "timed out waiting for current activation Read")
	410	+	consumer := input.claim()
	411	+	_, ok := input.take(context.Background(), consumer)
	412	+	require.False(t, ok, "a Read authorized before suspend/resume must not reach the new activation")
	413	+	input.revoke(consumer)
	414	+
	415	+	input.stop()
	416	+	reader.release <- terminalReadResult{}
	417	+	requireSignal(t, input.exited, "timed out waiting for terminal input pump shutdown")
	418	+	}
	419	+
270	420		func TestTerminalInputPumpCancellationAfterDeliveredCallbackDoesNotReplay(t *testing.T) {
271	421		input := newTerminalInputPump(nil)
272		-	input.enqueue(terminalReadResult{data: []byte("x")})
	422	+	input.enqueue(terminalReadResult{data: []byte("x")}, 1)
273	423		var activeGeneration atomic.Uint64
274	424		ctx, cancel := context.WithCancel(context.Background())
275	425		delivered := make(chan struct{})


@@ -7,6 +7,7 @@ import (
7		"sync"	7		"sync"
8		"sync/atomic"	8		"sync/atomic"
9		"testing"	9		"testing"
			10	+	"time"
10			11
11		"github.com/stretchr/testify/require"	12		"github.com/stretchr/testify/require"
12			13

@@ -71,6 +72,37 @@ func (r *lateStopReader) Read(p []byte) (int, error) {
71		return copy(p, result.data), result.err	72		return copy(p, result.data), result.err
72		}	73		}
73			74
			75	+	func requireSignal(t *testing.T, ch <-chan struct{}, message string) {
			76	+	t.Helper()
			77	+	select {
			78	+	case <-ch:
			79	+	case <-time.After(time.Second):
			80	+	t.Fatal(message)
			81	+	}
			82	+	}
			83	+
			84	+	func requireFrame(t *testing.T, ch <-chan ports.Frame, message string) ports.Frame {
			85	+	t.Helper()
			86	+	select {
			87	+	case frame := <-ch:
			88	+	return frame
			89	+	case <-time.After(time.Second):
			90	+	t.Fatal(message)
			91	+	return ports.Frame{}
			92	+	}
			93	+	}
			94	+
			95	+	func requireTimer(t testing.T, ch <-chan attachPaletteTimer, message string) *attachPaletteTimer {
			96	+	t.Helper()
			97	+	select {
			98	+	case timer := <-ch:
			99	+	return timer
			100	+	case <-time.After(time.Second):
			101	+	t.Fatal(message)
			102	+	return nil
			103	+	}
			104	+	}
			105	+
74		func TestPaletteSlotRejectsOutOfRangeIntBeforeNarrowing(t *testing.T) {	106		func TestPaletteSlotRejectsOutOfRangeIntBeforeNarrowing(t *testing.T) {
75		for _, tc := range []struct {	107		for _, tc := range []struct {
76		slot int	108		slot int

@@ -140,7 +172,7 @@ func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T)
140		// Both cancellation and a terminal read are ready before the old scanner	172		// Both cancellation and a terminal read are ready before the old scanner
141		// tries to dequeue. It must surrender the read to its replacement.	173		// tries to dequeue. It must surrender the read to its replacement.
142		want := []byte("first\x00second\xff")	174		want := []byte("first\x00second\xff")
143	-	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})	175	+	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)}, 1)
144			176
145		var activeGeneration atomic.Uint64	177		var activeGeneration atomic.Uint64
146		cancelledCtx, cancelCancelled := context.WithCancel(context.Background())	178		cancelledCtx, cancelCancelled := context.WithCancel(context.Background())

@@ -187,7 +219,7 @@ func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T)
187		func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {	219		func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
188		input := newTerminalInputPump(nil)	220		input := newTerminalInputPump(nil)
189		want := []byte("first\x00second\xff")	221		want := []byte("first\x00second\xff")
190	-	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})	222	+	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)}, 1)
191			223
192		ctx, cancel := context.WithCancel(context.Background())	224		ctx, cancel := context.WithCancel(context.Background())
193		consumer := input.claim()	225		consumer := input.claim()

@@ -223,7 +255,7 @@ func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
223			255
224		func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t *testing.T) {	256		func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t *testing.T) {
225		input := newTerminalInputPump(nil)	257		input := newTerminalInputPump(nil)
226	-	input.enqueue(terminalReadResult{data: []byte("\x1b")})	258	+	input.enqueue(terminalReadResult{data: []byte("\x1b")}, 1)
227		var activeGeneration atomic.Uint64	259		var activeGeneration atomic.Uint64
228			260
229		oldCtx, cancelOld := context.WithCancel(context.Background())	261		oldCtx, cancelOld := context.WithCancel(context.Background())

@@ -267,9 +299,127 @@ func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t
267		<-newDone	299		<-newDone
268		}	300		}
269			301
			302	+	func TestTerminalInputPumpCancellationAfterInputBeforeOSCPreservesOnlyUndeliveredSuffix(t *testing.T) {
			303	+	input := newTerminalInputPump(nil)
			304	+	input.enqueue(terminalReadResult{data: []byte("x\x1b]10;rgb:ffff/0000/0000\x07y")}, 1)
			305	+	var activeGeneration atomic.Uint64
			306	+	ctx, cancel := context.WithCancel(context.Background())
			307	+	delivered := make(chan struct{})
			308	+	done := make(chan struct{})
			309	+	out := make(chan ports.Frame, 1)
			310	+	go func() {
			311	+	(&stdinPump{
			312	+	ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
			313	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
			314	+	activeGeneration: &activeGeneration,
			315	+	afterInputDelivered: func() {
			316	+	cancel()
			317	+	close(delivered)
			318	+	},
			319	+	}).run()
			320	+	close(done)
			321	+	}()
			322	+	frame := requireFrame(t, out, "timed out waiting for delivered input prefix")
			323	+	got, err := ports.UnmarshalInput(frame.Payload)
			324	+	require.NoError(t, err)
			325	+	require.Equal(t, []byte("x"), got.Data)
			326	+	requireSignal(t, delivered, "timed out waiting for cancellation after input delivery")
			327	+	requireSignal(t, done, "timed out waiting for cancelled stdin pump")
			328	+
			329	+	replacementCtx, cancelReplacement := context.WithCancel(context.Background())
			330	+	replacementOut := make(chan ports.Frame, 1)
			331	+	replacementDone := make(chan struct{})
			332	+	go func() {
			333	+	(&stdinPump{
			334	+	ctx: replacementCtx, cancel: cancelReplacement, input: input,
			335	+	out: replacementOut, clock: newAttachPaletteClock(),
			336	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
			337	+	activeGeneration: &activeGeneration,
			338	+	}).run()
			339	+	close(replacementDone)
			340	+	}()
			341	+	frame = requireFrame(t, replacementOut, "timed out waiting for undelivered input suffix")
			342	+	got, err = ports.UnmarshalInput(frame.Payload)
			343	+	require.NoError(t, err)
			344	+	require.Equal(t, []byte("y"), got.Data, "replacement must receive only the undelivered ordinary suffix")
			345	+
			346	+	cancelReplacement()
			347	+	requireSignal(t, replacementDone, "timed out waiting for replacement stdin pump shutdown")
			348	+	}
			349	+
			350	+	func TestTerminalInputPumpCancellationPreservesCoalescerHeldSuffix(t *testing.T) {
			351	+	input := newTerminalInputPump(nil)
			352	+	held := ports.BracketedPasteOpenMarker[:3]
			353	+	raw := append([]byte("x\x1b]10;rgb:ffff/0000/0000\x07"), held...)
			354	+	input.enqueue(terminalReadResult{data: raw}, 1)
			355	+	var activeGeneration atomic.Uint64
			356	+	ctx, cancel := context.WithCancel(context.Background())
			357	+	out := make(chan ports.Frame, 1)
			358	+	done := make(chan struct{})
			359	+	go func() {
			360	+	(&stdinPump{
			361	+	ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
			362	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
			363	+	activeGeneration: &activeGeneration,
			364	+	afterInputDelivered: cancel,
			365	+	}).run()
			366	+	close(done)
			367	+	}()
			368	+	frame := requireFrame(t, out, "timed out waiting for delivered input prefix")
			369	+	got, err := ports.UnmarshalInput(frame.Payload)
			370	+	require.NoError(t, err)
			371	+	require.Equal(t, []byte("x"), got.Data)
			372	+	requireSignal(t, done, "timed out waiting for cancelled stdin pump")
			373	+
			374	+	replacementCtx, cancelReplacement := context.WithCancel(context.Background())
			375	+	replacementClock := newAttachPaletteClock()
			376	+	replacementOut := make(chan ports.Frame, 1)
			377	+	replacementDone := make(chan struct{})
			378	+	go func() {
			379	+	(&stdinPump{
			380	+	ctx: replacementCtx, cancel: cancelReplacement, input: input,
			381	+	out: replacementOut, clock: replacementClock,
			382	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
			383	+	activeGeneration: &activeGeneration,
			384	+	}).run()
			385	+	close(replacementDone)
			386	+	}()
			387	+	requireTimer(t, replacementClock.timers, "timed out waiting for paste-prefix ambiguity timer").fire()
			388	+	frame = requireFrame(t, replacementOut, "timed out waiting for coalescer-held suffix")
			389	+	got, err = ports.UnmarshalInput(frame.Payload)
			390	+	require.NoError(t, err)
			391	+	require.Equal(t, held, got.Data, "replacement must receive the coalescer-held suffix exactly once")
			392	+
			393	+	cancelReplacement()
			394	+	requireSignal(t, replacementDone, "timed out waiting for replacement stdin pump shutdown")
			395	+	}
			396	+
			397	+	func TestTerminalInputPumpRejectsReadFromPriorActivation(t *testing.T) {
			398	+	reader := newLateStopReader()
			399	+	input := newTerminalInputPump(reader)
			400	+	input.start()
			401	+	requireSignal(t, reader.started, "timed out waiting for initial activation Read")
			402	+
			403	+	input.suspend()
			404	+	input.resume()
			405	+	reader.release <- terminalReadResult{data: []byte("stale")}
			406	+
			407	+	// The old read must be discarded after resume, then the reader may begin a
			408	+	// new Read for the current activation.
			409	+	requireSignal(t, reader.started, "timed out waiting for current activation Read")
			410	+	consumer := input.claim()
			411	+	_, ok := input.take(context.Background(), consumer)
			412	+	require.False(t, ok, "a Read authorized before suspend/resume must not reach the new activation")
			413	+	input.revoke(consumer)
			414	+
			415	+	input.stop()
			416	+	reader.release <- terminalReadResult{}
			417	+	requireSignal(t, input.exited, "timed out waiting for terminal input pump shutdown")
			418	+	}
			419	+
270		func TestTerminalInputPumpCancellationAfterDeliveredCallbackDoesNotReplay(t *testing.T) {	420		func TestTerminalInputPumpCancellationAfterDeliveredCallbackDoesNotReplay(t *testing.T) {
271		input := newTerminalInputPump(nil)	421		input := newTerminalInputPump(nil)
272	-	input.enqueue(terminalReadResult{data: []byte("x")})	422	+	input.enqueue(terminalReadResult{data: []byte("x")}, 1)
273		var activeGeneration atomic.Uint64	423		var activeGeneration atomic.Uint64
274		ctx, cancel := context.WithCancel(context.Background())	424		ctx, cancel := context.WithCancel(context.Background())
275		delivered := make(chan struct{})	425		delivered := make(chan struct{})