0913a64 fix(client): keep remote UI responsive on flaky links · vev

A docs/plans/quic-resilience.md +66 -0

@@ -0,0 +1,66 @@
	1	+	# Plan: mosh-like remote resilience
	2	+
	3	+	Goal: on a lossy or intermittent link (4G, roaming, sleep), the UI never
	4	+	freezes, typed input is kept and delivered when the link returns, and SSH is
	5	+	used only to authenticate the first QUIC handshake.
	6	+
	7	+	Reference: mosh (`/tmp/mosh`, `src/network/transportsender-impl.h`,
	8	+	`src/frontend/terminaloverlay.cc`).
	9	+
	10	+	## Diagnosis (current main)
	11	+
	12	+	1. Input path blocks on the network: `sessionAttachmentWorker.send` waits for
	13	+	`SendClient`, which waits for the carriage write. While QUIC stalls, the
	14	+	worker loop stops reading keys, servicing overlays, and painting.
	15	+	2. Every reconnect re-runs SSH: `_broker-mux-quic-proxy` serves one
	16	+	connection with one-time credentials and exits with it.
	17	+	3. No connection migration: one UDP socket per connection, no `AddPath`.
	18	+	4. Slow detection, early give-up: keepalive 15 s / idle 60 s; 5 resume
	19	+	attempts over a few seconds, then back to the picker.
	20	+	5. `LinkState` is only ever `Offline` at teardown; `degraded`/`probing` are
	21	+	documented but never produced.
	22	+	6. Daemon ignores `Input.InputSeq`, so input in flight at a drop is lost or
	23	+	would be duplicated by a replay.
	24	+
	25	+	## Phase 1 — non-blocking client send (client only)
	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.
	34	+
	35	+	## Phase 2 — faster detection, patient resume (client + QUIC adapter)
	36	+
	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.
	41	+
	42	+	## Phase 3 — exactly-once input across resume (protocol bump)
	43	+
	44	+	- Daemon tracks last applied `InputSeq` per attachment; drops `<=` seqs.
	45	+	- `Welcome` on resume carries the last applied seq; client replays unacked
	46	+	input after it. Bump `protocol.Version`.
	47	+
	48	+	## Phase 4 — SSH only for first handshake (remote helpers, trust)
	49	+
	50	+	- Remote QUIC proxy outlives one connection for the park window and issues a
	51	+	resume ticket (bearer secret, bound to cert fingerprint + session, expiry,
	52	+	single active use, memory only).
	53	+	- Reconnect dials QUIC directly with the ticket; SSH only on rejection/expiry.
	54	+	- Picker host health uses the live link + a ticket-based QUIC probe, never
	55	+	SSH.
	56	+	- Needs a security review (ticket theft, replay, expiry) before merge.
	57	+
	58	+	## Phase 5 — real link states and path migration
	59	+
	60	+	- Derive `degraded`/`probing` from RTT and last-packet age in the adapter.
	61	+	- On stall or local address change, `AddPath` + probe + switch.
	62	+
	63	+	## Phase 6 — optional predictive local echo
	64	+
	65	+	- mosh-style overlay with underline-until-confirmed. Breaks the "thin client
	66	+	interprets nothing" rule; needs an explicit design decision.

@@ -0,0 +1,66 @@
	1	+	# Plan: mosh-like remote resilience
	2	+
	3	+	Goal: on a lossy or intermittent link (4G, roaming, sleep), the UI never
	4	+	freezes, typed input is kept and delivered when the link returns, and SSH is
	5	+	used only to authenticate the first QUIC handshake.
	6	+
	7	+	Reference: mosh (`/tmp/mosh`, `src/network/transportsender-impl.h`,
	8	+	`src/frontend/terminaloverlay.cc`).
	9	+
	10	+	## Diagnosis (current main)
	11	+
	12	+	1. Input path blocks on the network: `sessionAttachmentWorker.send` waits for
	13	+	`SendClient`, which waits for the carriage write. While QUIC stalls, the
	14	+	worker loop stops reading keys, servicing overlays, and painting.
	15	+	2. Every reconnect re-runs SSH: `_broker-mux-quic-proxy` serves one
	16	+	connection with one-time credentials and exits with it.
	17	+	3. No connection migration: one UDP socket per connection, no `AddPath`.
	18	+	4. Slow detection, early give-up: keepalive 15 s / idle 60 s; 5 resume
	19	+	attempts over a few seconds, then back to the picker.
	20	+	5. `LinkState` is only ever `Offline` at teardown; `degraded`/`probing` are
	21	+	documented but never produced.
	22	+	6. Daemon ignores `Input.InputSeq`, so input in flight at a drop is lost or
	23	+	would be duplicated by a replay.
	24	+
	25	+	## Phase 1 — non-blocking client send (client only)
	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.
	34	+
	35	+	## Phase 2 — faster detection, patient resume (client + QUIC adapter)
	36	+
	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.
	41	+
	42	+	## Phase 3 — exactly-once input across resume (protocol bump)
	43	+
	44	+	- Daemon tracks last applied `InputSeq` per attachment; drops `<=` seqs.
	45	+	- `Welcome` on resume carries the last applied seq; client replays unacked
	46	+	input after it. Bump `protocol.Version`.
	47	+
	48	+	## Phase 4 — SSH only for first handshake (remote helpers, trust)
	49	+
	50	+	- Remote QUIC proxy outlives one connection for the park window and issues a
	51	+	resume ticket (bearer secret, bound to cert fingerprint + session, expiry,
	52	+	single active use, memory only).
	53	+	- Reconnect dials QUIC directly with the ticket; SSH only on rejection/expiry.
	54	+	- Picker host health uses the live link + a ticket-based QUIC probe, never
	55	+	SSH.
	56	+	- Needs a security review (ticket theft, replay, expiry) before merge.
	57	+
	58	+	## Phase 5 — real link states and path migration
	59	+
	60	+	- Derive `degraded`/`probing` from RTT and last-packet age in the adapter.
	61	+	- On stall or local address change, `AddPath` + probe + switch.
	62	+
	63	+	## Phase 6 — optional predictive local echo
	64	+
	65	+	- mosh-style overlay with underline-until-confirmed. Breaks the "thin client
	66	+	interprets nothing" rule; needs an explicit design decision.

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


@@ -25,7 +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.
	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.
29	30		- If the reconnect window has expired but the session still exists, the client opens a fresh attachment to it.
30	31		- Several clients can attach to the same session. Each keeps its own window, tab, focus, and copy mode.
31	32


@@ -25,7 +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.	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.
29		- 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.
30		- 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.
31			32

M internal/adapters/quic/quic.go +6 -3


@@ -61,8 +61,11 @@ var (
61	61		// listener and none on the dialer, no unidirectional streams, keepalive /
62	62		// idle limits, bounded admission, no 0-RTT.
63	63		type Config struct {
64		-	// KeepAlivePeriod defaults to 15s; MaxIdleTimeout defaults to 60s;
65		-	// HandshakeIdleTimeout defaults to 5s.
	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.
66	69		KeepAlivePeriod time.Duration
67	70		MaxIdleTimeout time.Duration
68	71		HandshakeIdleTimeout time.Duration

@@ -86,7 +89,7 @@ const noIncomingStreams int64 = -1
86	89		func baseQUICConfig(config Config) *quicgo.Config {
87	90		keepAlive := config.KeepAlivePeriod
88	91		if keepAlive == 0 {
89		-	keepAlive = 15 * time.Second
	92	+	keepAlive = 2 * time.Second
90	93		}
91	94		maxIdle := config.MaxIdleTimeout
92	95		if maxIdle == 0 {

 // listener and none on the dialer, no unidirectional streams, keepalive /
 // idle limits, bounded admission, no 0-RTT.
 type Config struct {
-	// KeepAlivePeriod defaults to 15s; MaxIdleTimeout defaults to 60s;
+	// KeepAlivePeriod defaults to 2s; MaxIdleTimeout defaults to 60s;
-	// HandshakeIdleTimeout defaults to 5s.
+	// HandshakeIdleTimeout defaults to 5s. The short keepalive holds mobile
+	// CGNAT bindings open and surfaces loss quickly, while the long idle
+	// timeout lets one connection ride out short outages and deliver queued
+	// input without reconnecting.
 	KeepAlivePeriod      time.Duration
 	MaxIdleTimeout       time.Duration
 	HandshakeIdleTimeout time.Duration
 func baseQUICConfig(config Config) *quicgo.Config {
 	keepAlive := config.KeepAlivePeriod
 	if keepAlive == 0 {
-		keepAlive = 15 * time.Second
+		keepAlive = 2 * time.Second
 	}
 	maxIdle := config.MaxIdleTimeout
 	if maxIdle == 0 {

A internal/usecase/client/attachment_outbox.go +160 -0

@@ -0,0 +1,160 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"sync"
	6	+
	7	+	"github.com/bnema/vev/internal/ports"
	8	+	"github.com/bnema/vev/internal/protocol"
	9	+	)
	10	+
	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
	15	+
	16	+	// errOutboxStopped is returned by SendClient after the outbox stopped.
	17	+	var errOutboxStopped = errors.New("client: attachment outbox stopped")
	18	+
	19	+	// 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.
	24	+	type attachmentOutbox struct {
	25	+	ports.BrokerLogicalConnection
	26	+
	27	+	mu sync.Mutex
	28	+	queue []protocol.ClientMessage
	29	+	err error
	30	+	stopped bool
	31	+	wake chan struct{} // sender wake-up, capacity 1
	32	+	space chan struct{} // closed-and-replaced when the queue shrinks
	33	+	idle chan struct{} // closed-and-replaced when the queue drains
	34	+	failed chan struct{} // closed once on the first write failure
	35	+	done chan struct{} // closed when the sender exits
	36	+	}
	37	+
	38	+	func newAttachmentOutbox(stream ports.BrokerLogicalConnection) *attachmentOutbox {
	39	+	b := &attachmentOutbox{
	40	+	BrokerLogicalConnection: stream,
	41	+	wake: make(chan struct{}, 1),
	42	+	space: make(chan struct{}),
	43	+	idle: make(chan struct{}),
	44	+	failed: make(chan struct{}),
	45	+	done: make(chan struct{}),
	46	+	}
	47	+	close(b.idle)
	48	+	go b.run()
	49	+	return b
	50	+	}
	51	+
	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.
	54	+	func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
	55	+	b.mu.Lock()
	56	+	for {
	57	+	if b.err != nil {
	58	+	err := b.err
	59	+	b.mu.Unlock()
	60	+	return err
	61	+	}
	62	+	if b.stopped {
	63	+	b.mu.Unlock()
	64	+	return errOutboxStopped
	65	+	}
	66	+	if len(b.queue) < outboxMaxPending {
	67	+	break
	68	+	}
	69	+	space := b.space
	70	+	b.mu.Unlock()
	71	+	select {
	72	+	case <-space:
	73	+	case <-b.done:
	74	+	}
	75	+	b.mu.Lock()
	76	+	}
	77	+	if len(b.queue) == 0 {
	78	+	b.idle = make(chan struct{})
	79	+	}
	80	+	b.queue = append(b.queue, message)
	81	+	b.mu.Unlock()
	82	+	select {
	83	+	case b.wake <- struct{}{}:
	84	+	default:
	85	+	}
	86	+	return nil
	87	+	}
	88	+
	89	+	// Close stops the sender and closes the underlying stream, which unblocks a
	90	+	// write stalled on the link.
	91	+	func (b *attachmentOutbox) Close() error {
	92	+	b.stop()
	93	+	return b.BrokerLogicalConnection.Close()
	94	+	}
	95	+
	96	+	// stop ends the sender after its current write; queued messages are dropped.
	97	+	func (b *attachmentOutbox) stop() {
	98	+	b.mu.Lock()
	99	+	b.stopped = true
	100	+	b.mu.Unlock()
	101	+	select {
	102	+	case b.wake <- struct{}{}:
	103	+	default:
	104	+	}
	105	+	}
	106	+
	107	+	// Failed is closed once a write failed; failure returns the cause.
	108	+	func (b *attachmentOutbox) Failed() <-chan struct{} { return b.failed }
	109	+
	110	+	func (b *attachmentOutbox) failure() error {
	111	+	b.mu.Lock()
	112	+	defer b.mu.Unlock()
	113	+	return b.err
	114	+	}
	115	+
	116	+	// Idle is closed while nothing is queued or being written.
	117	+	func (b *attachmentOutbox) Idle() <-chan struct{} {
	118	+	b.mu.Lock()
	119	+	defer b.mu.Unlock()
	120	+	return b.idle
	121	+	}
	122	+
	123	+	func (b *attachmentOutbox) run() {
	124	+	defer close(b.done)
	125	+	for {
	126	+	b.mu.Lock()
	127	+	if b.stopped {
	128	+	b.mu.Unlock()
	129	+	return
	130	+	}
	131	+	if len(b.queue) == 0 {
	132	+	b.mu.Unlock()
	133	+	<-b.wake
	134	+	continue
	135	+	}
	136	+	message := b.queue[0]
	137	+	b.mu.Unlock()
	138	+
	139	+	err := b.BrokerLogicalConnection.SendClient(message)
	140	+
	141	+	b.mu.Lock()
	142	+	if err != nil {
	143	+	b.err = err
	144	+	b.queue = nil
	145	+	close(b.failed)
	146	+	close(b.idle)
	147	+	close(b.space)
	148	+	b.mu.Unlock()
	149	+	return
	150	+	}
	151	+	b.queue[0] = nil
	152	+	b.queue = b.queue[1:]
	153	+	close(b.space)
	154	+	b.space = make(chan struct{})
	155	+	if len(b.queue) == 0 {
	156	+	close(b.idle)
	157	+	}
	158	+	b.mu.Unlock()
	159	+	}
	160	+	}

@@ -0,0 +1,160 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"sync"
	6	+
	7	+	"github.com/bnema/vev/internal/ports"
	8	+	"github.com/bnema/vev/internal/protocol"
	9	+	)
	10	+
	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
	15	+
	16	+	// errOutboxStopped is returned by SendClient after the outbox stopped.
	17	+	var errOutboxStopped = errors.New("client: attachment outbox stopped")
	18	+
	19	+	// 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.
	24	+	type attachmentOutbox struct {
	25	+	ports.BrokerLogicalConnection
	26	+
	27	+	mu sync.Mutex
	28	+	queue []protocol.ClientMessage
	29	+	err error
	30	+	stopped bool
	31	+	wake chan struct{} // sender wake-up, capacity 1
	32	+	space chan struct{} // closed-and-replaced when the queue shrinks
	33	+	idle chan struct{} // closed-and-replaced when the queue drains
	34	+	failed chan struct{} // closed once on the first write failure
	35	+	done chan struct{} // closed when the sender exits
	36	+	}
	37	+
	38	+	func newAttachmentOutbox(stream ports.BrokerLogicalConnection) *attachmentOutbox {
	39	+	b := &attachmentOutbox{
	40	+	BrokerLogicalConnection: stream,
	41	+	wake: make(chan struct{}, 1),
	42	+	space: make(chan struct{}),
	43	+	idle: make(chan struct{}),
	44	+	failed: make(chan struct{}),
	45	+	done: make(chan struct{}),
	46	+	}
	47	+	close(b.idle)
	48	+	go b.run()
	49	+	return b
	50	+	}
	51	+
	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.
	54	+	func (b *attachmentOutbox) SendClient(message protocol.ClientMessage) error {
	55	+	b.mu.Lock()
	56	+	for {
	57	+	if b.err != nil {
	58	+	err := b.err
	59	+	b.mu.Unlock()
	60	+	return err
	61	+	}
	62	+	if b.stopped {
	63	+	b.mu.Unlock()
	64	+	return errOutboxStopped
	65	+	}
	66	+	if len(b.queue) < outboxMaxPending {
	67	+	break
	68	+	}
	69	+	space := b.space
	70	+	b.mu.Unlock()
	71	+	select {
	72	+	case <-space:
	73	+	case <-b.done:
	74	+	}
	75	+	b.mu.Lock()
	76	+	}
	77	+	if len(b.queue) == 0 {
	78	+	b.idle = make(chan struct{})
	79	+	}
	80	+	b.queue = append(b.queue, message)
	81	+	b.mu.Unlock()
	82	+	select {
	83	+	case b.wake <- struct{}{}:
	84	+	default:
	85	+	}
	86	+	return nil
	87	+	}
	88	+
	89	+	// Close stops the sender and closes the underlying stream, which unblocks a
	90	+	// write stalled on the link.
	91	+	func (b *attachmentOutbox) Close() error {
	92	+	b.stop()
	93	+	return b.BrokerLogicalConnection.Close()
	94	+	}
	95	+
	96	+	// stop ends the sender after its current write; queued messages are dropped.
	97	+	func (b *attachmentOutbox) stop() {
	98	+	b.mu.Lock()
	99	+	b.stopped = true
	100	+	b.mu.Unlock()
	101	+	select {
	102	+	case b.wake <- struct{}{}:
	103	+	default:
	104	+	}
	105	+	}
	106	+
	107	+	// Failed is closed once a write failed; failure returns the cause.
	108	+	func (b *attachmentOutbox) Failed() <-chan struct{} { return b.failed }
	109	+
	110	+	func (b *attachmentOutbox) failure() error {
	111	+	b.mu.Lock()
	112	+	defer b.mu.Unlock()
	113	+	return b.err
	114	+	}
	115	+
	116	+	// Idle is closed while nothing is queued or being written.
	117	+	func (b *attachmentOutbox) Idle() <-chan struct{} {
	118	+	b.mu.Lock()
	119	+	defer b.mu.Unlock()
	120	+	return b.idle
	121	+	}
	122	+
	123	+	func (b *attachmentOutbox) run() {
	124	+	defer close(b.done)
	125	+	for {
	126	+	b.mu.Lock()
	127	+	if b.stopped {
	128	+	b.mu.Unlock()
	129	+	return
	130	+	}
	131	+	if len(b.queue) == 0 {
	132	+	b.mu.Unlock()
	133	+	<-b.wake
	134	+	continue
	135	+	}
	136	+	message := b.queue[0]
	137	+	b.mu.Unlock()
	138	+
	139	+	err := b.BrokerLogicalConnection.SendClient(message)
	140	+
	141	+	b.mu.Lock()
	142	+	if err != nil {
	143	+	b.err = err
	144	+	b.queue = nil
	145	+	close(b.failed)
	146	+	close(b.idle)
	147	+	close(b.space)
	148	+	b.mu.Unlock()
	149	+	return
	150	+	}
	151	+	b.queue[0] = nil
	152	+	b.queue = b.queue[1:]
	153	+	close(b.space)
	154	+	b.space = make(chan struct{})
	155	+	if len(b.queue) == 0 {
	156	+	close(b.idle)
	157	+	}
	158	+	b.mu.Unlock()
	159	+	}
	160	+	}

A internal/usecase/client/attachment_outbox_test.go +95 -0

@@ -0,0 +1,95 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/stretchr/testify/require"
	9	+
	10	+	"github.com/bnema/vev/internal/protocol"
	11	+	)
	12	+
	13	+	// gatedSessionStream blocks every write until release is closed, like a
	14	+	// carriage stalled on a dead link, and can fail writes after release.
	15	+	type gatedSessionStream struct {
	16	+	*sessionTestStream
	17	+	release chan struct{}
	18	+	fail error
	19	+	}
	20	+
	21	+	func (s *gatedSessionStream) SendClient(message protocol.ClientMessage) error {
	22	+	select {
	23	+	case <-s.release:
	24	+	case <-s.closed:
	25	+	return errSessionTestClosed
	26	+	}
	27	+	if s.fail != nil {
	28	+	return s.fail
	29	+	}
	30	+	return s.sessionTestStream.SendClient(message)
	31	+	}
	32	+
	33	+	func TestAttachmentOutbox(t *testing.T) {
	34	+	writeErr := errors.New("link lost")
	35	+	tests := []struct {
	36	+	name string
	37	+	fail error
	38	+	wantErr error
	39	+	}{
	40	+	{name: "stalled link does not block enqueue and preserves order"},
	41	+	{name: "write failure is sticky and signalled", fail: writeErr, wantErr: writeErr},
	42	+	}
	43	+	for _, tt := range tests {
	44	+	t.Run(tt.name, func(t *testing.T) {
	45	+	stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{}), fail: tt.fail}
	46	+	outbox := newAttachmentOutbox(stream)
	47	+	defer outbox.stop()
	48	+
	49	+	messages := []protocol.ClientMessage{
	50	+	protocol.Input{InputSeq: 1, Data: []byte("a")},
	51	+	protocol.Input{InputSeq: 2, Data: []byte("b")},
	52	+	protocol.Input{InputSeq: 3, Data: []byte("c")},
	53	+	}
	54	+	enqueued := make(chan struct{})
	55	+	go func() {
	56	+	for _, m := range messages {
	57	+	if err := outbox.SendClient(m); err != nil {
	58	+	t.Errorf("SendClient: %v", err)
	59	+	}
	60	+	}
	61	+	close(enqueued)
	62	+	}()
	63	+	select {
	64	+	case <-enqueued:
	65	+	case <-time.After(5 * time.Second):
	66	+	t.Fatal("SendClient blocked on a stalled link")
	67	+	}
	68	+
	69	+	close(stream.release)
	70	+	if tt.wantErr != nil {
	71	+	<-outbox.Failed()
	72	+	require.ErrorIs(t, outbox.failure(), tt.wantErr)
	73	+	require.ErrorIs(t, outbox.SendClient(protocol.Detach{}), tt.wantErr)
	74	+	return
	75	+	}
	76	+	<-outbox.Idle()
	77	+	stream.mu.Lock()
	78	+	defer stream.mu.Unlock()
	79	+	require.Equal(t, messages, stream.sent)
	80	+	})
	81	+	}
	82	+	}
	83	+
	84	+	func TestAttachmentOutboxCloseUnblocksStalledWrite(t *testing.T) {
	85	+	stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
	86	+	outbox := newAttachmentOutbox(stream)
	87	+	require.NoError(t, outbox.SendClient(protocol.Detach{}))
	88	+
	89	+	require.NoError(t, outbox.Close())
	90	+	select {
	91	+	case <-outbox.done:
	92	+	case <-time.After(5 * time.Second):
	93	+	t.Fatal("sender did not exit after Close")
	94	+	}
	95	+	}

@@ -0,0 +1,95 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"testing"
	6	+	"time"
	7	+
	8	+	"github.com/stretchr/testify/require"
	9	+
	10	+	"github.com/bnema/vev/internal/protocol"
	11	+	)
	12	+
	13	+	// gatedSessionStream blocks every write until release is closed, like a
	14	+	// carriage stalled on a dead link, and can fail writes after release.
	15	+	type gatedSessionStream struct {
	16	+	*sessionTestStream
	17	+	release chan struct{}
	18	+	fail error
	19	+	}
	20	+
	21	+	func (s *gatedSessionStream) SendClient(message protocol.ClientMessage) error {
	22	+	select {
	23	+	case <-s.release:
	24	+	case <-s.closed:
	25	+	return errSessionTestClosed
	26	+	}
	27	+	if s.fail != nil {
	28	+	return s.fail
	29	+	}
	30	+	return s.sessionTestStream.SendClient(message)
	31	+	}
	32	+
	33	+	func TestAttachmentOutbox(t *testing.T) {
	34	+	writeErr := errors.New("link lost")
	35	+	tests := []struct {
	36	+	name string
	37	+	fail error
	38	+	wantErr error
	39	+	}{
	40	+	{name: "stalled link does not block enqueue and preserves order"},
	41	+	{name: "write failure is sticky and signalled", fail: writeErr, wantErr: writeErr},
	42	+	}
	43	+	for _, tt := range tests {
	44	+	t.Run(tt.name, func(t *testing.T) {
	45	+	stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{}), fail: tt.fail}
	46	+	outbox := newAttachmentOutbox(stream)
	47	+	defer outbox.stop()
	48	+
	49	+	messages := []protocol.ClientMessage{
	50	+	protocol.Input{InputSeq: 1, Data: []byte("a")},
	51	+	protocol.Input{InputSeq: 2, Data: []byte("b")},
	52	+	protocol.Input{InputSeq: 3, Data: []byte("c")},
	53	+	}
	54	+	enqueued := make(chan struct{})
	55	+	go func() {
	56	+	for _, m := range messages {
	57	+	if err := outbox.SendClient(m); err != nil {
	58	+	t.Errorf("SendClient: %v", err)
	59	+	}
	60	+	}
	61	+	close(enqueued)
	62	+	}()
	63	+	select {
	64	+	case <-enqueued:
	65	+	case <-time.After(5 * time.Second):
	66	+	t.Fatal("SendClient blocked on a stalled link")
	67	+	}
	68	+
	69	+	close(stream.release)
	70	+	if tt.wantErr != nil {
	71	+	<-outbox.Failed()
	72	+	require.ErrorIs(t, outbox.failure(), tt.wantErr)
	73	+	require.ErrorIs(t, outbox.SendClient(protocol.Detach{}), tt.wantErr)
	74	+	return
	75	+	}
	76	+	<-outbox.Idle()
	77	+	stream.mu.Lock()
	78	+	defer stream.mu.Unlock()
	79	+	require.Equal(t, messages, stream.sent)
	80	+	})
	81	+	}
	82	+	}
	83	+
	84	+	func TestAttachmentOutboxCloseUnblocksStalledWrite(t *testing.T) {
	85	+	stream := &gatedSessionStream{sessionTestStream: newSessionTestStream(), release: make(chan struct{})}
	86	+	outbox := newAttachmentOutbox(stream)
	87	+	require.NoError(t, outbox.SendClient(protocol.Detach{}))
	88	+
	89	+	require.NoError(t, outbox.Close())
	90	+	select {
	91	+	case <-outbox.done:
	92	+	case <-time.After(5 * time.Second):
	93	+	t.Fatal("sender did not exit after Close")
	94	+	}
	95	+	}

