c5fa06a fix(client): preserve scanner residuals across reconnect · vev

M internal/usecase/client/client.go +119 -32


@@ -503,6 +503,16 @@ func boundedPreWelcome(ctx context.Context, clk ports.Clock, transport ports.Tra
503	503		}
504	504		}
505	505
	506	+	// paletteSlot validates the scanner's signed slot before narrowing it for the
	507	+	// wire accumulator. Keep this boundary even though the in-tree Scanner also
	508	+	// validates slots: callbacks are an interface boundary.
	509	+	func paletteSlot(slot int) (uint8, bool) {
	510	+	if slot < 0 \|\| slot > 15 {
	511	+	return 0, false
	512	+	}
	513	+	return uint8(slot), true
	514	+	}
	515	+
506	516		// DetectTrueColor reports whether TERM/COLORTERM advertise direct color support.
507	517		func DetectTrueColor(termEnv, colorTerm string) bool {
508	518		switch strings.ToLower(strings.TrimSpace(colorTerm)) {

@@ -980,19 +990,25 @@ type terminalInputPump struct {
980	990		in io.Reader
981	991		done chan struct{}
982	992
983		-	mu sync.Mutex
984		-	pending *terminalReadResult
985		-	consumer uint64
986		-	delivering uint64
987		-	nextID uint64
988		-	closed bool
989		-	active bool
990		-	ready chan struct{}
991		-	space chan struct{}
992		-	state chan struct{}
993		-	exited chan struct{}
994		-	startMu sync.Once
995		-	stopMu sync.Once
	993	+	mu sync.Mutex
	994	+	// residual holds scanner bytes that were read but not delivered when an
	995	+	// attempt is cancelled. It is consumed before the next terminal Read so a
	996	+	// replacement scanner neither loses a partial marker nor replays bytes
	997	+	// whose callbacks already ran.
	998	+	residual []byte
	999	+	pending *terminalReadResult
	1000	+	consumer uint64
	1001	+	delivering uint64
	1002	+	deliveringResidual bool
	1003	+	nextID uint64
	1004	+	closed bool
	1005	+	active bool
	1006	+	ready chan struct{}
	1007	+	space chan struct{}
	1008	+	state chan struct{}
	1009	+	exited chan struct{}
	1010	+	startMu sync.Once
	1011	+	stopMu sync.Once
996	1012		}
997	1013
998	1014		func newTerminalInputPump(in io.Reader) *terminalInputPump {

@@ -1033,6 +1049,7 @@ func (p *terminalInputPump) revoke(consumer uint64) {
1033	1049		// pending and make it available to its replacement.
1034	1050		if p.delivering == consumer {
1035	1051		p.delivering = 0
	1052	+	p.deliveringResidual = false
1036	1053		}
1037	1054		p.mu.Unlock()
1038	1055		p.signalReady()

@@ -1070,20 +1087,48 @@ func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
1070	1087		// scanner delivery leaves the exact bytes available to the next attempt.
1071	1088		func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
1072	1089		p.mu.Lock()
1073		-	if ctx.Err() != nil \|\| p.consumer != consumer \|\| p.pending == nil \|\| p.delivering != 0 {
1074		-	pending := p.pending != nil && p.delivering == 0
	1090	+	if ctx.Err() != nil \|\| p.consumer != consumer \|\| p.delivering != 0 {
	1091	+	ready := (len(p.residual) != 0 \|\| p.pending != nil) && p.delivering == 0
1075	1092		p.mu.Unlock()
1076		-	if pending {
	1093	+	if ready {
1077	1094		p.signalReady()
1078	1095		}
1079	1096		return terminalReadResult{}, false
1080	1097		}
	1098	+	if len(p.residual) != 0 {
	1099	+	result := terminalReadResult{data: append([]byte(nil), p.residual...)}
	1100	+	p.delivering = consumer
	1101	+	p.deliveringResidual = true
	1102	+	p.mu.Unlock()
	1103	+	return result, true
	1104	+	}
	1105	+	if p.pending == nil {
	1106	+	p.mu.Unlock()
	1107	+	return terminalReadResult{}, false
	1108	+	}
1081	1109		result := *p.pending
1082	1110		p.delivering = consumer
	1111	+	p.deliveringResidual = false
1083	1112		p.mu.Unlock()
1084	1113		return result, true
1085	1114		}
1086	1115
	1116	+	// preserveResidual records the undecided marker prefix after callbacks for the
	1117	+	// rest of its read have completed. It is intentionally bounded by the marker
	1118	+	// scanner's fixed response prefixes and is replayed before later terminal
	1119	+	// reads on a replacement attempt.
	1120	+	func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
	1121	+	if len(data) == 0 {
	1122	+	return
	1123	+	}
	1124	+	p.mu.Lock()
	1125	+	defer p.mu.Unlock()
	1126	+	if p.consumer != consumer {
	1127	+	return
	1128	+	}
	1129	+	p.residual = append(p.residual[:0], data...)
	1130	+	}
	1131	+
1087	1132		// ack commits scanner delivery of a leased read and lets the lifecycle reader
1088	1133		// accept the next result. Only the consumer that holds the lease can ack it.
1089	1134		func (p *terminalInputPump) ack(consumer uint64) {

@@ -1092,9 +1137,22 @@ func (p *terminalInputPump) ack(consumer uint64) {
1092	1137		p.mu.Unlock()
1093	1138		return
1094	1139		}
1095		-	p.pending = nil
	1140	+	wasResidual := p.deliveringResidual
	1141	+	if wasResidual {
	1142	+	p.residual = nil
	1143	+	} else {
	1144	+	p.pending = nil
	1145	+	}
1096	1146		p.delivering = 0
	1147	+	p.deliveringResidual = false
	1148	+	more := len(p.residual) != 0 \|\| p.pending != nil
1097	1149		p.mu.Unlock()
	1150	+	if wasResidual {
	1151	+	if more {
	1152	+	p.signalReady()
	1153	+	}
	1154	+	return
	1155	+	}
1098	1156		p.space <- struct{}{}
1099	1157		}
1100	1158

@@ -1183,8 +1241,10 @@ func (p *terminalInputPump) suspend() {
1183	1241		p.mu.Lock()
1184	1242		p.active = false
1185	1243		discarded := p.pending != nil
	1244	+	p.residual = nil
1186	1245		p.pending = nil
1187	1246		p.delivering = 0
	1247	+	p.deliveringResidual = false
1188	1248		p.mu.Unlock()
1189	1249		if discarded {
1190	1250		select {

@@ -1224,19 +1284,20 @@ func (p *terminalInputPump) stop() {
1224	1284		// terminal directly, publishes a Theme, or writes terminal output;
1225	1285		// attachAttempt owns those operations.
1226	1286		type stdinPump struct {
1227		-	ctx context.Context
1228		-	cancel context.CancelFunc
1229		-	in io.Reader // compatibility for isolated scanner tests
1230		-	input *terminalInputPump
1231		-	consumer uint64
1232		-	out chan<- ports.Frame
1233		-	clock ports.Clock
1234		-	clipboard ports.ClipboardReader
1235		-	logger *slog.Logger
1236		-	paletteEvents chan<- paletteGenerationEvent
1237		-	activeGeneration *atomic.Uint64
1238		-	afterPaletteEvent func(paletteGenerationEvent) // test synchronization hook
1239		-	afterInputTake func() // test synchronization hook
	1287	+	ctx context.Context
	1288	+	cancel context.CancelFunc
	1289	+	in io.Reader // compatibility for isolated scanner tests
	1290	+	input *terminalInputPump
	1291	+	consumer uint64
	1292	+	out chan<- ports.Frame
	1293	+	clock ports.Clock
	1294	+	clipboard ports.ClipboardReader
	1295	+	logger *slog.Logger
	1296	+	paletteEvents chan<- paletteGenerationEvent
	1297	+	activeGeneration *atomic.Uint64
	1298	+	afterPaletteEvent func(paletteGenerationEvent) // test synchronization hook
	1299	+	afterInputTake func() // test synchronization hook
	1300	+	afterInputDelivered func() // test synchronization hook
1240	1301		}
1241	1302
1242	1303		func (p *stdinPump) run() {

@@ -1249,6 +1310,9 @@ func (p *stdinPump) run() {
1249	1310		send := func(frame ports.Frame) {
1250	1311		select {
1251	1312		case p.out <- frame:
	1313	+	if p.afterInputDelivered != nil {
	1314	+	p.afterInputDelivered()
	1315	+	}
1252	1316		case <-p.ctx.Done():
1253	1317		sendOK.Store(false)
1254	1318		}

@@ -1297,6 +1361,15 @@ func (p *stdinPump) run() {
1297	1361		consumer = input.claim()
1298	1362		defer input.revoke(consumer)
1299	1363		}
	1364	+	// A marker prefix is ordinary terminal input until it becomes a complete
	1365	+	// DECRQM reply. If cancellation wins while it is withheld, preserve just
	1366	+	// that undecided suffix; the rest of its source read has already been
	1367	+	// delivered and must not be replayed by the next attempt.
	1368	+	defer func() {
	1369	+	if p.ctx.Err() != nil {
	1370	+	input.preserveResidual(consumer, markers.takePending())
	1371	+	}
	1372	+	}()
1300	1373
1301	1374		var markerTimer ports.Timer
1302	1375		var markerTimerC <-chan time.Time

@@ -1371,7 +1444,14 @@ func (p *stdinPump) run() {
1371	1444		}
1372	1445		sendEvent(paletteGenerationEvent{id: readGeneration, kind: kindEvent, rgb: rgb})
1373	1446		}, func(slot int, rgb renderer.RGB) {
1374		-	sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventPalette, slot: uint8(slot), rgb: rgb})
	1447	+	// Scanner implementations are independently testable; retain the
	1448	+	// protocol boundary here too so a malformed implementation cannot
	1449	+	// wrap a negative or oversized int into an ANSI palette slot.
	1450	+	slot8, ok := paletteSlot(slot)
	1451	+	if !ok {
	1452	+	return
	1453	+	}
	1454	+	sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventPalette, slot: slot8, rgb: rgb})
1375	1455		}, func(light bool) {
1376	1456		sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventScheme, light: light})
1377	1457		}, func(data []byte) {

@@ -1396,10 +1476,17 @@ func (p *stdinPump) run() {
1396	1476		p.cancel()
1397	1477		return
1398	1478		}
1399		-	if !sendOK.Load() \|\| p.ctx.Err() != nil {
	1479	+	if !sendOK.Load() {
1400	1480		return
1401	1481		}
	1482	+	// Commit every callback from this raw read before observing a
	1483	+	// concurrent cancellation. Any remaining marker prefix is retained
	1484	+	// by the deferred handoff above, so reconnects never duplicate the
	1485	+	// delivered bytes or lose a standalone Escape.
1402	1486		input.ack(consumer)
	1487	+	if p.ctx.Err() != nil {
	1488	+	return
	1489	+	}
1403	1490		}
1404	1491		}
1405	1492		}

 	}
 }
+// paletteSlot validates the scanner's signed slot before narrowing it for the
+// wire accumulator. Keep this boundary even though the in-tree Scanner also
+// validates slots: callbacks are an interface boundary.
+func paletteSlot(slot int) (uint8, bool) {
+	if slot < 0 || slot > 15 {
+		return 0, false
+	}
+	return uint8(slot), true
+}
 // DetectTrueColor reports whether TERM/COLORTERM advertise direct color support.
 func DetectTrueColor(termEnv, colorTerm string) bool {
 	switch strings.ToLower(strings.TrimSpace(colorTerm)) {
 	in   io.Reader
 	done chan struct{}
-	mu         sync.Mutex
+	mu sync.Mutex
-	pending    *terminalReadResult
+	// residual holds scanner bytes that were read but not delivered when an
-	consumer   uint64
+	// attempt is cancelled. It is consumed before the next terminal Read so a
-	delivering uint64
+	// replacement scanner neither loses a partial marker nor replays bytes
-	nextID     uint64
+	// whose callbacks already ran.
-	closed     bool
+	residual           []byte
-	active     bool
+	pending            *terminalReadResult
-	ready      chan struct{}
+	consumer           uint64
-	space      chan struct{}
+	delivering         uint64
-	state      chan struct{}
+	deliveringResidual bool
-	exited     chan struct{}
+	nextID             uint64
-	startMu    sync.Once
+	closed             bool
-	stopMu     sync.Once
+	active             bool
+	ready              chan struct{}
+	space              chan struct{}
+	state              chan struct{}
+	exited             chan struct{}
+	startMu            sync.Once
+	stopMu             sync.Once
 }
 func newTerminalInputPump(in io.Reader) *terminalInputPump {
 	// pending and make it available to its replacement.
 	if p.delivering == consumer {
 		p.delivering = 0
+		p.deliveringResidual = false
 	}
 	p.mu.Unlock()
 	p.signalReady()
 // scanner delivery leaves the exact bytes available to the next attempt.
 func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
 	p.mu.Lock()
-	if ctx.Err() != nil || p.consumer != consumer || p.pending == nil || p.delivering != 0 {
+	if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
-		pending := p.pending != nil && p.delivering == 0
+		ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
 		p.mu.Unlock()
-		if pending {
+		if ready {
 			p.signalReady()
 		}
 		return terminalReadResult{}, false
 	}
+	if len(p.residual) != 0 {
+		result := terminalReadResult{data: append([]byte(nil), p.residual...)}
+		p.delivering = consumer
+		p.deliveringResidual = true
+		p.mu.Unlock()
+		return result, true
+	}
+	if p.pending == nil {
+		p.mu.Unlock()
+		return terminalReadResult{}, false
+	}
 	result := *p.pending
 	p.delivering = consumer
+	p.deliveringResidual = false
 	p.mu.Unlock()
 	return result, true
 }
+// preserveResidual records the undecided marker prefix after callbacks for the
+// rest of its read have completed. It is intentionally bounded by the marker
+// scanner's fixed response prefixes and is replayed before later terminal
+// reads on a replacement attempt.
+func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
+	if len(data) == 0 {
+		return
+	}
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	if p.consumer != consumer {
+		return
+	}
+	p.residual = append(p.residual[:0], data...)
+}
 // ack commits scanner delivery of a leased read and lets the lifecycle reader
 // accept the next result. Only the consumer that holds the lease can ack it.
 func (p *terminalInputPump) ack(consumer uint64) {
 		p.mu.Unlock()
 		return
 	}
-	p.pending = nil
+	wasResidual := p.deliveringResidual
+	if wasResidual {
+		p.residual = nil
+	} else {
+		p.pending = nil
+	}
 	p.delivering = 0
+	p.deliveringResidual = false
+	more := len(p.residual) != 0 || p.pending != nil
 	p.mu.Unlock()
+	if wasResidual {
+		if more {
+			p.signalReady()
+		}
+		return
+	}
 	p.space <- struct{}{}
 }
 	p.mu.Lock()
 	p.active = false
 	discarded := p.pending != nil
+	p.residual = nil
 	p.pending = nil
 	p.delivering = 0
+	p.deliveringResidual = false
 	p.mu.Unlock()
 	if discarded {
 		select {
 // terminal directly, publishes a Theme, or writes terminal output;
 // attachAttempt owns those operations.
 type stdinPump struct {
-	ctx               context.Context
+	ctx                 context.Context
-	cancel            context.CancelFunc
+	cancel              context.CancelFunc
-	in                io.Reader // compatibility for isolated scanner tests
+	in                  io.Reader // compatibility for isolated scanner tests
-	input             *terminalInputPump
+	input               *terminalInputPump
-	consumer          uint64
+	consumer            uint64
-	out               chan<- ports.Frame
+	out                 chan<- ports.Frame
-	clock             ports.Clock
+	clock               ports.Clock
-	clipboard         ports.ClipboardReader
+	clipboard           ports.ClipboardReader
-	logger            *slog.Logger
+	logger              *slog.Logger
-	paletteEvents     chan<- paletteGenerationEvent
+	paletteEvents       chan<- paletteGenerationEvent
-	activeGeneration  *atomic.Uint64
+	activeGeneration    *atomic.Uint64
-	afterPaletteEvent func(paletteGenerationEvent) // test synchronization hook
+	afterPaletteEvent   func(paletteGenerationEvent) // test synchronization hook
-	afterInputTake    func()                       // test synchronization hook
+	afterInputTake      func()                       // test synchronization hook
+	afterInputDelivered func()                       // test synchronization hook
 }
 func (p *stdinPump) run() {
 	send := func(frame ports.Frame) {
 		select {
 		case p.out <- frame:
+			if p.afterInputDelivered != nil {
+				p.afterInputDelivered()
+			}
 		case <-p.ctx.Done():
 			sendOK.Store(false)
 		}
 		consumer = input.claim()
 		defer input.revoke(consumer)
 	}
+	// A marker prefix is ordinary terminal input until it becomes a complete
+	// DECRQM reply. If cancellation wins while it is withheld, preserve just
+	// that undecided suffix; the rest of its source read has already been
+	// delivered and must not be replayed by the next attempt.
+	defer func() {
+		if p.ctx.Err() != nil {
+			input.preserveResidual(consumer, markers.takePending())
+		}
+	}()
 	var markerTimer ports.Timer
 	var markerTimerC <-chan time.Time
 					}
 					sendEvent(paletteGenerationEvent{id: readGeneration, kind: kindEvent, rgb: rgb})
 				}, func(slot int, rgb renderer.RGB) {
-					sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventPalette, slot: uint8(slot), rgb: rgb})
+					// Scanner implementations are independently testable; retain the
+					// protocol boundary here too so a malformed implementation cannot
+					// wrap a negative or oversized int into an ANSI palette slot.
+					slot8, ok := paletteSlot(slot)
+					if !ok {
+						return
+					}
+					sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventPalette, slot: slot8, rgb: rgb})
 				}, func(light bool) {
 					sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventScheme, light: light})
 				}, func(data []byte) {
 				p.cancel()
 				return
 			}
