fix(client): share terminal reader across reconnects

commit: cdca174ec4c76e59eb9251172f0aa26b8243c607
author: brice · 3mo ago · 2026-07-18 07:17 UTC
parent: 6e4ea87
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +219 -99

M internal/usecase/client/client.go +121 -61


@@ -77,20 +77,6 @@ func (s *terminalThemeState) snapshot() ports.Theme {
77	77		return s.theme
78	78		}
79	79
80		-	// clearPalette invalidates entries from a previous terminal color query while
81		-	// retaining independently reported foreground, background, and scheme data.
82		-	func (s *terminalThemeState) clearPalette() ports.Theme {
83		-	return s.update(func(theme *ports.Theme) {
84		-	theme.PaletteKnown = 0
85		-	theme.Palette = [16]renderer.RGB{}
86		-	})
87		-	}
88		-
89		-	func (s *terminalThemeState) reportedTheme() (ports.Theme, bool) {
90		-	theme := s.snapshot()
91		-	return theme, theme.HasForeground \|\| theme.HasBackground \|\| theme.SchemeKnown \|\| theme.PaletteKnown != 0
92		-	}
93		-
94	80		// milestones tracks lifecycle progress so a failure can report which stages
95	81		// were never reached — the key diagnostic when an attach dies early.
96	82		type milestones struct {

@@ -244,6 +230,23 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
244	230		rawEntered: &rawEntered,
245	231		stage: reconnectStageOfflineRetrying,
246	232		}
	233	+	var input *terminalInputPump
	234	+	defer func() {
	235	+	if input != nil {
	236	+	input.stop()
	237	+	}
	238	+	}()
	239	+	originalEnterRaw := enterRaw
	240	+	enterRaw = func() error {
	241	+	if err := originalEnterRaw(); err != nil {
	242	+	return err
	243	+	}
	244	+	if input == nil {
	245	+	input = newTerminalInputPump(r.term.In())
	246	+	input.start()
	247	+	}
	248	+	return nil
	249	+	}
