d0f41bb fix(client): correlate palette boundary response · vev

M internal/usecase/client/client.go +65 -26


@@ -8,7 +8,6 @@
8	8		package client
9	9
10	10		import (
11		-	"bytes"
12	11		"context"
13	12		"crypto/rand"
14	13		"errors"

@@ -602,22 +601,23 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
602	601		ackQueue := newCumulativeAckQueue()
603	602		sendErrCh := make(chan error, 1)
604	603
605		-	// Palette re-queries use DSR (CSI 5 n) as an ordered sentinel. The stdin
606		-	// pump drains terminal replies through CSI 0 n before it permits the main
607		-	// loop to publish the cleared scheme and issue QueryColors. This separates
608		-	// bytes already queued (including Scanner.pending) from replacement replies
609		-	// without allowing a second goroutine to write to the terminal.
	604	+	// Palette re-queries use a private-mode query as an ordered boundary. The
	605	+	// stdin pump drains terminal replies through its mode-specific response
	606	+	// before it permits the main loop to publish the cleared scheme and issue
	607	+	// QueryColors. This separates bytes already queued (including
	608	+	// Scanner.pending) from replacement replies without allowing a second
	609	+	// goroutine to write to the terminal.
610	610		colorQueryCh := make(chan colorQueryRequest)
611	611		paletteReadyCh := make(chan paletteQueryReady)
612	612		requestColors := func() (colorQueryRequest, bool) {
613		-	request := colorQueryRequest{sentinelWritten: make(chan struct{})}
	613	+	request := colorQueryRequest{boundaryWritten: make(chan struct{})}
614	614		select {
615	615		case colorQueryCh <- request:
616	616		case <-loopCtx.Done():
617	617		return colorQueryRequest{}, false
618	618		}
619	619		select {
620		-	case <-request.sentinelWritten:
	620	+	case <-request.boundaryWritten:
621	621		return request, true
622	622		case <-loopCtx.Done():
623	623		return colorQueryRequest{}, false

@@ -669,17 +669,18 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
669	669		return welcomedResult(errLinkOffline)
670	670		}
671	671		case request := <-colorQueryCh:
672		-	if _, err := term.Out().Write([]byte("\x1b[5n")); err != nil {
673		-	return welcomedResult(fmt.Errorf("vev: writing palette sentinel: %w", err))
	672	+	if _, err := term.Out().Write(paletteBoundaryQuery); err != nil {
	673	+	return welcomedResult(fmt.Errorf("vev: writing palette boundary query: %w", err))
674	674		}
675	675		if err := term.Flush(); err != nil {
676		-	return welcomedResult(fmt.Errorf("vev: flushing palette sentinel: %w", err))
	676	+	return welcomedResult(fmt.Errorf("vev: flushing palette boundary query: %w", err))
677	677		}
678		-	close(request.sentinelWritten)
	678	+	close(request.boundaryWritten)
679	679		case ready := <-paletteReadyCh:
680		-	// CSI 0 n proves all earlier replies have been discarded by stdin.
681		-	// Publish the one cleared scheme snapshot before emitting the query;
682		-	// stdin remains paused until QueryColors has completed below.
	680	+	// The private-mode response proves all earlier replies have been
	681	+	// discarded by stdin. Publish the one cleared scheme snapshot before
	682	+	// emitting the query; stdin remains paused until QueryColors has
	683	+	// completed below.
683	684		clearedTheme := themeState.snapshot()
684	685		select {
685	686		case sendCh <- ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(clearedTheme)}:

@@ -848,7 +849,7 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
848	849		// terminal — and matches the standard pattern for stdin pumps; a
849	850		// closable stdin duplicate could lift it later if ever needed.
850	851		type colorQueryRequest struct {
851		-	sentinelWritten chan struct{}
	852	+	boundaryWritten chan struct{}
852	853		}
853	854
854	855		type paletteQueryReady struct {

@@ -868,22 +869,60 @@ const (
868	869		paletteAwaitingQuery
869	870		)
870	871
871		-	var paletteSentinel = []byte("\x1b[0n")
	872	+	var paletteBoundaryQuery = []byte("\x1b[?2031$p")
	873	+
	874	+	// paletteBoundaryResponses accepts valid DECRQM status values for private
	875	+	// mode 2031. The private mode number correlates this reply with the query
	876	+	// that starts a replacement palette generation.
	877	+	var paletteBoundaryResponses = [][]byte{
	878	+	[]byte("\x1b[?2031;0$y"),
	879	+	[]byte("\x1b[?2031;1$y"),
	880	+	[]byte("\x1b[?2031;2$y"),
	881	+	[]byte("\x1b[?2031;3$y"),
	882	+	[]byte("\x1b[?2031;4$y"),
	883	+	}
872	884
873		-	// scan consumes the DSR response while forwarding every other byte in order.
874		-	// It retains only a possible sentinel prefix, so a split CSI 0 n is consumed
875		-	// without dropping a user's partial escape sequence on cancellation.
876		-	func (d *paletteDrain) scan(data []byte, onBytes func([]byte), onSentinel func()) {
	885	+	// scan consumes the private-mode boundary response while forwarding every
	886	+	// other byte in order. It retains only a possible response prefix, so split
	887	+	// replies are consumed without dropping a user's partial escape sequence on
	888	+	// cancellation.
	889	+	func (d *paletteDrain) scan(data []byte, onBytes func([]byte), onBoundary func()) {
877	890		for _, b := range data {
878	891		d.pending = append(d.pending, b)
879		-	for len(d.pending) > 0 && !bytes.HasPrefix(paletteSentinel, d.pending) {
	892	+	for len(d.pending) > 0 {
	893	+	matched := false
	894	+	possible := false
	895	+	for _, response := range paletteBoundaryResponses {
	896	+	if len(d.pending) > len(response) {
	897	+	continue
	898	+	}
	899	+	match := true
	900	+	for i := range d.pending {
	901	+	if d.pending[i] != response[i] {
	902	+	match = false
	903	+	break
	904	+	}
	905	+	}
	906	+	if !match {
	907	+	continue
	908	+	}
	909	+	if len(d.pending) == len(response) {
	910	+	matched = true
	911	+	} else {
	912	+	possible = true
	913	+	}
	914	+	}
	915	+	if matched {
	916	+	d.pending = nil
	917	+	onBoundary()
	918	+	break
	919	+	}
	920	+	if possible {
	921	+	break
	922	+	}
880	923		onBytes(d.pending[:1])
881	924		d.pending = d.pending[1:]
882	925		}
883		-	if bytes.Equal(d.pending, paletteSentinel) {
884		-	d.pending = nil
885		-	onSentinel()
886		-	}
887	926		}
888	927		}
889	928

 package client
 import (
-	"bytes"
 	"context"
 	"crypto/rand"
 	"errors"
 	ackQueue := newCumulativeAckQueue()
 	sendErrCh := make(chan error, 1)
-	// Palette re-queries use DSR (CSI 5 n) as an ordered sentinel. The stdin
+	// Palette re-queries use a private-mode query as an ordered boundary. The
-	// pump drains terminal replies through CSI 0 n before it permits the main
+	// stdin pump drains terminal replies through its mode-specific response
-	// loop to publish the cleared scheme and issue QueryColors. This separates
+	// before it permits the main loop to publish the cleared scheme and issue
-	// bytes already queued (including Scanner.pending) from replacement replies
+	// QueryColors. This separates bytes already queued (including
-	// without allowing a second goroutine to write to the terminal.
+	// Scanner.pending) from replacement replies without allowing a second
+	// goroutine to write to the terminal.
 	colorQueryCh := make(chan colorQueryRequest)
 	paletteReadyCh := make(chan paletteQueryReady)
 	requestColors := func() (colorQueryRequest, bool) {
-		request := colorQueryRequest{sentinelWritten: make(chan struct{})}
+		request := colorQueryRequest{boundaryWritten: make(chan struct{})}
 		select {
 		case colorQueryCh <- request:
 		case <-loopCtx.Done():
 			return colorQueryRequest{}, false
 		}
 		select {
-		case <-request.sentinelWritten:
+		case <-request.boundaryWritten:
 			return request, true
 		case <-loopCtx.Done():
 			return colorQueryRequest{}, false
 				return welcomedResult(errLinkOffline)
 			}
 		case request := <-colorQueryCh:
-			if _, err := term.Out().Write([]byte("\x1b[5n")); err != nil {
+			if _, err := term.Out().Write(paletteBoundaryQuery); err != nil {
-				return welcomedResult(fmt.Errorf("vev: writing palette sentinel: %w", err))
+				return welcomedResult(fmt.Errorf("vev: writing palette boundary query: %w", err))
 			}
 			if err := term.Flush(); err != nil {
-				return welcomedResult(fmt.Errorf("vev: flushing palette sentinel: %w", err))
+				return welcomedResult(fmt.Errorf("vev: flushing palette boundary query: %w", err))
 			}
-			close(request.sentinelWritten)
+			close(request.boundaryWritten)
 		case ready := <-paletteReadyCh:
-			// CSI 0 n proves all earlier replies have been discarded by stdin.
+			// The private-mode response proves all earlier replies have been
-			// Publish the one cleared scheme snapshot before emitting the query;
+			// discarded by stdin. Publish the one cleared scheme snapshot before
-			// stdin remains paused until QueryColors has completed below.
+			// emitting the query; stdin remains paused until QueryColors has
+			// completed below.
 			clearedTheme := themeState.snapshot()
 			select {
 			case sendCh <- ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(clearedTheme)}:
 // terminal — and matches the standard pattern for stdin pumps; a
 // closable stdin duplicate could lift it later if ever needed.
 type colorQueryRequest struct {
-	sentinelWritten chan struct{}
+	boundaryWritten chan struct{}
 }
 type paletteQueryReady struct {
 	paletteAwaitingQuery
 )
-var paletteSentinel = []byte("\x1b[0n")
+var paletteBoundaryQuery = []byte("\x1b[?2031$p")
+// paletteBoundaryResponses accepts valid DECRQM status values for private
+// mode 2031. The private mode number correlates this reply with the query
+// that starts a replacement palette generation.
+var paletteBoundaryResponses = [][]byte{
+	[]byte("\x1b[?2031;0$y"),
+	[]byte("\x1b[?2031;1$y"),
+	[]byte("\x1b[?2031;2$y"),
+	[]byte("\x1b[?2031;3$y"),
+	[]byte("\x1b[?2031;4$y"),
+}
-// scan consumes the DSR response while forwarding every other byte in order.
+// scan consumes the private-mode boundary response while forwarding every
-// It retains only a possible sentinel prefix, so a split CSI 0 n is consumed
+// other byte in order. It retains only a possible response prefix, so split
-// without dropping a user's partial escape sequence on cancellation.
+// replies are consumed without dropping a user's partial escape sequence on
-func (d *paletteDrain) scan(data []byte, onBytes func([]byte), onSentinel func()) {
+// cancellation.
+func (d *paletteDrain) scan(data []byte, onBytes func([]byte), onBoundary func()) {
 	for _, b := range data {
 		d.pending = append(d.pending, b)
-		for len(d.pending) > 0 && !bytes.HasPrefix(paletteSentinel, d.pending) {
+		for len(d.pending) > 0 {
+			matched := false
+			possible := false
+			for _, response := range paletteBoundaryResponses {
+				if len(d.pending) > len(response) {
+					continue
+				}
+				match := true
+				for i := range d.pending {
+					if d.pending[i] != response[i] {
+						match = false
+						break
+					}
+				}
+				if !match {
+					continue
+				}
+				if len(d.pending) == len(response) {
+					matched = true
+				} else {
+					possible = true
+				}
+			}
+			if matched {
+				d.pending = nil
+				onBoundary()
+				break
+			}
+			if possible {
+				break
+			}
 			onBytes(d.pending[:1])
 			d.pending = d.pending[1:]
 		}
-		if bytes.Equal(d.pending, paletteSentinel) {
-			d.pending = nil
-			onSentinel()
-		}
 	}
 }

M internal/usecase/client/client_test.go +9 -9


@@ -1048,11 +1048,11 @@ func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) {
1048	1048		require.Equal(t, int32(1), term.restoreCount.Load())
1049	1049		}
1050	1050
1051		-	func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T) {
	1051	+	func TestAttachSchemeRequeryCorrelatesSplitBoundaryAndDiscardsStalePalette(t *testing.T) {
1052	1052		input := newChunkedBlockingReader(
1053		-	[]byte("a\x1b]4;1;#112233\a\x1b[?997;2n\x1b]4;2;#44"),
1054		-	[]byte("5566\ab\x1b["),
1055		-	[]byte("0nc\x1b]4;3;#778899\ad"),
	1053	+	[]byte("a\x1b]4;1;#112233\a\x1b[?997;2n"),
	1054	+	[]byte("\x1b[0n\x1b]4;2;#445566\ab\x1b[?2031;"),
	1055	+	[]byte("2$yc\x1b]4;3;#778899\ad"),
1056	1056		)
1057	1057		defer input.unblock()
1058	1058

@@ -1107,7 +1107,7 @@ func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T)
1107	1107		tr.EXPECT().Close().Return(nil).Once()
1108	1108
1109	1109		require.NoError(t, runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral}))
1110		-	require.Equal(t, "\x1b[5n", out.String(), "the DSR sentinel must be the only main-loop terminal write")
	1110	+	require.Equal(t, "\x1b[?2031$p", out.String(), "the correlated boundary query must be the only main-loop terminal write")
1111	1111		require.Equal(t, int32(1), restoreCount.Load())
1112	1112
1113	1113		cleared := <-themes

@@ -1115,10 +1115,10 @@ func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T)
1115	1115		require.True(t, cleared.SchemeKnown)
1116	1116		require.True(t, cleared.Light)
1117	1117		require.Zero(t, cleared.PaletteKnown, "only the cleared scheme snapshot may cross the boundary")
1118		-	require.Equal(t, uint16(1<<3), fresh.PaletteKnown, "only colors after the sentinel and replacement query are accepted")
	1118	+	require.Equal(t, uint16(1<<3), fresh.PaletteKnown, "only colors after the boundary response and replacement query are accepted")
1119	1119		require.Zero(t, fresh.PaletteKnown&(1<<1\|1<<2), "pre-boundary palette slots must not leak into the new generation")
1120	1120		inputMu.Lock()
1121		-	require.Equal(t, []byte("abcd"), inputBytes, "ordinary input around split stale and sentinel responses must survive")
	1121	+	require.Equal(t, []byte("a\x1b[0nbcd"), inputBytes, "the old status response and ordinary input around the boundary must survive")
1122	1122		inputMu.Unlock()
1123	1123		}
1124	1124

@@ -1129,7 +1129,7 @@ func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1129	1129		input := newChunkedBlockingReader(
1130	1130		[]byte("\x1b]11;#010203\a\x1b]4;1;#112233\a\x1b]4;14;#778899\a"),
1131	1131		[]byte("\x1b[?997;2n"),
1132		-	[]byte("\x1b[0n"),
	1132	+	[]byte("\x1b[?2031;1$y"),
1133	1133		)
1134	1134		defer input.unblock()
1135	1135

@@ -1190,7 +1190,7 @@ func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1190	1190		require.Len(t, themes, 2, "the old palette snapshot and one cleared scheme snapshot are sent")
1191	1191		require.Equal(t, uint16(1<<1\|1<<14), themes[0].PaletteKnown)
1192	1192		require.True(t, themes[0].HasBackground)
1193		-	require.Zero(t, themes[1].PaletteKnown, "the DSR-delimited re-query must invalidate stale palette entries")
	1193	+	require.Zero(t, themes[1].PaletteKnown, "the correlated boundary re-query must invalidate stale palette entries")
1194	1194		require.Equal(t, themes[0].Background, themes[1].Background)
1195	1195		require.True(t, themes[1].SchemeKnown)
1196	1196		require.True(t, themes[1].Light)


@@ -1048,11 +1048,11 @@ func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) {
1048		require.Equal(t, int32(1), term.restoreCount.Load())	1048		require.Equal(t, int32(1), term.restoreCount.Load())
1049		}	1049		}
1050			1050
1051	-	func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T) {	1051	+	func TestAttachSchemeRequeryCorrelatesSplitBoundaryAndDiscardsStalePalette(t *testing.T) {
1052		input := newChunkedBlockingReader(	1052		input := newChunkedBlockingReader(
1053	-	[]byte("a\x1b]4;1;#112233\a\x1b[?997;2n\x1b]4;2;#44"),	1053	+	[]byte("a\x1b]4;1;#112233\a\x1b[?997;2n"),
1054	-	[]byte("5566\ab\x1b["),	1054	+	[]byte("\x1b[0n\x1b]4;2;#445566\ab\x1b[?2031;"),
1055	-	[]byte("0nc\x1b]4;3;#778899\ad"),	1055	+	[]byte("2$yc\x1b]4;3;#778899\ad"),
1056		)	1056		)
1057		defer input.unblock()	1057		defer input.unblock()
1058			1058

@@ -1107,7 +1107,7 @@ func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T)
1107		tr.EXPECT().Close().Return(nil).Once()	1107		tr.EXPECT().Close().Return(nil).Once()
1108			1108
1109		require.NoError(t, runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral}))	1109		require.NoError(t, runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral}))
1110	-	require.Equal(t, "\x1b[5n", out.String(), "the DSR sentinel must be the only main-loop terminal write")	1110	+	require.Equal(t, "\x1b[?2031$p", out.String(), "the correlated boundary query must be the only main-loop terminal write")
1111		require.Equal(t, int32(1), restoreCount.Load())	1111		require.Equal(t, int32(1), restoreCount.Load())
1112			1112
1113		cleared := <-themes	1113		cleared := <-themes

