feat(daemon): construct remote view links

commit: d3ad7ec14ffac203d79460c2172391e72d950fcb
author: brice · 2mo ago · 2026-08-06 19:07 UTC
parent: 52f10a1
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +1117 -27

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


@@ -61,6 +61,13 @@ type remoteView struct {
61	61		screen *vt.Screen
62	62		metadata ports.SessionMeta
63	63		displayOrigin string
	64	+	// link and linkGeneration fence every remote transport callback. They are
	65	+	// owned by the view rather than the local-session registry, and neither
	66	+	// is ever used while view.mu is held for transport I/O.
	67	+	link *remoteLink
	68	+	linkGeneration uint64
	69	+	output remoteOutputState
	70	+	resetRequested bool
64	71		}
65	72
66	73		func (*remoteView) attachmentOwner() {}


@@ -61,6 +61,13 @@ type remoteView struct {
61	screen *vt.Screen	61		screen *vt.Screen
62	metadata ports.SessionMeta	62		metadata ports.SessionMeta
63	displayOrigin string	63		displayOrigin string
		64	+	// link and linkGeneration fence every remote transport callback. They are
		65	+	// owned by the view rather than the local-session registry, and neither
		66	+	// is ever used while view.mu is held for transport I/O.
		67	+	link *remoteLink
		68	+	linkGeneration uint64
		69	+	output remoteOutputState
		70	+	resetRequested bool
64	}	71		}
65		72
66	func (*remoteView) attachmentOwner() {}	73		func (*remoteView) attachmentOwner() {}

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


@@ -934,6 +934,16 @@ func (d *Daemon) resizeAttachmentForLease(token attachmentConnectionToken, size
934	934		if sameSize \|\| !token.attachmentEffectCurrent() \|\| !view.resizeScreen(ac.sizeSnapshot()) {
935	935		return false
936	936		}
	937	+	resizePayload, err := ports.MarshalResize(ports.Resize{Size: contentSize(ac.sizeSnapshot())})
	938	+	if err != nil {
	939	+	return false
	940	+	}
	941	+	if err := d.sendRemoteViewFrame(view, ports.Frame{Type: ports.MsgResize, Payload: resizePayload}); err != nil {
	942	+	// A disconnected remote link must not suppress the attachment-local
	943	+	// rebase: retained content remains locally renderable while Phase 4
	944	+	// reconnect policy restores the remote transport.
	945	+	d.log.Warn("forwarding remote view resize failed", "endpoint", view.key.endpoint, "session", view.key.sessionName, "err", err)
	946	+	}
937	947		return d.paintRemoteView(view, ac, true, token) == paintEmitted
938	948		}
939	949


@@ -934,6 +934,16 @@ func (d *Daemon) resizeAttachmentForLease(token attachmentConnectionToken, size
934	if sameSize \|\| !token.attachmentEffectCurrent() \|\| !view.resizeScreen(ac.sizeSnapshot()) {	934		if sameSize \|\| !token.attachmentEffectCurrent() \|\| !view.resizeScreen(ac.sizeSnapshot()) {
935	return false	935		return false
936	}	936		}
		937	+	resizePayload, err := ports.MarshalResize(ports.Resize{Size: contentSize(ac.sizeSnapshot())})
		938	+	if err != nil {
		939	+	return false
		940	+	}
		941	+	if err := d.sendRemoteViewFrame(view, ports.Frame{Type: ports.MsgResize, Payload: resizePayload}); err != nil {
		942	+	// A disconnected remote link must not suppress the attachment-local
		943	+	// rebase: retained content remains locally renderable while Phase 4
		944	+	// reconnect policy restores the remote transport.
		945	+	d.log.Warn("forwarding remote view resize failed", "endpoint", view.key.endpoint, "session", view.key.sessionName, "err", err)
		946	+	}
937	return d.paintRemoteView(view, ac, true, token) == paintEmitted	947		return d.paintRemoteView(view, ac, true, token) == paintEmitted
938	}	948		}
939		949

M internal/usecase/daemon/daemon.go +49 -26