247	250
248	251		for {
249	252		transport, err := r.dialer.Dial(ctx)

@@ -279,6 +282,9 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
279	282		enterRaw: enterRaw,
280	283		reconnect: reconnect,
281	284		linkEvents: linkEvents,
	285	+	terminalInput: func() *terminalInputPump {
	286	+	return input
	287	+	},
282	288		}).run(ctx)
283	289		if result.welcomed {
284	290		backoff = defaultReconnectBackoff.initial

@@ -418,16 +424,17 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time
418	424		}
419	425
420	426		type attachAttempt struct {
421		-	runner *Runner
422		-	transport ports.Transport
423		-	request AttachRequest
424		-	resumeToken uint64
425		-	clientID [16]byte
426		-	milestones *milestones
427		-	themeState *terminalThemeState
428		-	enterRaw func() error
429		-	reconnect *reconnectUI
430		-	linkEvents <-chan ports.LinkEvent
	427	+	runner *Runner
	428	+	transport ports.Transport
	429	+	request AttachRequest
	430	+	resumeToken uint64
	431	+	clientID [16]byte
	432	+	milestones *milestones
	433	+	themeState *terminalThemeState
	434	+	enterRaw func() error
	435	+	reconnect *reconnectUI
	436	+	linkEvents <-chan ports.LinkEvent
	437	+	terminalInput func() *terminalInputPump
431	438		}
432	439
433	440		type attachResult struct {

@@ -592,6 +599,16 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
592	599		// coordinator owner.
593	600		loopCtx, cancel := context.WithCancel(ctx)
594	601		defer cancel()
	602	+	input := (*terminalInputPump)(nil)
	603	+	if a.terminalInput != nil {
	604	+	input = a.terminalInput()
	605	+	}
	606	+	ownedInput := input == nil
	607	+	if ownedInput {
	608	+	input = newTerminalInputPump(term.In())
	609	+	input.start()
	610	+	defer input.stop()
	611	+	}
595	612
596	613		sendCh := make(chan ports.Frame, sendQueueDepth)
597	614		ackQueue := newCumulativeAckQueue()

@@ -703,7 +720,18 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
703	720		}
704	721		senderStarted = true
705	722		go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)
706		-	go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
	723	+	stdinDone := make(chan struct{})
	724	+	go func() {
	725	+	defer close(stdinDone)
	726	+	(&stdinPump{ctx: loopCtx, cancel: cancel, input: input, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
	727	+	}()
	728	+	// A scanner must finish before the next attach starts. The lifecycle-owned
	729	+	// reader remains available for reconnect, but a cancelled scanner cannot
	730	+	// consume a later read result intended for its replacement.
	731	+	defer func() {
	732	+	cancel()
	733	+	<-stdinDone
	734	+	}()
707	735		go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
708	736
709	737		// 5. Output/main loop: the only goroutine that touches the terminal.

@@ -905,23 +933,71 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
905	933		}
906	934		}
907	935
908		-	// stdinPump pumps terminal input to the daemon as Input frames. A read error
909		-	// or EOF is treated as a detach: it best-effort sends a Detach frame and
910		-	// cancels the loop.
911		-	//
912		-	// Known MVP limitation: a bare io.Reader cannot be unblocked from outside,
913		-	// so on shutdown initiated elsewhere (daemon detach, transport loss) the
914		-	// reader goroutine stays parked in Read until the next byte arrives or the
915		-	// process exits. That is harmless here — Run has already returned and
916		-	// restored the terminal — and matches the standard pattern for stdin pumps;
917		-	// a closable stdin duplicate could lift it later if ever needed.
	936	+	// terminalReadResult is copied by terminalInputPump before the next Read so
	937	+	// terminal input remains owned by one lifecycle reader across reconnects.
	938	+	type terminalReadResult struct {
	939	+	data []byte
	940	+	err error
	941	+	}
	942	+
	943	+	// terminalInputPump is the sole consumer of the caller-owned terminal reader.
	944	+	// It is started once after raw mode is entered and is reused by each attach
	945	+	// scanner. A bare io.Reader cannot be interrupted, so one final Read may stay
	946	+	// blocked after Run exits; stop drops its result and never closes caller-owned
	947	+	// stdin. Reconnects never add another reader.
	948	+	type terminalInputPump struct {
	949	+	in io.Reader
	950	+	reads chan terminalReadResult
	951	+	done chan struct{}
	952	+	startMu sync.Once
	953	+	stopMu sync.Once
	954	+	}
	955	+
	956	+	func newTerminalInputPump(in io.Reader) *terminalInputPump {
	957	+	return &terminalInputPump{
	958	+	in: in,
	959	+	reads: make(chan terminalReadResult, 1),
	960	+	done: make(chan struct{}),
	961	+	}
	962	+	}
	963	+
	964	+	func (p *terminalInputPump) start() {
	965	+	p.startMu.Do(func() {
	966	+	go func() {
	967	+	defer close(p.reads)
	968	+	buf := make([]byte, stdinBufSize)
	969	+	for {
	970	+	n, err := p.in.Read(buf)
	971	+	result := terminalReadResult{err: err}
	972	+	if n > 0 {
	973	+	result.data = append([]byte(nil), buf[:n]...)
	974	+	}
	975	+	select {
	976	+	case p.reads <- result:
	977	+	case <-p.done:
	978	+	return
	979	+	}
	980	+	if err != nil {
	981	+	return
	982	+	}
	983	+	}
	984	+	}()
	985	+	})
	986	+	}
	987	+
	988	+	func (p *terminalInputPump) stop() {
	989	+	p.stopMu.Do(func() { close(p.done) })
	990	+	}
	991	+
918	992		// stdinPump parses terminal color reports into generation-tagged events while
919		-	// preserving all ordinary terminal input byte-for-byte. It never publishes a
920		-	// Theme and never writes terminal output; attachAttempt owns both operations.
	993	+	// preserving all ordinary terminal input byte-for-byte. It never reads the
	994	+	// terminal directly, publishes a Theme, or writes terminal output;
	995	+	// attachAttempt owns those operations.
