fix(client): synchronize palette query boundary

commit: 8f22d3a36bbc6b613ccaa302cf9d1d202676cf54
author: brice · 3mo ago · 2026-07-17 08:35 UTC
parent: 2274b26
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +336 -9

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


@@ -605,12 +605,23 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
605	605		// the main loop, which is the only goroutine allowed to touch term.Out()/
606	606		// term.Flush(); QueryColors writes through the same batched writer, so
607	607		// calling it directly from the stdin pump would race the main loop's
608		-	// output writes.
609		-	colorQueryCh := make(chan struct{}, 1)
610		-	requestColors := func() {
	608	+	// output writes. The acknowledgement creates a generation boundary: after
	609	+	// a scheme response, stdin must not read another chunk (which may contain
	610	+	// old OSC 4 replies) until this loop has cleared the palette and issued the
	611	+	// replacement query.
	612	+	colorQueryCh := make(chan colorQueryRequest)
	613	+	requestColors := func() bool {
	614	+	request := colorQueryRequest{acknowledged: make(chan struct{})}
611	615		select {
612		-	case colorQueryCh <- struct{}{}:
613		-	default:
	616	+	case colorQueryCh <- request:
	617	+	case <-loopCtx.Done():
	618	+	return false
	619	+	}
	620	+	select {
	621	+	case <-request.acknowledged:
	622	+	return true
	623	+	case <-loopCtx.Done():
	624	+	return false
614	625		}
615	626		}
616	627

@@ -658,7 +669,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
658	669		// token. Run closes this transport on the way out.
659	670		return welcomedResult(errLinkOffline)
660	671		}
661		-	case <-colorQueryCh:
	672	+	case request := <-colorQueryCh:
662	673		// A fresh query supersedes every prior OSC 4 response. Queue the
663	674		// cleared snapshot before writing the query so the daemon never
664	675		// applies stale palette entries while it awaits the new responses.

@@ -671,6 +682,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
671	682		if err := term.QueryColors(); err != nil {
672	683		log.Warn("querying terminal colors", "err", err)
673	684		}
	685	+	close(request.acknowledged)