M internal/usecase/client/attachment_session.go +33 -1


@@ -301,9 +301,13 @@ var (
301	301
302	302		// pumpAttached publishes output and concurrently forwards the foreground's
303	303		// authorized input, latest geometry and explicit lifecycle decision.
304		-	func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, state outputApplyState, incoming <-chan serverReceive) AttachmentEvent {
	304	+	func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg AttachmentForeground, raw ports.BrokerLogicalConnection, token AttachmentToken, state outputApplyState, incoming <-chan serverReceive) AttachmentEvent {
305	305		pumpCtx, cancel := context.WithCancel(ctx)
306	306		defer cancel()
	307	+	// Every send while attached goes through the outbox so a stalled link
	308	+	// never blocks key handling, overlays, or painting.
	309	+	stream := newAttachmentOutbox(raw)
	310	+	defer stream.stop()
307	311		outgoing := make(chan attachmentClientEvent)
308	312		go pumpAttachmentInput(pumpCtx, fg, outgoing)
309	313		go pumpAttachmentGeometry(pumpCtx, fg, outgoing)

@@ -572,8 +576,15 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
572	576		}
573	577		}
574	578		if event.action != 0 {
	579	+	// The lifecycle message must reach the daemon before the
	580	+	// supervisor closes the stream.
	581	+	if err := w.drainOutbox(ctx, fg, stream); err != nil {
	582	+	return w.settle(ctx, fg, stream, token, err)
	583	+	}
575	584		return attachmentLifecycleEnd(token, event.action)
576	585		}
	586	+	case <-stream.Failed():
	587	+	return w.settle(ctx, fg, stream, token, stream.failure())
