337ac71 fix(client): resolve a pending Esc before a resume phase ends · vev

M internal/usecase/client/attachment_outbox_test.go +55 -0


@@ -212,6 +212,61 @@ func TestResumeWatchKeepsBacklogVerbatim(t *testing.T) {
212	212		}
213	213		}
214	214
	215	+	// TestResumeWatchResolvesEscapeAtRelease presses Esc just before the phase
	216	+	// ends: release waits for the escape delay, so a lone Esc still cancels and an
	217	+	// escape sequence completed in time is kept for the session.
	218	+	func TestResumeWatchResolvesEscapeAtRelease(t *testing.T) {
	219	+	for _, tt := range []struct {
	220	+	name, after string
	221	+	wantCancel bool
	222	+	wantResidual string
	223	+	}{
	224	+	{name: "lone esc cancels", wantCancel: true},
	225	+	{name: "arrow key is kept", after: "[A", wantResidual: "\x1b[A"},
	226	+	} {
	227	+	t.Run(tt.name, func(t *testing.T) {
	228	+	reader := &inputTestReader{chunks: make(chan []byte, 2)}
	229	+	pump := newTerminalInputPump(reader)
	230	+	pump.start()
	231	+	t.Cleanup(func() {
	232	+	pump.stop()
	233	+	close(reader.chunks)
	234	+	})
	235	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
	236	+	clock := newSupervisorTestClock()
	237	+	watch := lifetime.watchResume(clock)
	238	+	reader.chunks <- []byte("\x1b")
	239	+	escape := clock.awaitTimer(t)
	240	+	released := make(chan struct{})
	241	+	go func() { watch.release(); close(released) }()
	242	+	select {
	243	+	case <-released:
	244	+	t.Fatal("release returned before the escape was resolved")
	245	+	case <-time.After(50 * time.Millisecond):
	246	+	}
	247	+	if tt.after != "" {
	248	+	reader.chunks <- []byte(tt.after)
	249	+	} else {
	250	+	escape.fire()
	251	+	}
	252	+	select {
	253	+	case <-released:
	254	+	case <-time.After(5 * time.Second):
	255	+	t.Fatal("release never returned")
	256	+	}
	257	+	select {
	258	+	case <-watch.cancelled:
	259	+	require.True(t, tt.wantCancel, "escape sequence cancelled the resume")
	260	+	default:
	261	+	require.False(t, tt.wantCancel, "lone esc at release did not cancel")
	262	+	}
	263	+	pump.mu.Lock()
	264	+	defer pump.mu.Unlock()
	265	+	require.Equal(t, tt.wantResidual, string(pump.residual))
	266	+	})
	267	+	}
	268	+	}
	269	+
215	270		func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
216	271		w := &resumeWatch{}
217	272		w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))


@@ -212,6 +212,61 @@ func TestResumeWatchKeepsBacklogVerbatim(t *testing.T) {
212	}	212		}
213	}	213		}
214		214
		215	+	// TestResumeWatchResolvesEscapeAtRelease presses Esc just before the phase
		216	+	// ends: release waits for the escape delay, so a lone Esc still cancels and an
		217	+	// escape sequence completed in time is kept for the session.
		218	+	func TestResumeWatchResolvesEscapeAtRelease(t *testing.T) {
		219	+	for _, tt := range []struct {
		220	+	name, after string
		221	+	wantCancel bool
		222	+	wantResidual string
		223	+	}{
		224	+	{name: "lone esc cancels", wantCancel: true},
		225	+	{name: "arrow key is kept", after: "[A", wantResidual: "\x1b[A"},
		226	+	} {
		227	+	t.Run(tt.name, func(t *testing.T) {
		228	+	reader := &inputTestReader{chunks: make(chan []byte, 2)}
		229	+	pump := newTerminalInputPump(reader)
		230	+	pump.start()
		231	+	t.Cleanup(func() {
		232	+	pump.stop()
		233	+	close(reader.chunks)
		234	+	})
		235	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
		236	+	clock := newSupervisorTestClock()
		237	+	watch := lifetime.watchResume(clock)
		238	+	reader.chunks <- []byte("\x1b")
		239	+	escape := clock.awaitTimer(t)
		240	+	released := make(chan struct{})
		241	+	go func() { watch.release(); close(released) }()
		242	+	select {
		243	+	case <-released:
		244	+	t.Fatal("release returned before the escape was resolved")
		245	+	case <-time.After(50 * time.Millisecond):
		246	+	}
		247	+	if tt.after != "" {
		248	+	reader.chunks <- []byte(tt.after)
		249	+	} else {
		250	+	escape.fire()
		251	+	}
		252	+	select {
		253	+	case <-released:
		254	+	case <-time.After(5 * time.Second):
		255	+	t.Fatal("release never returned")
		256	+	}
		257	+	select {
		258	+	case <-watch.cancelled:
		259	+	require.True(t, tt.wantCancel, "escape sequence cancelled the resume")
		260	+	default:
		261	+	require.False(t, tt.wantCancel, "lone esc at release did not cancel")
		262	+	}
		263	+	pump.mu.Lock()
		264	+	defer pump.mu.Unlock()
		265	+	require.Equal(t, tt.wantResidual, string(pump.residual))
		266	+	})
		267	+	}
		268	+	}
		269	+
