Merge pull request #13 from bnema/fix/bracketed-paste-marker-splits

fix(client,keys): keep bracketed paste markers intact across input frames

commit: 71edbd43255fdca35592d03d18b757dbef9512ef
author: bnema · 3mo ago · 2026-07-04 16:39 UTC
committer: GitHub · 3mo ago · 2026-07-04 16:39 UTC
merge of: 8f91483 + 5d67c2f · diff against the first parent
signature: GPG UnverifiedB5690EEEBB952194 The key is unknown

9 files · +699 -40

M internal/app/run.go +3 -3


@@ -299,7 +299,7 @@ type runAttachDeps struct {
299	299		attachLocal func(context.Context, uint8, string, *slog.Logger) error // compatibility hook for focused tests
300	300		localDialer func() ports.Dialer
301	301		remoteDialer func(target, session string, log *slog.Logger) ports.Dialer
302		-	runClient func(context.Context, ports.Dialer, ports.Terminal, uint8, string, *slog.Logger) error
	302	+	runClient func(context.Context, ports.Dialer, ports.Terminal, ports.Clock, uint8, string, *slog.Logger) error
303	303		createDetached func(context.Context, string) error
304	304		}
305	305

@@ -323,7 +323,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac
323	323		if log != nil {
324	324		log.Info("attaching to remote session", "target", remoteTarget, "name", name)
325	325		}
326		-	return runClient(ctx, remoteDialer(remoteTarget, name, log), term.New(), intent, name, log)
	326	+	return runClient(ctx, remoteDialer(remoteTarget, name, log), term.New(), clock.New(), intent, name, log)
327	327		}
328	328
329	329		attachLocal := deps.attachLocal

@@ -336,7 +336,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac
336	336		if log != nil {
337	337		log.Info("attaching to local session", "intent", intent, "name", name)
338	338		}
339		-	return runClient(ctx, localDialer(), term.New(), intent, name, log)
	339	+	return runClient(ctx, localDialer(), term.New(), clock.New(), intent, name, log)
340	340		}
341	341		}
342	342		return runLocalAttachWithRecovery(ctx, intent, name, attachRecoveryDeps{


@@ -299,7 +299,7 @@ type runAttachDeps struct {
299		attachLocal func(context.Context, uint8, string, *slog.Logger) error // compatibility hook for focused tests	299		attachLocal func(context.Context, uint8, string, *slog.Logger) error // compatibility hook for focused tests
300		localDialer func() ports.Dialer	300		localDialer func() ports.Dialer
301		remoteDialer func(target, session string, log *slog.Logger) ports.Dialer	301		remoteDialer func(target, session string, log *slog.Logger) ports.Dialer
302	-	runClient func(context.Context, ports.Dialer, ports.Terminal, uint8, string, *slog.Logger) error	302	+	runClient func(context.Context, ports.Dialer, ports.Terminal, ports.Clock, uint8, string, *slog.Logger) error
303		createDetached func(context.Context, string) error	303		createDetached func(context.Context, string) error
304		}	304		}
305			305

@@ -323,7 +323,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac
323		if log != nil {	323		if log != nil {
324		log.Info("attaching to remote session", "target", remoteTarget, "name", name)	324		log.Info("attaching to remote session", "target", remoteTarget, "name", name)
325		}	325		}
326	-	return runClient(ctx, remoteDialer(remoteTarget, name, log), term.New(), intent, name, log)	326	+	return runClient(ctx, remoteDialer(remoteTarget, name, log), term.New(), clock.New(), intent, name, log)
327		}	327		}
328			328
329		attachLocal := deps.attachLocal	329		attachLocal := deps.attachLocal