577	588		case <-ctx.Done():
578	589		return w.settle(ctx, fg, stream, token, ctx.Err())
579	590		case <-fg.Done():

@@ -816,6 +827,22 @@ func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForegro
816	827		}
817	828		}
818	829
	830	+	// drainOutbox waits until every queued message was written, or the run ends.
	831	+	func (w sessionAttachmentWorker) drainOutbox(ctx context.Context, fg AttachmentForeground, outbox attachmentOutbox) error {
	832	+	select {
	833	+	case <-outbox.Idle():
	834	+	return outbox.failure()
	835	+	case <-outbox.Failed():
	836	+	return outbox.failure()
	837	+	case <-ctx.Done():
	838	+	_ = outbox.Close()
	839	+	return ctx.Err()
	840	+	case <-fg.Done():
	841	+	_ = outbox.Close()
	842	+	return errAttachmentForegroundRevoked
	843	+	}
	844	+	}
	845	+
819	846		// awaitServer waits for the next server message while the run is live.
820	847		func (w *sessionAttachmentWorker) awaitServer(ctx context.Context, fg AttachmentForeground, incoming <-chan serverReceive) (protocol.ServerMessage, error) {
821	848		select {

@@ -861,6 +888,11 @@ func (w *sessionAttachmentWorker) sendClosed(stream ports.BrokerLogicalConnectio
861	888		if supervisorNil(stream) {
862	889		return
863	890		}
	891	+	if outbox, ok := stream.(*attachmentOutbox); ok {
	892	+	// Bypass the queue: the run is ending and its sender stops with it.
	893	+	outbox.stop()
	894	+	stream = outbox.BrokerLogicalConnection
	895	+	}
864	896		sent := make(chan error, 1)
865	897		go func() { sent <- stream.SendClient(protocol.Detach{Closed: true}) }()
866	898		timer := w.cfg.Clock.NewTimer(closedDetachTimeout)


@@ -301,9 +301,13 @@ var (
301			301
302		// pumpAttached publishes output and concurrently forwards the foreground's	302		// pumpAttached publishes output and concurrently forwards the foreground's
303		// authorized input, latest geometry and explicit lifecycle decision.	303		// authorized input, latest geometry and explicit lifecycle decision.
304	-	func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, state outputApplyState, incoming <-chan serverReceive) AttachmentEvent {	304	+	func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg AttachmentForeground, raw ports.BrokerLogicalConnection, token AttachmentToken, state outputApplyState, incoming <-chan serverReceive) AttachmentEvent {
305		pumpCtx, cancel := context.WithCancel(ctx)	305		pumpCtx, cancel := context.WithCancel(ctx)
306		defer cancel()	306		defer cancel()
			307	+	// Every send while attached goes through the outbox so a stalled link
			308	+	// never blocks key handling, overlays, or painting.
			309	+	stream := newAttachmentOutbox(raw)
			310	+	defer stream.stop()
307		outgoing := make(chan attachmentClientEvent)	311		outgoing := make(chan attachmentClientEvent)
308		go pumpAttachmentInput(pumpCtx, fg, outgoing)	312		go pumpAttachmentInput(pumpCtx, fg, outgoing)
309		go pumpAttachmentGeometry(pumpCtx, fg, outgoing)	313		go pumpAttachmentGeometry(pumpCtx, fg, outgoing)

@@ -572,8 +576,15 @@ func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg Attachmen
572		}	576		}
573		}	577		}
574		if event.action != 0 {	578		if event.action != 0 {
			579	+	// The lifecycle message must reach the daemon before the
			580	+	// supervisor closes the stream.
			581	+	if err := w.drainOutbox(ctx, fg, stream); err != nil {
			582	+	return w.settle(ctx, fg, stream, token, err)
			583	+	}
575		return attachmentLifecycleEnd(token, event.action)	584		return attachmentLifecycleEnd(token, event.action)
576		}	585		}
			586	+	case <-stream.Failed():
			587	+	return w.settle(ctx, fg, stream, token, stream.failure())