215	func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {	270		func TestResumeWatchHeldOverflowIsSticky(t *testing.T) {
216	w := &resumeWatch{}	271		w := &resumeWatch{}
217	w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))	272		w.hold(bytes.Repeat([]byte("a"), resumeHeldInputLimit-1))

M internal/usecase/client/resume_input.go +10 -0


@@ -2,12 +2,17 @@ package client
2	2
3	3		import (
4	4		"bytes"
	5	+	"time"
5	6
6	7		"github.com/bnema/vev/internal/usecase/keys/kittykey"
7	8		)
8	9
9	10		const resumeEscapeLimit = 256
10	11
	12	+	// resumeEscapeDelay is how long a lone Esc stays undecided before it counts
	13	+	// as a cancel rather than the start of an escape sequence.
	14	+	const resumeEscapeDelay = 50 * time.Millisecond
	15	+
11	16		// resumeInputDecoder retains only an undecided escape prefix. Held bytes stay
12	17		// in their original terminal encoding; paste content is never a cancel key.
13	18		type resumeInputDecoder struct {

@@ -75,6 +80,11 @@ func (d *resumeInputDecoder) feed(data []byte) (held []byte, cancel bool) {
75	80		return held, false
76	81		}
77	82
	83	+	// undecided reports whether an escape prefix is still waiting for more bytes.
	84	+	func (d *resumeInputDecoder) undecided() bool {
	85	+	return len(d.prefix) != 0
	86	+	}
	87	+
78	88		func (d *resumeInputDecoder) loneEscape() bool {
79	89		return !d.paste && bytes.Equal(d.prefix, []byte{0x1b})
80	90		}


@@ -2,12 +2,17 @@ package client
2		2
3	import (	3		import (
4	"bytes"	4		"bytes"
		5	+	"time"
5		6
6	"github.com/bnema/vev/internal/usecase/keys/kittykey"	7		"github.com/bnema/vev/internal/usecase/keys/kittykey"
7	)	8		)
8		9
9	const resumeEscapeLimit = 256	10		const resumeEscapeLimit = 256
10		11
		12	+	// resumeEscapeDelay is how long a lone Esc stays undecided before it counts
		13	+	// as a cancel rather than the start of an escape sequence.
		14	+	const resumeEscapeDelay = 50 * time.Millisecond
		15	+
11	// resumeInputDecoder retains only an undecided escape prefix. Held bytes stay	16		// resumeInputDecoder retains only an undecided escape prefix. Held bytes stay
12	// in their original terminal encoding; paste content is never a cancel key.	17		// in their original terminal encoding; paste content is never a cancel key.
13	type resumeInputDecoder struct {	18		type resumeInputDecoder struct {

@@ -75,6 +80,11 @@ func (d *resumeInputDecoder) feed(data []byte) (held []byte, cancel bool) {
75	return held, false	80		return held, false
76	}	81		}
77		82
		83	+	// undecided reports whether an escape prefix is still waiting for more bytes.
		84	+	func (d *resumeInputDecoder) undecided() bool {
		85	+	return len(d.prefix) != 0
		86	+	}
		87	+
78	func (d *resumeInputDecoder) loneEscape() bool {	88		func (d *resumeInputDecoder) loneEscape() bool {
79	return !d.paste && bytes.Equal(d.prefix, []byte{0x1b})	89		return !d.paste && bytes.Equal(d.prefix, []byte{0x1b})
80	}	90		}

M internal/usecase/client/supervisor.go +30 -9


@@ -1507,7 +1507,8 @@ type resumeWatch struct {
1507	1507
1508	1508		// backlog counts deliveries already waiting at claim: session input the
1509	1509		// lost attachment kept, held verbatim and never read as a cancel key.
1510		-	backlog int
	1510	+	backlog int
	1511	+	releaseOnce sync.Once
1511	1512
1512	1513		mu sync.Mutex
1513	1514		held []byte

@@ -1542,13 +1543,13 @@ func (l terminalInputLifetime) watchResume(clock ports.Clock) resumeWatch {
1542	1543		close(w.done)
1543	1544		return w
1544	1545		}
1545		-	id, ok := l.pump.tryClaim()
	1546	+	id, backlog, ok := l.pump.claimWithBacklog()
1546	1547		if !ok {
1547	1548		close(w.done)
1548	1549		return w
1549	1550		}
1550	1551		w.id = id
1551		-	w.backlog = l.pump.backlog(id)
	1552	+	w.backlog = backlog
1552	1553		go w.run()
1553	1554		return w
1554	1555		}

@@ -1564,6 +1565,7 @@ func (w *resumeWatch) run() {
1564	1565		}
1565	1566		w.hold(decoder.flush())
1566	1567		}()
	1568	+	stop := w.stop
1567	1569		for {
1568	1570		var escapeC <-chan time.Time
1569	1571		if escape != nil {

@@ -1574,10 +1576,21 @@ func (w *resumeWatch) run() {
1574	1576		select {
1575	1577		case <-pump.readyFor(w.id):
1576	1578		case <-escapeC:
1577		-	close(w.cancelled)
1578		-	return
1579		-	case <-w.stop:
	1579	+	if decoder.loneEscape() {
	1580	+	close(w.cancelled)
	1581	+	}
1580	1582		return
	1583	+	case <-stop:
	1584	+	if !decoder.undecided() {
	1585	+	return
	1586	+	}
	1587	+	// Resolve an escape typed just before the phase ended: the
	1588	+	// next watch would hold it as backlog, so a cancel pressed at
	1589	+	// the boundary would reach the session instead.
	1590	+	stop = nil
	1591	+	if escape == nil {
	1592	+	escape = w.clock.NewTimer(resumeEscapeDelay)
	1593	+	}
1581	1594		case <-pump.done:
1582	1595		return
1583	1596		}

@@ -1607,18 +1620,26 @@ func (w *resumeWatch) run() {
1607	1620		escape.Stop()
1608	1621		escape = nil
1609	1622		}
1610		-	if decoder.loneEscape() {
1611		-	escape = w.clock.NewTimer(50 * time.Millisecond)
	1623	+	if stop == nil && !decoder.undecided() {
	1624	+	return
	1625	+	}
	1626	+	if decoder.loneEscape() \|\| stop == nil {
	1627	+	escape = w.clock.NewTimer(resumeEscapeDelay)
1612	1628		}
1613	1629		}
1614	1630		}
1615	1631
1616	1632		// release stops the watch and hands kept keys back to the pump, unless the
1617		-	// user cancelled, in which case they are dropped with the session.
	1633	+	// user cancelled, in which case they are dropped with the session. An escape
	1634	+	// still undecided is resolved first, so it may cancel. Later calls are no-ops.
1618	1635		func (w *resumeWatch) release() {
1619	1636		if w.id == 0 {
1620	1637		return
1621	1638		}
	1639	+	w.releaseOnce.Do(w.releaseClaim)
	1640	+	}
	1641	+
	1642	+	func (w *resumeWatch) releaseClaim() {
1622	1643		close(w.stop)
1623	1644		<-w.done
1624	1645		pump := w.lifetime.pump

 	// backlog counts deliveries already waiting at claim: session input the
 	// lost attachment kept, held verbatim and never read as a cancel key.
-	backlog int
+	backlog     int
+	releaseOnce sync.Once
 	mu   sync.Mutex
 	held []byte
 		close(w.done)
 		return w
 	}
-	id, ok := l.pump.tryClaim()
+	id, backlog, ok := l.pump.claimWithBacklog()
 	if !ok {
 		close(w.done)
 		return w
 	}
 	w.id = id
-	w.backlog = l.pump.backlog(id)
+	w.backlog = backlog
 	go w.run()
 	return w
 }
 		}
 		w.hold(decoder.flush())
 	}()
+	stop := w.stop
 	for {
 		var escapeC <-chan time.Time
 		if escape != nil {
 			select {
 			case <-pump.readyFor(w.id):
 			case <-escapeC:
-				close(w.cancelled)
+				if decoder.loneEscape() {
-				return
+					close(w.cancelled)
-			case <-w.stop:
+				}
 				return
+			case <-stop:
+				if !decoder.undecided() {
+					return
+				}
+				// Resolve an escape typed just before the phase ended: the
+				// next watch would hold it as backlog, so a cancel pressed at
+				// the boundary would reach the session instead.
+				stop = nil
+				if escape == nil {
+					escape = w.clock.NewTimer(resumeEscapeDelay)
+				}
 			case <-pump.done:
 				return
 			}
 			escape.Stop()
 			escape = nil
 		}
-		if decoder.loneEscape() {
+		if stop == nil && !decoder.undecided() {
-			escape = w.clock.NewTimer(50 * time.Millisecond)
+			return
+		}
+		if decoder.loneEscape() || stop == nil {
+			escape = w.clock.NewTimer(resumeEscapeDelay)
 		}
 	}
 }
 // release stops the watch and hands kept keys back to the pump, unless the
-// user cancelled, in which case they are dropped with the session.
+// user cancelled, in which case they are dropped with the session. An escape
+// still undecided is resolved first, so it may cancel. Later calls are no-ops.
 func (w *resumeWatch) release() {
 	if w.id == 0 {
 		return
 	}
+	w.releaseOnce.Do(w.releaseClaim)
+}
+func (w *resumeWatch) releaseClaim() {
 	close(w.stop)
 	<-w.done
 	pump := w.lifetime.pump

M internal/usecase/client/supervisor_attach.go +10 -0


@@ -1011,6 +1011,16 @@ func (s Supervisor) waitResume(ctx context.Context, input terminalInputLifetim
1011	1011		s.transition(supervisorEvent{kind: supervisorAttachEnded})
1012	1012		return false, nil, true
1013	1013		case <-timer.C():
	1014	+	// Release resolves an Esc typed just before the backoff ended,
	1015	+	// which may still cancel.
	1016	+	keys.release()
	1017	+	select {
	1018	+	case <-keys.cancelled:
	1019	+	s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
	1020	+	s.transition(supervisorEvent{kind: supervisorAttachEnded})
	1021	+	return false, nil, true
	1022	+	default:
	1023	+	}
1014	1024		return false, nil, false
1015	1025		case <-ctx.Done():
1016	1026		return true, ctx.Err(), false


@@ -1011,6 +1011,16 @@ func (s Supervisor) waitResume(ctx context.Context, input terminalInputLifetim
1011	s.transition(supervisorEvent{kind: supervisorAttachEnded})	1011		s.transition(supervisorEvent{kind: supervisorAttachEnded})
1012	return false, nil, true	1012		return false, nil, true
1013	case <-timer.C():	1013		case <-timer.C():
		1014	+	// Release resolves an Esc typed just before the backoff ended,
		1015	+	// which may still cancel.
		1016	+	keys.release()
		1017	+	select {
		1018	+	case <-keys.cancelled:
		1019	+	s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
		1020	+	s.transition(supervisorEvent{kind: supervisorAttachEnded})
		1021	+	return false, nil, true
		1022	+	default:
		1023	+	}
1014	return false, nil, false	1024		return false, nil, false
1015	case <-ctx.Done():	1025		case <-ctx.Done():
1016	return true, ctx.Err(), false	1026		return true, ctx.Err(), false

M internal/usecase/client/terminal_input_pump.go +17 -21


@@ -105,10 +105,25 @@ func newTerminalInputPump(in io.Reader) *terminalInputPump {
105	105
106	106		// tryClaim lets competing foreground owners decline admission without panicking.
107	107		func (p *terminalInputPump) tryClaim() (uint64, bool) {
	108	+	consumer, _, ok := p.claimWithBacklog()
	109	+	return consumer, ok
	110	+	}
	111	+
	112	+	// claimWithBacklog claims the pump and reports, atomically with the claim, how
	113	+	// many deliveries were already waiting: the residual, then the pending read.
	114	+	// They were read for an earlier owner, not typed for this one.
	115	+	func (p *terminalInputPump) claimWithBacklog() (uint64, int, bool) {
108	116		p.mu.Lock()
109	117		if p.consumer != 0 {
110	118		p.mu.Unlock()
111		-	return 0, false
	119	+	return 0, 0, false
	120	+	}
	121	+	backlog := 0
	122	+	if len(p.residual) != 0 {
	123	+	backlog++
	124	+	}
	125	+	if p.pending != nil {
	126	+	backlog++
112	127		}
113	128		p.nextID++
114	129		p.consumer = p.nextID

@@ -122,26 +137,7 @@ func (p *terminalInputPump) tryClaim() (uint64, bool) {
122	137		if ready {
123	138		p.signalReady(consumer)
124	139		}
125		-	return consumer, true
126		-	}
127		-
128		-	// backlog reports how many deliveries were already waiting when consumer
129		-	// claimed the pump: the residual, then the pending read. They were read for an
130		-	// earlier owner, not typed for consumer.
131		-	func (p *terminalInputPump) backlog(consumer uint64) int {
132		-	p.mu.Lock()
133		-	defer p.mu.Unlock()
134		-	if p.consumer != consumer {
135		-	return 0
136		-	}
137		-	n := 0
138		-	if len(p.residual) != 0 {
139		-	n++
140		-	}
141		-	if p.pending != nil {
142		-	n++
143		-	}
144		-	return n
	140	+	return consumer, backlog, true
145	141		}
146	142
147	143		// revoke invalidates an attempt before its replacement is allowed to claim

 // tryClaim lets competing foreground owners decline admission without panicking.
 func (p *terminalInputPump) tryClaim() (uint64, bool) {
+	consumer, _, ok := p.claimWithBacklog()
+	return consumer, ok
+}
+// claimWithBacklog claims the pump and reports, atomically with the claim, how
+// many deliveries were already waiting: the residual, then the pending read.
+// They were read for an earlier owner, not typed for this one.
+func (p *terminalInputPump) claimWithBacklog() (uint64, int, bool) {
 	p.mu.Lock()
 	if p.consumer != 0 {
 		p.mu.Unlock()
-		return 0, false
+		return 0, 0, false
+	}
+	backlog := 0
+	if len(p.residual) != 0 {
+		backlog++
+	}
+	if p.pending != nil {
+		backlog++
 	}
 	p.nextID++
 	p.consumer = p.nextID
 	if ready {
 		p.signalReady(consumer)
 	}
-	return consumer, true
+	return consumer, backlog, true
-}
-// backlog reports how many deliveries were already waiting when consumer
-// claimed the pump: the residual, then the pending read. They were read for an
-// earlier owner, not typed for consumer.
-func (p *terminalInputPump) backlog(consumer uint64) int {
-	p.mu.Lock()
-	defer p.mu.Unlock()
-	if p.consumer != consumer {
-		return 0
-	}
-	n := 0
-	if len(p.residual) != 0 {
-		n++
-	}
-	if p.pending != nil {
-		n++
-	}
-	return n
 }
 // revoke invalidates an attempt before its replacement is allowed to claim