@@ -336,7 +336,7 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac
336		if log != nil {	336		if log != nil {
337		log.Info("attaching to local session", "intent", intent, "name", name)	337		log.Info("attaching to local session", "intent", intent, "name", name)
338		}	338		}
339	-	return runClient(ctx, localDialer(), term.New(), intent, name, log)	339	+	return runClient(ctx, localDialer(), term.New(), clock.New(), intent, name, log)
340		}	340		}
341		}	341		}
342		return runLocalAttachWithRecovery(ctx, intent, name, attachRecoveryDeps{	342		return runLocalAttachWithRecovery(ctx, intent, name, attachRecoveryDeps{

M internal/app/run_test.go +2 -2


@@ -401,7 +401,7 @@ func TestRunAttachWithDepsBuildsRemoteDialer(t *testing.T) {
401	401		gotTarget, gotSession = target, session
402	402		return namedDialer{name: "remote"}
403	403		},
404		-	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, intent uint8, name string, _ *slog.Logger) error {
	404	+	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, _ ports.Clock, intent uint8, name string, _ *slog.Logger) error {
405	405		gotDialer = d.(namedDialer).name
406	406		if intent != ports.IntentAttach \|\| name != "work" {
407	407		t.Fatalf("intent/name = %d/%q, want attach/work", intent, name)

@@ -421,7 +421,7 @@ func TestRunAttachWithDepsBuildsLocalDialer(t *testing.T) {
421	421		var gotDialer string
422	422		err := runAttachWithDeps(context.Background(), ports.IntentEphemeral, "", "", "", nil, runAttachDeps{
423	423		localDialer: func() ports.Dialer { return namedDialer{name: "local"} },
424		-	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, intent uint8, name string, _ *slog.Logger) error {
	424	+	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, _ ports.Clock, intent uint8, name string, _ *slog.Logger) error {
425	425		gotDialer = d.(namedDialer).name
426	426		if intent != ports.IntentEphemeral \|\| name != "" {
427	427		t.Fatalf("intent/name = %d/%q, want ephemeral/empty", intent, name)


@@ -401,7 +401,7 @@ func TestRunAttachWithDepsBuildsRemoteDialer(t *testing.T) {
401		gotTarget, gotSession = target, session	401		gotTarget, gotSession = target, session
402		return namedDialer{name: "remote"}	402		return namedDialer{name: "remote"}
403		},	403		},
404	-	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, intent uint8, name string, _ *slog.Logger) error {	404	+	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, _ ports.Clock, intent uint8, name string, _ *slog.Logger) error {
405		gotDialer = d.(namedDialer).name	405		gotDialer = d.(namedDialer).name
406		if intent != ports.IntentAttach \|\| name != "work" {	406		if intent != ports.IntentAttach \|\| name != "work" {
407		t.Fatalf("intent/name = %d/%q, want attach/work", intent, name)	407		t.Fatalf("intent/name = %d/%q, want attach/work", intent, name)

@@ -421,7 +421,7 @@ func TestRunAttachWithDepsBuildsLocalDialer(t *testing.T) {
421		var gotDialer string	421		var gotDialer string
422		err := runAttachWithDeps(context.Background(), ports.IntentEphemeral, "", "", "", nil, runAttachDeps{	422		err := runAttachWithDeps(context.Background(), ports.IntentEphemeral, "", "", "", nil, runAttachDeps{
423		localDialer: func() ports.Dialer { return namedDialer{name: "local"} },	423		localDialer: func() ports.Dialer { return namedDialer{name: "local"} },
424	-	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, intent uint8, name string, _ *slog.Logger) error {	424	+	runClient: func(_ context.Context, d ports.Dialer, _ ports.Terminal, _ ports.Clock, intent uint8, name string, _ *slog.Logger) error {
425		gotDialer = d.(namedDialer).name	425		gotDialer = d.(namedDialer).name
426		if intent != ports.IntentEphemeral \|\| name != "" {	426		if intent != ports.IntentEphemeral \|\| name != "" {
427		t.Fatalf("intent/name = %d/%q, want ephemeral/empty", intent, name)	427		t.Fatalf("intent/name = %d/%q, want ephemeral/empty", intent, name)

A internal/ports/paste.go +9 -0

@@ -0,0 +1,9 @@
	1	+	package ports
	2	+
	3	+	// Bracketed-paste markers bracket terminal paste payloads when bracketed-paste
	4	+	// mode is enabled. The client coalescer and daemon key router both depend on
	5	+	// these bytes staying in sync.
	6	+	var (
	7	+	BracketedPasteOpenMarker = []byte("\x1b[200~")
	8	+	BracketedPasteCloseMarker = []byte("\x1b[201~")
	9	+	)

@@ -0,0 +1,9 @@
	1	+	package ports
	2	+
	3	+	// Bracketed-paste markers bracket terminal paste payloads when bracketed-paste
	4	+	// mode is enabled. The client coalescer and daemon key router both depend on
	5	+	// these bytes staying in sync.
	6	+	var (
	7	+	BracketedPasteOpenMarker = []byte("\x1b[200~")
	8	+	BracketedPasteCloseMarker = []byte("\x1b[201~")
	9	+	)

M internal/usecase/client/client.go +27 -21


@@ -16,6 +16,7 @@ import (
16	16		"log/slog"
17	17		"os"
18	18		"strings"
	19	+	"sync/atomic"
19	20		"time"
20	21
21	22		"github.com/bnema/vev/internal/domain"

@@ -118,7 +119,7 @@ const (
118	119		// Run connects via dialer and runs the attach client. It owns the terminal
119	120		// lifecycle above attach attempts so raw mode remains active while a live
120	121		// client process redials a lost link.
121		-	func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, intent uint8, name string, log *slog.Logger) (retErr error) {
	122	+	func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, log *slog.Logger) (retErr error) {
122	123		if log == nil {
123	124		log = slog.Default()
124	125		}

@@ -191,7 +192,7 @@ func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, intent u
191	192		}
192	193		ms.dialed = true
193	194
194		-	result := attachOnce(ctx, transport, term, attemptIntent, name, resumeToken, processClientID, &ms, enterRaw, clearStatus, log)
	195	+	result := attachOnce(ctx, transport, term, clk, attemptIntent, name, resumeToken, processClientID, &ms, enterRaw, clearStatus, log)
195	196		if result.welcomed {
196	197		backoff = defaultReconnectBackoff.initial
197	198		}

@@ -253,8 +254,8 @@ func shouldReconnect(err error) bool {
253	254		// disappears, or the context is cancelled.
254	255		//
255	256		// It is kept as a compatibility wrapper for callers that still own dialing.
256		-	func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, intent uint8, name string) error {
257		-	return Run(ctx, singleTransportDialer{transport: transport}, term, intent, name, slog.Default())
	257	+	func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string) error {
	258	+	return Run(ctx, singleTransportDialer{transport: transport}, term, clk, intent, name, slog.Default())
258	259		}
259	260
260	261		type singleTransportDialer struct{ transport ports.Transport }

@@ -269,7 +270,7 @@ type attachResult struct {
269	270		err error
270	271		}
271	272
272		-	func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms milestones, enterRaw func() error, clearStatus func(), log slog.Logger) attachResult {
	273	+	func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms milestones, enterRaw func() error, clearStatus func(), log slog.Logger) attachResult {
273	274		// 1. Handshake: send Hello with our size and TERM.
274	275		size, err := term.Size()
275	276		if err != nil {

@@ -336,7 +337,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
336	337		sendErrCh := make(chan error, 1)
337	338
338	339		go runSender(loopCtx, cancel, transport, sendCh, sendErrCh, log)
339		-	go runStdin(loopCtx, cancel, term.In(), sendCh, trueColor, log)
	340	+	go runStdin(loopCtx, cancel, term.In(), sendCh, clk, trueColor, log)
340	341		go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
341	342
342	343		// 5. Output/main loop: the only goroutine that touches the terminal.

@@ -460,24 +461,36 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T
460	461		// exits. That is harmless here — Attach has already returned and restored
461	462		// the terminal — and matches the standard pattern for stdin pumps; a
462	463		// closable stdin duplicate could lift it later if ever needed.
463		-	func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame, trueColor bool, log *slog.Logger) {
	464	+	func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame, clk ports.Clock, trueColor bool, log *slog.Logger) {
464	465		defer log.Debug("stdin pump exited")
465	466		buf := make([]byte, stdinBufSize)
466	467		var scanner theme.Scanner
467	468		var inputSeq uint64
468	469		current := ports.Theme{TrueColor: trueColor}
469		-	send := func(frame ports.Frame) bool {
	470	+	var sendOK atomic.Bool
	471	+	sendOK.Store(true)
	472	+	send := func(frame ports.Frame) {
470	473		select {
471	474		case out <- frame:
472		-	return true
473	475		case <-ctx.Done():
474		-	return false
	476	+	sendOK.Store(false)
475	477		}
476	478		}
	479	+	// The coalescer reframes a bracketed paste split across reads into one
	480	+	// MsgInput frame, so a marker boundary can never leave a lone ESC on the
	481	+	// wire. Its emit runs one MsgInput per call, keeping the inputSeq contract.
	482	+	coalescer := newPasteCoalescer(clk, func(data []byte) {
	483	+	if len(data) == 0 {
	484	+	return
	485	+	}
	486	+	copyData := append([]byte(nil), data...)
	487	+	inputSeq++
	488	+	send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: copyData})})
	489	+	})
	490	+	defer coalescer.Close()
477	491		for {
478	492		n, rerr := in.Read(buf)
479	493		if n > 0 {
480		-	ok := true
481	494		scanner.Scan(buf[:n], func(kind int, rgb renderer.RGB) {
482	495		switch kind {
483	496		case 10:

@@ -489,16 +502,9 @@ func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out
489	502		default:
490	503		return
491	504		}
492		-	ok = send(ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(current)})
493		-	}, func(data []byte) {
494		-	if len(data) == 0 \|\| !ok {
495		-	return
496		-	}
497		-	copyData := append([]byte(nil), data...)
498		-	inputSeq++
499		-	ok = send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: copyData})})
500		-	})
501		-	if !ok {
	505	+	send(ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(current)})
	506	+	}, coalescer.Scan)
	507	+	if !sendOK.Load() {
502	508		return
503	509		}
504	510		}

 	"log/slog"
 	"os"
 	"strings"
+	"sync/atomic"
 	"time"
 	"github.com/bnema/vev/internal/domain"
 // Run connects via dialer and runs the attach client. It owns the terminal
 // lifecycle above attach attempts so raw mode remains active while a live
 // client process redials a lost link.
-func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, intent uint8, name string, log *slog.Logger) (retErr error) {
+func Run(ctx context.Context, dialer ports.Dialer, term ports.Terminal, clk ports.Clock, intent uint8, name string, log *slog.Logger) (retErr error) {
 	if log == nil {
 		log = slog.Default()
 	}
 		}
 		ms.dialed = true
-		result := attachOnce(ctx, transport, term, attemptIntent, name, resumeToken, processClientID, &ms, enterRaw, clearStatus, log)
+		result := attachOnce(ctx, transport, term, clk, attemptIntent, name, resumeToken, processClientID, &ms, enterRaw, clearStatus, log)
 		if result.welcomed {
 			backoff = defaultReconnectBackoff.initial
 		}
 // disappears, or the context is cancelled.
 //
 // It is kept as a compatibility wrapper for callers that still own dialing.
-func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, intent uint8, name string) error {
+func Attach(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string) error {
-	return Run(ctx, singleTransportDialer{transport: transport}, term, intent, name, slog.Default())
+	return Run(ctx, singleTransportDialer{transport: transport}, term, clk, intent, name, slog.Default())
 }
 type singleTransportDialer struct{ transport ports.Transport }
 	err         error
 }
-func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, enterRaw func() error, clearStatus func(), log *slog.Logger) attachResult {
+func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, enterRaw func() error, clearStatus func(), log *slog.Logger) attachResult {
 	// 1. Handshake: send Hello with our size and TERM.
 	size, err := term.Size()
 	if err != nil {
 	sendErrCh := make(chan error, 1)
 	go runSender(loopCtx, cancel, transport, sendCh, sendErrCh, log)
-	go runStdin(loopCtx, cancel, term.In(), sendCh, trueColor, log)
+	go runStdin(loopCtx, cancel, term.In(), sendCh, clk, trueColor, log)
 	go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
 	// 5. Output/main loop: the only goroutine that touches the terminal.
 // exits. That is harmless here — Attach 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.
-func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame, trueColor bool, log *slog.Logger) {
+func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame, clk ports.Clock, trueColor bool, log *slog.Logger) {
 	defer log.Debug("stdin pump exited")
 	buf := make([]byte, stdinBufSize)
 	var scanner theme.Scanner
 	var inputSeq uint64
 	current := ports.Theme{TrueColor: trueColor}
-	send := func(frame ports.Frame) bool {
+	var sendOK atomic.Bool
+	sendOK.Store(true)
+	send := func(frame ports.Frame) {
 		select {
 		case out <- frame:
-			return true
 		case <-ctx.Done():
-			return false
+			sendOK.Store(false)
 		}
 	}
+	// The coalescer reframes a bracketed paste split across reads into one
+	// MsgInput frame, so a marker boundary can never leave a lone ESC on the
+	// wire. Its emit runs one MsgInput per call, keeping the inputSeq contract.
+	coalescer := newPasteCoalescer(clk, func(data []byte) {
+		if len(data) == 0 {
+			return
+		}
+		copyData := append([]byte(nil), data...)
+		inputSeq++
+		send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: copyData})})
+	})
+	defer coalescer.Close()
 	for {
 		n, rerr := in.Read(buf)
 		if n > 0 {
-			ok := true
 			scanner.Scan(buf[:n], func(kind int, rgb renderer.RGB) {
 				switch kind {
 				case 10:
 				default:
 					return
 				}
-				ok = send(ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(current)})
+				send(ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(current)})
-			}, func(data []byte) {
+			}, coalescer.Scan)
-				if len(data) == 0 || !ok {
+			if !sendOK.Load() {
-					return
-				}
-				copyData := append([]byte(nil), data...)
-				inputSeq++
-				ok = send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: copyData})})
-			})
-			if !ok {
 				return
 			}
 		}

M internal/usecase/client/client_test.go +114 -14


@@ -25,6 +25,21 @@ func init() {
25	25		slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))
26	26		}
27	27
	28	+	// realClock is a stdlib-backed ports.Clock for tests that exercise the stdin
	29	+	// pump end to end; the paste coalescer's flush timer runs on wall time here.
	30	+	type realClock struct{}
	31	+
	32	+	func (realClock) Now() time.Time { return time.Now() }
	33	+	func (realClock) NewTimer(d time.Duration) ports.Timer {
	34	+	return &realTimer{t: time.NewTimer(d)}
	35	+	}
	36	+
	37	+	type realTimer struct{ t *time.Timer }
	38	+
	39	+	func (r *realTimer) C() <-chan time.Time { return r.t.C }
	40	+	func (r *realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) }
	41	+	func (r *realTimer) Stop() bool { return r.t.Stop() }
	42	+