@@ -1115,10 +1115,10 @@ func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T)
1115		require.True(t, cleared.SchemeKnown)	1115		require.True(t, cleared.SchemeKnown)
1116		require.True(t, cleared.Light)	1116		require.True(t, cleared.Light)
1117		require.Zero(t, cleared.PaletteKnown, "only the cleared scheme snapshot may cross the boundary")	1117		require.Zero(t, cleared.PaletteKnown, "only the cleared scheme snapshot may cross the boundary")
1118	-	require.Equal(t, uint16(1<<3), fresh.PaletteKnown, "only colors after the sentinel and replacement query are accepted")	1118	+	require.Equal(t, uint16(1<<3), fresh.PaletteKnown, "only colors after the boundary response and replacement query are accepted")
1119		require.Zero(t, fresh.PaletteKnown&(1<<1\|1<<2), "pre-boundary palette slots must not leak into the new generation")	1119		require.Zero(t, fresh.PaletteKnown&(1<<1\|1<<2), "pre-boundary palette slots must not leak into the new generation")
1120		inputMu.Lock()	1120		inputMu.Lock()
1121	-	require.Equal(t, []byte("abcd"), inputBytes, "ordinary input around split stale and sentinel responses must survive")	1121	+	require.Equal(t, []byte("a\x1b[0nbcd"), inputBytes, "the old status response and ordinary input around the boundary must survive")
1122		inputMu.Unlock()	1122		inputMu.Unlock()
1123		}	1123		}
1124			1124

