445049f fix(client): make terminal reader claims transactional · vev

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


@@ -610,6 +610,17 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
610	610		defer input.stop()
611	611		}
612	612		inputConsumer := input.claim()
	613	+	claimActive := true
	614	+	revokeInputClaim := func() {
	615	+	if claimActive {
	616	+	input.revoke(inputConsumer)
	617	+	claimActive = false
	618	+	}
	619	+	}
	620	+	// Register revocation before any palette publication or terminal I/O: both
	621	+	// can fail before the stdin scanner is started, and a reconnect must still
	622	+	// be able to claim this lifecycle-owned reader.
	623	+	defer revokeInputClaim()
613	624
614	625		sendCh := make(chan ports.Frame, sendQueueDepth)
615	626		ackQueue := newCumulativeAckQueue()

@@ -729,7 +740,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
729	740		// Revoke this scanner's dequeue permission before waiting for it to exit.
730	741		// The lifecycle-owned reader retains queued bytes for the next attempt.
731	742		defer func() {
732		-	input.revoke(inputConsumer)
	743	+	// Revoke before scanner teardown so replacement attempts can claim the
	744	+	// lifecycle-owned reader as soon as this attempt releases it.
	745	+	revokeInputClaim()
733	746		cancel()
734	747		<-stdinDone
735	748		}()

@@ -950,15 +963,16 @@ type terminalInputPump struct {
950	963		in io.Reader
951	964		done chan struct{}
952	965
953		-	mu sync.Mutex
954		-	pending *terminalReadResult
955		-	consumer uint64
956		-	nextID uint64
957		-	closed bool
958		-	ready chan struct{}
959		-	space chan struct{}
960		-	startMu sync.Once
961		-	stopMu sync.Once
	966	+	mu sync.Mutex
	967	+	pending *terminalReadResult
	968	+	consumer uint64
	969	+	delivering uint64
	970	+	nextID uint64
	971	+	closed bool
	972	+	ready chan struct{}
	973	+	space chan struct{}
	974	+	startMu sync.Once
	975	+	stopMu sync.Once
962	976		}
963	977
964	978		func newTerminalInputPump(in io.Reader) *terminalInputPump {

@@ -992,6 +1006,11 @@ func (p *terminalInputPump) revoke(consumer uint64) {
992	1006		if p.consumer == consumer {
993	1007		p.consumer = 0
994	1008		}
	1009	+	// An unacknowledged read was never delivered by this scanner. Leave it
	1010	+	// pending and make it available to its replacement.
	1011	+	if p.delivering == consumer {
	1012	+	p.delivering = 0
	1013	+	}
995	1014		p.mu.Unlock()
996	1015		p.signalReady()
997	1016		}

@@ -1023,13 +1042,13 @@ func (p *terminalInputPump) enqueue(result terminalReadResult) bool {
1023	1042		return true
1024	1043		}
1025	1044
1026		-	// take gives a ready result only to the current, non-cancelled consumer. It
1027		-	// checks cancellation after readiness is observed, so cancellation wins a
1028		-	// simultaneous cancellation/read race and leaves the result queued.
	1045	+	// take leases a ready result only to the current, non-cancelled consumer.
	1046	+	// The raw read remains pending until ack, so revocation or cancellation before
	1047	+	// scanner delivery leaves the exact bytes available to the next attempt.
1029	1048		func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
1030	1049		p.mu.Lock()
1031		-	if ctx.Err() != nil \|\| p.consumer != consumer \|\| p.pending == nil {
1032		-	pending := p.pending != nil
	1050	+	if ctx.Err() != nil \|\| p.consumer != consumer \|\| p.pending == nil \|\| p.delivering != 0 {
	1051	+	pending := p.pending != nil && p.delivering == 0
1033	1052		p.mu.Unlock()
1034	1053		if pending {
1035	1054		p.signalReady()

@@ -1037,10 +1056,23 @@ func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminal
1037	1056		return terminalReadResult{}, false
1038	1057		}
1039	1058		result := *p.pending
	1059	+	p.delivering = consumer
	1060	+	p.mu.Unlock()
	1061	+	return result, true
	1062	+	}
	1063	+
	1064	+	// ack commits scanner delivery of a leased read and lets the lifecycle reader
	1065	+	// accept the next result. Only the consumer that holds the lease can ack it.
	1066	+	func (p *terminalInputPump) ack(consumer uint64) {
	1067	+	p.mu.Lock()
	1068	+	if p.consumer != consumer \|\| p.delivering != consumer {
	1069	+	p.mu.Unlock()
	1070	+	return
	1071	+	}
1040	1072		p.pending = nil
	1073	+	p.delivering = 0
1041	1074		p.mu.Unlock()
1042	1075		p.space <- struct{}{}
1043		-	return result, true
1044	1076		}
1045	1077
1046	1078		func (p *terminalInputPump) finish() {

@@ -1093,6 +1125,7 @@ type stdinPump struct {
1093	1125		paletteEvents chan<- paletteGenerationEvent
1094	1126		activeGeneration *atomic.Uint64
1095	1127		afterPaletteEvent func(paletteGenerationEvent) // test synchronization hook
	1128	+	afterInputTake func() // test synchronization hook
1096	1129		}
1097	1130
1098	1131		func (p *stdinPump) run() {

@@ -1205,6 +1238,14 @@ func (p *stdinPump) run() {
1205	1238		}
1206	1239		continue
1207	1240		}
	1241	+	if p.afterInputTake != nil {
	1242	+	p.afterInputTake()
	1243	+	}
	1244	+	// The lease is not acknowledged until scanner delivery completes.
	1245	+	// This second check closes the cancellation window after take.
	1246	+	if p.ctx.Err() != nil {
	1247	+	return
	1248	+	}
1208	1249		disarmMarkerDeadline()
1209	1250		// All scanner callbacks from this Read must retain this one generation.
1210	1251		// A scheme notification can start its replacement before a later byte in

@@ -1244,6 +1285,10 @@ func (p *stdinPump) run() {
1244	1285		p.cancel()
1245	1286		return
1246	1287		}
	1288	+	if !sendOK.Load() \|\| p.ctx.Err() != nil {
	1289	+	return
	1290	+	}
	1291	+	input.ack(consumer)
1247	1292		}
1248	1293		}
1249	1294		}

 		defer input.stop()
 	}
 	inputConsumer := input.claim()