28	43		// recvItem is one scripted Recv result.
29	44		type recvItem struct {
30	45		f ports.Frame

@@ -92,6 +107,31 @@ func (r *oneShotBlockingReader) Read(p []byte) (int, error) {
92	107
93	108		func (r *oneShotBlockingReader) unblock() { close(r.done) }
94	109
	110	+	type chunkedBlockingReader struct {
	111	+	chunks [][]byte
	112	+	done chan struct{}
	113	+	mu sync.Mutex
	114	+	next int
	115	+	}
	116	+
	117	+	func newChunkedBlockingReader(chunks ...[]byte) *chunkedBlockingReader {
	118	+	return &chunkedBlockingReader{chunks: chunks, done: make(chan struct{})}
	119	+	}
	120	+
	121	+	func (r *chunkedBlockingReader) Read(p []byte) (int, error) {
	122	+	r.mu.Lock()
	123	+	defer r.mu.Unlock()
	124	+	if r.next < len(r.chunks) {
	125	+	chunk := r.chunks[r.next]
	126	+	r.next++
	127	+	return copy(p, chunk), nil
	128	+	}
	129	+	<-r.done
	130	+	return 0, io.EOF
	131	+	}
	132	+
	133	+	func (r *chunkedBlockingReader) unblock() { close(r.done) }
	134	+
95	135		func newHappyTerminal(t testing.T, out bytes.Buffer, restoreCount atomic.Int32, resizeCh chan domain.Size) (portsmocks.MockTerminal, *blockingReader) {
96	136		tm := portsmocks.NewMockTerminal(t)
97	137		tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()

@@ -128,7 +168,7 @@ func TestAttachHappyPath(t *testing.T) {
128	168		defer unblock()
129	169		tr.EXPECT().Close().Return(nil).Once()
130	170
131		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	171	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
132	172		require.NoError(t, err)
133	173		require.Equal(t, "hello world", out.String())
134	174		require.Equal(t, int32(1), restoreCount.Load(), "restore must run exactly once")

@@ -147,7 +187,7 @@ func TestAttachVersionMismatch(t *testing.T) {
147	187		).Once()
148	188		tr.EXPECT().Close().Return(nil).Once()
149	189
150		-	err := client.Attach(context.Background(), tr, tm, ports.IntentAttach, "main")
	190	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentAttach, "main")
151	191		require.Error(t, err)
152	192		var pe *client.ProtocolError
153	193		require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)

@@ -171,7 +211,7 @@ func TestAttachRestoredOnRecvErrorMidStream(t *testing.T) {
171	211		defer unblock()
172	212		tr.EXPECT().Close().Return(nil).Once()
173	213
174		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	214	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
175	215		require.Error(t, err)
176	216		require.Equal(t, int32(1), restoreCount.Load(), "restore must run after mid-stream Recv error")
177	217		}

@@ -192,7 +232,7 @@ func TestAttachDaemonVanishedOnEOF(t *testing.T) {
192	232		defer unblock()
193	233		tr.EXPECT().Close().Return(nil).Once()
194	234
195		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	235	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
196	236		require.Error(t, err)
197	237		require.ErrorIs(t, err, io.EOF)
198	238		require.Equal(t, int32(1), restoreCount.Load())

@@ -264,7 +304,7 @@ func TestAttachStdinOSCColorResponseSendsThemeAndPreservesInput(t *testing.T) {
264	304		}).Maybe()
265	305		tr.EXPECT().Close().Return(nil).Once()
266	306
267		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	307	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
268	308		require.NoError(t, err)
269	309		require.Equal(t, int32(1), restoreCount.Load())
270	310

@@ -342,7 +382,7 @@ func TestAttachStdinThemeTrueColorFalseWhenCOLORTERMNotTruecolor(t *testing.T) {
342	382		}).Maybe()
343	383		tr.EXPECT().Close().Return(nil).Once()
344	384
345		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	385	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
346	386		require.NoError(t, err)
347	387		select {
348	388		case th := <-gotTheme:

@@ -402,7 +442,7 @@ func TestAttachForwardsStandaloneEscapeInput(t *testing.T) {
402	442		}).Maybe()
403	443		tr.EXPECT().Close().Return(nil).Once()
404	444
405		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	445	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
406	446		require.NoError(t, err)
407	447		select {
408	448		case got := <-gotInput:

@@ -461,7 +501,7 @@ func TestAttachStdinForwardsSGRMouseReportAsSingleFrame(t *testing.T) {
461	501		}).Maybe()
462	502		tr.EXPECT().Close().Return(nil).Once()
463	503
464		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	504	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
465	505		require.NoError(t, err)
466	506		select {
467	507		case got := <-gotInput:

@@ -471,6 +511,66 @@ func TestAttachStdinForwardsSGRMouseReportAsSingleFrame(t *testing.T) {
471	511		}
472	512		}
473	513
	514	+	func TestAttachStdinCoalescesSplitBracketedPaste(t *testing.T) {
	515	+	var out bytes.Buffer
	516	+	var restoreCount atomic.Int32
	517	+	resizeCh := make(chan domain.Size)
	518	+	paste := []byte("\x1b[200~hello\nworld\x1b[201~")
	519	+	input := newChunkedBlockingReader([]byte("\x1b[200~hello\n"), []byte("world\x1b[201~"))
	520	+	defer input.unblock()
	521	+
	522	+	tm := portsmocks.NewMockTerminal(t)
	523	+	tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
	524	+	tm.EXPECT().EnterRaw().Return(func() error { restoreCount.Add(1); return nil }, nil).Once()
	525	+	tm.EXPECT().In().Return(input).Maybe()
	526	+	tm.EXPECT().Out().Return(&out).Maybe()
	527	+	tm.EXPECT().Flush().Return(nil).Maybe()
	528	+	tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
	529	+
	530	+	gotInput := make(chan []byte, 1)
	531	+	allowDetach := make(chan struct{})
	532	+	tr := portsmocks.NewMockTransport(t)
	533	+	tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
	534	+	tr.EXPECT().Send(isType(ports.MsgInput)).RunAndReturn(func(f ports.Frame) error {
	535	+	in, err := ports.UnmarshalInput(f.Payload)
	536	+	require.NoError(t, err)
	537	+	gotInput <- in.Data
	538	+	close(allowDetach)
	539	+	return nil
	540	+	}).Once()
	541	+	welcome := frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))
	542	+	detached := frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))
	543	+	recvCh := make(chan recvItem, 1)
	544	+	recvCh <- recvItem{f: welcome}
	545	+	closed := make(chan struct{})
	546	+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
	547	+	select {
	548	+	case it := <-recvCh:
	549	+	return it.f, it.err
	550	+	case <-allowDetach:
	551	+	select {
	552	+	case <-closed:
	553	+	return ports.Frame{}, io.EOF
	554	+	default:
	555	+	close(closed)
	556	+	return detached, nil
	557	+	}
	558	+	case <-closed:
	559	+	return ports.Frame{}, io.EOF
	560	+	}
	561	+	}).Maybe()
	562	+	tr.EXPECT().Close().Return(nil).Once()
	563	+
	564	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
	565	+	require.NoError(t, err)
	566	+	select {
	567	+	case got := <-gotInput:
	568	+	require.Equal(t, paste, got, "split bracketed paste must arrive as one intact MsgInput frame")
	569	+	case <-time.After(2 * time.Second):
	570	+	t.Fatal("split bracketed paste input was not sent")
	571	+	}
	572	+	}
	573	+
474	574		func TestAttachForwardsResize(t *testing.T) {
475	575		var out bytes.Buffer
476	576		var restoreCount atomic.Int32

@@ -521,7 +621,7 @@ func TestAttachForwardsResize(t *testing.T) {
521	621		resizeCh <- domain.Size{Cols: 120, Rows: 40}
522	622		}()
523	623
524		-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
	624	+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
525	625		require.NoError(t, err)
526	626		select {
527	627		case r := <-gotResize:

@@ -550,7 +650,7 @@ func TestRunRestoresRawModeAfterAttachError(t *testing.T) {
550	650		d := portsmocks.NewMockDialer(t)
551	651		d.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
552	652
553		-	err := client.Run(context.Background(), d, tm, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
	653	+	err := client.Run(context.Background(), d, tm, realClock{}, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
554	654		require.Error(t, err)
555	655		require.Equal(t, int32(1), restoreCount.Load(), "Run must restore raw mode after attachOnce errors")
556	656		}

@@ -570,7 +670,7 @@ func TestRunDoesNotEnterRawBeforePreWelcomeError(t *testing.T) {
570	670		d := portsmocks.NewMockDialer(t)
571	671		d.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
572	672
573		-	err := client.Run(context.Background(), d, tm, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
	673	+	err := client.Run(context.Background(), d, tm, realClock{}, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
574	674		require.Error(t, err)
575	675		var pe *client.ProtocolError
576	676		require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)

@@ -582,7 +682,7 @@ func TestRunPhaseASingleAttempt(t *testing.T) {
582	682		d.EXPECT().Dial(mock.Anything).Return(nil, dialErr).Once()
583	683		tm := portsmocks.NewMockTerminal(t)
584	684
585		-	err := client.Run(context.Background(), d, tm, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
	685	+	err := client.Run(context.Background(), d, tm, realClock{}, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
586	686		require.ErrorIs(t, err, dialErr)
587	687		}
588	688

@@ -671,7 +771,7 @@ func TestRunReconnectsWithRotatedTokenAndSameClientID(t *testing.T) {
671	771		tr3 := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(33)}, {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))}}}
672	772		d := &sequenceDialer{trs: []ports.Transport{tr1, tr2, tr3}}
673	773
674		-	err := client.Run(context.Background(), d, term, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
	774	+	err := client.Run(context.Background(), d, term, realClock{}, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
675	775		require.NoError(t, err)
676	776		require.Equal(t, int32(3), d.calls.Load())
677	777		require.Equal(t, int32(1), term.rawCount.Load())

@@ -698,7 +798,7 @@ func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) {
698	798		tr := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonSessionKilled}))}}}
699	799		d := &sequenceDialer{trs: []ports.Transport{tr}}
700	800
701		-	err := client.Run(context.Background(), d, term, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
	801	+	err := client.Run(context.Background(), d, term, realClock{}, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
702	802		require.Error(t, err)
703	803		var de *client.DetachedError
704	804		require.True(t, errors.As(err, &de))

 	slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))
 }
+// realClock is a stdlib-backed ports.Clock for tests that exercise the stdin
+// pump end to end; the paste coalescer's flush timer runs on wall time here.
+type realClock struct{}
+func (realClock) Now() time.Time { return time.Now() }
+func (realClock) NewTimer(d time.Duration) ports.Timer {
+	return &realTimer{t: time.NewTimer(d)}
+}
+type realTimer struct{ t *time.Timer }
+func (r *realTimer) C() <-chan time.Time        { return r.t.C }
+func (r *realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) }
+func (r *realTimer) Stop() bool                 { return r.t.Stop() }
 // recvItem is one scripted Recv result.
 type recvItem struct {
 	f   ports.Frame
 func (r *oneShotBlockingReader) unblock() { close(r.done) }
+type chunkedBlockingReader struct {
+	chunks [][]byte
+	done   chan struct{}
+	mu     sync.Mutex
+	next   int
+}
+func newChunkedBlockingReader(chunks ...[]byte) *chunkedBlockingReader {
+	return &chunkedBlockingReader{chunks: chunks, done: make(chan struct{})}
+}
+func (r *chunkedBlockingReader) Read(p []byte) (int, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.next < len(r.chunks) {
+		chunk := r.chunks[r.next]
+		r.next++
+		return copy(p, chunk), nil
+	}
+	<-r.done
+	return 0, io.EOF
+}
+func (r *chunkedBlockingReader) unblock() { close(r.done) }
 func newHappyTerminal(t *testing.T, out *bytes.Buffer, restoreCount *atomic.Int32, resizeCh chan domain.Size) (*portsmocks.MockTerminal, *blockingReader) {
 	tm := portsmocks.NewMockTerminal(t)
 	tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
 	defer unblock()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.NoError(t, err)
 	require.Equal(t, "hello world", out.String())
 	require.Equal(t, int32(1), restoreCount.Load(), "restore must run exactly once")
 	).Once()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentAttach, "main")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentAttach, "main")
 	require.Error(t, err)
 	var pe *client.ProtocolError
 	require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
 	defer unblock()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.Error(t, err)
 	require.Equal(t, int32(1), restoreCount.Load(), "restore must run after mid-stream Recv error")
 }
 	defer unblock()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.Error(t, err)
 	require.ErrorIs(t, err, io.EOF)
 	require.Equal(t, int32(1), restoreCount.Load())
 	}).Maybe()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.NoError(t, err)
 	require.Equal(t, int32(1), restoreCount.Load())
 	}).Maybe()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.NoError(t, err)
 	select {
 	case th := <-gotTheme:
 	}).Maybe()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.NoError(t, err)
 	select {
 	case got := <-gotInput:
 	}).Maybe()
 	tr.EXPECT().Close().Return(nil).Once()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.NoError(t, err)
 	select {
 	case got := <-gotInput:
 	}
 }