@@ -77,7 +77,11 @@ type Daemon struct {
77	77		// sessions beyond concrete local *session values. Both are guarded by mu.
78	78		remoteViews map[remoteViewID]*remoteView
79	79		remoteViewsByKey map[remoteViewKey]remoteViewID
80		-	nextRemoteViewID remoteViewID
	80	+	// remoteViewConstructions reserves a remote key while its dial and
	81	+	// handshake run outside d.mu. A waiter never dials and can cancel without
	82	+	// disturbing the elected constructor.
	83	+	remoteViewConstructions map[remoteViewKey]*remoteViewConstruction
	84	+	nextRemoteViewID remoteViewID
81	85		// creating reserves names while durable creation I/O runs without mu.
82	86		creating map[string]struct{}
83	87		nextID uint64

@@ -600,30 +604,31 @@ func New(ptys ports.PTYFactory, clock ports.Clock, log *slog.Logger, opts ...Opt
600	604		}
601	605		paneProcessCtx, paneProcessCancel := context.WithCancel(context.Background())
602	606		d := &Daemon{
603		-	sessions: make(map[domain.SessionID]*session),
604		-	stopped: make(map[string]stoppedSession),
605		-	remoteViews: make(map[remoteViewID]*remoteView),
606		-	remoteViewsByKey: make(map[remoteViewKey]remoteViewID),
607		-	creating: make(map[string]struct{}),
608		-	parked: make(map[uint64]*parkedAttachment),
609		-	parking: make(map[uint64]*parkingAttachment),
610		-	paneProcessCtx: paneProcessCtx,
611		-	paneProcessCancel: paneProcessCancel,
612		-	ptys: ptys,
613		-	clock: clock,
614		-	log: log,
615		-	baseEnv: os.Environ(),
616		-	shell: shell,
617		-	dirOrHome: dirOrHome,
618		-	done: make(chan struct{}),
619		-	restoreDone: make(chan struct{}),
620		-	animWake: make(chan struct{}, 1),
621		-	snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
622		-	snapshotAdmitted: make(map[*snapshotCapture]struct{}),
623		-	snapshotWake: make(chan struct{}, 1),
624		-	notices: newNoticeCenter(),
625		-	remoteCatalog: newRemoteCatalogState(),
626		-	resumeParkGrace: defaultResumeParkGrace,
	607	+	sessions: make(map[domain.SessionID]*session),
	608	+	stopped: make(map[string]stoppedSession),
	609	+	remoteViews: make(map[remoteViewID]*remoteView),
	610	+	remoteViewsByKey: make(map[remoteViewKey]remoteViewID),
	611	+	remoteViewConstructions: make(map[remoteViewKey]*remoteViewConstruction),
	612	+	creating: make(map[string]struct{}),
	613	+	parked: make(map[uint64]*parkedAttachment),
	614	+	parking: make(map[uint64]*parkingAttachment),
	615	+	paneProcessCtx: paneProcessCtx,
	616	+	paneProcessCancel: paneProcessCancel,
	617	+	ptys: ptys,
	618	+	clock: clock,
	619	+	log: log,
	620	+	baseEnv: os.Environ(),
	621	+	shell: shell,
	622	+	dirOrHome: dirOrHome,
	623	+	done: make(chan struct{}),
	624	+	restoreDone: make(chan struct{}),
	625	+	animWake: make(chan struct{}, 1),
	626	+	snapshotJobs: make(chan *snapshotCapture, snapshotQueueCapacity),
	627	+	snapshotAdmitted: make(map[*snapshotCapture]struct{}),
	628	+	snapshotWake: make(chan struct{}, 1),
	629	+	notices: newNoticeCenter(),
	630	+	remoteCatalog: newRemoteCatalogState(),
	631	+	resumeParkGrace: defaultResumeParkGrace,
627	632		barScripts: &barScriptState{
628	633		cfg: barConfigFromDomain(domain.Defaults().Bar),
629	634		outputs: make(map[domain.SessionID]barScriptOutputs),

@@ -808,18 +813,36 @@ func (d Daemon) shutdownAllWithSnapshotDeadline(reason uint8, deadline snapsho
808	813		for _, view := range d.remoteViews {
809	814		remoteViews = append(remoteViews, view)
810	815		}
	816	+	constructions := make([]context.CancelFunc, 0, len(d.remoteViewConstructions))
	817	+	for _, construction := range d.remoteViewConstructions {
	818	+	if construction.cancel != nil {
	819	+	constructions = append(constructions, construction.cancel)
	820	+	}
	821	+	}
811	822		clear(d.remoteViews)
812	823		clear(d.remoteViewsByKey)
	824	+	clear(d.remoteViewConstructions)
813	825		snapshot := d.sessionsSnapshotLocked()
814	826		empty := len(snapshot) == 0
815	827		d.mu.Unlock()
816	828		// A remote view is not a daemon-liveness root. Retire each exact local
817	829		// attachment after releasing d.mu so a token cannot remain valid and a
818		-	// blocked local-client Recv/Send cannot outlive daemon shutdown.
	830	+	// blocked local-client Recv/Send cannot outlive daemon shutdown. Candidate
	831	+	// construction uses the same exact-close rule, even before publication.
	832	+	for _, cancel := range constructions {
	833	+	cancel()
	834	+	}
	835	+	remoteLinks := make([]*remoteLink, 0, len(remoteViews))
819	836		for _, view := range remoteViews {
820	837		for _, ac := range view.close() {
821	838		d.retireShutdownRemoteAttachment(view, ac, reason)
822	839		}
	840	+	if link := d.stopRemoteViewLink(view); link != nil {
	841	+	remoteLinks = append(remoteLinks, link)
	842	+	}
	843	+	}
	844	+	for _, link := range remoteLinks {
	845	+	<-link.done
823	846		}
824	847		d.finishParkedAttachmentRetirements(parkedRetirements)
825	848		d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot))

 	// sessions beyond concrete local *session values. Both are guarded by mu.
 	remoteViews      map[remoteViewID]*remoteView
 	remoteViewsByKey map[remoteViewKey]remoteViewID
-	nextRemoteViewID remoteViewID
+	// remoteViewConstructions reserves a remote key while its dial and
+	// handshake run outside d.mu. A waiter never dials and can cancel without
+	// disturbing the elected constructor.
+	remoteViewConstructions map[remoteViewKey]*remoteViewConstruction
+	nextRemoteViewID        remoteViewID
 	// creating reserves names while durable creation I/O runs without mu.
 	creating map[string]struct{}
 	nextID   uint64
 	}
 	paneProcessCtx, paneProcessCancel := context.WithCancel(context.Background())
 	d := &Daemon{
-		sessions:          make(map[domain.SessionID]*session),
+		sessions:                make(map[domain.SessionID]*session),
-		stopped:           make(map[string]stoppedSession),
+		stopped:                 make(map[string]stoppedSession),
-		remoteViews:       make(map[remoteViewID]*remoteView),
+		remoteViews:             make(map[remoteViewID]*remoteView),
-		remoteViewsByKey:  make(map[remoteViewKey]remoteViewID),
+		remoteViewsByKey:        make(map[remoteViewKey]remoteViewID),
-		creating:          make(map[string]struct{}),
+		remoteViewConstructions: make(map[remoteViewKey]*remoteViewConstruction),
-		parked:            make(map[uint64]*parkedAttachment),
+		creating:                make(map[string]struct{}),
-		parking:           make(map[uint64]*parkingAttachment),
+		parked:                  make(map[uint64]*parkedAttachment),
-		paneProcessCtx:    paneProcessCtx,
+		parking:                 make(map[uint64]*parkingAttachment),
-		paneProcessCancel: paneProcessCancel,
+		paneProcessCtx:          paneProcessCtx,
-		ptys:              ptys,
+		paneProcessCancel:       paneProcessCancel,
-		clock:             clock,
+		ptys:                    ptys,
-		log:               log,
+		clock:                   clock,
-		baseEnv:           os.Environ(),
+		log:                     log,
-		shell:             shell,
+		baseEnv:                 os.Environ(),
-		dirOrHome:         dirOrHome,
+		shell:                   shell,
-		done:              make(chan struct{}),
+		dirOrHome:               dirOrHome,
-		restoreDone:       make(chan struct{}),
+		done:                    make(chan struct{}),
-		animWake:          make(chan struct{}, 1),
+		restoreDone:             make(chan struct{}),
-		snapshotJobs:      make(chan *snapshotCapture, snapshotQueueCapacity),
+		animWake:                make(chan struct{}, 1),
-		snapshotAdmitted:  make(map[*snapshotCapture]struct{}),
+		snapshotJobs:            make(chan *snapshotCapture, snapshotQueueCapacity),
-		snapshotWake:      make(chan struct{}, 1),
+		snapshotAdmitted:        make(map[*snapshotCapture]struct{}),
-		notices:           newNoticeCenter(),
+		snapshotWake:            make(chan struct{}, 1),
-		remoteCatalog:     newRemoteCatalogState(),
+		notices:                 newNoticeCenter(),
-		resumeParkGrace:   defaultResumeParkGrace,
+		remoteCatalog:           newRemoteCatalogState(),
+		resumeParkGrace:         defaultResumeParkGrace,
 		barScripts: &barScriptState{
 			cfg:         barConfigFromDomain(domain.Defaults().Bar),
 			outputs:     make(map[domain.SessionID]barScriptOutputs),
 	for _, view := range d.remoteViews {
 		remoteViews = append(remoteViews, view)
 	}
+	constructions := make([]context.CancelFunc, 0, len(d.remoteViewConstructions))
+	for _, construction := range d.remoteViewConstructions {
+		if construction.cancel != nil {
+			constructions = append(constructions, construction.cancel)
+		}
+	}
 	clear(d.remoteViews)
 	clear(d.remoteViewsByKey)
+	clear(d.remoteViewConstructions)
 	snapshot := d.sessionsSnapshotLocked()
 	empty := len(snapshot) == 0
 	d.mu.Unlock()
 	// A remote view is not a daemon-liveness root. Retire each exact local
 	// attachment after releasing d.mu so a token cannot remain valid and a
-	// blocked local-client Recv/Send cannot outlive daemon shutdown.
+	// blocked local-client Recv/Send cannot outlive daemon shutdown. Candidate
+	// construction uses the same exact-close rule, even before publication.
+	for _, cancel := range constructions {
+		cancel()
+	}
+	remoteLinks := make([]*remoteLink, 0, len(remoteViews))
 	for _, view := range remoteViews {
 		for _, ac := range view.close() {
 			d.retireShutdownRemoteAttachment(view, ac, reason)
 		}
+		if link := d.stopRemoteViewLink(view); link != nil {
+			remoteLinks = append(remoteLinks, link)
+		}
+	}
+	for _, link := range remoteLinks {
+		<-link.done
 	}
 	d.finishParkedAttachmentRetirements(parkedRetirements)
 	d.log.Info("graceful shutdown begin", "reason", reason, "sessions", len(snapshot))

M internal/usecase/daemon/input.go +5 -1


@@ -21,10 +21,14 @@ func (d Daemon) handleSequencedInput(sess session, ac *attachedClient, _ uint6
21	21		d.handleInput(sess, ac, data)
22	22		}
23	23
24		-	func (d *Daemon) handleSequencedInputForAttachment(token attachmentConnectionToken, _ uint64, data []byte) {
	24	+	func (d *Daemon) handleSequencedInputForAttachment(token attachmentConnectionToken, inputSeq uint64, data []byte) {
25	25		if !token.attachmentEffectCurrent() {
26	26		return
27	27		}
	28	+	if view, remote := token.owner.(*remoteView); remote {
	29	+	d.handleRemoteViewInput(view, token, inputSeq, data)
	30	+	return
	31	+	}
28	32		if token.ac.renderMode == ports.RenderModeProxiedContent {
29	33		d.handleProxiedInputForAttachment(token, data)
30	34		return

 	d.handleInput(sess, ac, data)
 }
-func (d *Daemon) handleSequencedInputForAttachment(token attachmentConnectionToken, _ uint64, data []byte) {
+func (d *Daemon) handleSequencedInputForAttachment(token attachmentConnectionToken, inputSeq uint64, data []byte) {
 	if !token.attachmentEffectCurrent() {
 		return
 	}
+	if view, remote := token.owner.(*remoteView); remote {
+		d.handleRemoteViewInput(view, token, inputSeq, data)
+		return
+	}
 	if token.ac.renderMode == ports.RenderModeProxiedContent {
 		d.handleProxiedInputForAttachment(token, data)
 		return

A internal/usecase/daemon/remote_link.go +663 -0

@@ -0,0 +1,663 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"crypto/rand"
	6	+	"errors"
	7	+	"fmt"
	8	+	"sync"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	"github.com/bnema/vev/pkg/vt"
	13	+	)
	14	+
	15	+	var (
	16	+	errRemoteViewUnavailable = errors.New("remote view: link is unavailable")
	17	+	errRemoteViewStale = errors.New("remote view: link is stale")
	18	+	)
	19	+
	20	+	// remoteViewConstruction reserves one remote-view key while its candidate
	21	+	// dials and handshakes. It is guarded by Daemon.mu. cancel is invoked only
	22	+	// after the daemon lock is released, because it may interrupt adapter I/O.
	23	+	type remoteViewConstruction struct {
	24	+	done chan struct{}
	25	+	cancel context.CancelFunc
	26	+	err error
	27	+	}
	28	+
	29	+	// remoteLink is one exact transport generation for a remote view. sendMu
	30	+	// orders every outbound remote frame; view.mu guards whether this exact link
	31	+	// remains current. Neither lock is held while transport I/O runs.
	32	+	type remoteLink struct {
	33	+	view *remoteView
	34	+	generation uint64
	35	+	target domain.RemoteSessionTarget
	36	+	clientID [16]byte
	37	+	transport ports.Transport
	38	+	cancel context.CancelFunc
	39	+
	40	+	sendMu sync.Mutex
	41	+	startOnce sync.Once
	42	+	done chan struct{}
	43	+	active bool // guarded by view.mu
	44	+	}
	45	+
	46	+	// remoteOutputState is the remote-link equivalent of the thin client's
	47	+	// dependency chain. It remains private to the daemon because this link owns a
	48	+	// persistent VT rather than forwarding output bytes to a terminal.
	49	+	type remoteOutputState struct {
	50	+	epoch uint64
	51	+	state uint64
	52	+	viewRevision uint64
	53	+	initialized bool
	54	+	}
	55	+
	56	+	func (s remoteOutputState) next(output ports.Output) (remoteOutputState, bool) {
	57	+	if output.Epoch == 0 {
	58	+	return remoteOutputState{}, false
	59	+	}
	60	+	if !s.initialized {
	61	+	if output.New == 0 {
	62	+	return s, true
	63	+	}
	64	+	if output.Base != 0 \|\| !output.Full {
	65	+	return remoteOutputState{}, false
	66	+	}
	67	+	return remoteOutputState{epoch: output.Epoch, state: output.New, viewRevision: output.ViewRevision, initialized: true}, true
	68	+	}
	69	+	if output.Epoch < s.epoch {
	70	+	return remoteOutputState{}, false
	71	+	}
	72	+	if output.Epoch == s.epoch {
	73	+	if output.ViewRevision != s.viewRevision {
	74	+	return remoteOutputState{}, false
	75	+	}
	76	+	if output.New == 0 {
	77	+	if output.Base != 0 \|\| output.Full {
	78	+	return remoteOutputState{}, false
	79	+	}
	80	+	return s, true
	81	+	}
	82	+	if output.Full \|\| output.Base != s.state \|\| output.New != output.Base+1 {
	83	+	return remoteOutputState{}, false
	84	+	}
	85	+	return remoteOutputState{epoch: s.epoch, state: output.New, viewRevision: s.viewRevision, initialized: true}, true
	86	+	}
	87	+	if !output.Full \|\| output.Base != 0 {
	88	+	return remoteOutputState{}, false
	89	+	}
	90	+	return remoteOutputState{epoch: output.Epoch, state: output.New, viewRevision: output.ViewRevision, initialized: true}, true
	91	+	}
	92	+
	93	+	// openRemoteView returns the exact warm view for target or elects one
	94	+	// constructor. The reservation is protected only by Daemon.mu; the dial and
	95	+	// all handshake I/O happen after every architecture lock is released.
	96	+	func (d Daemon) openRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (remoteView, error) {
	97	+	if d == nil {
	98	+	return nil, errors.New("remote view: nil daemon")
	99	+	}
	100	+	if ctx == nil {
	101	+	ctx = context.Background()
	102	+	}
	103	+	if err := target.Validate(); err != nil {
	104	+	return nil, fmt.Errorf("remote view: invalid target: %w", err)
	105	+	}
	106	+	if !size.Valid() {
	107	+	return nil, fmt.Errorf("remote view: invalid size %dx%d", size.Cols, size.Rows)
	108	+	}
	109	+	if d.remoteDialerFactory == nil {
	110	+	return nil, errors.New("remote view: remote dialer is not configured")
	111	+	}
	112	+	key, err := remoteViewKeyForTarget(target)
	113	+	if err != nil {
	114	+	return nil, fmt.Errorf("remote view: key: %w", err)
	115	+	}
	116	+
	117	+	for {
	118	+	if err := ctx.Err(); err != nil {
	119	+	return nil, err
	120	+	}
	121	+	d.mu.Lock()
	122	+	if existing := d.remoteViewByKeyLocked(key); existing != nil {
	123	+	d.mu.Unlock()
	124	+	return existing, nil
	125	+	}
	126	+	if d.closing {
	127	+	d.mu.Unlock()
	128	+	return nil, errors.New("remote view: daemon is shutting down")
	129	+	}
	130	+	if construction := d.remoteViewConstructions[key]; construction != nil {
	131	+	done := construction.done
	132	+	d.mu.Unlock()
	133	+	select {
	134	+	case <-ctx.Done():
	135	+	return nil, ctx.Err()
	136	+	case <-done:
	137	+	}
	138	+	if err := ctx.Err(); err != nil {
	139	+	return nil, err
	140	+	}
	141	+	d.mu.Lock()
	142	+	winner := d.remoteViewByKeyLocked(key)
	143	+	constructionErr := construction.err
	144	+	d.mu.Unlock()
	145	+	if winner != nil {
	146	+	return winner, nil
	147	+	}
	148	+	if constructionErr != nil && !errors.Is(constructionErr, context.Canceled) && !errors.Is(constructionErr, context.DeadlineExceeded) {
	149	+	return nil, constructionErr
	150	+	}
	151	+	// A canceled candidate never poisons an independent waiter. Re-elect
	152	+	// if the caller still wants this exact remote view.
	153	+	continue
	154	+	}
	155	+
	156	+	constructionCtx, cancel := context.WithCancel(ctx)
	157	+	construction := &remoteViewConstruction{done: make(chan struct{}), cancel: cancel}
	158	+	if d.remoteViewConstructions == nil {
	159	+	d.remoteViewConstructions = make(map[remoteViewKey]*remoteViewConstruction)
	160	+	}
	161	+	d.remoteViewConstructions[key] = construction
	162	+	d.mu.Unlock()
	163	+
	164	+	candidate, constructionErr := d.constructRemoteView(constructionCtx, target, size)
	165	+	if constructionErr == nil {
	166	+	constructionErr = ctx.Err()
	167	+	}
	168	+	cancel()
	169	+
	170	+	var winner *remoteView
	171	+	publishErr := constructionErr
	172	+	d.mu.Lock()
	173	+	if publishErr == nil {
	174	+	if d.closing {
	175	+	publishErr = errors.New("remote view: daemon is shutting down")
	176	+	} else if existing := d.remoteViewByKeyLocked(key); existing != nil {
	177	+	winner = existing
	178	+	} else if err := d.registerRemoteViewLocked(candidate); err != nil {
	179	+	publishErr = fmt.Errorf("remote view: publish candidate: %w", err)
	180	+	} else {
	181	+	winner = candidate
	182	+	// Starting the receiver while d.mu still protects publication keeps
	183	+	// shutdown from observing a registry-visible link with no worker
	184	+	// to join.
	185	+	d.startRemoteLink(candidate.link)
	186	+	}
	187	+	}
	188	+	construction.err = publishErr
	189	+	if d.remoteViewConstructions[key] == construction {
	190	+	delete(d.remoteViewConstructions, key)
	191	+	}
	192	+	close(construction.done)
	193	+	d.mu.Unlock()
	194	+
	195	+	if candidate != nil && candidate != winner {
	196	+	d.stopUnpublishedRemoteView(candidate)
	197	+	}
	198	+	if winner != nil {
	199	+	return winner, nil
	200	+	}
	201	+	return nil, publishErr
	202	+	}
	203	+	}
	204	+
	205	+	// constructRemoteView builds an unexposed candidate. Its caller owns a
	206	+	// constructor reservation but holds no daemon, session, or view lock.
	207	+	func (d Daemon) constructRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (_ remoteView, err error) {
	208	+	content := contentSize(size)
	209	+	if !content.Valid() {
	210	+	return nil, fmt.Errorf("remote view: invalid content size %dx%d", content.Cols, content.Rows)
	211	+	}
	212	+	key, err := remoteViewKeyForTarget(target)
	213	+	if err != nil {
	214	+	return nil, err
	215	+	}
	216	+	var clientID [16]byte
	217	+	if _, err := rand.Read(clientID[:]); err != nil {
	218	+	return nil, fmt.Errorf("remote view: generate client ID: %w", err)
	219	+	}
	220	+	root := context.Background()
	221	+	if d.serveCtx != nil {
	222	+	root = d.serveCtx
	223	+	}
	224	+	linkCtx, cancelLink := context.WithCancel(root)
	225	+	stopCallerCancellation := context.AfterFunc(ctx, cancelLink)
	226	+	defer stopCallerCancellation()
	227	+	view := &remoteView{
	228	+	key: key,
	229	+	displayOrigin: target.DisplayOrigin,
	230	+	screen: vt.NewScreen(content.Cols, content.Rows),
	231	+	linkGeneration: 1,
	232	+	}
	233	+	link := &remoteLink{
	234	+	view: view,
	235	+	generation: view.linkGeneration,
	236	+	target: target,
	237	+	clientID: clientID,
	238	+	cancel: cancelLink,
	239	+	done: make(chan struct{}),
	240	+	active: true,
	241	+	}
	242	+	view.link = link
	243	+	published := false
	244	+	defer func() {
	245	+	if !published {
	246	+	d.stopUnpublishedRemoteView(view)
	247	+	}
	248	+	}()
	249	+
	250	+	handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(linkCtx)
	251	+	defer finishHandshake()
	252	+	dialer, err := d.remoteDialerFactory.DialerForRemote(target.Endpoint, target.SessionName, d.remoteTransportMode, d.log)
	253	+	if err != nil {
	254	+	return nil, fmt.Errorf("remote view: select remote dialer: %w", err)
	255	+	}
	256	+	if dialer == nil {
	257	+	return nil, errors.New("remote view: remote dialer factory returned nil dialer")
	258	+	}
	259	+	transport, err := dialer.Dial(handshakeCtx)
	260	+	if err != nil {
	261	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: dial: %w", err))
	262	+	}
	263	+	if transport == nil {
	264	+	return nil, errors.New("remote view: remote dialer returned nil transport")
	265	+	}
	266	+	link.transport = transport
	267	+	stopWatch := watchHandshakeTransport(handshakeCtx, transport)
	268	+	defer stopWatch()
	269	+
	270	+	helloTarget := target
	271	+	hello := ports.Hello{
	272	+	Version: ports.ProtocolVersion,
	273	+	Intent: ports.IntentAttach,
	274	+	ClientID: clientID,
	275	+	Name: target.SessionName,
	276	+	Size: content,
	277	+	RenderMode: ports.RenderModeProxiedContent,
	278	+	MaxOutputInFlight: remoteLinkOutputWindow(transport),
	279	+	RemoteTarget: &helloTarget,
	280	+	EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
	281	+	}
	282	+	if err := link.send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
	283	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: send hello: %w", err))
	284	+	}
	285	+
	286	+	welcomeFrame, err := recvRemoteLinkHandshakeFrame(handshakeCtx, transport)
	287	+	if err != nil {
	288	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: receive welcome: %w", err))
	289	+	}
	290	+	if err := validateRemoteLinkWelcome(welcomeFrame, target); err != nil {
	291	+	return nil, err
	292	+	}
	293	+	metadataFrame, err := recvRemoteLinkHandshakeFrame(handshakeCtx, transport)
	294	+	if err != nil {
	295	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: receive metadata: %w", err))
	296	+	}
	297	+	metadata, err := validateInitialRemoteLinkMetadata(metadataFrame, target)
	298	+	if err != nil {
	299	+	return nil, err
	300	+	}
	301	+	if !view.applyRemoteMetadata(link, metadata, true) {
	302	+	return nil, errors.New("remote view: initial metadata became stale")
	303	+	}
	304	+	outputFrame, err := recvRemoteLinkHandshakeFrame(handshakeCtx, transport)
	305	+	if err != nil {
	306	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: receive initial output: %w", err))
	307	+	}
	308	+	if outputFrame.Type != ports.MsgOutput {
	309	+	return nil, fmt.Errorf("remote view: third server frame is %d, want Output", outputFrame.Type)
	310	+	}
	311	+	output, err := ports.UnmarshalOutput(outputFrame.Payload)
	312	+	if err != nil {
	313	+	return nil, fmt.Errorf("remote view: decode initial output: %w", err)
	314	+	}
	315	+	ack, reset, _, accepted := view.applyRemoteOutput(link, output)
	316	+	if !accepted \|\| reset \|\| output.New == 0 \|\| !output.Full {
	317	+	return nil, errors.New("remote view: initial output is not an authoritative reset")
	318	+	}
	319	+	if ack != 0 {
	320	+	ackPayload, err := ports.MarshalAck(ports.Ack{Epoch: output.Epoch, State: ack})
	321	+	if err != nil {
	322	+	return nil, fmt.Errorf("remote view: encode initial output acknowledgement: %w", err)
	323	+	}
	324	+	if err := link.send(ports.Frame{Type: ports.MsgAck, Payload: ackPayload}); err != nil {
	325	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: acknowledge initial output: %w", err))
	326	+	}
	327	+	}
	328	+	if err := handshakeCtx.Err(); err != nil {
	329	+	return nil, handshakeContextError(ctx, timedOut, err)
	330	+	}
	331	+	published = true
	332	+	return view, nil
	333	+	}
	334	+
	335	+	func remoteLinkOutputWindow(transport ports.Transport) uint8 {
	336	+	if _, datagram := transport.(ports.DatagramTransport); datagram {
	337	+	return 1
	338	+	}
	339	+	return maxUnackedOutputStates
	340	+	}
	341	+
	342	+	func recvRemoteLinkHandshakeFrame(ctx context.Context, transport ports.Transport) (ports.Frame, error) {
	343	+	var frame ports.Frame
	344	+	err := boundedHandshakeOperation(ctx, transport, func() error {
	345	+	var recvErr error
	346	+	frame, recvErr = transport.Recv()
	347	+	return recvErr
	348	+	})
	349	+	if err != nil {
	350	+	return ports.Frame{}, err
	351	+	}
	352	+	return frame, nil
	353	+	}
	354	+
	355	+	func validateRemoteLinkWelcome(frame ports.Frame, target domain.RemoteSessionTarget) error {
	356	+	if frame.Type == ports.MsgError {
	357	+	remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
	358	+	if err != nil {
	359	+	return fmt.Errorf("remote view: malformed remote error: %w", err)
	360	+	}
	361	+	return fmt.Errorf("remote view: remote rejected handshake (%d): %s", remoteErr.Code, remoteErr.Text)
	362	+	}
	363	+	if frame.Type != ports.MsgWelcome {
	364	+	return fmt.Errorf("remote view: first server frame is %d, want Welcome", frame.Type)
	365	+	}
	366	+	welcome, err := ports.UnmarshalWelcome(frame.Payload)
	367	+	if err != nil {
	368	+	return fmt.Errorf("remote view: decode welcome: %w", err)
	369	+	}
	370	+	if welcome.SessionID == "" \|\| welcome.SessionName != target.SessionName \|\| welcome.RenderMode != ports.RenderModeProxiedContent {
	371	+	return errors.New("remote view: welcome identity or render mode mismatch")
	372	+	}
	373	+	return nil
	374	+	}
	375	+
	376	+	func validateInitialRemoteLinkMetadata(frame ports.Frame, target domain.RemoteSessionTarget) (ports.SessionMeta, error) {
	377	+	if frame.Type != ports.MsgSessionMeta {
	378	+	return ports.SessionMeta{}, fmt.Errorf("remote view: second server frame is %d, want SessionMeta", frame.Type)
	379	+	}
	380	+	metadata, err := ports.UnmarshalSessionMeta(frame.Payload)
	381	+	if err != nil {
	382	+	return ports.SessionMeta{}, fmt.Errorf("remote view: decode initial metadata: %w", err)
	383	+	}
	384	+	if metadata.LifecycleID != target.LifecycleID \|\| metadata.SessionName != target.SessionName \|\| metadata.Revision != 1 {
	385	+	return ports.SessionMeta{}, errors.New("remote view: metadata identity mismatch")
	386	+	}
	387	+	if !target.Stopped && metadata.ActiveTabID != target.LiveTabID {
	388	+	return ports.SessionMeta{}, errors.New("remote view: metadata active tab mismatch")
	389	+	}
	390	+	return metadata, nil
	391	+	}
	392	+
	393	+	func cloneRemoteSessionMeta(metadata ports.SessionMeta) ports.SessionMeta {
	394	+	metadata.Tabs = append([]ports.SessionTabMeta(nil), metadata.Tabs...)
	395	+	return metadata
	396	+	}
	397	+
	398	+	// applyRemoteMetadata commits one validated update only for the exact current
	399	+	// link. The view lock protects content state only; no renderer or callback runs
	400	+	// before it is released.
	401	+	func (v remoteView) applyRemoteMetadata(link remoteLink, metadata ports.SessionMeta, initial bool) bool {
	402	+	if v == nil \|\| link == nil {
	403	+	return false
	404	+	}
	405	+	v.mu.Lock()
	406	+	defer v.mu.Unlock()
	407	+	if v.closed \|\| v.link != link \|\| v.linkGeneration != link.generation {
	408	+	return false
	409	+	}
	410	+	if initial {
	411	+	if v.metadata.Revision != 0 {
	412	+	return false
	413	+	}
	414	+	} else if metadata.Revision <= v.metadata.Revision {
	415	+	return false
	416	+	}
	417	+	v.metadata = cloneRemoteSessionMeta(metadata)
	418	+	return true
	419	+	}
	420	+
	421	+	// applyRemoteOutput validates and applies state-bearing ANSI to the persistent
	422	+	// private VT. It returns only immutable post-commit work; callers send ACK or
	423	+	// reset frames and repaint attachments after view.mu is released.
	424	+	func (v remoteView) applyRemoteOutput(link remoteLink, output ports.Output) (ack uint64, reset bool, attachments []*attachedClient, accepted bool) {
	425	+	if v == nil \|\| link == nil {
	426	+	return 0, false, nil, false
	427	+	}
	428	+	v.mu.Lock()
	429	+	defer v.mu.Unlock()
	430	+	if v.closed \|\| v.link != link \|\| v.linkGeneration != link.generation {
	431	+	return 0, false, nil, false
	432	+	}
	433	+	next, valid := v.output.next(output)
	434	+	if !valid \|\| (output.New != 0 && (v.screen == nil \|\| v.screen.Frame.Width != output.Size.Cols \|\| v.screen.Frame.Height != output.Size.Rows)) {
	435	+	if v.resetRequested {
	436	+	return 0, false, nil, false
	437	+	}
	438	+	v.resetRequested = true
	439	+	return 0, true, nil, false
	440	+	}
	441	+	if output.New == 0 {
	442	+	return 0, false, nil, true
	443	+	}
	444	+	v.output = next
	445	+	v.resetRequested = false
	446	+	if len(output.Data) != 0 {
	447	+	v.screen.Write(output.Data)
	448	+	}
	449	+	attachments = make([]*attachedClient, 0, len(v.attachments))
	450	+	for attachment := range v.attachments {
	451	+	attachments = append(attachments, attachment)
	452	+	}
	453	+	return output.New, false, attachments, true
	454	+	}
	455	+
	456	+	func (d Daemon) startRemoteLink(link remoteLink) {
	457	+	if d == nil \|\| link == nil \|\| link.transport == nil {
	458	+	return
	459	+	}
	460	+	link.startOnce.Do(func() {
	461	+	go d.runRemoteLink(link)
	462	+	})
	463	+	}
	464	+
	465	+	func (d Daemon) runRemoteLink(link remoteLink) {
	466	+	defer close(link.done)
	467	+	for {
	468	+	frame, err := link.transport.Recv()
	469	+	if err != nil {
	470	+	d.markRemoteLinkUnavailable(link)
	471	+	return
	472	+	}
	473	+	if err := d.handleRemoteLinkFrame(link, frame); err != nil {
	474	+	d.log.Warn("remote view link stopped", "endpoint", link.target.Endpoint, "session", link.target.SessionName, "err", err)
	475	+	d.markRemoteLinkUnavailable(link)
	476	+	return
	477	+	}
	478	+	}
	479	+	}
	480	+
	481	+	func (d Daemon) handleRemoteLinkFrame(link remoteLink, frame ports.Frame) error {
	482	+	switch frame.Type {
	483	+	case ports.MsgOutput:
	484	+	output, err := ports.UnmarshalOutput(frame.Payload)
	485	+	if err != nil {
	486	+	return fmt.Errorf("remote view: decode output: %w", err)
	487	+	}
	488	+	ack, reset, attachments, accepted := link.view.applyRemoteOutput(link, output)
	489	+	if reset {
	490	+	if err := link.send(ports.Frame{Type: ports.MsgOutputResetRequest, Payload: ports.MarshalOutputResetRequest(ports.OutputResetRequest{})}); err != nil {
	491	+	return fmt.Errorf("remote view: request output reset: %w", err)
	492	+	}
	493	+	return nil
	494	+	}
	495	+	if !accepted {
	496	+	return nil
	497	+	}
	498	+	if ack != 0 {
	499	+	ackPayload, err := ports.MarshalAck(ports.Ack{Epoch: output.Epoch, State: ack})
	500	+	if err != nil {
	501	+	return fmt.Errorf("remote view: encode output acknowledgement: %w", err)
	502	+	}
	503	+	if err := link.send(ports.Frame{Type: ports.MsgAck, Payload: ackPayload}); err != nil {
	504	+	return fmt.Errorf("remote view: acknowledge output: %w", err)
	505	+	}
	506	+	}
	507	+	for _, attachment := range attachments {
	508	+	token := attachmentOwnerToken(link.view, attachment, attachment.transport())
	509	+	if token.ac != nil {
	510	+	go d.paintRemoteView(link.view, attachment, false, token)
	511	+	}
	512	+	}
	513	+	return nil
	514	+	case ports.MsgSessionMeta:
	515	+	metadata, err := ports.UnmarshalSessionMeta(frame.Payload)
	516	+	if err != nil {
	517	+	return fmt.Errorf("remote view: decode metadata: %w", err)
	518	+	}
	519	+	if metadata.LifecycleID != link.target.LifecycleID \|\| metadata.SessionName != link.target.SessionName \|\| !link.view.applyRemoteMetadata(link, metadata, false) {
	520	+	return errors.New("remote view: invalid metadata update")
	521	+	}
	522	+	return nil
	523	+	case ports.MsgPing:
	524	+	if _, err := ports.UnmarshalPing(frame.Payload); err != nil {
	525	+	return fmt.Errorf("remote view: decode ping: %w", err)
	526	+	}
	527	+	return link.send(ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})})
	528	+	case ports.MsgPong:
	529	+	_, err := ports.UnmarshalPong(frame.Payload)
	530	+	return err
	531	+	case ports.MsgDetached:
	532	+	_, err := ports.UnmarshalDetached(frame.Payload)
	533	+	if err != nil {
	534	+	return fmt.Errorf("remote view: decode detached: %w", err)
	535	+	}
	536	+	return errors.New("remote view: remote session detached")
	537	+	case ports.MsgError:
	538	+	remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
	539	+	if err != nil {
	540	+	return fmt.Errorf("remote view: malformed remote error: %w", err)
	541	+	}
	542	+	return fmt.Errorf("remote view: remote error (%d): %s", remoteErr.Code, remoteErr.Text)
	543	+	default:
	544	+	return fmt.Errorf("remote view: unexpected server frame type %d", frame.Type)
	545	+	}
	546	+	}
	547	+
	548	+	// send serializes the complete outbound remote stream. It captures the exact
	549	+	// current transport under view.mu, releases every architecture lock, then lets
	550	+	// Close interrupt a blocked Send concurrently.
	551	+	func (link *remoteLink) send(frame ports.Frame) error {
	552	+	if link == nil \|\| link.view == nil {
	553	+	return errRemoteViewUnavailable
	554	+	}
	555	+	link.sendMu.Lock()
	556	+	view := link.view
	557	+	view.mu.Lock()
	558	+	current := !view.closed && view.link == link && view.linkGeneration == link.generation && link.active
	559	+	transport := link.transport
	560	+	view.mu.Unlock()
	561	+	if !current \|\| transport == nil {
	562	+	link.sendMu.Unlock()
	563	+	return errRemoteViewStale
	564	+	}
	565	+	err := transport.Send(frame)
	566	+	link.sendMu.Unlock()
	567	+	if err != nil {
	568	+	link.cancel()
	569	+	_ = transport.Close()
	570	+	}
	571	+	return err
	572	+	}
	573	+
	574	+	// handleRemoteViewInput forwards the raw local-terminal bytes and their
	575	+	// original sequence through the exact current remote link. No local key,
	576	+	// overlay, mouse, PTY, or persistence path observes those bytes.
	577	+	func (d Daemon) handleRemoteViewInput(view remoteView, token attachmentConnectionToken, inputSeq uint64, data []byte) {
	578	+	if view == nil \|\| len(data) == 0 \|\| !token.attachmentEffectCurrent() {
	579	+	return
	580	+	}
	581	+	payload := ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: append([]byte(nil), data...)})
	582	+	if err := d.sendRemoteViewFrame(view, ports.Frame{Type: ports.MsgInput, Payload: payload}); err != nil {
	583	+	d.log.Warn("forwarding remote view input failed", "endpoint", view.key.endpoint, "session", view.key.sessionName, "err", err)
	584	+	}
	585	+	}
	586	+
	587	+	// sendRemoteViewFrame snapshots the exact link while holding only view.mu;
	588	+	// remoteLink.send performs the generation check and serialized transport I/O.
	589	+	func (d Daemon) sendRemoteViewFrame(view remoteView, frame ports.Frame) error {
	590	+	if view == nil {
	591	+	return errRemoteViewUnavailable
	592	+	}
	593	+	view.mu.Lock()
	594	+	link := view.link
	595	+	view.mu.Unlock()
	596	+	if link == nil {
	597	+	return errRemoteViewUnavailable
	598	+	}
	599	+	return link.send(frame)
	600	+	}
	601	+
	602	+	func (d Daemon) markRemoteLinkUnavailable(link remoteLink) {
	603	+	if link == nil \|\| link.view == nil {
	604	+	return
	605	+	}
	606	+	view := link.view
	607	+	view.mu.Lock()
	608	+	current := !view.closed && view.link == link && view.linkGeneration == link.generation
	609	+	if current {
	610	+	link.active = false
	611	+	}
	612	+	view.mu.Unlock()
	613	+	if current {
	614	+	link.cancel()
	615	+	_ = link.transport.Close()
	616	+	}
	617	+	}
	618	+
	619	+	// stopUnpublishedRemoteView is used for candidate failure and lost-election
	620	+	// cleanup. It never touches the daemon registry and closes only the candidate
	621	+	// transport generation after view.mu is released.
	622	+	func (d Daemon) stopUnpublishedRemoteView(view remoteView) {
	623	+	if view == nil {
	624	+	return
	625	+	}
	626	+	view.mu.Lock()
	627	+	view.closed = true
	628	+	link := view.link
	629	+	view.link = nil
	630	+	view.linkGeneration++
	631	+	view.mu.Unlock()
	632	+	if link != nil {
	633	+	link.cancel()
	634	+	if link.transport != nil {
	635	+	_ = link.transport.Close()
	636	+	}
	637	+	}
	638	+	}
	639	+
	640	+	// stopRemoteViewLink interrupts only the exact currently installed remote
	641	+	// transport. It is called after registry publication has been revoked, so it
	642	+	// cannot affect a successor view or link generation. The returned link has a
	643	+	// started worker, which the terminal shutdown path joins after Close.
	644	+	func (d Daemon) stopRemoteViewLink(view remoteView) *remoteLink {
	645	+	if view == nil {
	646	+	return nil
	647	+	}
	648	+	view.mu.Lock()
	649	+	link := view.link
	650	+	if link != nil {
	651	+	view.link = nil
	652	+	view.linkGeneration++
	653	+	link.active = false
	654	+	}
	655	+	view.mu.Unlock()
	656	+	if link != nil {
	657	+	link.cancel()
	658	+	if link.transport != nil {
	659	+	_ = link.transport.Close()
	660	+	}
	661	+	}
	662	+	return link
	663	+	}

@@ -0,0 +1,663 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"crypto/rand"
	6	+	"errors"
	7	+	"fmt"
	8	+	"sync"
	9	+
	10	+	"github.com/bnema/vev/internal/domain"
	11	+	"github.com/bnema/vev/internal/ports"
	12	+	"github.com/bnema/vev/pkg/vt"
	13	+	)
	14	+
	15	+	var (
	16	+	errRemoteViewUnavailable = errors.New("remote view: link is unavailable")
	17	+	errRemoteViewStale = errors.New("remote view: link is stale")
	18	+	)
	19	+
	20	+	// remoteViewConstruction reserves one remote-view key while its candidate
	21	+	// dials and handshakes. It is guarded by Daemon.mu. cancel is invoked only
	22	+	// after the daemon lock is released, because it may interrupt adapter I/O.
	23	+	type remoteViewConstruction struct {
	24	+	done chan struct{}
	25	+	cancel context.CancelFunc
	26	+	err error
	27	+	}
	28	+
	29	+	// remoteLink is one exact transport generation for a remote view. sendMu
	30	+	// orders every outbound remote frame; view.mu guards whether this exact link
	31	+	// remains current. Neither lock is held while transport I/O runs.
	32	+	type remoteLink struct {
	33	+	view *remoteView
	34	+	generation uint64
	35	+	target domain.RemoteSessionTarget
	36	+	clientID [16]byte
	37	+	transport ports.Transport
	38	+	cancel context.CancelFunc
	39	+
	40	+	sendMu sync.Mutex
	41	+	startOnce sync.Once
	42	+	done chan struct{}
	43	+	active bool // guarded by view.mu
	44	+	}
	45	+
	46	+	// remoteOutputState is the remote-link equivalent of the thin client's
	47	+	// dependency chain. It remains private to the daemon because this link owns a
	48	+	// persistent VT rather than forwarding output bytes to a terminal.
	49	+	type remoteOutputState struct {
	50	+	epoch uint64
	51	+	state uint64
	52	+	viewRevision uint64
	53	+	initialized bool
	54	+	}
	55	+
	56	+	func (s remoteOutputState) next(output ports.Output) (remoteOutputState, bool) {
	57	+	if output.Epoch == 0 {
	58	+	return remoteOutputState{}, false
	59	+	}
	60	+	if !s.initialized {
	61	+	if output.New == 0 {
	62	+	return s, true
	63	+	}
	64	+	if output.Base != 0 \|\| !output.Full {
	65	+	return remoteOutputState{}, false
	66	+	}
	67	+	return remoteOutputState{epoch: output.Epoch, state: output.New, viewRevision: output.ViewRevision, initialized: true}, true
	68	+	}
	69	+	if output.Epoch < s.epoch {
	70	+	return remoteOutputState{}, false
	71	+	}
	72	+	if output.Epoch == s.epoch {
	73	+	if output.ViewRevision != s.viewRevision {
	74	+	return remoteOutputState{}, false
	75	+	}
	76	+	if output.New == 0 {
	77	+	if output.Base != 0 \|\| output.Full {
	78	+	return remoteOutputState{}, false
	79	+	}
	80	+	return s, true
	81	+	}
	82	+	if output.Full \|\| output.Base != s.state \|\| output.New != output.Base+1 {
	83	+	return remoteOutputState{}, false
	84	+	}
	85	+	return remoteOutputState{epoch: s.epoch, state: output.New, viewRevision: s.viewRevision, initialized: true}, true
	86	+	}
	87	+	if !output.Full \|\| output.Base != 0 {
	88	+	return remoteOutputState{}, false
	89	+	}
	90	+	return remoteOutputState{epoch: output.Epoch, state: output.New, viewRevision: output.ViewRevision, initialized: true}, true
	91	+	}
	92	+
	93	+	// openRemoteView returns the exact warm view for target or elects one
	94	+	// constructor. The reservation is protected only by Daemon.mu; the dial and
	95	+	// all handshake I/O happen after every architecture lock is released.
	96	+	func (d Daemon) openRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (remoteView, error) {
	97	+	if d == nil {
	98	+	return nil, errors.New("remote view: nil daemon")
	99	+	}
	100	+	if ctx == nil {
	101	+	ctx = context.Background()
	102	+	}
	103	+	if err := target.Validate(); err != nil {
	104	+	return nil, fmt.Errorf("remote view: invalid target: %w", err)
	105	+	}
	106	+	if !size.Valid() {
	107	+	return nil, fmt.Errorf("remote view: invalid size %dx%d", size.Cols, size.Rows)
	108	+	}
	109	+	if d.remoteDialerFactory == nil {
	110	+	return nil, errors.New("remote view: remote dialer is not configured")
	111	+	}
	112	+	key, err := remoteViewKeyForTarget(target)
	113	+	if err != nil {
	114	+	return nil, fmt.Errorf("remote view: key: %w", err)
	115	+	}
	116	+
	117	+	for {
	118	+	if err := ctx.Err(); err != nil {
	119	+	return nil, err
	120	+	}
	121	+	d.mu.Lock()
	122	+	if existing := d.remoteViewByKeyLocked(key); existing != nil {
	123	+	d.mu.Unlock()
	124	+	return existing, nil
	125	+	}
	126	+	if d.closing {
	127	+	d.mu.Unlock()
	128	+	return nil, errors.New("remote view: daemon is shutting down")
	129	+	}
	130	+	if construction := d.remoteViewConstructions[key]; construction != nil {
	131	+	done := construction.done
	132	+	d.mu.Unlock()
	133	+	select {
	134	+	case <-ctx.Done():
	135	+	return nil, ctx.Err()
	136	+	case <-done:
	137	+	}
	138	+	if err := ctx.Err(); err != nil {
	139	+	return nil, err
	140	+	}
	141	+	d.mu.Lock()
	142	+	winner := d.remoteViewByKeyLocked(key)
	143	+	constructionErr := construction.err
	144	+	d.mu.Unlock()
	145	+	if winner != nil {
	146	+	return winner, nil
	147	+	}
	148	+	if constructionErr != nil && !errors.Is(constructionErr, context.Canceled) && !errors.Is(constructionErr, context.DeadlineExceeded) {
	149	+	return nil, constructionErr
	150	+	}
	151	+	// A canceled candidate never poisons an independent waiter. Re-elect
	152	+	// if the caller still wants this exact remote view.
	153	+	continue
	154	+	}
	155	+
	156	+	constructionCtx, cancel := context.WithCancel(ctx)
	157	+	construction := &remoteViewConstruction{done: make(chan struct{}), cancel: cancel}
	158	+	if d.remoteViewConstructions == nil {
	159	+	d.remoteViewConstructions = make(map[remoteViewKey]*remoteViewConstruction)
	160	+	}
	161	+	d.remoteViewConstructions[key] = construction
	162	+	d.mu.Unlock()
	163	+
	164	+	candidate, constructionErr := d.constructRemoteView(constructionCtx, target, size)
	165	+	if constructionErr == nil {
	166	+	constructionErr = ctx.Err()
	167	+	}
	168	+	cancel()
	169	+
	170	+	var winner *remoteView
	171	+	publishErr := constructionErr
	172	+	d.mu.Lock()
	173	+	if publishErr == nil {
	174	+	if d.closing {
	175	+	publishErr = errors.New("remote view: daemon is shutting down")
	176	+	} else if existing := d.remoteViewByKeyLocked(key); existing != nil {
	177	+	winner = existing
	178	+	} else if err := d.registerRemoteViewLocked(candidate); err != nil {
	179	+	publishErr = fmt.Errorf("remote view: publish candidate: %w", err)
	180	+	} else {
	181	+	winner = candidate
	182	+	// Starting the receiver while d.mu still protects publication keeps
	183	+	// shutdown from observing a registry-visible link with no worker
	184	+	// to join.
	185	+	d.startRemoteLink(candidate.link)
	186	+	}
	187	+	}
	188	+	construction.err = publishErr
	189	+	if d.remoteViewConstructions[key] == construction {
	190	+	delete(d.remoteViewConstructions, key)
	191	+	}
	192	+	close(construction.done)
	193	+	d.mu.Unlock()
	194	+
	195	+	if candidate != nil && candidate != winner {
	196	+	d.stopUnpublishedRemoteView(candidate)
	197	+	}
	198	+	if winner != nil {
	199	+	return winner, nil
	200	+	}
	201	+	return nil, publishErr
	202	+	}
	203	+	}
	204	+
	205	+	// constructRemoteView builds an unexposed candidate. Its caller owns a
	206	+	// constructor reservation but holds no daemon, session, or view lock.
	207	+	func (d Daemon) constructRemoteView(ctx context.Context, target domain.RemoteSessionTarget, size domain.Size) (_ remoteView, err error) {
	208	+	content := contentSize(size)
	209	+	if !content.Valid() {
	210	+	return nil, fmt.Errorf("remote view: invalid content size %dx%d", content.Cols, content.Rows)
	211	+	}
	212	+	key, err := remoteViewKeyForTarget(target)
	213	+	if err != nil {
	214	+	return nil, err
	215	+	}
	216	+	var clientID [16]byte
	217	+	if _, err := rand.Read(clientID[:]); err != nil {
	218	+	return nil, fmt.Errorf("remote view: generate client ID: %w", err)
	219	+	}
	220	+	root := context.Background()
	221	+	if d.serveCtx != nil {
	222	+	root = d.serveCtx
	223	+	}
	224	+	linkCtx, cancelLink := context.WithCancel(root)
	225	+	stopCallerCancellation := context.AfterFunc(ctx, cancelLink)
	226	+	defer stopCallerCancellation()
	227	+	view := &remoteView{
	228	+	key: key,
	229	+	displayOrigin: target.DisplayOrigin,
	230	+	screen: vt.NewScreen(content.Cols, content.Rows),
	231	+	linkGeneration: 1,
	232	+	}
	233	+	link := &remoteLink{
	234	+	view: view,
	235	+	generation: view.linkGeneration,
	236	+	target: target,
	237	+	clientID: clientID,
	238	+	cancel: cancelLink,
	239	+	done: make(chan struct{}),
	240	+	active: true,
	241	+	}
	242	+	view.link = link
	243	+	published := false
	244	+	defer func() {
	245	+	if !published {
	246	+	d.stopUnpublishedRemoteView(view)
	247	+	}
	248	+	}()
	249	+
	250	+	handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(linkCtx)
	251	+	defer finishHandshake()
	252	+	dialer, err := d.remoteDialerFactory.DialerForRemote(target.Endpoint, target.SessionName, d.remoteTransportMode, d.log)
	253	+	if err != nil {
	254	+	return nil, fmt.Errorf("remote view: select remote dialer: %w", err)
	255	+	}
	256	+	if dialer == nil {
	257	+	return nil, errors.New("remote view: remote dialer factory returned nil dialer")
	258	+	}
	259	+	transport, err := dialer.Dial(handshakeCtx)
	260	+	if err != nil {
	261	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: dial: %w", err))
	262	+	}
	263	+	if transport == nil {
	264	+	return nil, errors.New("remote view: remote dialer returned nil transport")
	265	+	}
	266	+	link.transport = transport
	267	+	stopWatch := watchHandshakeTransport(handshakeCtx, transport)
	268	+	defer stopWatch()
	269	+
	270	+	helloTarget := target
	271	+	hello := ports.Hello{
	272	+	Version: ports.ProtocolVersion,
	273	+	Intent: ports.IntentAttach,
	274	+	ClientID: clientID,
	275	+	Name: target.SessionName,
	276	+	Size: content,
	277	+	RenderMode: ports.RenderModeProxiedContent,
	278	+	MaxOutputInFlight: remoteLinkOutputWindow(transport),
	279	+	RemoteTarget: &helloTarget,
	280	+	EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
	281	+	}
	282	+	if err := link.send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)}); err != nil {
	283	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: send hello: %w", err))
	284	+	}
	285	+
	286	+	welcomeFrame, err := recvRemoteLinkHandshakeFrame(handshakeCtx, transport)
	287	+	if err != nil {
	288	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: receive welcome: %w", err))
	289	+	}
	290	+	if err := validateRemoteLinkWelcome(welcomeFrame, target); err != nil {
	291	+	return nil, err
	292	+	}
	293	+	metadataFrame, err := recvRemoteLinkHandshakeFrame(handshakeCtx, transport)
	294	+	if err != nil {
	295	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: receive metadata: %w", err))
	296	+	}
	297	+	metadata, err := validateInitialRemoteLinkMetadata(metadataFrame, target)
	298	+	if err != nil {
	299	+	return nil, err
	300	+	}
	301	+	if !view.applyRemoteMetadata(link, metadata, true) {
	302	+	return nil, errors.New("remote view: initial metadata became stale")
	303	+	}
	304	+	outputFrame, err := recvRemoteLinkHandshakeFrame(handshakeCtx, transport)
	305	+	if err != nil {
	306	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: receive initial output: %w", err))
	307	+	}
	308	+	if outputFrame.Type != ports.MsgOutput {
	309	+	return nil, fmt.Errorf("remote view: third server frame is %d, want Output", outputFrame.Type)
	310	+	}
	311	+	output, err := ports.UnmarshalOutput(outputFrame.Payload)
	312	+	if err != nil {
	313	+	return nil, fmt.Errorf("remote view: decode initial output: %w", err)
	314	+	}
	315	+	ack, reset, _, accepted := view.applyRemoteOutput(link, output)
	316	+	if !accepted \|\| reset \|\| output.New == 0 \|\| !output.Full {
	317	+	return nil, errors.New("remote view: initial output is not an authoritative reset")
	318	+	}
	319	+	if ack != 0 {
	320	+	ackPayload, err := ports.MarshalAck(ports.Ack{Epoch: output.Epoch, State: ack})
	321	+	if err != nil {
	322	+	return nil, fmt.Errorf("remote view: encode initial output acknowledgement: %w", err)
	323	+	}
	324	+	if err := link.send(ports.Frame{Type: ports.MsgAck, Payload: ackPayload}); err != nil {
	325	+	return nil, handshakeContextError(ctx, timedOut, fmt.Errorf("remote view: acknowledge initial output: %w", err))
	326	+	}
	327	+	}
	328	+	if err := handshakeCtx.Err(); err != nil {
	329	+	return nil, handshakeContextError(ctx, timedOut, err)
	330	+	}
	331	+	published = true
	332	+	return view, nil
	333	+	}
	334	+
	335	+	func remoteLinkOutputWindow(transport ports.Transport) uint8 {
	336	+	if _, datagram := transport.(ports.DatagramTransport); datagram {
	337	+	return 1
	338	+	}
	339	+	return maxUnackedOutputStates
	340	+	}
	341	+
	342	+	func recvRemoteLinkHandshakeFrame(ctx context.Context, transport ports.Transport) (ports.Frame, error) {
	343	+	var frame ports.Frame
	344	+	err := boundedHandshakeOperation(ctx, transport, func() error {
	345	+	var recvErr error
	346	+	frame, recvErr = transport.Recv()
	347	+	return recvErr
	348	+	})
	349	+	if err != nil {
	350	+	return ports.Frame{}, err
	351	+	}
	352	+	return frame, nil
	353	+	}
	354	+
	355	+	func validateRemoteLinkWelcome(frame ports.Frame, target domain.RemoteSessionTarget) error {
	356	+	if frame.Type == ports.MsgError {
	357	+	remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
	358	+	if err != nil {
	359	+	return fmt.Errorf("remote view: malformed remote error: %w", err)
	360	+	}
	361	+	return fmt.Errorf("remote view: remote rejected handshake (%d): %s", remoteErr.Code, remoteErr.Text)
	362	+	}
	363	+	if frame.Type != ports.MsgWelcome {
	364	+	return fmt.Errorf("remote view: first server frame is %d, want Welcome", frame.Type)
	365	+	}
	366	+	welcome, err := ports.UnmarshalWelcome(frame.Payload)
	367	+	if err != nil {
	368	+	return fmt.Errorf("remote view: decode welcome: %w", err)
	369	+	}
	370	+	if welcome.SessionID == "" \|\| welcome.SessionName != target.SessionName \|\| welcome.RenderMode != ports.RenderModeProxiedContent {
	371	+	return errors.New("remote view: welcome identity or render mode mismatch")
	372	+	}
	373	+	return nil
	374	+	}
	375	+
	376	+	func validateInitialRemoteLinkMetadata(frame ports.Frame, target domain.RemoteSessionTarget) (ports.SessionMeta, error) {
	377	+	if frame.Type != ports.MsgSessionMeta {
	378	+	return ports.SessionMeta{}, fmt.Errorf("remote view: second server frame is %d, want SessionMeta", frame.Type)
	379	+	}
	380	+	metadata, err := ports.UnmarshalSessionMeta(frame.Payload)
	381	+	if err != nil {
	382	+	return ports.SessionMeta{}, fmt.Errorf("remote view: decode initial metadata: %w", err)
	383	+	}
	384	+	if metadata.LifecycleID != target.LifecycleID \|\| metadata.SessionName != target.SessionName \|\| metadata.Revision != 1 {
	385	+	return ports.SessionMeta{}, errors.New("remote view: metadata identity mismatch")
	386	+	}
	387	+	if !target.Stopped && metadata.ActiveTabID != target.LiveTabID {
	388	+	return ports.SessionMeta{}, errors.New("remote view: metadata active tab mismatch")
	389	+	}
	390	+	return metadata, nil
	391	+	}
	392	+
	393	+	func cloneRemoteSessionMeta(metadata ports.SessionMeta) ports.SessionMeta {
	394	+	metadata.Tabs = append([]ports.SessionTabMeta(nil), metadata.Tabs...)
	395	+	return metadata
	396	+	}
	397	+
	398	+	// applyRemoteMetadata commits one validated update only for the exact current
	399	+	// link. The view lock protects content state only; no renderer or callback runs
	400	+	// before it is released.
	401	+	func (v remoteView) applyRemoteMetadata(link remoteLink, metadata ports.SessionMeta, initial bool) bool {
	402	+	if v == nil \|\| link == nil {
	403	+	return false
	404	+	}
	405	+	v.mu.Lock()
	406	+	defer v.mu.Unlock()
	407	+	if v.closed \|\| v.link != link \|\| v.linkGeneration != link.generation {
	408	+	return false
	409	+	}
	410	+	if initial {
	411	+	if v.metadata.Revision != 0 {
	412	+	return false
	413	+	}
	414	+	} else if metadata.Revision <= v.metadata.Revision {
	415	+	return false
	416	+	}
	417	+	v.metadata = cloneRemoteSessionMeta(metadata)
	418	+	return true
	419	+	}
	420	+
	421	+	// applyRemoteOutput validates and applies state-bearing ANSI to the persistent
	422	+	// private VT. It returns only immutable post-commit work; callers send ACK or
	423	+	// reset frames and repaint attachments after view.mu is released.
	424	+	func (v remoteView) applyRemoteOutput(link remoteLink, output ports.Output) (ack uint64, reset bool, attachments []*attachedClient, accepted bool) {
	425	+	if v == nil \|\| link == nil {
	426	+	return 0, false, nil, false
	427	+	}
	428	+	v.mu.Lock()
	429	+	defer v.mu.Unlock()
	430	+	if v.closed \|\| v.link != link \|\| v.linkGeneration != link.generation {
	431	+	return 0, false, nil, false
	432	+	}
	433	+	next, valid := v.output.next(output)
	434	+	if !valid \|\| (output.New != 0 && (v.screen == nil \|\| v.screen.Frame.Width != output.Size.Cols \|\| v.screen.Frame.Height != output.Size.Rows)) {
	435	+	if v.resetRequested {
	436	+	return 0, false, nil, false
	437	+	}
	438	+	v.resetRequested = true
	439	+	return 0, true, nil, false
	440	+	}
	441	+	if output.New == 0 {
	442	+	return 0, false, nil, true
	443	+	}
	444	+	v.output = next
	445	+	v.resetRequested = false
	446	+	if len(output.Data) != 0 {
	447	+	v.screen.Write(output.Data)
	448	+	}
	449	+	attachments = make([]*attachedClient, 0, len(v.attachments))
	450	+	for attachment := range v.attachments {
	451	+	attachments = append(attachments, attachment)
	452	+	}
	453	+	return output.New, false, attachments, true
	454	+	}
	455	+
	456	+	func (d Daemon) startRemoteLink(link remoteLink) {
	457	+	if d == nil \|\| link == nil \|\| link.transport == nil {
	458	+	return
	459	+	}
	460	+	link.startOnce.Do(func() {
	461	+	go d.runRemoteLink(link)
	462	+	})
	463	+	}
	464	+
	465	+	func (d Daemon) runRemoteLink(link remoteLink) {
	466	+	defer close(link.done)
	467	+	for {
	468	+	frame, err := link.transport.Recv()
	469	+	if err != nil {
	470	+	d.markRemoteLinkUnavailable(link)
	471	+	return
	472	+	}
	473	+	if err := d.handleRemoteLinkFrame(link, frame); err != nil {
	474	+	d.log.Warn("remote view link stopped", "endpoint", link.target.Endpoint, "session", link.target.SessionName, "err", err)
	475	+	d.markRemoteLinkUnavailable(link)
	476	+	return
	477	+	}
	478	+	}
	479	+	}
	480	+
	481	+	func (d Daemon) handleRemoteLinkFrame(link remoteLink, frame ports.Frame) error {
	482	+	switch frame.Type {
	483	+	case ports.MsgOutput:
	484	+	output, err := ports.UnmarshalOutput(frame.Payload)
	485	+	if err != nil {
	486	+	return fmt.Errorf("remote view: decode output: %w", err)
	487	+	}
	488	+	ack, reset, attachments, accepted := link.view.applyRemoteOutput(link, output)
	489	+	if reset {
	490	+	if err := link.send(ports.Frame{Type: ports.MsgOutputResetRequest, Payload: ports.MarshalOutputResetRequest(ports.OutputResetRequest{})}); err != nil {
	491	+	return fmt.Errorf("remote view: request output reset: %w", err)
	492	+	}
	493	+	return nil
	494	+	}
	495	+	if !accepted {
	496	+	return nil
	497	+	}
	498	+	if ack != 0 {
	499	+	ackPayload, err := ports.MarshalAck(ports.Ack{Epoch: output.Epoch, State: ack})
	500	+	if err != nil {
	501	+	return fmt.Errorf("remote view: encode output acknowledgement: %w", err)
	502	+	}
	503	+	if err := link.send(ports.Frame{Type: ports.MsgAck, Payload: ackPayload}); err != nil {
	504	+	return fmt.Errorf("remote view: acknowledge output: %w", err)
	505	+	}
	506	+	}
	507	+	for _, attachment := range attachments {
	508	+	token := attachmentOwnerToken(link.view, attachment, attachment.transport())
	509	+	if token.ac != nil {
	510	+	go d.paintRemoteView(link.view, attachment, false, token)
	511	+	}
	512	+	}
	513	+	return nil
	514	+	case ports.MsgSessionMeta:
	515	+	metadata, err := ports.UnmarshalSessionMeta(frame.Payload)
	516	+	if err != nil {
	517	+	return fmt.Errorf("remote view: decode metadata: %w", err)
	518	+	}
	519	+	if metadata.LifecycleID != link.target.LifecycleID \|\| metadata.SessionName != link.target.SessionName \|\| !link.view.applyRemoteMetadata(link, metadata, false) {
	520	+	return errors.New("remote view: invalid metadata update")
	521	+	}
	522	+	return nil
	523	+	case ports.MsgPing:
	524	+	if _, err := ports.UnmarshalPing(frame.Payload); err != nil {
	525	+	return fmt.Errorf("remote view: decode ping: %w", err)
	526	+	}
	527	+	return link.send(ports.Frame{Type: ports.MsgPong, Payload: ports.MarshalPong(ports.Pong{})})
	528	+	case ports.MsgPong:
	529	+	_, err := ports.UnmarshalPong(frame.Payload)
	530	+	return err
	531	+	case ports.MsgDetached:
	532	+	_, err := ports.UnmarshalDetached(frame.Payload)
	533	+	if err != nil {
	534	+	return fmt.Errorf("remote view: decode detached: %w", err)
	535	+	}
	536	+	return errors.New("remote view: remote session detached")
	537	+	case ports.MsgError:
	538	+	remoteErr, err := ports.UnmarshalErrorMsg(frame.Payload)
	539	+	if err != nil {
	540	+	return fmt.Errorf("remote view: malformed remote error: %w", err)
	541	+	}
	542	+	return fmt.Errorf("remote view: remote error (%d): %s", remoteErr.Code, remoteErr.Text)
	543	+	default:
	544	+	return fmt.Errorf("remote view: unexpected server frame type %d", frame.Type)
	545	+	}
	546	+	}
	547	+
	548	+	// send serializes the complete outbound remote stream. It captures the exact
	549	+	// current transport under view.mu, releases every architecture lock, then lets
	550	+	// Close interrupt a blocked Send concurrently.
	551	+	func (link *remoteLink) send(frame ports.Frame) error {
	552	+	if link == nil \|\| link.view == nil {
	553	+	return errRemoteViewUnavailable
	554	+	}
	555	+	link.sendMu.Lock()
	556	+	view := link.view
	557	+	view.mu.Lock()
	558	+	current := !view.closed && view.link == link && view.linkGeneration == link.generation && link.active
	559	+	transport := link.transport
	560	+	view.mu.Unlock()
	561	+	if !current \|\| transport == nil {
	562	+	link.sendMu.Unlock()
	563	+	return errRemoteViewStale
	564	+	}
	565	+	err := transport.Send(frame)
	566	+	link.sendMu.Unlock()
	567	+	if err != nil {
	568	+	link.cancel()
	569	+	_ = transport.Close()
	570	+	}
	571	+	return err
	572	+	}
	573	+
	574	+	// handleRemoteViewInput forwards the raw local-terminal bytes and their
	575	+	// original sequence through the exact current remote link. No local key,
	576	+	// overlay, mouse, PTY, or persistence path observes those bytes.
	577	+	func (d Daemon) handleRemoteViewInput(view remoteView, token attachmentConnectionToken, inputSeq uint64, data []byte) {
	578	+	if view == nil \|\| len(data) == 0 \|\| !token.attachmentEffectCurrent() {
	579	+	return
	580	+	}
	581	+	payload := ports.MarshalInput(ports.Input{InputSeq: inputSeq, Data: append([]byte(nil), data...)})
	582	+	if err := d.sendRemoteViewFrame(view, ports.Frame{Type: ports.MsgInput, Payload: payload}); err != nil {
	583	+	d.log.Warn("forwarding remote view input failed", "endpoint", view.key.endpoint, "session", view.key.sessionName, "err", err)
	584	+	}
	585	+	}
	586	+
	587	+	// sendRemoteViewFrame snapshots the exact link while holding only view.mu;
	588	+	// remoteLink.send performs the generation check and serialized transport I/O.
	589	+	func (d Daemon) sendRemoteViewFrame(view remoteView, frame ports.Frame) error {
	590	+	if view == nil {
	591	+	return errRemoteViewUnavailable
	592	+	}
	593	+	view.mu.Lock()
	594	+	link := view.link
	595	+	view.mu.Unlock()
	596	+	if link == nil {
	597	+	return errRemoteViewUnavailable
	598	+	}
	599	+	return link.send(frame)
	600	+	}
	601	+
	602	+	func (d Daemon) markRemoteLinkUnavailable(link remoteLink) {
	603	+	if link == nil \|\| link.view == nil {
	604	+	return
	605	+	}
	606	+	view := link.view
	607	+	view.mu.Lock()
	608	+	current := !view.closed && view.link == link && view.linkGeneration == link.generation
	609	+	if current {
	610	+	link.active = false
	611	+	}
	612	+	view.mu.Unlock()
	613	+	if current {
	614	+	link.cancel()
	615	+	_ = link.transport.Close()
	616	+	}
	617	+	}
	618	+
	619	+	// stopUnpublishedRemoteView is used for candidate failure and lost-election
	620	+	// cleanup. It never touches the daemon registry and closes only the candidate
	621	+	// transport generation after view.mu is released.
	622	+	func (d Daemon) stopUnpublishedRemoteView(view remoteView) {
	623	+	if view == nil {
	624	+	return
	625	+	}
	626	+	view.mu.Lock()
	627	+	view.closed = true
	628	+	link := view.link
	629	+	view.link = nil
	630	+	view.linkGeneration++
	631	+	view.mu.Unlock()
	632	+	if link != nil {
	633	+	link.cancel()
	634	+	if link.transport != nil {
	635	+	_ = link.transport.Close()
	636	+	}
	637	+	}
	638	+	}
	639	+
	640	+	// stopRemoteViewLink interrupts only the exact currently installed remote
	641	+	// transport. It is called after registry publication has been revoked, so it
	642	+	// cannot affect a successor view or link generation. The returned link has a
	643	+	// started worker, which the terminal shutdown path joins after Close.
	644	+	func (d Daemon) stopRemoteViewLink(view remoteView) *remoteLink {
	645	+	if view == nil {
	646	+	return nil
	647	+	}
	648	+	view.mu.Lock()
	649	+	link := view.link
	650	+	if link != nil {
	651	+	view.link = nil
	652	+	view.linkGeneration++
	653	+	link.active = false
	654	+	}
	655	+	view.mu.Unlock()
	656	+	if link != nil {
	657	+	link.cancel()
	658	+	if link.transport != nil {
	659	+	_ = link.transport.Close()
	660	+	}
	661	+	}
	662	+	return link
	663	+	}

