f67cb47 Merge pull request 'fix: v0.7.0 review findings' (#12) from fix/v0.7.0-polish into main · vev

M internal/protocol/routes.go +5 -3


@@ -505,9 +505,11 @@ func (a RouteCreateSessionAction) Validate() error {
505	505		return nil
506	506		}
507	507
508		-	// SamePeerSwitchRequest confirms a daemon-offered endpoint-empty target. It
509		-	// carries only the exact lifecycle identity and the client-owned tab cursor;
510		-	// transport origin remains proven by the existing authenticated connection.
	508	+	// SamePeerSwitchRequest moves the attachment to another session of the same
	509	+	// daemon. The client sends it from the picker or to confirm a daemon offer; a
	510	+	// pending offer pins the only accepted target. It carries only the exact
	511	+	// lifecycle identity and the client-owned tab cursor; transport origin remains
	512	+	// proven by the existing authenticated connection.
511	513		type SamePeerSwitchRequest struct {
512	514		RequestID uint64
513	515		Target ExactSessionTarget

 	return nil
 }
-// SamePeerSwitchRequest confirms a daemon-offered endpoint-empty target. It
+// SamePeerSwitchRequest moves the attachment to another session of the same
-// carries only the exact lifecycle identity and the client-owned tab cursor;
+// daemon. The client sends it from the picker or to confirm a daemon offer; a
-// transport origin remains proven by the existing authenticated connection.
+// pending offer pins the only accepted target. It carries only the exact
+// lifecycle identity and the client-owned tab cursor; transport origin remains
+// proven by the existing authenticated connection.
 type SamePeerSwitchRequest struct {
 	RequestID      uint64
 	Target         ExactSessionTarget

M internal/protocol/session.go +6 -0


@@ -229,6 +229,12 @@ func (c ViewContext) Validate() error {
229	229		return domain.ValidatePaneStableID(c.FocusedPaneID)
230	230		}
231	231
	232	+	// EchoAckDelay is how long sequenced input must have been applied before the
	233	+	// daemon acknowledges it in Output.Echo. Like mosh's ECHO_TIMEOUT, it gives the
	234	+	// PTY time to echo the input, so the screen published with the acknowledgement
	235	+	// already contains that echo. Clients remove it from round-trip samples.
	236	+	const EchoAckDelay = 50 * time.Millisecond
	237	+
232	238		type Output struct {
233	239		Epoch uint64
234	240		Base uint64


@@ -229,6 +229,12 @@ func (c ViewContext) Validate() error {
229	return domain.ValidatePaneStableID(c.FocusedPaneID)	229		return domain.ValidatePaneStableID(c.FocusedPaneID)
230	}	230		}
231		231
		232	+	// EchoAckDelay is how long sequenced input must have been applied before the
		233	+	// daemon acknowledges it in Output.Echo. Like mosh's ECHO_TIMEOUT, it gives the
		234	+	// PTY time to echo the input, so the screen published with the acknowledgement
		235	+	// already contains that echo. Clients remove it from round-trip samples.
		236	+	const EchoAckDelay = 50 * time.Millisecond
		237	+
232	type Output struct {	238		type Output struct {
233	Epoch uint64	239		Epoch uint64
234	Base uint64	240		Base uint64

M internal/snapshotcodec/codec.go +4 -0


@@ -34,6 +34,10 @@ const (
34	34		maxSnapshotDecodedAllocation = 256 << 20
35	35		)
36	36
	37	+	// MaxGenerationPayloadBytes bounds the inflated object payloads of one
	38	+	// incremental generation, the same budget an uncompressed checkpoint had.
	39	+	const MaxGenerationPayloadBytes = 256 << 20
	40	+
37	41		// Unmarshal first measures the complete manifest and its VT blobs without
38	42		// allocating. After accepting aggregate budgets, it validates references and
39	43		// blob semantics before decoding the session.


@@ -34,6 +34,10 @@ const (
34	maxSnapshotDecodedAllocation = 256 << 20	34		maxSnapshotDecodedAllocation = 256 << 20
35	)	35		)
36		36
		37	+	// MaxGenerationPayloadBytes bounds the inflated object payloads of one
		38	+	// incremental generation, the same budget an uncompressed checkpoint had.
		39	+	const MaxGenerationPayloadBytes = 256 << 20
		40	+
37	// Unmarshal first measures the complete manifest and its VT blobs without	41		// Unmarshal first measures the complete manifest and its VT blobs without
38	// allocating. After accepting aggregate budgets, it validates references and	42		// allocating. After accepting aggregate budgets, it validates references and
39	// blob semantics before decoding the session.	43		// blob semantics before decoding the session.

M internal/snapshotcodec/object_compress.go +16 -3


@@ -12,7 +12,12 @@ import (
12	12		// Object payloads are stored zlib-compressed. History cells are highly
13	13		// repetitive, so a sealed 10k-row history shrinks several times on disk.
14	14		// BestSpeed keeps snapshot encoding cheap; Go's deflate output is
15		-	// deterministic for one level, so content addressing stays stable.
	15	+	// deterministic for one toolchain and level, so content addressing stays
	16	+	// stable between checkpoints of one build.
	17	+
	18	+	// inflateInitialCap bounds the up-front allocation for one payload, so a
	19	+	// corrupt declared length cannot allocate more than the stream produces.
	20	+	const inflateInitialCap = 1 << 20
16	21
17	22		var objectCompressorPool = sync.Pool{New: func() any {
18	23		w, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)

@@ -53,10 +58,18 @@ func inflateObjectPayload(compressed []byte, size uint32) ([]byte, error) {
53	58		return nil, invalid(err)
54	59		}
55	60		defer objectDecompressorPool.Put(r)
56		-	payload := make([]byte, size)
57		-	if _, err := io.ReadFull(r, payload); err != nil {
	61	+	out := bytes.NewBuffer(make([]byte, 0, min(int(size), inflateInitialCap)))
	62	+	if _, err := io.Copy(out, io.LimitReader(r, int64(size))); err != nil {
58	63		return nil, invalid(err)
59	64		}
	65	+	if out.Len() != int(size) {
	66	+	return nil, invalid(io.ErrUnexpectedEOF)
	67	+	}
	68	+	payload := out.Bytes()
	69	+	if cap(payload) > len(payload)+len(payload)/4 {
	70	+	// Restored payloads are retained; drop the growth slack.
	71	+	payload = bytes.Clone(payload)
	72	+	}
60	73		var extra [1]byte
61	74		if n, err := r.Read(extra[:]); n != 0 \|\| err != io.EOF {
62	75		if err == nil \|\| err == io.EOF {

 // Object payloads are stored zlib-compressed. History cells are highly
 // repetitive, so a sealed 10k-row history shrinks several times on disk.
 // BestSpeed keeps snapshot encoding cheap; Go's deflate output is
-// deterministic for one level, so content addressing stays stable.
+// deterministic for one toolchain and level, so content addressing stays
+// stable between checkpoints of one build.
+// inflateInitialCap bounds the up-front allocation for one payload, so a
+// corrupt declared length cannot allocate more than the stream produces.
+const inflateInitialCap = 1 << 20
 var objectCompressorPool = sync.Pool{New: func() any {
 	w, err := zlib.NewWriterLevel(io.Discard, zlib.BestSpeed)
 		return nil, invalid(err)
 	}
 	defer objectDecompressorPool.Put(r)
-	payload := make([]byte, size)
+	out := bytes.NewBuffer(make([]byte, 0, min(int(size), inflateInitialCap)))
-	if _, err := io.ReadFull(r, payload); err != nil {
+	if _, err := io.Copy(out, io.LimitReader(r, int64(size))); err != nil {
 		return nil, invalid(err)
 	}
+	if out.Len() != int(size) {
+		return nil, invalid(io.ErrUnexpectedEOF)
+	}
+	payload := out.Bytes()
+	if cap(payload) > len(payload)+len(payload)/4 {
+		// Restored payloads are retained; drop the growth slack.
+		payload = bytes.Clone(payload)
+	}
 	var extra [1]byte
 	if n, err := r.Read(extra[:]); n != 0 || err != io.EOF {
 		if err == nil || err == io.EOF {

M internal/snapshotcodec/object_compress_test.go +4 -0


@@ -46,6 +46,10 @@ func TestObjectCompressionRejectsMalformedPayloads(t *testing.T) {
46	46		binary.BigEndian.PutUint32(b[1:5], uint32(len(payload)+1))
47	47		return b
48	48		}, want: ErrInvalidData},
	49	+	{name: "declared maximum on a small stream", mutate: func(b []byte) []byte {
	50	+	binary.BigEndian.PutUint32(b[1:5], maxObjectPayloadSize)
	51	+	return b
	52	+	}, want: ErrInvalidData},
49	53		{name: "truncated stream", mutate: func(b []byte) []byte { return b[:len(b)-1] }, want: ErrInvalidData},
50	54		{name: "corrupt stream", mutate: func(b []byte) []byte { b[len(b)-6] ^= 0xff; return b }, want: ErrInvalidData},
51	55		{name: "trailing garbage", mutate: func(b []byte) []byte { return append(b, 0) }, want: ErrTrailingBytes},


@@ -46,6 +46,10 @@ func TestObjectCompressionRejectsMalformedPayloads(t *testing.T) {
46	binary.BigEndian.PutUint32(b[1:5], uint32(len(payload)+1))	46		binary.BigEndian.PutUint32(b[1:5], uint32(len(payload)+1))
47	return b	47		return b
48	}, want: ErrInvalidData},	48		}, want: ErrInvalidData},
		49	+	{name: "declared maximum on a small stream", mutate: func(b []byte) []byte {
		50	+	binary.BigEndian.PutUint32(b[1:5], maxObjectPayloadSize)
		51	+	return b
		52	+	}, want: ErrInvalidData},
49	{name: "truncated stream", mutate: func(b []byte) []byte { return b[:len(b)-1] }, want: ErrInvalidData},	53		{name: "truncated stream", mutate: func(b []byte) []byte { return b[:len(b)-1] }, want: ErrInvalidData},
50	{name: "corrupt stream", mutate: func(b []byte) []byte { b[len(b)-6] ^= 0xff; return b }, want: ErrInvalidData},	54		{name: "corrupt stream", mutate: func(b []byte) []byte { b[len(b)-6] ^= 0xff; return b }, want: ErrInvalidData},
51	{name: "trailing garbage", mutate: func(b []byte) []byte { return append(b, 0) }, want: ErrTrailingBytes},	55		{name: "trailing garbage", mutate: func(b []byte) []byte { return append(b, 0) }, want: ErrTrailingBytes},

M internal/usecase/client/attachment_echo.go +16 -0


@@ -53,6 +53,22 @@ func (e *attachmentEcho) input(seq uint64, data []byte) error {
53	53		return e.draw()
54	54		}
55	55
	56	+	// beforeOutput erases drawn guesses before the foreground writes a daemon
	57	+	// frame, so the frame never scrolls or overwrites around a guess.
	58	+	func (e *attachmentEcho) beforeOutput(output protocol.Output) error {
	59	+	if e == nil \|\| len(output.Data) == 0 {
	60	+	return nil
	61	+	}
	62	+	data := e.predictor.undraw()
	63	+	if len(data) == 0 {
	64	+	return nil
	65	+	}
	66	+	if _, err := e.out.writePrediction(data); err != nil {
	67	+	return fmt.Errorf("erasing predicted echo: %w", err)
	68	+	}
	69	+	return nil
	70	+	}
	71	+
56	72		// output mirrors one Output after the foreground wrote it.
57	73		func (e *attachmentEcho) output(output protocol.Output) error {
58	74		if e == nil {


@@ -53,6 +53,22 @@ func (e *attachmentEcho) input(seq uint64, data []byte) error {
53	return e.draw()	53		return e.draw()
54	}	54		}
55		55
		56	+	// beforeOutput erases drawn guesses before the foreground writes a daemon
		57	+	// frame, so the frame never scrolls or overwrites around a guess.
		58	+	func (e *attachmentEcho) beforeOutput(output protocol.Output) error {
		59	+	if e == nil \|\| len(output.Data) == 0 {
		60	+	return nil
		61	+	}
		62	+	data := e.predictor.undraw()
		63	+	if len(data) == 0 {
		64	+	return nil
		65	+	}
		66	+	if _, err := e.out.writePrediction(data); err != nil {
		67	+	return fmt.Errorf("erasing predicted echo: %w", err)
		68	+	}
		69	+	return nil
		70	+	}
		71	+
56	// output mirrors one Output after the foreground wrote it.	72		// output mirrors one Output after the foreground wrote it.
57	func (e *attachmentEcho) output(output protocol.Output) error {	73		func (e *attachmentEcho) output(output protocol.Output) error {
58	if e == nil {	74		if e == nil {

M internal/usecase/client/attachment_outbox_test.go +11 -4


@@ -126,7 +126,7 @@ func TestResumeWatch(t *testing.T) {
126	126		})
127	127		lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
128	128		clock := newSupervisorTestClock()
129		-	watch := lifetime.watchResume(clock)
	129	+	watch := lifetime.watchResume(clock, nil)
130	130		for _, read := range tt.reads {
131	131		reader.chunks <- []byte(read)
132	132		}

@@ -187,7 +187,7 @@ func TestResumeWatchKeepsBacklogVerbatim(t *testing.T) {
187	187		}, 5*time.Second, time.Millisecond)
188	188		}
189	189		lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
190		-	watch := lifetime.watchResume(newSupervisorTestClock())
	190	+	watch := lifetime.watchResume(newSupervisorTestClock(), nil)
191	191		require.Eventually(t, func() bool { return watch.heldLen() == len(tt.residual)+len(tt.pending) }, 5*time.Second, time.Millisecond)
192	192		if tt.typed != "" {
193	193		reader.chunks <- []byte(tt.typed)

@@ -234,7 +234,7 @@ func TestResumeWatchResolvesEscapeAtRelease(t *testing.T) {
234	234		})
235	235		lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
236	236		clock := newSupervisorTestClock()
237		-	watch := lifetime.watchResume(clock)
	237	+	watch := lifetime.watchResume(clock, nil)
238	238		reader.chunks <- []byte("\x1b")
239	239		escape := clock.awaitTimer(t)
240	240		released := make(chan struct{})

@@ -268,11 +268,18 @@ func TestResumeWatchResolvesEscapeAtRelease(t *testing.T) {
268	268		}
269	269
270	270		func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
271		-	w := &resumeWatch{}
	271	+	lifetime := &terminalInputLifetime{}
	272	+	w := &resumeWatch{lifetime: lifetime}
272	273		w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
273	274		w.hold([]byte("bb"))
274	275		w.hold([]byte("c"))
275	276		require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
	277	+
	278	+	// The next resume phase of the same outage starts a new watch; it must
	279	+	// keep dropping so the session never receives input with a gap.
	280	+	next := &resumeWatch{lifetime: lifetime}
	281	+	next.hold([]byte("d"))
	282	+	require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")
276	283		}
277	284
278	285		// stallingSessionStream passes writes through until stall is set; stalled

 			})
 			lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
 			clock := newSupervisorTestClock()
-			watch := lifetime.watchResume(clock)
+			watch := lifetime.watchResume(clock, nil)
 			for _, read := range tt.reads {
 				reader.chunks <- []byte(read)
 			}
 				}, 5*time.Second, time.Millisecond)
 			}
 			lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
-			watch := lifetime.watchResume(newSupervisorTestClock())
+			watch := lifetime.watchResume(newSupervisorTestClock(), nil)
 			require.Eventually(t, func() bool { return watch.heldLen() == len(tt.residual)+len(tt.pending) }, 5*time.Second, time.Millisecond)
 			if tt.typed != "" {
 				reader.chunks <- []byte(tt.typed)
 			})
 			lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
 			clock := newSupervisorTestClock()
-			watch := lifetime.watchResume(clock)
+			watch := lifetime.watchResume(clock, nil)
 			reader.chunks <- []byte("\x1b")
 			escape := clock.awaitTimer(t)
 			released := make(chan struct{})
 }
 func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
-	w := &resumeWatch{}
+	lifetime := &terminalInputLifetime{}
+	w := &resumeWatch{lifetime: lifetime}
 	w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))
 	w.hold([]byte("bb"))
 	w.hold([]byte("c"))
 	require.Equal(t, resumeHeldInputLimit-1, w.heldLen(), "keys after an overflow must be dropped too")
+	// The next resume phase of the same outage starts a new watch; it must
+	// keep dropping so the session never receives input with a gap.
+	next := &resumeWatch{lifetime: lifetime}
+	next.hold([]byte("d"))
+	require.Zero(t, next.heldLen(), "a later phase must not accept keys after an overflow")
 }
 // stallingSessionStream passes writes through until stall is set; stalled

M internal/usecase/client/attachment_session.go +3 -0


@@ -542,6 +542,9 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
542	542		}
543	543		}
544	544		w.noteCommitted(fg, state.context)
	545	+	if err := echo.beforeOutput(typed); err != nil {
	546	+	return w.settle(ctx, fg, stream, token, err)
	547	+	}
545	548		if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil {
546	549		return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing output: %w", err)}
547	550		}


@@ -542,6 +542,9 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
542	}	542		}
543	}	543		}
544	w.noteCommitted(fg, state.context)	544		w.noteCommitted(fg, state.context)
		545	+	if err := echo.beforeOutput(typed); err != nil {
		546	+	return w.settle(ctx, fg, stream, token, err)
		547	+	}
545	if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil {	548		if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil {
546	return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing output: %w", err)}	549		return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing output: %w", err)}
547	}	550		}

M internal/usecase/client/predictive_echo.go +54 -20