+func TestAttachStdinCoalescesSplitBracketedPaste(t *testing.T) {
+	var out bytes.Buffer
+	var restoreCount atomic.Int32
+	resizeCh := make(chan domain.Size)
+	paste := []byte("\x1b[200~hello\nworld\x1b[201~")
+	input := newChunkedBlockingReader([]byte("\x1b[200~hello\n"), []byte("world\x1b[201~"))
+	defer input.unblock()
+	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().ResizeEvents().Return(resizeCh).Maybe()
+	gotInput := make(chan []byte, 1)
+	allowDetach := make(chan struct{})
+	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)
+		require.NoError(t, err)
+		gotInput <- in.Data
+		close(allowDetach)
+		return nil
+	}).Once()
+	welcome := frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))
+	detached := frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))
+	recvCh := make(chan recvItem, 1)
+	recvCh <- recvItem{f: welcome}
+	closed := make(chan struct{})
+	tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
+		select {
+		case it := <-recvCh:
+			return it.f, it.err
+		case <-allowDetach:
+			select {
+			case <-closed:
+				return ports.Frame{}, io.EOF
+			default:
+				close(closed)
+				return detached, nil
+			}
+		case <-closed:
+			return ports.Frame{}, io.EOF
+		}
+	}).Maybe()
+	tr.EXPECT().Close().Return(nil).Once()
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
+	require.NoError(t, err)
+	select {
+	case got := <-gotInput:
+		require.Equal(t, paste, got, "split bracketed paste must arrive as one intact MsgInput frame")
+	case <-time.After(2 * time.Second):
+		t.Fatal("split bracketed paste input was not sent")
+	}
+}
 func TestAttachForwardsResize(t *testing.T) {
 	var out bytes.Buffer
 	var restoreCount atomic.Int32
 		resizeCh <- domain.Size{Cols: 120, Rows: 40}
 	}()
-	err := client.Attach(context.Background(), tr, tm, ports.IntentEphemeral, "")
+	err := client.Attach(context.Background(), tr, tm, realClock{}, ports.IntentEphemeral, "")
 	require.NoError(t, err)
 	select {
 	case r := <-gotResize:
 	d := portsmocks.NewMockDialer(t)
 	d.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
-	err := client.Run(context.Background(), d, tm, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
+	err := client.Run(context.Background(), d, tm, realClock{}, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
 	require.Error(t, err)
 	require.Equal(t, int32(1), restoreCount.Load(), "Run must restore raw mode after attachOnce errors")
 }
 	d := portsmocks.NewMockDialer(t)
 	d.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
-	err := client.Run(context.Background(), d, tm, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
+	err := client.Run(context.Background(), d, tm, realClock{}, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
 	require.Error(t, err)
 	var pe *client.ProtocolError
 	require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
 	d.EXPECT().Dial(mock.Anything).Return(nil, dialErr).Once()
 	tm := portsmocks.NewMockTerminal(t)
-	err := client.Run(context.Background(), d, tm, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
+	err := client.Run(context.Background(), d, tm, realClock{}, ports.IntentEphemeral, "", slog.New(slog.DiscardHandler))
 	require.ErrorIs(t, err, dialErr)
 }
 	tr3 := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(33)}, {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))}}}
 	d := &sequenceDialer{trs: []ports.Transport{tr1, tr2, tr3}}
-	err := client.Run(context.Background(), d, term, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
+	err := client.Run(context.Background(), d, term, realClock{}, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
 	require.NoError(t, err)
 	require.Equal(t, int32(3), d.calls.Load())
 	require.Equal(t, int32(1), term.rawCount.Load())
 	tr := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonSessionKilled}))}}}
 	d := &sequenceDialer{trs: []ports.Transport{tr}}
-	err := client.Run(context.Background(), d, term, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
+	err := client.Run(context.Background(), d, term, realClock{}, ports.IntentAttach, "main", slog.New(slog.DiscardHandler))
 	require.Error(t, err)
 	var de *client.DetachedError
 	require.True(t, errors.As(err, &de))

A internal/usecase/client/paste_coalescer.go +253 -0

@@ -0,0 +1,253 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"sync"
	6	+	"time"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	)
	10	+
	11	+	const (
	12	+	// pasteFlushDelay bounds how long a trailing strict prefix of the opening
	13	+	// marker is held before it is emitted as ordinary bytes. It only has to
	14	+	// bridge a local kernel read split: the rest of the marker is already in
	15	+	// the tty buffer and arrives on the next Read microseconds later. It is NOT
	16	+	// meant to bridge a network-scale gap — once the opening marker completes,
	17	+	// buffering mode holds the whole paste (up to the 2s safety timer), so a
	18	+	// close-marker split at any realistic latency is covered by whole-paste
	19	+	// framing rather than this timer. Kept well under the daemon router's
	20	+	// ESCDelay so a real lone Esc keypress still reaches the pane promptly.
	21	+	pasteFlushDelay = 20 * time.Millisecond
	22	+	// pasteSafetyDelay caps how long the coalescer waits for a closing marker
	23	+	// before flushing a partial paste, so a paste with a lost closing marker
	24	+	// degrades to today's per-frame behavior instead of stalling input.
	25	+	pasteSafetyDelay = 2 * time.Second
	26	+	// maxPasteBuffer caps buffered paste bytes. Beyond it the buffer is flushed
	27	+	// verbatim and normal scanning resumes; input is never dropped.
	28	+	maxPasteBuffer = 2 << 20 // 2 MiB
	29	+	)
	30	+
	31	+	// pasteCoalescer reframes a bracketed paste that arrives split across input
	32	+	// reads into a single emit call. Over a remote attach, stdin is read in fixed
	33	+	// chunks and forwarded frame by frame; an input-frame boundary can land inside
	34	+	// a paste marker (e.g. between the marker's ESC and its '['). When the daemon's
	35	+	// keys router then sees a lone trailing ESC it retains it for ESCDelay and, if
	36	+	// the next frame is late, forwards it as a standalone Esc keypress — breaking
	37	+	// the opening marker so the pane's TUI never enters paste mode. Coalescing the
	38	+	// whole paste into one emit keeps both markers contiguous on the wire.
	39	+	//
	40	+	// It is safe for concurrent use: Scan runs on the stdin pump goroutine while a
	41	+	// flush/safety timer may fire on its own goroutine; both take the mutex, so
	42	+	// emit calls are serialised.
	43	+	type pasteCoalescer struct {
	44	+	clock ports.Clock
	45	+	emit func([]byte)
	46	+
	47	+	mu sync.Mutex
	48	+	buffering bool // inside a paste: accumulating until the closing marker
	49	+	buf []byte // paste bytes accumulated in buffering mode
	50	+	pending []byte // held trailing strict prefix of the opening marker
	51	+
	52	+	flushTimer ports.Timer
	53	+	flushDone chan struct{}
	54	+	safetyTimer ports.Timer
	55	+	safetyDone chan struct{}
	56	+	closed bool
	57	+	}
	58	+
	59	+	func newPasteCoalescer(clock ports.Clock, emit func([]byte)) *pasteCoalescer {
	60	+	return &pasteCoalescer{clock: clock, emit: emit}
	61	+	}
	62	+
	63	+	// Scan consumes one run of ordinary input bytes, emitting either pass-through
	64	+	// bytes immediately or a whole bracketed paste as a single emit call once its
	65	+	// closing marker is seen.
	66	+	func (c *pasteCoalescer) Scan(data []byte) {
	67	+	if len(data) == 0 {
	68	+	return
	69	+	}
	70	+	c.mu.Lock()
	71	+	defer c.mu.Unlock()
	72	+	if len(c.pending) > 0 {
	73	+	// New bytes arrived before the flush timer fired: the held prefix was
	74	+	// the start of this data, not a lone Esc. Recombine and rescan.
	75	+	c.stopFlushTimer()
	76	+	data = append(c.pending, data...)
	77	+	c.pending = nil
	78	+	}
	79	+	c.process(data)
	80	+	}
	81	+
	82	+	// process runs the scan/buffer state machine with the mutex held.
	83	+	func (c *pasteCoalescer) process(data []byte) {
	84	+	for {
	85	+	if c.buffering {
	86	+	searchFrom := len(c.buf)
	87	+	if overlap := len(ports.BracketedPasteCloseMarker) - 1; searchFrom > overlap {
	88	+	searchFrom -= overlap
	89	+	} else {
	90	+	searchFrom = 0
	91	+	}
	92	+	c.buf = append(c.buf, data...)
	93	+	idx := bytes.Index(c.buf[searchFrom:], ports.BracketedPasteCloseMarker)
	94	+	if idx >= 0 {
	95	+	idx += searchFrom
	96	+	}
	97	+	if idx < 0 {
	98	+	if len(c.buf) > maxPasteBuffer {
	99	+	c.emit(c.buf)
	100	+	c.resetBuffering()
	101	+	}
	102	+	return
	103	+	}
	104	+	end := idx + len(ports.BracketedPasteCloseMarker)
	105	+	c.emit(c.buf[:end])
	106	+	rest := append([]byte(nil), c.buf[end:]...)
	107	+	c.resetBuffering()
	108	+	data = rest
	109	+	// Fall through to scan any bytes trailing the paste.
	110	+	}
	111	+
	112	+	if len(data) == 0 {
	113	+	return
	114	+	}
	115	+
	116	+	if idx := bytes.Index(data, ports.BracketedPasteOpenMarker); idx >= 0 {
	117	+	if idx > 0 {
	118	+	c.emit(data[:idx])
	119	+	}
	120	+	c.buffering = true
	121	+	c.startSafetyTimer()
	122	+	data = data[idx:] // marker onward; re-enter the buffering branch
	123	+	continue
	124	+	}
	125	+
	126	+	if p := trailingOpenPrefixLen(data); p > 0 {
	127	+	if len(data) > p {
	128	+	c.emit(data[:len(data)-p])
	129	+	}
	130	+	c.pending = append([]byte(nil), data[len(data)-p:]...)
	131	+	c.startFlushTimer()
	132	+	return
	133	+	}
	134	+
	135	+	c.emit(data)
	136	+	return
	137	+	}
	138	+	}
	139	+
	140	+	// resetBuffering leaves paste mode and stops the safety timer.
	141	+	func (c *pasteCoalescer) resetBuffering() {
	142	+	c.buffering = false
	143	+	c.buf = nil
	144	+	c.stopSafetyTimer()
	145	+	}
	146	+
	147	+	// Close stops any pending timers and their goroutines. Held bytes are dropped:
	148	+	// Close runs only as the stdin pump unwinds on detach, when there is nowhere
	149	+	// left to deliver them.
	150	+	func (c *pasteCoalescer) Close() {
	151	+	c.mu.Lock()
	152	+	defer c.mu.Unlock()
	153	+	c.closed = true
	154	+	c.stopFlushTimer()
	155	+	c.stopSafetyTimer()
	156	+	}
	157	+
	158	+	func (c *pasteCoalescer) startFlushTimer() {
	159	+	c.stopFlushTimer()
	160	+	if c.closed {
	161	+	return
	162	+	}
	163	+	timer := c.clock.NewTimer(pasteFlushDelay)
	164	+	done := make(chan struct{})
	165	+	c.flushTimer = timer
	166	+	c.flushDone = done
	167	+	go func() {
	168	+	select {
	169	+	case <-timer.C():
	170	+	case <-done:
	171	+	return
	172	+	}
	173	+	c.mu.Lock()
	174	+	defer c.mu.Unlock()
	175	+	if c.flushTimer != timer \|\| len(c.pending) == 0 {
	176	+	return
	177	+	}
	178	+	c.flushTimer = nil
	179	+	c.flushDone = nil
	180	+	held := c.pending
	181	+	c.pending = nil
	182	+	c.emit(held)
	183	+	}()
	184	+	}
	185	+
	186	+	func (c *pasteCoalescer) stopFlushTimer() {
	187	+	if c.flushTimer != nil {
	188	+	c.flushTimer.Stop()
	189	+	c.flushTimer = nil
	190	+	}
	191	+	if c.flushDone != nil {
	192	+	close(c.flushDone)
	193	+	c.flushDone = nil
	194	+	}
	195	+	}
	196	+
	197	+	func (c *pasteCoalescer) startSafetyTimer() {
	198	+	c.stopSafetyTimer()
	199	+	if c.closed {
	200	+	return
	201	+	}
	202	+	timer := c.clock.NewTimer(pasteSafetyDelay)
	203	+	done := make(chan struct{})
	204	+	c.safetyTimer = timer
	205	+	c.safetyDone = done
	206	+	go func() {
	207	+	select {
	208	+	case <-timer.C():
	209	+	case <-done:
	210	+	return
	211	+	}
	212	+	c.mu.Lock()
	213	+	defer c.mu.Unlock()
	214	+	if c.safetyTimer != timer \|\| !c.buffering {
	215	+	return
	216	+	}
	217	+	buffered := c.buf
	218	+	c.buffering = false
	219	+	c.buf = nil
	220	+	c.safetyTimer = nil
	221	+	c.safetyDone = nil
	222	+	if len(buffered) > 0 {
	223	+	c.emit(buffered)
	224	+	}
	225	+	}()
	226	+	}
	227	+
	228	+	func (c *pasteCoalescer) stopSafetyTimer() {
	229	+	if c.safetyTimer != nil {
	230	+	c.safetyTimer.Stop()
	231	+	c.safetyTimer = nil
	232	+	}
	233	+	if c.safetyDone != nil {
	234	+	close(c.safetyDone)
	235	+	c.safetyDone = nil
	236	+	}
	237	+	}
	238	+
	239	+	// trailingOpenPrefixLen returns the length of the longest suffix of data that
	240	+	// is a non-empty strict prefix of the opening marker (e.g. a trailing "\x1b" or
	241	+	// "\x1b[20"). A full marker match is handled by the caller before this runs.
	242	+	func trailingOpenPrefixLen(data []byte) int {
	243	+	matchLimit := len(ports.BracketedPasteOpenMarker) - 1
	244	+	if matchLimit > len(data) {
	245	+	matchLimit = len(data)
	246	+	}
	247	+	for n := matchLimit; n >= 1; n-- {
	248	+	if bytes.Equal(data[len(data)-n:], ports.BracketedPasteOpenMarker[:n]) {
	249	+	return n
	250	+	}
	251	+	}
	252	+	return 0
	253	+	}