A internal/usecase/daemon/remote_link_test.go +383 -0

@@ -0,0 +1,383 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"io"
	6	+	"log/slog"
	7	+	"sync"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	"github.com/stretchr/testify/require"
	14	+	)
	15	+
	16	+	type remoteLinkTestReceive struct {
	17	+	frame ports.Frame
	18	+	err error
	19	+	}
	20	+
	21	+	type remoteLinkTestTransport struct {
	22	+	recv chan remoteLinkTestReceive
	23	+	sent chan ports.Frame
	24	+	closed chan struct{}
	25	+	closeOnce sync.Once
	26	+	}
	27	+
	28	+	func newRemoteLinkTestTransport() *remoteLinkTestTransport {
	29	+	return &remoteLinkTestTransport{
	30	+	recv: make(chan remoteLinkTestReceive, 8),
	31	+	sent: make(chan ports.Frame, 8),
	32	+	closed: make(chan struct{}),
	33	+	}
	34	+	}
	35	+
	36	+	func (t *remoteLinkTestTransport) Send(frame ports.Frame) error {
	37	+	select {
	38	+	case <-t.closed:
	39	+	return io.EOF
	40	+	case t.sent <- frame:
	41	+	return nil
	42	+	}
	43	+	}
	44	+
	45	+	func (t *remoteLinkTestTransport) Recv() (ports.Frame, error) {
	46	+	select {
	47	+	case <-t.closed:
	48	+	return ports.Frame{}, io.EOF
	49	+	case received := <-t.recv:
	50	+	return received.frame, received.err
	51	+	}
	52	+	}
	53	+
	54	+	func (t *remoteLinkTestTransport) Close() error {
	55	+	t.closeOnce.Do(func() { close(t.closed) })
	56	+	return nil
	57	+	}
	58	+
	59	+	type remoteLinkTestDialer struct {
	60	+	transport ports.Transport
	61	+	started chan<- struct{}
	62	+	release <-chan struct{}
	63	+	once sync.Once
	64	+	}
	65	+
	66	+	func (d *remoteLinkTestDialer) Dial(ctx context.Context) (ports.Transport, error) {
	67	+	if d.started != nil {
	68	+	d.once.Do(func() { close(d.started) })
	69	+	}
	70	+	if d.release != nil {
	71	+	select {
	72	+	case <-ctx.Done():
	73	+	return nil, ctx.Err()
	74	+	case <-d.release:
	75	+	}
	76	+	}
	77	+	return d.transport, nil
	78	+	}
	79	+
	80	+	type remoteLinkTestFactory struct {
	81	+	dialer ports.Dialer
	82	+	calls atomic.Int32
	83	+	}
	84	+
	85	+	func (f remoteLinkTestFactory) DialerForRemote(_ string, _ string, _ ports.RemoteTransportMode, _ slog.Logger) (ports.Dialer, error) {
	86	+	f.calls.Add(1)
	87	+	return f.dialer, nil
	88	+	}
	89	+
	90	+	func remoteLinkTestTarget() domain.RemoteSessionTarget {
	91	+	return domain.RemoteSessionTarget{
	92	+	Endpoint: "remote.example",
	93	+	DisplayOrigin: "remote.example",
	94	+	LifecycleID: domain.SessionLifecycleID{1},
	95	+	SessionName: "remote",
	96	+	LiveTabID: "tab-1",
	97	+	}
	98	+	}
	99	+
	100	+	func enqueueRemoteLinkHandshake(t testing.T, transport remoteLinkTestTransport, target domain.RemoteSessionTarget, size domain.Size) {
	101	+	t.Helper()
	102	+	content := contentSize(size)
	103	+	metaPayload, err := ports.MarshalSessionMeta(ports.SessionMeta{
	104	+	LifecycleID: target.LifecycleID,
	105	+	Revision: 1,
	106	+	SessionName: target.SessionName,
	107	+	ActiveTabID: target.LiveTabID,
	108	+	Tabs: []ports.SessionTabMeta{{ID: target.LiveTabID, Name: "main"}},
	109	+	})
	110	+	require.NoError(t, err)
	111	+	outputPayload, err := ports.MarshalOutput(ports.Output{
	112	+	Epoch: 1, New: 1, Size: content, Full: true, Data: []byte("remote marker"),
	113	+	})
	114	+	require.NoError(t, err)
	115	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{
	116	+	SessionID: "remote-id", SessionName: target.SessionName, RenderMode: ports.RenderModeProxiedContent,
	117	+	Capabilities: ports.CapabilityResume,
	118	+	})}}
	119	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgSessionMeta, Payload: metaPayload}}
	120	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: outputPayload}}
	121	+	}
	122	+
	123	+	func TestOpenRemoteViewPublishesHandshakeReadyCandidate(t *testing.T) {
	124	+	d := newTestDaemon(t, nil, stubClock{})
	125	+	target := remoteLinkTestTarget()
	126	+	size := domain.Size{Cols: 80, Rows: 24}
	127	+	transport := newRemoteLinkTestTransport()
	128	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	129	+	factory := &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	130	+	d.remoteDialerFactory = factory
	131	+	d.remoteTransportMode = ports.RemoteTransportUDP
	132	+
	133	+	view, err := d.openRemoteView(context.Background(), target, size)
	134	+	require.NoError(t, err)
	135	+	require.NotNil(t, view)
	136	+	require.Equal(t, int32(1), factory.calls.Load())
	137	+
	138	+	helloFrame := awaitFrame(t, transport.sent, ports.MsgHello)
	139	+	hello, err := ports.UnmarshalHello(helloFrame.Payload)
	140	+	require.NoError(t, err)
	141	+	require.Equal(t, ports.IntentAttach, hello.Intent)
	142	+	require.Equal(t, ports.RenderModeProxiedContent, hello.RenderMode)
	143	+	require.Equal(t, contentSize(size), hello.Size)
	144	+	require.Equal(t, ports.EnvironmentPolicyDaemonOwned, hello.EnvironmentPolicy)
	145	+	require.Equal(t, &target, hello.RemoteTarget)
	146	+
	147	+	ackFrame := awaitFrame(t, transport.sent, ports.MsgAck)
	148	+	ack, err := ports.UnmarshalAck(ackFrame.Payload)
	149	+	require.NoError(t, err)
	150	+	require.Equal(t, ports.Ack{Epoch: 1, State: 1}, ack)
	151	+
	152	+	d.mu.Lock()
	153	+	require.Same(t, view, d.remoteViewByKeyLocked(view.key))
	154	+	d.mu.Unlock()
	155	+	view.mu.Lock()
	156	+	require.Contains(t, screenLineText(view.screen, 0), "remote marker")
	157	+	require.Equal(t, target.LifecycleID, view.metadata.LifecycleID)
	158	+	require.Equal(t, target.LiveTabID, view.metadata.ActiveTabID)
	159	+	view.mu.Unlock()
	160	+
	161	+	require.NoError(t, transport.Close())
	162	+	}
	163	+
	164	+	func TestOpenRemoteViewCancellationClosesUnpublishedTransport(t *testing.T) {
	165	+	d := newTestDaemon(t, nil, stubClock{})
	166	+	target := remoteLinkTestTarget()
	167	+	size := domain.Size{Cols: 80, Rows: 24}
	168	+	transport := newRemoteLinkTestTransport()
	169	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	170	+
	171	+	ctx, cancel := context.WithCancel(context.Background())
	172	+	type result struct {
	173	+	view *remoteView
	174	+	err error
	175	+	}
	176	+	resultCh := make(chan result, 1)
	177	+	go func() {
	178	+	view, err := d.openRemoteView(ctx, target, size)
	179	+	resultCh <- result{view: view, err: err}
	180	+	}()
	181	+	awaitFrame(t, transport.sent, ports.MsgHello)
	182	+	cancel()
	183	+
	184	+	got := <-resultCh
	185	+	require.Nil(t, got.view)
	186	+	require.ErrorIs(t, got.err, context.Canceled)
	187	+	select {
	188	+	case <-transport.closed:
	189	+	default:
	190	+	t.Fatal("canceled candidate transport was not closed")
	191	+	}
	192	+	d.mu.Lock()
	193	+	require.Empty(t, d.remoteViews)
	194	+	require.Empty(t, d.remoteViewConstructions)
	195	+	d.mu.Unlock()
	196	+	}
	197	+
	198	+	func TestShutdownAllCancelsUnpublishedRemoteViewConstruction(t *testing.T) {
	199	+	d := newTestDaemon(t, nil, stubClock{})
	200	+	target := remoteLinkTestTarget()
	201	+	size := domain.Size{Cols: 80, Rows: 24}
	202	+	transport := newRemoteLinkTestTransport()
	203	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	204	+
	205	+	type result struct {
	206	+	view *remoteView
	207	+	err error
	208	+	}
	209	+	resultCh := make(chan result, 1)
	210	+	go func() {
	211	+	view, err := d.openRemoteView(context.Background(), target, size)
	212	+	resultCh <- result{view: view, err: err}
	213	+	}()
	214	+	awaitFrame(t, transport.sent, ports.MsgHello)
	215	+
	216	+	d.shutdownAll(ports.ReasonServerShutdown)
	217	+	got := awaitTestValue(t, resultCh, "remote view construction did not stop at shutdown")
	218	+	require.Nil(t, got.view)
	219	+	require.ErrorIs(t, got.err, context.Canceled)
	220	+	select {
	221	+	case <-transport.closed:
	222	+	default:
	223	+	t.Fatal("shutdown did not close the unpublished remote transport")
	224	+	}
	225	+	d.mu.Lock()
	226	+	require.Empty(t, d.remoteViews)
	227	+	require.Empty(t, d.remoteViewConstructions)
	228	+	d.mu.Unlock()
	229	+	}
	230	+
	231	+	func TestRemoteLinkAppliesAcceptedOutputAndRequestsOneResetForInvalidState(t *testing.T) {
	232	+	d := newTestDaemon(t, nil, stubClock{})
	233	+	target := remoteLinkTestTarget()
	234	+	size := domain.Size{Cols: 80, Rows: 24}
	235	+	transport := newRemoteLinkTestTransport()
	236	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	237	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	238	+
	239	+	view, err := d.openRemoteView(context.Background(), target, size)
	240	+	require.NoError(t, err)
	241	+	awaitFrame(t, transport.sent, ports.MsgHello)
	242	+	awaitFrame(t, transport.sent, ports.MsgAck)
	243	+
	244	+	content := contentSize(size)
	245	+	incrementalPayload, err := ports.MarshalOutput(ports.Output{
	246	+	Epoch: 1, Base: 1, New: 2, Size: content, Data: []byte(" updated"),
	247	+	})
	248	+	require.NoError(t, err)
	249	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: incrementalPayload}}
	250	+	ackFrame := awaitFrame(t, transport.sent, ports.MsgAck)
	251	+	ack, err := ports.UnmarshalAck(ackFrame.Payload)
	252	+	require.NoError(t, err)
	253	+	require.Equal(t, ports.Ack{Epoch: 1, State: 2}, ack)
	254	+	view.mu.Lock()
	255	+	require.Contains(t, screenLineText(view.screen, 0), "remote marker updated")
	256	+	view.mu.Unlock()
	257	+
	258	+	invalidPayload, err := ports.MarshalOutput(ports.Output{
	259	+	Epoch: 1, Base: 9, New: 10, Size: content, Data: []byte("must not apply"),
	260	+	})
	261	+	require.NoError(t, err)
	262	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: invalidPayload}}
	263	+	awaitFrame(t, transport.sent, ports.MsgOutputResetRequest)
	264	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: invalidPayload}}
	265	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})}}
	266	+	afterInvalid := awaitTestValue(t, transport.sent, "remote link did not process duplicate invalid output")
	267	+	require.Equal(t, ports.MsgPong, afterInvalid.Type, "duplicate invalid state must not trigger another reset")
	268	+
	269	+	view.mu.Lock()
	270	+	require.NotContains(t, screenLineText(view.screen, 0), "must not apply")
	271	+	view.mu.Unlock()
	272	+	require.NoError(t, transport.Close())
	273	+	}
	274	+
	275	+	func TestShutdownAllInterruptsPublishedRemoteViewLink(t *testing.T) {
	276	+	d := newTestDaemon(t, nil, stubClock{})
	277	+	target := remoteLinkTestTarget()
	278	+	size := domain.Size{Cols: 80, Rows: 24}
	279	+	transport := newRemoteLinkTestTransport()
	280	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	281	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	282	+
	283	+	view, err := d.openRemoteView(context.Background(), target, size)
	284	+	require.NoError(t, err)
	285	+	awaitFrame(t, transport.sent, ports.MsgHello)
	286	+	awaitFrame(t, transport.sent, ports.MsgAck)
	287	+	view.mu.Lock()
	288	+	link := view.link
	289	+	view.mu.Unlock()
	290	+	require.NotNil(t, link)
	291	+
	292	+	d.shutdownAll(ports.ReasonServerShutdown)
	293	+	select {
	294	+	case <-transport.closed:
	295	+	default:
	296	+	t.Fatal("shutdown did not close the exact remote transport")
	297	+	}
	298	+	awaitTestCompletion(t, link.done, "remote link worker did not stop after shutdown")
	299	+	d.mu.Lock()
	300	+	require.Empty(t, d.remoteViews)
	301	+	d.mu.Unlock()
	302	+	}
	303	+
	304	+	func TestRemoteLinkForwardsRemoteViewInputAndContentResize(t *testing.T) {
	305	+	d := newTestDaemon(t, nil, stubClock{})
	306	+	target := remoteLinkTestTarget()
	307	+	size := domain.Size{Cols: 80, Rows: 24}
	308	+	transport := newRemoteLinkTestTransport()
	309	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	310	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	311	+
	312	+	view, err := d.openRemoteView(context.Background(), target, size)
	313	+	require.NoError(t, err)
	314	+	awaitFrame(t, transport.sent, ports.MsgHello)
	315	+	awaitFrame(t, transport.sent, ports.MsgAck)
	316	+
	317	+	clientTransport, _ := newCapturingTransport(t)
	318	+	ac := &attachedClient{tr: clientTransport, output: newOutputStateStream(), size: size}
	319	+	ac.initOverlays()
	320	+	ac.setAttachmentOwner(view)
	321	+	view.mu.Lock()
	322	+	require.True(t, view.registerAttachmentLocked(ac))
	323	+	view.mu.Unlock()
	324	+	token := attachmentOwnerToken(view, ac, clientTransport)
	325	+	require.True(t, token.attachmentCurrent())
	326	+
	327	+	d.handleSequencedInputForAttachment(token, 42, []byte("remote input"))
	328	+	inputFrame := awaitFrame(t, transport.sent, ports.MsgInput)
	329	+	input, err := ports.UnmarshalInput(inputFrame.Payload)
	330	+	require.NoError(t, err)
	331	+	require.Equal(t, ports.Input{InputSeq: 42, Data: []byte("remote input")}, input)
	332	+
	333	+	require.True(t, d.resizeAttachmentForLease(token, domain.Size{Cols: 100, Rows: 30}))
	334	+	resizeFrame := awaitFrame(t, transport.sent, ports.MsgResize)
	335	+	resize, err := ports.UnmarshalResize(resizeFrame.Payload)
	336	+	require.NoError(t, err)
	337	+	require.Equal(t, domain.Size{Cols: 100, Rows: 28}, resize.Size)
	338	+
	339	+	require.NoError(t, transport.Close())
	340	+	}
	341	+
	342	+	func TestOpenRemoteViewElectsOneConstructorAndCanceledWaiterDoesNotPoisonWinner(t *testing.T) {
	343	+	d := newTestDaemon(t, nil, stubClock{})
	344	+	target := remoteLinkTestTarget()
	345	+	size := domain.Size{Cols: 80, Rows: 24}
	346	+	transport := newRemoteLinkTestTransport()
	347	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	348	+	started := make(chan struct{})
	349	+	release := make(chan struct{})
	350	+	factory := &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport, started: started, release: release}}
	351	+	d.remoteDialerFactory = factory
	352	+
	353	+	type result struct {
	354	+	view *remoteView
	355	+	err error
	356	+	}
	357	+	ownerResult := make(chan result, 1)
	358	+	go func() {
	359	+	view, err := d.openRemoteView(context.Background(), target, size)
	360	+	ownerResult <- result{view: view, err: err}
	361	+	}()
	362	+	<-started
	363	+
	364	+	waiterCtx, cancelWaiter := context.WithCancel(context.Background())
	365	+	cancelWaiter()
	366	+	view, err := d.openRemoteView(waiterCtx, target, size)
	367	+	require.Nil(t, view)
	368	+	require.ErrorIs(t, err, context.Canceled)
	369	+	require.Equal(t, int32(1), factory.calls.Load())
	370	+
	371	+	close(release)
	372	+	owner := <-ownerResult
	373	+	require.NoError(t, owner.err)
	374	+	require.NotNil(t, owner.view)
	375	+	require.Equal(t, int32(1), factory.calls.Load())
	376	+
	377	+	warm, err := d.openRemoteView(context.Background(), target, size)
	378	+	require.NoError(t, err)
	379	+	require.Same(t, owner.view, warm)
	380	+	require.Equal(t, int32(1), factory.calls.Load())
	381	+
	382	+	require.NoError(t, transport.Close())
	383	+	}