577		case <-ctx.Done():	588		case <-ctx.Done():
578		return w.settle(ctx, fg, stream, token, ctx.Err())	589		return w.settle(ctx, fg, stream, token, ctx.Err())
579		case <-fg.Done():	590		case <-fg.Done():

@@ -816,6 +827,22 @@ func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForegro
816		}	827		}
817		}	828		}
818			829
			830	+	// drainOutbox waits until every queued message was written, or the run ends.
			831	+	func (w sessionAttachmentWorker) drainOutbox(ctx context.Context, fg AttachmentForeground, outbox attachmentOutbox) error {
			832	+	select {
			833	+	case <-outbox.Idle():
			834	+	return outbox.failure()
			835	+	case <-outbox.Failed():
			836	+	return outbox.failure()
			837	+	case <-ctx.Done():
			838	+	_ = outbox.Close()
			839	+	return ctx.Err()
			840	+	case <-fg.Done():
			841	+	_ = outbox.Close()
			842	+	return errAttachmentForegroundRevoked
			843	+	}
			844	+	}
			845	+
819		// awaitServer waits for the next server message while the run is live.	846		// awaitServer waits for the next server message while the run is live.
820		func (w *sessionAttachmentWorker) awaitServer(ctx context.Context, fg AttachmentForeground, incoming <-chan serverReceive) (protocol.ServerMessage, error) {	847		func (w *sessionAttachmentWorker) awaitServer(ctx context.Context, fg AttachmentForeground, incoming <-chan serverReceive) (protocol.ServerMessage, error) {
821		select {	848		select {

@@ -861,6 +888,11 @@ func (w *sessionAttachmentWorker) sendClosed(stream ports.BrokerLogicalConnectio
861		if supervisorNil(stream) {	888		if supervisorNil(stream) {
862		return	889		return
863		}	890		}
			891	+	if outbox, ok := stream.(*attachmentOutbox); ok {
			892	+	// Bypass the queue: the run is ending and its sender stops with it.
			893	+	outbox.stop()
			894	+	stream = outbox.BrokerLogicalConnection
			895	+	}
864		sent := make(chan error, 1)	896		sent := make(chan error, 1)
865		go func() { sent <- stream.SendClient(protocol.Detach{Closed: true}) }()	897		go func() { sent <- stream.SendClient(protocol.Detach{Closed: true}) }()
866		timer := w.cfg.Clock.NewTimer(closedDetachTimeout)	898		timer := w.cfg.Clock.NewTimer(closedDetachTimeout)

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


@@ -616,7 +616,8 @@ func (s Supervisor) runResolvedAttachment(ctx context.Context, input terminalI
616	616		// same target again.
617	617		outcome.target = target
618	618		}
619		-	attempt, ok := budget.spend(outcome.kind, s.cfg.Clock.Now().Sub(attemptStart))
	619	+	now := s.cfg.Clock.Now()
	620	+	attempt, ok := budget.spend(outcome.kind, now.Sub(attemptStart), now)
620	621		if !ok {
621	622		s.logger.Warn("client_attachment_resume_exhausted", "endpoint", outcome.target.request.Endpoint, "session", outcome.target.request.Target.SessionName, "attempts", attempt-1, "error", outcome.cause)
622	623		s.transition(supervisorEvent{kind: supervisorAttachEnded, err: outcome.cause})

@@ -640,6 +641,8 @@ func (s Supervisor) runResolvedAttachment(ctx context.Context, input terminalI
640	641		type attachmentResumeBudget struct {
641	642		used int
642	643		resuming bool
	644	+	// since is when the current outage began; zero before the first spend.
	645	+	since time.Time
643	646		}
644	647
645	648		// spend records one resume after an attempt of the given kind that ran for

@@ -647,14 +650,18 @@ type attachmentResumeBudget struct {
647	650		// fresh budget; one that drops right after attaching keeps counting, so a
648	651		// flapping session gives up.
649	652		// It returns the 1-based resume attempt, and false once the budget is
650		-	// exhausted.
651		-	func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration) (int, bool) {
	653	+	// exhausted: at least maxAttachmentResumes attempts and resumeOutageWindow
	654	+	// since the outage began.
	655	+	func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration, now time.Time) (int, bool) {
652	656		if kind == attachmentResume && up >= resumeStableAttachment {
653	657		b.used = 0
654	658		}
	659	+	if b.used == 0 {
	660	+	b.since = now
	661	+	}
655	662		b.used++
656	663		b.resuming = true
657		-	return b.used, b.used <= maxAttachmentResumes
	664	+	return b.used, b.used <= maxAttachmentResumes \|\| now.Sub(b.since) < resumeOutageWindow
658	665		}
659	666
660	667		// withFreshStream gives target a newly allocated stream identity. It is

@@ -898,11 +905,17 @@ settlement:
898	905		return attachmentEndedOutcome()
899	906		}
900	907
901		-	// maxAttachmentResumes bounds how many times one lost attachment reconnects
902		-	// before the client gives up and returns to the picker. With the supervisor
903		-	// backoff this spans a few seconds, enough to ride out a transient drop.
	908	+	// maxAttachmentResumes is the minimum number of resume attempts before the
	909	+	// client may give up on a lost attachment. Giving up also requires the outage
	910	+	// to have lasted resumeOutageWindow, so a flaky mobile link keeps resuming
	911	+	// instead of dropping to the picker after a few seconds.
904	912		const maxAttachmentResumes = 5
905	913
	914	+	// resumeOutageWindow is how long one outage keeps resuming. It matches the
	915	+	// daemon's park window for a detached remote attachment: past it the session
	916	+	// view is gone anyway.
	917	+	const resumeOutageWindow = 15 * time.Minute
	918	+
906	919		// resumeStableAttachment is how long a resumed attachment must stay up before
907	920		// its next loss gets a fresh resume budget.
908	921		const resumeStableAttachment = 30 * time.Second

 			// same target again.
 			outcome.target = target
 		}
-		attempt, ok := budget.spend(outcome.kind, s.cfg.Clock.Now().Sub(attemptStart))
+		now := s.cfg.Clock.Now()
+		attempt, ok := budget.spend(outcome.kind, now.Sub(attemptStart), now)
 		if !ok {
 			s.logger.Warn("client_attachment_resume_exhausted", "endpoint", outcome.target.request.Endpoint, "session", outcome.target.request.Target.SessionName, "attempts", attempt-1, "error", outcome.cause)
 			s.transition(supervisorEvent{kind: supervisorAttachEnded, err: outcome.cause})
 type attachmentResumeBudget struct {
 	used     int
 	resuming bool
+	// since is when the current outage began; zero before the first spend.
+	since time.Time
 }
 // spend records one resume after an attempt of the given kind that ran for
 // 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.
+// exhausted: at least maxAttachmentResumes attempts and resumeOutageWindow
-func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration) (int, bool) {
+// since the outage began.
+func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration, now time.Time) (int, bool) {
 	if kind == attachmentResume && up >= resumeStableAttachment {
 		b.used = 0
 	}
+	if b.used == 0 {
+		b.since = now
+	}
 	b.used++
 	b.resuming = true
-	return b.used, b.used <= maxAttachmentResumes
+	return b.used, b.used <= maxAttachmentResumes || now.Sub(b.since) < resumeOutageWindow
 }
 // withFreshStream gives target a newly allocated stream identity. It is
 	return attachmentEndedOutcome()
 }
-// maxAttachmentResumes bounds how many times one lost attachment reconnects
+// maxAttachmentResumes is the minimum number of resume attempts before the
-// before the client gives up and returns to the picker. With the supervisor
+// client may give up on a lost attachment. Giving up also requires the outage
-// backoff this spans a few seconds, enough to ride out a transient drop.
+// to have lasted resumeOutageWindow, so a flaky mobile link keeps resuming
+// instead of dropping to the picker after a few seconds.
 const maxAttachmentResumes = 5
+// resumeOutageWindow is how long one outage keeps resuming. It matches the
+// daemon's park window for a detached remote attachment: past it the session
+// view is gone anyway.
+const resumeOutageWindow = 15 * time.Minute
 // resumeStableAttachment is how long a resumed attachment must stay up before
 // its next loss gets a fresh resume budget.
 const resumeStableAttachment = 30 * time.Second

M internal/usecase/client/supervisor_attach_test.go +29 -6


@@ -1307,11 +1307,19 @@ func resumeTestJitter() float64 { return 0.5 }
1307	1307		// fireResumeTimer fires the resume backoff timer for one attempt. Other timers
1308	1308		// share the clock, so it matches the exact backoff delay.
1309	1309		func fireResumeTimer(t testing.T, clock supervisorTestClock, attempt int) {
	1310	+	t.Helper()
	1311	+	fireResumeTimerAfter(t, clock, attempt, func() {})
	1312	+	}
	1313	+
	1314	+	// fireResumeTimerAfter waits for the attempt's backoff timer, runs before, then
	1315	+	// fires it.
	1316	+	func fireResumeTimerAfter(t testing.T, clock supervisorTestClock, attempt int, before func()) {
1310	1317		t.Helper()
1311	1318		want := supervisorBackoffDelay(uint64(attempt), resumeTestJitter)
1312	1319		for {
1313	1320		timer := clock.awaitTimer(t)
1314	1321		if timer.delay == want {
	1322	+	before()
1315	1323		timer.fire()
1316	1324		return
1317	1325		}

@@ -1320,22 +1328,29 @@ func fireResumeTimer(t testing.T, clock supervisorTestClock, attempt int) {
1320	1328
1321	1329		func TestAttachmentResumeBudget(t *testing.T) {
1322	1330		brief := time.Second
	1331	+	// step is the clock advance between spends: past the outage window, the
	1332	+	// attempt count alone decides; within it, resuming never gives up.
	1333	+	past := resumeOutageWindow / maxAttachmentResumes
1323	1334		tests := []struct {
1324	1335		name string
1325	1336		steps []attachmentOutcomeKind
1326	1337		ups []time.Duration
	1338	+	step time.Duration
1327	1339		want []bool
1328	1340		}{
1329		-	{name: "flapping resumes exhaust", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), want: append(repeatBools(true, maxAttachmentResumes), false)},
1330		-	{name: "stable attachment restores the budget", steps: repeatKinds(attachmentResume, maxAttachmentResumes+2), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+2), want: repeatBools(true, maxAttachmentResumes+2)},
1331		-	{name: "retries never restore the budget", steps: repeatKinds(attachmentRetry, maxAttachmentResumes+1), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+1), 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)},
	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: "outage window keeps resuming", steps: repeatKinds(attachmentRetry, 3maxAttachmentResumes), ups: repeatDurations(brief, 3maxAttachmentResumes), step: time.Second, want: repeatBools(true, 3*maxAttachmentResumes)},
1332	1345		}
1333	1346		for _, tt := range tests {
1334	1347		t.Run(tt.name, func(t *testing.T) {
1335	1348		var budget attachmentResumeBudget
1336	1349		require.False(t, budget.resuming)
	1350	+	now := time.Unix(0, 0)
1337	1351		for i, kind := range tt.steps {
1338		-	_, ok := budget.spend(kind, tt.ups[i])
	1352	+	_, ok := budget.spend(kind, tt.ups[i], now)
	1353	+	now = now.Add(tt.step)
1339	1354		require.Equal(t, tt.want[i], ok, "step %d", i)
1340	1355		require.True(t, budget.resuming)
1341	1356		}

@@ -1443,7 +1458,7 @@ func TestSupervisorAttachmentLossResumesSameSession(t *testing.T) {
1443	1458		first.fail(testStreamLoss())
1444	1459
1445	1460		attempt := 1
1446		-	for _, s := range tc.steps {
	1461	+	for i, s := range tc.steps {
1447	1462		if s == cancelDuringBackoff {
1448	1463		backoff := supervisorBackoffDelay(1, resumeTestJitter)
1449	1464		for harness.clock.awaitTimer(t).delay != backoff {

@@ -1456,7 +1471,15 @@ func TestSupervisorAttachmentLossResumesSameSession(t *testing.T) {
1456	1471		refusing.Store(s == refuse)
1457	1472		unavailable.Store(s == openUnavailable)
1458	1473		opened := len(harness.service.openedRequests())
1459		-	fireResumeTimer(t, harness.clock, attempt)
	1474	+	// Giving up also requires the outage window to pass. The
	1475	+	// clock moves only once the last attempt's backoff is
	1476	+	// scheduled, so no attempt measures it as uptime.
	1477	+	last := tc.wantPicker && i == len(tc.steps)-1
	1478	+	fireResumeTimerAfter(t, harness.clock, attempt, func() {
	1479	+	if last {
	1480	+	advancePickerClock(harness.clock, resumeOutageWindow)
	1481	+	}
	1482	+	})
1460	1483		if s == refuse {
1461	1484		break
1462	1485		}

 // fireResumeTimer fires the resume backoff timer for one attempt. Other timers
 // share the clock, so it matches the exact backoff delay.
 func fireResumeTimer(t *testing.T, clock *supervisorTestClock, attempt int) {
+	t.Helper()
+	fireResumeTimerAfter(t, clock, attempt, func() {})
+}
+// fireResumeTimerAfter waits for the attempt's backoff timer, runs before, then
+// fires it.
+func fireResumeTimerAfter(t *testing.T, clock *supervisorTestClock, attempt int, before func()) {
 	t.Helper()
 	want := supervisorBackoffDelay(uint64(attempt), resumeTestJitter)
 	for {
 		timer := clock.awaitTimer(t)
 		if timer.delay == want {
+			before()
 			timer.fire()
 			return
 		}
 func TestAttachmentResumeBudget(t *testing.T) {
 	brief := time.Second
+	// step is the clock advance between spends: past the outage window, the
+	// attempt count alone decides; within it, resuming never gives up.
+	past := resumeOutageWindow / maxAttachmentResumes
 	tests := []struct {
 		name  string
 		steps []attachmentOutcomeKind
 		ups   []time.Duration
+		step  time.Duration
 		want  []bool
 	}{
-		{name: "flapping resumes exhaust", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), want: append(repeatBools(true, maxAttachmentResumes), false)},
+		{name: "flapping resumes exhaust", steps: repeatKinds(attachmentResume, maxAttachmentResumes+1), ups: repeatDurations(brief, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
-		{name: "stable attachment restores the budget", steps: repeatKinds(attachmentResume, maxAttachmentResumes+2), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+2), want: repeatBools(true, maxAttachmentResumes+2)},
+		{name: "stable attachment restores the budget", steps: repeatKinds(attachmentResume, maxAttachmentResumes+2), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+2), step: past, want: repeatBools(true, maxAttachmentResumes+2)},
-		{name: "retries never restore the budget", steps: repeatKinds(attachmentRetry, maxAttachmentResumes+1), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+1), want: append(repeatBools(true, maxAttachmentResumes), false)},
+		{name: "retries never restore the budget", steps: repeatKinds(attachmentRetry, maxAttachmentResumes+1), ups: repeatDurations(resumeStableAttachment, maxAttachmentResumes+1), step: past, want: append(repeatBools(true, maxAttachmentResumes), false)},
+		{name: "outage window keeps resuming", steps: repeatKinds(attachmentRetry, 3*maxAttachmentResumes), ups: repeatDurations(brief, 3*maxAttachmentResumes), step: time.Second, want: repeatBools(true, 3*maxAttachmentResumes)},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			var budget attachmentResumeBudget
 			require.False(t, budget.resuming)
+			now := time.Unix(0, 0)
 			for i, kind := range tt.steps {
-				_, ok := budget.spend(kind, tt.ups[i])
+				_, ok := budget.spend(kind, tt.ups[i], now)
+				now = now.Add(tt.step)
 				require.Equal(t, tt.want[i], ok, "step %d", i)
 				require.True(t, budget.resuming)
 			}
 			first.fail(testStreamLoss())
 			attempt := 1
-			for _, s := range tc.steps {
+			for i, s := range tc.steps {
 				if s == cancelDuringBackoff {
 					backoff := supervisorBackoffDelay(1, resumeTestJitter)
 					for harness.clock.awaitTimer(t).delay != backoff {
 				refusing.Store(s == refuse)
 				unavailable.Store(s == openUnavailable)
 				opened := len(harness.service.openedRequests())
-				fireResumeTimer(t, harness.clock, attempt)
+				// Giving up also requires the outage window to pass. The
+				// clock moves only once the last attempt's backoff is
+				// scheduled, so no attempt measures it as uptime.
+				last := tc.wantPicker && i == len(tc.steps)-1
+				fireResumeTimerAfter(t, harness.clock, attempt, func() {
+					if last {
+						advancePickerClock(harness.clock, resumeOutageWindow)
+					}
+				})
 				if s == refuse {
 					break
 				}