@@ -0,0 +1,253 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"sync"
	6	+	"time"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	)
	10	+
	11	+	const (
	12	+	// pasteFlushDelay bounds how long a trailing strict prefix of the opening
	13	+	// marker is held before it is emitted as ordinary bytes. It only has to
	14	+	// bridge a local kernel read split: the rest of the marker is already in
	15	+	// the tty buffer and arrives on the next Read microseconds later. It is NOT
	16	+	// meant to bridge a network-scale gap — once the opening marker completes,
	17	+	// buffering mode holds the whole paste (up to the 2s safety timer), so a
	18	+	// close-marker split at any realistic latency is covered by whole-paste
	19	+	// framing rather than this timer. Kept well under the daemon router's
	20	+	// ESCDelay so a real lone Esc keypress still reaches the pane promptly.
	21	+	pasteFlushDelay = 20 * time.Millisecond
	22	+	// pasteSafetyDelay caps how long the coalescer waits for a closing marker
	23	+	// before flushing a partial paste, so a paste with a lost closing marker
	24	+	// degrades to today's per-frame behavior instead of stalling input.
	25	+	pasteSafetyDelay = 2 * time.Second
	26	+	// maxPasteBuffer caps buffered paste bytes. Beyond it the buffer is flushed
	27	+	// verbatim and normal scanning resumes; input is never dropped.
	28	+	maxPasteBuffer = 2 << 20 // 2 MiB
	29	+	)
	30	+
	31	+	// pasteCoalescer reframes a bracketed paste that arrives split across input
	32	+	// reads into a single emit call. Over a remote attach, stdin is read in fixed
	33	+	// chunks and forwarded frame by frame; an input-frame boundary can land inside
	34	+	// a paste marker (e.g. between the marker's ESC and its '['). When the daemon's
	35	+	// keys router then sees a lone trailing ESC it retains it for ESCDelay and, if
	36	+	// the next frame is late, forwards it as a standalone Esc keypress — breaking
	37	+	// the opening marker so the pane's TUI never enters paste mode. Coalescing the
	38	+	// whole paste into one emit keeps both markers contiguous on the wire.
	39	+	//
	40	+	// It is safe for concurrent use: Scan runs on the stdin pump goroutine while a
	41	+	// flush/safety timer may fire on its own goroutine; both take the mutex, so
	42	+	// emit calls are serialised.
	43	+	type pasteCoalescer struct {
	44	+	clock ports.Clock
	45	+	emit func([]byte)
	46	+
	47	+	mu sync.Mutex
	48	+	buffering bool // inside a paste: accumulating until the closing marker
	49	+	buf []byte // paste bytes accumulated in buffering mode
	50	+	pending []byte // held trailing strict prefix of the opening marker
	51	+
	52	+	flushTimer ports.Timer
	53	+	flushDone chan struct{}
	54	+	safetyTimer ports.Timer
	55	+	safetyDone chan struct{}
	56	+	closed bool
	57	+	}
	58	+
	59	+	func newPasteCoalescer(clock ports.Clock, emit func([]byte)) *pasteCoalescer {
	60	+	return &pasteCoalescer{clock: clock, emit: emit}
	61	+	}
	62	+
	63	+	// Scan consumes one run of ordinary input bytes, emitting either pass-through
	64	+	// bytes immediately or a whole bracketed paste as a single emit call once its
	65	+	// closing marker is seen.
	66	+	func (c *pasteCoalescer) Scan(data []byte) {
	67	+	if len(data) == 0 {
	68	+	return
	69	+	}
	70	+	c.mu.Lock()
	71	+	defer c.mu.Unlock()
	72	+	if len(c.pending) > 0 {
	73	+	// New bytes arrived before the flush timer fired: the held prefix was
	74	+	// the start of this data, not a lone Esc. Recombine and rescan.
	75	+	c.stopFlushTimer()
	76	+	data = append(c.pending, data...)
	77	+	c.pending = nil
	78	+	}
	79	+	c.process(data)
	80	+	}
	81	+
	82	+	// process runs the scan/buffer state machine with the mutex held.
	83	+	func (c *pasteCoalescer) process(data []byte) {
	84	+	for {
	85	+	if c.buffering {
	86	+	searchFrom := len(c.buf)
	87	+	if overlap := len(ports.BracketedPasteCloseMarker) - 1; searchFrom > overlap {
	88	+	searchFrom -= overlap
	89	+	} else {
	90	+	searchFrom = 0
	91	+	}
	92	+	c.buf = append(c.buf, data...)
	93	+	idx := bytes.Index(c.buf[searchFrom:], ports.BracketedPasteCloseMarker)
	94	+	if idx >= 0 {
	95	+	idx += searchFrom
	96	+	}
	97	+	if idx < 0 {
	98	+	if len(c.buf) > maxPasteBuffer {
	99	+	c.emit(c.buf)
	100	+	c.resetBuffering()
	101	+	}
	102	+	return
	103	+	}
	104	+	end := idx + len(ports.BracketedPasteCloseMarker)
	105	+	c.emit(c.buf[:end])
	106	+	rest := append([]byte(nil), c.buf[end:]...)
	107	+	c.resetBuffering()
	108	+	data = rest
	109	+	// Fall through to scan any bytes trailing the paste.
	110	+	}
	111	+
	112	+	if len(data) == 0 {
	113	+	return
	114	+	}
	115	+
	116	+	if idx := bytes.Index(data, ports.BracketedPasteOpenMarker); idx >= 0 {
	117	+	if idx > 0 {
	118	+	c.emit(data[:idx])
	119	+	}
	120	+	c.buffering = true
	121	+	c.startSafetyTimer()
	122	+	data = data[idx:] // marker onward; re-enter the buffering branch
	123	+	continue
	124	+	}
	125	+
	126	+	if p := trailingOpenPrefixLen(data); p > 0 {
	127	+	if len(data) > p {
	128	+	c.emit(data[:len(data)-p])
	129	+	}
	130	+	c.pending = append([]byte(nil), data[len(data)-p:]...)
	131	+	c.startFlushTimer()
	132	+	return
	133	+	}
	134	+
	135	+	c.emit(data)
	136	+	return
	137	+	}
	138	+	}
	139	+
	140	+	// resetBuffering leaves paste mode and stops the safety timer.
	141	+	func (c *pasteCoalescer) resetBuffering() {
	142	+	c.buffering = false
	143	+	c.buf = nil
	144	+	c.stopSafetyTimer()
	145	+	}
	146	+
	147	+	// Close stops any pending timers and their goroutines. Held bytes are dropped:
	148	+	// Close runs only as the stdin pump unwinds on detach, when there is nowhere
	149	+	// left to deliver them.
	150	+	func (c *pasteCoalescer) Close() {
	151	+	c.mu.Lock()
	152	+	defer c.mu.Unlock()
	153	+	c.closed = true
	154	+	c.stopFlushTimer()
	155	+	c.stopSafetyTimer()
	156	+	}
	157	+
	158	+	func (c *pasteCoalescer) startFlushTimer() {
	159	+	c.stopFlushTimer()
	160	+	if c.closed {
	161	+	return
	162	+	}
	163	+	timer := c.clock.NewTimer(pasteFlushDelay)
	164	+	done := make(chan struct{})
	165	+	c.flushTimer = timer
	166	+	c.flushDone = done
	167	+	go func() {
	168	+	select {
	169	+	case <-timer.C():
	170	+	case <-done:
	171	+	return
	172	+	}
	173	+	c.mu.Lock()
	174	+	defer c.mu.Unlock()
	175	+	if c.flushTimer != timer \|\| len(c.pending) == 0 {
	176	+	return
	177	+	}
	178	+	c.flushTimer = nil
	179	+	c.flushDone = nil
	180	+	held := c.pending
	181	+	c.pending = nil
	182	+	c.emit(held)
	183	+	}()
	184	+	}
	185	+
	186	+	func (c *pasteCoalescer) stopFlushTimer() {
	187	+	if c.flushTimer != nil {
	188	+	c.flushTimer.Stop()
	189	+	c.flushTimer = nil
	190	+	}
	191	+	if c.flushDone != nil {
	192	+	close(c.flushDone)
	193	+	c.flushDone = nil
	194	+	}
	195	+	}
	196	+
	197	+	func (c *pasteCoalescer) startSafetyTimer() {
	198	+	c.stopSafetyTimer()
	199	+	if c.closed {
	200	+	return
	201	+	}
	202	+	timer := c.clock.NewTimer(pasteSafetyDelay)
	203	+	done := make(chan struct{})
	204	+	c.safetyTimer = timer
	205	+	c.safetyDone = done
	206	+	go func() {
	207	+	select {
	208	+	case <-timer.C():
	209	+	case <-done:
	210	+	return
	211	+	}
	212	+	c.mu.Lock()
	213	+	defer c.mu.Unlock()
	214	+	if c.safetyTimer != timer \|\| !c.buffering {
	215	+	return
	216	+	}
	217	+	buffered := c.buf
	218	+	c.buffering = false
	219	+	c.buf = nil
	220	+	c.safetyTimer = nil
	221	+	c.safetyDone = nil
	222	+	if len(buffered) > 0 {
	223	+	c.emit(buffered)
	224	+	}
	225	+	}()
	226	+	}
	227	+
	228	+	func (c *pasteCoalescer) stopSafetyTimer() {
	229	+	if c.safetyTimer != nil {
	230	+	c.safetyTimer.Stop()
	231	+	c.safetyTimer = nil
	232	+	}
	233	+	if c.safetyDone != nil {
	234	+	close(c.safetyDone)
	235	+	c.safetyDone = nil
	236	+	}
	237	+	}
	238	+
	239	+	// trailingOpenPrefixLen returns the length of the longest suffix of data that
	240	+	// is a non-empty strict prefix of the opening marker (e.g. a trailing "\x1b" or
	241	+	// "\x1b[20"). A full marker match is handled by the caller before this runs.
	242	+	func trailingOpenPrefixLen(data []byte) int {
	243	+	matchLimit := len(ports.BracketedPasteOpenMarker) - 1
	244	+	if matchLimit > len(data) {
	245	+	matchLimit = len(data)
	246	+	}
	247	+	for n := matchLimit; n >= 1; n-- {
	248	+	if bytes.Equal(data[len(data)-n:], ports.BracketedPasteOpenMarker[:n]) {
	249	+	return n
	250	+	}
	251	+	}
	252	+	return 0
	253	+	}