@@ -47,10 +47,9 @@ const (
47	47		echoGlitchRepairCount = 10
48	48		echoGlitchRepairMinInterval = 150 * time.Millisecond
49	49
50		-	// echoServerDelay is the daemon's echo acknowledgement delay
51		-	// (usecase/daemon echoAckDelay). It is removed from each round-trip
52		-	// sample so the thresholds compare network time, as mosh's SRTT does.
53		-	echoServerDelay = 50 * time.Millisecond
	50	+	// echoServerDelay is removed from each round-trip sample so the
	51	+	// thresholds compare network time, as mosh's SRTT does.
	52	+	echoServerDelay = protocol.EchoAckDelay
54	53
55	54		// maxEchoSent bounds the send-time history used for round-trip samples.
56	55		maxEchoSent = 256

@@ -123,17 +122,22 @@ type echoPredictor struct {
123	122		mode domain.EchoPredictMode
124	123		screen *vt.Screen
125	124
126		-	sent []echoSent
127		-	echo uint64
128		-	srtt time.Duration
129		-	srttKnown bool
130		-	srttOn bool
131		-	flagging bool
132		-	glitch int
133		-	lastQuick time.Time
134		-	predEpoch uint64
135		-	confEpoch uint64
136		-	holdUntil uint64
	125	+	sent []echoSent
	126	+	echo uint64
	127	+	srtt time.Duration
	128	+	srttKnown bool
	129	+	srttOn bool
	130	+	flagging bool
	131	+	glitch int
	132	+	lastQuick time.Time
	133	+	predEpoch uint64
	134	+	confEpoch uint64
	135	+	holdUntil uint64
	136	+	// held is set by the first daemon frame after an unknown key (Enter)
	137	+	// that is not its acknowledgement: that frame may move the guessed row,
	138	+	// so guesses stay hidden until the key is acknowledged. Before such a
	139	+	// frame, drawn guesses stay put.
	140	+	held bool
137	141		row int
138	142		cells []echoCell
139	143		cursor echoCursor

@@ -158,17 +162,19 @@ func (p *echoPredictor) applyOutput(output protocol.Output, now time.Time) {
158	162		p.forget()
159	163		}
160	164		if output.Full {
161		-	// A full frame repainted every drawn guess.
162		-	p.drawn = p.drawn[:0]
163		-	p.cursorMove = false
	165	+	// A full frame repaints everything: guesses placed against the old
	166	+	// frame would land on unrelated content.
	167	+	p.forget()
164	168		}
165		-	if data := p.apc.strip(output.Data); len(data) > 0 {
	169	+	data := p.apc.strip(output.Data)
	170	+	if len(data) > 0 {
166	171		p.screen.Write(data)
167	172		}
168	173		if output.Echo > p.echo {
169	174		p.sample(output.Echo, now)
170	175		p.echo = output.Echo
171	176		}
	177	+	p.held = p.echo < p.holdUntil && (p.held \|\| len(data) > 0)
172	178		p.cull(now)
173	179		}
174	180

@@ -608,6 +614,32 @@ func (p *echoPredictor) needsTick() bool {
608	614		return p.mode == domain.EchoPredictAdaptive && p.active() && !(p.glitch > 0 && p.flagging)
609	615		}
610	616
	617	+	// undraw returns the bytes that repaint every drawn guess from the mirror, so
	618	+	// a daemon frame lands on the screen the daemon believes it has. A frame that
	619	+	// scrolls would otherwise move a guess where nothing ever repairs it. The
	620	+	// guesses stay pending and render draws them again after the frame.
	621	+	func (p *echoPredictor) undraw() []byte {
	622	+	if len(p.drawn) == 0 {
	623	+	return nil
	624	+	}
	625	+	var out bytes.Buffer
	626	+	if p.drawnRow < p.screen.Rows() {
	627	+	for _, col := range p.drawn {
	628	+	if col < p.screen.Columns() {
	629	+	p.restoreCell(&out, p.drawnRow, col)
	630	+	}
	631	+	}
	632	+	}
	633	+	p.drawn = p.drawn[:0]
	634	+	if out.Len() > 0 \|\| p.cursorMove {
	635	+	// The daemon frame is relative to its own cursor, not the predicted one.
	636	+	out.WriteString("\x1b[0m")
	637	+	writeEchoCUP(&out, p.screen.CursorRow(), p.screen.CursorCol())
	638	+	p.cursorMove = false
	639	+	}
	640	+	return out.Bytes()
	641	+	}
	642	+
611	643		// render returns the terminal bytes that bring the screen from the last
612	644		// render to the current guesses: drawn cells that no longer show a guess are
613	645		// repainted from the mirror, shown guesses are drawn again (a daemon frame

@@ -617,7 +649,9 @@ func (p *echoPredictor) render() []byte {
617	649		var out bytes.Buffer
618	650		show := p.displaying() && p.enabled()
619	651		var want []int
620		-	if show && p.cells != nil && p.row < p.screen.Rows() {
	652	+	// Once a frame arrived while an unknown key (Enter) is unanswered, the
	653	+	// guesses' row may have moved, so none is drawn until the answer.
	654	+	if show && p.cells != nil && p.row < p.screen.Rows() && !(p.held && p.echo < p.holdUntil) {
621	655		for col := range p.cells {
622	656		cell := &p.cells[col]
623	657		if !cell.active \|\| cell.unknown \|\| cell.tentative(p.confEpoch) {

 	echoGlitchRepairCount       = 10
 	echoGlitchRepairMinInterval = 150 * time.Millisecond
-	// echoServerDelay is the daemon's echo acknowledgement delay
+	// echoServerDelay is removed from each round-trip sample so the
-	// (usecase/daemon echoAckDelay). It is removed from each round-trip
+	// thresholds compare network time, as mosh's SRTT does.
-	// sample so the thresholds compare network time, as mosh's SRTT does.
+	echoServerDelay = protocol.EchoAckDelay
-	echoServerDelay = 50 * time.Millisecond
 	// maxEchoSent bounds the send-time history used for round-trip samples.
 	maxEchoSent = 256
 	mode   domain.EchoPredictMode
 	screen *vt.Screen
-	sent       []echoSent
+	sent      []echoSent
-	echo       uint64
+	echo      uint64
-	srtt       time.Duration
+	srtt      time.Duration
-	srttKnown  bool
+	srttKnown bool
-	srttOn     bool
+	srttOn    bool
-	flagging   bool
+	flagging  bool
-	glitch     int
+	glitch    int
-	lastQuick  time.Time
+	lastQuick time.Time
-	predEpoch  uint64
+	predEpoch uint64
-	confEpoch  uint64
+	confEpoch uint64
-	holdUntil  uint64
+	holdUntil uint64
+	// held is set by the first daemon frame after an unknown key (Enter)
+	// that is not its acknowledgement: that frame may move the guessed row,
+	// so guesses stay hidden until the key is acknowledged. Before such a
+	// frame, drawn guesses stay put.
+	held       bool
 	row        int
 	cells      []echoCell
 	cursor     echoCursor
 		p.forget()
 	}
 	if output.Full {
-		// A full frame repainted every drawn guess.
+		// A full frame repaints everything: guesses placed against the old
-		p.drawn = p.drawn[:0]
+		// frame would land on unrelated content.
-		p.cursorMove = false
+		p.forget()
 	}
-	if data := p.apc.strip(output.Data); len(data) > 0 {
+	data := p.apc.strip(output.Data)
+	if len(data) > 0 {
 		p.screen.Write(data)
 	}
 	if output.Echo > p.echo {
 		p.sample(output.Echo, now)
 		p.echo = output.Echo
 	}
+	p.held = p.echo < p.holdUntil && (p.held || len(data) > 0)
 	p.cull(now)
 }
 	return p.mode == domain.EchoPredictAdaptive && p.active() && !(p.glitch > 0 && p.flagging)
 }
+// undraw returns the bytes that repaint every drawn guess from the mirror, so
+// a daemon frame lands on the screen the daemon believes it has. A frame that
+// scrolls would otherwise move a guess where nothing ever repairs it. The
+// guesses stay pending and render draws them again after the frame.
+func (p *echoPredictor) undraw() []byte {
+	if len(p.drawn) == 0 {
+		return nil
+	}
+	var out bytes.Buffer
+	if p.drawnRow < p.screen.Rows() {
+		for _, col := range p.drawn {
+			if col < p.screen.Columns() {
+				p.restoreCell(&out, p.drawnRow, col)
+			}
+		}
+	}
+	p.drawn = p.drawn[:0]
+	if out.Len() > 0 || p.cursorMove {
+		// The daemon frame is relative to its own cursor, not the predicted one.
+		out.WriteString("\x1b[0m")
+		writeEchoCUP(&out, p.screen.CursorRow(), p.screen.CursorCol())
+		p.cursorMove = false
+	}
+	return out.Bytes()
+}
 // render returns the terminal bytes that bring the screen from the last
 // render to the current guesses: drawn cells that no longer show a guess are
 // repainted from the mirror, shown guesses are drawn again (a daemon frame
 	var out bytes.Buffer
 	show := p.displaying() && p.enabled()
 	var want []int
-	if show && p.cells != nil && p.row < p.screen.Rows() {
+	// Once a frame arrived while an unknown key (Enter) is unanswered, the
+	// guesses' row may have moved, so none is drawn until the answer.
+	if show && p.cells != nil && p.row < p.screen.Rows() && !(p.held && p.echo < p.holdUntil) {
 		for col := range p.cells {
 			cell := &p.cells[col]
 			if !cell.active || cell.unknown || cell.tentative(p.confEpoch) {

M internal/usecase/client/predictive_echo_test.go +65 -0


@@ -90,6 +90,71 @@ func TestEchoPredictorDisplay(t *testing.T) {
90	90		}
91	91		}
92	92
	93	+	// TestEchoPredictorNeverDrawsStaleGuesses covers daemon frames that arrive
	94	+	// before the echo acknowledgement and move or replace the guessed row.
	95	+	func TestEchoPredictorNeverDrawsStaleGuesses(t *testing.T) {
	96	+	t.Parallel()
	97	+	tests := []struct {
	98	+	name string
	99	+	input string
	100	+	frame protocol.Output
	101	+	}{
	102	+	{
	103	+	name: "enter then a scrolled prompt before the ack",
	104	+	input: "b\r",
	105	+	frame: protocol.Output{Epoch: 1, New: 2, Size: domain.Size{Cols: 20, Rows: 3}, Data: []byte("b\r\nfoo\r\nbar\r\n$ ")},
	106	+	},
	107	+	{
	108	+	name: "full frame before the ack",
	109	+	input: "b",
	110	+	frame: protocol.Output{Epoch: 1, New: 2, Full: true, Size: domain.Size{Cols: 20, Rows: 3}, Data: []byte("\x1b[H\x1b[2Jother")},
	111	+	},
	112	+	}
	113	+	for _, tt := range tests {
	114	+	t.Run(tt.name, func(t *testing.T) {
	115	+	t.Parallel()
	116	+	h := newEchoHarness(t, domain.EchoPredictAlways)
	117	+	h.warm(100 * time.Millisecond)
	118	+	h.confirm()
	119	+	h.typed(tt.input)
	120	+	h.p.undraw()
	121	+	h.p.applyOutput(tt.frame, h.now)
	122	+	if got := h.p.render(); bytes.Contains(got, []byte("b")) {
	123	+	t.Fatalf("render = %q redraws a guess over the new frame", got)
	124	+	}
	125	+	})
	126	+	}
	127	+	}
	128	+
	129	+	func TestEchoPredictorEnterKeepsDrawnGuessesUntilAFrame(t *testing.T) {
	130	+	t.Parallel()
	131	+	h := newEchoHarness(t, domain.EchoPredictAlways)
	132	+	h.warm(100 * time.Millisecond)
	133	+	h.confirm()
	134	+	if got := h.typed("b"); !bytes.Contains(got, []byte("b")) {
	135	+	t.Fatalf("render = %q, want the guess drawn", got)
	136	+	}
	137	+	if got := h.typed("\r"); bytes.Contains(got, []byte(" ")) {
	138	+	t.Fatalf("render after Enter = %q erases the typed guess before any frame", got)
	139	+	}
	140	+	}
	141	+
	142	+	func TestEchoPredictorUndrawRestoresMirror(t *testing.T) {
	143	+	t.Parallel()
	144	+	h := newEchoHarness(t, domain.EchoPredictAlways)
	145	+	h.warm(100 * time.Millisecond)
	146	+	h.confirm()
	147	+	if got := h.typed("b"); !bytes.Contains(got, []byte("b")) {
	148	+	t.Fatalf("render = %q, want the guess drawn", got)
	149	+	}
	150	+	if got, want := string(h.p.undraw()), "\x1b[1;4H\x1b[0m \x1b[0m\x1b[1;4H"; got != want {
	151	+	t.Fatalf("undraw = %q, want %q", got, want)
	152	+	}
	153	+	if got := h.p.undraw(); len(got) != 0 {
	154	+	t.Fatalf("second undraw = %q, want nothing", got)
	155	+	}
	156	+	}
	157	+
93	158		func TestEchoPredictorConfirmationRepaintsFromMirror(t *testing.T) {
94	159		t.Parallel()
95	160		h := newEchoHarness(t, domain.EchoPredictAdaptive)


@@ -90,6 +90,71 @@ func TestEchoPredictorDisplay(t *testing.T) {
90	}	90		}
91	}	91		}
92		92
		93	+	// TestEchoPredictorNeverDrawsStaleGuesses covers daemon frames that arrive
		94	+	// before the echo acknowledgement and move or replace the guessed row.
		95	+	func TestEchoPredictorNeverDrawsStaleGuesses(t *testing.T) {
		96	+	t.Parallel()
		97	+	tests := []struct {
		98	+	name string
		99	+	input string
		100	+	frame protocol.Output
		101	+	}{
		102	+	{
		103	+	name: "enter then a scrolled prompt before the ack",
		104	+	input: "b\r",
		105	+	frame: protocol.Output{Epoch: 1, New: 2, Size: domain.Size{Cols: 20, Rows: 3}, Data: []byte("b\r\nfoo\r\nbar\r\n$ ")},
		106	+	},
		107	+	{
		108	+	name: "full frame before the ack",
		109	+	input: "b",
		110	+	frame: protocol.Output{Epoch: 1, New: 2, Full: true, Size: domain.Size{Cols: 20, Rows: 3}, Data: []byte("\x1b[H\x1b[2Jother")},
		111	+	},
		112	+	}
		113	+	for _, tt := range tests {
		114	+	t.Run(tt.name, func(t *testing.T) {
		115	+	t.Parallel()
		116	+	h := newEchoHarness(t, domain.EchoPredictAlways)
		117	+	h.warm(100 * time.Millisecond)
		118	+	h.confirm()
		119	+	h.typed(tt.input)
		120	+	h.p.undraw()
		121	+	h.p.applyOutput(tt.frame, h.now)
		122	+	if got := h.p.render(); bytes.Contains(got, []byte("b")) {
		123	+	t.Fatalf("render = %q redraws a guess over the new frame", got)
		124	+	}
		125	+	})
		126	+	}
		127	+	}
		128	+
		129	+	func TestEchoPredictorEnterKeepsDrawnGuessesUntilAFrame(t *testing.T) {
		130	+	t.Parallel()
		131	+	h := newEchoHarness(t, domain.EchoPredictAlways)
		132	+	h.warm(100 * time.Millisecond)
		133	+	h.confirm()
		134	+	if got := h.typed("b"); !bytes.Contains(got, []byte("b")) {
		135	+	t.Fatalf("render = %q, want the guess drawn", got)
		136	+	}
		137	+	if got := h.typed("\r"); bytes.Contains(got, []byte(" ")) {
		138	+	t.Fatalf("render after Enter = %q erases the typed guess before any frame", got)
		139	+	}
		140	+	}
		141	+
		142	+	func TestEchoPredictorUndrawRestoresMirror(t *testing.T) {
		143	+	t.Parallel()
		144	+	h := newEchoHarness(t, domain.EchoPredictAlways)
		145	+	h.warm(100 * time.Millisecond)
		146	+	h.confirm()
		147	+	if got := h.typed("b"); !bytes.Contains(got, []byte("b")) {
		148	+	t.Fatalf("render = %q, want the guess drawn", got)
		149	+	}
		150	+	if got, want := string(h.p.undraw()), "\x1b[1;4H\x1b[0m \x1b[0m\x1b[1;4H"; got != want {
		151	+	t.Fatalf("undraw = %q, want %q", got, want)
		152	+	}
		153	+	if got := h.p.undraw(); len(got) != 0 {
		154	+	t.Fatalf("second undraw = %q, want nothing", got)
		155	+	}
		156	+	}
		157	+
93	func TestEchoPredictorConfirmationRepaintsFromMirror(t *testing.T) {	158		func TestEchoPredictorConfirmationRepaintsFromMirror(t *testing.T) {
94	t.Parallel()	159		t.Parallel()
95	h := newEchoHarness(t, domain.EchoPredictAdaptive)	160		h := newEchoHarness(t, domain.EchoPredictAdaptive)

M internal/usecase/client/supervisor.go +22 -11


@@ -1387,6 +1387,10 @@ type terminalInputLifetime struct {
1387	1387
1388	1388		mu sync.Mutex
1389	1389		pickerID uint64
	1390	+	// resumeOverflowed spans every watch of one outage: once held keys
	1391	+	// overflowed, later phases drop keys too, so the resumed session never
	1392	+	// receives input with a gap. The picker taking input clears it.
	1393	+	resumeOverflowed atomic.Bool
1390	1394		}
1391	1395
1392	1396		// startTerminalInputLifetime starts the single terminal read. A nil reader is

@@ -1436,6 +1440,7 @@ func (l *terminalInputLifetime) acquirePicker() {
1436	1440		}
1437	1441		if id, ok := l.pump.tryClaim(); ok {
1438	1442		l.pickerID = id
	1443	+	l.resumeOverflowed.Store(false)
1439	1444		l.mu.Unlock()
1440	1445		return
1441	1446		}

@@ -1514,9 +1519,8 @@ func (l *terminalInputLifetime) stop() {
1514	1519		}
1515	1520
1516	1521		// resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
1517		-	// Past it further keys are dropped, like a full terminal buffer.
1518		-	// TODO: the overflow flag is per resume watch; a later phase can accept keys
1519		-	// again and leave a gap. Make it sticky on the input lifetime until attach.
	1522	+	// Past it further keys are dropped, like a full terminal buffer, until the
	1523	+	// picker takes input again.
1520	1524		const resumeHeldInputLimit = 64 << 10
1521	1525
1522	1526		// resumeWatch reads the terminal while a lost attachment waits to resume.

@@ -1529,6 +1533,7 @@ type resumeWatch struct {
1529	1533		lifetime *terminalInputLifetime
1530	1534		id uint64
1531	1535		clock ports.Clock
	1536	+	logger *slog.Logger
1532	1537
1533	1538		// backlog counts deliveries already waiting at claim: session input the
1534	1539		// lost attachment kept, held verbatim and never read as a cancel key.

@@ -1537,9 +1542,6 @@ type resumeWatch struct {
1537	1542
1538	1543		mu sync.Mutex
1539	1544		held []byte
1540		-	// overflowed drops every key after the first one past the bound, so the
1541		-	// resumed session never receives input with a gap in the middle.
1542		-	overflowed bool
1543	1545		}
1544	1546
1545	1547		// heldLen reports how many kept bytes the watch holds.

@@ -1553,8 +1555,11 @@ func (w *resumeWatch) heldLen() int {
1553	1555		func (w *resumeWatch) hold(data []byte) {
1554	1556		w.mu.Lock()
1555	1557		defer w.mu.Unlock()
1556		-	if w.overflowed \|\| len(w.held)+len(data) > resumeHeldInputLimit {
1557		-	w.overflowed = w.overflowed \|\| len(data) != 0
	1558	+	overflowed := &w.lifetime.resumeOverflowed
	1559	+	if overflowed.Load() \|\| len(w.held)+len(data) > resumeHeldInputLimit {
	1560	+	if len(data) != 0 {
	1561	+	overflowed.Store(true)
	1562	+	}
1558	1563		return
1559	1564		}
1560	1565		w.held = append(w.held, data...)

@@ -1562,8 +1567,8 @@ func (w *resumeWatch) hold(data []byte) {
1562	1567
1563	1568		// watchResume claims terminal input for one resume backoff. When the claim
1564	1569		// is unavailable it returns a watch that never cancels.
1565		-	func (l terminalInputLifetime) watchResume(clock ports.Clock) resumeWatch {
1566		-	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock}
	1570	+	func (l terminalInputLifetime) watchResume(clock ports.Clock, logger slog.Logger) *resumeWatch {
	1571	+	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock, logger: logger}
1567	1572		if l == nil \|\| l.pump == nil {
1568	1573		close(w.done)
1569	1574		return w

@@ -1672,7 +1677,13 @@ func (w *resumeWatch) releaseClaim() {
1672	1677		case <-w.cancelled:
1673	1678		default:
1674	1679		w.mu.Lock()
1675		-	pump.appendResidual(w.id, w.held, resumeHeldInputLimit)
	1680	+	if !pump.appendResidual(w.id, w.held, resumeHeldInputLimit) && len(w.held) != 0 {
	1681	+	// Replaying only part of the held keys would leave a gap.
	1682	+	w.lifetime.resumeOverflowed.Store(true)
	1683	+	if w.logger != nil {
	1684	+	w.logger.Warn("client_resume_input_dropped", "bytes", len(w.held))
	1685	+	}
	1686	+	}
1676	1687		w.mu.Unlock()
1677	1688		}
1678	1689		pump.revoke(w.id)

 	mu       sync.Mutex
 	pickerID uint64
+	// resumeOverflowed spans every watch of one outage: once held keys
+	// overflowed, later phases drop keys too, so the resumed session never
+	// receives input with a gap. The picker taking input clears it.
+	resumeOverflowed atomic.Bool
 }
 // startTerminalInputLifetime starts the single terminal read. A nil reader is
 	}
 	if id, ok := l.pump.tryClaim(); ok {
 		l.pickerID = id
+		l.resumeOverflowed.Store(false)
 		l.mu.Unlock()
 		return
 	}
 }
 // resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
-// Past it further keys are dropped, like a full terminal buffer.
+// Past it further keys are dropped, like a full terminal buffer, until the
-// TODO: the overflow flag is per resume watch; a later phase can accept keys
+// picker takes input again.
-// again and leave a gap. Make it sticky on the input lifetime until attach.
 const resumeHeldInputLimit = 64 << 10
 // resumeWatch reads the terminal while a lost attachment waits to resume.
 	lifetime  *terminalInputLifetime
 	id        uint64
 	clock     ports.Clock
+	logger    *slog.Logger
 	// backlog counts deliveries already waiting at claim: session input the
 	// lost attachment kept, held verbatim and never read as a cancel key.
 	mu   sync.Mutex
 	held []byte
-	// overflowed drops every key after the first one past the bound, so the
-	// resumed session never receives input with a gap in the middle.
-	overflowed bool
 }
 // heldLen reports how many kept bytes the watch holds.
 func (w *resumeWatch) hold(data []byte) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	if w.overflowed || len(w.held)+len(data) > resumeHeldInputLimit {
+	overflowed := &w.lifetime.resumeOverflowed
-		w.overflowed = w.overflowed || len(data) != 0
+	if overflowed.Load() || len(w.held)+len(data) > resumeHeldInputLimit {
+		if len(data) != 0 {
+			overflowed.Store(true)
+		}
 		return
 	}
 	w.held = append(w.held, data...)
 // watchResume claims terminal input for one resume backoff. When the claim
 // is unavailable it returns a watch that never cancels.
-func (l *terminalInputLifetime) watchResume(clock ports.Clock) *resumeWatch {
+func (l *terminalInputLifetime) watchResume(clock ports.Clock, logger *slog.Logger) *resumeWatch {
-	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock}
+	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l, clock: clock, logger: logger}
 	if l == nil || l.pump == nil {
 		close(w.done)
 		return w
 	case <-w.cancelled:
 	default:
 		w.mu.Lock()
-		pump.appendResidual(w.id, w.held, resumeHeldInputLimit)
+		if !pump.appendResidual(w.id, w.held, resumeHeldInputLimit) && len(w.held) != 0 {
+			// Replaying only part of the held keys would leave a gap.
+			w.lifetime.resumeOverflowed.Store(true)
+			if w.logger != nil {
+				w.logger.Warn("client_resume_input_dropped", "bytes", len(w.held))
+			}
+		}
 		w.mu.Unlock()
 	}
 	pump.revoke(w.id)

M internal/usecase/client/supervisor_attach.go +7 -2


@@ -744,7 +744,7 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
744	744		var cancelOpen context.CancelFunc
745	745		var watchDone chan struct{}
746	746		if resuming {
747		-	keys = input.watchResume(s.cfg.Clock)
	747	+	keys = input.watchResume(s.cfg.Clock, s.logger)
748	748		openCtx, cancelOpen = context.WithCancel(openCtx)
749	749		watchDone = make(chan struct{})
750	750		go func() {

@@ -814,6 +814,11 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
814	814		s.transition(supervisorEvent{kind: supervisorAttachEnded})
815	815		return attachmentEndedOutcome()
816	816		}
	817	+	if input != nil {
	818	+	// The attachment owns input again: keys typed from now on reach the
	819	+	// session, so the outage's overflow ends here, not at stream open.
	820	+	input.resumeOverflowed.Store(false)
	821	+	}
817	822		return s.settleAttachment(ctx, input, service, deadline, run, request, resuming)
818	823		}
819	824

@@ -1043,7 +1048,7 @@ func (s Supervisor) waitResume(ctx context.Context, input terminalInputLifetim
1043	1048		delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
1044	1049		timer := s.cfg.Clock.NewTimer(delay)
1045	1050		defer stopSupervisorTimer(timer)
1046		-	keys := input.watchResume(s.cfg.Clock)
	1051	+	keys := input.watchResume(s.cfg.Clock, s.logger)
1047	1052		defer keys.release()
1048	1053		for {
1049	1054		select {

 	var cancelOpen context.CancelFunc
 	var watchDone chan struct{}
 	if resuming {
-		keys = input.watchResume(s.cfg.Clock)
+		keys = input.watchResume(s.cfg.Clock, s.logger)
 		openCtx, cancelOpen = context.WithCancel(openCtx)
 		watchDone = make(chan struct{})
 		go func() {
 		s.transition(supervisorEvent{kind: supervisorAttachEnded})
 		return attachmentEndedOutcome()
 	}
+	if input != nil {
+		// The attachment owns input again: keys typed from now on reach the
+		// session, so the outage's overflow ends here, not at stream open.
+		input.resumeOverflowed.Store(false)
+	}
 	return s.settleAttachment(ctx, input, service, deadline, run, request, resuming)
 }
 	delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
 	timer := s.cfg.Clock.NewTimer(delay)
 	defer stopSupervisorTimer(timer)
-	keys := input.watchResume(s.cfg.Clock)
+	keys := input.watchResume(s.cfg.Clock, s.logger)
 	defer keys.release()
 	for {
 		select {

M internal/usecase/copy/mode.go +0 -42


@@ -501,48 +501,6 @@ func FindMatches(doc *Document, query string) []SearchMatch {
501	501		return findMatches(query, doc.Snapshot().rangeText)
502	502		}
503	503
504		-	// RefineMatches returns FindMatches(doc, query) given previous, the complete
505		-	// result for a query that query extends. Any row matching the longer query also
506		-	// matched the shorter one, so only rows listed in previous are scanned again.
507		-	func RefineMatches(doc *Document, previous []SearchMatch, query string) []SearchMatch {
508		-	query = strings.TrimSpace(query)
509		-	if doc == nil \|\| query == "" {
510		-	return nil
511		-	}
512		-	snapshot := doc.Snapshot()
513		-	return findMatches(query, func(yield func(int, []rune, []int, int) bool) error {
514		-	var cells []renderer.Cell
515		-	var runes []rune
516		-	var columns []int
517		-	last := -1
518		-	for _, match := range previous {
519		-	if match.Row == last {
520		-	continue
521		-	}
522		-	last = match.Row
523		-	width := snapshot.RowWidth(match.Row)
524		-	cells = slices.Grow(cells[:0], width)[:width]
525		-	cells = cells[:snapshot.CopyRow(match.Row, cells)]
526		-	runes, columns = runes[:0], columns[:0]
527		-	for x, cell := range cells {
528		-	if cell.Continuation {
529		-	continue
530		-	}
531		-	r := cell.Rune
532		-	if r == 0 {
533		-	r = ' '
534		-	}
535		-	runes = append(runes, r)
536		-	columns = append(columns, x)
537		-	}
538		-	if !yield(match.Row, runes, columns, len(cells)) {
539		-	return nil
540		-	}
541		-	}
542		-	return nil
543		-	})
544		-	}
545		-
546	504		func findMatches(query string, rows textRows) []SearchMatch {
547	505		needle := lowerRunes(query)
548	506		matches := []SearchMatch{}


@@ -501,48 +501,6 @@ func FindMatches(doc *Document, query string) []SearchMatch {
501		return findMatches(query, doc.Snapshot().rangeText)	501	return findMatches(query, doc.Snapshot().rangeText)
502		}	502	}
503			503
504	-	// RefineMatches returns FindMatches(doc, query) given previous, the complete
505	-	// result for a query that query extends. Any row matching the longer query also
506	-	// matched the shorter one, so only rows listed in previous are scanned again.
507	-	func RefineMatches(doc *Document, previous []SearchMatch, query string) []SearchMatch {
508	-	query = strings.TrimSpace(query)
509	-	if doc == nil \|\| query == "" {
510	-	return nil
511	-	}
512	-	snapshot := doc.Snapshot()
513	-	return findMatches(query, func(yield func(int, []rune, []int, int) bool) error {
514	-	var cells []renderer.Cell
515	-	var runes []rune
516	-	var columns []int
517	-	last := -1
518	-	for _, match := range previous {
519	-	if match.Row == last {
520	-	continue
521	-	}
522	-	last = match.Row
523	-	width := snapshot.RowWidth(match.Row)
524	-	cells = slices.Grow(cells[:0], width)[:width]
525	-	cells = cells[:snapshot.CopyRow(match.Row, cells)]
526	-	runes, columns = runes[:0], columns[:0]
527	-	for x, cell := range cells {
528	-	if cell.Continuation {
529	-	continue
530	-	}
531	-	r := cell.Rune
532	-	if r == 0 {
533	-	r = ' '
534	-	}
535	-	runes = append(runes, r)
536	-	columns = append(columns, x)
537	-	}
538	-	if !yield(match.Row, runes, columns, len(cells)) {
539	-	return nil
540	-	}
541	-	}
542	-	return nil
543	-	})
544	-	}
545	-
546		func findMatches(query string, rows textRows) []SearchMatch {	504	func findMatches(query string, rows textRows) []SearchMatch {
547		needle := lowerRunes(query)	505	needle := lowerRunes(query)
548		matches := []SearchMatch{}	506	matches := []SearchMatch{}

M internal/usecase/daemon/echo_ack.go +3 -5


@@ -5,13 +5,11 @@ import (
5	5		"time"
6	6
7	7		"github.com/bnema/vev/internal/ports"
	8	+	"github.com/bnema/vev/internal/protocol"
8	9		)
9	10
10		-	// echoAckDelay is how long sequenced input must have been applied before the
11		-	// daemon acknowledges it in Output.Echo. Like mosh's ECHO_TIMEOUT, it gives the
12		-	// PTY time to echo the input, so the screen published with the acknowledgement
13		-	// already contains whatever echo the input produced.
14		-	const echoAckDelay = 50 * time.Millisecond
	11	+	// echoAckDelay is the protocol's echo acknowledgement delay.
	12	+	const echoAckDelay = protocol.EchoAckDelay
15	13
16	14		// maxPendingEchoes bounds the per-attachment input history. Acknowledgements
17	15		// are cumulative, so dropping the oldest entry only delays its confirmation to

 	"time"
 	"github.com/bnema/vev/internal/ports"
+	"github.com/bnema/vev/internal/protocol"
 )
-// echoAckDelay is how long sequenced input must have been applied before the
+// echoAckDelay is the protocol's echo acknowledgement delay.
-// daemon acknowledges it in Output.Echo. Like mosh's ECHO_TIMEOUT, it gives the
+const echoAckDelay = protocol.EchoAckDelay
-// PTY time to echo the input, so the screen published with the acknowledgement
-// already contains whatever echo the input produced.
-const echoAckDelay = 50 * time.Millisecond
 // maxPendingEchoes bounds the per-attachment input history. Acknowledgements
 // are cumulative, so dropping the oldest entry only delays its confirmation to

M internal/usecase/daemon/echo_ack_test.go +5 -3


@@ -65,11 +65,13 @@ func (c echoTestClock) nextTimer(t testing.T) (*manualAttentionTimer, time.Dur
65	65
66	66		func TestEchoAckWaitsForEchoTimeoutAndSendsAckOnlyOutput(t *testing.T) {
67	67		clock := newEchoTestClock()
68		-	d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
	68	+	d, sess, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
69	69
70		-	d.noteInputApplied(ac, 1)
	70	+	// Drive the production input path so the acknowledgement wiring is
	71	+	// covered, not just the tracker.
	72	+	d.handleSequencedInput(sess, ac, 1, []byte("a"))
71	73		clock.advance(20 * time.Millisecond)
72		-	d.noteInputApplied(ac, 2)
	74	+	d.handleSequencedInput(sess, ac, 2, []byte("b"))
73	75		timer, delay := clock.nextTimer(t)
74	76		require.Equal(t, echoAckDelay, delay)
75	77		require.Zero(t, ac.echoAck.Load(), "input is not echoed when applied")

 func TestEchoAckWaitsForEchoTimeoutAndSendsAckOnlyOutput(t *testing.T) {
 	clock := newEchoTestClock()
-	d, _, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
+	d, sess, ac, sends := newManualSessionWithPTYsClock(t, clock, newQuietPTY())
-	d.noteInputApplied(ac, 1)
+	// Drive the production input path so the acknowledgement wiring is
+	// covered, not just the tracker.
+	d.handleSequencedInput(sess, ac, 1, []byte("a"))
 	clock.advance(20 * time.Millisecond)
-	d.noteInputApplied(ac, 2)
+	d.handleSequencedInput(sess, ac, 2, []byte("b"))
 	timer, delay := clock.nextTimer(t)
 	require.Equal(t, echoAckDelay, delay)
 	require.Zero(t, ac.echoAck.Load(), "input is not echoed when applied")

M internal/usecase/daemon/floating.go +1 -1


@@ -407,7 +407,7 @@ func (d Daemon) openAndInstallFloating(sess session, tb *tab, spec floatingLau
407	407		// Claim the session spare when it matches this launch; a missing or
408	408		// mismatched spare falls through to a fresh launch. Once a spare was
409	409		// used up, the session starts the next one in the background.
410		-	p, had := sess.takeFloatingSpare(spec.cwd, spec.floatingCommand)
	410	+	p, had := sess.takeFloatingSpare(spec)
411	411		if had {
412	412		defer d.ensureFloatingSpare(sess, tb)
413	413		}


@@ -407,7 +407,7 @@ func (d Daemon) openAndInstallFloating(sess session, tb *tab, spec floatingLau
407		// Claim the session spare when it matches this launch; a missing or	407		// Claim the session spare when it matches this launch; a missing or
408		// mismatched spare falls through to a fresh launch. Once a spare was	408		// mismatched spare falls through to a fresh launch. Once a spare was
409		// used up, the session starts the next one in the background.	409		// used up, the session starts the next one in the background.
410	-	p, had := sess.takeFloatingSpare(spec.cwd, spec.floatingCommand)	410	+	p, had := sess.takeFloatingSpare(spec)
411		if had {	411		if had {
412		defer d.ensureFloatingSpare(sess, tb)	412		defer d.ensureFloatingSpare(sess, tb)
413		}	413		}

M internal/usecase/daemon/floating_spare.go +115 -17

@@ -1,6 +1,9 @@
1	1		package daemon
2	2
3	3		import (
	4	+	"strings"
	5	+
	6	+	"github.com/bnema/vev/internal/domain"
4	7		"github.com/bnema/vev/internal/domain/layout"
5	8		)
6	9

@@ -8,20 +11,39 @@ import (
8	11		// than by a tab. The first tab that opens its floating pane claims it, and the
9	12		// session then starts a new spare in the background, so a session keeps at
10	13		// most one idle floating shell instead of one hidden shell per tab. A spare is
11		-	// never persisted or restored. Its reader starts only once a tab installs it;
12		-	// until then the kernel PTY buffer holds its first prompt. Its VEV identity
13		-	// names the tab that started it; floating panes are not --self targets.
	14	+	// never persisted or restored. Its reader starts with the shell, so startup
	15	+	// terminal queries are answered before any claim; with no owner its screen
	16	+	// changes publish nothing. Its VEV identity names the tab that started it;
	17	+	// floating panes are not --self targets.
14	18		//
15		-	// A claim takes the claiming tab's geometry (resize before the reader starts)
16		-	// and cwd: a spare started in another directory is discarded and the claim
17		-	// launches a fresh shell instead.
	19	+	// A claim takes the claiming tab's geometry and requires the same cwd,
	20	+	// floating.command, shell and environment: a mismatched spare is discarded and
	21	+	// the claim launches a fresh shell instead.
18	22		//
19	23		// The spare is guarded by session.floatingLaunchMu, which is never held across
20	24		// PTY operations or architecture locks.
21	25		type floatingSpare struct {
22		-	pane *pane
23		-	cwd string
24		-	command string
	26	+	pane *pane
	27	+	key floatingSpareKey
	28	+	}
	29	+
	30	+	// floatingSpareKey is every launch input a claim must share with the spare.
	31	+	// env excludes the per-launch VEV identity.
	32	+	type floatingSpareKey struct {
	33	+	cwd, floatingCommand, command, args, env string
	34	+	}
	35	+
	36	+	func spareKey(spec floatingLaunchSpec) floatingSpareKey {
	37	+	env := make([]string, 0, len(spec.env))
	38	+	for _, entry := range spec.env {
	39	+	if key, _, _ := environmentEntry(entry); key != "VEV" {
	40	+	env = append(env, entry)
	41	+	}
	42	+	}
	43	+	return floatingSpareKey{
	44	+	cwd: spec.cwd, floatingCommand: spec.floatingCommand, command: spec.command,
	45	+	args: strings.Join(spec.args, "\x00"), env: strings.Join(env, "\x00"),
	46	+	}
25	47		}
26	48
27	49		// ensureFloatingSpare starts the session's spare when none exists or is

@@ -58,10 +80,28 @@ func (d Daemon) ensureFloatingSpare(sess session, tb *tab) {
58	80		d.sessWg.Go(func() {
59	81		defer sess.finishFloatingLaunch(launch)
60	82		p := d.openFloatingPane(sess, spec)
	83	+	if p != nil {
	84	+	// The reader answers color and scheme queries, so the spare
	85	+	// starts with the session theme, as a new tab does.
	86	+	d.applySessionThemeToPane(sess, p)
	87	+	}
61	88		sess.floatingLaunchMu.Lock()
62	89		sess.floatingSpareStarting = false
63	90		if p != nil && !sess.floatingLaunchStopping {
64		-	sess.floatingSpare = &floatingSpare{pane: p, cwd: spec.cwd, command: cfg.Command}
	91	+	sess.floatingSpare = &floatingSpare{pane: p, key: spareKey(spec)}
	92	+	p.prestarted = true
	93	+	spare := p
	94	+	spare.onExit = func() {
	95	+	spare.spareExited.Store(true)
	96	+	// Still unclaimed: drop it. A claim takes the spare under the
	97	+	// same lock, then reaps after install when spareExited is set.
	98	+	if sess.dropFloatingSpare(spare) {
	99	+	return
	100	+	}
	101	+	d.reapInstalledFloating(spare)
	102	+	}
	103	+	d.sessWg.Add(1)
	104	+	go d.readPanePTY(p)
65	105		p = nil
66	106		}
67	107		sess.floatingLaunchMu.Unlock()

@@ -72,10 +112,10 @@ func (d Daemon) ensureFloatingSpare(sess session, tb *tab) {
72	112		}
73	113
74	114		// takeFloatingSpare removes the session spare. It returns the pane when the
75		-	// spare was started for cwd and command; a mismatched spare is closed because
76		-	// it can never serve this directory or config. had reports whether a spare
	115	+	// spare was started with the same launch inputs as spec; a mismatched spare is
	116	+	// closed because it can never serve this launch. had reports whether a spare
77	117		// existed, so the caller knows to start its replacement.
78		-	func (sess session) takeFloatingSpare(cwd, command string) (p pane, had bool) {
	118	+	func (sess session) takeFloatingSpare(spec floatingLaunchSpec) (p pane, had bool) {
79	119		sess.floatingLaunchMu.Lock()
80	120		spare := sess.floatingSpare
81	121		sess.floatingSpare = nil

@@ -83,13 +123,52 @@ func (sess session) takeFloatingSpare(cwd, command string) (p pane, had bool)
83	123		if spare == nil {
84	124		return nil, false
85	125		}
86		-	if spare.cwd != cwd \|\| spare.command != command {
	126	+	if spare.key != spareKey(spec) {
87	127		closeFloatingPane(spare.pane)
88	128		return nil, true
89	129		}
90	130		return spare.pane, true
91	131		}
92	132
	133	+	// dropFloatingSpare closes p when it is still the session's unclaimed spare
	134	+	// and reports whether it did. A claimed spare is left to its tab.
	135	+	func (sess session) dropFloatingSpare(p pane) bool {
	136	+	sess.floatingLaunchMu.Lock()
	137	+	unclaimed := sess.floatingSpare != nil && sess.floatingSpare.pane == p
	138	+	if unclaimed {
	139	+	sess.floatingSpare = nil
	140	+	}
	141	+	sess.floatingLaunchMu.Unlock()
	142	+	if unclaimed {
	143	+	closeFloatingPane(p)
	144	+	}
	145	+	return unclaimed
	146	+	}
	147	+
	148	+	// applySessionThemeToPane gives a pane the theme of the session's first
	149	+	// attachment, as createTab does for a new tab. A headless session leaves the
	150	+	// pane unthemed until the next theme application.
	151	+	func (d Daemon) applySessionThemeToPane(sess session, p *pane) {
	152	+	attachments := sess.snapshotAttachments()
	153	+	if len(attachments) == 0 {
	154	+	return
	155	+	}
	156	+	t := d.effectiveTheme(attachments[0].getClientTheme())
	157	+	p.mu.Lock()
	158	+	applyPaneThemeLocked(p, t, false)
	159	+	p.mu.Unlock()
	160	+	}
	161	+
	162	+	// floatingSparePane returns the session's unclaimed spare pane, if any.
	163	+	func (sess session) floatingSparePane() pane {
	164	+	sess.floatingLaunchMu.Lock()
	165	+	defer sess.floatingLaunchMu.Unlock()
	166	+	if sess.floatingSpare == nil {
	167	+	return nil
	168	+	}
	169	+	return sess.floatingSpare.pane
	170	+	}
	171	+
93	172		// closeFloatingSpare releases the spare on session teardown. The caller has
94	173		// already set floatingLaunchStopping, so no launch can publish a new one.
95	174		func (sess *session) closeFloatingSpare() {

@@ -134,8 +213,19 @@ func (d Daemon) openFloatingPane(sess session, spec floatingLaunchSpec) *pane
134	213		// and installs it in the current slot generation. It runs in the launch worker
135	214		// without any architecture lock, because PTY.Resize is external.
136	215		func (d Daemon) adoptFloatingSpare(sess session, tb tab, p pane, spec floatingLaunchSpec, generation uint64) {
137		-	if p.geometry != spec.ptyGeometry {
138		-	if err := p.pty.Resize(spec.ptyGeometry); err != nil {
	216	+	p.resizeMu.Lock()
	217	+	p.mu.Lock()
	218	+	resize := p.geometry != spec.ptyGeometry
	219	+	if resize {
	220	+	// The spare's reader is running: buffer its output across the
	221	+	// resize so the redraw is parsed at the new size.
	222	+	p.resizeApplying = true
	223	+	}
	224	+	p.mu.Unlock()
	225	+	if resize {
	226	+	if err := resizePTYGeometry(p.pty, spec.ptyGeometry); err != nil {
	227	+	d.replayResizePending(sess, tb, p, false, domain.Rect{})
	228	+	p.resizeMu.Unlock()
139	229		closeFloatingPane(p)
140	230		d.failFloatingLaunch(sess, tb, generation, true, spec.sessionName, err)
141	231		return

@@ -147,6 +237,14 @@ func (d Daemon) adoptFloatingSpare(sess session, tb tab, p pane, spec floati
147	237		p.rect = spec.geometry.ptyRect()
148	238		p.popupGeometry = spec.geometry
149	239		p.mu.Unlock()
150		-	p.onExit = func() { d.reapInstalledFloating(p) }
	240	+	if resize {
	241	+	d.replayResizePending(sess, tb, p, false, domain.Rect{})
	242	+	}
	243	+	p.resizeMu.Unlock()
151	244		d.installFloating(sess, tb, p, generation)
	245	+	if p.spareExited.Load() {
	246	+	// The shell exited before the owner was published, so its reader
	247	+	// could not reap the slot.
	248	+	d.reapInstalledFloating(p)
	249	+	}
152	250		}

@@ -1,6 +1,9 @@
1		package daemon	1		package daemon
2			2
3		import (	3		import (
			4	+	"strings"
			5	+
			6	+	"github.com/bnema/vev/internal/domain"
4		"github.com/bnema/vev/internal/domain/layout"	7		"github.com/bnema/vev/internal/domain/layout"
5		)	8		)
6			9

@@ -8,20 +11,39 @@ import (
8		// than by a tab. The first tab that opens its floating pane claims it, and the	11		// than by a tab. The first tab that opens its floating pane claims it, and the
9		// session then starts a new spare in the background, so a session keeps at	12		// session then starts a new spare in the background, so a session keeps at
10		// most one idle floating shell instead of one hidden shell per tab. A spare is	13		// most one idle floating shell instead of one hidden shell per tab. A spare is
11	-	// never persisted or restored. Its reader starts only once a tab installs it;	14	+	// never persisted or restored. Its reader starts with the shell, so startup
12	-	// until then the kernel PTY buffer holds its first prompt. Its VEV identity	15	+	// terminal queries are answered before any claim; with no owner its screen
13	-	// names the tab that started it; floating panes are not --self targets.	16	+	// changes publish nothing. Its VEV identity names the tab that started it;
			17	+	// floating panes are not --self targets.
14		//	18		//
15	-	// A claim takes the claiming tab's geometry (resize before the reader starts)	19	+	// A claim takes the claiming tab's geometry and requires the same cwd,
16	-	// and cwd: a spare started in another directory is discarded and the claim	20	+	// floating.command, shell and environment: a mismatched spare is discarded and
17	-	// launches a fresh shell instead.	21	+	// the claim launches a fresh shell instead.
18		//	22		//
19		// The spare is guarded by session.floatingLaunchMu, which is never held across	23		// The spare is guarded by session.floatingLaunchMu, which is never held across
20		// PTY operations or architecture locks.	24		// PTY operations or architecture locks.
21		type floatingSpare struct {	25		type floatingSpare struct {
22	-	pane *pane	26	+	pane *pane
23	-	cwd string	27	+	key floatingSpareKey
24	-	command string	28	+	}
			29	+
			30	+	// floatingSpareKey is every launch input a claim must share with the spare.
			31	+	// env excludes the per-launch VEV identity.
			32	+	type floatingSpareKey struct {
			33	+	cwd, floatingCommand, command, args, env string
			34	+	}
			35	+
			36	+	func spareKey(spec floatingLaunchSpec) floatingSpareKey {
			37	+	env := make([]string, 0, len(spec.env))
			38	+	for _, entry := range spec.env {
			39	+	if key, _, _ := environmentEntry(entry); key != "VEV" {
			40	+	env = append(env, entry)
			41	+	}
			42	+	}
			43	+	return floatingSpareKey{
			44	+	cwd: spec.cwd, floatingCommand: spec.floatingCommand, command: spec.command,
			45	+	args: strings.Join(spec.args, "\x00"), env: strings.Join(env, "\x00"),
			46	+	}
25		}	47		}
26			48
27		// ensureFloatingSpare starts the session's spare when none exists or is	49		// ensureFloatingSpare starts the session's spare when none exists or is

@@ -58,10 +80,28 @@ func (d Daemon) ensureFloatingSpare(sess session, tb *tab) {
58		d.sessWg.Go(func() {	80		d.sessWg.Go(func() {
59		defer sess.finishFloatingLaunch(launch)	81		defer sess.finishFloatingLaunch(launch)
60		p := d.openFloatingPane(sess, spec)	82		p := d.openFloatingPane(sess, spec)
			83	+	if p != nil {
			84	+	// The reader answers color and scheme queries, so the spare
			85	+	// starts with the session theme, as a new tab does.
			86	+	d.applySessionThemeToPane(sess, p)
			87	+	}
61		sess.floatingLaunchMu.Lock()	88		sess.floatingLaunchMu.Lock()
62		sess.floatingSpareStarting = false	89		sess.floatingSpareStarting = false
63		if p != nil && !sess.floatingLaunchStopping {	90		if p != nil && !sess.floatingLaunchStopping {
64	-	sess.floatingSpare = &floatingSpare{pane: p, cwd: spec.cwd, command: cfg.Command}	91	+	sess.floatingSpare = &floatingSpare{pane: p, key: spareKey(spec)}
			92	+	p.prestarted = true
			93	+	spare := p
			94	+	spare.onExit = func() {
			95	+	spare.spareExited.Store(true)
			96	+	// Still unclaimed: drop it. A claim takes the spare under the
			97	+	// same lock, then reaps after install when spareExited is set.
			98	+	if sess.dropFloatingSpare(spare) {
			99	+	return
			100	+	}
			101	+	d.reapInstalledFloating(spare)
			102	+	}
			103	+	d.sessWg.Add(1)
			104	+	go d.readPanePTY(p)
65		p = nil	105		p = nil
66		}	106		}
67		sess.floatingLaunchMu.Unlock()	107		sess.floatingLaunchMu.Unlock()

@@ -72,10 +112,10 @@ func (d Daemon) ensureFloatingSpare(sess session, tb *tab) {
72		}	112		}
73			113
74		// takeFloatingSpare removes the session spare. It returns the pane when the	114		// takeFloatingSpare removes the session spare. It returns the pane when the
75	-	// spare was started for cwd and command; a mismatched spare is closed because	115	+	// spare was started with the same launch inputs as spec; a mismatched spare is
76	-	// it can never serve this directory or config. had reports whether a spare	116	+	// closed because it can never serve this launch. had reports whether a spare
77		// existed, so the caller knows to start its replacement.	117		// existed, so the caller knows to start its replacement.
78	-	func (sess session) takeFloatingSpare(cwd, command string) (p pane, had bool) {	118	+	func (sess session) takeFloatingSpare(spec floatingLaunchSpec) (p pane, had bool) {
79		sess.floatingLaunchMu.Lock()	119		sess.floatingLaunchMu.Lock()
80		spare := sess.floatingSpare	120		spare := sess.floatingSpare
81		sess.floatingSpare = nil	121		sess.floatingSpare = nil

@@ -83,13 +123,52 @@ func (sess session) takeFloatingSpare(cwd, command string) (p pane, had bool)
83		if spare == nil {	123		if spare == nil {
84		return nil, false	124		return nil, false
85		}	125		}
86	-	if spare.cwd != cwd \|\| spare.command != command {	126	+	if spare.key != spareKey(spec) {
87		closeFloatingPane(spare.pane)	127		closeFloatingPane(spare.pane)
88		return nil, true	128		return nil, true
89		}	129		}
90		return spare.pane, true	130		return spare.pane, true
91		}	131		}
92			132
			133	+	// dropFloatingSpare closes p when it is still the session's unclaimed spare
			134	+	// and reports whether it did. A claimed spare is left to its tab.
			135	+	func (sess session) dropFloatingSpare(p pane) bool {
			136	+	sess.floatingLaunchMu.Lock()
			137	+	unclaimed := sess.floatingSpare != nil && sess.floatingSpare.pane == p
			138	+	if unclaimed {
			139	+	sess.floatingSpare = nil
			140	+	}
			141	+	sess.floatingLaunchMu.Unlock()
			142	+	if unclaimed {
			143	+	closeFloatingPane(p)
			144	+	}
			145	+	return unclaimed
			146	+	}
			147	+
			148	+	// applySessionThemeToPane gives a pane the theme of the session's first
			149	+	// attachment, as createTab does for a new tab. A headless session leaves the
			150	+	// pane unthemed until the next theme application.
			151	+	func (d Daemon) applySessionThemeToPane(sess session, p *pane) {
			152	+	attachments := sess.snapshotAttachments()
			153	+	if len(attachments) == 0 {
			154	+	return
			155	+	}
			156	+	t := d.effectiveTheme(attachments[0].getClientTheme())
			157	+	p.mu.Lock()
			158	+	applyPaneThemeLocked(p, t, false)
			159	+	p.mu.Unlock()
			160	+	}
			161	+
			162	+	// floatingSparePane returns the session's unclaimed spare pane, if any.
			163	+	func (sess session) floatingSparePane() pane {
			164	+	sess.floatingLaunchMu.Lock()
			165	+	defer sess.floatingLaunchMu.Unlock()
			166	+	if sess.floatingSpare == nil {
			167	+	return nil
			168	+	}
			169	+	return sess.floatingSpare.pane
			170	+	}
			171	+
93		// closeFloatingSpare releases the spare on session teardown. The caller has	172		// closeFloatingSpare releases the spare on session teardown. The caller has
94		// already set floatingLaunchStopping, so no launch can publish a new one.	173		// already set floatingLaunchStopping, so no launch can publish a new one.
95		func (sess *session) closeFloatingSpare() {	174		func (sess *session) closeFloatingSpare() {

@@ -134,8 +213,19 @@ func (d Daemon) openFloatingPane(sess session, spec floatingLaunchSpec) *pane
134		// and installs it in the current slot generation. It runs in the launch worker	213		// and installs it in the current slot generation. It runs in the launch worker
135		// without any architecture lock, because PTY.Resize is external.	214		// without any architecture lock, because PTY.Resize is external.
136		func (d Daemon) adoptFloatingSpare(sess session, tb tab, p pane, spec floatingLaunchSpec, generation uint64) {	215		func (d Daemon) adoptFloatingSpare(sess session, tb tab, p pane, spec floatingLaunchSpec, generation uint64) {
137	-	if p.geometry != spec.ptyGeometry {	216	+	p.resizeMu.Lock()
138	-	if err := p.pty.Resize(spec.ptyGeometry); err != nil {	217	+	p.mu.Lock()
			218	+	resize := p.geometry != spec.ptyGeometry
			219	+	if resize {
			220	+	// The spare's reader is running: buffer its output across the
			221	+	// resize so the redraw is parsed at the new size.
			222	+	p.resizeApplying = true
			223	+	}
			224	+	p.mu.Unlock()
			225	+	if resize {
			226	+	if err := resizePTYGeometry(p.pty, spec.ptyGeometry); err != nil {
			227	+	d.replayResizePending(sess, tb, p, false, domain.Rect{})
			228	+	p.resizeMu.Unlock()
139		closeFloatingPane(p)	229		closeFloatingPane(p)
140		d.failFloatingLaunch(sess, tb, generation, true, spec.sessionName, err)	230		d.failFloatingLaunch(sess, tb, generation, true, spec.sessionName, err)
141		return	231		return

@@ -147,6 +237,14 @@ func (d Daemon) adoptFloatingSpare(sess session, tb tab, p pane, spec floati
147		p.rect = spec.geometry.ptyRect()	237		p.rect = spec.geometry.ptyRect()
148		p.popupGeometry = spec.geometry	238		p.popupGeometry = spec.geometry
149		p.mu.Unlock()	239		p.mu.Unlock()
150	-	p.onExit = func() { d.reapInstalledFloating(p) }	240	+	if resize {
			241	+	d.replayResizePending(sess, tb, p, false, domain.Rect{})
			242	+	}
			243	+	p.resizeMu.Unlock()
151		d.installFloating(sess, tb, p, generation)	244		d.installFloating(sess, tb, p, generation)
			245	+	if p.spareExited.Load() {
			246	+	// The shell exited before the owner was published, so its reader
			247	+	// could not reap the slot.
			248	+	d.reapInstalledFloating(p)
			249	+	}
152		}	250		}

M internal/usecase/daemon/floating_spare_test.go +87 -0


@@ -7,10 +7,12 @@ import (
7	7		"testing"
8	8		"time"
9	9
	10	+	renderer "github.com/bnema/vev-vt"
10	11		"github.com/stretchr/testify/require"
11	12
12	13		"github.com/bnema/vev/internal/domain"
13	14		"github.com/bnema/vev/internal/ports"
	15	+	themeui "github.com/bnema/vev/internal/usecase/theme"
14	16		)
15	17
16	18		// spareOpen records one PTY Open issued for a floating shell.

@@ -183,3 +185,88 @@ type pidPTY struct {
183	185		}
184	186
185	187		func (p *pidPTY) Pid() int { return p.pid }
	188	+
	189	+	// queryPTY is a floating shell that sends one terminal query at startup and
	190	+	// records what the daemon writes back.
	191	+	type queryPTY struct {
	192	+	sparePTY
	193	+	query string
	194	+	sent chan struct{}
	195	+	replied chan []byte
	196	+	}
	197	+
	198	+	func (p *queryPTY) Read(b []byte) (int, error) {
	199	+	select {
	200	+	case <-p.sent:
	201	+	return p.sparePTY.Read(b)
	202	+	default:
	203	+	close(p.sent)
	204	+	return copy(b, p.query), nil
	205	+	}
	206	+	}
	207	+
	208	+	func (p *queryPTY) Write(b []byte) (int, error) {
	209	+	p.replied <- append([]byte(nil), b...)
	210	+	return len(b), nil
	211	+	}
	212	+
	213	+	type queryFactory struct{ pty *queryPTY }
	214	+
	215	+	func (f *queryFactory) Open(ctx context.Context, _ string, _ []string, _ []string, _ string, _ domain.Geometry) (ports.PTY, error) {
	216	+	f.pty.ctx = ctx
	217	+	return f.pty, nil
	218	+	}
	219	+
	220	+	func TestFloatingSpareAnswersStartupQueriesBeforeClaim(t *testing.T) {
	221	+	for _, tt := range []struct {
	222	+	name string
	223	+	query string
	224	+	want string
	225	+	}{
	226	+	{name: "primary device attributes", query: "\x1b[c", want: "\x1b[?"},
	227	+	{name: "background color from the session theme", query: "\x1b]11;?\x07", want: "\x1b]11;rgb:"},
	228	+	} {
	229	+	t.Run(tt.name, func(t *testing.T) {
	230	+	pty := &queryPTY{sparePTY: sparePTY{closed: make(chan struct{})}, query: tt.query, sent: make(chan struct{}), replied: make(chan []byte, 4)}
	231	+	d := newTestDaemon(t, &queryFactory{pty: pty}, stubClock{})
	232	+	d.ApplyConfig(domain.Config{Floating: domain.FloatingConfig{Width: 80, Height: 80}})
	233	+	tb := newTab(&pidPTY{pid: 1}, domain.Size{Cols: 80, Rows: 24})
	234	+	tb.ctx, tb.cancel = context.WithCancel(t.Context())
	235	+	ac := &attachedClient{}
	236	+	ac.setClientTheme(themeui.Theme{Known: true, HasFG: true, HasBG: true,
	237	+	Foreground: renderer.RGB{R: 0xee, G: 0xee, B: 0xee}, Background: renderer.RGB{R: 0x11, G: 0x11, B: 0x11}})
	238	+	sess := &session{sessionCore: sessionCore{name: "work", attachments: map[attachedClient]struct{}{ac: {}}}, tabs: []tab{tb}, ctx: t.Context()}
	239	+
	240	+	d.ensureFloatingWarm(sess, tb)
	241	+	select {
	242	+	case reply := <-pty.replied:
	243	+	require.Contains(t, string(reply), tt.want, "the unclaimed spare must answer its shell's startup query")
	244	+	case <-time.After(5 * time.Second):
	245	+	t.Fatal("the unclaimed spare never answered its shell's startup query")
	246	+	}
	247	+
	248	+	sess.stopInMemoryLifecycle()
	249	+	d.sessWg.Wait()
	250	+	})
	251	+	}
	252	+	}
	253	+
	254	+	func TestFloatingSpareExitBeforeClaimDropsIt(t *testing.T) {
	255	+	factory := &spareFactory{}
	256	+	d := newTestDaemon(t, factory, stubClock{})
	257	+	d.ApplyConfig(domain.Config{Floating: domain.FloatingConfig{Width: 80, Height: 80}})
	258	+	tb := newTab(&pidPTY{pid: 1}, domain.Size{Cols: 80, Rows: 24})
	259	+	tb.ctx, tb.cancel = context.WithCancel(t.Context())
	260	+	sess := &session{sessionCore: sessionCore{name: "work"}, tabs: []*tab{tb}, ctx: t.Context()}
	261	+
	262	+	d.ensureFloatingWarm(sess, tb)
	263	+	waitSpareReady(t, sess)
	264	+	spare := sess.floatingSparePane()
	265	+	require.NotNil(t, spare)
	266	+	require.NoError(t, factory.snapshot()[0].pty.Close(), "the spare's shell exits")
	267	+	require.Eventually(t, func() bool { return sess.floatingSparePane() == nil }, time.Second, time.Millisecond,
	268	+	"an exited unclaimed spare must be dropped so the next claim launches a fresh shell")
	269	+
	270	+	sess.stopInMemoryLifecycle()
	271	+	d.sessWg.Wait()
	272	+	}


@@ -7,10 +7,12 @@ import (
7	"testing"	7		"testing"
8	"time"	8		"time"
9		9
		10	+	renderer "github.com/bnema/vev-vt"
10	"github.com/stretchr/testify/require"	11		"github.com/stretchr/testify/require"
11		12
12	"github.com/bnema/vev/internal/domain"	13		"github.com/bnema/vev/internal/domain"
13	"github.com/bnema/vev/internal/ports"	14		"github.com/bnema/vev/internal/ports"
		15	+	themeui "github.com/bnema/vev/internal/usecase/theme"
14	)	16		)
15		17
16	// spareOpen records one PTY Open issued for a floating shell.	18		// spareOpen records one PTY Open issued for a floating shell.

@@ -183,3 +185,88 @@ type pidPTY struct {
183	}	185		}
184		186
185	func (p *pidPTY) Pid() int { return p.pid }	187		func (p *pidPTY) Pid() int { return p.pid }
		188	+
		189	+	// queryPTY is a floating shell that sends one terminal query at startup and
		190	+	// records what the daemon writes back.
		191	+	type queryPTY struct {
		192	+	sparePTY
		193	+	query string
		194	+	sent chan struct{}
		195	+	replied chan []byte
		196	+	}
		197	+
		198	+	func (p *queryPTY) Read(b []byte) (int, error) {
		199	+	select {
		200	+	case <-p.sent:
		201	+	return p.sparePTY.Read(b)
		202	+	default:
		203	+	close(p.sent)
		204	+	return copy(b, p.query), nil
		205	+	}
		206	+	}
		207	+
		208	+	func (p *queryPTY) Write(b []byte) (int, error) {
		209	+	p.replied <- append([]byte(nil), b...)
		210	+	return len(b), nil
		211	+	}
		212	+
		213	+	type queryFactory struct{ pty *queryPTY }
		214	+
		215	+	func (f *queryFactory) Open(ctx context.Context, _ string, _ []string, _ []string, _ string, _ domain.Geometry) (ports.PTY, error) {
		216	+	f.pty.ctx = ctx
		217	+	return f.pty, nil
		218	+	}
		219	+
		220	+	func TestFloatingSpareAnswersStartupQueriesBeforeClaim(t *testing.T) {
		221	+	for _, tt := range []struct {
		222	+	name string
		223	+	query string
		224	+	want string
		225	+	}{
		226	+	{name: "primary device attributes", query: "\x1b[c", want: "\x1b[?"},
		227	+	{name: "background color from the session theme", query: "\x1b]11;?\x07", want: "\x1b]11;rgb:"},
		228	+	} {
		229	+	t.Run(tt.name, func(t *testing.T) {
		230	+	pty := &queryPTY{sparePTY: sparePTY{closed: make(chan struct{})}, query: tt.query, sent: make(chan struct{}), replied: make(chan []byte, 4)}
		231	+	d := newTestDaemon(t, &queryFactory{pty: pty}, stubClock{})
		232	+	d.ApplyConfig(domain.Config{Floating: domain.FloatingConfig{Width: 80, Height: 80}})
		233	+	tb := newTab(&pidPTY{pid: 1}, domain.Size{Cols: 80, Rows: 24})
		234	+	tb.ctx, tb.cancel = context.WithCancel(t.Context())
		235	+	ac := &attachedClient{}
		236	+	ac.setClientTheme(themeui.Theme{Known: true, HasFG: true, HasBG: true,
		237	+	Foreground: renderer.RGB{R: 0xee, G: 0xee, B: 0xee}, Background: renderer.RGB{R: 0x11, G: 0x11, B: 0x11}})
		238	+	sess := &session{sessionCore: sessionCore{name: "work", attachments: map[attachedClient]struct{}{ac: {}}}, tabs: []tab{tb}, ctx: t.Context()}
		239	+
		240	+	d.ensureFloatingWarm(sess, tb)
		241	+	select {
		242	+	case reply := <-pty.replied:
		243	+	require.Contains(t, string(reply), tt.want, "the unclaimed spare must answer its shell's startup query")
		244	+	case <-time.After(5 * time.Second):
		245	+	t.Fatal("the unclaimed spare never answered its shell's startup query")
		246	+	}
		247	+
		248	+	sess.stopInMemoryLifecycle()
		249	+	d.sessWg.Wait()
		250	+	})
		251	+	}
		252	+	}
		253	+
		254	+	func TestFloatingSpareExitBeforeClaimDropsIt(t *testing.T) {
		255	+	factory := &spareFactory{}
		256	+	d := newTestDaemon(t, factory, stubClock{})
		257	+	d.ApplyConfig(domain.Config{Floating: domain.FloatingConfig{Width: 80, Height: 80}})
		258	+	tb := newTab(&pidPTY{pid: 1}, domain.Size{Cols: 80, Rows: 24})
		259	+	tb.ctx, tb.cancel = context.WithCancel(t.Context())
		260	+	sess := &session{sessionCore: sessionCore{name: "work"}, tabs: []*tab{tb}, ctx: t.Context()}
		261	+
		262	+	d.ensureFloatingWarm(sess, tb)
		263	+	waitSpareReady(t, sess)
		264	+	spare := sess.floatingSparePane()
		265	+	require.NotNil(t, spare)
		266	+	require.NoError(t, factory.snapshot()[0].pty.Close(), "the spare's shell exits")
		267	+	require.Eventually(t, func() bool { return sess.floatingSparePane() == nil }, time.Second, time.Millisecond,
		268	+	"an exited unclaimed spare must be dropped so the next claim launches a fresh shell")
		269	+
		270	+	sess.stopInMemoryLifecycle()
		271	+	d.sessWg.Wait()
		272	+	}

M internal/usecase/daemon/pane.go +7 -0


@@ -145,6 +145,13 @@ type pane struct {
145	145		// onExit is set before the reader starts and never changed. Floating panes
146	146		// use it to reap their independent slot without touching the layout tree.
147	147		onExit func()
	148	+	// prestarted marks a floating spare whose reader runs before any owner is
	149	+	// published, so terminal queries from a starting shell are answered.
	150	+	// Set before the reader starts and never changed.
	151	+	prestarted bool
	152	+	// spareExited records that a prestarted spare's reader ended, so a claim
	153	+	// racing that exit still reaps the slot it installed.
	154	+	spareExited atomic.Bool
148	155		}
149	156
150	157		func closePaneProcess(p *pane) {


@@ -145,6 +145,13 @@ type pane struct {
145	// onExit is set before the reader starts and never changed. Floating panes	145		// onExit is set before the reader starts and never changed. Floating panes
146	// use it to reap their independent slot without touching the layout tree.	146		// use it to reap their independent slot without touching the layout tree.
147	onExit func()	147		onExit func()
		148	+	// prestarted marks a floating spare whose reader runs before any owner is
		149	+	// published, so terminal queries from a starting shell are answered.
		150	+	// Set before the reader starts and never changed.
		151	+	prestarted bool
		152	+	// spareExited records that a prestarted spare's reader ended, so a claim
		153	+	// racing that exit still reaps the slot it installed.
		154	+	spareExited atomic.Bool
148	}	155		}
149		156
150	func closePaneProcess(p *pane) {	157		func closePaneProcess(p *pane) {

M internal/usecase/daemon/render.go +5 -0


@@ -213,6 +213,11 @@ func filterUnsupportedKittyAnimationResponses(data []byte) []byte {
213	213		// bytes that belonged to the old owner.
214	214		func (d Daemon) applyPanePTYEffects(p pane, effects panePTYEffects) {
215	215		owner := effects.lease.owner
	216	+	if owner == nil && p.prestarted && len(effects.responses) > 0 {
	217	+	// An unclaimed spare still answers its shell's startup queries; a
	218	+	// closed spare only fails the write.
	219	+	_, _ = p.pty.Write(effects.responses)
	220	+	}
216	221		if owner == nil \|\| owner.session == nil \|\| owner.tab == nil {
217	222		return
218	223		}


@@ -213,6 +213,11 @@ func filterUnsupportedKittyAnimationResponses(data []byte) []byte {
213	// bytes that belonged to the old owner.	213		// bytes that belonged to the old owner.
214	func (d Daemon) applyPanePTYEffects(p pane, effects panePTYEffects) {	214		func (d Daemon) applyPanePTYEffects(p pane, effects panePTYEffects) {
215	owner := effects.lease.owner	215		owner := effects.lease.owner
		216	+	if owner == nil && p.prestarted && len(effects.responses) > 0 {
		217	+	// An unclaimed spare still answers its shell's startup queries; a
		218	+	// closed spare only fails the write.
		219	+	_, _ = p.pty.Write(effects.responses)
		220	+	}
216	if owner == nil \|\| owner.session == nil \|\| owner.tab == nil {	221		if owner == nil \|\| owner.session == nil \|\| owner.tab == nil {
217	return	222		return
218	}	223		}

M internal/usecase/daemon/same_peer_switch.go +19 -0

@@ -1,6 +1,8 @@
1	1		package daemon
2	2
3	3		import (
	4	+	"sync"
	5	+
4	6		"github.com/bnema/vev/internal/domain"
5	7		"github.com/bnema/vev/internal/protocol"
6	8		)

@@ -72,6 +74,22 @@ func (d Daemon) switchSamePeerForAttachment(effect attachmentEffect, request p
72	74		return
73	75		}
74	76
	77	+	// The switch moves an attachment between sessions, so it must not run
	78	+	// inside the exact set a KillAll purge has captured.
	79	+	d.mu.Lock()
	80	+	if d.purgeAdmissionClosedLocked() {
	81	+	d.mu.Unlock()
	82	+	d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)
	83	+	return
	84	+	}
	85	+	d.purgeAdmissionActive++
	86	+	d.mu.Unlock()
	87	+	// The reservation covers only the membership change; first paint can
	88	+	// block on a slow client and must not stall a purge's drain.
	89	+	var releaseOnce sync.Once
	90	+	release := func() { releaseOnce.Do(d.releasePurgeAdmission) }
	91	+	defer release()
	92	+
75	93		target, targetTabIndex, ok := d.samePeerTarget(request)
76	94		if !ok \|\| target == effect.sess {
77	95		d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)

@@ -100,6 +118,7 @@ func (d Daemon) switchSamePeerForAttachment(effect attachmentEffect, request p
100	118		d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)
101	119		return
102	120		}
	121	+	release()
103	122
104	123		if fresh, admitted := effect.ac.beginAttachmentEffect(transition.published); admitted {
105	124		fresh.End()

@@ -1,6 +1,8 @@
1	package daemon	1		package daemon
2		2
3	import (	3		import (
		4	+	"sync"
		5	+
4	"github.com/bnema/vev/internal/domain"	6		"github.com/bnema/vev/internal/domain"
5	"github.com/bnema/vev/internal/protocol"	7		"github.com/bnema/vev/internal/protocol"
6	)	8		)

@@ -72,6 +74,22 @@ func (d Daemon) switchSamePeerForAttachment(effect attachmentEffect, request p
72	return	74		return
73	}	75		}
74		76
		77	+	// The switch moves an attachment between sessions, so it must not run
		78	+	// inside the exact set a KillAll purge has captured.
		79	+	d.mu.Lock()
		80	+	if d.purgeAdmissionClosedLocked() {
		81	+	d.mu.Unlock()
		82	+	d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)
		83	+	return
		84	+	}
		85	+	d.purgeAdmissionActive++
		86	+	d.mu.Unlock()
		87	+	// The reservation covers only the membership change; first paint can
		88	+	// block on a slow client and must not stall a purge's drain.
		89	+	var releaseOnce sync.Once
		90	+	release := func() { releaseOnce.Do(d.releasePurgeAdmission) }
		91	+	defer release()
		92	+
75	target, targetTabIndex, ok := d.samePeerTarget(request)	93		target, targetTabIndex, ok := d.samePeerTarget(request)
76	if !ok \|\| target == effect.sess {	94		if !ok \|\| target == effect.sess {
77	d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)	95		d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)

@@ -100,6 +118,7 @@ func (d Daemon) switchSamePeerForAttachment(effect attachmentEffect, request p
100	d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)	118		d.sendSamePeerSwitchFailure(effect, request.RequestID, protocol.SamePeerSwitchStaleTarget)
101	return	119		return
102	}	120		}
		121	+	release()
103		122
104	if fresh, admitted := effect.ac.beginAttachmentEffect(transition.published); admitted {	123		if fresh, admitted := effect.ac.beginAttachmentEffect(transition.published); admitted {
105	fresh.End()	124		fresh.End()

M internal/usecase/daemon/same_peer_switch_test.go +9 -0


@@ -131,11 +131,13 @@ func TestSamePeerSwitchClientInitiated(t *testing.T) {
131	131		tests := []struct {
132	132		name string
133	133		offer *protocol.ExactSessionTarget
	134	+	purging bool
134	135		switched bool
135	136		}{
136	137		{name: "no pending offer switches", switched: true},
137	138		{name: "matching offer switches", offer: &requestTarget, switched: true},
138	139		{name: "another pending offer refuses", offer: &otherOffer},
	140	+	{name: "a running purge refuses", purging: true},
139	141		}
140	142		for _, tt := range tests {
141	143		t.Run(tt.name, func(t *testing.T) {

@@ -155,6 +157,9 @@ func TestSamePeerSwitchClientInitiated(t *testing.T) {
155	157		if tt.offer != nil {
156	158		ac.offerSamePeerTarget(*tt.offer)
157	159		}
	160	+	d.mu.Lock()
	161	+	d.purgeAdmissionClosing = tt.purging
	162	+	d.mu.Unlock()
158	163
159	164		token := source.captureAttachmentCapability(ac, ac.transport())
160	165		effect, admitted := ac.beginAttachmentEffect(token)

@@ -162,6 +167,10 @@ func TestSamePeerSwitchClientInitiated(t *testing.T) {
162	167		defer effect.End()
163	168		d.switchSamePeerForAttachment(effect, protocol.SamePeerSwitchRequest{RequestID: 1, Target: requestTarget})
164	169
	170	+	d.mu.Lock()
	171	+	require.Zero(t, d.purgeAdmissionActive, "the switch releases its purge admission")
	172	+	d.purgeAdmissionClosing = false
	173	+	d.mu.Unlock()
165	174		if tt.switched {
166	175		require.Same(t, target, ac.currentAttachmentSession())
167	176		identity := decodeServerMessage(t, awaitFrame(t, sends, "CommittedRouteIdentity")).(protocol.CommittedRouteIdentity)


@@ -131,11 +131,13 @@ func TestSamePeerSwitchClientInitiated(t *testing.T) {
131	tests := []struct {	131		tests := []struct {
132	name string	132		name string
133	offer *protocol.ExactSessionTarget	133		offer *protocol.ExactSessionTarget
		134	+	purging bool
134	switched bool	135		switched bool
135	}{	136		}{
136	{name: "no pending offer switches", switched: true},	137		{name: "no pending offer switches", switched: true},
137	{name: "matching offer switches", offer: &requestTarget, switched: true},	138		{name: "matching offer switches", offer: &requestTarget, switched: true},
138	{name: "another pending offer refuses", offer: &otherOffer},	139		{name: "another pending offer refuses", offer: &otherOffer},
		140	+	{name: "a running purge refuses", purging: true},
139	}	141		}
140	for _, tt := range tests {	142		for _, tt := range tests {
141	t.Run(tt.name, func(t *testing.T) {	143		t.Run(tt.name, func(t *testing.T) {

@@ -155,6 +157,9 @@ func TestSamePeerSwitchClientInitiated(t *testing.T) {
155	if tt.offer != nil {	157		if tt.offer != nil {
156	ac.offerSamePeerTarget(*tt.offer)	158		ac.offerSamePeerTarget(*tt.offer)
157	}	159		}
		160	+	d.mu.Lock()
		161	+	d.purgeAdmissionClosing = tt.purging
		162	+	d.mu.Unlock()
158		163
159	token := source.captureAttachmentCapability(ac, ac.transport())	164		token := source.captureAttachmentCapability(ac, ac.transport())
160	effect, admitted := ac.beginAttachmentEffect(token)	165		effect, admitted := ac.beginAttachmentEffect(token)

@@ -162,6 +167,10 @@ func TestSamePeerSwitchClientInitiated(t *testing.T) {
162	defer effect.End()	167		defer effect.End()
163	d.switchSamePeerForAttachment(effect, protocol.SamePeerSwitchRequest{RequestID: 1, Target: requestTarget})	168		d.switchSamePeerForAttachment(effect, protocol.SamePeerSwitchRequest{RequestID: 1, Target: requestTarget})
164		169
		170	+	d.mu.Lock()
		171	+	require.Zero(t, d.purgeAdmissionActive, "the switch releases its purge admission")
		172	+	d.purgeAdmissionClosing = false
		173	+	d.mu.Unlock()
165	if tt.switched {	174		if tt.switched {
166	require.Same(t, target, ac.currentAttachmentSession())	175		require.Same(t, target, ac.currentAttachmentSession())
167	identity := decodeServerMessage(t, awaitFrame(t, sends, "CommittedRouteIdentity")).(protocol.CommittedRouteIdentity)	176		identity := decodeServerMessage(t, awaitFrame(t, sends, "CommittedRouteIdentity")).(protocol.CommittedRouteIdentity)

M internal/usecase/daemon/session.go +4 -0


@@ -896,6 +896,10 @@ func (d Daemon) startPaneGoroutines(sess session, tb tab, p pane) {
896	896		name := sess.name
897	897		sess.mu.Unlock()
898	898		d.log.Info("pane created", "session", name, "pane", p.id)
	899	+	if p.prestarted {
	900	+	// A claimed spare keeps the reader it started with.
	901	+	return
	902	+	}
899	903		// Scheduler ownership was removed; this launch creates exactly one reader.
900	904		d.sessWg.Add(1)
901	905		go d.readPanePTY(p)


@@ -896,6 +896,10 @@ func (d Daemon) startPaneGoroutines(sess session, tb tab, p pane) {
896	name := sess.name	896		name := sess.name
897	sess.mu.Unlock()	897		sess.mu.Unlock()
898	d.log.Info("pane created", "session", name, "pane", p.id)	898		d.log.Info("pane created", "session", name, "pane", p.id)
		899	+	if p.prestarted {
		900	+	// A claimed spare keeps the reader it started with.
		901	+	return
		902	+	}
899	// Scheduler ownership was removed; this launch creates exactly one reader.	903		// Scheduler ownership was removed; this launch creates exactly one reader.
900	d.sessWg.Add(1)	904		d.sessWg.Add(1)
901	go d.readPanePTY(p)	905		go d.readPanePTY(p)

M internal/usecase/daemon/snapshot_restore_incremental.go +17 -3


@@ -395,21 +395,35 @@ func sessionFromGeneration(generation ports.SnapshotGeneration) (snapcodec.Sessi
395	395		return snapcodec.Session{}, fmt.Errorf("snapshot: generation identity mismatch")
396	396		}
397	397		result := snapcodec.Session{Name: manifest.Name, CreatedAt: manifest.CreatedAt, Active: manifest.Active, Tabs: make([]snapcodec.Tab, 0, len(manifest.Tabs))}
	398	+	// Objects are compressed, so the on-disk read budget no longer bounds the
	399	+	// inflated generation; budget its payloads as one uncompressed checkpoint.
	400	+	budget := snapcodec.MaxGenerationPayloadBytes
	401	+	object := func(ref snapcodec.ObjectRef, kind snapcodec.ObjectKind) ([]byte, error) {
	402	+	data, err := generationObject(generation, ref, kind)
	403	+	if err != nil {
	404	+	return nil, err
	405	+	}
	406	+	if len(data) > budget {
	407	+	return nil, fmt.Errorf("snapshot: generation payload too large")
	408	+	}
	409	+	budget -= len(data)
	410	+	return data, nil
	411	+	}
398	412		for _, tab := range manifest.Tabs {
399	413		outTab := snapcodec.Tab{StableID: tab.StableID, Cols: tab.Cols, Rows: tab.Rows, NextPaneID: tab.NextPaneID, Focus: tab.Focus, Tree: tab.Tree, Panes: make([]snapcodec.Pane, 0, len(tab.Panes))}
400	414		for _, pane := range tab.Panes {
401	415		outPane := snapcodec.Pane{ID: pane.ID, StableID: pane.StableID, Cwd: pane.Cwd, Process: pane.Process}
402	416		for _, ref := range pane.Sealed {
403		-	data, err := generationObject(generation, ref, snapcodec.HistoryChunk)
	417	+	data, err := object(ref, snapcodec.HistoryChunk)
404	418		if err != nil {
405	419		return snapcodec.Session{}, err
406	420		}
407	421		outPane.SealedChunks = append(outPane.SealedChunks, data)
408	422		}
409		-	if outPane.Tail, err = generationObject(generation, pane.Tail, snapcodec.HistoryTail); err != nil {
	423	+	if outPane.Tail, err = object(pane.Tail, snapcodec.HistoryTail); err != nil {
410	424		return snapcodec.Session{}, err
411	425		}
412		-	if outPane.Transcript, err = generationObject(generation, pane.Transcript, snapcodec.RecoveryTranscript); err != nil {
	426	+	if outPane.Transcript, err = object(pane.Transcript, snapcodec.RecoveryTranscript); err != nil {
413	427		return snapcodec.Session{}, err
414	428		}
415	429		outTab.Panes = append(outTab.Panes, outPane)

 		return snapcodec.Session{}, fmt.Errorf("snapshot: generation identity mismatch")
 	}
 	result := snapcodec.Session{Name: manifest.Name, CreatedAt: manifest.CreatedAt, Active: manifest.Active, Tabs: make([]snapcodec.Tab, 0, len(manifest.Tabs))}
+	// Objects are compressed, so the on-disk read budget no longer bounds the
+	// inflated generation; budget its payloads as one uncompressed checkpoint.
+	budget := snapcodec.MaxGenerationPayloadBytes
+	object := func(ref snapcodec.ObjectRef, kind snapcodec.ObjectKind) ([]byte, error) {
+		data, err := generationObject(generation, ref, kind)
+		if err != nil {
+			return nil, err
+		}
+		if len(data) > budget {
+			return nil, fmt.Errorf("snapshot: generation payload too large")
+		}
+		budget -= len(data)
+		return data, nil
+	}
 	for _, tab := range manifest.Tabs {
 		outTab := snapcodec.Tab{StableID: tab.StableID, Cols: tab.Cols, Rows: tab.Rows, NextPaneID: tab.NextPaneID, Focus: tab.Focus, Tree: tab.Tree, Panes: make([]snapcodec.Pane, 0, len(tab.Panes))}
 		for _, pane := range tab.Panes {
 			outPane := snapcodec.Pane{ID: pane.ID, StableID: pane.StableID, Cwd: pane.Cwd, Process: pane.Process}
 			for _, ref := range pane.Sealed {
-				data, err := generationObject(generation, ref, snapcodec.HistoryChunk)
+				data, err := object(ref, snapcodec.HistoryChunk)
 				if err != nil {
 					return snapcodec.Session{}, err
 				}
 				outPane.SealedChunks = append(outPane.SealedChunks, data)
 			}
-			if outPane.Tail, err = generationObject(generation, pane.Tail, snapcodec.HistoryTail); err != nil {
+			if outPane.Tail, err = object(pane.Tail, snapcodec.HistoryTail); err != nil {
 				return snapcodec.Session{}, err
 			}
-			if outPane.Transcript, err = generationObject(generation, pane.Transcript, snapcodec.RecoveryTranscript); err != nil {
+			if outPane.Transcript, err = object(pane.Transcript, snapcodec.RecoveryTranscript); err != nil {
 				return snapcodec.Session{}, err
 			}
 			outTab.Panes = append(outTab.Panes, outPane)

M internal/usecase/daemon/theme_resolve.go +6 -0


@@ -51,6 +51,12 @@ func (d Daemon) applyHostThemeLocked(sess session, ac *attachedClient, t theme
51	51		p.mu.Unlock()
52	52		}
53	53		}
	54	+	// An unclaimed spare answers color queries too, so it follows the theme.
	55	+	if spare := sess.floatingSparePane(); spare != nil {
	56	+	spare.mu.Lock()
	57	+	applyPaneThemeLocked(spare, t, clearUnknownScheme)
	58	+	spare.mu.Unlock()
	59	+	}
54	60		return true
55	61		}
56	62


@@ -51,6 +51,12 @@ func (d Daemon) applyHostThemeLocked(sess session, ac *attachedClient, t theme
51	p.mu.Unlock()	51		p.mu.Unlock()
52	}	52		}
53	}	53		}
		54	+	// An unclaimed spare answers color queries too, so it follows the theme.
		55	+	if spare := sess.floatingSparePane(); spare != nil {
		56	+	spare.mu.Lock()
		57	+	applyPaneThemeLocked(spare, t, clearUnknownScheme)
		58	+	spare.mu.Unlock()
		59	+	}
54	return true	60		return true
55	}	61		}
56		62

M internal/usecase/theme/neutral_session_test.go +31 -0


@@ -4,9 +4,40 @@ import (
4	4		"testing"
5	5
6	6		renderer "github.com/bnema/vev-vt"
	7	+	"github.com/bnema/vev/internal/domain"
7	8		"github.com/stretchr/testify/require"
8	9		)
9	10
	11	+	// TestIndexedFallbackKeepsActiveTitleReadable covers a terminal that reports
	12	+	// its default colors but no palette: the scheme blue fallback must not recolor
	13	+	// the title drawn on the full-strength active surface.
	14	+	func TestIndexedFallbackKeepsActiveTitleReadable(t *testing.T) {
	15	+	black := renderer.RGB{}
	16	+	white := renderer.RGB{R: 255, G: 255, B: 255}
	17	+	for _, tt := range []struct {
	18	+	name string
	19	+	foreground, background renderer.RGB
	20	+	light bool
	21	+	}{
	22	+	{name: "dark", foreground: white, background: black},
	23	+	{name: "light", foreground: black, background: white, light: true},
	24	+	} {
	25	+	t.Run(tt.name, func(t *testing.T) {
	26	+	theme := Theme{
	27	+	Known: true, SchemeKnown: true, Light: tt.light, UsePalette: true,
	28	+	Foreground: tt.foreground, Background: tt.background, HasFG: true, HasBG: true,
	29	+	}
	30	+	resolved := Resolve(theme, domain.ThemeAccent{Mode: domain.ThemeAccentAuto})
	31	+	require.True(t, resolved.Accent.IndexedOnly)
	32	+
	33	+	want := MutedVariantStyle(resolved.Styles.SurfaceActive, theme)
	34	+	require.Equal(t, want, resolved.Styles.TabActiveTitle)
	35	+	require.Equal(t, want, resolved.Styles.TabTitleActive)
	36	+	require.Equal(t, int(resolved.Accent.Slot), resolved.Styles.TabInactiveTitle.Foreground)
	37	+	})
	38	+	}
	39	+	}
	40	+
10	41		func TestNeutralSessionSurfaceHierarchy(t *testing.T) {
11	42		black := renderer.RGB{}
12	43		white := renderer.RGB{R: 255, G: 255, B: 255}


@@ -4,9 +4,40 @@ import (
4	"testing"	4		"testing"
5		5
6	renderer "github.com/bnema/vev-vt"	6		renderer "github.com/bnema/vev-vt"
		7	+	"github.com/bnema/vev/internal/domain"
7	"github.com/stretchr/testify/require"	8		"github.com/stretchr/testify/require"
8	)	9		)
9		10
		11	+	// TestIndexedFallbackKeepsActiveTitleReadable covers a terminal that reports
		12	+	// its default colors but no palette: the scheme blue fallback must not recolor
		13	+	// the title drawn on the full-strength active surface.
		14	+	func TestIndexedFallbackKeepsActiveTitleReadable(t *testing.T) {
		15	+	black := renderer.RGB{}
		16	+	white := renderer.RGB{R: 255, G: 255, B: 255}
		17	+	for _, tt := range []struct {
		18	+	name string
		19	+	foreground, background renderer.RGB
		20	+	light bool
		21	+	}{
		22	+	{name: "dark", foreground: white, background: black},
		23	+	{name: "light", foreground: black, background: white, light: true},
		24	+	} {
		25	+	t.Run(tt.name, func(t *testing.T) {
		26	+	theme := Theme{
		27	+	Known: true, SchemeKnown: true, Light: tt.light, UsePalette: true,
		28	+	Foreground: tt.foreground, Background: tt.background, HasFG: true, HasBG: true,
		29	+	}
		30	+	resolved := Resolve(theme, domain.ThemeAccent{Mode: domain.ThemeAccentAuto})
		31	+	require.True(t, resolved.Accent.IndexedOnly)
		32	+
		33	+	want := MutedVariantStyle(resolved.Styles.SurfaceActive, theme)
		34	+	require.Equal(t, want, resolved.Styles.TabActiveTitle)
		35	+	require.Equal(t, want, resolved.Styles.TabTitleActive)
		36	+	require.Equal(t, int(resolved.Accent.Slot), resolved.Styles.TabInactiveTitle.Foreground)
		37	+	})
		38	+	}
		39	+	}
		40	+
10	func TestNeutralSessionSurfaceHierarchy(t *testing.T) {	41		func TestNeutralSessionSurfaceHierarchy(t *testing.T) {
11	black := renderer.RGB{}	42		black := renderer.RGB{}
12	white := renderer.RGB{R: 255, G: 255, B: 255}	43		white := renderer.RGB{R: 255, G: 255, B: 255}

M internal/usecase/theme/resolved_styles.go +5 -1


@@ -95,7 +95,11 @@ func indexedStyles(styles Styles, slot uint8) Styles {
95	95		styles.BorderActive = indexedForeground(styles.BorderActive, slot)
96	96		styles.BorderWarn = indexedForeground(styles.BorderWarn, slot)
97	97		styles.TabInactiveTitle = indexedForeground(styles.TabInactiveTitle, slot)
98		-	styles.TabActiveTitle = indexedForeground(styles.TabActiveTitle, slot)
	98	+	// On the full-strength neutral surface an indexed title color cannot be
	99	+	// kept readable, so the active title keeps its RGB secondary text.
	100	+	if !styles.TabActiveTitle.HasBackgroundRGB {
	101	+	styles.TabActiveTitle = indexedForeground(styles.TabActiveTitle, slot)
	102	+	}
99	103		styles.PickerDescription = indexedForeground(styles.PickerDescription, slot)
100	104		styles.PickerSeparator = indexedForeground(styles.PickerSeparator, slot)
101	105

 	styles.BorderActive = indexedForeground(styles.BorderActive, slot)
 	styles.BorderWarn = indexedForeground(styles.BorderWarn, slot)
 	styles.TabInactiveTitle = indexedForeground(styles.TabInactiveTitle, slot)
-	styles.TabActiveTitle = indexedForeground(styles.TabActiveTitle, slot)
+	// On the full-strength neutral surface an indexed title color cannot be
+	// kept readable, so the active title keeps its RGB secondary text.
+	if !styles.TabActiveTitle.HasBackgroundRGB {
+		styles.TabActiveTitle = indexedForeground(styles.TabActiveTitle, slot)
+	}
 	styles.PickerDescription = indexedForeground(styles.PickerDescription, slot)
 	styles.PickerSeparator = indexedForeground(styles.PickerSeparator, slot)

M internal/usecase/visualsearch/model.go +3 -14


@@ -2,7 +2,6 @@ package visualsearch
2	2
3	3		import (
4	4		"strconv"
5		-	"strings"
6	5
7	6		renderer "github.com/bnema/vev-vt"
8	7		"github.com/bnema/vev/internal/domain"

@@ -14,10 +13,8 @@ type Model struct {
14	13		snapshot scopy.Snapshot
15	14		input ui.TextInput
16	15		matches []scopy.SearchMatch
17		-	// matchedQuery is the trimmed query matches was computed for.
18		-	matchedQuery string
19		-	selected int
20		-	scroll int
	16	+	selected int
	17	+	scroll int
21	18		}
22	19
23	20		func New(snapshot scopy.Snapshot) *Model {

@@ -118,15 +115,7 @@ func (m *Model) Matches() []scopy.SearchMatch {
118	115		}
119	116
120	117		func (m *Model) refresh() {
121		-	doc := scopy.NewDocument(m.snapshot, "")
122		-	query := strings.TrimSpace(m.input.Value())
123		-	// Typing extends the query, so only rows that already matched can match.
124		-	if m.matchedQuery != "" && strings.HasPrefix(query, m.matchedQuery) {
125		-	m.matches = scopy.RefineMatches(doc, m.matches, query)
126		-	} else {
127		-	m.matches = scopy.FindMatches(doc, query)
128		-	}
129		-	m.matchedQuery = query
	118	+	m.matches = scopy.FindMatches(scopy.NewDocument(m.snapshot, ""), m.input.Value())
130	119		m.clamp()
131	120		m.ensureVisible(defaultVisibleRows)
132	121		}

 import (
 	"strconv"
-	"strings"
 	renderer "github.com/bnema/vev-vt"
 	"github.com/bnema/vev/internal/domain"
 	snapshot scopy.Snapshot
 	input    ui.TextInput
 	matches  []scopy.SearchMatch
-	// matchedQuery is the trimmed query matches was computed for.
+	selected int
-	matchedQuery string
+	scroll   int
-	selected     int
-	scroll       int
 }
 func New(snapshot scopy.Snapshot) *Model {
 }
 func (m *Model) refresh() {
-	doc := scopy.NewDocument(m.snapshot, "")
+	m.matches = scopy.FindMatches(scopy.NewDocument(m.snapshot, ""), m.input.Value())
-	query := strings.TrimSpace(m.input.Value())
-	// Typing extends the query, so only rows that already matched can match.
-	if m.matchedQuery != "" && strings.HasPrefix(query, m.matchedQuery) {
-		m.matches = scopy.RefineMatches(doc, m.matches, query)
-	} else {
-		m.matches = scopy.FindMatches(doc, query)
-	}
-	m.matchedQuery = query
 	m.clamp()
 	m.ensureVisible(defaultVisibleRows)
 }

M scripts/remote-resilience-harness/relay/main.go +20 -3


@@ -35,6 +35,7 @@ type relay struct {
35	35		ctx context.Context
36	36		cancel context.CancelFunc
37	37		remote string
	38	+	listen net.IP
38	39		degraded bool
39	40		blackout atomic.Bool
40	41		bytes atomic.Uint64

@@ -48,21 +49,37 @@ type relay struct {
48	49		func main() {
49	50		remote := flag.String("remote", "remote", "fixture upstream host")
50	51		control := flag.String("control", "", "required private Unix control path")
51		-	tcp := flag.String("tcp", ":2222", "SSH listener")
	52	+	// The relay forwards without authentication, so it listens on loopback
	53	+	// unless the disposable fixture asks for every interface.
	54	+	tcp := flag.String("tcp", "127.0.0.1:2222", "SSH listener")
	55	+	listen := flag.String("listen", "127.0.0.1", "UDP proxy listen IP")
52	56		degraded := flag.Bool("degraded", false, "enable seeded degraded profile")
53	57		flag.Parse()
54	58		if *control == "" {
55	59		fmt.Fprintln(os.Stderr, "-control is required")
56	60		os.Exit(2)
57	61		}
	62	+	listenIP := net.ParseIP(*listen)
	63	+	if listenIP == nil {
	64	+	fmt.Fprintln(os.Stderr, "-listen must be an IP address")
	65	+	os.Exit(2)
	66	+	}
58	67		ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
59		-	r := &relay{ctx: ctx, cancel: cancel, remote: remote, degraded: degraded, connections: make(map[net.Conn]struct{})}
	68	+	r := &relay{ctx: ctx, cancel: cancel, remote: remote, listen: listenIP, degraded: degraded, connections: make(map[net.Conn]struct{})}
60	69		if err := r.serve(tcp, control); err != nil {
61	70		fmt.Fprintln(os.Stderr, err)
62	71		os.Exit(1)
63	72		}
64	73		}
65	74
	75	+	// listenIP is the UDP proxy bind address, loopback when unset.
	76	+	func (r *relay) listenIP() net.IP {
	77	+	if r.listen == nil {
	78	+	return net.IPv4(127, 0, 0, 1)
	79	+	}
	80	+	return r.listen
	81	+	}
	82	+
66	83		func (r *relay) serve(addr, path string) error {
67	84		listener, err := net.Listen("tcp", addr)
68	85		if err != nil {

@@ -172,7 +189,7 @@ func (r *relay) apply(c command) (any, error) {
172	189		if len(r.proxies) >= 32 {
173	190		return nil, errors.New("UDP listener limit")
174	191		}
175		-	p, err := quicnettest.New(quicnettest.Config{ServerAddr: upstream, ListenAddr: &net.UDPAddr{IP: net.IPv4zero}, Seed: 317, ToServer: link, ToClient: link})
	192	+	p, err := quicnettest.New(quicnettest.Config{ServerAddr: upstream, ListenAddr: &net.UDPAddr{IP: r.listenIP()}, Seed: 317, ToServer: link, ToClient: link})
176	193		if err != nil {
177	194		return nil, err
178	195		}

 	ctx         context.Context
 	cancel      context.CancelFunc
 	remote      string
+	listen      net.IP
 	degraded    bool
 	blackout    atomic.Bool
 	bytes       atomic.Uint64
 func main() {
 	remote := flag.String("remote", "remote", "fixture upstream host")
 	control := flag.String("control", "", "required private Unix control path")
-	tcp := flag.String("tcp", ":2222", "SSH listener")
+	// The relay forwards without authentication, so it listens on loopback
+	// unless the disposable fixture asks for every interface.
+	tcp := flag.String("tcp", "127.0.0.1:2222", "SSH listener")
+	listen := flag.String("listen", "127.0.0.1", "UDP proxy listen IP")
 	degraded := flag.Bool("degraded", false, "enable seeded degraded profile")
 	flag.Parse()
 	if *control == "" {
 		fmt.Fprintln(os.Stderr, "-control is required")
 		os.Exit(2)
 	}
+	listenIP := net.ParseIP(*listen)
+	if listenIP == nil {
+		fmt.Fprintln(os.Stderr, "-listen must be an IP address")
+		os.Exit(2)
+	}
 	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
-	r := &relay{ctx: ctx, cancel: cancel, remote: *remote, degraded: *degraded, connections: make(map[net.Conn]struct{})}
+	r := &relay{ctx: ctx, cancel: cancel, remote: *remote, listen: listenIP, degraded: *degraded, connections: make(map[net.Conn]struct{})}
 	if err := r.serve(*tcp, *control); err != nil {
 		fmt.Fprintln(os.Stderr, err)
 		os.Exit(1)
 	}
 }
+// listenIP is the UDP proxy bind address, loopback when unset.
+func (r *relay) listenIP() net.IP {
+	if r.listen == nil {
+		return net.IPv4(127, 0, 0, 1)
+	}
+	return r.listen
+}
 func (r *relay) serve(addr, path string) error {
 	listener, err := net.Listen("tcp", addr)
 	if err != nil {
 		if len(r.proxies) >= 32 {
 			return nil, errors.New("UDP listener limit")
 		}
-		p, err := quicnettest.New(quicnettest.Config{ServerAddr: upstream, ListenAddr: &net.UDPAddr{IP: net.IPv4zero}, Seed: 317, ToServer: link, ToClient: link})
+		p, err := quicnettest.New(quicnettest.Config{ServerAddr: upstream, ListenAddr: &net.UDPAddr{IP: r.listenIP()}, Seed: 317, ToServer: link, ToClient: link})
 		if err != nil {
 			return nil, err
 		}

M scripts/remote-resilience-harness/visual.py +1 -1


@@ -76,7 +76,7 @@ def main():
76	76		if transport not in ("quic", "ssh"):
77	77		raise ValueError("VEV_RESILIENCE_TRANSPORT must be quic or ssh")
78	78		relay_args = ["-degraded"] if os.environ.get("VEV_RESILIENCE_DEGRADED") == "1" else []
79		-	relay = subprocess.Popen(["docker", "exec", client, "/tmp/relay", "-remote", "upstream", "-control", "/home/demo/relay/control.sock", *relay_args])
	79	+	relay = subprocess.Popen(["docker", "exec", client, "/tmp/relay", "-remote", "upstream", "-control", "/home/demo/relay/control.sock", "-tcp", ":2222", "-listen", "0.0.0.0", *relay_args])
80	80		processes.append(relay)
81	81		control_id = 0
82	82		def control(op, **fields):


@@ -76,7 +76,7 @@ def main():
76		if transport not in ("quic", "ssh"):	76		if transport not in ("quic", "ssh"):
77		raise ValueError("VEV_RESILIENCE_TRANSPORT must be quic or ssh")	77		raise ValueError("VEV_RESILIENCE_TRANSPORT must be quic or ssh")
78		relay_args = ["-degraded"] if os.environ.get("VEV_RESILIENCE_DEGRADED") == "1" else []	78		relay_args = ["-degraded"] if os.environ.get("VEV_RESILIENCE_DEGRADED") == "1" else []
79	-	relay = subprocess.Popen(["docker", "exec", client, "/tmp/relay", "-remote", "upstream", "-control", "/home/demo/relay/control.sock", *relay_args])	79	+	relay = subprocess.Popen(["docker", "exec", client, "/tmp/relay", "-remote", "upstream", "-control", "/home/demo/relay/control.sock", "-tcp", ":2222", "-listen", "0.0.0.0", *relay_args])
80		processes.append(relay)	80		processes.append(relay)
81		control_id = 0	81		control_id = 0
82		def control(op, **fields):	82		def control(op, **fields):