+	claimActive := true
+	revokeInputClaim := func() {
+		if claimActive {
+			input.revoke(inputConsumer)
+			claimActive = false
+		}
+	}
+	// Register revocation before any palette publication or terminal I/O: both
+	// can fail before the stdin scanner is started, and a reconnect must still
+	// be able to claim this lifecycle-owned reader.
+	defer revokeInputClaim()
 	sendCh := make(chan ports.Frame, sendQueueDepth)
 	ackQueue := newCumulativeAckQueue()
 	// Revoke this scanner's dequeue permission before waiting for it to exit.
 	// The lifecycle-owned reader retains queued bytes for the next attempt.
 	defer func() {
-		input.revoke(inputConsumer)
+		// Revoke before scanner teardown so replacement attempts can claim the
+		// lifecycle-owned reader as soon as this attempt releases it.
+		revokeInputClaim()
 		cancel()
 		<-stdinDone
 	}()
 	in   io.Reader
 	done chan struct{}
-	mu       sync.Mutex
+	mu         sync.Mutex
-	pending  *terminalReadResult
+	pending    *terminalReadResult
-	consumer uint64
+	consumer   uint64
-	nextID   uint64
+	delivering uint64
-	closed   bool
+	nextID     uint64
-	ready    chan struct{}
+	closed     bool
-	space    chan struct{}
+	ready      chan struct{}
-	startMu  sync.Once
+	space      chan struct{}
-	stopMu   sync.Once
+	startMu    sync.Once
+	stopMu     sync.Once
 }
 func newTerminalInputPump(in io.Reader) *terminalInputPump {
 	if p.consumer == consumer {
 		p.consumer = 0
 	}
+	// An unacknowledged read was never delivered by this scanner. Leave it
+	// pending and make it available to its replacement.
+	if p.delivering == consumer {
+		p.delivering = 0
+	}
 	p.mu.Unlock()
 	p.signalReady()
 }
 	return true
 }
-// take gives a ready result only to the current, non-cancelled consumer. It
+// take leases a ready result only to the current, non-cancelled consumer.
-// checks cancellation after readiness is observed, so cancellation wins a
+// The raw read remains pending until ack, so revocation or cancellation before
-// simultaneous cancellation/read race and leaves the result queued.
+// scanner delivery leaves the exact bytes available to the next attempt.
 func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
 	p.mu.Lock()