A internal/usecase/client/paste_coalescer_test.go +227 -0

@@ -0,0 +1,227 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"strconv"
	6	+	"sync"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	)
	14	+
	15	+	// pcFakeClock hands out fake timers the test fires by hand. The coalescer only
	16	+	// ever has one timer live at a time (a flush timer in normal mode or a safety
	17	+	// timer while buffering), so fireLast drives whichever is pending.
	18	+	type pcFakeClock struct {
	19	+	mu sync.Mutex
	20	+	timers []*pcFakeTimer
	21	+	}
	22	+
	23	+	func (c *pcFakeClock) Now() time.Time { return time.Time{} }
	24	+
	25	+	func (c *pcFakeClock) NewTimer(d time.Duration) ports.Timer {
	26	+	c.mu.Lock()
	27	+	defer c.mu.Unlock()
	28	+	t := &pcFakeTimer{ch: make(chan time.Time, 1), d: d}
	29	+	c.timers = append(c.timers, t)
	30	+	return t
	31	+	}
	32	+
	33	+	func (c *pcFakeClock) fireLast() {
	34	+	c.mu.Lock()
	35	+	defer c.mu.Unlock()
	36	+	for i := len(c.timers) - 1; i >= 0; i-- {
	37	+	if !c.timers[i].stopped && !c.timers[i].fired {
	38	+	c.timers[i].fired = true
	39	+	c.timers[i].ch <- time.Time{}
	40	+	return
	41	+	}
	42	+	}
	43	+	}
	44	+
	45	+	type pcFakeTimer struct {
	46	+	ch chan time.Time
	47	+	d time.Duration
	48	+	stopped bool
	49	+	fired bool
	50	+	}
	51	+
	52	+	func (t *pcFakeTimer) C() <-chan time.Time { return t.ch }
	53	+	func (t *pcFakeTimer) Reset(d time.Duration) bool { t.d = d; return !t.stopped }
	54	+	func (t *pcFakeTimer) Stop() bool { s := t.stopped; t.stopped = true; return !s }
	55	+
	56	+	// pcCollector records emit calls under a mutex so a timer goroutine's emit is
	57	+	// safe to observe from the test goroutine.
	58	+	type pcCollector struct {
	59	+	mu sync.Mutex
	60	+	emits [][]byte
	61	+	}
	62	+
	63	+	func (c *pcCollector) emit(data []byte) {
	64	+	c.mu.Lock()
	65	+	defer c.mu.Unlock()
	66	+	c.emits = append(c.emits, append([]byte(nil), data...))
	67	+	}
	68	+
	69	+	func (c *pcCollector) snapshot() [][]byte {
	70	+	c.mu.Lock()
	71	+	defer c.mu.Unlock()
	72	+	out := make([][]byte, len(c.emits))
	73	+	copy(out, c.emits)
	74	+	return out
	75	+	}
	76	+
	77	+	func (c pcCollector) waitEmits(t testing.T, n int) {
	78	+	t.Helper()
	79	+	require.Eventually(t, func() bool {
	80	+	c.mu.Lock()
	81	+	defer c.mu.Unlock()
	82	+	return len(c.emits) >= n
	83	+	}, time.Second, time.Millisecond)
	84	+	}
	85	+
	86	+	func newTestCoalescer() (pasteCoalescer, pcFakeClock, *pcCollector) {
	87	+	clk := &pcFakeClock{}
	88	+	col := &pcCollector{}
	89	+	return newPasteCoalescer(clk, col.emit), clk, col
	90	+	}
	91	+
	92	+	// samplePaste is a multiline bracketed paste whose content deliberately embeds
	93	+	// sequences that would otherwise be interpreted (a lone ESC+letter, an Alt
	94	+	// arrow CSI, a marker-looking prefix) to prove content is carried verbatim.
	95	+	var samplePaste = []byte("\x1b[200~line one\nline\x1btwo\n\x1b[1;3Athree\x1b[20not-a-marker\x1b[201~")
	96	+
	97	+	func TestPasteCoalescerSplitAtEveryOffsetTwoReads(t *testing.T) {
	98	+	for s := 0; s <= len(samplePaste); s++ {
	99	+	t.Run("split_"+strconv.Itoa(s), func(t *testing.T) {
	100	+	pc, _, col := newTestCoalescer()
	101	+	defer pc.Close()
	102	+	pc.Scan(samplePaste[:s])
	103	+	pc.Scan(samplePaste[s:])
	104	+	require.Equal(t, [][]byte{samplePaste}, col.snapshot(),
	105	+	"paste split at offset %d must emit exactly once, byte-identical", s)
	106	+	})
	107	+	}
	108	+	}
	109	+
	110	+	func TestPasteCoalescerByteByByte(t *testing.T) {
	111	+	pc, _, col := newTestCoalescer()
	112	+	defer pc.Close()
	113	+	for i := 0; i < len(samplePaste); i++ {
	114	+	pc.Scan(samplePaste[i : i+1])
	115	+	}
	116	+	require.Equal(t, [][]byte{samplePaste}, col.snapshot(),
	117	+	"paste fed one byte at a time must still emit as one contiguous paste")
	118	+	}
	119	+
	120	+	func TestPasteCoalescerBackToBackPastesInOneRead(t *testing.T) {
	121	+	pc, _, col := newTestCoalescer()
	122	+	defer pc.Close()
	123	+
	124	+	first := []byte("\x1b[200~first\x1b[201~")
	125	+	second := []byte("\x1b[200~second\x1b[201~")
	126	+	pc.Scan(append(append([]byte(nil), first...), second...))
	127	+
	128	+	require.Equal(t, [][]byte{first, second}, col.snapshot(),
	129	+	"back-to-back pastes in one read must emit as two intact paste frames")
	130	+	}
	131	+
	132	+	func TestPasteCoalescerLoneEscapeFlushedAfterTimer(t *testing.T) {
	133	+	pc, clk, col := newTestCoalescer()
	134	+	defer pc.Close()
	135	+
	136	+	pc.Scan([]byte{0x1b})
	137	+	require.Empty(t, col.snapshot(), "a trailing ESC must be held, not emitted immediately")
	138	+
	139	+	clk.fireLast()
	140	+	col.waitEmits(t, 1)
	141	+	require.Equal(t, [][]byte{{0x1b}}, col.snapshot(), "held ESC must flush unchanged after the timer")
	142	+	}
	143	+
	144	+	func TestPasteCoalescerEscapeThenOpenMarkerNextRead(t *testing.T) {
	145	+	pc, _, col := newTestCoalescer()
	146	+	defer pc.Close()
	147	+
	148	+	pc.Scan([]byte{0x1b})
	149	+	require.Empty(t, col.snapshot(), "ESC held pending the rest of the marker")
	150	+
	151	+	pc.Scan(samplePaste[1:]) // "[200~...\x1b[201~"
	152	+	require.Equal(t, [][]byte{samplePaste}, col.snapshot(),
	153	+	"ESC recombined with the following marker must emit a single paste")
	154	+	}
	155	+
	156	+	func TestPasteCoalescerPlainTextPassthrough(t *testing.T) {
	157	+	pc, _, col := newTestCoalescer()
	158	+	defer pc.Close()
	159	+
	160	+	pc.Scan([]byte("hello world"))
	161	+	pc.Scan([]byte("second read"))
	162	+	require.Equal(t, [][]byte{[]byte("hello world"), []byte("second read")}, col.snapshot(),
	163	+	"plain text must pass through unchanged at the same call granularity")
	164	+	}
	165	+
	166	+	func TestPasteCoalescerTextPasteTextInOneRead(t *testing.T) {
	167	+	pc, _, col := newTestCoalescer()
	168	+	defer pc.Close()
	169	+
	170	+	paste := []byte("\x1b[200~pasted\x1b[201~")
	171	+	read := append(append([]byte("AB"), paste...), []byte("CD")...)
	172	+	pc.Scan(read)
	173	+	require.Equal(t, [][]byte{[]byte("AB"), paste, []byte("CD")}, col.snapshot(),
	174	+	"text + paste + text in one read must emit three chunks in order")
	175	+	}
	176	+
	177	+	func TestPasteCoalescerBufferCapOverflowFlushesVerbatim(t *testing.T) {
	178	+	pc, _, col := newTestCoalescer()
	179	+	defer pc.Close()
	180	+
	181	+	big := append([]byte("\x1b[200~"), bytes.Repeat([]byte("x"), maxPasteBuffer+16)...)
	182	+	pc.Scan(big)
	183	+	require.Equal(t, [][]byte{big}, col.snapshot(),
	184	+	"an oversized unterminated paste must flush verbatim without dropping bytes")
	185	+
	186	+	// The coalescer must be back in normal mode and keep working.
	187	+	pc.Scan([]byte("after"))
	188	+	require.Equal(t, [][]byte{big, []byte("after")}, col.snapshot())
	189	+	}
	190	+
	191	+	func TestPasteCoalescerSafetyTimeoutFlushesVerbatim(t *testing.T) {
	192	+	pc, clk, col := newTestCoalescer()
	193	+	defer pc.Close()
	194	+
	195	+	partial := []byte("\x1b[200~unterminated paste")
	196	+	pc.Scan(partial)
	197	+	require.Empty(t, col.snapshot(), "an open paste with no closing marker is buffered, not emitted")
	198	+
	199	+	clk.fireLast() // safety timer
	200	+	col.waitEmits(t, 1)
	201	+	require.Equal(t, [][]byte{partial}, col.snapshot(),
	202	+	"the safety timeout must flush buffered paste bytes verbatim")
	203	+
	204	+	pc.Scan([]byte("normal"))
	205	+	require.Equal(t, [][]byte{partial, []byte("normal")}, col.snapshot(),
	206	+	"after the safety flush the coalescer must resume normal scanning")
	207	+	}
	208	+
	209	+	func TestTrailingOpenPrefixLen(t *testing.T) {
	210	+	cases := []struct {
	211	+	in string
	212	+	want int
	213	+	}{
	214	+	{"", 0},
	215	+	{"abc", 0},
	216	+	{"abc\x1b", 1},
	217	+	{"abc\x1b[", 2},
	218	+	{"\x1b[2", 3},
	219	+	{"\x1b[20", 4},
	220	+	{"\x1b[200", 5},
	221	+	{"\x1b[200~", 0}, // full marker is not a strict prefix
	222	+	{"[200", 0}, // no ESC: not a marker start
	223	+	}
	224	+	for _, tc := range cases {
	225	+	require.Equal(t, tc.want, trailingOpenPrefixLen([]byte(tc.in)), "input %q", tc.in)
	226	+	}
	227	+	}