@@ -1129,7 +1129,7 @@ func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1129		input := newChunkedBlockingReader(	1129		input := newChunkedBlockingReader(
1130		[]byte("\x1b]11;#010203\a\x1b]4;1;#112233\a\x1b]4;14;#778899\a"),	1130		[]byte("\x1b]11;#010203\a\x1b]4;1;#112233\a\x1b]4;14;#778899\a"),
1131		[]byte("\x1b[?997;2n"),	1131		[]byte("\x1b[?997;2n"),
1132	-	[]byte("\x1b[0n"),	1132	+	[]byte("\x1b[?2031;1$y"),
1133		)	1133		)
1134		defer input.unblock()	1134		defer input.unblock()
1135			1135

@@ -1190,7 +1190,7 @@ func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1190		require.Len(t, themes, 2, "the old palette snapshot and one cleared scheme snapshot are sent")	1190		require.Len(t, themes, 2, "the old palette snapshot and one cleared scheme snapshot are sent")
1191		require.Equal(t, uint16(1<<1\|1<<14), themes[0].PaletteKnown)	1191		require.Equal(t, uint16(1<<1\|1<<14), themes[0].PaletteKnown)
1192		require.True(t, themes[0].HasBackground)	1192		require.True(t, themes[0].HasBackground)
1193	-	require.Zero(t, themes[1].PaletteKnown, "the DSR-delimited re-query must invalidate stale palette entries")	1193	+	require.Zero(t, themes[1].PaletteKnown, "the correlated boundary re-query must invalidate stale palette entries")
1194		require.Equal(t, themes[0].Background, themes[1].Background)	1194		require.Equal(t, themes[0].Background, themes[1].Background)
1195		require.True(t, themes[1].SchemeKnown)	1195		require.True(t, themes[1].SchemeKnown)
1196		require.True(t, themes[1].Light)	1196		require.True(t, themes[1].Light)

