client.go RawHistoryFolder

1// Package client implements vev's thin attach client: it speaks the framed
2// IPC protocol to a daemon, pumping terminal input, resize events, and PTY
3// output between the controlling terminal and the session.
4//
5// The package depends only on ports (transport, terminal, wire codecs) and
6// domain, never on a concrete transport or terminal implementation; the app
7// layer wires those in.
8package client
9
10import (
11	"context"
12	"crypto/rand"
13	"errors"
14	"fmt"
15	"io"
16	"log/slog"
17	"os"
18	"sync"
19	"sync/atomic"
20	"time"
21
22	renderer "github.com/bnema/vev-vt"
23	"github.com/bnema/vev/internal/domain"
24	"github.com/bnema/vev/internal/ports"
25	"github.com/bnema/vev/internal/usecase/theme"
26)
27
28// ProtocolError is a session- or protocol-level failure reported by the
29// daemon before attach could proceed (a Hello rejected with an ErrorMsg).
30// The app layer prints it to the user.
31type ProtocolError struct {
32	Code uint16
33	Text string
34}
35
36func (e *ProtocolError) Error() string {
37	if e.Text == "" {
38		return fmt.Sprintf("vev: daemon rejected attach (code %d)", e.Code)
39	}
40	return fmt.Sprintf("vev: %s", e.Text)
41}
42
43// DetachedError reports that the daemon detached this client for a reason
44// other than an ordinary user-initiated detach.
45type DetachedError struct {
46	Reason uint8
47	Text   string
48}
49
50func (e *DetachedError) Error() string { return "vev: " + e.Text }
51
52// terminalThemeState retains the latest terminal-reported colors across
53// transport attempts so a replacement attachment can restore the daemon theme
54// without relying on another OSC response.
55type terminalThemeState struct {
56	mu    sync.Mutex
57	theme ports.Theme
58}
59
60func (s *terminalThemeState) setTrueColor(enabled bool) {
61	s.mu.Lock()
62	s.theme.TrueColor = enabled
63	s.mu.Unlock()
64}
65
66func (s *terminalThemeState) update(update func(*ports.Theme)) ports.Theme {
67	s.mu.Lock()
68	defer s.mu.Unlock()
69	update(&s.theme)
70	return s.theme
71}
72
73func (s *terminalThemeState) snapshot() ports.Theme {
74	s.mu.Lock()
75	defer s.mu.Unlock()
76	return s.theme
77}
78
79// milestones tracks lifecycle progress so a failure can report which stages
80// were never reached — the key diagnostic when an attach dies early.
81type milestones struct {
82	dialed      bool
83	helloSent   bool
84	welcomed    bool
85	rawEntered  bool
86	firstOutput bool
87	detached    bool
88}
89
90func (m milestones) missing() []string {
91	var out []string
92	if !m.dialed {
93		out = append(out, "dialed")
94	}
95	if !m.helloSent {
96		out = append(out, "hello_sent")
97	}
98	if !m.welcomed {
99		out = append(out, "welcomed")
100	}
101	if !m.rawEntered {
102		out = append(out, "raw_entered")
103	}
104	if !m.firstOutput {
105		out = append(out, "first_output")
106	}
107	if !m.detached {
108		out = append(out, "detached")
109	}
110	return out
111}
112
113// processClientID identifies this client process across links. It is kept in
114// memory only; a fresh process gets a fresh ID.
115var processClientID = newClientID()
116
117func newClientID() [16]byte {
118	var id [16]byte
119	if _, err := rand.Read(id[:]); err != nil {
120		panic("crypto/rand failed generating client ID: " + err.Error())
121	}
122	return id
123}
124
125// stdinBufSize is the read buffer for the terminal input pump.
126const stdinBufSize = 4096
127
128// sendQueueDepth bounds the buffered frames awaiting the single sender
129// goroutine, decoupling the input/resize pumps from transport back-pressure.
130const sendQueueDepth = 64
131
132// preWelcomeTimeout remains the focused-test name for the shared handshake
133// budget, which now also covers connect and initial publication.
134const preWelcomeTimeout = ports.HandshakeTimeout
135
136// remoteHostLearnerShutdownTimeout bounds how long Run waits for async remote
137// host learning after terminal restoration. Learning is best-effort; a stalled
138// learner must not block shell prompt return indefinitely.
139const remoteHostLearnerShutdownTimeout = time.Second
140
141var reconnectSleep = sleepReconnect
142var reconnectSleepWithResize = sleepReconnectWithResizeEvents
143
144const (
145	statusReconnect = "\r\x1b[2Kreconnecting…"
146	statusClear     = "\r\x1b[2K"
147)
148
149// Dependencies supplies the collaborators required by a Runner.
150type AttachHandoffFunc func(ports.AttachTarget) (ports.Dialer, AttachRequest, error)
151
152type Dependencies struct {
153	Dialer            ports.Dialer
154	Terminal          ports.Terminal
155	Clock             ports.Clock
156	Clipboard         ports.ClipboardReader
157	Logger            *slog.Logger
158	RuntimeObserver   ports.SerializedRuntimeObserver
159	RemoteHostLearner ports.RemoteHostLearner
160	// AttachHandoff keeps one Runner and one terminal/input ownership across a
161	// local daemon's structured remote handoff. It is nil for direct CLI attach.
162	AttachHandoff AttachHandoffFunc
163	// Remote selects client-side carriage presentation only; it never enters
164	// the daemon-facing session request.
165	Remote bool
166	// Origin is explicit composition metadata for the initial route. A zero
167	// value is normalized to local for direct package callers and old tests.
168	Origin    ports.RouteOrigin
169	OriginKey string
170}
171
172// AttachRequest identifies the session for one client run. Remote is
173// client-only presentation metadata. RemoteTarget is an optional exact picker
174// handoff and is serialized into Hello only when present.
175type AttachRequest struct {
176	Intent                 uint8
177	SessionName            string
178	Remote                 bool
179	Origin                 ports.RouteOrigin
180	OriginKey              string
181	RemoteTarget           *domain.RemoteSessionTarget
182	ExactTarget            *ports.ExactSessionTarget
183	PreferredTabID         domain.TabStableID
184	RemoteOrigin           string
185	EnvironmentPolicy      ports.EnvironmentPolicy
186	NavigationCapabilities ports.NavigationCapabilities
187	StartupOverlay         ports.StartupOverlay
188}
189
190// attachRoute captures the dialer, request, and resume token needed to
191// restore a previous attachment route after a navigation handoff.
192type attachRoute struct {
193	dialer      ports.Dialer
194	request     AttachRequest
195	resumeToken uint64
196}
197
198type Runner struct {
199	dialer            ports.Dialer
200	term              ports.Terminal
201	clock             ports.Clock
202	clipboard         ports.ClipboardReader
203	logger            *slog.Logger
204	runtimeObserver   ports.SerializedRuntimeObserver
205	remoteHostLearner ports.RemoteHostLearner
206	attachHandoff     AttachHandoffFunc
207	remote            bool
208	origin            ports.RouteOrigin
209	ledger            *routeLedger
210	routeFailure      *ports.RouteNavigationFailure
211
212	inputMu sync.Mutex
213	input   *terminalInputPump
214}
215
216// NewRunner constructs a client runner. A nil logger uses the process default.
217func NewRunner(deps Dependencies) *Runner {
218	log := deps.Logger
219	if log == nil {
220		log = slog.Default()
221	}
222	return &Runner{
223		dialer:            deps.Dialer,
224		term:              deps.Terminal,
225		clock:             deps.Clock,
226		clipboard:         deps.Clipboard,
227		logger:            log,
228		runtimeObserver:   deps.RuntimeObserver,
229		remoteHostLearner: deps.RemoteHostLearner,
230		attachHandoff:     deps.AttachHandoff,
231		remote:            deps.Remote,
232		origin:            normalizeRouteOrigin(deps.Origin, deps.Remote),
233		ledger:            newRouteLedger(),
234	}
235}
236
237// terminalInput returns the Runner-owned sole reader. It survives sequential
238// Run calls because a caller-owned reader may remain blocked after an attach
239// exits and therefore cannot safely be replaced by another reader.
240func (r *Runner) terminalInput() *terminalInputPump {
241	r.inputMu.Lock()
242	defer r.inputMu.Unlock()
243
244	if r.input == nil || r.input.hasExited() || r.input.isClosed() {
245		r.input = newTerminalInputPump(r.term.In())
246		r.input.start()
247	}
248	r.input.resume()
249	return r.input
250}
251
252func validateAttachRequest(request AttachRequest) error {
253	if request.Origin != 0 {
254		if err := request.Origin.Validate(); err != nil {
255			return fmt.Errorf("vev: invalid route origin: %w", err)
256		}
257	}
258	if request.OriginKey != "" {
259		if err := ports.ValidateRouteLabel(request.OriginKey, false); err != nil {
260			return fmt.Errorf("vev: invalid route origin key: %w", err)
261		}
262	}
263	if request.ExactTarget != nil {
264		if err := request.ExactTarget.Validate(); err != nil {
265			return fmt.Errorf("vev: invalid exact target: %w", err)
266		}
267		if request.SessionName != request.ExactTarget.SessionName {
268			return errors.New("vev: exact target session mismatch")
269		}
270		if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != request.ExactTarget.LifecycleID || request.RemoteTarget.SessionName != request.ExactTarget.SessionName) {
271			return errors.New("vev: exact target does not match remote target")
272		}
273	}
274	if err := (SessionTarget{Intent: request.Intent, SessionName: request.SessionName}).validate(); err != nil {
275		return fmt.Errorf("vev: invalid session target: %w", err)
276	}
277	if err := ports.ValidateNavigation(request.NavigationCapabilities, request.StartupOverlay, request.RemoteTarget != nil || request.EnvironmentPolicy == ports.EnvironmentPolicyDaemonOwned); err != nil {
278		return fmt.Errorf("vev: invalid navigation route: %w", err)
279	}
280	if request.RemoteTarget == nil {
281		if request.EnvironmentPolicy == ports.EnvironmentPolicyClientOwned {
282			return nil
283		}
284		if request.Remote && request.EnvironmentPolicy == ports.EnvironmentPolicyDaemonOwned {
285			return nil
286		}
287		return errors.New("vev: attach without remote target requires a matching environment policy")
288	}
289	if err := request.RemoteTarget.Validate(); err != nil {
290		return fmt.Errorf("vev: invalid remote attach target: %w", err)
291	}
292	if request.SessionName != request.RemoteTarget.SessionName {
293		return errors.New("vev: remote attach target session mismatch")
294	}
295	if request.EnvironmentPolicy != ports.EnvironmentPolicyDaemonOwned {
296		return errors.New("vev: remote attach target requires daemon-owned environment")
297	}
298	if request.Intent != ports.IntentAttach && request.Intent != ports.IntentResume {
299		return errors.New("vev: remote attach target requires attach or resume")
300	}
301	return nil
302}
303
304// Run connects and runs the attach client. It owns the terminal lifecycle
305// above attach attempts so raw mode remains active while a live client process
306// redials a lost link.
307func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) {
308	request = cloneAttachRequest(request)
309	if request.Origin == 0 {
310		request.Origin = r.origin
311	}
312	if err := validateAttachRequest(request); err != nil {
313		return err
314	}
315	ms := milestones{}
316
317	defer func() {
318		if retErr != nil {
319			r.logger.Error("attach ended with error", "err", retErr, "missing_milestones", ms.missing())
320		}
321	}()
322
323	var restore func() error
324	rawEntered := false
325	enterRaw := func() error {
326		if rawEntered {
327			return nil
328		}
329		var err error
330		restore, err = r.term.EnterRaw()
331		if err != nil {
332			return fmt.Errorf("vev: entering raw mode: %w", err)
333		}
334		rawEntered = true
335		ms.rawEntered = true
336		return nil
337	}
338
339	resumeToken := uint64(0)
340	attemptRequest := request
341	dialer := r.dialer
342	var homeRoute *attachRoute
343	var returnRoute *attachRoute
344	returnResumeFallback := false
345	homeNavigationPending := false
346	returnNavigationPending := false
347	routeNavigationPending := false
348	routeNavigationResumeFallback := false
349	var routeNavigationSelection *routeNavigationSelection
350	var routeNavigationFallback *attachRoute
351	var routeNavigationAction *ports.RouteNavigationAction
352	backoff := defaultReconnectBackoff.initial
353	themeState := &terminalThemeState{}
354	var rememberOnce sync.Once
355	var rememberWG sync.WaitGroup
356	var learnerStarted bool
357	rememberRemoteHost := func() {
358		if r.remoteHostLearner == nil {
359			return
360		}
361		rememberOnce.Do(func() {
362			learnerStarted = true
363			rememberWG.Go(func() {
364				if err := r.remoteHostLearner.RememberRemoteHost(); err != nil {
365					r.logger.Warn("remembering remote host failed", "err", err)
366				}
367			})
368		})
369	}
370	defer func() {
371		if !learnerStarted {
372			return
373		}
374		done := make(chan struct{})
375		go func() {
376			rememberWG.Wait()
377			close(done)
378		}()
379		timer := r.clock.NewTimer(remoteHostLearnerShutdownTimeout)
380		defer timer.Stop()
381		select {
382		case <-done:
383		case <-timer.C():
384			r.logger.Warn("remote host learner stalled past shutdown timeout")
385		}
386	}()
387	defer func() {
388		if rawEntered {
389			if rerr := restore(); rerr != nil && retErr == nil {
390				retErr = fmt.Errorf("vev: restoring terminal: %w", rerr)
391			}
392		}
393	}()
394	remote := request.Remote || r.remote
395	reconnect := &reconnectUI{
396		term:       r.term,
397		remote:     remote,
398		rawEntered: &rawEntered,
399		stage:      reconnectStageOfflineRetrying,
400	}
401	restoreReturnRoute := func(route attachRoute) {
402		returnNavigationPending = true
403		dialer = route.dialer
404		attemptRequest = route.request
405		if route.resumeToken != 0 {
406			attemptRequest.Intent = ports.IntentResume
407		} else {
408			attemptRequest.Intent = ports.IntentAttach
409		}
410		attemptRequest.StartupOverlay = ports.StartupOverlayNone
411		attemptRequest.NavigationCapabilities &^= ports.NavigationCapabilityBack | ports.NavigationCapabilityHomePicker
412		resumeToken = route.resumeToken
413		returnResumeFallback = route.resumeToken != 0
414		remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote)
415		backoff = defaultReconnectBackoff.initial
416	}
417	enterHomePicker := func() {
418		dialer = homeRoute.dialer
419		attemptRequest = homeRoute.request
420		attemptRequest.Intent = ports.IntentAttach
421		attemptRequest.NavigationCapabilities = ports.NavigationCapabilityBack
422		attemptRequest.StartupOverlay = ports.StartupOverlaySessionPicker
423		attemptRequest.RemoteTarget = nil
424		attemptRequest.EnvironmentPolicy = ports.EnvironmentPolicyClientOwned
425		attemptRequest.Remote = homeRoute.request.Remote || r.remote
426		resumeToken = 0
427		remote = syncReconnectRemote(reconnect, homeRoute.request.Remote || r.remote)
428		backoff = defaultReconnectBackoff.initial
429	}
430	var input *terminalInputPump
431	defer func() {
432		if input != nil {
433			input.suspend()
434		}
435	}()
436	originalEnterRaw := enterRaw
437	enterRaw = func() error {
438		if err := originalEnterRaw(); err != nil {
439			return err
440		}
441		if input == nil {
442			input = r.terminalInput()
443		}
444		return nil
445	}
446
447	for {
448		if err := validateAttachRequest(attemptRequest); err != nil {
449			return err
450		}
451		handshakeCtx, timedOut, finishHandshake := newHandshakeContext(ctx, r.clock)
452		transport, err := boundedDial(handshakeCtx, dialer)
453		if err != nil {
454			err = handshakeContextError(ctx, timedOut, err)
455			finishHandshake()
456			if resumeToken == 0 || ctx.Err() != nil {
457				if clearErr := reconnect.clear(); clearErr != nil {
458					return errors.Join(err, clearErr)
459				}
460				return err
461			}
462			r.logger.Warn("reconnect dial failed", "err", err, "backoff", backoff)
463			if drawErr := reconnect.draw(); drawErr != nil {
464				return errors.Join(err, drawErr)
465			}
466			slept, sleepErr := reconnect.sleep(ctx, r.clock, backoff)
467			if sleepErr != nil {
468				return errors.Join(err, sleepErr)
469			}
470			if !slept {
471				if clearErr := reconnect.clear(); clearErr != nil {
472					return errors.Join(ctx.Err(), clearErr)
473				}
474				return ctx.Err()
475			}
476			backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max)
477			attemptRequest.Intent = ports.IntentResume
478			continue
479		}
480		ms.dialed = true
481		if transport == nil {
482			finishHandshake()
483			return errors.New("vev: dialer returned nil transport")
484		}
485		stopHandshakeTransport := watchHandshakeTransport(handshakeCtx, transport)
486		connection, err := NewSessionConnection(transport, SessionTarget{
487			Intent:      attemptRequest.Intent,
488			SessionName: attemptRequest.SessionName,
489		})
490		if err != nil {
491			stopHandshakeTransport()
492			finishHandshake()
493			_ = transport.Close()
494			return err
495		}
496
497		var linkEvents <-chan ports.LinkEvent
498		if reporter, ok := connection.Transport().(ports.LinkStateReporter); ok {
499			linkEvents = reporter.LinkEvents()
500		}
501		result := (&attachAttempt{
502			runner:                   r,
503			dialer:                   dialer,
504			connection:               connection,
505			remote:                   remote,
506			handshakeCtx:             handshakeCtx,
507			handshakeTimedOut:        timedOut,
508			finishHandshake:          finishHandshake,
509			stopHandshakeTransport:   stopHandshakeTransport,
510			request:                  attemptRequest,
511			resumeToken:              resumeToken,
512			routeNavigationSelection: routeNavigationSelection,
513			clientID:                 processClientID,
514			milestones:               &ms,
515			themeState:               themeState,
516			enterRaw:                 enterRaw,
517			reconnect:                reconnect,
518			linkEvents:               linkEvents,
519			rememberRemoteHost:       rememberRemoteHost,
520			terminalInput: func() *terminalInputPump {
521				return input
522			},
523		}).run(ctx)
524		stopHandshakeTransport()
525		finishHandshake()
526		if result.welcomed {
527			backoff = defaultReconnectBackoff.initial
528		}
529		if !result.transportClosed {
530			_ = connection.Close()
531		}
532		// Apply metadata from every processed frame before dispatching its
533		// navigation action. A committed identity can arrive immediately before
534		// a home/back action and must become the route request's authority first.
535		if result.resumeToken != 0 {
536			resumeToken = result.resumeToken
537		}
538		if result.sessionName != "" {
539			attemptRequest.SessionName = result.sessionName
540		}
541		if result.committedIdentity != nil {
542			identity := *result.committedIdentity
543			if attemptRequest.ExactTarget != nil && *attemptRequest.ExactTarget != identity.Target {
544				attemptRequest.PreferredTabID = ""
545			}
546			if remoteTarget := attemptRequest.RemoteTarget; remoteTarget != nil &&
547				(remoteTarget.LifecycleID != identity.Target.LifecycleID || remoteTarget.SessionName != identity.Target.SessionName) {
548				attemptRequest.RemoteTarget = nil
549			}
550			attemptRequest.ExactTarget = &identity.Target
551		}
552		if result.routePosition != nil && attemptRequest.ExactTarget != nil &&
553			*attemptRequest.ExactTarget == result.routePosition.Target {
554			attemptRequest.PreferredTabID = result.routePosition.ActiveTabID
555			attemptRequest.RemoteTarget = nil
556		}
557		if result.err == nil && result.welcomed && routeNavigationPending {
558			routeNavigationPending = false
559			routeNavigationSelection = nil
560			routeNavigationResumeFallback = false
561			routeNavigationFallback = nil
562			routeNavigationAction = nil
563		}
564		if result.routeAction != nil {
565			if r.ledger == nil {
566				return errors.New("vev: route ledger unavailable")
567			}
568			selection, valid := r.ledger.navigationSelection(*result.routeAction)
569			if !valid {
570				return errors.New("vev: stale route navigation action")
571			}
572			if selection.noOp {
573				continue
574			}
575			action := *result.routeAction
576			routeNavigationAction = &action
577			routeNavigationFallback = &attachRoute{dialer: selection.prior.dialer, request: selection.prior.request, resumeToken: selection.prior.resumeToken}
578			dialer = selection.selected.dialer
579			attemptRequest = cloneAttachRequest(selection.selected.request)
580			if selection.selected.resumeToken != 0 {
581				attemptRequest.Intent = ports.IntentResume
582			} else {
583				attemptRequest.Intent = ports.IntentAttach
584			}
585			// Back is transient to the picker overlay. Re-derive the home-picker
586			// capability from the selected route instead of trusting request
587			// metadata, which may have been cleared after a daemon-side switch.
588			attemptRequest.StartupOverlay = ports.StartupOverlayNone
589			attemptRequest.NavigationCapabilities = 0
590			if homeRoute != nil && selection.selected.presentation.kind == ports.RouteKindRemote {
591				attemptRequest.NavigationCapabilities = ports.NavigationCapabilityHomePicker
592			}
593			resumeToken = selection.selected.resumeToken
594			routeNavigationPending = true
595			routeNavigationSelection = &selection
596			routeNavigationResumeFallback = selection.selected.resumeToken != 0
597			remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote)
598			backoff = defaultReconnectBackoff.initial
599			continue
600		}
601		if result.action != 0 {
602			switch result.action {
603			case ports.NavigationOpenHomePicker:
604				// The daemon only sends this action after accepting a Hello that
605				// advertised Home support. The durable client-side prerequisite is
606				// the captured route itself; request metadata may have been rebased by
607				// committed identity or cursor updates since that Hello.
608				if homeRoute == nil {
609					return errors.New("vev: stale home navigation action")
610				}
611				returnRequest := attemptRequest
612				if result.sessionName != "" {
613					returnRequest.SessionName = result.sessionName
614				}
615				returnRoute = &attachRoute{dialer: dialer, request: returnRequest, resumeToken: result.resumeToken}
616				homeNavigationPending = true
617				enterHomePicker()
618				continue
619			case ports.NavigationBack:
620				// As with Home, the daemon accepted Back in Hello before sending
621				// this action. The retained concrete route is the durable client
622				// prerequisite; request metadata may have been rebased meanwhile.
623				if returnRoute == nil {
624					return errors.New("vev: stale return navigation action")
625				}
626				route := *returnRoute
627				returnRoute = nil
628				homeNavigationPending = false
629				returnNavigationPending = true
630				restoreReturnRoute(route)
631				continue
632			default:
633				return errors.New("vev: unsupported navigation action")
634			}
635		}
636		if result.target != nil {
637			if result.target.RemoteTarget != nil || result.target.Endpoint != "" {
638				if homeRoute == nil && attemptRequest.RemoteTarget == nil {
639					routeRequest := attemptRequest
640					if result.sessionName != "" {
641						routeRequest.SessionName = result.sessionName
642					}
643					routeRequest.Intent = ports.IntentAttach
644					routeRequest.NavigationCapabilities = 0
645					routeRequest.StartupOverlay = ports.StartupOverlayNone
646					homeRoute = &attachRoute{dialer: dialer, request: routeRequest}
647				}
648			}
649			var (
650				nextDialer  ports.Dialer
651				nextRequest AttachRequest
652				handoffErr  error
653			)
654			if result.target.Endpoint == "" {
655				if r.ledger == nil {
656					return errors.New("vev: route ledger unavailable")
657				}
658				nextDialer = dialer
659				nextRequest = r.ledger.samePeerHandoff(attemptRequest, *result.target)
660			} else {
661				if r.attachHandoff == nil {
662					return &AttachTargetError{Target: *result.target}
663				}
664				nextDialer, nextRequest, handoffErr = r.attachHandoff(*result.target)
665				if handoffErr != nil {
666					return handoffErr
667				}
668			}
669			if result.target.ExactTarget != nil {
670				nextRequest.ExactTarget = result.target.ExactTarget
671			}
672			if homeRoute != nil && (nextRequest.RemoteTarget != nil || nextRequest.EnvironmentPolicy == ports.EnvironmentPolicyDaemonOwned) {
673				nextRequest.NavigationCapabilities |= ports.NavigationCapabilityHomePicker
674			}
675			if attemptRequest.StartupOverlay == ports.StartupOverlaySessionPicker {
676				returnRoute = nil
677				returnResumeFallback = false
678			}
679			nextRequest = cloneAttachRequest(nextRequest)
680			if err := validateAttachRequest(nextRequest); err != nil {
681				return fmt.Errorf("vev: invalid route handoff request: %w", err)
682			}
683			if nextDialer == nil {
684				return errors.New("vev: route handoff returned nil dialer")
685			}
686			dialer = nextDialer
687			attemptRequest = nextRequest
688			remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote)
689			resumeToken = 0
690			backoff = defaultReconnectBackoff.initial
691			continue
692		}
693		if result.err == nil {
694			if result.welcomed && returnNavigationPending {
695				returnNavigationPending = false
696				returnResumeFallback = false
697			}
698			if clearErr := reconnect.clear(); clearErr != nil {
699				return clearErr
700			}
701			return nil
702		}
703		if result.welcomed {
704			homeNavigationPending = false
705		}
706		if homeNavigationPending && returnRoute != nil {
707			route := *returnRoute
708			returnRoute = nil
709			homeNavigationPending = false
710			returnNavigationPending = true
711			restoreReturnRoute(route)
712			continue
713		}
714		if routeNavigationPending && routeNavigationResumeFallback && resumeNeedsExactAttach(result.err) {
715			attemptRequest.Intent = ports.IntentAttach
716			resumeToken = 0
717			routeNavigationResumeFallback = false
718			continue
719		}
720		if routeNavigationPending && routeNavigationFallback != nil {
721			if routeNavigationAction != nil {
722				action := *routeNavigationAction
723				r.routeFailure = &ports.RouteNavigationFailure{Key: action.Key, Generation: action.Generation, Code: routeFailureCode(result.err)}
724				routeNavigationAction = nil
725			}
726			route := *routeNavigationFallback
727			routeNavigationFallback = nil
728			routeNavigationPending = false
729			routeNavigationSelection = nil
730			routeNavigationResumeFallback = false
731			returnNavigationPending = true
732			restoreReturnRoute(route)
733			continue
734		}
735		if returnResumeFallback && resumeNeedsExactAttach(result.err) {
736			attemptRequest.Intent = ports.IntentAttach
737			resumeToken = 0
738			returnResumeFallback = false
739			continue
740		}
741		if returnNavigationPending && !returnResumeFallback && homeRoute != nil {
742			returnRoute = nil
743			returnNavigationPending = false
744			enterHomePicker()
745			continue
746		}
747		if !shouldReconnect(result.err) || resumeToken == 0 || ctx.Err() != nil {
748			if clearErr := reconnect.clear(); clearErr != nil {
749				return errors.Join(result.err, clearErr)
750			}
751			return result.err
752		}
753		r.logger.Warn("reconnecting after attach error", "err", result.err, "backoff", backoff)
754		if drawErr := reconnect.drawStage(reconnectStageSSH); drawErr != nil {
755			return errors.Join(result.err, drawErr)
756		}
757		slept, sleepErr := reconnect.sleep(ctx, r.clock, backoff)
758		if sleepErr != nil {
759			return errors.Join(result.err, sleepErr)
760		}
761		if !slept {
762			if clearErr := reconnect.clear(); clearErr != nil {
763				return errors.Join(ctx.Err(), clearErr)
764			}
765			return ctx.Err()
766		}
767		backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max)
768		attemptRequest.Intent = ports.IntentResume
769	}
770}
771
772// reconnectUI owns reconnect presentation while Runner owns the lifecycle.
773// Its methods run only on Runner's goroutine, preserving the terminal's
774// single-writer rule.
775type reconnectUI struct {
776	term       ports.Terminal
777	remote     bool
778	rawEntered *bool
779	showing    bool
780	stage      reconnectStage
781}
782
783func syncReconnectRemote(reconnect *reconnectUI, remote bool) bool {
784	if reconnect != nil {
785		reconnect.remote = remote
786	}
787	return remote
788}
789
790func (u *reconnectUI) redraw(size domain.Size) error {
791	if !*u.rawEntered || !u.remote {
792		return nil
793	}
794	_, err := drawReconnectToastStage(u.term.Out(), size, u.stage)
795	if err != nil {
796		return fmt.Errorf("drawing reconnect toast: %w", err)
797	}
798	if err := u.term.Flush(); err != nil {
799		return fmt.Errorf("flushing reconnect toast: %w", err)
800	}
801	u.showing = true
802	return nil
803}
804
805func (u *reconnectUI) drawStage(stage reconnectStage) error {
806	u.stage = stage
807	if u.showing && u.remote {
808		size, err := u.term.Size()
809		if err != nil {
810			return fmt.Errorf("reading terminal size for reconnect toast: %w", err)
811		}
812		return u.redraw(size)
813	}
814	return u.draw()
815}
816
817func (u *reconnectUI) draw() error {
818	if !*u.rawEntered || u.showing {
819		return nil
820	}
821	if u.remote {
822		size, err := u.term.Size()
823		if err != nil {
824			return fmt.Errorf("reading terminal size for reconnect toast: %w", err)
825		}
826		return u.redraw(size)
827	}
828	if _, err := u.term.Out().Write([]byte(statusReconnect)); err != nil {
829		return fmt.Errorf("writing reconnect status: %w", err)
830	}
831	if err := u.term.Flush(); err != nil {
832		return fmt.Errorf("flushing reconnect status: %w", err)
833	}
834	u.showing = true
835	return nil
836}
837
838func (u *reconnectUI) clear() error {
839	if !u.showing {
840		return nil
841	}
842	if !u.remote {
843		if _, err := u.term.Out().Write([]byte(statusClear)); err != nil {
844			return fmt.Errorf("clearing reconnect status: %w", err)
845		}
846		if err := u.term.Flush(); err != nil {
847			return fmt.Errorf("flushing reconnect status clear: %w", err)
848		}
849	}
850	u.showing = false
851	return nil
852}
853
854func (u *reconnectUI) sleep(ctx context.Context, clk ports.Clock, d time.Duration) (bool, error) {
855	if u.remote && u.showing {
856		return reconnectSleepWithResize(ctx, clk, d, u.term.ResizeEvents(), u.redraw)
857	}
858	return reconnectSleep(ctx, clk, d), nil
859}
860
861func sleepReconnect(ctx context.Context, clk ports.Clock, d time.Duration) bool {
862	t := clk.NewTimer(d)
863	defer t.Stop()
864	select {
865	case <-t.C():
866		return true
867	case <-ctx.Done():
868		return false
869	}
870}
871
872func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time.Duration, resizeEvents <-chan domain.Size, onResize func(domain.Size) error) (bool, error) {
873	t := clk.NewTimer(d)
874	defer t.Stop()
875	for {
876		select {
877		case <-t.C():
878			return true, nil
879		case <-ctx.Done():
880			return false, nil
881		case size, ok := <-resizeEvents:
882			if !ok {
883				resizeEvents = nil
884				continue
885			}
886			if onResize != nil {
887				if err := onResize(size); err != nil {
888					return false, err
889				}
890			}
891		}
892	}
893}
894
895type attachAttempt struct {
896	runner                   *Runner
897	dialer                   ports.Dialer
898	connection               *SessionConnection
899	remote                   bool
900	transport                ports.Transport
901	handshakeCtx             context.Context
902	handshakeTimedOut        <-chan struct{}
903	finishHandshake          func()
904	stopHandshakeTransport   func()
905	request                  AttachRequest
906	resumeToken              uint64
907	routeNavigationSelection *routeNavigationSelection
908	clientID                 [16]byte
909	milestones               *milestones
910	themeState               *terminalThemeState
911	enterRaw                 func() error
912	reconnect                *reconnectUI
913	linkEvents               <-chan ports.LinkEvent
914	rememberRemoteHost       func()
915	terminalInput            func() *terminalInputPump
916}
917
918type attachResult struct {
919	resumeToken       uint64
920	sessionName       string
921	committedIdentity *ports.CommittedRouteIdentity
922	routePosition     *ports.RoutePosition
923	welcomed          bool
924	transportClosed   bool
925	target            *ports.AttachTarget
926	action            ports.NavigationAction
927	routeAction       *ports.RouteNavigationAction
928	err               error
929}
930
931// AttachTargetError requests that the composition root replace the current
932// transport with the selected endpoint. The daemon never opens that endpoint.
933type AttachTargetError struct {
934	Target ports.AttachTarget
935}
936
937func (e *AttachTargetError) Error() string {
938	if e == nil {
939		return "client: attach target handoff"
940	}
941	return fmt.Sprintf("client: attach target %q/%q", e.Target.Endpoint, e.Target.Session)
942}
943
944// paletteSlot validates the scanner's signed slot before narrowing it for the
945// wire accumulator. Keep this boundary even though the in-tree Scanner also
946// validates slots: callbacks are an interface boundary.
947func paletteSlot(slot int) (uint8, bool) {
948	if slot < 0 || slot > 15 {
949		return 0, false
950	}
951	return uint8(slot), true
952}
953
954// DetectTrueColor reports whether TERM/COLORTERM advertise direct color support.
955func DetectTrueColor(termEnv, colorTerm string) bool {
956	return ports.DetectTrueColor(termEnv, colorTerm)
957}
958
959func requestedOutputWindow(transport ports.Transport) uint8 {
960	if _, ok := transport.(ports.DatagramTransport); ok {
961		return 1
962	}
963	return 8
964}
965
966func (a *attachAttempt) run(ctx context.Context) attachResult {
967	transport := a.transport
968	if a.connection != nil {
969		transport = a.connection.Transport()
970	}
971	request := a.request
972	term := a.runner.term
973	clk := a.runner.clock
974	intent := request.Intent
975	name := request.SessionName
976	resumeToken := a.resumeToken
977	clientID := a.clientID
978	ms := a.milestones
979	themeState := a.themeState
980	enterRaw := a.enterRaw
981	reconnect := a.reconnect
982	linkEvents := a.linkEvents
983	remote := a.remote || a.request.Remote
984	clipboard := a.runner.clipboard
985	log := a.runner.logger
986	observer := a.runner.runtimeObserver
987	handshakeCtx := a.handshakeCtx
988	handshakeTimedOut := a.handshakeTimedOut
989	finishHandshake := a.finishHandshake
990	stopHandshakeTransport := a.stopHandshakeTransport
991	if handshakeCtx == nil {
992		handshakeCtx, handshakeTimedOut, finishHandshake = newHandshakeContext(ctx, clk)
993		stopHandshakeTransport = watchHandshakeTransport(handshakeCtx, transport)
994	}
995	handshakeFinished := false
996	endHandshake := func() {
997		if handshakeFinished {
998			return
999		}
1000		handshakeFinished = true
1001		if stopHandshakeTransport != nil {
1002			stopHandshakeTransport()
1003		}
1004		if finishHandshake != nil {
1005			finishHandshake()
1006		}
1007	}
1008	defer endHandshake()
1009	handshakeFailure := func(stage string, err error) attachResult {
1010		if handshakeTimedOut != nil {
1011			err = handshakeContextError(ctx, handshakeTimedOut, err)
1012		}
1013		return attachResult{transportClosed: handshakeCtx.Err() != nil, err: fmt.Errorf("vev: %s: %w", stage, err)}
1014	}
1015	sendHandshake := func(operation func() error) error {
1016		err := boundedHandshakeOperation(handshakeCtx, transport, operation)
1017		if err != nil {
1018			return handshakeContextError(ctx, handshakeTimedOut, err)
1019		}
1020		return nil
1021	}
1022	// 1. Handshake: send Hello with our size and TERM.
1023	size, err := term.Size()
1024	if err != nil {
1025		return attachResult{err: fmt.Errorf("vev: reading terminal size: %w", err)}
1026	}
1027	cwd, err := os.Getwd()
1028	if err != nil {
1029		cwd = ""
1030	}
1031	termEnv := os.Getenv("TERM")
1032	colorTerm := os.Getenv("COLORTERM")
1033	trueColor := DetectTrueColor(termEnv, colorTerm)
1034	themeState.setTrueColor(trueColor)
1035	exactTarget := request.ExactTarget
1036	if exactTarget == nil && request.RemoteTarget != nil {
1037		exactTarget = &ports.ExactSessionTarget{
1038			LifecycleID: request.RemoteTarget.LifecycleID,
1039			SessionName: request.RemoteTarget.SessionName,
1040		}
1041	}
1042	hello := ports.Hello{
1043		Version:                ports.ProtocolVersion,
1044		Intent:                 intent,
1045		ClientID:               clientID,
1046		ResumeToken:            resumeToken,
1047		Name:                   name,
1048		Size:                   size,
1049		TermEnv:                termEnv,
1050		Cwd:                    cwd,
1051		TrueColor:              trueColor,
1052		MaxOutputInFlight:      requestedOutputWindow(transport),
1053		Env:                    os.Environ(),
1054		RemoteTarget:           request.RemoteTarget,
1055		ExactTarget:            exactTarget,
1056		PreferredTabID:         request.PreferredTabID,
1057		RemoteOrigin:           request.RemoteOrigin,
1058		EnvironmentPolicy:      request.EnvironmentPolicy,
1059		NavigationCapabilities: request.NavigationCapabilities,
1060		StartupOverlay:         request.StartupOverlay,
1061	}
1062	if err := sendHandshake(func() error {
1063		return transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)})
1064	}); err != nil {
1065		return handshakeFailure("sending hello", err)
1066	}
1067	ms.helloSent = true
1068
1069	// 2. Await Welcome or a typed rejection.
1070	var reply ports.Frame
1071	if err := boundedHandshakeOperation(handshakeCtx, transport, func() error {
1072		var recvErr error
1073		reply, recvErr = transport.Recv()
1074		return recvErr
1075	}); err != nil {
1076		return handshakeFailure("awaiting welcome", err)
1077	}
1078	var committedIdentity *ports.CommittedRouteIdentity
1079	var routePosition *ports.RoutePosition
1080	result := func(welcomed bool, err error) attachResult {
1081		return attachResult{
1082			resumeToken:       resumeToken,
1083			sessionName:       name,
1084			committedIdentity: cloneCommittedIdentity(committedIdentity),
1085			routePosition:     cloneRoutePosition(routePosition),
1086			welcomed:          welcomed,
1087			err:               err,
1088		}
1089	}
1090	switch reply.Type {
1091	case ports.MsgWelcome:
1092		welcome, derr := ports.UnmarshalWelcome(reply.Payload)
1093		if derr != nil {
1094			return result(false, fmt.Errorf("vev: decoding welcome: %w", derr))
1095		}
1096		resumeToken = welcome.ResumeToken
1097		name = welcome.SessionName
1098		committedIdentity = cloneCommittedIdentity(welcome.CommittedIdentity)
1099		ms.welcomed = true
1100		log.Debug("welcomed by daemon", "resume_token_present", resumeToken != 0)
1101		if remote {
1102			if remember := a.rememberRemoteHost; remember != nil {
1103				remember()
1104			}
1105		}
1106	case ports.MsgError:
1107		em, derr := ports.UnmarshalErrorMsg(reply.Payload)
1108		if derr != nil {
1109			return result(false, fmt.Errorf("vev: decoding error reply: %w", derr))
1110		}
1111		return result(false, &ProtocolError{Code: em.Code, Text: em.Text})
1112	default:
1113		return result(false, fmt.Errorf("vev: unexpected reply type %d before welcome", reply.Type))
1114	}
1115	if err := handshakeCtx.Err(); err != nil {
1116		return handshakeFailure("validating welcome", err)
1117	}
1118	welcomedResult := func(err error) attachResult { return result(true, err) }
1119	if committedIdentity != nil && a.runner.ledger != nil {
1120		if request.ExactTarget != nil && *request.ExactTarget != committedIdentity.Target {
1121			return welcomedResult(errRouteTargetChanged)
1122		}
1123		if request.RemoteTarget != nil && (request.RemoteTarget.LifecycleID != committedIdentity.Target.LifecycleID || request.RemoteTarget.SessionName != committedIdentity.Target.SessionName) {
1124			return welcomedResult(errRouteTargetChanged)
1125		}
1126		candidate := routeCandidateForAttach(request, *committedIdentity, a.dialer, resumeToken)
1127		var commitErr error
1128		if a.routeNavigationSelection != nil {
1129			selection := *a.routeNavigationSelection
1130			commitErr = a.runner.ledger.commitTransition(selection.selected.identity, selection.snapshotGeneration, selection.active, candidate)
1131		} else {
1132			_, commitErr = a.runner.ledger.commitAttach(candidate)
1133		}
1134		if commitErr != nil {
1135			return welcomedResult(fmt.Errorf("vev: committing route identity: %w", commitErr))
1136		}
1137		snapshotPayload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot())
1138		if err != nil {
1139			return welcomedResult(fmt.Errorf("vev: encoding route snapshot: %w", err))
1140		}
1141		attentionPayload, err := ports.MarshalRouteAttentionSubscription(a.runner.ledger.attentionSubscription())
1142		if err != nil {
1143			return welcomedResult(fmt.Errorf("vev: encoding route attention subscription: %w", err))
1144		}
1145		if err := sendHandshake(func() error {
1146			if err := transport.Send(ports.Frame{Type: ports.MsgRecentRouteSnapshot, Payload: snapshotPayload}); err != nil {
1147				return err
1148			}
1149			return transport.Send(ports.Frame{Type: ports.MsgRouteAttentionSubscription, Payload: attentionPayload})
1150		}); err != nil {
1151			return welcomedResult(fmt.Errorf("vev: publishing route snapshot: %w", err))
1152		}
1153	}
1154	if a.runner.routeFailure != nil {
1155		payload, err := ports.MarshalRouteNavigationFailure(*a.runner.routeFailure)
1156		if err != nil {
1157			return welcomedResult(fmt.Errorf("vev: encoding route navigation failure: %w", err))
1158		}
1159		if err := sendHandshake(func() error {
1160			return transport.Send(ports.Frame{Type: ports.MsgRouteNavigationFailure, Payload: payload})
1161		}); err != nil {
1162			return welcomedResult(fmt.Errorf("vev: publishing route navigation failure: %w", err))
1163		}
1164		a.runner.routeFailure = nil
1165	}
1166
1167	// 3. Enter raw mode after Welcome; Run owns restoration.
1168	if err := enterRaw(); err != nil {
1169		return welcomedResult(err)
1170	}
1171
1172	// 4. Derive a cancellable context so the pumps always unwind when the
1173	// loop returns. The attach loop is the sole terminal writer and palette
1174	// coordinator owner.
1175	loopCtx, cancel := context.WithCancel(ctx)
1176	defer cancel()
1177	input := (*terminalInputPump)(nil)
1178	if a.terminalInput != nil {
1179		input = a.terminalInput()
1180	}
1181	ownedInput := input == nil
1182	if ownedInput {
1183		input = newTerminalInputPump(term.In())
1184		input.start()
1185		defer input.stop()
1186	}
1187	inputConsumer := input.claim()
1188	claimActive := true
1189	revokeInputClaim := func() {
1190		if claimActive {
1191			input.revoke(inputConsumer)
1192			claimActive = false
1193		}
1194	}
1195	// Register revocation before any palette publication or terminal I/O: both
1196	// can fail before the stdin scanner is started, and a reconnect must still
1197	// be able to claim this lifecycle-owned reader.
1198	defer revokeInputClaim()
1199
1200	sendCh := make(chan ports.Frame, sendQueueDepth)
1201	ackQueue := newCumulativeAckQueue()
1202	sendErrCh := make(chan error, 1)
1203	senderStarted := false
1204	paletteEvents := make(chan paletteGenerationEvent, 32)
1205	var activeGeneration atomic.Uint64
1206	coordinator := newPaletteGenerationCoordinator()
1207	type paletteTimer struct {
1208		timer  ports.Timer
1209		cancel chan struct{}
1210	}
1211	drainTimers := map[paletteGenerationID]paletteTimer{}
1212	completionTimers := map[paletteGenerationID]paletteTimer{}
1213	cancelTimer := func(timers map[paletteGenerationID]paletteTimer, id paletteGenerationID) {
1214		if timer, ok := timers[id]; ok {
1215			timer.timer.Stop()
1216			close(timer.cancel)
1217			delete(timers, id)
1218		}
1219	}
1220	publish := func(snapshot ports.Theme) error {
1221		frame := ports.Frame{Type: ports.MsgTheme, Payload: ports.MarshalTheme(snapshot)}
1222		// The initial cleared snapshot is sent before the sender exists. This
1223		// keeps the generation publication ordered before the first query and
1224		// avoids leaving a queued Theme behind an immediate detach.
1225		if !senderStarted {
1226			if err := sendHandshake(func() error { return transport.Send(frame) }); err != nil {
1227				return fmt.Errorf("sending initial theme: %w", err)
1228			}
1229			return nil
1230		}
1231		select {
1232		case sendCh <- frame:
1233			return nil
1234		case <-loopCtx.Done():
1235			return context.Canceled
1236		}
1237	}
1238	processPaletteActions := func(actions []paletteGenerationAction) error {
1239		for _, action := range actions {
1240			switch action.kind {
1241			case paletteActionPublishCleared:
1242				if err := publish(action.theme); err != nil {
1243					return fmt.Errorf("publishing theme: %w", err)
1244				}
1245			case paletteActionPublishFinal:
1246				themeState.update(func(current *ports.Theme) { *current = action.theme })
1247				if err := publish(action.theme); err != nil {
1248					return fmt.Errorf("publishing theme: %w", err)
1249				}
1250			case paletteActionWriteDrain, paletteActionWriteBatch:
1251				if _, err := term.Out().Write([]byte(action.bytes)); err != nil {
1252					return fmt.Errorf("writing palette query: %w", err)
1253				}
1254				if err := term.Flush(); err != nil {
1255					return fmt.Errorf("flushing palette query: %w", err)
1256				}
1257			case paletteActionArmDrainDeadline, paletteActionArmCompletionDeadline:
1258				timer := clk.NewTimer(action.deadline)
1259				entry := paletteTimer{timer: timer, cancel: make(chan struct{})}
1260				timers := drainTimers
1261				kind := paletteEventDrainDeadline
1262				if action.kind == paletteActionArmCompletionDeadline {
1263					timers = completionTimers
1264					kind = paletteEventCompletionDeadline
1265				}
1266				timers[action.id] = entry
1267				go func(id paletteGenerationID, eventKind paletteGenerationEventKind, timer paletteTimer) {
1268					select {
1269					case <-timer.timer.C():
1270						select {
1271						case paletteEvents <- paletteGenerationEvent{id: id, kind: eventKind}:
1272						case <-loopCtx.Done():
1273						}
1274					case <-timer.cancel:
1275					case <-loopCtx.Done():
1276					}
1277				}(action.id, kind, entry)
1278			case paletteActionCancelDrainDeadline:
1279				cancelTimer(drainTimers, action.id)
1280			case paletteActionCancelCompletionDeadline:
1281				cancelTimer(completionTimers, action.id)
1282			}
1283		}
1284		return nil
1285	}
1286	defer func() {
1287		for id := range drainTimers {
1288			cancelTimer(drainTimers, id)
1289		}
1290		for id := range completionTimers {
1291			cancelTimer(completionTimers, id)
1292		}
1293	}()
1294
1295	// Initial acquisition is a generation too. A reconnect publishes its one
1296	// cleared snapshot from retained definitive colors, never a restore frame.
1297	retained := themeState.snapshot()
1298	initialActions := coordinator.start(retained, false)
1299	activeGeneration.Store(uint64(coordinator.current.id))
1300	if err := processPaletteActions(initialActions); err != nil {
1301		return welcomedResult(err)
1302	}
1303
1304	var clip ports.ClipboardReader
1305	if remote {
1306		clip = clipboard
1307	}
1308	awaitingReconnectReset := false
1309	outputState := outputApplyState{}
1310	outputResetRequested := false
1311	if reconnect.showing {
1312		if reconnect.remote {
1313			// Before the asynchronous sender starts, preserve transport ordering by
1314			// requesting the reset synchronously after the initial Theme publication.
1315			resetErr := sendHandshake(func() error {
1316				payload, err := ports.MarshalResize(ports.Resize{Size: size})
1317				if err != nil {
1318					return err
1319				}
1320				return transport.Send(ports.Frame{Type: ports.MsgResize, Payload: payload})
1321			})
1322			if resetErr != nil {
1323				resetErr = handshakeContextError(ctx, handshakeTimedOut, resetErr)
1324				resetResult := welcomedResult(fmt.Errorf("vev: requesting reconnect reconciliation: %w", resetErr))
1325				resetResult.transportClosed = handshakeCtx.Err() != nil
1326				return resetResult
1327			}
1328			awaitingReconnectReset = true
1329		}
1330		if err := reconnect.clear(); err != nil {
1331			return welcomedResult(err)
1332		}
1333	}
1334	// Welcome and every synchronous initial publication have committed. The
1335	// handshake deadline must not own the long-lived runtime transport: a quiet
1336	// or unchanged screen may legitimately produce no initial Output frame.
1337	endHandshake()
1338	senderStarted = true
1339	go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log)
1340	stdinDone := make(chan struct{})
1341	go func() {
1342		defer close(stdinDone)
1343		(&stdinPump{ctx: loopCtx, cancel: cancel, input: input, consumer: inputConsumer, out: sendCh, clock: clk, clipboard: clip, logger: log, paletteEvents: paletteEvents, activeGeneration: &activeGeneration}).run()
1344	}()
1345	// Scanner cancellation can leave an undecided marker suffix (including a
1346	// standalone Escape) that it must hand back to the lifecycle-owned reader.
1347	// Wait for that handoff before revoking the claim: otherwise a replacement
1348	// scanner could claim the reader between revocation and preservation and
1349	// lose the suffix.
1350	defer func() {
1351		cancel()
1352		<-stdinDone
1353		revokeInputClaim()
1354	}()
1355	go runResize(loopCtx, term.ResizeEvents(), sendCh, log)
1356
1357	// 5. Output/main loop: the only goroutine that touches the terminal.
1358	recvCh := make(chan recvResult, 1)
1359	go runRecv(loopCtx, transport, recvCh, log)
1360
1361	requestReconnectReset := func() error {
1362		if awaitingReconnectReset {
1363			return nil
1364		}
1365		size, serr := term.Size()
1366		if serr != nil {
1367			return fmt.Errorf("reading terminal size for reconnect reconciliation: %w", serr)
1368		}
1369		payload, err := ports.MarshalResize(ports.Resize{Size: size})
1370		if err != nil {
1371			return fmt.Errorf("encoding terminal size for reconnect reconciliation: %w", err)
1372		}
1373		select {
1374		case sendCh <- ports.Frame{Type: ports.MsgResize, Payload: payload}:
1375			awaitingReconnectReset = true
1376			return nil
1377		case <-loopCtx.Done():
1378			return context.Canceled
1379		}
1380	}
1381	publishRouteSnapshot := func() error {
1382		if a.runner.ledger == nil {
1383			return errors.New("vev: route ledger unavailable")
1384		}
1385		snapshotPayload, err := ports.MarshalRecentRouteSnapshot(a.runner.ledger.snapshot())
1386		if err != nil {
1387			return err
1388		}
1389		attentionPayload, err := ports.MarshalRouteAttentionSubscription(a.runner.ledger.attentionSubscription())
1390		if err != nil {
1391			return err
1392		}
1393		for _, frame := range []ports.Frame{
1394			{Type: ports.MsgRecentRouteSnapshot, Payload: snapshotPayload},
1395			{Type: ports.MsgRouteAttentionSubscription, Payload: attentionPayload},
1396		} {
1397			select {
1398			case sendCh <- frame:
1399			case <-loopCtx.Done():
1400				return context.Canceled
1401			}
1402		}
1403		return nil
1404	}
1405	sendRouteFailure := func(action ports.RouteNavigationAction, code ports.RouteFailureCode) error {
1406		payload, err := ports.MarshalRouteNavigationFailure(ports.RouteNavigationFailure{
1407			Key:        action.Key,
1408			Generation: action.Generation,
1409			Code:       code,
1410		})
1411		if err != nil {
1412			return err
1413		}
1414		select {
1415		case sendCh <- ports.Frame{Type: ports.MsgRouteNavigationFailure, Payload: payload}:
1416			return nil
1417		case <-loopCtx.Done():
1418			return context.Canceled
1419		}
1420	}
1421	dismissReconnect := func() error {
1422		if !reconnect.showing {
1423			return nil
1424		}
1425		if err := reconnect.clear(); err != nil {
1426			return err
1427		}
1428		if !reconnect.remote {
1429			return nil
1430		}
1431		return requestReconnectReset()
1432	}
1433	loopCanceledResult := func() attachResult {
1434		// A pump initiated shutdown (stdin EOF/detach) or the parent context
1435		// was cancelled. A queued sender error takes priority.
1436		select {
1437		case serr := <-sendErrCh:
1438			return welcomedResult(fmt.Errorf("vev: sending to daemon: %w", serr))
1439		default:
1440			return welcomedResult(nil)
1441		}
1442	}
1443	for {
1444		select {
1445		case <-loopCtx.Done():
1446			return loopCanceledResult()
1447		case ev, ok := <-linkEvents:
1448			if !ok {
1449				linkEvents = nil
1450				continue
1451			}
1452			if ev.State == ports.LinkStateConnected {
1453				if err := dismissReconnect(); err != nil {
1454					if errors.Is(err, context.Canceled) {
1455						return loopCanceledResult()
1456					}
1457					return welcomedResult(fmt.Errorf("dismissing reconnect toast: %w", err))
1458				}
1459				select {
1460				case sendCh <- ports.Frame{Type: ports.MsgClientNotice, Payload: ports.MarshalClientNotice(ports.ClientNotice{Action: ports.ClientNoticeLinkConnected})}:
1461				case <-loopCtx.Done():
1462					return welcomedResult(nil)
1463				}
1464				continue
1465			}
1466			if ev.State == ports.LinkStateDegraded {
1467				log.Warn("UDP link degraded")
1468				select {
1469				case sendCh <- ports.Frame{Type: ports.MsgClientNotice, Payload: ports.MarshalClientNotice(ports.ClientNotice{Action: ports.ClientNoticeLinkDegraded})}:
1470				case <-loopCtx.Done():
1471					return welcomedResult(nil)
1472				}
1473				continue
1474			}
1475			stage := stageForLinkState(ev.State)
1476			if reconnect.remote && reconnect.showing && reconnect.stage != stage {
1477				reconnect.stage = stage
1478				if err := requestReconnectReset(); err != nil {
1479					if errors.Is(err, context.Canceled) {
1480						return loopCanceledResult()
1481					}
1482					return welcomedResult(fmt.Errorf("requesting reconnect reconciliation: %w", err))
1483				}
1484			} else if err := reconnect.drawStage(stage); err != nil {
1485				return welcomedResult(err)
1486			}
1487			if ev.State == ports.LinkStateOffline {
1488				// Stop waiting for the transport to reach Dead at 60s: exit with
1489				// a retryable error so Run re-dials over ssh with the resume
1490				// token. Run closes this transport on the way out.
1491				return welcomedResult(errLinkOffline)
1492			}
1493		case event := <-paletteEvents:
1494			if event.kind == paletteEventScheme {
1495				retained := themeState.update(func(current *ports.Theme) {
1496					current.SchemeKnown = true
1497					current.Light = event.light
1498				})
1499				actions := coordinator.start(retained, true)
1500				activeGeneration.Store(uint64(coordinator.current.id))
1501				if err := processPaletteActions(actions); err != nil {
1502					return welcomedResult(err)
1503				}
1504				continue
1505			}
1506			if err := processPaletteActions(coordinator.handle(event)); err != nil {
1507				return welcomedResult(err)
1508			}
1509		case r := <-recvCh:
1510			if r.err != nil {
1511				if errors.Is(r.err, io.EOF) {
1512					return welcomedResult(fmt.Errorf("vev: daemon vanished (missing: %v): %w", ms.missing(), r.err))
1513				}
1514				return welcomedResult(fmt.Errorf("vev: receiving from daemon: %w", r.err))
1515			}
1516			switch r.frame.Type {
1517			case ports.MsgOutput:
1518				o, derr := ports.UnmarshalOutput(r.frame.Payload)
1519				if derr != nil {
1520					return welcomedResult(fmt.Errorf("vev: decoding output: %w", derr))
1521				}
1522				nextState, accepted := outputState.next(o)
1523				if !accepted {
1524					// A discarded frame was never written and must never be ACKed.
1525					// Ask the daemon for one authoritative full reset, coalescing
1526					// repeated gaps while that reset is in flight.
1527					if !outputResetRequested {
1528						select {
1529						case sendCh <- ports.Frame{Type: ports.MsgOutputResetRequest, Payload: ports.MarshalOutputResetRequest(ports.OutputResetRequest{})}:
1530							outputResetRequested = true
1531						case <-loopCtx.Done():
1532							return loopCanceledResult()
1533						}
1534					}
1535					continue
1536				}
1537				if _, werr := term.Out().Write(o.Data); werr != nil {
1538					return welcomedResult(fmt.Errorf("vev: writing terminal output: %w", werr))
1539				}
1540				if awaitingReconnectReset && o.Full && o.Base == 0 && o.New != 0 {
1541					awaitingReconnectReset = false
1542				}
1543				if reconnect.showing && reconnect.remote {
1544					size, serr := term.Size()
1545					if serr != nil {
1546						return welcomedResult(fmt.Errorf("vev: reading terminal size for reconnect redraw: %w", serr))
1547					}
1548					if rerr := reconnect.redraw(size); rerr != nil {
1549						return welcomedResult(fmt.Errorf("vev: redrawing reconnect toast: %w", rerr))
1550					}
1551				} else if ferr := term.Flush(); ferr != nil {
1552					return welcomedResult(fmt.Errorf("vev: flushing terminal: %w", ferr))
1553				}
1554				outputState = nextState
1555				if o.New != 0 {
1556					ackQueue.offer(o.Epoch, o.New)
1557					outputResetRequested = false
1558				}
1559				// The terminal boundary is after a successful flush and before ACK.
1560				if observer != nil {
1561					observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeTerminalFlushed, uint64(len(o.Data)), true))
1562				}
1563				if !ms.firstOutput {
1564					ms.firstOutput = true
1565					log.Debug("received first output")
1566					endHandshake()
1567				}
1568			case ports.MsgAttachTarget:
1569				target, derr := ports.UnmarshalAttachTarget(r.frame.Payload)
1570				if derr != nil {
1571					return welcomedResult(fmt.Errorf("vev: decoding attach target: %w", derr))
1572				}
1573				handoff := welcomedResult(nil)
1574				handoff.target = &target
1575				return handoff
1576			case ports.MsgNavigationAction:
1577				action, derr := ports.UnmarshalNavigationAction(r.frame.Payload)
1578				if derr != nil {
1579					return welcomedResult(fmt.Errorf("vev: decoding navigation action: %w", derr))
1580				}
1581				navigation := welcomedResult(nil)
1582				navigation.action = action
1583				return navigation
1584			case ports.MsgNavigateRecentRoute:
1585				action, derr := ports.UnmarshalRouteNavigationAction(r.frame.Payload)
1586				if derr != nil {
1587					return welcomedResult(fmt.Errorf("vev: decoding route navigation action: %w", derr))
1588				}
1589				if a.runner.ledger == nil {
1590					if ferr := sendRouteFailure(action, ports.RouteFailureUnavailable); ferr != nil {
1591						return welcomedResult(ferr)
1592					}
1593					continue
1594				}
1595				_, valid, noOp := a.runner.ledger.navigationRecord(action)
1596				if !valid {
1597					if ferr := sendRouteFailure(action, ports.RouteFailureStaleSelection); ferr != nil {
1598						return welcomedResult(ferr)
1599					}
1600					continue
1601				}
1602				if noOp {
1603					continue
1604				}
1605				navigation := welcomedResult(nil)
1606				navigation.routeAction = &action
1607				return navigation
1608			case ports.MsgCommittedRouteIdentity:
1609				identity, derr := ports.UnmarshalCommittedRouteIdentity(r.frame.Payload)
1610				if derr != nil {
1611					return welcomedResult(fmt.Errorf("vev: decoding committed route identity: %w", derr))
1612				}
1613				name = identity.Target.SessionName
1614				committedIdentity = cloneCommittedIdentity(&identity)
1615				if a.runner.ledger == nil {
1616					return welcomedResult(errors.New("vev: route ledger unavailable"))
1617				}
1618				if _, derr := a.runner.ledger.commitCommittedIdentity(identity); derr != nil {
1619					return welcomedResult(fmt.Errorf("vev: committing daemon-local route identity: %w", derr))
1620				}
1621				if derr := publishRouteSnapshot(); derr != nil {
1622					return welcomedResult(fmt.Errorf("vev: publishing daemon-local route snapshot: %w", derr))
1623				}
1624			case ports.MsgRoutePosition:
1625				position, derr := ports.UnmarshalRoutePosition(r.frame.Payload)
1626				if derr != nil {
1627					return welcomedResult(fmt.Errorf("vev: decoding route position: %w", derr))
1628				}
1629				if a.runner.ledger == nil {
1630					return welcomedResult(errors.New("vev: route ledger unavailable"))
1631				}
1632				if derr := a.runner.ledger.updateRoutePosition(position); derr != nil {
1633					return welcomedResult(fmt.Errorf("vev: remembering route position: %w", derr))
1634				}
1635				routePosition = cloneRoutePosition(&position)
1636			case ports.MsgRouteNavigationFailure:
1637				failure, derr := ports.UnmarshalRouteNavigationFailure(r.frame.Payload)
1638				if derr != nil {
1639					return welcomedResult(fmt.Errorf("vev: decoding route navigation failure: %w", derr))
1640				}
1641				log.Warn("route navigation rejected", "key", failure.Key, "generation", failure.Generation, "code", failure.Code)
1642			case ports.MsgDetached:
1643				d, derr := ports.UnmarshalDetached(r.frame.Payload)
1644				if derr != nil {
1645					return welcomedResult(fmt.Errorf("vev: decoding detached: %w", derr))
1646				}
1647				ms.detached = true
1648				if d.Reason == ports.ReasonDetach {
1649					log.Info("detached cleanly", "reason", d.Reason)
1650				} else {
1651					log.Warn("detached by daemon", "reason", d.Reason)
1652				}
1653				return welcomedResult(detachedResult(d.Reason))
1654			case ports.MsgPong:
1655				// Liveness reply; nothing to do.
1656			default:
1657				// Unknown/out-of-band message types are ignored so a newer
1658				// daemon can add server->client messages without breaking us.
1659			}
1660		}
1661	}
1662}
1663
1664// recvResult carries one framed message (or a read error) from the receive
1665// pump to the main loop.
1666type recvResult struct {
1667	frame ports.Frame
1668	err   error
1669}
1670
1671// runRecv reads frames until an error, forwarding each to the main loop.
1672// It exits on the first error or when the loop context is cancelled.
1673func runRecv(ctx context.Context, transport ports.Transport, out chan<- recvResult, log *slog.Logger) {
1674	defer log.Debug("receive pump exited")
1675	for {
1676		f, err := transport.Recv()
1677		select {
1678		case out <- recvResult{frame: f, err: err}:
1679		case <-ctx.Done():
1680			return
1681		}
1682		if err != nil {
1683			return
1684		}
1685	}
1686}
1687
1688// runSender is the sole caller of transport.Send: serialising the input and
1689// resize pumps through one goroutine keeps the transport's single-writer
1690// contract intact. A send failure cancels the loop and is surfaced once.
1691type cumulativeAckQueue struct {
1692	mu          sync.Mutex
1693	latestEpoch uint64
1694	latestState uint64
1695	wake        chan struct{}
1696}
1697
1698func newCumulativeAckQueue() *cumulativeAckQueue {
1699	return &cumulativeAckQueue{wake: make(chan struct{}, 1)}
1700}
1701
1702func (q *cumulativeAckQueue) offer(epoch, state uint64) {
1703	if epoch == 0 || state == 0 {
1704		return
1705	}
1706	q.mu.Lock()
1707	if epoch > q.latestEpoch || (epoch == q.latestEpoch && state > q.latestState) {
1708		q.latestEpoch, q.latestState = epoch, state
1709	}
1710	q.mu.Unlock()
1711	select {
1712	case q.wake <- struct{}{}:
1713	default:
1714	}
1715}
1716
1717func (q *cumulativeAckQueue) take() (uint64, uint64) {
1718	q.mu.Lock()
1719	defer q.mu.Unlock()
1720	epoch, state := q.latestEpoch, q.latestState
1721	q.latestEpoch, q.latestState = 0, 0
1722	return epoch, state
1723}
1724
1725func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.Transport, in <-chan ports.Frame, acks *cumulativeAckQueue, errCh chan<- error, log *slog.Logger) {
1726	defer log.Debug("sender pump exited")
1727	send := func(f ports.Frame) bool {
1728		if err := transport.Send(f); err != nil {
1729			select {
1730			case errCh <- err:
1731			default:
1732			}
1733			cancel()
1734			return false
1735		}
1736		return true
1737	}
1738	sendAck := func() bool {
1739		epoch, state := acks.take()
1740		if state == 0 {
1741			return true
1742		}
1743		payload, err := ports.MarshalAck(ports.Ack{Epoch: epoch, State: state})
1744		if err != nil {
1745			select {
1746			case errCh <- fmt.Errorf("encoding output ACK: %w", err):
1747			default:
1748			}
1749			cancel()
1750			return false
1751		}
1752		return send(ports.Frame{Type: ports.MsgAck, Payload: payload})
1753	}
1754	for {
1755		select {
1756		case <-acks.wake:
1757			if !sendAck() {
1758				return
1759			}
1760			continue
1761		default:
1762		}
1763		select {
1764		case <-acks.wake:
1765			if !sendAck() {
1766				return
1767			}
1768		case f := <-in:
1769			select {
1770			case <-ctx.Done():
1771				return
1772			default:
1773			}
1774			if !send(f) {
1775				return
1776			}
1777		case <-ctx.Done():
1778			return
1779		}
1780	}
1781}
1782
1783// terminalReadResult is copied by terminalInputPump before the next Read so
1784// terminal input remains owned by one lifecycle reader across reconnects.
1785type terminalReadResult struct {
1786	data []byte
1787	err  error
1788}
1789
1790// terminalInputPump is the sole consumer of the caller-owned terminal reader.
1791// It is started once after raw mode is entered and is reused by each attach
1792// scanner. A bare io.Reader cannot be interrupted, so a Read may stay blocked
1793// after Run exits; suspend drops any result received while inactive and never
1794// closes caller-owned stdin. Reconnects and later runs never add another reader.
1795type terminalInputPump struct {
1796	in   io.Reader
1797	done chan struct{}
1798
1799	mu sync.Mutex
1800	// residual holds scanner bytes that were read but not delivered when an
1801	// attempt is cancelled. It is consumed before the next terminal Read so a
1802	// replacement scanner neither loses a partial marker nor replays bytes
1803	// whose callbacks already ran.
1804	residual           []byte
1805	pending            *terminalReadResult
1806	consumer           uint64
1807	delivering         uint64
1808	deliveringResidual bool
1809	nextID             uint64
1810	activation         uint64
1811	closed             bool
1812	active             bool
1813	ready              chan struct{}
1814	space              chan struct{}
1815	state              chan struct{}
1816	exited             chan struct{}
1817	afterRevoke        func() // test synchronization hook
1818	startMu            sync.Once
1819	stopMu             sync.Once
1820}
1821
1822func newTerminalInputPump(in io.Reader) *terminalInputPump {
1823	space := make(chan struct{}, 1)
1824	space <- struct{}{}
1825	return &terminalInputPump{
1826		in:         in,
1827		done:       make(chan struct{}),
1828		activation: 1,
1829		active:     true,
1830		ready:      make(chan struct{}, 1),
1831		space:      space,
1832		state:      make(chan struct{}, 1),
1833		exited:     make(chan struct{}),
1834	}
1835}
1836
1837// claim grants one attach scanner exclusive permission to dequeue terminal
1838// input. The reader itself remains lifecycle-owned across reconnects.
1839func (p *terminalInputPump) claim() uint64 {
1840	p.mu.Lock()
1841	defer p.mu.Unlock()
1842	if p.consumer != 0 {
1843		panic("terminal input consumer already claimed")
1844	}
1845	p.nextID++
1846	p.consumer = p.nextID
1847	return p.consumer
1848}
1849
1850// revoke invalidates an attempt before its replacement is allowed to claim
1851// input. Pending bytes are deliberately retained for that replacement.
1852func (p *terminalInputPump) revoke(consumer uint64) {
1853	p.mu.Lock()
1854	if p.consumer == consumer {
1855		p.consumer = 0
1856	}
1857	// An unacknowledged read was never delivered by this scanner. Leave it
1858	// pending and make it available to its replacement.
1859	if p.delivering == consumer {
1860		p.delivering = 0
1861		p.deliveringResidual = false
1862	}
1863	p.mu.Unlock()
1864	p.signalReady()
1865	if p.afterRevoke != nil {
1866		p.afterRevoke()
1867	}
1868}
1869
1870func (p *terminalInputPump) signalReady() {
1871	select {
1872	case p.ready <- struct{}{}:
1873	default:
1874	}
1875}
1876
1877// enqueue retains at most one completed terminal read. The reader blocks
1878// before publishing another result, keeping handoff buffering bounded.
1879func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
1880	select {
1881	case <-p.done:
1882		return false
1883	case <-p.space:
1884	}
1885	p.mu.Lock()
1886	if p.closed || !p.active || p.activation != activation {
1887		p.mu.Unlock()
1888		p.space <- struct{}{}
1889		return false
1890	}
1891	p.pending = &result
1892	p.mu.Unlock()
1893	p.signalReady()
1894	return true
1895}
1896
1897// take leases a ready result only to the current, non-cancelled consumer.
1898// The raw read remains pending until ack, so revocation or cancellation before
1899// scanner delivery leaves the exact bytes available to the next attempt.
1900func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
1901	p.mu.Lock()
1902	if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
1903		ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
1904		p.mu.Unlock()
1905		if ready {
1906			p.signalReady()
1907		}
1908		return terminalReadResult{}, false
1909	}
1910	if len(p.residual) != 0 {
1911		result := terminalReadResult{data: append([]byte(nil), p.residual...)}
1912		p.delivering = consumer
1913		p.deliveringResidual = true
1914		p.mu.Unlock()
1915		return result, true
1916	}
1917	if p.pending == nil {
1918		p.mu.Unlock()
1919		return terminalReadResult{}, false
1920	}
1921	result := *p.pending
1922	p.delivering = consumer
1923	p.deliveringResidual = false
1924	p.mu.Unlock()
1925	return result, true
1926}
1927
1928// preserveResidual records the undecided marker prefix after callbacks for the
1929// rest of its read have completed. It is intentionally bounded by the marker
1930// scanner's fixed response prefixes and is replayed before later terminal
1931// reads on a replacement attempt.
1932func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
1933	if len(data) == 0 {
1934		return
1935	}
1936	p.mu.Lock()
1937	defer p.mu.Unlock()
1938	if p.consumer != consumer {
1939		return
1940	}
1941	p.residual = append(p.residual[:0], data...)
1942}
1943
1944// ack commits scanner delivery of a leased read and lets the lifecycle reader
1945// accept the next result. Only the consumer that holds the lease can ack it.
1946func (p *terminalInputPump) ack(consumer uint64) {
1947	p.mu.Lock()
1948	if p.consumer != consumer || p.delivering != consumer {
1949		p.mu.Unlock()
1950		return
1951	}
1952	wasResidual := p.deliveringResidual
1953	if wasResidual {
1954		p.residual = nil
1955	} else {
1956		p.pending = nil
1957	}
1958	p.delivering = 0
1959	p.deliveringResidual = false
1960	more := len(p.residual) != 0 || p.pending != nil
1961	p.mu.Unlock()
1962	if wasResidual {
1963		if more {
1964			p.signalReady()
1965		}
1966		return
1967	}
1968	p.space <- struct{}{}
1969}
1970
1971func (p *terminalInputPump) finish() {
1972	p.mu.Lock()
1973	p.closed = true
1974	p.mu.Unlock()
1975	p.signalReady()
1976	p.signalState()
1977}
1978
1979func (p *terminalInputPump) signalState() {
1980	select {
1981	case p.state <- struct{}{}:
1982	default:
1983	}
1984}
1985
1986func (p *terminalInputPump) waitForActive() (uint64, bool) {
1987	for {
1988		p.mu.Lock()
1989		active, closed, activation := p.active, p.closed, p.activation
1990		p.mu.Unlock()
1991		if closed {
1992			return 0, false
1993		}
1994		if active {
1995			return activation, true
1996		}
1997		select {
1998		case <-p.done:
1999			return 0, false
2000		case <-p.state:
2001		}
2002	}
2003}
2004
2005func (p *terminalInputPump) start() {
2006	p.startMu.Do(func() {
2007		go func() {
2008			defer close(p.exited)
2009			buf := make([]byte, stdinBufSize)
2010			for {
2011				activation, ok := p.waitForActive()
2012				if !ok {
2013					return
2014				}
2015				n, err := p.in.Read(buf)
2016				result := terminalReadResult{err: err}
2017				if n > 0 {
2018					result.data = append([]byte(nil), buf[:n]...)
2019				}
2020				if !p.enqueue(result, activation) {
2021					if p.isClosed() {
2022						return
2023					}
2024					continue
2025				}
2026				if err != nil {
2027					p.finish()
2028					return
2029				}
2030			}
2031		}()
2032	})
2033}
2034
2035func (p *terminalInputPump) hasExited() bool {
2036	select {
2037	case <-p.exited:
2038		return true
2039	default:
2040		return false
2041	}
2042}
2043
2044func (p *terminalInputPump) isClosed() bool {
2045	p.mu.Lock()
2046	defer p.mu.Unlock()
2047	return p.closed
2048}
2049
2050// suspend drops input completed after an attach ends and parks the lifecycle
2051// reader before its next Read. A later Run resumes this same reader, avoiding
2052// both input replay between runs and a competing read on caller-owned stdin.
2053func (p *terminalInputPump) suspend() {
2054	p.mu.Lock()
2055	p.active = false
2056	discarded := p.pending != nil
2057	p.residual = nil
2058	p.pending = nil
2059	p.delivering = 0
2060	p.deliveringResidual = false
2061	p.mu.Unlock()
2062	if discarded {
2063		select {
2064		case p.space <- struct{}{}:
2065		default:
2066		}
2067	}
2068	p.signalState()
2069	p.signalReady()
2070}
2071
2072func (p *terminalInputPump) resume() {
2073	p.mu.Lock()
2074	if !p.closed && !p.active {
2075		p.activation++
2076		p.active = true
2077	}
2078	p.mu.Unlock()
2079	p.signalState()
2080}
2081
2082func (p *terminalInputPump) stop() {
2083	p.stopMu.Do(func() {
2084		// Mark closure while holding the same mutex that publishes a completed
2085		// Read. A Read which returns after stop therefore cannot publish even
2086		// when both done and space are selectable in enqueue.
2087		p.mu.Lock()
2088		p.closed = true
2089		p.mu.Unlock()
2090		close(p.done)
2091		p.signalReady()
2092		p.signalState()
2093	})
2094}
2095
2096// stdinPump parses terminal color reports into generation-tagged events while
2097// preserving all ordinary terminal input byte-for-byte. It never reads the
2098// terminal directly, publishes a Theme, or writes terminal output;
2099// attachAttempt owns those operations.
2100type stdinPump struct {
2101	ctx                 context.Context
2102	cancel              context.CancelFunc
2103	in                  io.Reader // compatibility for isolated scanner tests
2104	input               *terminalInputPump
2105	consumer            uint64
2106	out                 chan<- ports.Frame
2107	clock               ports.Clock
2108	clipboard           ports.ClipboardReader
2109	logger              *slog.Logger
2110	paletteEvents       chan<- paletteGenerationEvent
2111	activeGeneration    *atomic.Uint64
2112	afterPaletteEvent   func(paletteGenerationEvent) // test synchronization hook
2113	afterInputTake      func()                       // test synchronization hook
2114	afterInputDelivered func()                       // test synchronization hook
2115}
2116
2117func (p *stdinPump) run() {
2118	defer p.logger.Debug("stdin pump exited")
2119	var scanner theme.Scanner
2120	var markers paletteMarkerScanner
2121	var inputSeq uint64
2122	var sendOK atomic.Bool
2123	var undeliveredMu sync.Mutex
2124	var undelivered []byte
2125	sendOK.Store(true)
2126	send := func(frame ports.Frame) bool {
2127		if !sendOK.Load() {
2128			return false
2129		}
2130		select {
2131		case p.out <- frame:
2132			if p.afterInputDelivered != nil {
2133				p.afterInputDelivered()
2134			}
2135			return true
2136		case <-p.ctx.Done():
2137			sendOK.Store(false)
2138			return false
2139		}
2140	}
2141	sendEvent := func(event paletteGenerationEvent) bool {
2142		if !sendOK.Load() {
2143			return false
2144		}
2145		select {
2146		case p.paletteEvents <- event:
2147			if p.afterPaletteEvent != nil {
2148				p.afterPaletteEvent(event)
2149			}
2150			return true
2151		case <-p.ctx.Done():
2152			sendOK.Store(false)
2153			return false
2154		}
2155	}
2156	coalescer := newPasteCoalescer(p.clock, func(data []byte) {
2157		if len(data) == 0 {
2158			return
2159		}
2160		inputSeq++
2161		if !send(ports.Frame{Type: ports.MsgInput, Payload: ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: append([]byte(nil), data...)})}) {
2162			undeliveredMu.Lock()
2163			undelivered = append(undelivered, data...)
2164			undeliveredMu.Unlock()
2165		}
2166	})
2167	defer coalescer.Close()
2168	sink := coalescer.Scan
2169	if p.clipboard != nil {
2170		ci := &clipboardIntercept{
2171			coalescer: coalescer,
2172			reader:    p.clipboard,
2173			log:       p.logger,
2174			sendNotice: func(action uint8) {
2175				send(ports.Frame{Type: ports.MsgClientNotice, Payload: ports.MarshalClientNotice(ports.ClientNotice{Action: action})})
2176			},
2177			sendImage: func(mime string, data []byte) {
2178				inputSeq++
2179				send(ports.Frame{Type: ports.MsgImagePush, Payload: ports.MarshalImagePush(ports.ImagePush{InputSeq: inputSeq, Mime: mime, Data: data})})
2180			},
2181			next: coalescer.Scan,
2182		}
2183		sink = ci.Scan
2184	}
2185	input := p.input
2186	ownedInput := input == nil
2187	if ownedInput {
2188		input = newTerminalInputPump(p.in)
2189		input.start()
2190		defer input.stop()
2191	}
2192	consumer := p.consumer
2193	if consumer == 0 {
2194		consumer = input.claim()
2195		defer input.revoke(consumer)
2196	}
2197	// A marker prefix is ordinary terminal input until it becomes a complete
2198	// DECRQM reply. If cancellation wins while it is withheld, preserve just
2199	// that undecided suffix; the rest of its source read has already been
2200	// delivered and must not be replayed by the next attempt.
2201	defer func() {
2202		if p.ctx.Err() != nil {
2203			// Closing flushes any bracketed-paste prefix or buffer through the
2204			// coalescer's emit callback. Failed ordinary-input deliveries are
2205			// retained ahead of a trailing undecided marker prefix, matching their
2206			// original order within the raw read.
2207			held := coalescer.CloseAndTakeHeld()
2208			undeliveredMu.Lock()
2209			residual := append([]byte(nil), undelivered...)
2210			undeliveredMu.Unlock()
2211			residual = append(residual, held...)
2212			residual = append(residual, markers.takePending()...)
2213			input.preserveResidual(consumer, residual)
2214		}
2215	}()
2216
2217	var markerTimer ports.Timer
2218	var markerTimerC <-chan time.Time
2219	flushMarkerPrefix := func() {
2220		markers.flush(sink)
2221		markerTimer = nil
2222		markerTimerC = nil
2223	}
2224	// Disarm before processing a subsequent read. A deadline already ready at
2225	// that boundary wins, so bytes cannot be retroactively consumed as a
2226	// marker after they were due to be forwarded as ordinary input.
2227	disarmMarkerDeadline := func() {
2228		if markerTimer == nil {
2229			return
2230		}
2231		select {
2232		case <-markerTimerC:
2233			flushMarkerPrefix()
2234			return
2235		default:
2236		}
2237		if !markerTimer.Stop() {
2238			// Stop reports false only after expiry or a prior stop. This timer is
2239			// owned only here, so expiry wins even if delivery to C races this
2240			// select; do not let a late value consume a newly arrived suffix.
2241			flushMarkerPrefix()
2242			return
2243		}
2244		markerTimer = nil
2245		markerTimerC = nil
2246	}
2247	armMarkerDeadline := func() {
2248		disarmMarkerDeadline()
2249		markerTimer = p.clock.NewTimer(paletteMarkerAmbiguityDeadline)
2250		markerTimerC = markerTimer.C()
2251	}
2252	defer disarmMarkerDeadline()
2253
2254	for {
2255		select {
2256		case <-p.ctx.Done():
2257			return
2258		case <-markerTimerC:
2259			flushMarkerPrefix()
2260		case <-input.ready:
2261			result, ok := input.take(p.ctx, consumer)
2262			if !ok {
2263				if p.ctx.Err() != nil {
2264					return
2265				}
2266				continue
2267			}
2268			if p.afterInputTake != nil {
2269				p.afterInputTake()
2270			}
2271			// The lease is not acknowledged until scanner delivery completes.
2272			// This second check closes the cancellation window after take.
2273			if p.ctx.Err() != nil {
2274				return
2275			}
2276			disarmMarkerDeadline()
2277			// All scanner callbacks from this Read must retain this one generation.
2278			// A scheme notification can start its replacement before a later byte in
2279			// the same read is scanned; reloading there would misclassify an old
2280			// completion marker as the replacement's drain response.
2281			readGeneration := paletteGenerationID(p.activeGeneration.Load())
2282			if len(result.data) > 0 {
2283				scanner.Scan(result.data, func(kind int, rgb renderer.RGB) {
2284					kindEvent := paletteEventForeground
2285					if kind == 11 {
2286						kindEvent = paletteEventBackground
2287					}
2288					sendEvent(paletteGenerationEvent{id: readGeneration, kind: kindEvent, rgb: rgb})
2289				}, func(slot int, rgb renderer.RGB) {
2290					// Scanner implementations are independently testable; retain the
2291					// protocol boundary here too so a malformed implementation cannot
2292					// wrap a negative or oversized int into an ANSI palette slot.
2293					slot8, ok := paletteSlot(slot)
2294					if !ok {
2295						return
2296					}
2297					sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventPalette, slot: slot8, rgb: rgb})
2298				}, func(light bool) {
2299					sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventScheme, light: light})
2300				}, func(data []byte) {
2301					markers.scan(data, sink, func() {
2302						sendEvent(paletteGenerationEvent{id: readGeneration, kind: paletteEventMarker})
2303					})
2304				})
2305				if !sendOK.Load() {
2306					// Commit the source read so callbacks accepted before cancellation
2307					// are never replayed. The deferred handoff retains only ordinary
2308					// bytes whose callback could not be delivered, plus any undecided
2309					// trailing marker prefix.
2310					input.ack(consumer)
2311					return
2312				}
2313			}
2314			if markers.hasPendingPrefix() {
2315				armMarkerDeadline()
2316			}
2317			if result.err != nil {
2318				disarmMarkerDeadline()
2319				markers.flush(sink)
2320				select {
2321				case p.out <- ports.Frame{Type: ports.MsgDetach, Payload: ports.MarshalDetach(ports.Detach{})}:
2322				case <-p.ctx.Done():
2323				}
2324				p.cancel()
2325				return
2326			}
2327			// Commit every callback from this raw read before observing a
2328			// concurrent cancellation. Any remaining marker prefix is retained
2329			// by the deferred handoff above, so reconnects never duplicate the
2330			// delivered bytes or lose a standalone Escape.
2331			input.ack(consumer)
2332			if p.ctx.Err() != nil {
2333				return
2334			}
2335		}
2336	}
2337}
2338
2339// runResize forwards coalesced terminal resize events to the daemon. It
2340// tolerates an already-closed resize channel (which the terminal adapter
2341// hands back when restore ran before ResizeEvents was first called).
2342func runResize(ctx context.Context, events <-chan domain.Size, out chan<- ports.Frame, log *slog.Logger) {
2343	defer log.Debug("resize pump exited")
2344	for {
2345		select {
2346		case sz, ok := <-events:
2347			if !ok {
2348				return
2349			}
2350			payload, err := ports.MarshalResize(ports.Resize{Size: sz})
2351			if err != nil {
2352				log.Error("encoding terminal resize", "error", err)
2353				continue
2354			}
2355			frame := ports.Frame{Type: ports.MsgResize, Payload: payload}
2356			select {
2357			case out <- frame:
2358			case <-ctx.Done():
2359				return
2360			}
2361		case <-ctx.Done():
2362			return
2363		}
2364	}
2365}
2366
2367// detachedResult maps a Detached reason to a return value: an ordinary
2368// user-initiated detach is a clean (nil) exit; anything else is surfaced so
2369// the user learns why the session went away.
2370func detachedResult(reason uint8) error {
2371	switch reason {
2372	case ports.ReasonDetach:
2373		return nil
2374	case ports.ReasonSessionKilled:
2375		return &DetachedError{Reason: reason, Text: "session was killed"}
2376	case ports.ReasonServerShutdown:
2377		return &DetachedError{Reason: reason, Text: "daemon shut down"}
2378	case ports.ReasonReplaced:
2379		return &DetachedError{Reason: reason, Text: "session taken over by another client"}
2380	default:
2381		return &DetachedError{Reason: reason, Text: "detached by daemon"}
2382	}
2383}