@@ -0,0 +1,227 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"bytes"
	5	+	"strconv"
	6	+	"sync"
	7	+	"testing"
	8	+	"time"
	9	+
	10	+	"github.com/stretchr/testify/require"
	11	+
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	)
	14	+
	15	+	// pcFakeClock hands out fake timers the test fires by hand. The coalescer only
	16	+	// ever has one timer live at a time (a flush timer in normal mode or a safety
	17	+	// timer while buffering), so fireLast drives whichever is pending.
	18	+	type pcFakeClock struct {
	19	+	mu sync.Mutex
	20	+	timers []*pcFakeTimer
	21	+	}
	22	+
	23	+	func (c *pcFakeClock) Now() time.Time { return time.Time{} }
	24	+
	25	+	func (c *pcFakeClock) NewTimer(d time.Duration) ports.Timer {
	26	+	c.mu.Lock()
	27	+	defer c.mu.Unlock()
	28	+	t := &pcFakeTimer{ch: make(chan time.Time, 1), d: d}
	29	+	c.timers = append(c.timers, t)
	30	+	return t
	31	+	}
	32	+
	33	+	func (c *pcFakeClock) fireLast() {
	34	+	c.mu.Lock()
	35	+	defer c.mu.Unlock()
	36	+	for i := len(c.timers) - 1; i >= 0; i-- {
	37	+	if !c.timers[i].stopped && !c.timers[i].fired {
	38	+	c.timers[i].fired = true
	39	+	c.timers[i].ch <- time.Time{}
	40	+	return
	41	+	}
	42	+	}
	43	+	}
	44	+
	45	+	type pcFakeTimer struct {
	46	+	ch chan time.Time
	47	+	d time.Duration
	48	+	stopped bool
	49	+	fired bool
	50	+	}
	51	+
	52	+	func (t *pcFakeTimer) C() <-chan time.Time { return t.ch }
	53	+	func (t *pcFakeTimer) Reset(d time.Duration) bool { t.d = d; return !t.stopped }
	54	+	func (t *pcFakeTimer) Stop() bool { s := t.stopped; t.stopped = true; return !s }
	55	+
	56	+	// pcCollector records emit calls under a mutex so a timer goroutine's emit is
	57	+	// safe to observe from the test goroutine.
	58	+	type pcCollector struct {
	59	+	mu sync.Mutex
	60	+	emits [][]byte
	61	+	}
	62	+
	63	+	func (c *pcCollector) emit(data []byte) {
	64	+	c.mu.Lock()
	65	+	defer c.mu.Unlock()
	66	+	c.emits = append(c.emits, append([]byte(nil), data...))
	67	+	}
	68	+
	69	+	func (c *pcCollector) snapshot() [][]byte {
	70	+	c.mu.Lock()
	71	+	defer c.mu.Unlock()
	72	+	out := make([][]byte, len(c.emits))
	73	+	copy(out, c.emits)
	74	+	return out
	75	+	}
	76	+
	77	+	func (c pcCollector) waitEmits(t testing.T, n int) {
	78	+	t.Helper()
	79	+	require.Eventually(t, func() bool {
	80	+	c.mu.Lock()
	81	+	defer c.mu.Unlock()
	82	+	return len(c.emits) >= n
	83	+	}, time.Second, time.Millisecond)
	84	+	}
	85	+
	86	+	func newTestCoalescer() (pasteCoalescer, pcFakeClock, *pcCollector) {
	87	+	clk := &pcFakeClock{}
	88	+	col := &pcCollector{}
	89	+	return newPasteCoalescer(clk, col.emit), clk, col
	90	+	}
	91	+
	92	+	// samplePaste is a multiline bracketed paste whose content deliberately embeds
	93	+	// sequences that would otherwise be interpreted (a lone ESC+letter, an Alt
	94	+	// arrow CSI, a marker-looking prefix) to prove content is carried verbatim.
	95	+	var samplePaste = []byte("\x1b[200~line one\nline\x1btwo\n\x1b[1;3Athree\x1b[20not-a-marker\x1b[201~")
	96	+
	97	+	func TestPasteCoalescerSplitAtEveryOffsetTwoReads(t *testing.T) {
	98	+	for s := 0; s <= len(samplePaste); s++ {
	99	+	t.Run("split_"+strconv.Itoa(s), func(t *testing.T) {
	100	+	pc, _, col := newTestCoalescer()
	101	+	defer pc.Close()
	102	+	pc.Scan(samplePaste[:s])
	103	+	pc.Scan(samplePaste[s:])
	104	+	require.Equal(t, [][]byte{samplePaste}, col.snapshot(),
	105	+	"paste split at offset %d must emit exactly once, byte-identical", s)
	106	+	})
	107	+	}
	108	+	}
	109	+
	110	+	func TestPasteCoalescerByteByByte(t *testing.T) {
	111	+	pc, _, col := newTestCoalescer()
	112	+	defer pc.Close()
	113	+	for i := 0; i < len(samplePaste); i++ {
	114	+	pc.Scan(samplePaste[i : i+1])
	115	+	}
	116	+	require.Equal(t, [][]byte{samplePaste}, col.snapshot(),
	117	+	"paste fed one byte at a time must still emit as one contiguous paste")
	118	+	}
	119	+
	120	+	func TestPasteCoalescerBackToBackPastesInOneRead(t *testing.T) {
	121	+	pc, _, col := newTestCoalescer()
	122	+	defer pc.Close()
	123	+
	124	+	first := []byte("\x1b[200~first\x1b[201~")
	125	+	second := []byte("\x1b[200~second\x1b[201~")
	126	+	pc.Scan(append(append([]byte(nil), first...), second...))
	127	+
	128	+	require.Equal(t, [][]byte{first, second}, col.snapshot(),
	129	+	"back-to-back pastes in one read must emit as two intact paste frames")
	130	+	}
	131	+
	132	+	func TestPasteCoalescerLoneEscapeFlushedAfterTimer(t *testing.T) {
	133	+	pc, clk, col := newTestCoalescer()
	134	+	defer pc.Close()
	135	+
	136	+	pc.Scan([]byte{0x1b})
	137	+	require.Empty(t, col.snapshot(), "a trailing ESC must be held, not emitted immediately")
	138	+
	139	+	clk.fireLast()
	140	+	col.waitEmits(t, 1)
	141	+	require.Equal(t, [][]byte{{0x1b}}, col.snapshot(), "held ESC must flush unchanged after the timer")
	142	+	}
	143	+
	144	+	func TestPasteCoalescerEscapeThenOpenMarkerNextRead(t *testing.T) {
	145	+	pc, _, col := newTestCoalescer()
	146	+	defer pc.Close()
	147	+
	148	+	pc.Scan([]byte{0x1b})
	149	+	require.Empty(t, col.snapshot(), "ESC held pending the rest of the marker")
	150	+
	151	+	pc.Scan(samplePaste[1:]) // "[200~...\x1b[201~"
	152	+	require.Equal(t, [][]byte{samplePaste}, col.snapshot(),
	153	+	"ESC recombined with the following marker must emit a single paste")
	154	+	}
	155	+
	156	+	func TestPasteCoalescerPlainTextPassthrough(t *testing.T) {
	157	+	pc, _, col := newTestCoalescer()
	158	+	defer pc.Close()
	159	+
	160	+	pc.Scan([]byte("hello world"))
	161	+	pc.Scan([]byte("second read"))
	162	+	require.Equal(t, [][]byte{[]byte("hello world"), []byte("second read")}, col.snapshot(),
	163	+	"plain text must pass through unchanged at the same call granularity")
	164	+	}
	165	+
	166	+	func TestPasteCoalescerTextPasteTextInOneRead(t *testing.T) {
	167	+	pc, _, col := newTestCoalescer()
	168	+	defer pc.Close()
	169	+
	170	+	paste := []byte("\x1b[200~pasted\x1b[201~")
	171	+	read := append(append([]byte("AB"), paste...), []byte("CD")...)
	172	+	pc.Scan(read)
	173	+	require.Equal(t, [][]byte{[]byte("AB"), paste, []byte("CD")}, col.snapshot(),
	174	+	"text + paste + text in one read must emit three chunks in order")
	175	+	}
	176	+
	177	+	func TestPasteCoalescerBufferCapOverflowFlushesVerbatim(t *testing.T) {
	178	+	pc, _, col := newTestCoalescer()
	179	+	defer pc.Close()
	180	+
	181	+	big := append([]byte("\x1b[200~"), bytes.Repeat([]byte("x"), maxPasteBuffer+16)...)
	182	+	pc.Scan(big)
	183	+	require.Equal(t, [][]byte{big}, col.snapshot(),
	184	+	"an oversized unterminated paste must flush verbatim without dropping bytes")
	185	+
	186	+	// The coalescer must be back in normal mode and keep working.
	187	+	pc.Scan([]byte("after"))
	188	+	require.Equal(t, [][]byte{big, []byte("after")}, col.snapshot())
	189	+	}
	190	+
	191	+	func TestPasteCoalescerSafetyTimeoutFlushesVerbatim(t *testing.T) {
	192	+	pc, clk, col := newTestCoalescer()
	193	+	defer pc.Close()
	194	+
	195	+	partial := []byte("\x1b[200~unterminated paste")
	196	+	pc.Scan(partial)
	197	+	require.Empty(t, col.snapshot(), "an open paste with no closing marker is buffered, not emitted")
	198	+
	199	+	clk.fireLast() // safety timer
	200	+	col.waitEmits(t, 1)
	201	+	require.Equal(t, [][]byte{partial}, col.snapshot(),
	202	+	"the safety timeout must flush buffered paste bytes verbatim")
	203	+
	204	+	pc.Scan([]byte("normal"))
	205	+	require.Equal(t, [][]byte{partial, []byte("normal")}, col.snapshot(),
	206	+	"after the safety flush the coalescer must resume normal scanning")
	207	+	}
	208	+
	209	+	func TestTrailingOpenPrefixLen(t *testing.T) {
	210	+	cases := []struct {
	211	+	in string
	212	+	want int
	213	+	}{
	214	+	{"", 0},
	215	+	{"abc", 0},
	216	+	{"abc\x1b", 1},
	217	+	{"abc\x1b[", 2},
	218	+	{"\x1b[2", 3},
	219	+	{"\x1b[20", 4},
	220	+	{"\x1b[200", 5},
	221	+	{"\x1b[200~", 0}, // full marker is not a strict prefix
	222	+	{"[200", 0}, // no ESC: not a marker start
	223	+	}
	224	+	for _, tc := range cases {
	225	+	require.Equal(t, tc.want, trailingOpenPrefixLen([]byte(tc.in)), "input %q", tc.in)
	226	+	}
	227	+	}