-	if ctx.Err() != nil || p.consumer != consumer || p.pending == nil {
+	if ctx.Err() != nil || p.consumer != consumer || p.pending == nil || p.delivering != 0 {
-		pending := p.pending != nil
+		pending := p.pending != nil && p.delivering == 0
 		p.mu.Unlock()
 		if pending {
 			p.signalReady()
 		return terminalReadResult{}, false
 	}
 	result := *p.pending
+	p.delivering = consumer
+	p.mu.Unlock()
+	return result, true
+}
+// ack commits scanner delivery of a leased read and lets the lifecycle reader
+// accept the next result. Only the consumer that holds the lease can ack it.
+func (p *terminalInputPump) ack(consumer uint64) {
+	p.mu.Lock()
+	if p.consumer != consumer || p.delivering != consumer {
+		p.mu.Unlock()
+		return
+	}
 	p.pending = nil
+	p.delivering = 0
 	p.mu.Unlock()
 	p.space <- struct{}{}
-	return result, true
 }
 func (p *terminalInputPump) finish() {
 	paletteEvents     chan<- paletteGenerationEvent
 	activeGeneration  *atomic.Uint64
 	afterPaletteEvent func(paletteGenerationEvent) // test synchronization hook
+	afterInputTake    func()                       // test synchronization hook
 }
 func (p *stdinPump) run() {
 				}
 				continue
 			}
+			if p.afterInputTake != nil {
+				p.afterInputTake()
+			}
+			// The lease is not acknowledged until scanner delivery completes.
+			// This second check closes the cancellation window after take.
+			if p.ctx.Err() != nil {
+				return
+			}
 			disarmMarkerDeadline()
 			// All scanner callbacks from this Read must retain this one generation.
 			// A scheme notification can start its replacement before a later byte in
 				p.cancel()
 				return
 			}
+			if !sendOK.Load() || p.ctx.Err() != nil {
+				return
+			}
+			input.ack(consumer)
 		}
 	}
 }

M internal/usecase/client/palette_attach_generation_test.go +54 -9


@@ -61,6 +61,7 @@ type attachPaletteWriter struct {
61	61		mu sync.Mutex
62	62		writes []string
63	63		wrote chan string
	64	+	err error
64	65		}
65	66
66	67		func newAttachPaletteWriter() *attachPaletteWriter {

@@ -70,7 +71,11 @@ func (w *attachPaletteWriter) Write(data []byte) (int, error) {
70	71		copyData := string(data)
71	72		w.mu.Lock()
72	73		w.writes = append(w.writes, copyData)
	74	+	err := w.err
73	75		w.mu.Unlock()
	76	+	if err != nil {
	77	+	return 0, err
	78	+	}
74	79		w.wrote <- copyData
75	80		return len(data), nil
76	81		}

@@ -167,9 +172,18 @@ func startAttachPaletteHarness(state terminalThemeState) attachPaletteHarness
167	172		}
168	173
169	174		func startAttachPaletteHarnessWithTransport(state terminalThemeState, transport attachPaletteTransport) *attachPaletteHarness {
	175	+	return startAttachPaletteHarnessWithInput(state, transport, nil)
	176	+	}
	177	+
	178	+	func startAttachPaletteHarnessWithInput(state terminalThemeState, transport attachPaletteTransport, input terminalInputPump) attachPaletteHarness {
	179	+	return startAttachPaletteHarnessWithInputAndWriteError(state, transport, input, nil)
	180	+	}
	181	+
	182	+	func startAttachPaletteHarnessWithInputAndWriteError(state terminalThemeState, transport attachPaletteTransport, input terminalInputPump, writeErr error) attachPaletteHarness {
170	183		clock := newAttachPaletteClock()
171	184		reader := newAttachPaletteReader()
172	185		writer := newAttachPaletteWriter()
	186	+	writer.err = writeErr
173	187		term := &attachPaletteTerminalHarness{in: reader, out: writer, resize: make(chan domain.Size)}
174	188		ms := &milestones{}
175	189		runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)}

@@ -178,6 +192,9 @@ func startAttachPaletteHarnessWithTransport(state *terminalThemeState, transport
178	192		milestones: ms, themeState: state,
179	193		enterRaw: func() error { ms.rawEntered = true; return nil },
180	194		reconnect: &reconnectUI{term: term, rawEntered: new(bool)},
	195	+	terminalInput: func() *terminalInputPump {
	196	+	return input
	197	+	},
181	198		}
182	199		h := &attachPaletteHarness{clock: clock, reader: reader, writer: writer, transport: transport, done: make(chan attachResult, 1)}
183	200		go func() { h.done <- attempt.run(context.Background()) }()