M internal/usecase/client/theme_state_test.go +35 -2


@@ -187,7 +187,7 @@ func (r *chunksReader) Read(p []byte) (int, error) {
187	187		return copy(p, chunk), nil
188	188		}
189	189
190		-	func TestStdinPumpDefersRepeatedSchemeChunksUntilSentinel(t *testing.T) {
	190	+	func TestStdinPumpDefersRepeatedSchemeChunksUntilBoundary(t *testing.T) {
191	191		ctx, cancel := context.WithCancel(context.Background())
192	192		defer cancel()
193	193		requests := 0

@@ -208,7 +208,40 @@ func TestStdinPumpDefersRepeatedSchemeChunksUntilSentinel(t *testing.T) {
208	208		}
209	209
210	210		pump.run()
211		-	require.Equal(t, 1, requests, "a later scheme response is drained until the first sentinel arrives")
	211	+	require.Equal(t, 1, requests, "a later scheme response is drained until the boundary response arrives")
	212	+	}
	213	+
	214	+	func TestPaletteDrainConsumesOnlyPrivateModeBoundaryResponses(t *testing.T) {
	215	+	for _, status := range []byte{'0', '1', '2', '3', '4'} {
	216	+	t.Run(string(status), func(t *testing.T) {
	217	+	response := []byte("\x1b[?2031;" + string(status) + "$y")
	218	+	var got []byte
	219	+	boundaries := 0
	220	+	drain := paletteDrain{}
	221	+	drain.scan(append([]byte("x"), response[:len(response)-2]...), func(data []byte) {
	222	+	got = append(got, data...)
	223	+	}, func() { boundaries++ })
	224	+	drain.scan(append(response[len(response)-2:], 'y'), func(data []byte) {
	225	+	got = append(got, data...)
	226	+	}, func() { boundaries++ })
	227	+
	228	+	require.Equal(t, []byte("xy"), got)
	229	+	require.Equal(t, 1, boundaries)
	230	+	})
	231	+	}
	232	+	}
	233	+
	234	+	func TestPaletteDrainForwardsNonmatchingAndPartialBytes(t *testing.T) {
	235	+	var got []byte
	236	+	boundaries := 0
	237	+	drain := paletteDrain{}
	238	+	emit := func(data []byte) { got = append(got, data...) }
	239	+	drain.scan([]byte("\x1b[0n\x1b[?2031;5$y"), emit, func() { boundaries++ })
	240	+	drain.scan([]byte("\x1b[?2031;"), emit, func() { boundaries++ })
	241	+	drain.flush(emit)
	242	+
	243	+	require.Equal(t, []byte("\x1b[0n\x1b[?2031;5$y\x1b[?2031;"), got)
	244	+	require.Zero(t, boundaries)
212	245		}
213	246
214	247		func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {


@@ -187,7 +187,7 @@ func (r *chunksReader) Read(p []byte) (int, error) {
187		return copy(p, chunk), nil	187		return copy(p, chunk), nil
188		}	188		}
189			189
190	-	func TestStdinPumpDefersRepeatedSchemeChunksUntilSentinel(t *testing.T) {	190	+	func TestStdinPumpDefersRepeatedSchemeChunksUntilBoundary(t *testing.T) {
191		ctx, cancel := context.WithCancel(context.Background())	191		ctx, cancel := context.WithCancel(context.Background())
192		defer cancel()	192		defer cancel()
193		requests := 0	193		requests := 0

@@ -208,7 +208,40 @@ func TestStdinPumpDefersRepeatedSchemeChunksUntilSentinel(t *testing.T) {
208		}	208		}
209			209
210		pump.run()	210		pump.run()
211	-	require.Equal(t, 1, requests, "a later scheme response is drained until the first sentinel arrives")	211	+	require.Equal(t, 1, requests, "a later scheme response is drained until the boundary response arrives")
			212	+	}
			213	+
			214	+	func TestPaletteDrainConsumesOnlyPrivateModeBoundaryResponses(t *testing.T) {
			215	+	for _, status := range []byte{'0', '1', '2', '3', '4'} {
			216	+	t.Run(string(status), func(t *testing.T) {
			217	+	response := []byte("\x1b[?2031;" + string(status) + "$y")
			218	+	var got []byte
			219	+	boundaries := 0
			220	+	drain := paletteDrain{}
			221	+	drain.scan(append([]byte("x"), response[:len(response)-2]...), func(data []byte) {
			222	+	got = append(got, data...)
			223	+	}, func() { boundaries++ })
			224	+	drain.scan(append(response[len(response)-2:], 'y'), func(data []byte) {
			225	+	got = append(got, data...)
			226	+	}, func() { boundaries++ })
			227	+
			228	+	require.Equal(t, []byte("xy"), got)
			229	+	require.Equal(t, 1, boundaries)
			230	+	})
			231	+	}
			232	+	}
			233	+
			234	+	func TestPaletteDrainForwardsNonmatchingAndPartialBytes(t *testing.T) {
			235	+	var got []byte
			236	+	boundaries := 0
			237	+	drain := paletteDrain{}
			238	+	emit := func(data []byte) { got = append(got, data...) }
			239	+	drain.scan([]byte("\x1b[0n\x1b[?2031;5$y"), emit, func() { boundaries++ })
			240	+	drain.scan([]byte("\x1b[?2031;"), emit, func() { boundaries++ })
			241	+	drain.flush(emit)
			242	+
			243	+	require.Equal(t, []byte("\x1b[0n\x1b[?2031;5$y\x1b[?2031;"), got)
			244	+	require.Zero(t, boundaries)
212		}	245		}
213			246
214		func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {	247		func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {