a4e74c2 fix(client): address resilience review findings · vev

M docs/plans/quic-resilience.md +22 -14


@@ -22,22 +22,23 @@ Reference: mosh (`/tmp/mosh`, `src/network/transportsender-impl.h`,
22	22		6. Daemon ignores `Input.InputSeq`, so input in flight at a drop is lost or
23	23		would be duplicated by a replay.
24	24
25		-	## Phase 1 — non-blocking client send (client only)
	25	+	## Phase 1 — non-blocking client send (done)
26	26
27		-	- Per-attachment ordered outbound queue drained by one sender goroutine;
28		-	the worker loop enqueues and keeps running. Failure of the sender surfaces
29		-	as one event into the loop, which settles as today.
30		-	- Bounded by bytes; when full, the loop stops reading terminal input
31		-	(backpressure) instead of blocking UI work.
32		-	- Tests: a stalled `SendClient` must not block overlay/repaint/output
33		-	handling; ordering preserved; error settles the run.
	27	+	- Per-attachment ordered outbox drained by one sender goroutine; the worker
	28	+	loop enqueues and keeps running. Write failure settles the run as before.
	29	+	- Bounded by count and bytes. An input delivery stays undecided until the
	30	+	outbox wrote it, so the pump reads no newer keys meanwhile (backpressure)
	31	+	and a link failure preserves exactly the unwritten keys for the resume.
	32	+	- Lifecycle exits drain the outbox for at most 2 s.
34	33
35		-	## Phase 2 — faster detection, patient resume (client + QUIC adapter)
	34	+	## Phase 2 — patient resume (done, except status text)
36	35
37		-	- QUIC keepalive ~2 s; keep idle timeout long (resume covers longer gaps).
38		-	- Resume loop: retry with capped backoff until the remote park window
39		-	(15 min) expires instead of 5 attempts; Ctrl-C/detach still exit.
40		-	- Status bar shows "last contact Ns ago" while resuming.
	36	+	- QUIC keepalive 2 s; idle timeout stays 60 s.
	37	+	- A route that stays down keeps resuming for the 15-minute park window; a
	38	+	session that drops right after attaching still gives up after 5 flaps.
	39	+	- During the resume backoff, Ctrl-C or a lone Esc returns to the picker;
	40	+	other keys are kept (bounded) and replayed into the resumed session.
	41	+	- TODO: status bar "last contact Ns ago" while resuming.
41	42
42	43		## Phase 3 — exactly-once input across resume (protocol bump)
43	44

@@ -60,7 +61,14 @@ Reference: mosh (`/tmp/mosh`, `src/network/transportsender-impl.h`,
60	61		- Derive `degraded`/`probing` from RTT and last-packet age in the adapter.
61	62		- On stall or local address change, `AddPath` + probe + switch.
62	63
63		-	## Phase 6 — optional predictive local echo
	64	+	## Phase 6 — client-rendered palette on remote sessions
	65	+
	66	+	- The command palette is rendered by the serving daemon, so each keystroke
	67	+	costs a full round trip on a remote session. Render it client-side like
	68	+	the session picker, fed by the daemon's command catalogue, and send only
	69	+	the chosen command.
	70	+
	71	+	## Phase 7 — optional predictive local echo
64	72
65	73		- mosh-style overlay with underline-until-confirmed. Breaks the "thin client
66	74		interprets nothing" rule; needs an explicit design decision.


@@ -22,22 +22,23 @@ Reference: mosh (`/tmp/mosh`, `src/network/transportsender-impl.h`,
22		6. Daemon ignores `Input.InputSeq`, so input in flight at a drop is lost or	22		6. Daemon ignores `Input.InputSeq`, so input in flight at a drop is lost or
23		would be duplicated by a replay.	23		would be duplicated by a replay.
24			24
25	-	## Phase 1 — non-blocking client send (client only)	25	+	## Phase 1 — non-blocking client send (done)
26			26
27	-	- Per-attachment ordered outbound queue drained by one sender goroutine;	27	+	- Per-attachment ordered outbox drained by one sender goroutine; the worker
28	-	the worker loop enqueues and keeps running. Failure of the sender surfaces	28	+	loop enqueues and keeps running. Write failure settles the run as before.
29	-	as one event into the loop, which settles as today.	29	+	- Bounded by count and bytes. An input delivery stays undecided until the
30	-	- Bounded by bytes; when full, the loop stops reading terminal input	30	+	outbox wrote it, so the pump reads no newer keys meanwhile (backpressure)
31	-	(backpressure) instead of blocking UI work.	31	+	and a link failure preserves exactly the unwritten keys for the resume.
32	-	- Tests: a stalled `SendClient` must not block overlay/repaint/output	32	+	- Lifecycle exits drain the outbox for at most 2 s.
33	-	handling; ordering preserved; error settles the run.
34			33
35	-	## Phase 2 — faster detection, patient resume (client + QUIC adapter)	34	+	## Phase 2 — patient resume (done, except status text)
36			35
37	-	- QUIC keepalive ~2 s; keep idle timeout long (resume covers longer gaps).	36	+	- QUIC keepalive 2 s; idle timeout stays 60 s.
38	-	- Resume loop: retry with capped backoff until the remote park window	37	+	- A route that stays down keeps resuming for the 15-minute park window; a
39	-	(15 min) expires instead of 5 attempts; Ctrl-C/detach still exit.	38	+	session that drops right after attaching still gives up after 5 flaps.
40	-	- Status bar shows "last contact Ns ago" while resuming.	39	+	- During the resume backoff, Ctrl-C or a lone Esc returns to the picker;
			40	+	other keys are kept (bounded) and replayed into the resumed session.
			41	+	- TODO: status bar "last contact Ns ago" while resuming.
41			42
42		## Phase 3 — exactly-once input across resume (protocol bump)	43		## Phase 3 — exactly-once input across resume (protocol bump)
43			44

@@ -60,7 +61,14 @@ Reference: mosh (`/tmp/mosh`, `src/network/transportsender-impl.h`,
60		- Derive `degraded`/`probing` from RTT and last-packet age in the adapter.	61		- Derive `degraded`/`probing` from RTT and last-packet age in the adapter.
61		- On stall or local address change, `AddPath` + probe + switch.	62		- On stall or local address change, `AddPath` + probe + switch.
62			63
63	-	## Phase 6 — optional predictive local echo	64	+	## Phase 6 — client-rendered palette on remote sessions
			65	+
			66	+	- The command palette is rendered by the serving daemon, so each keystroke
			67	+	costs a full round trip on a remote session. Render it client-side like
			68	+	the session picker, fed by the daemon's command catalogue, and send only
			69	+	the chosen command.
			70	+
			71	+	## Phase 7 — optional predictive local echo
64			72
65		- mosh-style overlay with underline-until-confirmed. Breaks the "thin client	73		- mosh-style overlay with underline-until-confirmed. Breaks the "thin client
66		interprets nothing" rule; needs an explicit design decision.	74		interprets nothing" rule; needs an explicit design decision.

M docs/remote-resilience.md +2 -2


@@ -25,8 +25,8 @@ The transport is fixed when the host is added (`quic` by default). To switch, ru
25	25		## What survives a network problem
26	26
27	27		- The remote session and its shells keep running while you are offline.
28		-	- The client reconnects by itself after a network change, a VPN switch, or a laptop sleep, and keeps retrying for up to 15 minutes of outage before it returns to the picker.
29		-	- A stalled link never freezes the client: typing, overlays, and prompts stay responsive, and input is sent in order once the link moves again.
	28	+	- The client reconnects by itself after a network change, a VPN switch, or a laptop sleep, and keeps retrying for up to 15 minutes of outage before it returns to the picker. Press Ctrl-C or Esc while it reconnects to give up and return to the picker.
	29	+	- A stalled link does not block the client loop. Keys typed while the link stalls or reconnects are kept and sent in order once it recovers; keys that never reached the remote before a reconnect are replayed to the resumed session.
30	30		- If the reconnect window has expired but the session still exists, the client opens a fresh attachment to it.
31	31		- Several clients can attach to the same session. Each keeps its own window, tab, focus, and copy mode.
32	32


@@ -25,8 +25,8 @@ The transport is fixed when the host is added (`quic` by default). To switch, ru
25		## What survives a network problem	25		## What survives a network problem
26			26
27		- The remote session and its shells keep running while you are offline.	27		- The remote session and its shells keep running while you are offline.
28	-	- The client reconnects by itself after a network change, a VPN switch, or a laptop sleep, and keeps retrying for up to 15 minutes of outage before it returns to the picker.	28	+	- The client reconnects by itself after a network change, a VPN switch, or a laptop sleep, and keeps retrying for up to 15 minutes of outage before it returns to the picker. Press Ctrl-C or Esc while it reconnects to give up and return to the picker.
29	-	- A stalled link never freezes the client: typing, overlays, and prompts stay responsive, and input is sent in order once the link moves again.	29	+	- A stalled link does not block the client loop. Keys typed while the link stalls or reconnects are kept and sent in order once it recovers; keys that never reached the remote before a reconnect are replayed to the resumed session.
30		- If the reconnect window has expired but the session still exists, the client opens a fresh attachment to it.	30		- If the reconnect window has expired but the session still exists, the client opens a fresh attachment to it.
31		- Several clients can attach to the same session. Each keeps its own window, tab, focus, and copy mode.	31		- Several clients can attach to the same session. Each keeps its own window, tab, focus, and copy mode.
32			32

M internal/adapters/quic/quic.go +7 -4


@@ -62,10 +62,13 @@ var (
62	62		// idle limits, bounded admission, no 0-RTT.
63	63		type Config struct {
64	64		// KeepAlivePeriod defaults to 2s; MaxIdleTimeout defaults to 60s;
65		-	// HandshakeIdleTimeout defaults to 5s. The short keepalive holds mobile
66		-	// CGNAT bindings open and surfaces loss quickly, while the long idle
67		-	// timeout lets one connection ride out short outages and deliver queued
68		-	// input without reconnecting.
	65	+	// HandshakeIdleTimeout defaults to 5s. The short keepalive keeps mobile
	66	+	// CGNAT bindings open and probes the path often enough that loss
	67	+	// recovery restarts promptly once the link returns. It does not shorten
	68	+	// failure detection: a dead path is still declared only after
	69	+	// MaxIdleTimeout, which stays long so one connection rides out short
	70	+	// outages and delivers queued input without reconnecting. Both roles use
	71	+	// it; a 2s PING is a few dozen bytes per connection.
69	72		KeepAlivePeriod time.Duration
70	73		MaxIdleTimeout time.Duration
71	74		HandshakeIdleTimeout time.Duration

 // idle limits, bounded admission, no 0-RTT.
 type Config struct {
 	// KeepAlivePeriod defaults to 2s; MaxIdleTimeout defaults to 60s;
-	// HandshakeIdleTimeout defaults to 5s. The short keepalive holds mobile
+	// HandshakeIdleTimeout defaults to 5s. The short keepalive keeps mobile
-	// CGNAT bindings open and surfaces loss quickly, while the long idle
+	// CGNAT bindings open and probes the path often enough that loss
-	// timeout lets one connection ride out short outages and deliver queued
+	// recovery restarts promptly once the link returns. It does not shorten
-	// input without reconnecting.
+	// failure detection: a dead path is still declared only after
+	// MaxIdleTimeout, which stays long so one connection rides out short
+	// outages and delivers queued input without reconnecting. Both roles use
+	// it; a 2s PING is a few dozen bytes per connection.
 	KeepAlivePeriod      time.Duration
 	MaxIdleTimeout       time.Duration
 	HandshakeIdleTimeout time.Duration

M internal/usecase/client/attachment_input_test.go +14 -2


@@ -33,6 +33,7 @@ func (r *inputTestReader) Read(p []byte) (int, error) {
33	33
34	34		type inputHarness struct {
35	35		stream *sessionTestStream
	36	+	pump *terminalInputPump
36	37		term *workerTestTerminal
37	38		clock *supervisorTestClock
38	39		reader *inputTestReader

@@ -49,6 +50,16 @@ func startInputHarness(t testing.T, themes terminalThemeState) *inputHarness {
49	50
50	51		func startInputHarnessWithConfig(t testing.T, themes terminalThemeState, request ports.BrokerOpenStreamRequest, clipboard ports.ClipboardReader) *inputHarness {
51	52		t.Helper()
	53	+	return startInputHarnessOn(t, themes, request, clipboard, newSessionTestStream(), nil)
	54	+	}
	55	+
	56	+	// startInputHarnessOn runs the harness over stream; transport, when non-nil,
	57	+	// is what the worker writes through (a wrapper over stream).
	58	+	func startInputHarnessOn(t testing.T, themes terminalThemeState, request ports.BrokerOpenStreamRequest, clipboard ports.ClipboardReader, stream sessionTestStream, transport ports.BrokerLogicalConnection) inputHarness {
	59	+	t.Helper()
	60	+	if transport == nil {
	61	+	transport = stream
	62	+	}
52	63		reader := &inputTestReader{chunks: make(chan []byte, 16)}
53	64		pump := newTerminalInputPump(reader)
54	65		pump.start()

@@ -57,7 +68,8 @@ func startInputHarnessWithConfig(t testing.T, themes terminalThemeState, reque
57	68		close(reader.chunks)
58	69		})
59	70		h := &inputHarness{
60		-	stream: newSessionTestStream(),
	71	+	stream: stream,
	72	+	pump: pump,
61	73		term: newWorkerTestTerminal(),
62	74		clock: newSupervisorTestClock(),
63	75		reader: reader,

@@ -72,7 +84,7 @@ func startInputHarnessWithConfig(t testing.T, themes terminalThemeState, reque
72	84		worker, err := newSessionAttachmentWorker(cfg)
73	85		require.NoError(t, err)
74	86		go func() {
75		-	event, _ := host.Run(context.Background(), AttachmentToken{Generation: 4, Attempt: 1}, worker, h.stream)
	87	+	event, _ := host.Run(context.Background(), AttachmentToken{Generation: 4, Attempt: 1}, worker, transport)
76	88		h.events <- event
77	89		}()
78	90		awaitHello(t, h.stream)

 type inputHarness struct {
 	stream *sessionTestStream
+	pump   *terminalInputPump
 	term   *workerTestTerminal
 	clock  *supervisorTestClock
 	reader *inputTestReader
 func startInputHarnessWithConfig(t *testing.T, themes *terminalThemeState, request ports.BrokerOpenStreamRequest, clipboard ports.ClipboardReader) *inputHarness {
 	t.Helper()
+	return startInputHarnessOn(t, themes, request, clipboard, newSessionTestStream(), nil)
+}
+// startInputHarnessOn runs the harness over stream; transport, when non-nil,
+// is what the worker writes through (a wrapper over stream).
+func startInputHarnessOn(t *testing.T, themes *terminalThemeState, request ports.BrokerOpenStreamRequest, clipboard ports.ClipboardReader, stream *sessionTestStream, transport ports.BrokerLogicalConnection) *inputHarness {
+	t.Helper()
+	if transport == nil {
+		transport = stream
+	}
 	reader := &inputTestReader{chunks: make(chan []byte, 16)}
 	pump := newTerminalInputPump(reader)
 	pump.start()
 		close(reader.chunks)
 	})
 	h := &inputHarness{
-		stream: newSessionTestStream(),
+		stream: stream,
+		pump:   pump,
 		term:   newWorkerTestTerminal(),
 		clock:  newSupervisorTestClock(),
 		reader: reader,
 	worker, err := newSessionAttachmentWorker(cfg)
 	require.NoError(t, err)
 	go func() {
-		event, _ := host.Run(context.Background(), AttachmentToken{Generation: 4, Attempt: 1}, worker, h.stream)
+		event, _ := host.Run(context.Background(), AttachmentToken{Generation: 4, Attempt: 1}, worker, transport)
 		h.events <- event
 	}()
 	awaitHello(t, h.stream)

M internal/usecase/client/attachment_outbox.go +99 -13

@@ -1,6 +1,7 @@
1	1		package client
2	2
3	3		import (
	4	+	"context"
4	5		"errors"
5	6		"sync"
6	7

@@ -8,28 +9,35 @@ import (
8	9		"github.com/bnema/vev/internal/protocol"
9	10		)
10	11
11		-	// outboxMaxPending bounds client messages accepted but not yet written. Past
12		-	// it SendClient blocks, which applies backpressure to the worker loop only
13		-	// after a large amount of input has piled up behind a stalled link.
14		-	const outboxMaxPending = 4096
	12	+	// Outbox bounds. Past either, enqueue blocks until the sender makes progress
	13	+	// or the caller is cancelled. Input is acknowledged to the terminal pump only
	14	+	// once written, so the pump stops reading new keys long before these limits;
	15	+	// they bound acks, focus reports, and other control traffic.
	16	+	const (
	17	+	outboxMaxPending = 4096
	18	+	outboxMaxBytes = 4 << 20
	19	+	)
15	20
16	21		// errOutboxStopped is returned by SendClient after the outbox stopped.
17	22		var errOutboxStopped = errors.New("client: attachment outbox stopped")
18	23
19	24		// attachmentOutbox decouples the attachment worker loop from the carriage:
20		-	// SendClient enqueues and returns, and one sender goroutine writes messages in
21		-	// order. A stalled link therefore no longer freezes key handling, overlays, or
22		-	// painting. The first write failure is sticky: it is reported on failed and
23		-	// returned by every later SendClient.
	25	+	// enqueue returns once the message is accepted, and one sender goroutine
	26	+	// writes messages in order. A stalled link therefore no longer freezes key
	27	+	// handling, overlays, or painting. The first write failure is sticky: it is
	28	+	// reported on Failed and returned by every later enqueue.
24	29		type attachmentOutbox struct {
25	30		ports.BrokerLogicalConnection
26	31
27	32		mu sync.Mutex
28	33		queue []protocol.ClientMessage
	34	+	bytes int // payload bytes queued
	35	+	queued uint64 // messages ever accepted
	36	+	written uint64 // messages written successfully
29	37		err error
30	38		stopped bool
31	39		wake chan struct{} // sender wake-up, capacity 1
32		-	space chan struct{} // closed-and-replaced when the queue shrinks
	40	+	space chan struct{} // closed-and-replaced after each write, closed on failure
33	41		idle chan struct{} // closed-and-replaced when the queue drains
34	42		failed chan struct{} // closed once on the first write failure
35	43		done chan struct{} // closed when the sender exits

@@ -49,9 +57,28 @@ func newAttachmentOutbox(stream ports.BrokerLogicalConnection) *attachmentOutbox
49	57		return b
50	58		}
51	59
52		-	// SendClient enqueues message for ordered transmission. It blocks only while
53		-	// the queue is full, and returns the sticky failure once the sender failed.
	60	+	// outboxPayload is the variable-size part of a message counted against
	61	+	// outboxMaxBytes.
	62	+	func outboxPayload(message protocol.ClientMessage) int {
	63	+	switch m := message.(type) {
	64	+	case protocol.Input:
	65	+	return len(m.Data)
	66	+	case protocol.ImagePush:
	67	+	return len(m.Data)
	68	+	}
	69	+	return 0
	70	+	}
	71	+
	72	+	// SendClient enqueues message without a cancellation source.
54	73		func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
	74	+	return b.enqueue(context.Background(), nil, message)
	75	+	}
	76	+
	77	+	// enqueue accepts message for ordered transmission. It blocks only while the
	78	+	// outbox is full, until space frees, ctx ends, or cancel closes. It returns
	79	+	// the sticky failure once the sender failed.
	80	+	func (b *attachmentOutbox) enqueue(ctx context.Context, cancel <-chan struct{}, message protocol.ClientMessage) error {
	81	+	size := outboxPayload(message)
55	82		b.mu.Lock()
56	83		for {
57	84		if b.err != nil {

@@ -63,7 +90,8 @@ func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
63	90		b.mu.Unlock()
64	91		return errOutboxStopped
65	92		}
66		-	if len(b.queue) < outboxMaxPending {
	93	+	// An empty queue always admits, so one oversized message still goes.
	94	+	if len(b.queue) == 0 \|\| (len(b.queue) < outboxMaxPending && b.bytes+size <= outboxMaxBytes) {
67	95		break
68	96		}
69	97		space := b.space

@@ -71,6 +99,10 @@ func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
71	99		select {
72	100		case <-space:
73	101		case <-b.done:
	102	+	case <-ctx.Done():
	103	+	return ctx.Err()
	104	+	case <-cancel:
	105	+	return errAttachmentForegroundRevoked
74	106		}
75	107		b.mu.Lock()
76	108		}

@@ -78,6 +110,8 @@ func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
78	110		b.idle = make(chan struct{})
79	111		}
80	112		b.queue = append(b.queue, message)
	113	+	b.bytes += size
	114	+	b.queued++
81	115		b.mu.Unlock()
82	116		select {
83	117		case b.wake <- struct{}{}:

@@ -94,6 +128,8 @@ func (b *attachmentOutbox) Close() error {
94	128		}
95	129
96	130		// stop ends the sender after its current write; queued messages are dropped.
	131	+	// Session input among them was never acknowledged to the terminal pump, so
	132	+	// the worker preserves it like a failed synchronous send.
97	133		func (b *attachmentOutbox) stop() {
98	134		b.mu.Lock()
99	135		b.stopped = true

@@ -104,6 +140,53 @@ func (b *attachmentOutbox) stop() {
104	140		}
105	141		}
106	142
	143	+	// unsentInput returns the session input bytes of accepted messages numbered
	144	+	// (from, to] that were never written. It is final only after join.
	145	+	func (b *attachmentOutbox) unsentInput(from, to uint64) []byte {
	146	+	b.mu.Lock()
	147	+	defer b.mu.Unlock()
	148	+	var data []byte
	149	+	// queue[i] is accepted message number written+1+i.
	150	+	for i, message := range b.queue {
	151	+	n := b.written + 1 + uint64(i)
	152	+	if n <= from \|\| n > to {
	153	+	continue
	154	+	}
	155	+	if input, ok := message.(protocol.Input); ok {
	156	+	data = append(data, input.Data...)
	157	+	}
	158	+	}
	159	+	return data
	160	+	}
	161	+
	162	+	// join stops the sender, closes the stream so a stalled write returns, and
	163	+	// waits for the sender to exit. Afterwards writtenCount is final.
	164	+	func (b *attachmentOutbox) join() {
	165	+	_ = b.Close()
	166	+	<-b.done
	167	+	}
	168	+
	169	+	// accepted reports how many messages were ever accepted.
	170	+	func (b *attachmentOutbox) accepted() uint64 {
	171	+	b.mu.Lock()
	172	+	defer b.mu.Unlock()
	173	+	return b.queued
	174	+	}
	175	+
	176	+	// writtenCount reports how many messages were written successfully.
	177	+	func (b *attachmentOutbox) writtenCount() uint64 {
	178	+	b.mu.Lock()
	179	+	defer b.mu.Unlock()
	180	+	return b.written
	181	+	}
	182	+
	183	+	// Progress is closed after the next successful write or on failure.
	184	+	func (b *attachmentOutbox) Progress() <-chan struct{} {
	185	+	b.mu.Lock()
	186	+	defer b.mu.Unlock()
	187	+	return b.space
	188	+	}
	189	+
107	190		// Failed is closed once a write failed; failure returns the cause.
108	191		func (b *attachmentOutbox) Failed() <-chan struct{} { return b.failed }
109	192

@@ -140,8 +223,9 @@ func (b *attachmentOutbox) run() {
140	223
141	224		b.mu.Lock()
142	225		if err != nil {
	226	+	// The queue is kept, failed write first, so unsentInput can
	227	+	// hand undelivered keys back to the terminal pump.
143	228		b.err = err
144		-	b.queue = nil
145	229		close(b.failed)
146	230		close(b.idle)
147	231		close(b.space)

@@ -150,6 +234,8 @@ func (b *attachmentOutbox) run() {
150	234		}
151	235		b.queue[0] = nil
152	236		b.queue = b.queue[1:]
	237	+	b.bytes -= outboxPayload(message)
	238	+	b.written++
153	239		close(b.space)
154	240		b.space = make(chan struct{})
155	241		if len(b.queue) == 0 {

 package client
 import (
+	"context"
 	"errors"
 	"sync"
 	"github.com/bnema/vev/internal/protocol"
 )
-// outboxMaxPending bounds client messages accepted but not yet written. Past
+// Outbox bounds. Past either, enqueue blocks until the sender makes progress
-// it SendClient blocks, which applies backpressure to the worker loop only
+// or the caller is cancelled. Input is acknowledged to the terminal pump only
-// after a large amount of input has piled up behind a stalled link.
+// once written, so the pump stops reading new keys long before these limits;
-const outboxMaxPending = 4096
+// they bound acks, focus reports, and other control traffic.
+const (
+	outboxMaxPending = 4096
+	outboxMaxBytes   = 4 << 20
+)
 // errOutboxStopped is returned by SendClient after the outbox stopped.
 var errOutboxStopped = errors.New("client: attachment outbox stopped")
 // attachmentOutbox decouples the attachment worker loop from the carriage:
-// SendClient enqueues and returns, and one sender goroutine writes messages in
+// enqueue returns once the message is accepted, and one sender goroutine
-// order. A stalled link therefore no longer freezes key handling, overlays, or
+// writes messages in order. A stalled link therefore no longer freezes key
-// painting. The first write failure is sticky: it is reported on failed and
+// handling, overlays, or painting. The first write failure is sticky: it is
-// returned by every later SendClient.
+// reported on Failed and returned by every later enqueue.
 type attachmentOutbox struct {
 	ports.BrokerLogicalConnection
 	mu      sync.Mutex
 	queue   []protocol.ClientMessage
+	bytes   int    // payload bytes queued
+	queued  uint64 // messages ever accepted
+	written uint64 // messages written successfully
 	err     error
 	stopped bool
 	wake    chan struct{} // sender wake-up, capacity 1
-	space   chan struct{} // closed-and-replaced when the queue shrinks
+	space   chan struct{} // closed-and-replaced after each write, closed on failure
 	idle    chan struct{} // closed-and-replaced when the queue drains
 	failed  chan struct{} // closed once on the first write failure
 	done    chan struct{} // closed when the sender exits
 	return b
 }
-// SendClient enqueues message for ordered transmission. It blocks only while
+// outboxPayload is the variable-size part of a message counted against
-// the queue is full, and returns the sticky failure once the sender failed.
+// outboxMaxBytes.
+func outboxPayload(message protocol.ClientMessage) int {
+	switch m := message.(type) {
+	case protocol.Input:
+		return len(m.Data)
+	case protocol.ImagePush:
+		return len(m.Data)
+	}
+	return 0
+}
+// SendClient enqueues message without a cancellation source.
 func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
+	return b.enqueue(context.Background(), nil, message)
+}
+// enqueue accepts message for ordered transmission. It blocks only while the
+// outbox is full, until space frees, ctx ends, or cancel closes. It returns
+// the sticky failure once the sender failed.
+func (b *attachmentOutbox) enqueue(ctx context.Context, cancel <-chan struct{}, message protocol.ClientMessage) error {
+	size := outboxPayload(message)
 	b.mu.Lock()
 	for {
 		if b.err != nil {
 			b.mu.Unlock()
 			return errOutboxStopped
 		}
-		if len(b.queue) < outboxMaxPending {
+		// An empty queue always admits, so one oversized message still goes.
+		if len(b.queue) == 0 || (len(b.queue) < outboxMaxPending && b.bytes+size <= outboxMaxBytes) {
 			break
 		}
 		space := b.space
 		select {
 		case <-space:
 		case <-b.done:
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-cancel:
+			return errAttachmentForegroundRevoked
 		}
 		b.mu.Lock()
 	}
 		b.idle = make(chan struct{})
 	}
 	b.queue = append(b.queue, message)
+	b.bytes += size
+	b.queued++
 	b.mu.Unlock()
 	select {
 	case b.wake <- struct{}{}:
 }
 // stop ends the sender after its current write; queued messages are dropped.
+// Session input among them was never acknowledged to the terminal pump, so
+// the worker preserves it like a failed synchronous send.
 func (b *attachmentOutbox) stop() {
 	b.mu.Lock()
 	b.stopped = true
 	}
 }
+// unsentInput returns the session input bytes of accepted messages numbered
+// (from, to] that were never written. It is final only after join.
+func (b *attachmentOutbox) unsentInput(from, to uint64) []byte {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	var data []byte
+	// queue[i] is accepted message number written+1+i.
+	for i, message := range b.queue {
+		n := b.written + 1 + uint64(i)
+		if n <= from || n > to {
+			continue
+		}
+		if input, ok := message.(protocol.Input); ok {
+			data = append(data, input.Data...)
+		}
+	}
+	return data
+}
+// join stops the sender, closes the stream so a stalled write returns, and
+// waits for the sender to exit. Afterwards writtenCount is final.
+func (b *attachmentOutbox) join() {
+	_ = b.Close()
+	<-b.done
+}
+// accepted reports how many messages were ever accepted.
+func (b *attachmentOutbox) accepted() uint64 {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.queued
+}
+// writtenCount reports how many messages were written successfully.
+func (b *attachmentOutbox) writtenCount() uint64 {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.written
+}
+// Progress is closed after the next successful write or on failure.
+func (b *attachmentOutbox) Progress() <-chan struct{} {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.space
+}
 // Failed is closed once a write failed; failure returns the cause.
 func (b *attachmentOutbox) Failed() <-chan struct{} { return b.failed }
 		b.mu.Lock()
 		if err != nil {
+			// The queue is kept, failed write first, so unsentInput can
+			// hand undelivered keys back to the terminal pump.
 			b.err = err
-			b.queue = nil
 			close(b.failed)
 			close(b.idle)
 			close(b.space)
 		}
 		b.queue[0] = nil
 		b.queue = b.queue[1:]
+		b.bytes -= outboxPayload(message)
+		b.written++
 		close(b.space)
 		b.space = make(chan struct{})
 		if len(b.queue) == 0 {

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


@@ -2,6 +2,8 @@ package client
2	2
3	3		import (
4	4		"errors"
	5	+	"strings"
	6	+	"sync/atomic"
5	7		"testing"
6	8		"time"
7	9

@@ -93,3 +95,150 @@ func TestAttachmentOutboxCloseUnblocksStalledWrite(t *testing.T) {
93	95		t.Fatal("sender did not exit after Close")
94	96		}
95	97		}
	98	+
	99	+	// TestResumeWatch proves the resume backoff reads the terminal: Ctrl-C or a
	100	+	// lone Esc cancels, anything else is replayed to the next owner in order.
	101	+	func TestResumeWatch(t *testing.T) {
	102	+	tests := []struct {
	103	+	name string
	104	+	reads []string
	105	+	wantCancel bool
	106	+	wantResidual string
	107	+	}{
	108	+	{name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true},
	109	+	{name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true},
	110	+	{name: "escape sequence is kept", reads: []string{"\x1b[A"}, wantResidual: "\x1b[A"},
	111	+	{name: "typed keys are kept in order", reads: []string{"l", "s", "\r"}, wantResidual: "ls\r"},
	112	+	}
	113	+	for _, tt := range tests {
	114	+	t.Run(tt.name, func(t *testing.T) {
	115	+	reader := &inputTestReader{chunks: make(chan []byte, len(tt.reads))}
	116	+	pump := newTerminalInputPump(reader)
	117	+	pump.start()
	118	+	t.Cleanup(func() {
	119	+	pump.stop()
	120	+	close(reader.chunks)
	121	+	})
	122	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
	123	+	watch := lifetime.watchResume()
	124	+	for _, read := range tt.reads {
	125	+	reader.chunks <- []byte(read)
	126	+	}
	127	+	if tt.wantCancel {
	128	+	select {
	129	+	case <-watch.cancelled:
	130	+	case <-time.After(5 * time.Second):
	131	+	t.Fatal("cancel key ignored")
	132	+	}
	133	+	} else {
	134	+	require.Eventually(t, func() bool { return watch.heldLen() == len(tt.wantResidual) }, 5*time.Second, time.Millisecond)
	135	+	}
	136	+	watch.release()
	137	+	pump.mu.Lock()
	138	+	defer pump.mu.Unlock()
	139	+	require.Equal(t, tt.wantResidual, string(pump.residual))
	140	+	require.Zero(t, pump.consumer, "release returns the claim")
	141	+	})
	142	+	}
	143	+	}
	144	+
	145	+	// stallingSessionStream passes writes through until stall is set; stalled
	146	+	// writes block until release closes, then fail with failErr when set.
	147	+	type stallingSessionStream struct {
	148	+	*sessionTestStream
	149	+	stall atomic.Bool
	150	+	entered chan struct{}
	151	+	release chan struct{}
	152	+	failErr error
	153	+	}
	154	+
	155	+	func (s *stallingSessionStream) SendClient(message protocol.ClientMessage) error {
	156	+	if s.stall.Load() {
	157	+	select {
	158	+	case s.entered <- struct{}{}:
	159	+	default:
	160	+	}
	161	+	select {
	162	+	case <-s.release:
	163	+	case <-s.closed:
	164	+	return errSessionTestClosed
	165	+	}
	166	+	if s.failErr != nil {
	167	+	return s.failErr
	168	+	}
	169	+	}
	170	+	return s.sessionTestStream.SendClient(message)
	171	+	}
	172	+
	173	+	// TestAttachmentWorkerStalledLink drives the real worker over a link that
	174	+	// stops accepting writes after attach.
	175	+	func TestAttachmentWorkerStalledLink(t *testing.T) {
	176	+	linkErr := errors.New("link lost")
	177	+	tests := []struct {
	178	+	name string
	179	+	run func(t testing.T, h inputHarness, s *stallingSessionStream)
	180	+	}{
	181	+	{
	182	+	name: "output keeps painting while a write is stalled",
	183	+	run: func(t testing.T, h inputHarness, s *stallingSessionStream) {
	184	+	h.send("a")
	185	+	<-s.entered
	186	+	later := sessionTestOutput(2, "\x1b[Hlater")
	187	+	later.Epoch = 2
	188	+	h.stream.deliver(later)
	189	+	require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "later") }, 5*time.Second, time.Millisecond, "a stalled write blocked painting")
	190	+	close(s.release)
	191	+	require.Equal(t, []string{"a"}, h.awaitInputs(t, 1))
	192	+	h.end(t)
	193	+	},
	194	+	},
	195	+	{
	196	+	name: "keys behind a failed write are preserved for the next attempt",
	197	+	run: func(t testing.T, h inputHarness, s *stallingSessionStream) {
	198	+	s.failErr = linkErr
	199	+	h.send("a")
	200	+	<-s.entered
	201	+	close(s.release)
	202	+	select {
	203	+	case event := <-h.events:
	204	+	require.Equal(t, AttachmentEventFailed, event.Kind)
	205	+	require.ErrorIs(t, event.Err, linkErr)
	206	+	case <-time.After(5 * time.Second):
	207	+	t.Fatal("the failed write never settled the run")
	208	+	}
	209	+	h.pump.mu.Lock()
	210	+	defer h.pump.mu.Unlock()
	211	+	require.Equal(t, "a", string(h.pump.residual), "the unsent key is kept for the resume")
	212	+	},
	213	+	},
	214	+	{
	215	+	name: "detach on a stalled link is bounded",
	216	+	run: func(t testing.T, h inputHarness, s *stallingSessionStream) {
	217	+	h.send("a")
	218	+	<-s.entered
	219	+	h.stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
	220	+	select {
	221	+	case event := <-h.events:
	222	+	require.Equal(t, AttachmentEventEnded, event.Kind)
	223	+	case <-time.After(5 * time.Second):
	224	+	t.Fatal("the daemon-confirmed detach never settled")
	225	+	}
	226	+	h.pump.mu.Lock()
	227	+	defer h.pump.mu.Unlock()
	228	+	require.Equal(t, "a", string(h.pump.residual), "a key the link never took is not lost")
	229	+	},
	230	+	},
	231	+	}
	232	+	for _, tt := range tests {
	233	+	t.Run(tt.name, func(t *testing.T) {
	234	+	base := newSessionTestStream()
	235	+	s := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})}
	236	+	h := startInputHarnessOn(t, nil, sessionTestRequest(true), nil, base, s)
	237	+	require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "ready") }, 5*time.Second, time.Millisecond)
	238	+	// Let the initial Ack and focus report through, then stall.
	239	+	require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond)
	240	+	s.stall.Store(true)
	241	+	tt.run(t, h, s)
	242	+	})
	243	+	}
	244	+	}


@@ -2,6 +2,8 @@ package client
2		2
3	import (	3		import (
4	"errors"	4		"errors"
		5	+	"strings"
		6	+	"sync/atomic"
5	"testing"	7		"testing"
6	"time"	8		"time"
7		9

@@ -93,3 +95,150 @@ func TestAttachmentOutboxCloseUnblocksStalledWrite(t *testing.T) {
93	t.Fatal("sender did not exit after Close")	95		t.Fatal("sender did not exit after Close")
94	}	96		}
95	}	97		}
		98	+
		99	+	// TestResumeWatch proves the resume backoff reads the terminal: Ctrl-C or a
		100	+	// lone Esc cancels, anything else is replayed to the next owner in order.
		101	+	func TestResumeWatch(t *testing.T) {
		102	+	tests := []struct {
		103	+	name string
		104	+	reads []string
		105	+	wantCancel bool
		106	+	wantResidual string
		107	+	}{
		108	+	{name: "ctrl-c cancels and drops held keys", reads: []string{"ls", "\x03"}, wantCancel: true},
		109	+	{name: "lone esc cancels", reads: []string{"\x1b"}, wantCancel: true},
		110	+	{name: "escape sequence is kept", reads: []string{"\x1b[A"}, wantResidual: "\x1b[A"},
		111	+	{name: "typed keys are kept in order", reads: []string{"l", "s", "\r"}, wantResidual: "ls\r"},
		112	+	}
		113	+	for _, tt := range tests {
		114	+	t.Run(tt.name, func(t *testing.T) {
		115	+	reader := &inputTestReader{chunks: make(chan []byte, len(tt.reads))}
		116	+	pump := newTerminalInputPump(reader)
		117	+	pump.start()
		118	+	t.Cleanup(func() {
		119	+	pump.stop()
		120	+	close(reader.chunks)
		121	+	})
		122	+	lifetime := &terminalInputLifetime{eof: make(chan error, 1), pump: pump}
		123	+	watch := lifetime.watchResume()
		124	+	for _, read := range tt.reads {
		125	+	reader.chunks <- []byte(read)
		126	+	}
		127	+	if tt.wantCancel {
		128	+	select {
		129	+	case <-watch.cancelled:
		130	+	case <-time.After(5 * time.Second):
		131	+	t.Fatal("cancel key ignored")
		132	+	}
		133	+	} else {
		134	+	require.Eventually(t, func() bool { return watch.heldLen() == len(tt.wantResidual) }, 5*time.Second, time.Millisecond)
		135	+	}
		136	+	watch.release()
		137	+	pump.mu.Lock()
		138	+	defer pump.mu.Unlock()
		139	+	require.Equal(t, tt.wantResidual, string(pump.residual))
		140	+	require.Zero(t, pump.consumer, "release returns the claim")
		141	+	})
		142	+	}
		143	+	}
		144	+
		145	+	// stallingSessionStream passes writes through until stall is set; stalled
		146	+	// writes block until release closes, then fail with failErr when set.
		147	+	type stallingSessionStream struct {
		148	+	*sessionTestStream
		149	+	stall atomic.Bool
		150	+	entered chan struct{}
		151	+	release chan struct{}
		152	+	failErr error
		153	+	}
		154	+
		155	+	func (s *stallingSessionStream) SendClient(message protocol.ClientMessage) error {
		156	+	if s.stall.Load() {
		157	+	select {
		158	+	case s.entered <- struct{}{}:
		159	+	default:
		160	+	}
		161	+	select {
		162	+	case <-s.release:
		163	+	case <-s.closed:
		164	+	return errSessionTestClosed
		165	+	}
		166	+	if s.failErr != nil {
		167	+	return s.failErr
		168	+	}
		169	+	}
		170	+	return s.sessionTestStream.SendClient(message)
		171	+	}
		172	+
		173	+	// TestAttachmentWorkerStalledLink drives the real worker over a link that
		174	+	// stops accepting writes after attach.
		175	+	func TestAttachmentWorkerStalledLink(t *testing.T) {
		176	+	linkErr := errors.New("link lost")
		177	+	tests := []struct {
		178	+	name string
		179	+	run func(t testing.T, h inputHarness, s *stallingSessionStream)
		180	+	}{
		181	+	{
		182	+	name: "output keeps painting while a write is stalled",
		183	+	run: func(t testing.T, h inputHarness, s *stallingSessionStream) {
		184	+	h.send("a")
		185	+	<-s.entered
		186	+	later := sessionTestOutput(2, "\x1b[Hlater")
		187	+	later.Epoch = 2
		188	+	h.stream.deliver(later)
		189	+	require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "later") }, 5*time.Second, time.Millisecond, "a stalled write blocked painting")
		190	+	close(s.release)
		191	+	require.Equal(t, []string{"a"}, h.awaitInputs(t, 1))
		192	+	h.end(t)
		193	+	},
		194	+	},
		195	+	{
		196	+	name: "keys behind a failed write are preserved for the next attempt",
		197	+	run: func(t testing.T, h inputHarness, s *stallingSessionStream) {
		198	+	s.failErr = linkErr
		199	+	h.send("a")
		200	+	<-s.entered
		201	+	close(s.release)
		202	+	select {
		203	+	case event := <-h.events:
		204	+	require.Equal(t, AttachmentEventFailed, event.Kind)
		205	+	require.ErrorIs(t, event.Err, linkErr)
		206	+	case <-time.After(5 * time.Second):
		207	+	t.Fatal("the failed write never settled the run")
		208	+	}
		209	+	h.pump.mu.Lock()
		210	+	defer h.pump.mu.Unlock()
		211	+	require.Equal(t, "a", string(h.pump.residual), "the unsent key is kept for the resume")
		212	+	},
		213	+	},
		214	+	{
		215	+	name: "detach on a stalled link is bounded",
		216	+	run: func(t testing.T, h inputHarness, s *stallingSessionStream) {
		217	+	h.send("a")
		218	+	<-s.entered
		219	+	h.stream.deliver(protocol.Detached{Reason: protocol.ReasonDetach})
		220	+	select {
		221	+	case event := <-h.events:
		222	+	require.Equal(t, AttachmentEventEnded, event.Kind)
		223	+	case <-time.After(5 * time.Second):
		224	+	t.Fatal("the daemon-confirmed detach never settled")
		225	+	}
		226	+	h.pump.mu.Lock()
		227	+	defer h.pump.mu.Unlock()
		228	+	require.Equal(t, "a", string(h.pump.residual), "a key the link never took is not lost")
		229	+	},
		230	+	},
		231	+	}
		232	+	for _, tt := range tests {
		233	+	t.Run(tt.name, func(t *testing.T) {
		234	+	base := newSessionTestStream()
		235	+	s := &stallingSessionStream{sessionTestStream: base, entered: make(chan struct{}, 1), release: make(chan struct{})}
		236	+	h := startInputHarnessOn(t, nil, sessionTestRequest(true), nil, base, s)
		237	+	require.Eventually(t, func() bool { return strings.Contains(h.term.written(), "ready") }, 5*time.Second, time.Millisecond)
		238	+	// Let the initial Ack and focus report through, then stall.
		239	+	require.Eventually(t, func() bool { return len(base.messages()) >= 2 }, 5*time.Second, time.Millisecond)
		240	+	s.stall.Store(true)
		241	+	tt.run(t, h, s)
		242	+	})
		243	+	}
		244	+	}

M internal/usecase/client/attachment_session.go +89 -7


@@ -349,7 +349,53 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
349	349		if err := sendFocus(); err != nil {
350	350		return w.settle(ctx, fg, stream, token, err)
351	351		}
	352	+	// pending is the one undecided input delivery waiting on the outbox.
	353	+	var pending *pendingInputAck
	354	+	defer func() {
	355	+	if pending == nil {
	356	+	return
	357	+	}
	358	+	// The run ends before the delivery was written: join the sender so
	359	+	// the outbox is final, then decide the delivery once, preserving
	360	+	// exactly the session input that never reached the carriage.
	361	+	stream.join()
	362	+	if unsent := stream.unsentInput(pending.from, pending.mark); len(unsent) != 0 {
	363	+	fg.PreserveInput(unsent)
	364	+	return
	365	+	}
	366	+	fg.AckInput()
	367	+	}()
	368	+	// ackWritten commits pending once the outbox wrote it, then fences its
	369	+	// UI action after it.
	370	+	ackWritten := func() error {
	371	+	if pending == nil \|\| stream.writtenCount() < pending.mark {
	372	+	return nil
	373	+	}
	374	+	actionID := pending.actionID
	375	+	pending = nil
	376	+	fg.AckInput()
	377	+	if actionID != 0 {
	378	+	return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID})
	379	+	}
	380	+	return nil
	381	+	}
	382	+	// await makes the delivery just routed pending until the outbox wrote
	383	+	// every message accepted since from.
	384	+	await := func(from, actionID uint64) error {
	385	+	pending = &pendingInputAck{from: from, mark: stream.accepted(), actionID: actionID}
	386	+	return ackWritten()
	387	+	}
352	388		for {
	389	+	var progress <-chan struct{}
	390	+	if pending != nil {
	391	+	progress = stream.Progress()
	392	+	if err := ackWritten(); err != nil {
	393	+	return w.settle(ctx, fg, stream, token, err)
	394	+	}
	395	+	if pending == nil {
	396	+	progress = nil
	397	+	}
	398	+	}
353	399		select {
354	400		case <-focus.changed:
355	401		if err := sendFocus(); err != nil {

@@ -521,13 +567,16 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
521	567		if event.input != nil && event.input.actionID == 0 {
522	568		// A physical read: replies are stripped and ordinary bytes
523	569		// reach the overlay or the session inside read.
	570	+	from := stream.accepted()
524	571		if err := input.read(ctx, state, event.input.Data); err != nil {
525	572		// Part of the read may already be delivered or held, so
526	573		// it is committed rather than replayed.
527	574		fg.AckInput()
528	575		return w.settle(ctx, fg, stream, token, err)
529	576		}
530		-	fg.AckInput()
	577	+	if err := await(from, 0); err != nil {
	578	+	return w.settle(ctx, fg, stream, token, err)
	579	+	}
531	580		continue
532	581		}
533	582		if event.input != nil {

@@ -561,6 +610,7 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
561	610		return w.settle(ctx, fg, stream, token, err)
562	611		}
563	612		}
	613	+	from := stream.accepted()
564	614		if err := w.send(ctx, fg, stream, event.message); err != nil {
565	615		if event.input != nil {
566	616		fg.PreserveInput(event.input.Data)

@@ -568,11 +618,12 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
568	618		return w.settle(ctx, fg, stream, token, err)
569	619		}
570	620		if event.input != nil {
571		-	fg.AckInput()
572		-	if event.input.actionID != 0 {
573		-	if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil {
574		-	return w.settle(ctx, fg, stream, token, err)
575		-	}
	621	+	// The delivery stays undecided until the outbox wrote it, so
	622	+	// a link failure preserves it for the next attempt exactly as
	623	+	// a failed synchronous send did. The pump hands out no newer
	624	+	// delivery meanwhile, which bounds queued input.
	625	+	if err := await(from, event.input.actionID); err != nil {
	626	+	return w.settle(ctx, fg, stream, token, err)
576	627		}
577	628		}
578	629		if event.action != 0 {

@@ -583,6 +634,10 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
583	634		}
584	635		return attachmentLifecycleEnd(token, event.action)
585	636		}
	637	+	case <-progress:
	638	+	if err := ackWritten(); err != nil {
	639	+	return w.settle(ctx, fg, stream, token, err)
	640	+	}
586	641		case <-stream.Failed():
587	642		return w.settle(ctx, fg, stream, token, stream.failure())
588	643		case <-ctx.Done():

@@ -593,6 +648,14 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
593	648		}
594	649		}
595	650
	651	+	// pendingInputAck is an input delivery whose messages the outbox accepted but
	652	+	// has not written yet: accepted messages numbered (from, mark].
	653	+	type pendingInputAck struct {
	654	+	from uint64
	655	+	mark uint64
	656	+	actionID uint64
	657	+	}
	658	+
596	659		// attachmentLifecycleEnd gives local and daemon-triggered lifecycle actions
597	660		// the same settlement path: drain/revoke foreground before publishing picker.
598	661		func attachmentLifecycleEnd(token AttachmentToken, action AttachmentLifecycleActionKind) AttachmentEvent {

@@ -811,6 +874,15 @@ func (w *sessionAttachmentWorker) hello(stream ports.BrokerLogicalConnection) pr
811	874		// wins a blocked send, closing the stream interrupts the carriage and the
812	875		// worker joins the send before returning, so no per-send goroutine is leaked.
813	876		func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, message protocol.ClientMessage) error {
	877	+	if outbox, ok := stream.(*attachmentOutbox); ok {
	878	+	// Enqueue returns at once unless the outbox is full, and then still
	879	+	// honours cancellation, so no helper goroutine is needed.
	880	+	err := outbox.enqueue(ctx, fg.Done(), message)
	881	+	if errors.Is(err, errAttachmentForegroundRevoked) \|\| (ctx.Err() != nil && errors.Is(err, ctx.Err())) {
	882	+	_ = outbox.Close()
	883	+	}
	884	+	return err
	885	+	}
814	886		sent := make(chan error, 1)
815	887		go func() { sent <- stream.SendClient(message) }()
816	888		select {

@@ -827,9 +899,19 @@ func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForegro
827	899		}
828	900		}
829	901
830		-	// drainOutbox waits until every queued message was written, or the run ends.
	902	+	// drainOutboxTimeout bounds how long a lifecycle exit waits for queued
	903	+	// messages on a stalled link. Past it the client leaves anyway; the daemon
	904	+	// parks the attachment when the detach never arrives.
	905	+	const drainOutboxTimeout = 2 * time.Second
	906	+
	907	+	// drainOutbox waits until every queued message was written, the bound
	908	+	// elapses, or the run ends.
831	909		func (w sessionAttachmentWorker) drainOutbox(ctx context.Context, fg AttachmentForeground, outbox attachmentOutbox) error {
	910	+	timer := w.cfg.Clock.NewTimer(drainOutboxTimeout)
	911	+	defer stopSupervisorTimer(timer)
832	912		select {
	913	+	case <-timer.C():
	914	+	return nil
833	915		case <-outbox.Idle():
834	916		return outbox.failure()
835	917		case <-outbox.Failed():


@@ -349,7 +349,53 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
349		if err := sendFocus(); err != nil {	349		if err := sendFocus(); err != nil {
350		return w.settle(ctx, fg, stream, token, err)	350		return w.settle(ctx, fg, stream, token, err)
351		}	351		}
			352	+	// pending is the one undecided input delivery waiting on the outbox.
			353	+	var pending *pendingInputAck
			354	+	defer func() {
			355	+	if pending == nil {
			356	+	return
			357	+	}
			358	+	// The run ends before the delivery was written: join the sender so
			359	+	// the outbox is final, then decide the delivery once, preserving
			360	+	// exactly the session input that never reached the carriage.
			361	+	stream.join()
			362	+	if unsent := stream.unsentInput(pending.from, pending.mark); len(unsent) != 0 {
			363	+	fg.PreserveInput(unsent)
			364	+	return
			365	+	}
			366	+	fg.AckInput()
			367	+	}()
			368	+	// ackWritten commits pending once the outbox wrote it, then fences its
			369	+	// UI action after it.
			370	+	ackWritten := func() error {
			371	+	if pending == nil \|\| stream.writtenCount() < pending.mark {
			372	+	return nil
			373	+	}
			374	+	actionID := pending.actionID
			375	+	pending = nil
			376	+	fg.AckInput()
			377	+	if actionID != 0 {
			378	+	return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID})
			379	+	}
			380	+	return nil
			381	+	}
			382	+	// await makes the delivery just routed pending until the outbox wrote
			383	+	// every message accepted since from.
			384	+	await := func(from, actionID uint64) error {
			385	+	pending = &pendingInputAck{from: from, mark: stream.accepted(), actionID: actionID}
			386	+	return ackWritten()
			387	+	}
352		for {	388		for {
			389	+	var progress <-chan struct{}
			390	+	if pending != nil {
			391	+	progress = stream.Progress()
			392	+	if err := ackWritten(); err != nil {
			393	+	return w.settle(ctx, fg, stream, token, err)
			394	+	}
			395	+	if pending == nil {
			396	+	progress = nil
			397	+	}
			398	+	}
353		select {	399		select {
354		case <-focus.changed:	400		case <-focus.changed:
355		if err := sendFocus(); err != nil {	401		if err := sendFocus(); err != nil {

@@ -521,13 +567,16 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
521		if event.input != nil && event.input.actionID == 0 {	567		if event.input != nil && event.input.actionID == 0 {
522		// A physical read: replies are stripped and ordinary bytes	568		// A physical read: replies are stripped and ordinary bytes
523		// reach the overlay or the session inside read.	569		// reach the overlay or the session inside read.
			570	+	from := stream.accepted()
524		if err := input.read(ctx, state, event.input.Data); err != nil {	571		if err := input.read(ctx, state, event.input.Data); err != nil {
525		// Part of the read may already be delivered or held, so	572		// Part of the read may already be delivered or held, so
526		// it is committed rather than replayed.	573		// it is committed rather than replayed.
527		fg.AckInput()	574		fg.AckInput()
528		return w.settle(ctx, fg, stream, token, err)	575		return w.settle(ctx, fg, stream, token, err)
529		}	576		}
530	-	fg.AckInput()	577	+	if err := await(from, 0); err != nil {
			578	+	return w.settle(ctx, fg, stream, token, err)
			579	+	}
531		continue	580		continue
532		}	581		}
533		if event.input != nil {	582		if event.input != nil {

@@ -561,6 +610,7 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
561		return w.settle(ctx, fg, stream, token, err)	610		return w.settle(ctx, fg, stream, token, err)
562		}	611		}
563		}	612		}
			613	+	from := stream.accepted()
564		if err := w.send(ctx, fg, stream, event.message); err != nil {	614		if err := w.send(ctx, fg, stream, event.message); err != nil {
565		if event.input != nil {	615		if event.input != nil {
566		fg.PreserveInput(event.input.Data)	616		fg.PreserveInput(event.input.Data)

@@ -568,11 +618,12 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
568		return w.settle(ctx, fg, stream, token, err)	618		return w.settle(ctx, fg, stream, token, err)
569		}	619		}
570		if event.input != nil {	620		if event.input != nil {
571	-	fg.AckInput()	621	+	// The delivery stays undecided until the outbox wrote it, so
572	-	if event.input.actionID != 0 {	622	+	// a link failure preserves it for the next attempt exactly as
573	-	if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil {	623	+	// a failed synchronous send did. The pump hands out no newer
574	-	return w.settle(ctx, fg, stream, token, err)	624	+	// delivery meanwhile, which bounds queued input.
575	-	}	625	+	if err := await(from, event.input.actionID); err != nil {
			626	+	return w.settle(ctx, fg, stream, token, err)
576		}	627		}
577		}	628		}
578		if event.action != 0 {	629		if event.action != 0 {

@@ -583,6 +634,10 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
583		}	634		}
584		return attachmentLifecycleEnd(token, event.action)	635		return attachmentLifecycleEnd(token, event.action)
585		}	636		}
			637	+	case <-progress:
			638	+	if err := ackWritten(); err != nil {
			639	+	return w.settle(ctx, fg, stream, token, err)
			640	+	}
586		case <-stream.Failed():	641		case <-stream.Failed():
587		return w.settle(ctx, fg, stream, token, stream.failure())	642		return w.settle(ctx, fg, stream, token, stream.failure())
588		case <-ctx.Done():	643		case <-ctx.Done():

@@ -593,6 +648,14 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
593		}	648		}
594		}	649		}
595			650
			651	+	// pendingInputAck is an input delivery whose messages the outbox accepted but
			652	+	// has not written yet: accepted messages numbered (from, mark].
			653	+	type pendingInputAck struct {
			654	+	from uint64
			655	+	mark uint64
			656	+	actionID uint64
			657	+	}
			658	+
596		// attachmentLifecycleEnd gives local and daemon-triggered lifecycle actions	659		// attachmentLifecycleEnd gives local and daemon-triggered lifecycle actions
597		// the same settlement path: drain/revoke foreground before publishing picker.	660		// the same settlement path: drain/revoke foreground before publishing picker.
598		func attachmentLifecycleEnd(token AttachmentToken, action AttachmentLifecycleActionKind) AttachmentEvent {	661		func attachmentLifecycleEnd(token AttachmentToken, action AttachmentLifecycleActionKind) AttachmentEvent {

@@ -811,6 +874,15 @@ func (w *sessionAttachmentWorker) hello(stream ports.BrokerLogicalConnection) pr
811		// wins a blocked send, closing the stream interrupts the carriage and the	874		// wins a blocked send, closing the stream interrupts the carriage and the
812		// worker joins the send before returning, so no per-send goroutine is leaked.	875		// worker joins the send before returning, so no per-send goroutine is leaked.
813		func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, message protocol.ClientMessage) error {	876		func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, message protocol.ClientMessage) error {
			877	+	if outbox, ok := stream.(*attachmentOutbox); ok {
			878	+	// Enqueue returns at once unless the outbox is full, and then still
			879	+	// honours cancellation, so no helper goroutine is needed.
			880	+	err := outbox.enqueue(ctx, fg.Done(), message)
			881	+	if errors.Is(err, errAttachmentForegroundRevoked) \|\| (ctx.Err() != nil && errors.Is(err, ctx.Err())) {
			882	+	_ = outbox.Close()
			883	+	}
			884	+	return err
			885	+	}
814		sent := make(chan error, 1)	886		sent := make(chan error, 1)
815		go func() { sent <- stream.SendClient(message) }()	887		go func() { sent <- stream.SendClient(message) }()
816		select {	888		select {

@@ -827,9 +899,19 @@ func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForegro
827		}	899		}
828		}	900		}
829			901
830	-	// drainOutbox waits until every queued message was written, or the run ends.	902	+	// drainOutboxTimeout bounds how long a lifecycle exit waits for queued
			903	+	// messages on a stalled link. Past it the client leaves anyway; the daemon
			904	+	// parks the attachment when the detach never arrives.
			905	+	const drainOutboxTimeout = 2 * time.Second
			906	+
			907	+	// drainOutbox waits until every queued message was written, the bound
			908	+	// elapses, or the run ends.
831		func (w sessionAttachmentWorker) drainOutbox(ctx context.Context, fg AttachmentForeground, outbox attachmentOutbox) error {	909		func (w sessionAttachmentWorker) drainOutbox(ctx context.Context, fg AttachmentForeground, outbox attachmentOutbox) error {
			910	+	timer := w.cfg.Clock.NewTimer(drainOutboxTimeout)
			911	+	defer stopSupervisorTimer(timer)
832		select {	912		select {
			913	+	case <-timer.C():
			914	+	return nil
833		case <-outbox.Idle():	915		case <-outbox.Idle():
834		return outbox.failure()	916		return outbox.failure()
835		case <-outbox.Failed():	917		case <-outbox.Failed():

M internal/usecase/client/supervisor.go +109 -0


@@ -1490,6 +1490,115 @@ func (l *terminalInputLifetime) stop() {
1490	1490		l.pump.suspend()
1491	1491		}
1492	1492
	1493	+	// resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
	1494	+	// Past it further keys are dropped, like a full terminal buffer.
	1495	+	const resumeHeldInputLimit = 64 << 10
	1496	+
	1497	+	// resumeWatch reads the terminal while a lost attachment waits to resume.
	1498	+	// cancelled closes on Ctrl-C or a lone Esc; any other bytes are handed back
	1499	+	// to the pump on release, so the resumed attachment receives them in order.
	1500	+	type resumeWatch struct {
	1501	+	cancelled chan struct{}
	1502	+	stop chan struct{}
	1503	+	done chan struct{}
	1504	+	lifetime *terminalInputLifetime
	1505	+	id uint64
	1506	+
	1507	+	mu sync.Mutex
	1508	+	held []byte
	1509	+	}
	1510	+
	1511	+	// heldLen reports how many kept bytes the watch holds.
	1512	+	func (w *resumeWatch) heldLen() int {
	1513	+	w.mu.Lock()
	1514	+	defer w.mu.Unlock()
	1515	+	return len(w.held)
	1516	+	}
	1517	+
	1518	+	// hold keeps data within the bound.
	1519	+	func (w *resumeWatch) hold(data []byte) {
	1520	+	w.mu.Lock()
	1521	+	defer w.mu.Unlock()
	1522	+	if len(w.held)+len(data) <= resumeHeldInputLimit {
	1523	+	w.held = append(w.held, data...)
	1524	+	}
	1525	+	}
	1526	+
	1527	+	// isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc. A
	1528	+	// longer read starting with Esc is an escape sequence (arrows, focus), not a
	1529	+	// cancel.
	1530	+	func isResumeCancelKey(data []byte) bool {
	1531	+	return len(data) == 1 && (data[0] == 0x03 \|\| data[0] == 0x1b)
	1532	+	}
	1533	+
	1534	+	// watchResume claims terminal input for one resume backoff. When the claim
	1535	+	// is unavailable it returns a watch that never cancels.
	1536	+	func (l terminalInputLifetime) watchResume() resumeWatch {
	1537	+	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l}
	1538	+	if l == nil \|\| l.pump == nil {
	1539	+	close(w.done)
	1540	+	return w
	1541	+	}
	1542	+	id, ok := l.pump.tryClaim()
	1543	+	if !ok {
	1544	+	close(w.done)
	1545	+	return w
	1546	+	}
	1547	+	w.id = id
	1548	+	go w.run()
	1549	+	return w
	1550	+	}
	1551	+
	1552	+	func (w *resumeWatch) run() {
	1553	+	defer close(w.done)
	1554	+	pump := w.lifetime.pump
	1555	+	for {
	1556	+	result, ok := pump.take(context.Background(), w.id)
	1557	+	if !ok {
	1558	+	select {
	1559	+	case <-pump.readyFor(w.id):
	1560	+	case <-w.stop:
	1561	+	return
	1562	+	case <-pump.done:
	1563	+	return
	1564	+	}
	1565	+	continue
	1566	+	}
	1567	+	if result.err != nil {
	1568	+	// Leave the final read for the terminal EOF watcher.
	1569	+	pump.ack(w.id)
	1570	+	w.hold(result.data)
	1571	+	w.lifetime.finish(result.err)
	1572	+	return
	1573	+	}
	1574	+	pump.ack(w.id)
	1575	+	if isResumeCancelKey(result.data) {
	1576	+	close(w.cancelled)
	1577	+	return
	1578	+	}
	1579	+	w.hold(result.data)
	1580	+	}
	1581	+	}
	1582	+
	1583	+	// release stops the watch and hands kept keys back to the pump, unless the
	1584	+	// user cancelled, in which case they are dropped with the session.
	1585	+	func (w *resumeWatch) release() {
	1586	+	if w.id == 0 {
	1587	+	return
	1588	+	}
	1589	+	close(w.stop)
	1590	+	<-w.done
	1591	+	pump := w.lifetime.pump
	1592	+	select {
	1593	+	case <-w.cancelled:
	1594	+	default:
	1595	+	w.mu.Lock()
	1596	+	pump.appendResidual(w.id, w.held, resumeHeldInputLimit)
	1597	+	w.mu.Unlock()
	1598	+	}
	1599	+	pump.revoke(w.id)
	1600	+	}
	1601	+
1493	1602		// finish publishes the terminal cause exactly once.
1494	1603		func (l *terminalInputLifetime) finish(err error) {
1495	1604		l.once.Do(func() { l.eof <- err })


@@ -1490,6 +1490,115 @@ func (l *terminalInputLifetime) stop() {
1490	l.pump.suspend()	1490		l.pump.suspend()
1491	}	1491		}
1492		1492
		1493	+	// resumeHeldInputLimit bounds keys kept while a lost attachment resumes.
		1494	+	// Past it further keys are dropped, like a full terminal buffer.
		1495	+	const resumeHeldInputLimit = 64 << 10
		1496	+
		1497	+	// resumeWatch reads the terminal while a lost attachment waits to resume.
		1498	+	// cancelled closes on Ctrl-C or a lone Esc; any other bytes are handed back
		1499	+	// to the pump on release, so the resumed attachment receives them in order.
		1500	+	type resumeWatch struct {
		1501	+	cancelled chan struct{}
		1502	+	stop chan struct{}
		1503	+	done chan struct{}
		1504	+	lifetime *terminalInputLifetime
		1505	+	id uint64
		1506	+
		1507	+	mu sync.Mutex
		1508	+	held []byte
		1509	+	}
		1510	+
		1511	+	// heldLen reports how many kept bytes the watch holds.
		1512	+	func (w *resumeWatch) heldLen() int {
		1513	+	w.mu.Lock()
		1514	+	defer w.mu.Unlock()
		1515	+	return len(w.held)
		1516	+	}
		1517	+
		1518	+	// hold keeps data within the bound.
		1519	+	func (w *resumeWatch) hold(data []byte) {
		1520	+	w.mu.Lock()
		1521	+	defer w.mu.Unlock()
		1522	+	if len(w.held)+len(data) <= resumeHeldInputLimit {
		1523	+	w.held = append(w.held, data...)
		1524	+	}
		1525	+	}
		1526	+
		1527	+	// isResumeCancelKey reports a read that is exactly Ctrl-C or a lone Esc. A
		1528	+	// longer read starting with Esc is an escape sequence (arrows, focus), not a
		1529	+	// cancel.
		1530	+	func isResumeCancelKey(data []byte) bool {
		1531	+	return len(data) == 1 && (data[0] == 0x03 \|\| data[0] == 0x1b)
		1532	+	}
		1533	+
		1534	+	// watchResume claims terminal input for one resume backoff. When the claim
		1535	+	// is unavailable it returns a watch that never cancels.
		1536	+	func (l terminalInputLifetime) watchResume() resumeWatch {
		1537	+	w := &resumeWatch{cancelled: make(chan struct{}), stop: make(chan struct{}), done: make(chan struct{}), lifetime: l}
		1538	+	if l == nil \|\| l.pump == nil {
		1539	+	close(w.done)
		1540	+	return w
		1541	+	}
		1542	+	id, ok := l.pump.tryClaim()
		1543	+	if !ok {
		1544	+	close(w.done)
		1545	+	return w
		1546	+	}
		1547	+	w.id = id
		1548	+	go w.run()
		1549	+	return w
		1550	+	}
		1551	+
		1552	+	func (w *resumeWatch) run() {
		1553	+	defer close(w.done)
		1554	+	pump := w.lifetime.pump
		1555	+	for {
		1556	+	result, ok := pump.take(context.Background(), w.id)
		1557	+	if !ok {
		1558	+	select {
		1559	+	case <-pump.readyFor(w.id):
		1560	+	case <-w.stop:
		1561	+	return
		1562	+	case <-pump.done:
		1563	+	return
		1564	+	}
		1565	+	continue
		1566	+	}
		1567	+	if result.err != nil {
		1568	+	// Leave the final read for the terminal EOF watcher.
		1569	+	pump.ack(w.id)
		1570	+	w.hold(result.data)
		1571	+	w.lifetime.finish(result.err)
		1572	+	return
		1573	+	}
		1574	+	pump.ack(w.id)
		1575	+	if isResumeCancelKey(result.data) {
		1576	+	close(w.cancelled)
		1577	+	return
		1578	+	}
		1579	+	w.hold(result.data)
		1580	+	}
		1581	+	}
		1582	+
		1583	+	// release stops the watch and hands kept keys back to the pump, unless the
		1584	+	// user cancelled, in which case they are dropped with the session.
		1585	+	func (w *resumeWatch) release() {
		1586	+	if w.id == 0 {
		1587	+	return
		1588	+	}
		1589	+	close(w.stop)
		1590	+	<-w.done
		1591	+	pump := w.lifetime.pump
		1592	+	select {
		1593	+	case <-w.cancelled:
		1594	+	default:
		1595	+	w.mu.Lock()
		1596	+	pump.appendResidual(w.id, w.held, resumeHeldInputLimit)
		1597	+	w.mu.Unlock()
		1598	+	}
		1599	+	pump.revoke(w.id)
		1600	+	}
		1601	+
1493	// finish publishes the terminal cause exactly once.	1602		// finish publishes the terminal cause exactly once.
1494	func (l *terminalInputLifetime) finish(err error) {	1603		func (l *terminalInputLifetime) finish(err error) {
1495	l.once.Do(func() { l.eof <- err })	1604		l.once.Do(func() { l.eof <- err })

M internal/usecase/client/supervisor_attach.go +32 -8


@@ -625,7 +625,7 @@ func (s Supervisor) runResolvedAttachment(ctx context.Context, input terminalI
625	625		s.notifyAttachment(outcome.cause)
626	626		return false, nil
627	627		}
628		-	if terminated, termErr, brokerLost := s.waitResume(ctx, input, service, attempt); terminated \|\| brokerLost {
	628	+	if terminated, termErr, stop := s.waitResume(ctx, input, service, attempt); terminated \|\| stop {
629	629		return terminated, termErr
630	630		}
631	631		next, ok := s.withFreshStream(service, outcome.target)

@@ -641,6 +641,10 @@ func (s Supervisor) runResolvedAttachment(ctx context.Context, input terminalI
641	641		type attachmentResumeBudget struct {
642	642		used int
643	643		resuming bool
	644	+	// flaps counts attachments lost before they became stable. They prove
	645	+	// the route works but the session drops, so they are capped by count
	646	+	// regardless of the outage window.
	647	+	flaps int
644	648		// since is when the current outage began; zero before the first spend.
645	649		since time.Time
646	650		}

@@ -650,17 +654,28 @@ type attachmentResumeBudget struct {
650	654		// fresh budget; one that drops right after attaching keeps counting, so a
651	655		// flapping session gives up.
652	656		// It returns the 1-based resume attempt, and false once the budget is
653		-	// exhausted: at least maxAttachmentResumes attempts and resumeOutageWindow
654		-	// since the outage began.
	657	+	// exhausted: more than maxAttachmentResumes flaps, or more than
	658	+	// maxAttachmentResumes attempts once resumeOutageWindow has passed since the
	659	+	// outage began. A route that stays down therefore keeps resuming for the
	660	+	// whole window, while a session that keeps dropping right after it attaches
	661	+	// gives up quickly.
655	662		func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration, now time.Time) (int, bool) {
656		-	if kind == attachmentResume && up >= resumeStableAttachment {
657		-	b.used = 0
	663	+	if kind == attachmentResume {
	664	+	if up >= resumeStableAttachment {
	665	+	b.used = 0
	666	+	b.flaps = 0
	667	+	} else {
	668	+	b.flaps++
	669	+	}
658	670		}
659	671		if b.used == 0 {
660	672		b.since = now
661	673		}
662	674		b.used++
663	675		b.resuming = true
	676	+	if b.flaps > maxAttachmentResumes {
	677	+	return b.used, false
	678	+	}
664	679		return b.used, b.used <= maxAttachmentResumes \|\| now.Sub(b.since) < resumeOutageWindow
665	680		}
666	681

@@ -937,15 +952,24 @@ func resumeTarget(request ports.BrokerOpenStreamRequest, fg *attachmentForegroun
937	952
938	953		// waitResume presents Connecting and waits one backoff interval before the
939	954		// next resume attempt. It settles early on cancellation or terminal EOF
940		-	// (terminated), or on broker loss, which ends the resume and leaves the ready
941		-	// loop to observe and retire the connection.
942		-	func (s Supervisor) waitResume(ctx context.Context, input terminalInputLifetime, service ports.BrokerNavigator, attempt int) (terminated bool, termErr error, brokerLost bool) {
	955	+	// (terminated), or with stop on broker loss, which leaves the ready loop to
	956	+	// observe and retire the connection, or when the user cancels the resume
	957	+	// with Ctrl-C or Esc, which returns to the picker. Other keys typed during
	958	+	// the outage are kept and replayed into the session once it resumes.
	959	+	func (s Supervisor) waitResume(ctx context.Context, input terminalInputLifetime, service ports.BrokerNavigator, attempt int) (terminated bool, termErr error, stop bool) {
943	960		s.transition(supervisorEvent{kind: supervisorAttachBegin})
944	961		delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
945	962		timer := s.cfg.Clock.NewTimer(delay)
946	963		defer stopSupervisorTimer(timer)
	964	+	keys := input.watchResume()
	965	+	defer keys.release()
947	966		for {
948	967		select {
	968	+	case <-keys.cancelled:
	969	+	// The user gave up on the outage: back to the picker, quietly.
	970	+	s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
	971	+	s.transition(supervisorEvent{kind: supervisorAttachEnded})
	972	+	return false, nil, true
949	973		case <-timer.C():
950	974		return false, nil, false
951	975		case <-ctx.Done():

 			s.notifyAttachment(outcome.cause)
 			return false, nil
 		}
-		if terminated, termErr, brokerLost := s.waitResume(ctx, input, service, attempt); terminated || brokerLost {
+		if terminated, termErr, stop := s.waitResume(ctx, input, service, attempt); terminated || stop {
 			return terminated, termErr
 		}
 		next, ok := s.withFreshStream(service, outcome.target)
 type attachmentResumeBudget struct {
 	used     int
 	resuming bool
+	// flaps counts attachments lost before they became stable. They prove
+	// the route works but the session drops, so they are capped by count
+	// regardless of the outage window.
+	flaps int
 	// since is when the current outage began; zero before the first spend.
 	since time.Time
 }
 // fresh budget; one that drops right after attaching keeps counting, so a
 // flapping session gives up.
 // It returns the 1-based resume attempt, and false once the budget is
-// exhausted: at least maxAttachmentResumes attempts and resumeOutageWindow
+// exhausted: more than maxAttachmentResumes flaps, or more than
-// since the outage began.
+// maxAttachmentResumes attempts once resumeOutageWindow has passed since the
+// outage began. A route that stays down therefore keeps resuming for the
+// whole window, while a session that keeps dropping right after it attaches
+// gives up quickly.
 func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration, now time.Time) (int, bool) {
-	if kind == attachmentResume && up >= resumeStableAttachment {
+	if kind == attachmentResume {
-		b.used = 0
+		if up >= resumeStableAttachment {
+			b.used = 0
+			b.flaps = 0
+		} else {
+			b.flaps++
+		}
 	}
 	if b.used == 0 {
 		b.since = now
 	}
 	b.used++
 	b.resuming = true
+	if b.flaps > maxAttachmentResumes {
+		return b.used, false
+	}
 	return b.used, b.used <= maxAttachmentResumes || now.Sub(b.since) < resumeOutageWindow
 }
 // waitResume presents Connecting and waits one backoff interval before the
 // next resume attempt. It settles early on cancellation or terminal EOF
-// (terminated), or on broker loss, which ends the resume and leaves the ready
+// (terminated), or with stop on broker loss, which leaves the ready loop to
-// loop to observe and retire the connection.
+// observe and retire the connection, or when the user cancels the resume
-func (s *Supervisor) waitResume(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, attempt int) (terminated bool, termErr error, brokerLost bool) {
+// with Ctrl-C or Esc, which returns to the picker. Other keys typed during
+// the outage are kept and replayed into the session once it resumes.
+func (s *Supervisor) waitResume(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, attempt int) (terminated bool, termErr error, stop bool) {
 	s.transition(supervisorEvent{kind: supervisorAttachBegin})
 	delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
 	timer := s.cfg.Clock.NewTimer(delay)
 	defer stopSupervisorTimer(timer)
+	keys := input.watchResume()
+	defer keys.release()
 	for {
 		select {
+		case <-keys.cancelled:
+			// The user gave up on the outage: back to the picker, quietly.
+			s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
+			s.transition(supervisorEvent{kind: supervisorAttachEnded})
+			return false, nil, true
 		case <-timer.C():
 			return false, nil, false
 		case <-ctx.Done():

M internal/usecase/client/supervisor_attach_test.go +1 -0


@@ -1341,6 +1341,7 @@ func TestAttachmentResumeBudget(t *testing.T) {
1341	1341		{name: "flapping resumes exhaust", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
1342	1342		{name: "stable attachment restores the budget", steps: repeatKinds(attachmentResume, maxAttachmentResumes+2), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+2), step: past, want: repeatBools(true, maxAttachmentResumes+2)},
1343	1343		{name: "retries never restore the budget", steps: repeatKinds(attachmentRetry, maxAttachmentResumes+1), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
	1344	+	{name: "flaps exhaust inside the outage window", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: time.Second, want: append(repeatBools(true, maxAttachmentResumes), false)},
1344	1345		{name: "outage window keeps resuming", steps: repeatKinds(attachmentRetry, 3maxAttachmentResumes), ups: repeatDurations(brief, 3maxAttachmentResumes), step: time.Second, want: repeatBools(true, 3*maxAttachmentResumes)},
1345	1346		}
1346	1347		for _, tt := range tests {


@@ -1341,6 +1341,7 @@ func TestAttachmentResumeBudget(t *testing.T) {
1341	{name: "flapping resumes exhaust", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},	1341		{name: "flapping resumes exhaust", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
1342	{name: "stable attachment restores the budget", steps: repeatKinds(attachmentResume, maxAttachmentResumes+2), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+2), step: past, want: repeatBools(true, maxAttachmentResumes+2)},	1342		{name: "stable attachment restores the budget", steps: repeatKinds(attachmentResume, maxAttachmentResumes+2), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+2), step: past, want: repeatBools(true, maxAttachmentResumes+2)},
1343	{name: "retries never restore the budget", steps: repeatKinds(attachmentRetry, maxAttachmentResumes+1), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},	1343		{name: "retries never restore the budget", steps: repeatKinds(attachmentRetry, maxAttachmentResumes+1), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
		1344	+	{name: "flaps exhaust inside the outage window", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: time.Second, want: append(repeatBools(true, maxAttachmentResumes), false)},
1344	{name: "outage window keeps resuming", steps: repeatKinds(attachmentRetry, 3maxAttachmentResumes), ups: repeatDurations(brief, 3maxAttachmentResumes), step: time.Second, want: repeatBools(true, 3*maxAttachmentResumes)},	1345		{name: "outage window keeps resuming", steps: repeatKinds(attachmentRetry, 3maxAttachmentResumes), ups: repeatDurations(brief, 3maxAttachmentResumes), step: time.Second, want: repeatBools(true, 3*maxAttachmentResumes)},
1345	}	1346		}
1346	for _, tt := range tests {	1347		for _, tt := range tests {

M internal/usecase/client/terminal_input_pump.go +13 -0


@@ -242,6 +242,19 @@ func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
242	242		p.residual = append(p.residual[:0], data...)
243	243		}
244	244
	245	+	// appendResidual appends bytes the consumer took but did not deliver, so the
	246	+	// next owner replays them after any earlier residual. It reports false when
	247	+	// the bound would be exceeded; the bytes are then dropped.
	248	+	func (p *terminalInputPump) appendResidual(consumer uint64, data []byte, limit int) bool {
	249	+	p.mu.Lock()
	250	+	defer p.mu.Unlock()
	251	+	if p.consumer != consumer \|\| len(p.residual)+len(data) > limit {
	252	+	return false
	253	+	}
	254	+	p.residual = append(p.residual, data...)
	255	+	return true
	256	+	}
	257	+
245	258		// ack commits scanner delivery of a leased read and lets the lifecycle reader
246	259		// accept the next result. Only the consumer that holds the lease can ack it.
247	260		// dropOwned disposes every undecided byte at an autonomous owner-class


@@ -242,6 +242,19 @@ func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
242	p.residual = append(p.residual[:0], data...)	242		p.residual = append(p.residual[:0], data...)
243	}	243		}
244		244
		245	+	// appendResidual appends bytes the consumer took but did not deliver, so the
		246	+	// next owner replays them after any earlier residual. It reports false when
		247	+	// the bound would be exceeded; the bytes are then dropped.
		248	+	func (p *terminalInputPump) appendResidual(consumer uint64, data []byte, limit int) bool {
		249	+	p.mu.Lock()
		250	+	defer p.mu.Unlock()
		251	+	if p.consumer != consumer \|\| len(p.residual)+len(data) > limit {
		252	+	return false
		253	+	}
		254	+	p.residual = append(p.residual, data...)
		255	+	return true
		256	+	}
		257	+
245	// ack commits scanner delivery of a leased read and lets the lifecycle reader	258		// ack commits scanner delivery of a leased read and lets the lifecycle reader
246	// accept the next result. Only the consumer that holds the lease can ack it.	259		// accept the next result. Only the consumer that holds the lease can ack it.
247	// dropOwned disposes every undecided byte at an autonomous owner-class	260		// dropOwned disposes every undecided byte at an autonomous owner-class