674	686		case r := <-recvCh:
675	687		if r.err != nil {
676	688		if errors.Is(r.err, io.EOF) {

@@ -828,6 +840,10 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
828	840		// exits. That is harmless here — Run has already returned and restored the
829	841		// terminal — and matches the standard pattern for stdin pumps; a
830	842		// closable stdin duplicate could lift it later if ever needed.
	843	+	type colorQueryRequest struct {
	844	+	acknowledged chan struct{}
	845	+	}
	846	+
831	847		type stdinPump struct {
832	848		ctx context.Context
833	849		cancel context.CancelFunc

@@ -837,7 +853,7 @@ type stdinPump struct {
837	853		themeState *terminalThemeState
838	854		clipboard ports.ClipboardReader
839	855		logger *slog.Logger
840		-	requestColors func()
	856	+	requestColors func() bool
841	857		}
842	858
843	859		func (p *stdinPump) run() {

@@ -932,8 +948,8 @@ func (p *stdinPump) run() {
932	948		// Defer the re-query until this read's complete snapshot is queued;
933	949		// otherwise the main loop could invalidate palette state midway
934	950		// through a chunk containing multiple terminal responses.
935		-	if queryColors && requestColors != nil {
936		-	requestColors()
	951	+	if queryColors && requestColors != nil && !requestColors() {
	952	+	return
937	953		}
938	954		if !sendOK.Load() {
939	955		return

 	// the main loop, which is the only goroutine allowed to touch term.Out()/
 	// term.Flush(); QueryColors writes through the same batched writer, so
 	// calling it directly from the stdin pump would race the main loop's
-	// output writes.
+	// output writes. The acknowledgement creates a generation boundary: after
-	colorQueryCh := make(chan struct{}, 1)
+	// a scheme response, stdin must not read another chunk (which may contain
-	requestColors := func() {
+	// old OSC 4 replies) until this loop has cleared the palette and issued the
+	// replacement query.
+	colorQueryCh := make(chan colorQueryRequest)
+	requestColors := func() bool {
+		request := colorQueryRequest{acknowledged: make(chan struct{})}
 		select {
-		case colorQueryCh <- struct{}{}:
+		case colorQueryCh <- request:
-		default:
+		case <-loopCtx.Done():
+			return false
+		}
+		select {
+		case <-request.acknowledged:
+			return true
+		case <-loopCtx.Done():
+			return false
 		}
 	}
 				// token. Run closes this transport on the way out.
 				return welcomedResult(errLinkOffline)
 			}
-		case <-colorQueryCh:
+		case request := <-colorQueryCh:
 			// A fresh query supersedes every prior OSC 4 response. Queue the
 			// cleared snapshot before writing the query so the daemon never
 			// applies stale palette entries while it awaits the new responses.
 			if err := term.QueryColors(); err != nil {
 				log.Warn("querying terminal colors", "err", err)
 			}
+			close(request.acknowledged)
 		case r := <-recvCh:
 			if r.err != nil {
 				if errors.Is(r.err, io.EOF) {
 // exits. That is harmless here — Run has already returned and restored the
 // terminal — and matches the standard pattern for stdin pumps; a
 // closable stdin duplicate could lift it later if ever needed.
+type colorQueryRequest struct {
+	acknowledged chan struct{}
+}
 type stdinPump struct {
 	ctx           context.Context
 	cancel        context.CancelFunc
 	themeState    *terminalThemeState
 	clipboard     ports.ClipboardReader
 	logger        *slog.Logger
-	requestColors func()
+	requestColors func() bool
 }
 func (p *stdinPump) run() {
 			// Defer the re-query until this read's complete snapshot is queued;
 			// otherwise the main loop could invalidate palette state midway
 			// through a chunk containing multiple terminal responses.
-			if queryColors && requestColors != nil {
+			if queryColors && requestColors != nil && !requestColors() {
-				requestColors()
+				return
 			}
 			if !sendOK.Load() {
 				return

M internal/usecase/client/client_test.go +135 -0


@@ -137,6 +137,40 @@ func (r *chunkedBlockingReader) Read(p []byte) (int, error) {
137	137
138	138		func (r *chunkedBlockingReader) unblock() { close(r.done) }
139	139
	140	+	// queryBoundaryReader reports an attempt to read the second chunk before
	141	+	// returning it, allowing a test to prove that a scheme re-query is a real
	142	+	// stdin generation boundary rather than a lossy notification.
	143	+	type queryBoundaryReader struct {
	144	+	first, second []byte
	145	+	secondRead chan struct{}
	146	+	allowSecond, done <-chan struct{}
	147	+	reads int
	148	+	}
	149	+
	150	+	func (r *queryBoundaryReader) Read(p []byte) (int, error) {
	151	+	switch r.reads {
	152	+	case 0:
	153	+	r.reads++
	154	+	return copy(p, r.first), nil
	155	+	case 1:
	156	+	select {
	157	+	case r.secondRead <- struct{}{}:
	158	+	case <-r.done:
	159	+	return 0, io.EOF
	160	+	}
	161	+	select {
	162	+	case <-r.allowSecond:
	163	+	r.reads++
	164	+	return copy(p, r.second), nil
	165	+	case <-r.done:
	166	+	return 0, io.EOF
	167	+	}
	168	+	default:
	169	+	<-r.done
	170	+	return 0, io.EOF
	171	+	}
	172	+	}
	173	+
140	174		func newHappyTerminal(t testing.T, out bytes.Buffer, restoreCount atomic.Int32, resizeCh chan domain.Size) (portsmocks.MockTerminal, *blockingReader) {
141	175		tm := portsmocks.NewMockTerminal(t)
142	176		tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()

@@ -1048,6 +1082,107 @@ func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) {
1048	1082		require.Equal(t, int32(1), term.restoreCount.Load())
1049	1083		}
1050	1084
	1085	+	func TestAttachSchemeRequeryCreatesPaletteGenerationBoundary(t *testing.T) {
	1086	+	ctx, cancel := context.WithCancel(context.Background())
	1087	+	defer cancel()
	1088	+	allowSecond := make(chan struct{})
	1089	+	inputDone := make(chan struct{})
	1090	+	input := &queryBoundaryReader{
	1091	+	first: []byte("\x1b]4;1;#112233\a\x1b[?997;2n"),
	1092	+	second: []byte("\x1b]4;2;#445566\a"),
	1093	+	secondRead: make(chan struct{}, 1),
	1094	+	allowSecond: allowSecond,
	1095	+	done: inputDone,
	1096	+	}
	1097	+	defer close(inputDone)
	1098	+
	1099	+	var out bytes.Buffer
	1100	+	var restoreCount atomic.Int32
	1101	+	resizeCh := make(chan domain.Size)
	1102	+	queryStarted := make(chan struct{})
	1103	+	allowQuery := make(chan struct{})
	1104	+	tm := portsmocks.NewMockTerminal(t)
	1105	+	tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
	1106	+	tm.EXPECT().EnterRaw().Return(func() error { restoreCount.Add(1); return nil }, nil).Once()
	1107	+	tm.EXPECT().In().Return(input).Maybe()
	1108	+	tm.EXPECT().Out().Return(&out).Maybe()
	1109	+	tm.EXPECT().Flush().Return(nil).Maybe()
	1110	+	tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
	1111	+	tm.EXPECT().QueryColors().RunAndReturn(func() error {
	1112	+	close(queryStarted)
	1113	+	<-allowQuery
	1114	+	return nil
	1115	+	}).Once()
	1116	+
	1117	+	themes := make(chan ports.Theme, 3)
	1118	+	recvRelease := make(chan struct{})
	1119	+	var welcomed atomic.Bool
	1120	+	tr := portsmocks.NewMockTransport(t)
	1121	+	tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
	1122	+	tr.EXPECT().Send(isType(ports.MsgTheme)).RunAndReturn(func(f ports.Frame) error {
	1123	+	theme, err := ports.UnmarshalTheme(f.Payload)
	1124	+	require.NoError(t, err)
	1125	+	themes <- theme
	1126	+	return nil
	1127	+	}).Times(3)
	1128	+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
	1129	+	if welcomed.CompareAndSwap(false, true) {
	1130	+	return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"})), nil
	1131	+	}
	1132	+	<-recvRelease
	1133	+	return ports.Frame{}, io.EOF
	1134	+	}).Maybe()
	1135	+	tr.EXPECT().Close().Return(nil).Once()
	1136	+
	1137	+	runDone := make(chan error, 1)
	1138	+	go func() {
	1139	+	runDone <- runTestClient(ctx, attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral})
	1140	+	}()
	1141	+
	1142	+	select {
	1143	+	case <-queryStarted:
	1144	+	case <-time.After(time.Second):
	1145	+	t.Fatal("main loop did not start the palette re-query")
	1146	+	}
	1147	+	select {
	1148	+	case <-input.secondRead:
	1149	+	t.Fatal("stdin read the next chunk before the cleared theme and query completed")
	1150	+	default:
	1151	+	}
	1152	+
	1153	+	close(allowQuery)
	1154	+	select {
	1155	+	case <-input.secondRead:
	1156	+	case <-time.After(time.Second):
	1157	+	t.Fatal("stdin did not resume after the palette re-query completed")
	1158	+	}
	1159	+	close(allowSecond)
	1160	+
	1161	+	var got [3]ports.Theme
	1162	+	for i := range got {
	1163	+	select {
	1164	+	case got[i] = <-themes:
	1165	+	case <-time.After(time.Second):
	1166	+	t.Fatalf("theme %d was not sent", i)
	1167	+	}
	1168	+	}
	1169	+	require.Equal(t, uint16(1<<1), got[0].PaletteKnown)
	1170	+	require.True(t, got[0].SchemeKnown)
	1171	+	require.Zero(t, got[1].PaletteKnown, "the old palette must be cleared before a later read is admitted")
	1172	+	require.True(t, got[1].SchemeKnown)
	1173	+	require.Equal(t, uint16(1<<2), got[2].PaletteKnown, "only the post-query chunk may populate the new palette generation")
	1174	+
	1175	+	cancel()
	1176	+	select {
	1177	+	case err := <-runDone:
	1178	+	require.NoError(t, err)
	1179	+	case <-time.After(time.Second):
	1180	+	t.Fatal("attach did not return after cancellation")
	1181	+	}
	1182	+	close(recvRelease)
	1183	+	require.Equal(t, int32(1), restoreCount.Load())
	1184	+	}
	1185	+
1051	1186		func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1052	1187		var out bytes.Buffer
1053	1188		var restoreCount atomic.Int32


@@ -137,6 +137,40 @@ func (r *chunkedBlockingReader) Read(p []byte) (int, error) {
137		137
138	func (r *chunkedBlockingReader) unblock() { close(r.done) }	138		func (r *chunkedBlockingReader) unblock() { close(r.done) }
139		139
		140	+	// queryBoundaryReader reports an attempt to read the second chunk before
		141	+	// returning it, allowing a test to prove that a scheme re-query is a real
		142	+	// stdin generation boundary rather than a lossy notification.
		143	+	type queryBoundaryReader struct {
		144	+	first, second []byte
		145	+	secondRead chan struct{}
		146	+	allowSecond, done <-chan struct{}
		147	+	reads int
		148	+	}
		149	+
		150	+	func (r *queryBoundaryReader) Read(p []byte) (int, error) {
		151	+	switch r.reads {
		152	+	case 0:
		153	+	r.reads++
		154	+	return copy(p, r.first), nil
		155	+	case 1:
		156	+	select {
		157	+	case r.secondRead <- struct{}{}:
		158	+	case <-r.done:
		159	+	return 0, io.EOF
		160	+	}
		161	+	select {
		162	+	case <-r.allowSecond:
		163	+	r.reads++
		164	+	return copy(p, r.second), nil
		165	+	case <-r.done:
		166	+	return 0, io.EOF
		167	+	}
		168	+	default:
		169	+	<-r.done
		170	+	return 0, io.EOF
		171	+	}
		172	+	}
		173	+
140	func newHappyTerminal(t testing.T, out bytes.Buffer, restoreCount atomic.Int32, resizeCh chan domain.Size) (portsmocks.MockTerminal, *blockingReader) {	174		func newHappyTerminal(t testing.T, out bytes.Buffer, restoreCount atomic.Int32, resizeCh chan domain.Size) (portsmocks.MockTerminal, *blockingReader) {
141	tm := portsmocks.NewMockTerminal(t)	175		tm := portsmocks.NewMockTerminal(t)
142	tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()	176		tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()

@@ -1048,6 +1082,107 @@ func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) {
1048	require.Equal(t, int32(1), term.restoreCount.Load())	1082		require.Equal(t, int32(1), term.restoreCount.Load())
1049	}	1083		}
1050		1084
		1085	+	func TestAttachSchemeRequeryCreatesPaletteGenerationBoundary(t *testing.T) {
		1086	+	ctx, cancel := context.WithCancel(context.Background())
		1087	+	defer cancel()
		1088	+	allowSecond := make(chan struct{})
		1089	+	inputDone := make(chan struct{})
		1090	+	input := &queryBoundaryReader{
		1091	+	first: []byte("\x1b]4;1;#112233\a\x1b[?997;2n"),
		1092	+	second: []byte("\x1b]4;2;#445566\a"),
		1093	+	secondRead: make(chan struct{}, 1),
		1094	+	allowSecond: allowSecond,
		1095	+	done: inputDone,
		1096	+	}
		1097	+	defer close(inputDone)
		1098	+
		1099	+	var out bytes.Buffer
		1100	+	var restoreCount atomic.Int32
		1101	+	resizeCh := make(chan domain.Size)
		1102	+	queryStarted := make(chan struct{})
		1103	+	allowQuery := make(chan struct{})
		1104	+	tm := portsmocks.NewMockTerminal(t)
		1105	+	tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
		1106	+	tm.EXPECT().EnterRaw().Return(func() error { restoreCount.Add(1); return nil }, nil).Once()
		1107	+	tm.EXPECT().In().Return(input).Maybe()
		1108	+	tm.EXPECT().Out().Return(&out).Maybe()
		1109	+	tm.EXPECT().Flush().Return(nil).Maybe()
		1110	+	tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
		1111	+	tm.EXPECT().QueryColors().RunAndReturn(func() error {
		1112	+	close(queryStarted)
		1113	+	<-allowQuery
		1114	+	return nil
		1115	+	}).Once()
		1116	+
		1117	+	themes := make(chan ports.Theme, 3)
		1118	+	recvRelease := make(chan struct{})
		1119	+	var welcomed atomic.Bool
		1120	+	tr := portsmocks.NewMockTransport(t)
		1121	+	tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
		1122	+	tr.EXPECT().Send(isType(ports.MsgTheme)).RunAndReturn(func(f ports.Frame) error {
		1123	+	theme, err := ports.UnmarshalTheme(f.Payload)
		1124	+	require.NoError(t, err)
		1125	+	themes <- theme
		1126	+	return nil
		1127	+	}).Times(3)
		1128	+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
		1129	+	if welcomed.CompareAndSwap(false, true) {
		1130	+	return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"})), nil
		1131	+	}
		1132	+	<-recvRelease
		1133	+	return ports.Frame{}, io.EOF
		1134	+	}).Maybe()
		1135	+	tr.EXPECT().Close().Return(nil).Once()
		1136	+
		1137	+	runDone := make(chan error, 1)
		1138	+	go func() {
		1139	+	runDone <- runTestClient(ctx, attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral})
		1140	+	}()
		1141	+
		1142	+	select {
		1143	+	case <-queryStarted:
		1144	+	case <-time.After(time.Second):
		1145	+	t.Fatal("main loop did not start the palette re-query")
		1146	+	}
		1147	+	select {
		1148	+	case <-input.secondRead:
		1149	+	t.Fatal("stdin read the next chunk before the cleared theme and query completed")
		1150	+	default:
		1151	+	}
		1152	+
		1153	+	close(allowQuery)
		1154	+	select {
		1155	+	case <-input.secondRead:
		1156	+	case <-time.After(time.Second):
		1157	+	t.Fatal("stdin did not resume after the palette re-query completed")
		1158	+	}
		1159	+	close(allowSecond)
		1160	+
		1161	+	var got [3]ports.Theme
		1162	+	for i := range got {
		1163	+	select {
		1164	+	case got[i] = <-themes:
		1165	+	case <-time.After(time.Second):
		1166	+	t.Fatalf("theme %d was not sent", i)
		1167	+	}
		1168	+	}
		1169	+	require.Equal(t, uint16(1<<1), got[0].PaletteKnown)
		1170	+	require.True(t, got[0].SchemeKnown)
		1171	+	require.Zero(t, got[1].PaletteKnown, "the old palette must be cleared before a later read is admitted")
		1172	+	require.True(t, got[1].SchemeKnown)
		1173	+	require.Equal(t, uint16(1<<2), got[2].PaletteKnown, "only the post-query chunk may populate the new palette generation")
		1174	+
		1175	+	cancel()
		1176	+	select {
		1177	+	case err := <-runDone:
		1178	+	require.NoError(t, err)
		1179	+	case <-time.After(time.Second):
		1180	+	t.Fatal("attach did not return after cancellation")
		1181	+	}
		1182	+	close(recvRelease)
		1183	+	require.Equal(t, int32(1), restoreCount.Load())
		1184	+	}
		1185	+
1051	func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {	1186		func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1052	var out bytes.Buffer	1187		var out bytes.Buffer
1053	var restoreCount atomic.Int32	1188		var restoreCount atomic.Int32

M internal/usecase/client/theme_state_test.go +176 -0


@@ -6,6 +6,7 @@ import (
6	6		"io"
7	7		"log/slog"
8	8		"testing"
	9	+	"time"
9	10
10	11		"github.com/stretchr/testify/require"
11	12

@@ -41,6 +42,37 @@ func TestTerminalThemeStateClearPalettePreservesReportedColors(t *testing.T) {
41	42		require.True(t, reported, "foreground/background remain usable after palette invalidation")
42	43		}
43	44
	45	+	type queryBoundaryReader struct {
	46	+	first, second []byte
	47	+	readSecond chan struct{}
	48	+	allowSecond, close <-chan struct{}
	49	+	reads int
	50	+	}
	51	+
	52	+	func (r *queryBoundaryReader) Read(p []byte) (int, error) {
	53	+	switch r.reads {
	54	+	case 0:
	55	+	r.reads++
	56	+	return copy(p, r.first), nil
	57	+	case 1:
	58	+	select {
	59	+	case r.readSecond <- struct{}{}:
	60	+	case <-r.close:
	61	+	return 0, io.EOF
	62	+	}
	63	+	select {
	64	+	case <-r.allowSecond:
	65	+	r.reads++
	66	+	return copy(p, r.second), nil
	67	+	case <-r.close:
	68	+	return 0, io.EOF
	69	+	}
	70	+	default:
	71	+	<-r.close
	72	+	return 0, io.EOF
	73	+	}
	74	+	}
	75	+
44	76		func TestStdinPumpCoalescesPaletteAndBackgroundThemePerRead(t *testing.T) {
45	77		state := &terminalThemeState{}
46	78		ctx, cancel := context.WithCancel(context.Background())

@@ -75,3 +107,147 @@ func TestStdinPumpCoalescesPaletteAndBackgroundThemePerRead(t *testing.T) {
75	107		default:
76	108		}
77	109		}
	110	+
	111	+	func TestStdinPumpWaitsForQueryAcknowledgementBeforeNextRead(t *testing.T) {
	112	+	ctx, cancel := context.WithCancel(context.Background())
	113	+	defer cancel()
	114	+	allowSecond := make(chan struct{})
	115	+	closeReader := make(chan struct{})
	116	+	reader := &queryBoundaryReader{
	117	+	first: []byte("\x1b]4;1;#112233\a\x1b[?997;2n"),
	118	+	second: []byte("\x1b]4;2;#445566\a"),
	119	+	readSecond: make(chan struct{}, 1),
	120	+	allowSecond: allowSecond,
	121	+	close: closeReader,
	122	+	}
	123	+	queryStarted := make(chan struct{})
	124	+	allowQuery := make(chan struct{})
	125	+	out := make(chan ports.Frame, 3)
	126	+	pump := stdinPump{
	127	+	ctx: ctx,
	128	+	cancel: cancel,
	129	+	in: reader,
	130	+	out: out,
	131	+	themeState: &terminalThemeState{},
	132	+	logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
	133	+	requestColors: func() bool {
	134	+	close(queryStarted)
	135	+	select {
	136	+	case <-allowQuery:
	137	+	return true
	138	+	case <-ctx.Done():
	139	+	return false
	140	+	}
	141	+	},
	142	+	}
	143	+	pumpDone := make(chan struct{})
	144	+	go func() {
	145	+	pump.run()
	146	+	close(pumpDone)
	147	+	}()
	148	+
	149	+	select {
	150	+	case <-queryStarted:
	151	+	case <-time.After(time.Second):
	152	+	t.Fatal("palette re-query was not requested")
	153	+	}
	154	+	select {
	155	+	case <-reader.readSecond:
	156	+	t.Fatal("stdin pump started the next read before the query acknowledgement")
	157	+	default:
	158	+	}
	159	+
	160	+	close(allowQuery)
	161	+	select {
	162	+	case <-reader.readSecond:
	163	+	case <-time.After(time.Second):
	164	+	t.Fatal("stdin pump did not resume after the query acknowledgement")
	165	+	}
	166	+
	167	+	cancel()
	168	+	close(closeReader)
	169	+	select {
	170	+	case <-pumpDone:
	171	+	case <-time.After(time.Second):
	172	+	t.Fatal("stdin pump did not unwind after cancellation")
	173	+	}
	174	+	}
	175	+
	176	+	type chunksReader struct {
	177	+	chunks [][]byte
	178	+	next int
	179	+	}
	180	+
	181	+	func (r *chunksReader) Read(p []byte) (int, error) {
	182	+	if r.next == len(r.chunks) {
	183	+	return 0, io.EOF
	184	+	}
	185	+	chunk := r.chunks[r.next]
	186	+	r.next++
	187	+	return copy(p, chunk), nil
	188	+	}
	189	+
	190	+	func TestStdinPumpAcknowledgesRepeatedSchemeChunks(t *testing.T) {
	191	+	ctx, cancel := context.WithCancel(context.Background())
	192	+	defer cancel()
	193	+	requests := 0
	194	+	pump := stdinPump{
	195	+	ctx: ctx,
	196	+	cancel: cancel,
	197	+	in: &chunksReader{chunks: [][]byte{
	198	+	[]byte("\x1b[?997;2n"),
	199	+	[]byte("\x1b[?997;1n"),
	200	+	}},
	201	+	out: make(chan ports.Frame, 3),
	202	+	themeState: &terminalThemeState{},
	203	+	logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
	204	+	requestColors: func() bool {
	205	+	requests++
	206	+	return true
	207	+	},
	208	+	}
	209	+
	210	+	pump.run()
	211	+	require.Equal(t, 2, requests, "each completed scheme chunk must receive its own acknowledgement")
	212	+	}
	213	+
	214	+	func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {
	215	+	ctx, cancel := context.WithCancel(context.Background())
	216	+	closeReader := make(chan struct{})
	217	+	reader := &queryBoundaryReader{
	218	+	first: []byte("\x1b[?997;2n"),
	219	+	close: closeReader,
	220	+	}
	221	+	requestStarted := make(chan struct{})
	222	+	pump := stdinPump{
	223	+	ctx: ctx,
	224	+	cancel: cancel,
	225	+	in: reader,
	226	+	out: make(chan ports.Frame, 2),
	227	+	themeState: &terminalThemeState{},
	228	+	logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
	229	+	requestColors: func() bool {
	230	+	close(requestStarted)
	231	+	<-ctx.Done()
	232	+	return false
	233	+	},
	234	+	}
	235	+	pumpDone := make(chan struct{})
	236	+	go func() {
	237	+	pump.run()
	238	+	close(pumpDone)
	239	+	}()
	240	+
	241	+	select {
	242	+	case <-requestStarted:
	243	+	case <-time.After(time.Second):
	244	+	t.Fatal("palette re-query was not requested")
	245	+	}
	246	+	cancel()
	247	+	select {
	248	+	case <-pumpDone:
	249	+	case <-time.After(time.Second):
	250	+	t.Fatal("stdin pump deadlocked waiting for a cancelled query acknowledgement")
	251	+	}
	252	+	close(closeReader)
	253	+	}


@@ -6,6 +6,7 @@ import (
6	"io"	6		"io"
7	"log/slog"	7		"log/slog"
8	"testing"	8		"testing"
		9	+	"time"
9		10
10	"github.com/stretchr/testify/require"	11		"github.com/stretchr/testify/require"
11		12

@@ -41,6 +42,37 @@ func TestTerminalThemeStateClearPalettePreservesReportedColors(t *testing.T) {
41	require.True(t, reported, "foreground/background remain usable after palette invalidation")	42		require.True(t, reported, "foreground/background remain usable after palette invalidation")
42	}	43		}
43		44
		45	+	type queryBoundaryReader struct {
		46	+	first, second []byte
		47	+	readSecond chan struct{}
		48	+	allowSecond, close <-chan struct{}
		49	+	reads int
		50	+	}
		51	+
		52	+	func (r *queryBoundaryReader) Read(p []byte) (int, error) {
		53	+	switch r.reads {
		54	+	case 0:
		55	+	r.reads++
		56	+	return copy(p, r.first), nil
		57	+	case 1:
		58	+	select {
		59	+	case r.readSecond <- struct{}{}:
		60	+	case <-r.close:
		61	+	return 0, io.EOF
		62	+	}
		63	+	select {
		64	+	case <-r.allowSecond:
		65	+	r.reads++
		66	+	return copy(p, r.second), nil
		67	+	case <-r.close:
		68	+	return 0, io.EOF
		69	+	}
		70	+	default:
		71	+	<-r.close
		72	+	return 0, io.EOF
		73	+	}
		74	+	}
		75	+
44	func TestStdinPumpCoalescesPaletteAndBackgroundThemePerRead(t *testing.T) {	76		func TestStdinPumpCoalescesPaletteAndBackgroundThemePerRead(t *testing.T) {
45	state := &terminalThemeState{}	77		state := &terminalThemeState{}
46	ctx, cancel := context.WithCancel(context.Background())	78		ctx, cancel := context.WithCancel(context.Background())

@@ -75,3 +107,147 @@ func TestStdinPumpCoalescesPaletteAndBackgroundThemePerRead(t *testing.T) {
75	default:	107		default:
76	}	108		}
77	}	109		}
		110	+
		111	+	func TestStdinPumpWaitsForQueryAcknowledgementBeforeNextRead(t *testing.T) {
		112	+	ctx, cancel := context.WithCancel(context.Background())
		113	+	defer cancel()
		114	+	allowSecond := make(chan struct{})
		115	+	closeReader := make(chan struct{})
		116	+	reader := &queryBoundaryReader{
		117	+	first: []byte("\x1b]4;1;#112233\a\x1b[?997;2n"),
		118	+	second: []byte("\x1b]4;2;#445566\a"),
		119	+	readSecond: make(chan struct{}, 1),
		120	+	allowSecond: allowSecond,
		121	+	close: closeReader,
		122	+	}
		123	+	queryStarted := make(chan struct{})
		124	+	allowQuery := make(chan struct{})
		125	+	out := make(chan ports.Frame, 3)
		126	+	pump := stdinPump{
		127	+	ctx: ctx,
		128	+	cancel: cancel,
		129	+	in: reader,
		130	+	out: out,
		131	+	themeState: &terminalThemeState{},
		132	+	logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
		133	+	requestColors: func() bool {
		134	+	close(queryStarted)
		135	+	select {
		136	+	case <-allowQuery:
		137	+	return true
		138	+	case <-ctx.Done():
		139	+	return false
		140	+	}
		141	+	},
		142	+	}
		143	+	pumpDone := make(chan struct{})
		144	+	go func() {
		145	+	pump.run()
		146	+	close(pumpDone)
		147	+	}()
		148	+
		149	+	select {
		150	+	case <-queryStarted:
		151	+	case <-time.After(time.Second):
		152	+	t.Fatal("palette re-query was not requested")
		153	+	}
		154	+	select {
		155	+	case <-reader.readSecond:
		156	+	t.Fatal("stdin pump started the next read before the query acknowledgement")
		157	+	default:
		158	+	}
		159	+
		160	+	close(allowQuery)
		161	+	select {
		162	+	case <-reader.readSecond:
		163	+	case <-time.After(time.Second):
		164	+	t.Fatal("stdin pump did not resume after the query acknowledgement")
		165	+	}
		166	+
		167	+	cancel()
		168	+	close(closeReader)
		169	+	select {
		170	+	case <-pumpDone:
		171	+	case <-time.After(time.Second):
		172	+	t.Fatal("stdin pump did not unwind after cancellation")
		173	+	}
		174	+	}
		175	+
		176	+	type chunksReader struct {
		177	+	chunks [][]byte
		178	+	next int
		179	+	}
		180	+
		181	+	func (r *chunksReader) Read(p []byte) (int, error) {
		182	+	if r.next == len(r.chunks) {
		183	+	return 0, io.EOF
		184	+	}
		185	+	chunk := r.chunks[r.next]
		186	+	r.next++
		187	+	return copy(p, chunk), nil
		188	+	}
		189	+
		190	+	func TestStdinPumpAcknowledgesRepeatedSchemeChunks(t *testing.T) {
		191	+	ctx, cancel := context.WithCancel(context.Background())
		192	+	defer cancel()
		193	+	requests := 0
		194	+	pump := stdinPump{
		195	+	ctx: ctx,
		196	+	cancel: cancel,
		197	+	in: &chunksReader{chunks: [][]byte{
		198	+	[]byte("\x1b[?997;2n"),
		199	+	[]byte("\x1b[?997;1n"),
		200	+	}},
		201	+	out: make(chan ports.Frame, 3),
		202	+	themeState: &terminalThemeState{},
		203	+	logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
		204	+	requestColors: func() bool {
		205	+	requests++
		206	+	return true
		207	+	},
		208	+	}
		209	+
		210	+	pump.run()
		211	+	require.Equal(t, 2, requests, "each completed scheme chunk must receive its own acknowledgement")
		212	+	}
		213	+
		214	+	func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {
		215	+	ctx, cancel := context.WithCancel(context.Background())
		216	+	closeReader := make(chan struct{})
		217	+	reader := &queryBoundaryReader{
		218	+	first: []byte("\x1b[?997;2n"),
		219	+	close: closeReader,
		220	+	}
		221	+	requestStarted := make(chan struct{})
		222	+	pump := stdinPump{
		223	+	ctx: ctx,
		224	+	cancel: cancel,
		225	+	in: reader,
		226	+	out: make(chan ports.Frame, 2),
		227	+	themeState: &terminalThemeState{},
		228	+	logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
		229	+	requestColors: func() bool {
		230	+	close(requestStarted)
		231	+	<-ctx.Done()
		232	+	return false
		233	+	},
		234	+	}
		235	+	pumpDone := make(chan struct{})
		236	+	go func() {
		237	+	pump.run()
		238	+	close(pumpDone)
		239	+	}()
		240	+
		241	+	select {
		242	+	case <-requestStarted:
		243	+	case <-time.After(time.Second):
		244	+	t.Fatal("palette re-query was not requested")
		245	+	}
		246	+	cancel()
		247	+	select {
		248	+	case <-pumpDone:
		249	+	case <-time.After(time.Second):
		250	+	t.Fatal("stdin pump deadlocked waiting for a cancelled query acknowledgement")
		251	+	}
		252	+	close(closeReader)
		253	+	}