-			if !sendOK.Load() || p.ctx.Err() != nil {
+			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
+			// delivered bytes or lose a standalone Escape.
 			input.ack(consumer)
+			if p.ctx.Err() != nil {
+				return
+			}
 		}
 	}
 }

M internal/usecase/client/palette_attach_test.go +44 -6


@@ -30,19 +30,34 @@ func (paletteAttachTimer) Reset(time.Duration) bool { return false }
30	30		func (paletteAttachTimer) Stop() bool { return true }
31	31
32	32		type paletteAttachReader struct {
	33	+	mu sync.Mutex
33	34		data []byte
	35	+	done chan struct{}
34	36		once sync.Once
35	37		}
36	38
	39	+	func newPaletteAttachReader(data []byte) *paletteAttachReader {
	40	+	return &paletteAttachReader{data: append([]byte(nil), data...), done: make(chan struct{})}
	41	+	}
	42	+
	43	+	// Read retains unread bytes after a short read and blocks only until close.
	44	+	// This keeps the test terminal reader faithful to io.Reader and makes a
	45	+	// failed test teardown deterministic instead of leaving a goroutine parked.
37	46		func (r *paletteAttachReader) Read(p []byte) (int, error) {
38		-	read := false
39		-	r.once.Do(func() { read = true })
40		-	if read {
41		-	return copy(p, r.data), nil
	47	+	r.mu.Lock()
	48	+	if len(r.data) != 0 {
	49	+	n := copy(p, r.data)
	50	+	r.data = r.data[n:]
	51	+	r.mu.Unlock()
	52	+	return n, nil
42	53		}
43		-	select {}
	54	+	r.mu.Unlock()
	55	+	<-r.done
	56	+	return 0, io.EOF
44	57		}
45	58
	59	+	func (r *paletteAttachReader) close() { r.once.Do(func() { close(r.done) }) }
	60	+
46	61		type paletteAttachTerminal struct {
47	62		in io.Reader
48	63		out bytes.Buffer

@@ -100,8 +115,27 @@ func (t *paletteAttachTransport) Recv() (ports.Frame, error) {
100	115		}
101	116		func (*paletteAttachTransport) Close() error { return nil }
102	117
	118	+	func TestPaletteAttachReaderPreservesShortReadRemainderAndCloses(t *testing.T) {
	119	+	reader := newPaletteAttachReader([]byte("abcdef"))
	120	+	first := make([]byte, 2)
	121	+	n, err := reader.Read(first)
	122	+	require.NoError(t, err)
	123	+	require.Equal(t, "ab", string(first[:n]))
	124	+
	125	+	rest := make([]byte, 8)
	126	+	n, err = reader.Read(rest)
	127	+	require.NoError(t, err)
	128	+	require.Equal(t, "cdef", string(rest[:n]))
	129	+
	130	+	reader.close()
	131	+	n, err = reader.Read(rest)
	132	+	require.Zero(t, n)
	133	+	require.ErrorIs(t, err, io.EOF)
	134	+	}
	135	+
103	136		func TestAttachPublishesOnlyClearedAndDefinitiveInitialPalette(t *testing.T) {
104		-	input := &paletteAttachReader{data: []byte("\x1b]10;#010203\a\x1b]11;#040506\a\x1b]4;2;#102030\a\x1b[?2031;1$y")}
	137	+	input := newPaletteAttachReader([]byte("\x1b]10;#010203\a\x1b]11;#040506\a\x1b]4;2;#102030\a\x1b[?2031;1$y"))
	138	+	t.Cleanup(input.close)
105	139		term := &paletteAttachTerminal{in: input, resize: make(chan domain.Size)}
106	140		transport := &paletteAttachTransport{finalSet: make(chan struct{})}
107	141		ms := &milestones{}

@@ -119,6 +153,10 @@ func TestAttachPublishesOnlyClearedAndDefinitiveInitialPalette(t *testing.T) {
119	153
120	154		transport.mu.Lock()
121	155		defer transport.mu.Unlock()
	156	+	require.Len(t, transport.sent, 3)
	157	+	require.Equal(t, []ports.MsgType{ports.MsgHello, ports.MsgTheme, ports.MsgTheme}, []ports.MsgType{
	158	+	transport.sent[0].Type, transport.sent[1].Type, transport.sent[2].Type,
	159	+	}, "the protocol requires Hello before the two palette publications")
122	160		require.Len(t, transport.themes, 2)
123	161		cleared, definitive := transport.themes[0], transport.themes[1]
124	162		require.Zero(t, cleared.PaletteKnown)


@@ -30,19 +30,34 @@ func (paletteAttachTimer) Reset(time.Duration) bool { return false }
30		func (paletteAttachTimer) Stop() bool { return true }	30		func (paletteAttachTimer) Stop() bool { return true }
31			31
32		type paletteAttachReader struct {	32		type paletteAttachReader struct {
			33	+	mu sync.Mutex
33		data []byte	34		data []byte
			35	+	done chan struct{}
34		once sync.Once	36		once sync.Once
35		}	37		}
36			38
			39	+	func newPaletteAttachReader(data []byte) *paletteAttachReader {
			40	+	return &paletteAttachReader{data: append([]byte(nil), data...), done: make(chan struct{})}
			41	+	}
			42	+
			43	+	// Read retains unread bytes after a short read and blocks only until close.
			44	+	// This keeps the test terminal reader faithful to io.Reader and makes a
			45	+	// failed test teardown deterministic instead of leaving a goroutine parked.
37		func (r *paletteAttachReader) Read(p []byte) (int, error) {	46		func (r *paletteAttachReader) Read(p []byte) (int, error) {
38	-	read := false	47	+	r.mu.Lock()
39	-	r.once.Do(func() { read = true })	48	+	if len(r.data) != 0 {
40	-	if read {	49	+	n := copy(p, r.data)
41	-	return copy(p, r.data), nil	50	+	r.data = r.data[n:]
			51	+	r.mu.Unlock()
			52	+	return n, nil
42		}	53		}
43	-	select {}	54	+	r.mu.Unlock()
			55	+	<-r.done
			56	+	return 0, io.EOF
44		}	57		}
45			58
			59	+	func (r *paletteAttachReader) close() { r.once.Do(func() { close(r.done) }) }
			60	+
46		type paletteAttachTerminal struct {	61		type paletteAttachTerminal struct {
47		in io.Reader	62		in io.Reader
48		out bytes.Buffer	63		out bytes.Buffer

@@ -100,8 +115,27 @@ func (t *paletteAttachTransport) Recv() (ports.Frame, error) {
100		}	115		}
101		func (*paletteAttachTransport) Close() error { return nil }	116		func (*paletteAttachTransport) Close() error { return nil }
102			117
			118	+	func TestPaletteAttachReaderPreservesShortReadRemainderAndCloses(t *testing.T) {
			119	+	reader := newPaletteAttachReader([]byte("abcdef"))
			120	+	first := make([]byte, 2)
			121	+	n, err := reader.Read(first)
			122	+	require.NoError(t, err)
			123	+	require.Equal(t, "ab", string(first[:n]))
			124	+
			125	+	rest := make([]byte, 8)
			126	+	n, err = reader.Read(rest)
			127	+	require.NoError(t, err)
			128	+	require.Equal(t, "cdef", string(rest[:n]))
			129	+
			130	+	reader.close()
			131	+	n, err = reader.Read(rest)
			132	+	require.Zero(t, n)
			133	+	require.ErrorIs(t, err, io.EOF)
			134	+	}
			135	+
103		func TestAttachPublishesOnlyClearedAndDefinitiveInitialPalette(t *testing.T) {	136		func TestAttachPublishesOnlyClearedAndDefinitiveInitialPalette(t *testing.T) {
104	-	input := &paletteAttachReader{data: []byte("\x1b]10;#010203\a\x1b]11;#040506\a\x1b]4;2;#102030\a\x1b[?2031;1$y")}	137	+	input := newPaletteAttachReader([]byte("\x1b]10;#010203\a\x1b]11;#040506\a\x1b]4;2;#102030\a\x1b[?2031;1$y"))
			138	+	t.Cleanup(input.close)
105		term := &paletteAttachTerminal{in: input, resize: make(chan domain.Size)}	139		term := &paletteAttachTerminal{in: input, resize: make(chan domain.Size)}
106		transport := &paletteAttachTransport{finalSet: make(chan struct{})}	140		transport := &paletteAttachTransport{finalSet: make(chan struct{})}
107		ms := &milestones{}	141		ms := &milestones{}

@@ -119,6 +153,10 @@ func TestAttachPublishesOnlyClearedAndDefinitiveInitialPalette(t *testing.T) {
119			153
120		transport.mu.Lock()	154		transport.mu.Lock()
121		defer transport.mu.Unlock()	155		defer transport.mu.Unlock()
			156	+	require.Len(t, transport.sent, 3)
			157	+	require.Equal(t, []ports.MsgType{ports.MsgHello, ports.MsgTheme, ports.MsgTheme}, []ports.MsgType{
			158	+	transport.sent[0].Type, transport.sent[1].Type, transport.sent[2].Type,
			159	+	}, "the protocol requires Hello before the two palette publications")
122		require.Len(t, transport.themes, 2)	160		require.Len(t, transport.themes, 2)
123		cleared, definitive := transport.themes[0], transport.themes[1]	161		cleared, definitive := transport.themes[0], transport.themes[1]
124		require.Zero(t, cleared.PaletteKnown)	162		require.Zero(t, cleared.PaletteKnown)

M internal/usecase/client/palette_generation.go +9 -0


@@ -308,4 +308,13 @@ func (s *paletteMarkerScanner) flush(onBytes func([]byte)) {
308	308		s.pending = nil
309	309		}
310	310
	311	+	// takePending transfers only an undecided DECRQM prefix to a replacement
	312	+	// scanner. Callers use it during cancellation after all other callbacks from
	313	+	// the source read have been committed, preventing both loss and replay.
	314	+	func (s *paletteMarkerScanner) takePending() []byte {
	315	+	pending := append([]byte(nil), s.pending...)
	316	+	s.pending = nil
	317	+	return pending
	318	+	}
	319	+
311	320		func (s *paletteMarkerScanner) hasPendingPrefix() bool { return len(s.pending) != 0 }


@@ -308,4 +308,13 @@ func (s *paletteMarkerScanner) flush(onBytes func([]byte)) {
308	s.pending = nil	308		s.pending = nil
309	}	309		}
310		310
		311	+	// takePending transfers only an undecided DECRQM prefix to a replacement
		312	+	// scanner. Callers use it during cancellation after all other callbacks from
		313	+	// the source read have been committed, preventing both loss and replay.
		314	+	func (s *paletteMarkerScanner) takePending() []byte {
		315	+	pending := append([]byte(nil), s.pending...)
		316	+	s.pending = nil
		317	+	return pending
		318	+	}
		319	+
311	func (s *paletteMarkerScanner) hasPendingPrefix() bool { return len(s.pending) != 0 }	320		func (s *paletteMarkerScanner) hasPendingPrefix() bool { return len(s.pending) != 0 }

M internal/usecase/client/reconnect_toast_test.go +62 -28


@@ -110,19 +110,6 @@ func reconnectToastDetach(reason uint8) ports.Frame {
110	110		return ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: reason})}
111	111		}
112	112
113		-	func mockReconnectTransport(t testing.T, recvs ...reconnectToastRecv) portsmocks.MockTransport {
114		-	t.Helper()
115		-	tr := portsmocks.NewMockTransport(t)
116		-	tr.EXPECT().Send(mock.MatchedBy(func(f ports.Frame) bool { return f.Type == ports.MsgTheme })).Return(nil).Maybe()
117		-	tr.EXPECT().Send(mock.Anything).Return(nil).Maybe()
118		-	for _, recv := range recvs {
119		-	tr.EXPECT().Recv().Return(recv.frame, recv.err).Once()
120		-	}
121		-	tr.EXPECT().Recv().Return(ports.Frame{}, io.EOF).Maybe()
122		-	tr.EXPECT().Close().Return(nil).Maybe()
123		-	return tr
124		-	}
125		-
126	113		type reconnectToastRecv struct {
127	114		frame ports.Frame
128	115		err error

@@ -143,11 +130,16 @@ func (d *reconnectToastSequenceDialer) Dial(context.Context) (ports.Transport, e
143	130		}
144	131
145	132		type reconnectToastRecordingTransport struct {
	133	+	mu sync.Mutex
146	134		recvs []reconnectToastRecv
147	135		sends []ports.Frame
148	136		closed bool
149	137		}
150	138
	139	+	func newReconnectToastRecordingTransport(recvs ...reconnectToastRecv) *reconnectToastRecordingTransport {
	140	+	return &reconnectToastRecordingTransport{recvs: append([]reconnectToastRecv(nil), recvs...)}
	141	+	}
	142	+
151	143		type reconnectToastLinkTransport struct {
152	144		recvCh chan reconnectToastRecv
153	145		sends []ports.Frame

@@ -180,11 +172,15 @@ func (t *reconnectToastLinkTransport) LinkEvents() <-chan ports.LinkEvent { retu
180	172		func (t *reconnectToastLinkTransport) LinkState() ports.LinkState { return t.state }
181	173
182	174		func (t *reconnectToastRecordingTransport) Send(f ports.Frame) error {
	175	+	t.mu.Lock()
	176	+	defer t.mu.Unlock()
183	177		t.sends = append(t.sends, f)
184	178		return nil
185	179		}
186	180
187	181		func (t *reconnectToastRecordingTransport) Recv() (ports.Frame, error) {
	182	+	t.mu.Lock()
	183	+	defer t.mu.Unlock()
188	184		if len(t.recvs) == 0 {
189	185		return ports.Frame{}, io.EOF
190	186		}

@@ -194,15 +190,47 @@ func (t *reconnectToastRecordingTransport) Recv() (ports.Frame, error) {
194	190		}
195	191
196	192		func (t *reconnectToastRecordingTransport) Close() error {
	193	+	t.mu.Lock()
	194	+	defer t.mu.Unlock()
197	195		t.closed = true
198	196		return nil
199	197		}
200	198
	199	+	func (t *reconnectToastRecordingTransport) sentFrames() []ports.Frame {
	200	+	t.mu.Lock()
	201	+	defer t.mu.Unlock()
	202	+	return append([]ports.Frame(nil), t.sends...)
	203	+	}
	204	+
	205	+	func (t *reconnectToastRecordingTransport) wasClosed() bool {
	206	+	t.mu.Lock()
	207	+	defer t.mu.Unlock()
	208	+	return t.closed
	209	+	}
	210	+
	211	+	// assertReconnectToastAttemptPublishesOnlyClearedTheme verifies the precise
	212	+	// handshake publication for these no-OSC-response reconnect paths. A final
	213	+	// Theme would be wrong here: no terminal palette response was supplied.
	214	+	func assertReconnectToastAttemptPublishesOnlyClearedTheme(t testing.T, tr reconnectToastRecordingTransport) {
	215	+	t.Helper()
	216	+	frames := tr.sentFrames()
	217	+	require.Len(t, frames, 2)
	218	+	require.Equal(t, ports.MsgHello, frames[0].Type)
	219	+	require.Equal(t, ports.MsgTheme, frames[1].Type)
	220	+	theme, err := ports.UnmarshalTheme(frames[1].Payload)
	221	+	require.NoError(t, err)
	222	+	require.False(t, theme.HasForeground)
	223	+	require.False(t, theme.HasBackground)
	224	+	require.Zero(t, theme.PaletteKnown)
	225	+	require.True(t, tr.wasClosed())
	226	+	}
	227	+
201	228		func reconnectToastHelloFromSend(t testing.T, tr reconnectToastRecordingTransport) ports.Hello {
202	229		t.Helper()
203		-	require.NotEmpty(t, tr.sends)
204		-	require.Equal(t, ports.MsgHello, tr.sends[0].Type)
205		-	hello, err := ports.UnmarshalHello(tr.sends[0].Payload)
	230	+	frames := tr.sentFrames()
	231	+	require.NotEmpty(t, frames)
	232	+	require.Equal(t, ports.MsgHello, frames[0].Type)
	233	+	hello, err := ports.UnmarshalHello(frames[0].Payload)
206	234		require.NoError(t, err)
207	235		return hello
208	236		}

@@ -494,8 +522,8 @@ func TestRemoteReconnectToastLifecycleWithWrappedTransportError(t *testing.T) {
494	522
495	523		err := NewRunner(Dependencies{Dialer: dialer, Terminal: term.term, Clock: newReconnectHandshakeClock(t), Logger: slog.New(slog.DiscardHandler)}).Run(context.Background(), AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true})
496	524		require.NoError(t, err)
497		-	require.True(t, tr1.closed)
498		-	require.True(t, tr2.closed)
	525	+	require.True(t, tr1.wasClosed())
	526	+	require.True(t, tr2.wasClosed())
499	527		require.Equal(t, 2, dialer.calls)
500	528
501	529		firstHello := reconnectToastHelloFromSend(t, tr1)

@@ -555,18 +583,19 @@ func TestRemoteEphemeralReconnectUsesAssignedSessionName(t *testing.T) {
555	583		func TestRemoteReconnectToastLifecycle(t *testing.T) {
556	584		tests := []struct {
557	585		name string
558		-	configure func(t testing.T, dialer portsmocks.MockDialer)
	586	+	configure func(t testing.T, dialer portsmocks.MockDialer) []*reconnectToastRecordingTransport
559	587		sleep func(cancel context.CancelFunc) bool
560	588		wantErr func(t *testing.T, err error)
561	589		wantBlank bool
562	590		}{
563	591		{
564	592		name: "clears on successful reconnect",
565		-	configure: func(t testing.T, dialer portsmocks.MockDialer) {
566		-	tr1 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
567		-	tr2 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)})
	593	+	configure: func(t testing.T, dialer portsmocks.MockDialer) []*reconnectToastRecordingTransport {
	594	+	tr1 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
	595	+	tr2 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)})
568	596		dialer.EXPECT().Dial(mock.Anything).Return(tr1, nil).Once()
569	597		dialer.EXPECT().Dial(mock.Anything).Return(tr2, nil).Once()
	598	+	return []*reconnectToastRecordingTransport{tr1, tr2}
570	599		},
571	600		sleep: func(context.CancelFunc) bool { return true },
572	601		wantErr: func(t *testing.T, err error) { require.NoError(t, err) },

@@ -574,9 +603,10 @@ func TestRemoteReconnectToastLifecycle(t *testing.T) {
574	603		},
575	604		{
576	605		name: "clears on cancellation",
577		-	configure: func(t testing.T, dialer portsmocks.MockDialer) {
578		-	tr := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
	606	+	configure: func(t testing.T, dialer portsmocks.MockDialer) []*reconnectToastRecordingTransport {
	607	+	tr := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
579	608		dialer.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
	609	+	return []*reconnectToastRecordingTransport{tr}
580	610		},
581	611		sleep: func(cancel context.CancelFunc) bool {
582	612		cancel()

@@ -587,11 +617,12 @@ func TestRemoteReconnectToastLifecycle(t *testing.T) {
587	617		},
588	618		{
589	619		name: "clears on final exit",
590		-	configure: func(t testing.T, dialer portsmocks.MockDialer) {
591		-	tr1 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
592		-	tr2 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonSessionKilled)})
	620	+	configure: func(t testing.T, dialer portsmocks.MockDialer) []*reconnectToastRecordingTransport {
	621	+	tr1 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
	622	+	tr2 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonSessionKilled)})
593	623		dialer.EXPECT().Dial(mock.Anything).Return(tr1, nil).Once()
594	624		dialer.EXPECT().Dial(mock.Anything).Return(tr2, nil).Once()
	625	+	return []*reconnectToastRecordingTransport{tr1, tr2}
595	626		},
596	627		sleep: func(context.CancelFunc) bool { return true },
597	628		wantErr: func(t *testing.T, err error) {

@@ -619,10 +650,13 @@ func TestRemoteReconnectToastLifecycle(t *testing.T) {
619	650		term := newReconnectToastTerminalHarness(t)
620	651		defer term.closeInput()
621	652		dialer := portsmocks.NewMockDialer(t)
622		-	tt.configure(t, dialer)
	653	+	transports := tt.configure(t, dialer)
623	654
624	655		err := NewRunner(Dependencies{Dialer: dialer, Terminal: term.term, Clock: newReconnectHandshakeClock(t), Logger: slog.New(slog.DiscardHandler)}).Run(ctx, AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true})
625	656		tt.wantErr(t, err)
	657	+	for _, transport := range transports {
	658	+	assertReconnectToastAttemptPublishesOnlyClearedTheme(t, transport)
	659	+	}
626	660		out := term.out.String()
627	661		require.Contains(t, out, reconnectToastMessage)
628	662		if tt.wantBlank {

 	return ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: reason})}
 }
-func mockReconnectTransport(t *testing.T, recvs ...reconnectToastRecv) *portsmocks.MockTransport {
-	t.Helper()
-	tr := portsmocks.NewMockTransport(t)
-	tr.EXPECT().Send(mock.MatchedBy(func(f ports.Frame) bool { return f.Type == ports.MsgTheme })).Return(nil).Maybe()
-	tr.EXPECT().Send(mock.Anything).Return(nil).Maybe()
-	for _, recv := range recvs {
-		tr.EXPECT().Recv().Return(recv.frame, recv.err).Once()
-	}
-	tr.EXPECT().Recv().Return(ports.Frame{}, io.EOF).Maybe()
-	tr.EXPECT().Close().Return(nil).Maybe()
-	return tr
-}
 type reconnectToastRecv struct {
 	frame ports.Frame
 	err   error
 }
 type reconnectToastRecordingTransport struct {
+	mu     sync.Mutex
 	recvs  []reconnectToastRecv
 	sends  []ports.Frame
 	closed bool
 }
+func newReconnectToastRecordingTransport(recvs ...reconnectToastRecv) *reconnectToastRecordingTransport {
+	return &reconnectToastRecordingTransport{recvs: append([]reconnectToastRecv(nil), recvs...)}
+}
 type reconnectToastLinkTransport struct {
 	recvCh chan reconnectToastRecv
 	sends  []ports.Frame
 func (t *reconnectToastLinkTransport) LinkState() ports.LinkState { return t.state }
 func (t *reconnectToastRecordingTransport) Send(f ports.Frame) error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
 	t.sends = append(t.sends, f)
 	return nil
 }
 func (t *reconnectToastRecordingTransport) Recv() (ports.Frame, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
 	if len(t.recvs) == 0 {
 		return ports.Frame{}, io.EOF
 	}
 }
 func (t *reconnectToastRecordingTransport) Close() error {
+	t.mu.Lock()
+	defer t.mu.Unlock()
 	t.closed = true
 	return nil
 }
+func (t *reconnectToastRecordingTransport) sentFrames() []ports.Frame {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return append([]ports.Frame(nil), t.sends...)
+}
+func (t *reconnectToastRecordingTransport) wasClosed() bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	return t.closed
+}
+// assertReconnectToastAttemptPublishesOnlyClearedTheme verifies the precise
+// handshake publication for these no-OSC-response reconnect paths. A final
+// Theme would be wrong here: no terminal palette response was supplied.
+func assertReconnectToastAttemptPublishesOnlyClearedTheme(t *testing.T, tr *reconnectToastRecordingTransport) {
+	t.Helper()
+	frames := tr.sentFrames()
+	require.Len(t, frames, 2)
+	require.Equal(t, ports.MsgHello, frames[0].Type)
+	require.Equal(t, ports.MsgTheme, frames[1].Type)
+	theme, err := ports.UnmarshalTheme(frames[1].Payload)
+	require.NoError(t, err)
+	require.False(t, theme.HasForeground)
+	require.False(t, theme.HasBackground)
+	require.Zero(t, theme.PaletteKnown)
+	require.True(t, tr.wasClosed())
+}
 func reconnectToastHelloFromSend(t *testing.T, tr *reconnectToastRecordingTransport) ports.Hello {
 	t.Helper()
-	require.NotEmpty(t, tr.sends)
+	frames := tr.sentFrames()
-	require.Equal(t, ports.MsgHello, tr.sends[0].Type)
+	require.NotEmpty(t, frames)
-	hello, err := ports.UnmarshalHello(tr.sends[0].Payload)
+	require.Equal(t, ports.MsgHello, frames[0].Type)
+	hello, err := ports.UnmarshalHello(frames[0].Payload)
 	require.NoError(t, err)
 	return hello
 }
 	err := NewRunner(Dependencies{Dialer: dialer, Terminal: term.term, Clock: newReconnectHandshakeClock(t), Logger: slog.New(slog.DiscardHandler)}).Run(context.Background(), AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true})
 	require.NoError(t, err)
-	require.True(t, tr1.closed)
+	require.True(t, tr1.wasClosed())
-	require.True(t, tr2.closed)
+	require.True(t, tr2.wasClosed())
 	require.Equal(t, 2, dialer.calls)
 	firstHello := reconnectToastHelloFromSend(t, tr1)
 func TestRemoteReconnectToastLifecycle(t *testing.T) {
 	tests := []struct {
 		name      string
-		configure func(t *testing.T, dialer *portsmocks.MockDialer)
+		configure func(t *testing.T, dialer *portsmocks.MockDialer) []*reconnectToastRecordingTransport
 		sleep     func(cancel context.CancelFunc) bool
 		wantErr   func(t *testing.T, err error)
 		wantBlank bool
 	}{
 		{
 			name: "clears on successful reconnect",
-			configure: func(t *testing.T, dialer *portsmocks.MockDialer) {
+			configure: func(t *testing.T, dialer *portsmocks.MockDialer) []*reconnectToastRecordingTransport {
-				tr1 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
+				tr1 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
-				tr2 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)})
+				tr2 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)})
 				dialer.EXPECT().Dial(mock.Anything).Return(tr1, nil).Once()
 				dialer.EXPECT().Dial(mock.Anything).Return(tr2, nil).Once()
+				return []*reconnectToastRecordingTransport{tr1, tr2}
 			},
 			sleep:     func(context.CancelFunc) bool { return true },
 			wantErr:   func(t *testing.T, err error) { require.NoError(t, err) },
 		},
 		{
 			name: "clears on cancellation",
-			configure: func(t *testing.T, dialer *portsmocks.MockDialer) {
+			configure: func(t *testing.T, dialer *portsmocks.MockDialer) []*reconnectToastRecordingTransport {
-				tr := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
+				tr := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
 				dialer.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
+				return []*reconnectToastRecordingTransport{tr}
 			},
 			sleep: func(cancel context.CancelFunc) bool {
 				cancel()
 		},
 		{
 			name: "clears on final exit",
-			configure: func(t *testing.T, dialer *portsmocks.MockDialer) {
+			configure: func(t *testing.T, dialer *portsmocks.MockDialer) []*reconnectToastRecordingTransport {
-				tr1 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
+				tr1 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(11)}, reconnectToastRecv{err: io.EOF})
-				tr2 := mockReconnectTransport(t, reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonSessionKilled)})
+				tr2 := newReconnectToastRecordingTransport(reconnectToastRecv{frame: reconnectToastWelcome(22)}, reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonSessionKilled)})
 				dialer.EXPECT().Dial(mock.Anything).Return(tr1, nil).Once()
 				dialer.EXPECT().Dial(mock.Anything).Return(tr2, nil).Once()