@@ -0,0 +1,383 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"io"
	6	+	"log/slog"
	7	+	"sync"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	"github.com/stretchr/testify/require"
	14	+	)
	15	+
	16	+	type remoteLinkTestReceive struct {
	17	+	frame ports.Frame
	18	+	err error
	19	+	}
	20	+
	21	+	type remoteLinkTestTransport struct {
	22	+	recv chan remoteLinkTestReceive
	23	+	sent chan ports.Frame
	24	+	closed chan struct{}
	25	+	closeOnce sync.Once
	26	+	}
	27	+
	28	+	func newRemoteLinkTestTransport() *remoteLinkTestTransport {
	29	+	return &remoteLinkTestTransport{
	30	+	recv: make(chan remoteLinkTestReceive, 8),
	31	+	sent: make(chan ports.Frame, 8),
	32	+	closed: make(chan struct{}),
	33	+	}
	34	+	}
	35	+
	36	+	func (t *remoteLinkTestTransport) Send(frame ports.Frame) error {
	37	+	select {
	38	+	case <-t.closed:
	39	+	return io.EOF
	40	+	case t.sent <- frame:
	41	+	return nil
	42	+	}
	43	+	}
	44	+
	45	+	func (t *remoteLinkTestTransport) Recv() (ports.Frame, error) {
	46	+	select {
	47	+	case <-t.closed:
	48	+	return ports.Frame{}, io.EOF
	49	+	case received := <-t.recv:
	50	+	return received.frame, received.err
	51	+	}
	52	+	}
	53	+
	54	+	func (t *remoteLinkTestTransport) Close() error {
	55	+	t.closeOnce.Do(func() { close(t.closed) })
	56	+	return nil
	57	+	}
	58	+
	59	+	type remoteLinkTestDialer struct {
	60	+	transport ports.Transport
	61	+	started chan<- struct{}
	62	+	release <-chan struct{}
	63	+	once sync.Once
	64	+	}
	65	+
	66	+	func (d *remoteLinkTestDialer) Dial(ctx context.Context) (ports.Transport, error) {
	67	+	if d.started != nil {
	68	+	d.once.Do(func() { close(d.started) })
	69	+	}
	70	+	if d.release != nil {
	71	+	select {
	72	+	case <-ctx.Done():
	73	+	return nil, ctx.Err()
	74	+	case <-d.release:
	75	+	}
	76	+	}
	77	+	return d.transport, nil
	78	+	}
	79	+
	80	+	type remoteLinkTestFactory struct {
	81	+	dialer ports.Dialer
	82	+	calls atomic.Int32
	83	+	}
	84	+
	85	+	func (f remoteLinkTestFactory) DialerForRemote(_ string, _ string, _ ports.RemoteTransportMode, _ slog.Logger) (ports.Dialer, error) {
	86	+	f.calls.Add(1)
	87	+	return f.dialer, nil
	88	+	}
	89	+
	90	+	func remoteLinkTestTarget() domain.RemoteSessionTarget {
	91	+	return domain.RemoteSessionTarget{
	92	+	Endpoint: "remote.example",
	93	+	DisplayOrigin: "remote.example",
	94	+	LifecycleID: domain.SessionLifecycleID{1},
	95	+	SessionName: "remote",
	96	+	LiveTabID: "tab-1",
	97	+	}
	98	+	}
	99	+
	100	+	func enqueueRemoteLinkHandshake(t testing.T, transport remoteLinkTestTransport, target domain.RemoteSessionTarget, size domain.Size) {
	101	+	t.Helper()
	102	+	content := contentSize(size)
	103	+	metaPayload, err := ports.MarshalSessionMeta(ports.SessionMeta{
	104	+	LifecycleID: target.LifecycleID,
	105	+	Revision: 1,
	106	+	SessionName: target.SessionName,
	107	+	ActiveTabID: target.LiveTabID,
	108	+	Tabs: []ports.SessionTabMeta{{ID: target.LiveTabID, Name: "main"}},
	109	+	})
	110	+	require.NoError(t, err)
	111	+	outputPayload, err := ports.MarshalOutput(ports.Output{
	112	+	Epoch: 1, New: 1, Size: content, Full: true, Data: []byte("remote marker"),
	113	+	})
	114	+	require.NoError(t, err)
	115	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgWelcome, Payload: ports.MarshalWelcome(ports.Welcome{
	116	+	SessionID: "remote-id", SessionName: target.SessionName, RenderMode: ports.RenderModeProxiedContent,
	117	+	Capabilities: ports.CapabilityResume,
	118	+	})}}
	119	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgSessionMeta, Payload: metaPayload}}
	120	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: outputPayload}}
	121	+	}
	122	+
	123	+	func TestOpenRemoteViewPublishesHandshakeReadyCandidate(t *testing.T) {
	124	+	d := newTestDaemon(t, nil, stubClock{})
	125	+	target := remoteLinkTestTarget()
	126	+	size := domain.Size{Cols: 80, Rows: 24}
	127	+	transport := newRemoteLinkTestTransport()
	128	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	129	+	factory := &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	130	+	d.remoteDialerFactory = factory
	131	+	d.remoteTransportMode = ports.RemoteTransportUDP
	132	+
	133	+	view, err := d.openRemoteView(context.Background(), target, size)
	134	+	require.NoError(t, err)
	135	+	require.NotNil(t, view)
	136	+	require.Equal(t, int32(1), factory.calls.Load())
	137	+
	138	+	helloFrame := awaitFrame(t, transport.sent, ports.MsgHello)
	139	+	hello, err := ports.UnmarshalHello(helloFrame.Payload)
	140	+	require.NoError(t, err)
	141	+	require.Equal(t, ports.IntentAttach, hello.Intent)
	142	+	require.Equal(t, ports.RenderModeProxiedContent, hello.RenderMode)
	143	+	require.Equal(t, contentSize(size), hello.Size)
	144	+	require.Equal(t, ports.EnvironmentPolicyDaemonOwned, hello.EnvironmentPolicy)
	145	+	require.Equal(t, &target, hello.RemoteTarget)
	146	+
	147	+	ackFrame := awaitFrame(t, transport.sent, ports.MsgAck)
	148	+	ack, err := ports.UnmarshalAck(ackFrame.Payload)
	149	+	require.NoError(t, err)
	150	+	require.Equal(t, ports.Ack{Epoch: 1, State: 1}, ack)
	151	+
	152	+	d.mu.Lock()
	153	+	require.Same(t, view, d.remoteViewByKeyLocked(view.key))
	154	+	d.mu.Unlock()
	155	+	view.mu.Lock()
	156	+	require.Contains(t, screenLineText(view.screen, 0), "remote marker")
	157	+	require.Equal(t, target.LifecycleID, view.metadata.LifecycleID)
	158	+	require.Equal(t, target.LiveTabID, view.metadata.ActiveTabID)
	159	+	view.mu.Unlock()
	160	+
	161	+	require.NoError(t, transport.Close())
	162	+	}
	163	+
	164	+	func TestOpenRemoteViewCancellationClosesUnpublishedTransport(t *testing.T) {
	165	+	d := newTestDaemon(t, nil, stubClock{})
	166	+	target := remoteLinkTestTarget()
	167	+	size := domain.Size{Cols: 80, Rows: 24}
	168	+	transport := newRemoteLinkTestTransport()
	169	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	170	+
	171	+	ctx, cancel := context.WithCancel(context.Background())
	172	+	type result struct {
	173	+	view *remoteView
	174	+	err error
	175	+	}
	176	+	resultCh := make(chan result, 1)
	177	+	go func() {
	178	+	view, err := d.openRemoteView(ctx, target, size)
	179	+	resultCh <- result{view: view, err: err}
	180	+	}()
	181	+	awaitFrame(t, transport.sent, ports.MsgHello)
	182	+	cancel()
	183	+
	184	+	got := <-resultCh
	185	+	require.Nil(t, got.view)
	186	+	require.ErrorIs(t, got.err, context.Canceled)
	187	+	select {
	188	+	case <-transport.closed:
	189	+	default:
	190	+	t.Fatal("canceled candidate transport was not closed")
	191	+	}
	192	+	d.mu.Lock()
	193	+	require.Empty(t, d.remoteViews)
	194	+	require.Empty(t, d.remoteViewConstructions)
	195	+	d.mu.Unlock()
	196	+	}
	197	+
	198	+	func TestShutdownAllCancelsUnpublishedRemoteViewConstruction(t *testing.T) {
	199	+	d := newTestDaemon(t, nil, stubClock{})
	200	+	target := remoteLinkTestTarget()
	201	+	size := domain.Size{Cols: 80, Rows: 24}
	202	+	transport := newRemoteLinkTestTransport()
	203	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	204	+
	205	+	type result struct {
	206	+	view *remoteView
	207	+	err error
	208	+	}
	209	+	resultCh := make(chan result, 1)
	210	+	go func() {
	211	+	view, err := d.openRemoteView(context.Background(), target, size)
	212	+	resultCh <- result{view: view, err: err}
	213	+	}()
	214	+	awaitFrame(t, transport.sent, ports.MsgHello)
	215	+
	216	+	d.shutdownAll(ports.ReasonServerShutdown)
	217	+	got := awaitTestValue(t, resultCh, "remote view construction did not stop at shutdown")
	218	+	require.Nil(t, got.view)
	219	+	require.ErrorIs(t, got.err, context.Canceled)
	220	+	select {
	221	+	case <-transport.closed:
	222	+	default:
	223	+	t.Fatal("shutdown did not close the unpublished remote transport")
	224	+	}
	225	+	d.mu.Lock()
	226	+	require.Empty(t, d.remoteViews)
	227	+	require.Empty(t, d.remoteViewConstructions)
	228	+	d.mu.Unlock()
	229	+	}
	230	+
	231	+	func TestRemoteLinkAppliesAcceptedOutputAndRequestsOneResetForInvalidState(t *testing.T) {
	232	+	d := newTestDaemon(t, nil, stubClock{})
	233	+	target := remoteLinkTestTarget()
	234	+	size := domain.Size{Cols: 80, Rows: 24}
	235	+	transport := newRemoteLinkTestTransport()
	236	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	237	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	238	+
	239	+	view, err := d.openRemoteView(context.Background(), target, size)
	240	+	require.NoError(t, err)
	241	+	awaitFrame(t, transport.sent, ports.MsgHello)
	242	+	awaitFrame(t, transport.sent, ports.MsgAck)
	243	+
	244	+	content := contentSize(size)
	245	+	incrementalPayload, err := ports.MarshalOutput(ports.Output{
	246	+	Epoch: 1, Base: 1, New: 2, Size: content, Data: []byte(" updated"),
	247	+	})
	248	+	require.NoError(t, err)
	249	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: incrementalPayload}}
	250	+	ackFrame := awaitFrame(t, transport.sent, ports.MsgAck)
	251	+	ack, err := ports.UnmarshalAck(ackFrame.Payload)
	252	+	require.NoError(t, err)
	253	+	require.Equal(t, ports.Ack{Epoch: 1, State: 2}, ack)
	254	+	view.mu.Lock()
	255	+	require.Contains(t, screenLineText(view.screen, 0), "remote marker updated")
	256	+	view.mu.Unlock()
	257	+
	258	+	invalidPayload, err := ports.MarshalOutput(ports.Output{
	259	+	Epoch: 1, Base: 9, New: 10, Size: content, Data: []byte("must not apply"),
	260	+	})
	261	+	require.NoError(t, err)
	262	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: invalidPayload}}
	263	+	awaitFrame(t, transport.sent, ports.MsgOutputResetRequest)
	264	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgOutput, Payload: invalidPayload}}
	265	+	transport.recv <- remoteLinkTestReceive{frame: ports.Frame{Type: ports.MsgPing, Payload: ports.MarshalPing(ports.Ping{})}}
	266	+	afterInvalid := awaitTestValue(t, transport.sent, "remote link did not process duplicate invalid output")
	267	+	require.Equal(t, ports.MsgPong, afterInvalid.Type, "duplicate invalid state must not trigger another reset")
	268	+
	269	+	view.mu.Lock()
	270	+	require.NotContains(t, screenLineText(view.screen, 0), "must not apply")
	271	+	view.mu.Unlock()
	272	+	require.NoError(t, transport.Close())
	273	+	}
	274	+
	275	+	func TestShutdownAllInterruptsPublishedRemoteViewLink(t *testing.T) {
	276	+	d := newTestDaemon(t, nil, stubClock{})
	277	+	target := remoteLinkTestTarget()
	278	+	size := domain.Size{Cols: 80, Rows: 24}
	279	+	transport := newRemoteLinkTestTransport()
	280	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	281	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	282	+
	283	+	view, err := d.openRemoteView(context.Background(), target, size)
	284	+	require.NoError(t, err)
	285	+	awaitFrame(t, transport.sent, ports.MsgHello)
	286	+	awaitFrame(t, transport.sent, ports.MsgAck)
	287	+	view.mu.Lock()
	288	+	link := view.link
	289	+	view.mu.Unlock()
	290	+	require.NotNil(t, link)
	291	+
	292	+	d.shutdownAll(ports.ReasonServerShutdown)
	293	+	select {
	294	+	case <-transport.closed:
	295	+	default:
	296	+	t.Fatal("shutdown did not close the exact remote transport")
	297	+	}
	298	+	awaitTestCompletion(t, link.done, "remote link worker did not stop after shutdown")
	299	+	d.mu.Lock()
	300	+	require.Empty(t, d.remoteViews)
	301	+	d.mu.Unlock()
	302	+	}
	303	+
	304	+	func TestRemoteLinkForwardsRemoteViewInputAndContentResize(t *testing.T) {
	305	+	d := newTestDaemon(t, nil, stubClock{})
	306	+	target := remoteLinkTestTarget()
	307	+	size := domain.Size{Cols: 80, Rows: 24}
	308	+	transport := newRemoteLinkTestTransport()
	309	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	310	+	d.remoteDialerFactory = &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport}}
	311	+
	312	+	view, err := d.openRemoteView(context.Background(), target, size)
	313	+	require.NoError(t, err)
	314	+	awaitFrame(t, transport.sent, ports.MsgHello)
	315	+	awaitFrame(t, transport.sent, ports.MsgAck)
	316	+
	317	+	clientTransport, _ := newCapturingTransport(t)
	318	+	ac := &attachedClient{tr: clientTransport, output: newOutputStateStream(), size: size}
	319	+	ac.initOverlays()
	320	+	ac.setAttachmentOwner(view)
	321	+	view.mu.Lock()
	322	+	require.True(t, view.registerAttachmentLocked(ac))
	323	+	view.mu.Unlock()
	324	+	token := attachmentOwnerToken(view, ac, clientTransport)
	325	+	require.True(t, token.attachmentCurrent())
	326	+
	327	+	d.handleSequencedInputForAttachment(token, 42, []byte("remote input"))
	328	+	inputFrame := awaitFrame(t, transport.sent, ports.MsgInput)
	329	+	input, err := ports.UnmarshalInput(inputFrame.Payload)
	330	+	require.NoError(t, err)
	331	+	require.Equal(t, ports.Input{InputSeq: 42, Data: []byte("remote input")}, input)
	332	+
	333	+	require.True(t, d.resizeAttachmentForLease(token, domain.Size{Cols: 100, Rows: 30}))
	334	+	resizeFrame := awaitFrame(t, transport.sent, ports.MsgResize)
	335	+	resize, err := ports.UnmarshalResize(resizeFrame.Payload)
	336	+	require.NoError(t, err)
	337	+	require.Equal(t, domain.Size{Cols: 100, Rows: 28}, resize.Size)
	338	+
	339	+	require.NoError(t, transport.Close())
	340	+	}
	341	+
	342	+	func TestOpenRemoteViewElectsOneConstructorAndCanceledWaiterDoesNotPoisonWinner(t *testing.T) {
	343	+	d := newTestDaemon(t, nil, stubClock{})
	344	+	target := remoteLinkTestTarget()
	345	+	size := domain.Size{Cols: 80, Rows: 24}
	346	+	transport := newRemoteLinkTestTransport()
	347	+	enqueueRemoteLinkHandshake(t, transport, target, size)
	348	+	started := make(chan struct{})
	349	+	release := make(chan struct{})
	350	+	factory := &remoteLinkTestFactory{dialer: &remoteLinkTestDialer{transport: transport, started: started, release: release}}
	351	+	d.remoteDialerFactory = factory
	352	+
	353	+	type result struct {
	354	+	view *remoteView
	355	+	err error
	356	+	}
	357	+	ownerResult := make(chan result, 1)
	358	+	go func() {
	359	+	view, err := d.openRemoteView(context.Background(), target, size)
	360	+	ownerResult <- result{view: view, err: err}
	361	+	}()
	362	+	<-started
	363	+
	364	+	waiterCtx, cancelWaiter := context.WithCancel(context.Background())
	365	+	cancelWaiter()
	366	+	view, err := d.openRemoteView(waiterCtx, target, size)
	367	+	require.Nil(t, view)
	368	+	require.ErrorIs(t, err, context.Canceled)
	369	+	require.Equal(t, int32(1), factory.calls.Load())
	370	+
	371	+	close(release)
	372	+	owner := <-ownerResult
	373	+	require.NoError(t, owner.err)
	374	+	require.NotNil(t, owner.view)
	375	+	require.Equal(t, int32(1), factory.calls.Load())
	376	+
	377	+	warm, err := d.openRemoteView(context.Background(), target, size)
	378	+	require.NoError(t, err)
	379	+	require.Same(t, owner.view, warm)
	380	+	require.Equal(t, int32(1), factory.calls.Load())
	381	+
	382	+	require.NoError(t, transport.Close())
	383	+	}