fix(client): delimit palette query generations

commit: 9bbe15fb7525671e951c76c2b335e35cd2710bcc
author: brice · 3mo ago · 2026-07-17 11:31 UTC
parent: a47fd09
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +182 -122

M internal/usecase/client/client.go +124 -33


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

@@ -601,27 +602,25 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
601	602		ackQueue := newCumulativeAckQueue()
602	603		sendErrCh := make(chan error, 1)
603	604
604		-	// colorQueryCh hands an OSC 10/11 re-query request from the stdin pump to
605		-	// the main loop, which is the only goroutine allowed to touch term.Out()/
606		-	// term.Flush(); QueryColors writes through the same batched writer, so
607		-	// calling it directly from the stdin pump would race the main loop's
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.
	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.
612	610		colorQueryCh := make(chan colorQueryRequest)
613		-	requestColors := func() bool {
614		-	request := colorQueryRequest{acknowledged: make(chan struct{})}
	611	+	paletteReadyCh := make(chan paletteQueryReady)
	612	+	requestColors := func() (colorQueryRequest, bool) {
	613	+	request := colorQueryRequest{sentinelWritten: make(chan struct{})}
615	614		select {
616	615		case colorQueryCh <- request:
617	616		case <-loopCtx.Done():
618		-	return false
	617	+	return colorQueryRequest{}, false
619	618		}
620	619		select {
621		-	case <-request.acknowledged:
622		-	return true
	620	+	case <-request.sentinelWritten:
	621	+	return request, true
623	622		case <-loopCtx.Done():
624		-	return false
	623	+	return colorQueryRequest{}, false
625	624		}
626	625		}
627	626

@@ -630,7 +629,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
630	629		clip = clipboard
631	630		}
632	631		go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)
633		-	go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, themeState: themeState, clipboard: clip, logger: log, requestColors: requestColors}).run()
	632	+	go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, themeState: themeState, clipboard: clip, logger: log, requestColors: requestColors, paletteReady: paletteReadyCh}).run()
634	633		go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
635	634
636	635		// 5. Output/main loop: the only goroutine that touches the terminal.

@@ -670,10 +669,18 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
670	669		return welcomedResult(errLinkOffline)
671	670		}
672	671		case request := <-colorQueryCh:
673		-	// A fresh query supersedes every prior OSC 4 response. Queue the
674		-	// cleared snapshot before writing the query so the daemon never
675		-	// applies stale palette entries while it awaits the new responses.
676		-	clearedTheme := themeState.clearPalette()
	672	+	if _, err := term.Out().Write([]byte("\x1b[5n")); err != nil {
	673	+	return welcomedResult(fmt.Errorf("vev: writing palette sentinel: %w", err))
	674	+	}
	675	+	if err := term.Flush(); err != nil {
	676	+	return welcomedResult(fmt.Errorf("vev: flushing palette sentinel: %w", err))
	677	+	}
	678	+	close(request.sentinelWritten)
	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.
	683	+	clearedTheme := themeState.snapshot()
677	684		select {
678	685		case sendCh <- ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(clearedTheme)}:
679	686		case <-loopCtx.Done():

@@ -682,7 +689,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
682	689		if err := term.QueryColors(); err != nil {
683	690		log.Warn("querying terminal colors", "err", err)
684	691		}
685		-	close(request.acknowledged)
	692	+	close(ready.queryIssued)