M internal/usecase/keys/router.go +16 -0


@@ -3,6 +3,7 @@
3	3		package keys
4	4
5	5		import (
	6	+	"bytes"
6	7		"slices"
7	8		"sync"
8	9		"time"

@@ -104,6 +105,21 @@ func (r *Router) route(data []byte) {
104	105		return
105	106		}
106	107		remaining := data[i+1:]
	108	+	if bytes.HasPrefix(data[i:], ports.BracketedPasteOpenMarker) {
	109	+	// A bracketed paste whose closing marker is in this same frame is
	110	+	// forwarded verbatim: pasted content bytes (including embedded
	111	+	// ESC+letter Alt lookalikes or ESC [ 1 ; 3 A sequences) must never
	112	+	// fire an Action mid-paste. Without a closing marker in-frame we
	113	+	// fall through to today's per-byte routing (the '[' passes through
	114	+	// as a control prefix); Part A keeps pastes single-frame in
	115	+	// practice, so no cross-frame paste state is tracked here.
	116	+	if rel := bytes.Index(data[i:], ports.BracketedPasteCloseMarker); rel >= 0 {
	117	+	end := i + rel + len(ports.BracketedPasteCloseMarker)
	118	+	buf = append(buf, data[i:end]...)
	119	+	i = end
	120	+	continue
	121	+	}
	122	+	}
107	123		if action, size, ok, partial := altArrowCSI(remaining); ok {
108	124		flush()
109	125		r.h.Action(action)


@@ -3,6 +3,7 @@
3	package keys	3		package keys
4		4
5	import (	5		import (
		6	+	"bytes"
6	"slices"	7		"slices"
7	"sync"	8		"sync"
8	"time"	9		"time"

@@ -104,6 +105,21 @@ func (r *Router) route(data []byte) {
104	return	105		return
105	}	106		}
106	remaining := data[i+1:]	107		remaining := data[i+1:]
		108	+	if bytes.HasPrefix(data[i:], ports.BracketedPasteOpenMarker) {
		109	+	// A bracketed paste whose closing marker is in this same frame is
		110	+	// forwarded verbatim: pasted content bytes (including embedded
		111	+	// ESC+letter Alt lookalikes or ESC [ 1 ; 3 A sequences) must never
		112	+	// fire an Action mid-paste. Without a closing marker in-frame we
		113	+	// fall through to today's per-byte routing (the '[' passes through
		114	+	// as a control prefix); Part A keeps pastes single-frame in
		115	+	// practice, so no cross-frame paste state is tracked here.
		116	+	if rel := bytes.Index(data[i:], ports.BracketedPasteCloseMarker); rel >= 0 {
		117	+	end := i + rel + len(ports.BracketedPasteCloseMarker)
		118	+	buf = append(buf, data[i:end]...)
		119	+	i = end
		120	+	continue
		121	+	}
		122	+	}
107	if action, size, ok, partial := altArrowCSI(remaining); ok {	123		if action, size, ok, partial := altArrowCSI(remaining); ok {
108	flush()	124		flush()
109	r.h.Action(action)	125		r.h.Action(action)

M internal/usecase/keys/router_test.go +48 -0


@@ -496,3 +496,51 @@ func TestRouterFlushesPendingESCBeforeOtherByte(t *testing.T) {
496	496		require.Equal(t, [][]byte{{ESC}, []byte("z")}, h.forwards)
497	497		require.Empty(t, h.actions)
498	498		}
	499	+
	500	+	// pasteBindingLookalikes are content bytes that, seen outside a paste, each map
	501	+	// to a binding: Alt-j, Alt-space, Alt-1, and an Alt-arrow CSI.
	502	+	var pasteBindingLookalikes = []byte("\x1bj\x1b \x1b1\x1b[1;3A")
	503	+
	504	+	func TestRouterForwardsSingleFramePasteVerbatim(t *testing.T) {
	505	+	clk := &fakeClock{}
	506	+	h := &captureHandler{}
	507	+	r := NewRouter(clk, h)
	508	+
	509	+	paste := append(append(append([]byte(nil), ports.BracketedPasteOpenMarker...), pasteBindingLookalikes...), ports.BracketedPasteCloseMarker...)
	510	+	r.Route(paste)
	511	+
	512	+	require.Empty(t, h.actions, "paste content must not fire any binding")
	513	+	require.Equal(t, [][]byte{paste}, h.forwards, "the whole paste must be forwarded byte-identical")
	514	+	require.Empty(t, clk.timers, "a fully bracketed paste retains no trailing ESC")
	515	+	}
	516	+
	517	+	func TestRouterFiresBindingsForSameBytesOutsidePaste(t *testing.T) {
	518	+	clk := &fakeClock{}
	519	+	h := &captureHandler{}
	520	+	r := NewRouter(clk, h)
	521	+
	522	+	r.Route(pasteBindingLookalikes)
	523	+
	524	+	require.Equal(t, []Action{
	525	+	ActionFocusPaneDown, // Alt-j
	526	+	ActionOpenPalette, // Alt-space
	527	+	ActionSwitchTab1, // Alt-1
	528	+	ActionFocusPaneUp, // Alt-arrow up
	529	+	}, h.actions)
	530	+	require.Empty(t, h.forwards)
	531	+	}
	532	+
	533	+	func TestRouterPasteMarkerSplitAcrossFramesKeepsCurrentBehavior(t *testing.T) {
	534	+	clk := &fakeClock{}
	535	+	h := &captureHandler{}
	536	+	r := NewRouter(clk, h)
	537	+
	538	+	// Opening marker + content in one frame, closing marker in the next: with
	539	+	// no closing marker in-frame the router keeps its historical per-byte
	540	+	// routing, forwarding each frame's bytes as they arrive.
	541	+	r.Route([]byte("\x1b[200~hello"))
	542	+	r.Route([]byte("\x1b[201~"))
	543	+
	544	+	require.Empty(t, h.actions)
	545	+	require.Equal(t, [][]byte{[]byte("\x1b[200~hello"), []byte("\x1b[201~")}, h.forwards)
	546	+	}


@@ -496,3 +496,51 @@ func TestRouterFlushesPendingESCBeforeOtherByte(t *testing.T) {
496	require.Equal(t, [][]byte{{ESC}, []byte("z")}, h.forwards)	496		require.Equal(t, [][]byte{{ESC}, []byte("z")}, h.forwards)
497	require.Empty(t, h.actions)	497		require.Empty(t, h.actions)
498	}	498		}
		499	+
		500	+	// pasteBindingLookalikes are content bytes that, seen outside a paste, each map
		501	+	// to a binding: Alt-j, Alt-space, Alt-1, and an Alt-arrow CSI.
		502	+	var pasteBindingLookalikes = []byte("\x1bj\x1b \x1b1\x1b[1;3A")
		503	+
		504	+	func TestRouterForwardsSingleFramePasteVerbatim(t *testing.T) {
		505	+	clk := &fakeClock{}
		506	+	h := &captureHandler{}
		507	+	r := NewRouter(clk, h)
		508	+
		509	+	paste := append(append(append([]byte(nil), ports.BracketedPasteOpenMarker...), pasteBindingLookalikes...), ports.BracketedPasteCloseMarker...)
		510	+	r.Route(paste)
		511	+
		512	+	require.Empty(t, h.actions, "paste content must not fire any binding")
		513	+	require.Equal(t, [][]byte{paste}, h.forwards, "the whole paste must be forwarded byte-identical")
		514	+	require.Empty(t, clk.timers, "a fully bracketed paste retains no trailing ESC")
		515	+	}
		516	+
		517	+	func TestRouterFiresBindingsForSameBytesOutsidePaste(t *testing.T) {
		518	+	clk := &fakeClock{}
		519	+	h := &captureHandler{}
		520	+	r := NewRouter(clk, h)
		521	+
		522	+	r.Route(pasteBindingLookalikes)
		523	+
		524	+	require.Equal(t, []Action{
		525	+	ActionFocusPaneDown, // Alt-j
		526	+	ActionOpenPalette, // Alt-space
		527	+	ActionSwitchTab1, // Alt-1
		528	+	ActionFocusPaneUp, // Alt-arrow up
		529	+	}, h.actions)
		530	+	require.Empty(t, h.forwards)
		531	+	}
		532	+
		533	+	func TestRouterPasteMarkerSplitAcrossFramesKeepsCurrentBehavior(t *testing.T) {
		534	+	clk := &fakeClock{}
		535	+	h := &captureHandler{}
		536	+	r := NewRouter(clk, h)
		537	+
		538	+	// Opening marker + content in one frame, closing marker in the next: with
		539	+	// no closing marker in-frame the router keeps its historical per-byte
		540	+	// routing, forwarding each frame's bytes as they arrive.
		541	+	r.Route([]byte("\x1b[200~hello"))
		542	+	r.Route([]byte("\x1b[201~"))
		543	+
		544	+	require.Empty(t, h.actions)
		545	+	require.Equal(t, [][]byte{[]byte("\x1b[200~hello"), []byte("\x1b[201~")}, h.forwards)
		546	+	}