655f64a fix(remote): finish navigation recovery review · vev

M internal/ports/frame_test.go +3 -0


@@ -21,6 +21,9 @@ func TestControlMsgTypes(t *testing.T) {
21	21		}
22	22
23	23		func TestMsgTypeUnique(t *testing.T) {
	24	+	if MsgNavigationAction != 23 {
	25	+	t.Fatalf("MsgNavigationAction = %d, want 23", MsgNavigationAction)
	26	+	}
24	27		tests := []struct {
25	28		name string
26	29		typ MsgType


@@ -21,6 +21,9 @@ func TestControlMsgTypes(t *testing.T) {
21	}	21		}
22		22
23	func TestMsgTypeUnique(t *testing.T) {	23		func TestMsgTypeUnique(t *testing.T) {
		24	+	if MsgNavigationAction != 23 {
		25	+	t.Fatalf("MsgNavigationAction = %d, want 23", MsgNavigationAction)
		26	+	}
24	tests := []struct {	27		tests := []struct {
25	name string	28		name string
26	typ MsgType	29		typ MsgType

M internal/usecase/client/client.go +6 -2


@@ -570,12 +570,16 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
570	570		returnRoute = nil
571	571		homeNavigationPending = false
572	572		returnNavigationPending = true
573		-	returnResumeFallback = true
574	573		dialer = route.dialer
575	574		attemptRequest = route.request
576		-	attemptRequest.Intent = ports.IntentResume
	575	+	if route.resumeToken != 0 {
	576	+	attemptRequest.Intent = ports.IntentResume
	577	+	} else {
	578	+	attemptRequest.Intent = ports.IntentAttach
	579	+	}
577	580		attemptRequest.StartupOverlay = ports.StartupOverlayNone
578	581		resumeToken = route.resumeToken
	582	+	returnResumeFallback = route.resumeToken != 0
579	583		remote = syncReconnectRemote(reconnect, attemptRequest.Remote \|\| r.remote)
580	584		backoff = defaultReconnectBackoff.initial
581	585		continue

 			returnRoute = nil
 			homeNavigationPending = false
 			returnNavigationPending = true
-			returnResumeFallback = true
 			dialer = route.dialer
 			attemptRequest = route.request
-			attemptRequest.Intent = ports.IntentResume
+			if route.resumeToken != 0 {
+				attemptRequest.Intent = ports.IntentResume
+			} else {
+				attemptRequest.Intent = ports.IntentAttach
+			}
 			attemptRequest.StartupOverlay = ports.StartupOverlayNone
 			resumeToken = route.resumeToken
+			returnResumeFallback = route.resumeToken != 0
 			remote = syncReconnectRemote(reconnect, attemptRequest.Remote || r.remote)
 			backoff = defaultReconnectBackoff.initial
 			continue

M internal/usecase/daemon/daemon.go +8 -4


@@ -1170,9 +1170,11 @@ func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <
1170	1170		failAttachment()
1171	1171		return
1172	1172		}
1173		-	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
	1173	+	welcomeDone, welcomeErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
1174	1174		return ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, d.resumeTokenSnapshot(ac)), welcomeTicket)
1175		-	}); err != nil {
	1175	+	})
	1176	+	if welcomeErr != nil {
	1177	+	<-welcomeDone
1176	1178		welcomeTicket.End()
1177	1179		failAttachment()
1178	1180		return

@@ -1202,10 +1204,12 @@ func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <
1202	1204		}
1203	1205		paintToken := sess.attachmentToken(ac, tr)
1204	1206		painted := make(chan bool, 1)
1205		-	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
	1207	+	paintDone, paintErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
1206	1208		painted <- d.firstPaintForTransition(paintToken)
1207	1209		return nil
1208		-	}); err != nil {
	1210	+	})
	1211	+	if paintErr != nil {
	1212	+	<-paintDone
1209	1213		postWelcomeTicket.End()
1210	1214		failAttachment()
1211	1215		return

 		failAttachment()
 		return
 	}
-	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
+	welcomeDone, welcomeErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
 		return ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, d.resumeTokenSnapshot(ac)), welcomeTicket)