686	693		case r := <-recvCh:
687	694		if r.err != nil {
688	695		if errors.Is(r.err, io.EOF) {

@@ -841,7 +848,50 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
841	848		// terminal — and matches the standard pattern for stdin pumps; a
842	849		// closable stdin duplicate could lift it later if ever needed.
843	850		type colorQueryRequest struct {
844		-	acknowledged chan struct{}
	851	+	sentinelWritten chan struct{}
	852	+	}
	853	+
	854	+	type paletteQueryReady struct {
	855	+	queryIssued chan struct{}
	856	+	}
	857	+
	858	+	type paletteDrain struct {
	859	+	pending []byte
	860	+	}
	861	+
	862	+	type palettePhase uint8
	863	+
	864	+	const (
	865	+	paletteAccepting palettePhase = iota
	866	+	paletteBoundaryRequested
	867	+	paletteDraining
	868	+	paletteAwaitingQuery
	869	+	)
	870	+
	871	+	var paletteSentinel = []byte("\x1b[0n")
	872	+
	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()) {
	877	+	for _, b := range data {
	878	+	d.pending = append(d.pending, b)
	879	+	for len(d.pending) > 0 && !bytes.HasPrefix(paletteSentinel, d.pending) {
	880	+	onBytes(d.pending[:1])
	881	+	d.pending = d.pending[1:]
	882	+	}
	883	+	if bytes.Equal(d.pending, paletteSentinel) {
	884	+	d.pending = nil
	885	+	onSentinel()
	886	+	}
	887	+	}
	888	+	}
	889	+
	890	+	func (d *paletteDrain) flush(onBytes func([]byte)) {
	891	+	if len(d.pending) != 0 {
	892	+	onBytes(d.pending)
	893	+	d.pending = nil
	894	+	}
845	895		}
846	896
847	897		type stdinPump struct {

@@ -853,7 +903,8 @@ type stdinPump struct {
853	903		themeState *terminalThemeState
854	904		clipboard ports.ClipboardReader
855	905		logger *slog.Logger
856		-	requestColors func() bool
	906	+	requestColors func() (colorQueryRequest, bool)
	907	+	paletteReady chan<- paletteQueryReady
857	908		}
858	909
859	910		func (p *stdinPump) run() {

@@ -911,12 +962,37 @@ func (p *stdinPump) run() {
911	962		sink = ci.Scan
912	963		}
913	964
	965	+	phase := paletteAccepting
	966	+	var drain paletteDrain
914	967		for {
915	968		n, rerr := in.Read(buf)
916	969		if n > 0 {
917	970		themeChanged := false
918		-	queryColors := false
	971	+	queryNeeded := false
	972	+	onBytes := func(data []byte) {
	973	+	if phase != paletteDraining {
	974	+	sink(data)
	975	+	return
	976	+	}
	977	+	drain.scan(data, sink, func() {
	978	+	phase = paletteAwaitingQuery
	979	+	ready := paletteQueryReady{queryIssued: make(chan struct{})}
	980	+	select {
	981	+	case p.paletteReady <- ready:
	982	+	case <-ctx.Done():
	983	+	return
	984	+	}
	985	+	select {
	986	+	case <-ready.queryIssued:
	987	+	phase = paletteAccepting
	988	+	case <-ctx.Done():
	989	+	}
	990	+	})
	991	+	}
919	992		scanner.Scan(buf[:n], func(kind int, rgb renderer.RGB) {
	993	+	if phase != paletteAccepting {
	994	+	return
	995	+	}
920	996		themeState.update(func(current *ports.Theme) {
921	997		switch kind {
922	998		case 10:

@@ -929,6 +1005,9 @@ func (p *stdinPump) run() {
929	1005		})
930	1006		themeChanged = true
931	1007		}, func(slot int, rgb renderer.RGB) {
	1008	+	if phase != paletteAccepting {
	1009	+	return
	1010	+	}
932	1011		themeState.update(func(current *ports.Theme) {
933	1012		current.PaletteKnown \|= uint16(1) << slot
934	1013		current.Palette[slot] = rgb

@@ -939,23 +1018,35 @@ func (p *stdinPump) run() {
939	1018		current.SchemeKnown = true
940	1019		current.Light = light
941	1020		})
942		-	themeChanged = true
943		-	queryColors = true
944		-	}, sink)
945		-	if themeChanged {
	1021	+	// The scheme itself begins a new palette generation. Suppress every
	1022	+	// color reply after it, including those remaining in this read, and
	1023	+	// defer invalidation until Scan completes so no partial snapshot is
	1024	+	// observable mid-chunk.
	1025	+	themeChanged = false
	1026	+	if phase == paletteAccepting {
	1027	+	phase = paletteBoundaryRequested
	1028	+	queryNeeded = true
	1029	+	}
	1030	+	}, onBytes)
	1031	+	if queryNeeded {
	1032	+	themeState.clearPalette()
	1033	+	if requestColors != nil {
	1034	+	if _, ok := requestColors(); !ok {
	1035	+	return
	1036	+	}
	1037	+	phase = paletteDraining
	1038	+	}
	1039	+	} else if themeChanged {
946	1040		send(ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(themeState.snapshot())})
947	1041		}
948		-	// Defer the re-query until this read's complete snapshot is queued;
949		-	// otherwise the main loop could invalidate palette state midway
950		-	// through a chunk containing multiple terminal responses.
951		-	if queryColors && requestColors != nil && !requestColors() {
952		-	return
953		-	}
954	1042		if !sendOK.Load() {
955	1043		return
956	1044		}
957	1045		}
958	1046		if rerr != nil {
	1047	+	if phase == paletteDraining {
	1048	+	drain.flush(sink)
	1049	+	}
959	1050		// Best-effort detach notification, then unwind.
960	1051		select {
961	1052		case out <- ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})}:

 package client
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
 	"errors"
 	ackQueue := newCumulativeAckQueue()
 	sendErrCh := make(chan error, 1)
-	// colorQueryCh hands an OSC 10/11 re-query request from the stdin pump to
+	// Palette re-queries use DSR (CSI 5 n) as an ordered sentinel. The stdin
-	// the main loop, which is the only goroutine allowed to touch term.Out()/
+	// pump drains terminal replies through CSI 0 n before it permits the main
-	// term.Flush(); QueryColors writes through the same batched writer, so
+	// loop to publish the cleared scheme and issue QueryColors. This separates
-	// calling it directly from the stdin pump would race the main loop's
+	// bytes already queued (including Scanner.pending) from replacement replies
-	// output writes. The acknowledgement creates a generation boundary: after
+	// without allowing a second goroutine to write to the terminal.
-	// a scheme response, stdin must not read another chunk (which may contain
-	// old OSC 4 replies) until this loop has cleared the palette and issued the
-	// replacement query.
 	colorQueryCh := make(chan colorQueryRequest)
-	requestColors := func() bool {
+	paletteReadyCh := make(chan paletteQueryReady)
-		request := colorQueryRequest{acknowledged: make(chan struct{})}
+	requestColors := func() (colorQueryRequest, bool) {
+		request := colorQueryRequest{sentinelWritten: make(chan struct{})}
 		select {
 		case colorQueryCh <- request:
 		case <-loopCtx.Done():
-			return false
+			return colorQueryRequest{}, false
 		}
 		select {
-		case <-request.acknowledged:
+		case <-request.sentinelWritten:
-			return true
+			return request, true
 		case <-loopCtx.Done():
-			return false
+			return colorQueryRequest{}, false
 		}
 	}
 		clip = clipboard
 	}
 	go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)
-	go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, themeState: themeState, clipboard: clip, logger: log, requestColors: requestColors}).run()
+	go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, themeState: themeState, clipboard: clip, logger: log, requestColors: requestColors, paletteReady: paletteReadyCh}).run()
 	go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
 	// 5. Output/main loop: the only goroutine that touches the terminal.
 				return welcomedResult(errLinkOffline)
 			}
 		case request := <-colorQueryCh:
-			// A fresh query supersedes every prior OSC 4 response. Queue the
+			if _, err := term.Out().Write([]byte("\x1b[5n")); err != nil {
-			// cleared snapshot before writing the query so the daemon never
+				return welcomedResult(fmt.Errorf("vev: writing palette sentinel: %w", err))
-			// applies stale palette entries while it awaits the new responses.
+			}
-			clearedTheme := themeState.clearPalette()
+			if err := term.Flush(); err != nil {
+				return welcomedResult(fmt.Errorf("vev: flushing palette sentinel: %w", err))
+			}
+			close(request.sentinelWritten)
+		case ready := <-paletteReadyCh:
+			// CSI 0 n proves all earlier replies have been discarded by stdin.
+			// Publish the one cleared scheme snapshot before 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)}:
 			case <-loopCtx.Done():
 			if err := term.QueryColors(); err != nil {
 				log.Warn("querying terminal colors", "err", err)
 			}
-			close(request.acknowledged)
+			close(ready.queryIssued)
 		case r := <-recvCh:
 			if r.err != nil {
 				if errors.Is(r.err, io.EOF) {
 // 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{}
+	sentinelWritten chan struct{}
+}
+type paletteQueryReady struct {
+	queryIssued chan struct{}
+}
+type paletteDrain struct {
+	pending []byte
+}
+type palettePhase uint8
+const (
+	paletteAccepting palettePhase = iota
+	paletteBoundaryRequested
+	paletteDraining
+	paletteAwaitingQuery
+)
+var paletteSentinel = []byte("\x1b[0n")
+// scan consumes the DSR response while forwarding every other byte in order.
+// It retains only a possible sentinel prefix, so a split CSI 0 n is consumed
+// without dropping a user's partial escape sequence on cancellation.
+func (d *paletteDrain) scan(data []byte, onBytes func([]byte), onSentinel func()) {
+	for _, b := range data {
+		d.pending = append(d.pending, b)
+		for len(d.pending) > 0 && !bytes.HasPrefix(paletteSentinel, d.pending) {
+			onBytes(d.pending[:1])
+			d.pending = d.pending[1:]
+		}
+		if bytes.Equal(d.pending, paletteSentinel) {
+			d.pending = nil
+			onSentinel()
+		}
+	}
+}
+func (d *paletteDrain) flush(onBytes func([]byte)) {
+	if len(d.pending) != 0 {
+		onBytes(d.pending)
+		d.pending = nil
+	}
 }
 type stdinPump struct {
 	themeState    *terminalThemeState
 	clipboard     ports.ClipboardReader
 	logger        *slog.Logger
-	requestColors func() bool
+	requestColors func() (colorQueryRequest, bool)
+	paletteReady  chan<- paletteQueryReady
 }
 func (p *stdinPump) run() {
 		sink = ci.Scan
 	}
+	phase := paletteAccepting
+	var drain paletteDrain
 	for {
 		n, rerr := in.Read(buf)
 		if n > 0 {
 			themeChanged := false
-			queryColors := false
+			queryNeeded := false
+			onBytes := func(data []byte) {
+				if phase != paletteDraining {
+					sink(data)
+					return
+				}
+				drain.scan(data, sink, func() {
+					phase = paletteAwaitingQuery
+					ready := paletteQueryReady{queryIssued: make(chan struct{})}
+					select {
+					case p.paletteReady <- ready:
+					case <-ctx.Done():
+						return
+					}
+					select {
+					case <-ready.queryIssued:
+						phase = paletteAccepting
+					case <-ctx.Done():
+					}
+				})
+			}
 			scanner.Scan(buf[:n], func(kind int, rgb renderer.RGB) {
+				if phase != paletteAccepting {
+					return
+				}
 				themeState.update(func(current *ports.Theme) {
 					switch kind {
 					case 10:
 				})
 				themeChanged = true
 			}, func(slot int, rgb renderer.RGB) {
+				if phase != paletteAccepting {
+					return
+				}
 				themeState.update(func(current *ports.Theme) {
 					current.PaletteKnown |= uint16(1) << slot
 					current.Palette[slot] = rgb
 					current.SchemeKnown = true
 					current.Light = light
 				})
-				themeChanged = true
+				// The scheme itself begins a new palette generation. Suppress every
-				queryColors = true
+				// color reply after it, including those remaining in this read, and
-			}, sink)
+				// defer invalidation until Scan completes so no partial snapshot is
-			if themeChanged {
+				// observable mid-chunk.
+				themeChanged = false
+				if phase == paletteAccepting {
+					phase = paletteBoundaryRequested
+					queryNeeded = true
+				}
+			}, onBytes)
+			if queryNeeded {
+				themeState.clearPalette()
+				if requestColors != nil {
+					if _, ok := requestColors(); !ok {
+						return
+					}
+					phase = paletteDraining
+				}
+			} else if themeChanged {
 				send(ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(themeState.snapshot())})
 			}
-			// 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 && !requestColors() {
-				return
-			}
 			if !sendOK.Load() {
 				return
 			}
 		}
 		if rerr != nil {
+			if phase == paletteDraining {
+				drain.flush(sink)
+			}
 			// Best-effort detach notification, then unwind.
 			select {
 			case out <- ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})}:

M internal/usecase/client/client_test.go +49 -80


@@ -1082,108 +1082,78 @@ func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) {
1082	1082		require.Equal(t, int32(1), term.restoreCount.Load())
1083	1083		}
1084	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)
	1085	+	func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T) {
	1086	+	input := newChunkedBlockingReader(
	1087	+	[]byte("a\x1b]4;1;#112233\a\x1b[?997;2n\x1b]4;2;#44"),
	1088	+	[]byte("5566\ab\x1b["),
	1089	+	[]byte("0nc\x1b]4;3;#778899\ad"),
	1090	+	)
	1091	+	defer input.unblock()
1098	1092
1099	1093		var out bytes.Buffer
1100	1094		var restoreCount atomic.Int32
1101	1095		resizeCh := make(chan domain.Size)
1102		-	queryStarted := make(chan struct{})
1103		-	allowQuery := make(chan struct{})
1104	1096		tm := portsmocks.NewMockTerminal(t)
1105	1097		tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
1106	1098		tm.EXPECT().EnterRaw().Return(func() error { restoreCount.Add(1); return nil }, nil).Once()
1107	1099		tm.EXPECT().In().Return(input).Maybe()
1108	1100		tm.EXPECT().Out().Return(&out).Maybe()
1109	1101		tm.EXPECT().Flush().Return(nil).Maybe()
	1102	+	tm.EXPECT().QueryColors().Return(nil).Once()
1110	1103		tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
1111		-	tm.EXPECT().QueryColors().RunAndReturn(func() error {
1112		-	close(queryStarted)
1113		-	<-allowQuery
1114		-	return nil
1115		-	}).Once()
1116	1104
1117		-	themes := make(chan ports.Theme, 3)
1118		-	recvRelease := make(chan struct{})
1119		-	var welcomed atomic.Bool
	1105	+	var inputBytes []byte
	1106	+	var inputMu sync.Mutex
	1107	+	themes := make(chan ports.Theme, 2)
	1108	+	freshTheme := make(chan struct{})
	1109	+	var freshOnce sync.Once
1120	1110		tr := portsmocks.NewMockTransport(t)
1121	1111		tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
	1112	+	tr.EXPECT().Send(isType(ports.MsgInput)).RunAndReturn(func(f ports.Frame) error {
	1113	+	in, err := ports.UnmarshalInput(f.Payload)
	1114	+	if err != nil {
	1115	+	return err
	1116	+	}
	1117	+	inputMu.Lock()
	1118	+	inputBytes = append(inputBytes, in.Data...)
	1119	+	inputMu.Unlock()
	1120	+	return nil
	1121	+	}).Maybe()
1122	1122		tr.EXPECT().Send(isType(ports.MsgTheme)).RunAndReturn(func(f ports.Frame) error {
1123	1123		theme, err := ports.UnmarshalTheme(f.Payload)
1124	1124		if err != nil {
1125		-	t.Errorf("decode sent theme: %v", err)
1126	1125		return err
1127	1126		}
1128	1127		themes <- theme
	1128	+	if theme.PaletteKnown == 1<<3 {
	1129	+	freshOnce.Do(func() { close(freshTheme) })
	1130	+	}
1129	1131		return nil
1130		-	}).Times(3)
	1132	+	}).Times(2)
	1133	+	var welcomed atomic.Bool