@@ -273,16 +290,44 @@ func TestAttachSchemeReplacementAndDeadlinesProgressWhileReadBlocks(t *testing.T
273	290		h.detach(t)
274	291		}
275	292
276		-	func TestAttachInitialClearedThemeSendReturnsTransportError(t *testing.T) {
	293	+	func TestAttachInitialPaletteFailureRevokesInputClaim(t *testing.T) {
277	294		t.Setenv("COLORTERM", "truecolor")
278		-	want := errors.New("initial theme send failed")
279		-	transport := newAttachPaletteTransport()
280		-	transport.themeSendErr = want
281		-	h := startAttachPaletteHarnessWithTransport(&terminalThemeState{}, transport)
282		-
283		-	err := (<-h.done).err
284		-	require.ErrorIs(t, err, want)
285		-	require.ErrorContains(t, err, "publishing theme")
	295	+	for _, tc := range []struct {
	296	+	name string
	297	+	start func(terminalInputPump) attachPaletteHarness
	298	+	contains string
	299	+	}{
	300	+	{
	301	+	name: "publish",
	302	+	start: func(input terminalInputPump) attachPaletteHarness {
	303	+	want := errors.New("initial theme send failed")
	304	+	transport := newAttachPaletteTransport()
	305	+	transport.themeSendErr = want
	306	+	return startAttachPaletteHarnessWithInput(&terminalThemeState{}, transport, input)
	307	+	},
	308	+	contains: "publishing theme",
	309	+	},
	310	+	{
	311	+	name: "write",
	312	+	start: func(input terminalInputPump) attachPaletteHarness {
	313	+	return startAttachPaletteHarnessWithInputAndWriteError(&terminalThemeState{}, newAttachPaletteTransport(), input, errors.New("palette write failed"))
	314	+	},
	315	+	contains: "writing palette query",
	316	+	},
	317	+	} {
	318	+	t.Run(tc.name, func(t *testing.T) {
	319	+	input := newTerminalInputPump(nil)
	320	+	failed := tc.start(input)
	321	+	err := (<-failed.done).err
	322	+	require.ErrorContains(t, err, tc.contains)
	323	+
	324	+	// The next attach must be able to claim the lifecycle-owned reader.
	325	+	// Before revocation was registered at claim time this panicked above.
	326	+	reconnected := startAttachPaletteHarnessWithInput(&terminalThemeState{}, newAttachPaletteTransport(), input)
	327	+	require.Equal(t, paletteColorBatch, reconnected.nextWrite(t))
	328	+	reconnected.detach(t)
	329	+	})
	330	+	}
286	331		}
287	332
288	333		func TestStdinPumpSameReadSchemeAndOldMarkerCannotAdvanceReplacement(t *testing.T) {


@@ -61,6 +61,7 @@ type attachPaletteWriter struct {
61		mu sync.Mutex	61		mu sync.Mutex
62		writes []string	62		writes []string
63		wrote chan string	63		wrote chan string
			64	+	err error
64		}	65		}
65			66
66		func newAttachPaletteWriter() *attachPaletteWriter {	67		func newAttachPaletteWriter() *attachPaletteWriter {

@@ -70,7 +71,11 @@ func (w *attachPaletteWriter) Write(data []byte) (int, error) {
70		copyData := string(data)	71		copyData := string(data)
71		w.mu.Lock()	72		w.mu.Lock()
72		w.writes = append(w.writes, copyData)	73		w.writes = append(w.writes, copyData)
			74	+	err := w.err
73		w.mu.Unlock()	75		w.mu.Unlock()
			76	+	if err != nil {
			77	+	return 0, err
			78	+	}
74		w.wrote <- copyData	79		w.wrote <- copyData
75		return len(data), nil	80		return len(data), nil
76		}	81		}

@@ -167,9 +172,18 @@ func startAttachPaletteHarness(state terminalThemeState) attachPaletteHarness
167		}	172		}
168			173
169		func startAttachPaletteHarnessWithTransport(state terminalThemeState, transport attachPaletteTransport) *attachPaletteHarness {	174		func startAttachPaletteHarnessWithTransport(state terminalThemeState, transport attachPaletteTransport) *attachPaletteHarness {
			175	+	return startAttachPaletteHarnessWithInput(state, transport, nil)
			176	+	}
			177	+
			178	+	func startAttachPaletteHarnessWithInput(state terminalThemeState, transport attachPaletteTransport, input terminalInputPump) attachPaletteHarness {
			179	+	return startAttachPaletteHarnessWithInputAndWriteError(state, transport, input, nil)
			180	+	}
			181	+
			182	+	func startAttachPaletteHarnessWithInputAndWriteError(state terminalThemeState, transport attachPaletteTransport, input terminalInputPump, writeErr error) attachPaletteHarness {
170		clock := newAttachPaletteClock()	183		clock := newAttachPaletteClock()
171		reader := newAttachPaletteReader()	184		reader := newAttachPaletteReader()
172		writer := newAttachPaletteWriter()	185		writer := newAttachPaletteWriter()
			186	+	writer.err = writeErr
173		term := &attachPaletteTerminalHarness{in: reader, out: writer, resize: make(chan domain.Size)}	187		term := &attachPaletteTerminalHarness{in: reader, out: writer, resize: make(chan domain.Size)}
174		ms := &milestones{}	188		ms := &milestones{}
175		runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)}	189		runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)}