-	}); err != nil {
+	})
+	if welcomeErr != nil {
+		<-welcomeDone
 		welcomeTicket.End()
 		failAttachment()
 		return
 	}
 	paintToken := sess.attachmentToken(ac, tr)
 	painted := make(chan bool, 1)
-	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
+	paintDone, paintErr := boundedHandshakeOperationTracked(handshakeCtx, tr, func() error {
 		painted <- d.firstPaintForTransition(paintToken)
 		return nil
-	}); err != nil {
+	})
+	if paintErr != nil {
+		<-paintDone
 		postWelcomeTicket.End()
 		failAttachment()
 		return

M internal/usecase/daemon/handshake.go +19 -5


@@ -75,24 +75,38 @@ func watchHandshakeTransport(ctx context.Context, transport ports.Transport) fun
75	75		// handshake context remains live. Closing the transport is the interruption
76	76		// mechanism because ports.Transport methods do not accept contexts.
77	77		func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
	78	+	_, err := boundedHandshakeOperationTracked(ctx, transport, operation)
	79	+	return err
	80	+	}
	81	+
	82	+	// boundedHandshakeOperationTracked also returns a completion signal for the
	83	+	// operation worker. Callers that own an effect gate must wait for this signal
	84	+	// before releasing the gate after cancellation; the transport operation itself
	85	+	// may still be unwinding after Close interrupts it.
	86	+	func boundedHandshakeOperationTracked(ctx context.Context, transport ports.Transport, operation func() error) (<-chan struct{}, error) {
	87	+	done := make(chan struct{})
78	88		if err := ctx.Err(); err != nil {
	89	+	close(done)
79	90		_ = transport.Close()
80		-	return err
	91	+	return done, err
81	92		}
82	93		completed := make(chan error, 1)
83		-	go func() { completed <- operation() }()
	94	+	go func() {
	95	+	defer close(done)
	96	+	completed <- operation()
	97	+	}()
84	98		select {
85	99		case err := <-completed:
86	100		if ctxErr := ctx.Err(); ctxErr != nil {
87	101		_ = transport.Close()
88		-	return ctxErr
	102	+	return done, ctxErr
89	103		}
90		-	return err
	104	+	return done, err
91	105		case <-ctx.Done():
92	106		_ = transport.Close()
93	107		// The result channel is buffered, so the operation can publish its
94	108		// completion after Close without keeping this cancellation path stuck.
95		-	return ctx.Err()
	109	+	return done, ctx.Err()
96	110		}
97	111		}
98	112

 // handshake context remains live. Closing the transport is the interruption
 // mechanism because ports.Transport methods do not accept contexts.
 func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
+	_, err := boundedHandshakeOperationTracked(ctx, transport, operation)
+	return err
+}
+// boundedHandshakeOperationTracked also returns a completion signal for the
+// operation worker. Callers that own an effect gate must wait for this signal
+// before releasing the gate after cancellation; the transport operation itself
+// may still be unwinding after Close interrupts it.
+func boundedHandshakeOperationTracked(ctx context.Context, transport ports.Transport, operation func() error) (<-chan struct{}, error) {
+	done := make(chan struct{})
 	if err := ctx.Err(); err != nil {
+		close(done)
 		_ = transport.Close()
-		return err
+		return done, err
 	}
 	completed := make(chan error, 1)
-	go func() { completed <- operation() }()
+	go func() {
+		defer close(done)
+		completed <- operation()
+	}()
 	select {
 	case err := <-completed:
 		if ctxErr := ctx.Err(); ctxErr != nil {
 			_ = transport.Close()
-			return ctxErr
+			return done, ctxErr
 		}
-		return err
+		return done, err
 	case <-ctx.Done():
 		_ = transport.Close()
 		// The result channel is buffered, so the operation can publish its
 		// completion after Close without keeping this cancellation path stuck.
-		return ctx.Err()
+		return done, ctx.Err()
 	}
 }

M internal/usecase/daemon/picker.go +10 -4