1131	1134		tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
1132	1135		if welcomed.CompareAndSwap(false, true) {
1133	1136		return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"})), nil
1134	1137		}
1135		-	<-recvRelease
1136		-	return ports.Frame{}, io.EOF
	1138	+	<-freshTheme
	1139	+	return frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach})), nil
1137	1140		}).Maybe()
1138	1141		tr.EXPECT().Close().Return(nil).Once()
1139	1142
1140		-	runDone := make(chan error, 1)
1141		-	go func() {
1142		-	runDone <- runTestClient(ctx, attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral})
1143		-	}()
1144		-
1145		-	select {
1146		-	case <-queryStarted:
1147		-	case <-time.After(time.Second):
1148		-	t.Fatal("main loop did not start the palette re-query")
1149		-	}
1150		-	select {
1151		-	case <-input.secondRead:
1152		-	t.Fatal("stdin read the next chunk before the cleared theme and query completed")
1153		-	default:
1154		-	}
1155		-
1156		-	close(allowQuery)
1157		-	select {
1158		-	case <-input.secondRead:
1159		-	case <-time.After(time.Second):
1160		-	t.Fatal("stdin did not resume after the palette re-query completed")
1161		-	}
1162		-	close(allowSecond)
1163		-
1164		-	var got [3]ports.Theme
1165		-	for i := range got {
1166		-	select {
1167		-	case got[i] = <-themes:
1168		-	case <-time.After(time.Second):
1169		-	t.Fatalf("theme %d was not sent", i)
1170		-	}
1171		-	}
1172		-	require.Equal(t, uint16(1<<1), got[0].PaletteKnown)
1173		-	require.True(t, got[0].SchemeKnown)
1174		-	require.Zero(t, got[1].PaletteKnown, "the old palette must be cleared before a later read is admitted")
1175		-	require.True(t, got[1].SchemeKnown)
1176		-	require.Equal(t, uint16(1<<2), got[2].PaletteKnown, "only the post-query chunk may populate the new palette generation")
1177		-
1178		-	cancel()
1179		-	select {
1180		-	case err := <-runDone:
1181		-	require.NoError(t, err)
1182		-	case <-time.After(time.Second):
1183		-	t.Fatal("attach did not return after cancellation")
1184		-	}
1185		-	close(recvRelease)
	1143	+	require.NoError(t, runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral}))
	1144	+	require.Equal(t, "\x1b[5n", out.String(), "the DSR sentinel must be the only main-loop terminal write")
1186	1145		require.Equal(t, int32(1), restoreCount.Load())
	1146	+
	1147	+	cleared := <-themes
	1148	+	fresh := <-themes
	1149	+	require.True(t, cleared.SchemeKnown)
	1150	+	require.True(t, cleared.Light)
	1151	+	require.Zero(t, cleared.PaletteKnown, "only the cleared scheme snapshot may cross the boundary")
	1152	+	require.Equal(t, uint16(1<<3), fresh.PaletteKnown, "only colors after the sentinel and replacement query are accepted")
	1153	+	require.Zero(t, fresh.PaletteKnown&(1<<1\|1<<2), "pre-boundary palette slots must not leak into the new generation")
	1154	+	inputMu.Lock()
	1155	+	require.Equal(t, []byte("abcd"), inputBytes, "ordinary input around split stale and sentinel responses must survive")
	1156	+	inputMu.Unlock()
1187	1157		}
1188	1158
1189	1159		func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {

@@ -1193,6 +1163,7 @@ func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1193	1163		input := newChunkedBlockingReader(
1194	1164		[]byte("\x1b]11;#010203\a\x1b]4;1;#112233\a\x1b]4;14;#778899\a"),
1195	1165		[]byte("\x1b[?997;2n"),
	1166	+	[]byte("\x1b[0n"),
1196	1167		)
1197	1168		defer input.unblock()
1198	1169

@@ -1224,7 +1195,7 @@ func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1224	1195		clearOnce.Do(func() { close(clearedPalette) })
1225	1196		}
1226	1197		return nil
1227		-	}).Times(3)
	1198	+	}).Times(2)