921	996		type stdinPump struct {
922	997		ctx context.Context
923	998		cancel context.CancelFunc
924		-	in io.Reader
	999	+	in io.Reader // compatibility for isolated scanner tests
	1000	+	input *terminalInputPump
925	1001		out chan<- ports.Frame
926	1002		clock ports.Clock
927	1003		clipboard ports.ClipboardReader

@@ -933,7 +1009,6 @@ type stdinPump struct {
933	1009
934	1010		func (p *stdinPump) run() {
935	1011		defer p.logger.Debug("stdin pump exited")
936		-	buf := make([]byte, stdinBufSize)
937	1012		var scanner theme.Scanner
938	1013		var markers paletteMarkerScanner
939	1014		var inputSeq uint64

@@ -978,29 +1053,14 @@ func (p *stdinPump) run() {
978	1053		}
979	1054		sink = ci.Scan
980	1055		}
981		-	type readResult struct {
982		-	data []byte
983		-	err error
	1056	+	input := p.input
	1057	+	ownedInput := input == nil
	1058	+	if ownedInput {
	1059	+	input = newTerminalInputPump(p.in)
	1060	+	input.start()
	1061	+	defer input.stop()
984	1062		}
985		-	reads := make(chan readResult, 1)
986		-	go func() {
987		-	defer close(reads)
988		-	for {
989		-	n, err := p.in.Read(buf)
990		-	result := readResult{err: err}
991		-	if n > 0 {
992		-	result.data = append([]byte(nil), buf[:n]...)
993		-	}
994		-	select {
995		-	case reads <- result:
996		-	case <-p.ctx.Done():
997		-	return
998		-	}
999		-	if err != nil {
1000		-	return
1001		-	}
1002		-	}
1003		-	}()
	1063	+	reads := input.reads
1004	1064
1005	1065		var markerTimer ports.Timer
1006	1066		var markerTimerC <-chan time.Time


@@ -77,20 +77,6 @@ func (s *terminalThemeState) snapshot() ports.Theme {
77		return s.theme	77		return s.theme
78		}	78		}
79			79
80	-	// clearPalette invalidates entries from a previous terminal color query while
81	-	// retaining independently reported foreground, background, and scheme data.
82	-	func (s *terminalThemeState) clearPalette() ports.Theme {
83	-	return s.update(func(theme *ports.Theme) {
84	-	theme.PaletteKnown = 0
85	-	theme.Palette = [16]renderer.RGB{}
86	-	})
87	-	}
88	-
89	-	func (s *terminalThemeState) reportedTheme() (ports.Theme, bool) {
90	-	theme := s.snapshot()
91	-	return theme, theme.HasForeground \|\| theme.HasBackground \|\| theme.SchemeKnown \|\| theme.PaletteKnown != 0
92	-	}
93	-
94		// milestones tracks lifecycle progress so a failure can report which stages	80		// milestones tracks lifecycle progress so a failure can report which stages
95		// were never reached — the key diagnostic when an attach dies early.	81		// were never reached — the key diagnostic when an attach dies early.
96		type milestones struct {	82		type milestones struct {

@@ -244,6 +230,23 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
244		rawEntered: &rawEntered,	230		rawEntered: &rawEntered,
245		stage: reconnectStageOfflineRetrying,	231		stage: reconnectStageOfflineRetrying,
246		}	232		}
			233	+	var input *terminalInputPump
			234	+	defer func() {
			235	+	if input != nil {
			236	+	input.stop()
			237	+	}
			238	+	}()
			239	+	originalEnterRaw := enterRaw
			240	+	enterRaw = func() error {
			241	+	if err := originalEnterRaw(); err != nil {
			242	+	return err
			243	+	}
			244	+	if input == nil {
			245	+	input = newTerminalInputPump(r.term.In())
			246	+	input.start()
			247	+	}
			248	+	return nil
			249	+	}
247			250
248		for {	251		for {
249		transport, err := r.dialer.Dial(ctx)	252		transport, err := r.dialer.Dial(ctx)

@@ -279,6 +282,9 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
279		enterRaw: enterRaw,	282		enterRaw: enterRaw,
280		reconnect: reconnect,	283		reconnect: reconnect,
281		linkEvents: linkEvents,	284		linkEvents: linkEvents,
			285	+	terminalInput: func() *terminalInputPump {
			286	+	return input
			287	+	},
282		}).run(ctx)	288		}).run(ctx)
283		if result.welcomed {	289		if result.welcomed {
284		backoff = defaultReconnectBackoff.initial	290		backoff = defaultReconnectBackoff.initial

@@ -418,16 +424,17 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time
418		}	424		}
419			425
420		type attachAttempt struct {	426		type attachAttempt struct {
421	-	runner *Runner	427	+	runner *Runner
422	-	transport ports.Transport	428	+	transport ports.Transport
423	-	request AttachRequest	429	+	request AttachRequest
424	-	resumeToken uint64	430	+	resumeToken uint64
425	-	clientID [16]byte	431	+	clientID [16]byte
426	-	milestones *milestones	432	+	milestones *milestones
427	-	themeState *terminalThemeState	433	+	themeState *terminalThemeState
428	-	enterRaw func() error	434	+	enterRaw func() error
429	-	reconnect *reconnectUI	435	+	reconnect *reconnectUI
430	-	linkEvents <-chan ports.LinkEvent	436	+	linkEvents <-chan ports.LinkEvent
			437	+	terminalInput func() *terminalInputPump
431		}	438		}
432			439
433		type attachResult struct {	440		type attachResult struct {

@@ -592,6 +599,16 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
592		// coordinator owner.	599		// coordinator owner.
593		loopCtx, cancel := context.WithCancel(ctx)	600		loopCtx, cancel := context.WithCancel(ctx)
594		defer cancel()	601		defer cancel()
			602	+	input := (*terminalInputPump)(nil)
			603	+	if a.terminalInput != nil {
			604	+	input = a.terminalInput()
			605	+	}
			606	+	ownedInput := input == nil
			607	+	if ownedInput {
			608	+	input = newTerminalInputPump(term.In())
			609	+	input.start()
			610	+	defer input.stop()
			611	+	}
595			612
596		sendCh := make(chan ports.Frame, sendQueueDepth)	613		sendCh := make(chan ports.Frame, sendQueueDepth)
597		ackQueue := newCumulativeAckQueue()	614		ackQueue := newCumulativeAckQueue()

@@ -703,7 +720,18 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
703		}	720		}
704		senderStarted = true	721		senderStarted = true
705		go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)	722		go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)
706	-	go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()	723	+	stdinDone := make(chan struct{})
			724	+	go func() {
			725	+	defer close(stdinDone)
			726	+	(&stdinPump{ctx: loopCtx, cancel: cancel, input: input, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
			727	+	}()
			728	+	// A scanner must finish before the next attach starts. The lifecycle-owned
			729	+	// reader remains available for reconnect, but a cancelled scanner cannot
			730	+	// consume a later read result intended for its replacement.
			731	+	defer func() {
			732	+	cancel()
			733	+	<-stdinDone
			734	+	}()
707		go runResize(loopCtx, term.ResizeEvents(), sendCh, log)	735		go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
708			736
709		// 5. Output/main loop: the only goroutine that touches the terminal.	737		// 5. Output/main loop: the only goroutine that touches the terminal.

@@ -905,23 +933,71 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
905		}	933		}
906		}	934		}
907			935
908	-	// stdinPump pumps terminal input to the daemon as Input frames. A read error	936	+	// terminalReadResult is copied by terminalInputPump before the next Read so
909	-	// or EOF is treated as a detach: it best-effort sends a Detach frame and	937	+	// terminal input remains owned by one lifecycle reader across reconnects.
910	-	// cancels the loop.	938	+	type terminalReadResult struct {
911	-	//	939	+	data []byte
912	-	// Known MVP limitation: a bare io.Reader cannot be unblocked from outside,	940	+	err error
913	-	// so on shutdown initiated elsewhere (daemon detach, transport loss) the	941	+	}
914	-	// reader goroutine stays parked in Read until the next byte arrives or the	942	+
915	-	// process exits. That is harmless here — Run has already returned and	943	+	// terminalInputPump is the sole consumer of the caller-owned terminal reader.
916	-	// restored the terminal — and matches the standard pattern for stdin pumps;	944	+	// It is started once after raw mode is entered and is reused by each attach
917	-	// a closable stdin duplicate could lift it later if ever needed.	945	+	// scanner. A bare io.Reader cannot be interrupted, so one final Read may stay
			946	+	// blocked after Run exits; stop drops its result and never closes caller-owned
			947	+	// stdin. Reconnects never add another reader.
			948	+	type terminalInputPump struct {
			949	+	in io.Reader
			950	+	reads chan terminalReadResult
			951	+	done chan struct{}
			952	+	startMu sync.Once
			953	+	stopMu sync.Once
			954	+	}
			955	+
			956	+	func newTerminalInputPump(in io.Reader) *terminalInputPump {
			957	+	return &terminalInputPump{
			958	+	in: in,
			959	+	reads: make(chan terminalReadResult, 1),
			960	+	done: make(chan struct{}),
			961	+	}
			962	+	}
			963	+
			964	+	func (p *terminalInputPump) start() {
			965	+	p.startMu.Do(func() {
			966	+	go func() {
			967	+	defer close(p.reads)
			968	+	buf := make([]byte, stdinBufSize)
			969	+	for {
			970	+	n, err := p.in.Read(buf)
			971	+	result := terminalReadResult{err: err}
			972	+	if n > 0 {
			973	+	result.data = append([]byte(nil), buf[:n]...)
			974	+	}
			975	+	select {
			976	+	case p.reads <- result:
			977	+	case <-p.done:
			978	+	return
			979	+	}
			980	+	if err != nil {
			981	+	return
			982	+	}
			983	+	}
			984	+	}()
			985	+	})
			986	+	}
			987	+
			988	+	func (p *terminalInputPump) stop() {
			989	+	p.stopMu.Do(func() { close(p.done) })
			990	+	}
			991	+