@@ -890,15 +890,21 @@ func (d *Daemon) switchToTargetForAttachment(token attachmentConnectionToken, ta
890	890		// the target is sent, the local attachment is detached and the client opens a
891	891		// fresh connection to the owning daemon.
892	892		func (d *Daemon) sendRemoteAttachTargetForAttachment(token attachmentConnectionToken, target picker.Target, key domain.RemoteSessionKey, guard sessionHandoffGuard, _ string) error {
	893	+	failUnavailable := func() error {
	894	+	if token.attachmentCurrent() {
	895	+	d.notifyRemotePickerUnavailable(token.sess, target)
	896	+	}
	897	+	return errAttachmentTransition
	898	+	}
893	899		var handoff ports.AttachTarget
894	900		if target.RemoteTarget != nil {
895	901		remoteTarget := *target.RemoteTarget
896	902		if err := remoteTarget.Validate(); err != nil \|\| target.Stopped != remoteTarget.Stopped \|\| !d.remotePickerTargetReadyTarget(remoteTarget) {
897		-	return errAttachmentTransition
	903	+	return failUnavailable()
898	904		}
899	905		if target.RemoteKey != nil {
900	906		if target.Session != target.RemoteKey.ID() \|\| target.RemoteKey.Host != remoteTarget.Endpoint \|\| target.RemoteKey.Name != remoteTarget.SessionName \|\| target.RemoteKey.LifecycleID != remoteTarget.LifecycleID {
901		-	return errAttachmentTransition
	907	+	return failUnavailable()
902	908		}
903	909		}
904	910		handoff = ports.AttachTarget{

@@ -910,13 +916,13 @@ func (d *Daemon) sendRemoteAttachTargetForAttachment(token attachmentConnectionT
910	916		}
911	917		} else {
912	918		if key.Validate() != nil \|\| target.Session != key.ID() \|\| target.Stopped \|\| target.Name != "" \|\| !d.remotePickerTargetReady(key) {
913		-	return errAttachmentTransition
	919	+	return failUnavailable()
914	920		}
915	921		handoff = ports.AttachTarget{Endpoint: key.Host, Session: key.Name, Intent: ports.IntentAttach, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned}
916	922		}
917	923		payload := ports.MarshalAttachTarget(handoff)
918	924		if payload == nil {
919		-	return errAttachmentTransition
	925	+	return failUnavailable()
920	926		}
921	927		if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: payload}); err != nil {
922	928		return err

 // the target is sent, the local attachment is detached and the client opens a
 // fresh connection to the owning daemon.
 func (d *Daemon) sendRemoteAttachTargetForAttachment(token attachmentConnectionToken, target picker.Target, key domain.RemoteSessionKey, guard sessionHandoffGuard, _ string) error {
+	failUnavailable := func() error {
+		if token.attachmentCurrent() {
+			d.notifyRemotePickerUnavailable(token.sess, target)
+		}
+		return errAttachmentTransition
+	}
 	var handoff ports.AttachTarget
 	if target.RemoteTarget != nil {
 		remoteTarget := *target.RemoteTarget
 		if err := remoteTarget.Validate(); err != nil || target.Stopped != remoteTarget.Stopped || !d.remotePickerTargetReadyTarget(remoteTarget) {
-			return errAttachmentTransition
+			return failUnavailable()
 		}
 		if target.RemoteKey != nil {
 			if target.Session != target.RemoteKey.ID() || target.RemoteKey.Host != remoteTarget.Endpoint || target.RemoteKey.Name != remoteTarget.SessionName || target.RemoteKey.LifecycleID != remoteTarget.LifecycleID {
-				return errAttachmentTransition
+				return failUnavailable()
 			}
 		}
 		handoff = ports.AttachTarget{
 		}
 	} else {
 		if key.Validate() != nil || target.Session != key.ID() || target.Stopped || target.Name != "" || !d.remotePickerTargetReady(key) {
-			return errAttachmentTransition
+			return failUnavailable()
 		}
 		handoff = ports.AttachTarget{Endpoint: key.Host, Session: key.Name, Intent: ports.IntentAttach, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned}
 	}
 	payload := ports.MarshalAttachTarget(handoff)
 	if payload == nil {
-		return errAttachmentTransition
+		return failUnavailable()
 	}
 	if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: payload}); err != nil {
 		return err

M internal/usecase/daemon/remote_attach_test.go +1 -1


@@ -71,7 +71,7 @@ func TestRemotePickerRichHandoffRejectsReplacedLifecycle(t *testing.T) {
71	71		require.ErrorIs(t, d.sendRemoteAttachTargetForAttachment(token, target, key, sessionHandoffGuard{}, "picker-select"), errAttachmentTransition)
72	72		select {
73	73		case frame := <-sends:
74		-	t.Fatalf("rejected handoff sent frame type %d", frame.Type)
	74	+	require.NotEqual(t, ports.MsgAttachTarget, frame.Type, "rejected handoff must not send an attach target")
75	75		default:
76	76		}
77	77		}


@@ -71,7 +71,7 @@ func TestRemotePickerRichHandoffRejectsReplacedLifecycle(t *testing.T) {
71		require.ErrorIs(t, d.sendRemoteAttachTargetForAttachment(token, target, key, sessionHandoffGuard{}, "picker-select"), errAttachmentTransition)	71		require.ErrorIs(t, d.sendRemoteAttachTargetForAttachment(token, target, key, sessionHandoffGuard{}, "picker-select"), errAttachmentTransition)
72		select {	72		select {
73		case frame := <-sends:	73		case frame := <-sends:
74	-	t.Fatalf("rejected handoff sent frame type %d", frame.Type)	74	+	require.NotEqual(t, ports.MsgAttachTarget, frame.Type, "rejected handoff must not send an attach target")
75		default:	75		default:
76		}	76		}
77		}	77		}

M internal/usecase/daemon/remote_route_test.go +2 -0


@@ -37,6 +37,8 @@ func TestFinishRouteAttachRollsBackCreatedSession(t *testing.T) {
37	37		Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: defaultSize,
38	38		RemoteTarget: &target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
39	39		}
	40	+	// finishRouteAttach follows the locked-caller contract: it releases d.mu
	41	+	// on both the success and error paths, so retain the lock across the call.
40	42		d.mu.Lock()
41	43		_, err = d.finishRouteAttach(sess, &closeTrackingTransport{}, defaultSize, terminalEnv{}, hello, true, true)
42	44		var protocol *protoErr


@@ -37,6 +37,8 @@ func TestFinishRouteAttachRollsBackCreatedSession(t *testing.T) {
37	Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: defaultSize,	37		Version: ports.ProtocolVersion, Intent: ports.IntentAttach, Name: "work", Size: defaultSize,
38	RemoteTarget: &target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,	38		RemoteTarget: &target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
39	}	39		}
		40	+	// finishRouteAttach follows the locked-caller contract: it releases d.mu
		41	+	// on both the success and error paths, so retain the lock across the call.