1228	1199
1229	1200		welcome := frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))
1230	1201		detached := frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))

@@ -1250,13 +1221,11 @@ func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
1250	1221
1251	1222		themesMu.Lock()
1252	1223		defer themesMu.Unlock()
1253		-	require.Len(t, themes, 3, "one complete snapshot per input chunk plus the re-query invalidation")
	1224	+	require.Len(t, themes, 2, "the old palette snapshot and one cleared scheme snapshot are sent")
1254	1225		require.Equal(t, uint16(1<<1\|1<<14), themes[0].PaletteKnown)
1255	1226		require.True(t, themes[0].HasBackground)
1256		-	require.Equal(t, uint16(1<<1\|1<<14), themes[1].PaletteKnown, "scheme update retains palette until re-query")
	1227	+	require.Zero(t, themes[1].PaletteKnown, "the DSR-delimited re-query must invalidate stale palette entries")
	1228	+	require.Equal(t, themes[0].Background, themes[1].Background)
1257	1229		require.True(t, themes[1].SchemeKnown)
1258		-	require.Zero(t, themes[2].PaletteKnown, "re-query must invalidate stale palette entries")
1259		-	require.Equal(t, themes[0].Background, themes[2].Background)
1260		-	require.True(t, themes[2].SchemeKnown)
1261		-	require.True(t, themes[2].Light)
	1230	+	require.True(t, themes[1].Light)
1262	1231		}

 	require.Equal(t, int32(1), term.restoreCount.Load())
 }
-func TestAttachSchemeRequeryCreatesPaletteGenerationBoundary(t *testing.T) {
+func TestAttachSchemeRequeryUsesSentinelToDiscardSplitStalePalette(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
+	input := newChunkedBlockingReader(
-	defer cancel()
+		[]byte("a\x1b]4;1;#112233\a\x1b[?997;2n\x1b]4;2;#44"),
-	allowSecond := make(chan struct{})
+		[]byte("5566\ab\x1b["),
-	inputDone := make(chan struct{})
+		[]byte("0nc\x1b]4;3;#778899\ad"),
-	input := &queryBoundaryReader{
+	)
-		first:       []byte("\x1b]4;1;#112233\a\x1b[?997;2n"),
+	defer input.unblock()
-		second:      []byte("\x1b]4;2;#445566\a"),
-		secondRead:  make(chan struct{}, 1),
-		allowSecond: allowSecond,
-		done:        inputDone,
-	}
-	defer close(inputDone)
 	var out bytes.Buffer
 	var restoreCount atomic.Int32
 	resizeCh := make(chan domain.Size)
-	queryStarted := make(chan struct{})
-	allowQuery := make(chan struct{})
 	tm := portsmocks.NewMockTerminal(t)
 	tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
 	tm.EXPECT().EnterRaw().Return(func() error { restoreCount.Add(1); return nil }, nil).Once()
 	tm.EXPECT().In().Return(input).Maybe()
 	tm.EXPECT().Out().Return(&out).Maybe()
 	tm.EXPECT().Flush().Return(nil).Maybe()
+	tm.EXPECT().QueryColors().Return(nil).Once()
 	tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
-	tm.EXPECT().QueryColors().RunAndReturn(func() error {
-		close(queryStarted)
-		<-allowQuery
-		return nil
-	}).Once()
-	themes := make(chan ports.Theme, 3)
+	var inputBytes []byte
-	recvRelease := make(chan struct{})
+	var inputMu sync.Mutex
-	var welcomed atomic.Bool
+	themes := make(chan ports.Theme, 2)
+	freshTheme := make(chan struct{})
+	var freshOnce sync.Once
 	tr := portsmocks.NewMockTransport(t)
 	tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
+	tr.EXPECT().Send(isType(ports.MsgInput)).RunAndReturn(func(f ports.Frame) error {
+		in, err := ports.UnmarshalInput(f.Payload)
+		if err != nil {
+			return err
+		}
+		inputMu.Lock()
+		inputBytes = append(inputBytes, in.Data...)
+		inputMu.Unlock()
+		return nil
+	}).Maybe()
 	tr.EXPECT().Send(isType(ports.MsgTheme)).RunAndReturn(func(f ports.Frame) error {
 		theme, err := ports.UnmarshalTheme(f.Payload)
 		if err != nil {
-			t.Errorf("decode sent theme: %v", err)
 			return err
 		}
 		themes <- theme
+		if theme.PaletteKnown == 1<<3 {
+			freshOnce.Do(func() { close(freshTheme) })
+		}
 		return nil
-	}).Times(3)
+	}).Times(2)
+	var welcomed atomic.Bool
 	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
 		if welcomed.CompareAndSwap(false, true) {
 			return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"})), nil
 		}