918		// stdinPump parses terminal color reports into generation-tagged events while	992		// stdinPump parses terminal color reports into generation-tagged events while
919	-	// preserving all ordinary terminal input byte-for-byte. It never publishes a	993	+	// preserving all ordinary terminal input byte-for-byte. It never reads the
920	-	// Theme and never writes terminal output; attachAttempt owns both operations.	994	+	// terminal directly, publishes a Theme, or writes terminal output;
			995	+	// attachAttempt owns those operations.
921		type stdinPump struct {	996		type stdinPump struct {
922		ctx context.Context	997		ctx context.Context
923		cancel context.CancelFunc	998		cancel context.CancelFunc
924	-	in io.Reader	999	+	in io.Reader // compatibility for isolated scanner tests
			1000	+	input *terminalInputPump
925		out chan<- ports.Frame	1001		out chan<- ports.Frame
926		clock ports.Clock	1002		clock ports.Clock
927		clipboard ports.ClipboardReader	1003		clipboard ports.ClipboardReader

@@ -933,7 +1009,6 @@ type stdinPump struct {
933			1009
934		func (p *stdinPump) run() {	1010		func (p *stdinPump) run() {
935		defer p.logger.Debug("stdin pump exited")	1011		defer p.logger.Debug("stdin pump exited")
936	-	buf := make([]byte, stdinBufSize)
937		var scanner theme.Scanner	1012		var scanner theme.Scanner
938		var markers paletteMarkerScanner	1013		var markers paletteMarkerScanner
939		var inputSeq uint64	1014		var inputSeq uint64

@@ -978,29 +1053,14 @@ func (p *stdinPump) run() {
978		}	1053		}
979		sink = ci.Scan	1054		sink = ci.Scan
980		}	1055		}
981	-	type readResult struct {	1056	+	input := p.input
982	-	data []byte	1057	+	ownedInput := input == nil
983	-	err error	1058	+	if ownedInput {
			1059	+	input = newTerminalInputPump(p.in)
			1060	+	input.start()
			1061	+	defer input.stop()
984		}	1062		}
985	-	reads := make(chan readResult, 1)	1063	+	reads := input.reads
986	-	go func() {
987	-	defer close(reads)
988	-	for {
989	-	n, err := p.in.Read(buf)
990	-	result := readResult{err: err}
991	-	if n > 0 {
992	-	result.data = append([]byte(nil), buf[:n]...)
993	-	}
994	-	select {
995	-	case reads <- result:
996	-	case <-p.ctx.Done():
997	-	return
998	-	}
999	-	if err != nil {
1000	-	return
1001	-	}
1002	-	}
1003	-	}()
1004			1064
1005		var markerTimer ports.Timer	1065		var markerTimer ports.Timer
1006		var markerTimerC <-chan time.Time	1066		var markerTimerC <-chan time.Time

A internal/usecase/client/terminal_input_pump_test.go +98 -0

@@ -0,0 +1,98 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"context"
	5	+	"io"
	6	+	"log/slog"
	7	+	"sync"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"github.com/bnema/vev/internal/ports"
	14	+	)
	15	+
	16	+	type lifecycleReader struct {
	17	+	chunks chan []byte
	18	+	started chan struct{}
	19	+
	20	+	mu sync.Mutex
	21	+	active int
	22	+	max int
	23	+	}
	24	+
	25	+	func newLifecycleReader() *lifecycleReader {
	26	+	return &lifecycleReader{chunks: make(chan []byte), started: make(chan struct{}, 4)}
	27	+	}
	28	+
	29	+	func (r *lifecycleReader) Read(p []byte) (int, error) {
	30	+	r.mu.Lock()
	31	+	r.active++
	32	+	if r.active > r.max {
	33	+	r.max = r.active
	34	+	}
	35	+	r.mu.Unlock()
	36	+	r.started <- struct{}{}
	37	+	data, ok := <-r.chunks
	38	+	r.mu.Lock()
	39	+	r.active--
	40	+	r.mu.Unlock()
	41	+	if !ok {
	42	+	return 0, io.EOF
	43	+	}
	44	+	return copy(p, data), nil
	45	+	}
	46	+
	47	+	func (r *lifecycleReader) maxActive() int {
	48	+	r.mu.Lock()
	49	+	defer r.mu.Unlock()
	50	+	return r.max
	51	+	}
	52	+
	53	+	func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
	54	+	reader := newLifecycleReader()
	55	+	input := newTerminalInputPump(reader)
	56	+	input.start()
	57	+	t.Cleanup(input.stop)
	58	+	<-reader.started // The lifecycle pump owns the sole blocking Read.
	59	+
	60	+	var activeGeneration atomic.Uint64
	61	+	firstCtx, cancelFirst := context.WithCancel(context.Background())
	62	+	firstDone := make(chan struct{})
	63	+	go func() {
	64	+	(&stdinPump{
	65	+	ctx: firstCtx, cancel: cancelFirst, input: input,
	66	+	out: make(chan ports.Frame), clock: newAttachPaletteClock(),
	67	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	68	+	activeGeneration: &activeGeneration,
	69	+	}).run()
	70	+	close(firstDone)
	71	+	}()
	72	+	cancelFirst()
	73	+	<-firstDone
	74	+
	75	+	secondCtx, cancelSecond := context.WithCancel(context.Background())
	76	+	defer cancelSecond()
	77	+	out := make(chan ports.Frame, 1)
	78	+	secondDone := make(chan struct{})
	79	+	go func() {
	80	+	(&stdinPump{
	81	+	ctx: secondCtx, cancel: cancelSecond, input: input,
	82	+	out: out, clock: newAttachPaletteClock(),
	83	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	84	+	activeGeneration: &activeGeneration,
	85	+	}).run()
	86	+	close(secondDone)
	87	+	}()
	88	+
	89	+	reader.chunks <- []byte("x")
	90	+	frame := <-out
	91	+	got, err := ports.UnmarshalInput(frame.Payload)
	92	+	require.NoError(t, err)
	93	+	require.Equal(t, []byte("x"), got.Data)
	94	+	require.Equal(t, 1, reader.maxActive(), "cancelled attempts must not leave competing reads")
	95	+
	96	+	cancelSecond()
	97	+	<-secondDone
	98	+	}