+				return []*reconnectToastRecordingTransport{tr1, tr2}
 			},
 			sleep: func(context.CancelFunc) bool { return true },
 			wantErr: func(t *testing.T, err error) {
 			term := newReconnectToastTerminalHarness(t)
 			defer term.closeInput()
 			dialer := portsmocks.NewMockDialer(t)
-			tt.configure(t, dialer)
+			transports := tt.configure(t, dialer)
 			err := NewRunner(Dependencies{Dialer: dialer, Terminal: term.term, Clock: newReconnectHandshakeClock(t), Logger: slog.New(slog.DiscardHandler)}).Run(ctx, AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true})
 			tt.wantErr(t, err)
+			for _, transport := range transports {
+				assertReconnectToastAttemptPublishesOnlyClearedTheme(t, transport)
+			}
 			out := term.out.String()
 			require.Contains(t, out, reconnectToastMessage)
 			if tt.wantBlank {

M internal/usecase/client/terminal_input_pump_test.go +99 -0


@@ -71,6 +71,26 @@ func (r *lateStopReader) Read(p []byte) (int, error) {
71	71		return copy(p, result.data), result.err
72	72		}
73	73
	74	+	func TestPaletteSlotRejectsOutOfRangeIntBeforeNarrowing(t *testing.T) {
	75	+	for _, tc := range []struct {
	76	+	slot int
	77	+	want uint8
	78	+	ok bool
	79	+	}{
	80	+	{slot: -1},
	81	+	{slot: 0, want: 0, ok: true},
	82	+	{slot: 15, want: 15, ok: true},
	83	+	{slot: 16},
	84	+	{slot: 256},
	85	+	} {
	86	+	got, ok := paletteSlot(tc.slot)
	87	+	require.Equal(t, tc.ok, ok, "slot %d", tc.slot)
	88	+	if ok {
	89	+	require.Equal(t, tc.want, got, "slot %d", tc.slot)
	90	+	}
	91	+	}
	92	+	}
	93	+
74	94		func TestTerminalInputPumpStopDropsLateReadWithoutStartingAnother(t *testing.T) {
75	95		reader := newLateStopReader()
76	96		input := newTerminalInputPump(reader)

@@ -201,6 +221,85 @@ func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
201	221		input.revoke(replacement)
202	222		}
203	223
	224	+	func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t *testing.T) {
	225	+	input := newTerminalInputPump(nil)
	226	+	input.enqueue(terminalReadResult{data: []byte("\x1b")})
	227	+	var activeGeneration atomic.Uint64
	228	+
	229	+	oldCtx, cancelOld := context.WithCancel(context.Background())
	230	+	oldClock := newAttachPaletteClock()
	231	+	oldDone := make(chan struct{})
	232	+	go func() {
	233	+	(&stdinPump{
	234	+	ctx: oldCtx, cancel: cancelOld, input: input, out: make(chan ports.Frame, 1),
	235	+	clock: oldClock, logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	236	+	activeGeneration: &activeGeneration,
	237	+	}).run()
	238	+	close(oldDone)
	239	+	}()
	240	+	// The ambiguity timer proves the ESC was scanned and its raw read was
	241	+	// committed; only the marker scanner's undecided suffix remains.
	242	+	<-oldClock.timers
	243	+	cancelOld()
	244	+	<-oldDone
	245	+
	246	+	newCtx, cancelNew := context.WithCancel(context.Background())
	247	+	defer cancelNew()
	248	+	newClock := newAttachPaletteClock()
	249	+	out := make(chan ports.Frame, 1)
	250	+	newDone := make(chan struct{})
	251	+	go func() {
	252	+	(&stdinPump{
	253	+	ctx: newCtx, cancel: cancelNew, input: input, out: out,
	254	+	clock: newClock, logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	255	+	activeGeneration: &activeGeneration,
	256	+	}).run()
	257	+	close(newDone)
	258	+	}()
	259	+	(<-newClock.timers).fire() // marker ambiguity deadline
	260	+	(<-newClock.timers).fire() // paste coalescer's lone-ESC deadline
	261	+	frame := <-out
	262	+	got, err := ports.UnmarshalInput(frame.Payload)
	263	+	require.NoError(t, err)
	264	+	require.Equal(t, []byte("\x1b"), got.Data)
	265	+
	266	+	cancelNew()
	267	+	<-newDone
	268	+	}
	269	+
	270	+	func TestTerminalInputPumpCancellationAfterDeliveredCallbackDoesNotReplay(t *testing.T) {
	271	+	input := newTerminalInputPump(nil)
	272	+	input.enqueue(terminalReadResult{data: []byte("x")})
	273	+	var activeGeneration atomic.Uint64
	274	+	ctx, cancel := context.WithCancel(context.Background())
	275	+	delivered := make(chan struct{})
	276	+	done := make(chan struct{})
	277	+	out := make(chan ports.Frame, 1)
	278	+	go func() {
	279	+	(&stdinPump{
	280	+	ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
	281	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	282	+	activeGeneration: &activeGeneration,
	283	+	afterInputDelivered: func() {
	284	+	cancel()
	285	+	close(delivered)
	286	+	},
	287	+	}).run()
	288	+	close(done)
	289	+	}()
	290	+	frame := <-out
	291	+	got, err := ports.UnmarshalInput(frame.Payload)
	292	+	require.NoError(t, err)
	293	+	require.Equal(t, []byte("x"), got.Data)
	294	+	<-delivered
	295	+	<-done
	296	+
	297	+	consumer := input.claim()
	298	+	_, ok := input.take(context.Background(), consumer)
	299	+	require.False(t, ok, "a callback already delivered to the sender must not replay after reconnect")
	300	+	input.revoke(consumer)
	301	+	}
	302	+