-		<-recvRelease
+		<-freshTheme
-		return ports.Frame{}, io.EOF
+		return frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach})), nil
 	}).Maybe()
 	tr.EXPECT().Close().Return(nil).Once()
-	runDone := make(chan error, 1)
+	require.NoError(t, runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral}))
-	go func() {
+	require.Equal(t, "\x1b[5n", out.String(), "the DSR sentinel must be the only main-loop terminal write")
-		runDone <- runTestClient(ctx, attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral})
-	}()
-	select {
-	case <-queryStarted:
-	case <-time.After(time.Second):
-		t.Fatal("main loop did not start the palette re-query")
-	}
-	select {
-	case <-input.secondRead:
-		t.Fatal("stdin read the next chunk before the cleared theme and query completed")
-	default:
-	}
-	close(allowQuery)
-	select {
-	case <-input.secondRead:
-	case <-time.After(time.Second):
-		t.Fatal("stdin did not resume after the palette re-query completed")
-	}
-	close(allowSecond)
-	var got [3]ports.Theme
-	for i := range got {
-		select {
-		case got[i] = <-themes:
-		case <-time.After(time.Second):
-			t.Fatalf("theme %d was not sent", i)
-		}
-	}
-	require.Equal(t, uint16(1<<1), got[0].PaletteKnown)
-	require.True(t, got[0].SchemeKnown)
-	require.Zero(t, got[1].PaletteKnown, "the old palette must be cleared before a later read is admitted")
-	require.True(t, got[1].SchemeKnown)
-	require.Equal(t, uint16(1<<2), got[2].PaletteKnown, "only the post-query chunk may populate the new palette generation")
-	cancel()
-	select {
-	case err := <-runDone:
-		require.NoError(t, err)
-	case <-time.After(time.Second):
-		t.Fatal("attach did not return after cancellation")
-	}
-	close(recvRelease)
 	require.Equal(t, int32(1), restoreCount.Load())
+	cleared := <-themes
+	fresh := <-themes
+	require.True(t, cleared.SchemeKnown)
+	require.True(t, cleared.Light)
+	require.Zero(t, cleared.PaletteKnown, "only the cleared scheme snapshot may cross the boundary")
+	require.Equal(t, uint16(1<<3), fresh.PaletteKnown, "only colors after the sentinel and replacement query are accepted")
+	require.Zero(t, fresh.PaletteKnown&(1<<1|1<<2), "pre-boundary palette slots must not leak into the new generation")
+	inputMu.Lock()
+	require.Equal(t, []byte("abcd"), inputBytes, "ordinary input around split stale and sentinel responses must survive")
+	inputMu.Unlock()
 }
 func TestAttachSchemeRequeryClearsStalePalette(t *testing.T) {
 	input := newChunkedBlockingReader(
 		[]byte("\x1b]11;#010203\a\x1b]4;1;#112233\a\x1b]4;14;#778899\a"),
 		[]byte("\x1b[?997;2n"),
+		[]byte("\x1b[0n"),
 	)
 	defer input.unblock()
 			clearOnce.Do(func() { close(clearedPalette) })
 		}
 		return nil
-	}).Times(3)
+	}).Times(2)
 	welcome := frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))
 	detached := frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))
 	themesMu.Lock()
 	defer themesMu.Unlock()
-	require.Len(t, themes, 3, "one complete snapshot per input chunk plus the re-query invalidation")
+	require.Len(t, themes, 2, "the old palette snapshot and one cleared scheme snapshot are sent")
 	require.Equal(t, uint16(1<<1|1<<14), themes[0].PaletteKnown)
 	require.True(t, themes[0].HasBackground)
-	require.Equal(t, uint16(1<<1|1<<14), themes[1].PaletteKnown, "scheme update retains palette until re-query")
+	require.Zero(t, themes[1].PaletteKnown, "the DSR-delimited re-query must invalidate stale palette entries")
+	require.Equal(t, themes[0].Background, themes[1].Background)
 	require.True(t, themes[1].SchemeKnown)
-	require.Zero(t, themes[2].PaletteKnown, "re-query must invalidate stale palette entries")
+	require.True(t, themes[1].Light)
-	require.Equal(t, themes[0].Background, themes[2].Background)
-	require.True(t, themes[2].SchemeKnown)
-	require.True(t, themes[2].Light)
 }

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