40	d.mu.Lock()	42		d.mu.Lock()
41	_, err = d.finishRouteAttach(sess, &closeTrackingTransport{}, defaultSize, terminalEnv{}, hello, true, true)	43		_, err = d.finishRouteAttach(sess, &closeTrackingTransport{}, defaultSize, terminalEnv{}, hello, true, true)
42	var protocol *protoErr	44		var protocol *protoErr

M internal/usecase/daemon/resume.go +1 -1


@@ -193,7 +193,7 @@ func (d *Daemon) waitParkingInFlight(h ports.Hello) bool {
193	193		}
194	194
195	195		func (d Daemon) prepareParkAttachment(sess session, ac *attachedClient) bool {
196		-	if !ac.resumeCapable {
	196	+	if sess == nil \|\| ac == nil \|\| !ac.resumeCapable {
197	197		return false
198	198		}
199	199		d.mu.Lock()


@@ -193,7 +193,7 @@ func (d *Daemon) waitParkingInFlight(h ports.Hello) bool {
193		}	193		}
194			194
195		func (d Daemon) prepareParkAttachment(sess session, ac *attachedClient) bool {	195		func (d Daemon) prepareParkAttachment(sess session, ac *attachedClient) bool {
196	-	if !ac.resumeCapable {	196	+	if sess == nil \|\| ac == nil \|\| !ac.resumeCapable {
197		return false	197		return false
198		}	198		}
199		d.mu.Lock()	199		d.mu.Lock()