@@ -0,0 +1,98 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"context"
	5	+	"io"
	6	+	"log/slog"
	7	+	"sync"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"github.com/bnema/vev/internal/ports"
	14	+	)
	15	+
	16	+	type lifecycleReader struct {
	17	+	chunks chan []byte
	18	+	started chan struct{}
	19	+
	20	+	mu sync.Mutex
	21	+	active int
	22	+	max int
	23	+	}
	24	+
	25	+	func newLifecycleReader() *lifecycleReader {
	26	+	return &lifecycleReader{chunks: make(chan []byte), started: make(chan struct{}, 4)}
	27	+	}
	28	+
	29	+	func (r *lifecycleReader) Read(p []byte) (int, error) {
	30	+	r.mu.Lock()
	31	+	r.active++
	32	+	if r.active > r.max {
	33	+	r.max = r.active
	34	+	}
	35	+	r.mu.Unlock()
	36	+	r.started <- struct{}{}
	37	+	data, ok := <-r.chunks
	38	+	r.mu.Lock()
	39	+	r.active--
	40	+	r.mu.Unlock()
	41	+	if !ok {
	42	+	return 0, io.EOF
	43	+	}
	44	+	return copy(p, data), nil
	45	+	}
	46	+
	47	+	func (r *lifecycleReader) maxActive() int {
	48	+	r.mu.Lock()
	49	+	defer r.mu.Unlock()
	50	+	return r.max
	51	+	}
	52	+
	53	+	func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
	54	+	reader := newLifecycleReader()
	55	+	input := newTerminalInputPump(reader)
	56	+	input.start()
	57	+	t.Cleanup(input.stop)
	58	+	<-reader.started // The lifecycle pump owns the sole blocking Read.
	59	+
	60	+	var activeGeneration atomic.Uint64
	61	+	firstCtx, cancelFirst := context.WithCancel(context.Background())
	62	+	firstDone := make(chan struct{})
	63	+	go func() {
	64	+	(&stdinPump{
	65	+	ctx: firstCtx, cancel: cancelFirst, input: input,
	66	+	out: make(chan ports.Frame), clock: newAttachPaletteClock(),
	67	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	68	+	activeGeneration: &activeGeneration,
	69	+	}).run()
	70	+	close(firstDone)
	71	+	}()
	72	+	cancelFirst()
	73	+	<-firstDone
	74	+
	75	+	secondCtx, cancelSecond := context.WithCancel(context.Background())
	76	+	defer cancelSecond()
	77	+	out := make(chan ports.Frame, 1)
	78	+	secondDone := make(chan struct{})
	79	+	go func() {
	80	+	(&stdinPump{
	81	+	ctx: secondCtx, cancel: cancelSecond, input: input,
	82	+	out: out, clock: newAttachPaletteClock(),
	83	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	84	+	activeGeneration: &activeGeneration,
	85	+	}).run()
	86	+	close(secondDone)
	87	+	}()
	88	+
	89	+	reader.chunks <- []byte("x")
	90	+	frame := <-out
	91	+	got, err := ports.UnmarshalInput(frame.Payload)
	92	+	require.NoError(t, err)
	93	+	require.Equal(t, []byte("x"), got.Data)
	94	+	require.Equal(t, 1, reader.maxActive(), "cancelled attempts must not leave competing reads")
	95	+
	96	+	cancelSecond()
	97	+	<-secondDone
	98	+	}