@@ -130,13 +130,13 @@ func TestStdinPumpWaitsForQueryAcknowledgementBeforeNextRead(t *testing.T) {
130	130		out: out,
131	131		themeState: &terminalThemeState{},
132	132		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
133		-	requestColors: func() bool {
	133	+	requestColors: func() (colorQueryRequest, bool) {
134	134		close(queryStarted)
135	135		select {
136	136		case <-allowQuery:
137		-	return true
	137	+	return colorQueryRequest{}, true
138	138		case <-ctx.Done():
139		-	return false
	139	+	return colorQueryRequest{}, false
140	140		}
141	141		},
142	142		}

@@ -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 TestStdinPumpAcknowledgesRepeatedSchemeChunks(t *testing.T) {
	190	+	func TestStdinPumpDefersRepeatedSchemeChunksUntilSentinel(t *testing.T) {
191	191		ctx, cancel := context.WithCancel(context.Background())
192	192		defer cancel()
193	193		requests := 0

@@ -201,14 +201,14 @@ func TestStdinPumpAcknowledgesRepeatedSchemeChunks(t *testing.T) {
201	201		out: make(chan ports.Frame, 3),
202	202		themeState: &terminalThemeState{},
203	203		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
204		-	requestColors: func() bool {
	204	+	requestColors: func() (colorQueryRequest, bool) {
205	205		requests++
206		-	return true
	206	+	return colorQueryRequest{}, true
207	207		},
208	208		}
209	209
210	210		pump.run()
211		-	require.Equal(t, 2, requests, "each completed scheme chunk must receive its own acknowledgement")
	211	+	require.Equal(t, 1, requests, "a later scheme response is drained until the first sentinel arrives")
212	212		}
213	213
214	214		func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {

@@ -226,10 +226,10 @@ func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T)
226	226		out: make(chan ports.Frame, 2),
227	227		themeState: &terminalThemeState{},
228	228		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
229		-	requestColors: func() bool {
	229	+	requestColors: func() (colorQueryRequest, bool) {
230	230		close(requestStarted)
231	231		<-ctx.Done()
232		-	return false
	232	+	return colorQueryRequest{}, false
233	233		},
234	234		}
235	235		pumpDone := make(chan struct{})


@@ -130,13 +130,13 @@ func TestStdinPumpWaitsForQueryAcknowledgementBeforeNextRead(t *testing.T) {
130		out: out,	130		out: out,
131		themeState: &terminalThemeState{},	131		themeState: &terminalThemeState{},
132		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),	132		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
133	-	requestColors: func() bool {	133	+	requestColors: func() (colorQueryRequest, bool) {
134		close(queryStarted)	134		close(queryStarted)
135		select {	135		select {
136		case <-allowQuery:	136		case <-allowQuery:
137	-	return true	137	+	return colorQueryRequest{}, true
138		case <-ctx.Done():	138		case <-ctx.Done():
139	-	return false	139	+	return colorQueryRequest{}, false
140		}	140		}
141		},	141		},
142		}	142		}

@@ -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 TestStdinPumpAcknowledgesRepeatedSchemeChunks(t *testing.T) {	190	+	func TestStdinPumpDefersRepeatedSchemeChunksUntilSentinel(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

@@ -201,14 +201,14 @@ func TestStdinPumpAcknowledgesRepeatedSchemeChunks(t *testing.T) {
201		out: make(chan ports.Frame, 3),	201		out: make(chan ports.Frame, 3),
202		themeState: &terminalThemeState{},	202		themeState: &terminalThemeState{},
203		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),	203		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
204	-	requestColors: func() bool {	204	+	requestColors: func() (colorQueryRequest, bool) {
205		requests++	205		requests++
206	-	return true	206	+	return colorQueryRequest{}, true
207		},	207		},
208		}	208		}
209			209
210		pump.run()	210		pump.run()
211	-	require.Equal(t, 2, requests, "each completed scheme chunk must receive its own acknowledgement")	211	+	require.Equal(t, 1, requests, "a later scheme response is drained until the first sentinel arrives")
212		}	212		}
213			213
214		func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {	214		func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T) {

@@ -226,10 +226,10 @@ func TestStdinPumpQueryAcknowledgementCancellationDoesNotDeadlock(t *testing.T)
226		out: make(chan ports.Frame, 2),	226		out: make(chan ports.Frame, 2),
227		themeState: &terminalThemeState{},	227		themeState: &terminalThemeState{},
228		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),	228		logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
229	-	requestColors: func() bool {	229	+	requestColors: func() (colorQueryRequest, bool) {
230		close(requestStarted)	230		close(requestStarted)
231		<-ctx.Done()	231		<-ctx.Done()
232	-	return false	232	+	return colorQueryRequest{}, false
233		},	233		},
234		}	234		}
235		pumpDone := make(chan struct{})	235		pumpDone := make(chan struct{})