@@ -178,6 +192,9 @@ func startAttachPaletteHarnessWithTransport(state *terminalThemeState, transport
178		milestones: ms, themeState: state,	192		milestones: ms, themeState: state,
179		enterRaw: func() error { ms.rawEntered = true; return nil },	193		enterRaw: func() error { ms.rawEntered = true; return nil },
180		reconnect: &reconnectUI{term: term, rawEntered: new(bool)},	194		reconnect: &reconnectUI{term: term, rawEntered: new(bool)},
			195	+	terminalInput: func() *terminalInputPump {
			196	+	return input
			197	+	},
181		}	198		}
182		h := &attachPaletteHarness{clock: clock, reader: reader, writer: writer, transport: transport, done: make(chan attachResult, 1)}	199		h := &attachPaletteHarness{clock: clock, reader: reader, writer: writer, transport: transport, done: make(chan attachResult, 1)}
183		go func() { h.done <- attempt.run(context.Background()) }()	200		go func() { h.done <- attempt.run(context.Background()) }()

@@ -273,16 +290,44 @@ func TestAttachSchemeReplacementAndDeadlinesProgressWhileReadBlocks(t *testing.T
273		h.detach(t)	290		h.detach(t)
274		}	291		}
275			292
276	-	func TestAttachInitialClearedThemeSendReturnsTransportError(t *testing.T) {	293	+	func TestAttachInitialPaletteFailureRevokesInputClaim(t *testing.T) {
277		t.Setenv("COLORTERM", "truecolor")	294		t.Setenv("COLORTERM", "truecolor")
278	-	want := errors.New("initial theme send failed")	295	+	for _, tc := range []struct {
279	-	transport := newAttachPaletteTransport()	296	+	name string
280	-	transport.themeSendErr = want	297	+	start func(terminalInputPump) attachPaletteHarness
281	-	h := startAttachPaletteHarnessWithTransport(&terminalThemeState{}, transport)	298	+	contains string
282	-		299	+	}{
283	-	err := (<-h.done).err	300	+	{
284	-	require.ErrorIs(t, err, want)	301	+	name: "publish",
285	-	require.ErrorContains(t, err, "publishing theme")	302	+	start: func(input terminalInputPump) attachPaletteHarness {
			303	+	want := errors.New("initial theme send failed")
			304	+	transport := newAttachPaletteTransport()
			305	+	transport.themeSendErr = want
			306	+	return startAttachPaletteHarnessWithInput(&terminalThemeState{}, transport, input)
			307	+	},
			308	+	contains: "publishing theme",
			309	+	},
			310	+	{
			311	+	name: "write",
			312	+	start: func(input terminalInputPump) attachPaletteHarness {
			313	+	return startAttachPaletteHarnessWithInputAndWriteError(&terminalThemeState{}, newAttachPaletteTransport(), input, errors.New("palette write failed"))
			314	+	},
			315	+	contains: "writing palette query",
			316	+	},
			317	+	} {
			318	+	t.Run(tc.name, func(t *testing.T) {
			319	+	input := newTerminalInputPump(nil)
			320	+	failed := tc.start(input)
			321	+	err := (<-failed.done).err
			322	+	require.ErrorContains(t, err, tc.contains)
			323	+
			324	+	// The next attach must be able to claim the lifecycle-owned reader.
			325	+	// Before revocation was registered at claim time this panicked above.
			326	+	reconnected := startAttachPaletteHarnessWithInput(&terminalThemeState{}, newAttachPaletteTransport(), input)
			327	+	require.Equal(t, paletteColorBatch, reconnected.nextWrite(t))
			328	+	reconnected.detach(t)
			329	+	})
			330	+	}
286		}	331		}
287			332
288		func TestStdinPumpSameReadSchemeAndOldMarkerCannotAdvanceReplacement(t *testing.T) {	333		func TestStdinPumpSameReadSchemeAndOldMarkerCannotAdvanceReplacement(t *testing.T) {

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


@@ -99,6 +99,43 @@ func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T)
99	99		<-replacementDone
100	100		}
101	101
	102	+	func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
	103	+	input := newTerminalInputPump(nil)
	104	+	want := []byte("first\x00second\xff")
	105	+	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})
	106	+
	107	+	ctx, cancel := context.WithCancel(context.Background())
	108	+	consumer := input.claim()
	109	+	taken := make(chan struct{})
	110	+	release := make(chan struct{})
	111	+	done := make(chan struct{})
	112	+	go func() {
	113	+	(&stdinPump{
	114	+	ctx: ctx, cancel: cancel, input: input, consumer: consumer,
	115	+	out: make(chan ports.Frame, 1), clock: newAttachPaletteClock(),
	116	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
	117	+	afterInputTake: func() {
	118	+	close(taken)
	119	+	<-release
	120	+	},
	121	+	}).run()
	122	+	close(done)
	123	+	}()
	124	+
	125	+	<-taken
	126	+	cancel()
	127	+	close(release)
	128	+	<-done
	129	+	input.revoke(consumer)
	130	+
	131	+	replacement := input.claim()
	132	+	got, ok := input.take(context.Background(), replacement)
	133	+	require.True(t, ok, "cancelled delivery must leave the read for the replacement scanner")
	134	+	require.Equal(t, want, got.data)
	135	+	input.ack(replacement)
	136	+	input.revoke(replacement)
	137	+	}
	138	+