D internal/usecase/client/theme_state_test.go +0 -38

@@ -1,38 +0,0 @@
1	-	package client
2	-
3	-	import (
4	-	"testing"
5	-
6	-	"github.com/stretchr/testify/require"
7	-
8	-	"github.com/bnema/vev/internal/ports"
9	-	"github.com/bnema/vev/pkg/renderer"
10	-	)
11	-
12	-	func TestTerminalThemeStateClearPalettePreservesReportedColors(t *testing.T) {
13	-	state := &terminalThemeState{}
14	-	state.update(func(theme *ports.Theme) {
15	-	theme.HasForeground = true
16	-	theme.Foreground = renderer.RGB{R: 1, G: 2, B: 3}
17	-	theme.HasBackground = true
18	-	theme.Background = renderer.RGB{R: 4, G: 5, B: 6}
19	-	theme.SchemeKnown = true
20	-	theme.Light = true
21	-	theme.PaletteKnown = 1<<1 \| 1<<14
22	-	theme.Palette[1] = renderer.RGB{R: 7, G: 8, B: 9}
23	-	theme.Palette[14] = renderer.RGB{R: 10, G: 11, B: 12}
24	-	})
25	-
26	-	got := state.clearPalette()
27	-	require.True(t, got.HasForeground)
28	-	require.Equal(t, renderer.RGB{R: 1, G: 2, B: 3}, got.Foreground)
29	-	require.True(t, got.HasBackground)
30	-	require.Equal(t, renderer.RGB{R: 4, G: 5, B: 6}, got.Background)
31	-	require.True(t, got.SchemeKnown)
32	-	require.True(t, got.Light)
33	-	require.Zero(t, got.PaletteKnown)
34	-	require.Equal(t, [16]renderer.RGB{}, got.Palette)
35	-
36	-	_, reported := state.reportedTheme()
37	-	require.True(t, reported, "foreground/background remain usable after palette invalidation")
38	-	}