204	303		func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
205	304		reader := newLifecycleReader()
206	305		input := newTerminalInputPump(reader)


@@ -71,6 +71,26 @@ func (r *lateStopReader) Read(p []byte) (int, error) {
71	return copy(p, result.data), result.err	71		return copy(p, result.data), result.err
72	}	72		}
73		73
		74	+	func TestPaletteSlotRejectsOutOfRangeIntBeforeNarrowing(t *testing.T) {
		75	+	for _, tc := range []struct {
		76	+	slot int
		77	+	want uint8
		78	+	ok bool
		79	+	}{
		80	+	{slot: -1},
		81	+	{slot: 0, want: 0, ok: true},
		82	+	{slot: 15, want: 15, ok: true},
		83	+	{slot: 16},
		84	+	{slot: 256},
		85	+	} {
		86	+	got, ok := paletteSlot(tc.slot)
		87	+	require.Equal(t, tc.ok, ok, "slot %d", tc.slot)
		88	+	if ok {
		89	+	require.Equal(t, tc.want, got, "slot %d", tc.slot)
		90	+	}
		91	+	}
		92	+	}
		93	+
74	func TestTerminalInputPumpStopDropsLateReadWithoutStartingAnother(t *testing.T) {	94		func TestTerminalInputPumpStopDropsLateReadWithoutStartingAnother(t *testing.T) {
75	reader := newLateStopReader()	95		reader := newLateStopReader()
76	input := newTerminalInputPump(reader)	96		input := newTerminalInputPump(reader)

@@ -201,6 +221,85 @@ func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
201	input.revoke(replacement)	221		input.revoke(replacement)
202	}	222		}
203		223
		224	+	func TestTerminalInputPumpCancellationPreservesStandaloneEscapeForReplacement(t *testing.T) {
		225	+	input := newTerminalInputPump(nil)
		226	+	input.enqueue(terminalReadResult{data: []byte("\x1b")})
		227	+	var activeGeneration atomic.Uint64
		228	+
		229	+	oldCtx, cancelOld := context.WithCancel(context.Background())
		230	+	oldClock := newAttachPaletteClock()
		231	+	oldDone := make(chan struct{})
		232	+	go func() {
		233	+	(&stdinPump{
		234	+	ctx: oldCtx, cancel: cancelOld, input: input, out: make(chan ports.Frame, 1),
		235	+	clock: oldClock, logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
		236	+	activeGeneration: &activeGeneration,
		237	+	}).run()
		238	+	close(oldDone)
		239	+	}()
		240	+	// The ambiguity timer proves the ESC was scanned and its raw read was
		241	+	// committed; only the marker scanner's undecided suffix remains.
		242	+	<-oldClock.timers
		243	+	cancelOld()
		244	+	<-oldDone
		245	+
		246	+	newCtx, cancelNew := context.WithCancel(context.Background())
		247	+	defer cancelNew()
		248	+	newClock := newAttachPaletteClock()
		249	+	out := make(chan ports.Frame, 1)
		250	+	newDone := make(chan struct{})
		251	+	go func() {
		252	+	(&stdinPump{
		253	+	ctx: newCtx, cancel: cancelNew, input: input, out: out,
		254	+	clock: newClock, logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
		255	+	activeGeneration: &activeGeneration,
		256	+	}).run()
		257	+	close(newDone)
		258	+	}()
		259	+	(<-newClock.timers).fire() // marker ambiguity deadline
		260	+	(<-newClock.timers).fire() // paste coalescer's lone-ESC deadline
		261	+	frame := <-out
		262	+	got, err := ports.UnmarshalInput(frame.Payload)
		263	+	require.NoError(t, err)
		264	+	require.Equal(t, []byte("\x1b"), got.Data)
		265	+
		266	+	cancelNew()
		267	+	<-newDone
		268	+	}
		269	+
		270	+	func TestTerminalInputPumpCancellationAfterDeliveredCallbackDoesNotReplay(t *testing.T) {
		271	+	input := newTerminalInputPump(nil)
		272	+	input.enqueue(terminalReadResult{data: []byte("x")})
		273	+	var activeGeneration atomic.Uint64
		274	+	ctx, cancel := context.WithCancel(context.Background())
		275	+	delivered := make(chan struct{})
		276	+	done := make(chan struct{})
		277	+	out := make(chan ports.Frame, 1)
		278	+	go func() {
		279	+	(&stdinPump{
		280	+	ctx: ctx, cancel: cancel, input: input, out: out, clock: newAttachPaletteClock(),
		281	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
		282	+	activeGeneration: &activeGeneration,
		283	+	afterInputDelivered: func() {
		284	+	cancel()
		285	+	close(delivered)
		286	+	},
		287	+	}).run()
		288	+	close(done)
		289	+	}()
		290	+	frame := <-out
		291	+	got, err := ports.UnmarshalInput(frame.Payload)
		292	+	require.NoError(t, err)
		293	+	require.Equal(t, []byte("x"), got.Data)
		294	+	<-delivered
		295	+	<-done
		296	+
		297	+	consumer := input.claim()
		298	+	_, ok := input.take(context.Background(), consumer)
		299	+	require.False(t, ok, "a callback already delivered to the sender must not replay after reconnect")
		300	+	input.revoke(consumer)
		301	+	}
		302	+
204	func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {	303		func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
205	reader := newLifecycleReader()	304		reader := newLifecycleReader()
206	input := newTerminalInputPump(reader)	305		input := newTerminalInputPump(reader)