102	139		func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
103	140		reader := newLifecycleReader()
104	141		input := newTerminalInputPump(reader)


@@ -99,6 +99,43 @@ func TestTerminalInputPumpCancellationHandoffPreservesQueuedBytes(t *testing.T)
99	<-replacementDone	99		<-replacementDone
100	}	100		}
101		101
		102	+	func TestTerminalInputPumpCancellationAfterTakeRequeuesRawRead(t *testing.T) {
		103	+	input := newTerminalInputPump(nil)
		104	+	want := []byte("first\x00second\xff")
		105	+	input.enqueue(terminalReadResult{data: append([]byte(nil), want...)})
		106	+
		107	+	ctx, cancel := context.WithCancel(context.Background())
		108	+	consumer := input.claim()
		109	+	taken := make(chan struct{})
		110	+	release := make(chan struct{})
		111	+	done := make(chan struct{})
		112	+	go func() {
		113	+	(&stdinPump{
		114	+	ctx: ctx, cancel: cancel, input: input, consumer: consumer,
		115	+	out: make(chan ports.Frame, 1), clock: newAttachPaletteClock(),
		116	+	logger: slog.New(slog.DiscardHandler), paletteEvents: make(chan paletteGenerationEvent),
		117	+	afterInputTake: func() {
		118	+	close(taken)
		119	+	<-release
		120	+	},
		121	+	}).run()
		122	+	close(done)
		123	+	}()
		124	+
		125	+	<-taken
		126	+	cancel()
		127	+	close(release)
		128	+	<-done
		129	+	input.revoke(consumer)
		130	+
		131	+	replacement := input.claim()
		132	+	got, ok := input.take(context.Background(), replacement)
		133	+	require.True(t, ok, "cancelled delivery must leave the read for the replacement scanner")
		134	+	require.Equal(t, want, got.data)
		135	+	input.ack(replacement)
		136	+	input.revoke(replacement)
		137	+	}
		138	+
102	func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {	139		func TestTerminalInputPumpReusesOneReaderAcrossCancelledAttempts(t *testing.T) {
103	reader := newLifecycleReader()	140		reader := newLifecycleReader()
104	input := newTerminalInputPump(reader)	141		input := newTerminalInputPump(reader)