@@ -1,38 +0,0 @@
1	-	package client
2	-
3	-	import (
4	-	"testing"
5	-
6	-	"github.com/stretchr/testify/require"
7	-
8	-	"github.com/bnema/vev/internal/ports"
9	-	"github.com/bnema/vev/pkg/renderer"
10	-	)
11	-
12	-	func TestTerminalThemeStateClearPalettePreservesReportedColors(t *testing.T) {
13	-	state := &terminalThemeState{}
14	-	state.update(func(theme *ports.Theme) {
15	-	theme.HasForeground = true
16	-	theme.Foreground = renderer.RGB{R: 1, G: 2, B: 3}
17	-	theme.HasBackground = true
18	-	theme.Background = renderer.RGB{R: 4, G: 5, B: 6}
19	-	theme.SchemeKnown = true
20	-	theme.Light = true
21	-	theme.PaletteKnown = 1<<1 \| 1<<14
22	-	theme.Palette[1] = renderer.RGB{R: 7, G: 8, B: 9}
23	-	theme.Palette[14] = renderer.RGB{R: 10, G: 11, B: 12}
24	-	})
25	-
26	-	got := state.clearPalette()
27	-	require.True(t, got.HasForeground)
28	-	require.Equal(t, renderer.RGB{R: 1, G: 2, B: 3}, got.Foreground)
29	-	require.True(t, got.HasBackground)
30	-	require.Equal(t, renderer.RGB{R: 4, G: 5, B: 6}, got.Background)
31	-	require.True(t, got.SchemeKnown)
32	-	require.True(t, got.Light)
33	-	require.Zero(t, got.PaletteKnown)
34	-	require.Equal(t, [16]renderer.RGB{}, got.Palette)
35	-
36	-	_, reported := state.reportedTheme()
37	-	require.True(t, reported, "foreground/background remain usable after palette invalidation")
38	-	}