c4be05e fix(handshake): bound unified session handshakes · vev

M internal/app/run.go +12 -74


@@ -47,7 +47,6 @@ import (
47	47		"github.com/bnema/vev/internal/platform"
48	48		"github.com/bnema/vev/internal/ports"
49	49		"github.com/bnema/vev/internal/usecase/client"
50		-	"github.com/bnema/vev/internal/usecase/confirm"
51	50		"github.com/bnema/vev/internal/usecase/daemon"
52	51		"github.com/bnema/vev/internal/usecase/recovery"
53	52		pdgram "github.com/bnema/vev/pkg/dgram"

@@ -868,70 +867,20 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac
868	867		if localDialer == nil {
869	868		localDialer = defaultLocalDialer
870	869		}
871		-	return runLocalAttachWithRecovery(ctx, intent, name, attachRecoveryDeps{
872		-	confirmer: confirm.NewConfirmer(os.Stdin, os.Stderr),
873		-	attach: func(ctx context.Context, intent uint8, name string) error {
874		-	if log != nil {
875		-	log.Info("attaching to local session", "intent", intent, "name", name)
876		-	}
877		-	return runClient(ctx, client.Dependencies{
878		-	Dialer: localDialer(),
879		-	Terminal: term.New(),
880		-	Clock: clock.New(),
881		-	Logger: log,
882		-	RuntimeObserver: deps.runtimeObserver,
883		-	Remote: false,
884		-	}, client.AttachRequest{Intent: intent, SessionName: name})
885		-	},
886		-	killDaemon: requestDaemonStop,
887		-	settleAfterKill: waitForDaemonStop,
888		-	})
889		-	}
890		-
891		-	type attachRecoveryDeps struct {
892		-	confirmer confirm.Confirmer
893		-	attach func(context.Context, uint8, string) error
894		-	killDaemon func(context.Context) error
895		-	settleAfterKill func(context.Context) error
896		-	}
897		-
898		-	const (
899		-	daemonStopTimeout = 2 * time.Second
900		-	daemonRestartSettle = 50 * time.Millisecond
901		-	legacyMalformedHelloSignal = "malformed hello"
902		-	)
903		-
904		-	func runLocalAttachWithRecovery(ctx context.Context, intent uint8, name string, deps attachRecoveryDeps) error {
905		-	err := deps.attach(ctx, intent, name)
906		-	if !isDaemonVersionDrift(err) {
907		-	return err
908		-	}
909		-
910		-	ok, promptErr := deps.confirmer.Confirm("Your vev version differs from the running daemon; kill it and restart?")
911		-	if promptErr != nil {
912		-	return fmt.Errorf("vev: reading confirmation: %w", promptErr)
913		-	}
914		-	if !ok {
915		-	return err
916		-	}
917		-	if killErr := deps.killDaemon(ctx); killErr != nil {
918		-	return killErr
919		-	}
920		-	if deps.settleAfterKill != nil {
921		-	if settleErr := deps.settleAfterKill(ctx); settleErr != nil {
922		-	return settleErr
923		-	}
924		-	}
925		-	return deps.attach(ctx, intent, name)
	870	+	if log != nil {
	871	+	log.Info("attaching to local session", "intent", intent, "name", name)
	872	+	}
	873	+	return runClient(ctx, client.Dependencies{
	874	+	Dialer: localDialer(),
	875	+	Terminal: term.New(),
	876	+	Clock: clock.New(),
	877	+	Logger: log,
	878	+	RuntimeObserver: deps.runtimeObserver,
	879	+	Remote: false,
	880	+	}, client.AttachRequest{Intent: intent, SessionName: name})
926	881		}
927	882
928		-	func isDaemonVersionDrift(err error) bool {
929		-	var protocolErr *client.ProtocolError
930		-	if !errors.As(err, &protocolErr) {
931		-	return false
932		-	}
933		-	return protocolErr.Code == ports.ErrVersionMismatch \|\| protocolErr.Text == legacyMalformedHelloSignal
934		-	}
	883	+	const daemonStopTimeout = 2 * time.Second
935	884
936	885		type localDaemonDialer struct {
937	886		dir string

@@ -1011,17 +960,6 @@ func createDetachedLocalSession(ctx context.Context, name string) error {
1011	960		}
1012	961		}
1013	962
1014		-	func waitForDaemonStop(ctx context.Context) error {
1015		-	timer := time.NewTimer(daemonRestartSettle)
1016		-	defer timer.Stop()
1017		-	select {
1018		-	case <-ctx.Done():
1019		-	return ctx.Err()
1020		-	case <-timer.C:
1021		-	return nil
1022		-	}
1023		-	}
1024		-
1025	963		func requestDaemonStop(ctx context.Context) error {
1026	964		ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)
1027	965		defer cancel()


@@ -47,7 +47,6 @@ import (
47		"github.com/bnema/vev/internal/platform"	47		"github.com/bnema/vev/internal/platform"
48		"github.com/bnema/vev/internal/ports"	48		"github.com/bnema/vev/internal/ports"
49		"github.com/bnema/vev/internal/usecase/client"	49		"github.com/bnema/vev/internal/usecase/client"
50	-	"github.com/bnema/vev/internal/usecase/confirm"
51		"github.com/bnema/vev/internal/usecase/daemon"	50		"github.com/bnema/vev/internal/usecase/daemon"
52		"github.com/bnema/vev/internal/usecase/recovery"	51		"github.com/bnema/vev/internal/usecase/recovery"
53		pdgram "github.com/bnema/vev/pkg/dgram"	52		pdgram "github.com/bnema/vev/pkg/dgram"

@@ -868,70 +867,20 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac
868		if localDialer == nil {	867		if localDialer == nil {
869		localDialer = defaultLocalDialer	868		localDialer = defaultLocalDialer
870		}	869		}
871	-	return runLocalAttachWithRecovery(ctx, intent, name, attachRecoveryDeps{	870	+	if log != nil {
872	-	confirmer: confirm.NewConfirmer(os.Stdin, os.Stderr),	871	+	log.Info("attaching to local session", "intent", intent, "name", name)
873	-	attach: func(ctx context.Context, intent uint8, name string) error {	872	+	}
874	-	if log != nil {	873	+	return runClient(ctx, client.Dependencies{
875	-	log.Info("attaching to local session", "intent", intent, "name", name)	874	+	Dialer: localDialer(),
876	-	}	875	+	Terminal: term.New(),
877	-	return runClient(ctx, client.Dependencies{	876	+	Clock: clock.New(),
878	-	Dialer: localDialer(),	877	+	Logger: log,
879	-	Terminal: term.New(),	878	+	RuntimeObserver: deps.runtimeObserver,
880	-	Clock: clock.New(),	879	+	Remote: false,
881	-	Logger: log,	880	+	}, client.AttachRequest{Intent: intent, SessionName: name})
882	-	RuntimeObserver: deps.runtimeObserver,
883	-	Remote: false,
884	-	}, client.AttachRequest{Intent: intent, SessionName: name})
885	-	},
886	-	killDaemon: requestDaemonStop,
887	-	settleAfterKill: waitForDaemonStop,
888	-	})
889	-	}
890	-
891	-	type attachRecoveryDeps struct {
892	-	confirmer confirm.Confirmer
893	-	attach func(context.Context, uint8, string) error
894	-	killDaemon func(context.Context) error
895	-	settleAfterKill func(context.Context) error
896	-	}
897	-
898	-	const (
899	-	daemonStopTimeout = 2 * time.Second
900	-	daemonRestartSettle = 50 * time.Millisecond
901	-	legacyMalformedHelloSignal = "malformed hello"
902	-	)
903	-
904	-	func runLocalAttachWithRecovery(ctx context.Context, intent uint8, name string, deps attachRecoveryDeps) error {
905	-	err := deps.attach(ctx, intent, name)
906	-	if !isDaemonVersionDrift(err) {
907	-	return err
908	-	}
909	-
910	-	ok, promptErr := deps.confirmer.Confirm("Your vev version differs from the running daemon; kill it and restart?")
911	-	if promptErr != nil {
912	-	return fmt.Errorf("vev: reading confirmation: %w", promptErr)
913	-	}
914	-	if !ok {
915	-	return err
916	-	}
917	-	if killErr := deps.killDaemon(ctx); killErr != nil {
918	-	return killErr
919	-	}
920	-	if deps.settleAfterKill != nil {
921	-	if settleErr := deps.settleAfterKill(ctx); settleErr != nil {
922	-	return settleErr
923	-	}
924	-	}
925	-	return deps.attach(ctx, intent, name)
926		}	881		}
927			882
928	-	func isDaemonVersionDrift(err error) bool {	883	+	const daemonStopTimeout = 2 * time.Second
929	-	var protocolErr *client.ProtocolError
930	-	if !errors.As(err, &protocolErr) {
931	-	return false
932	-	}
933	-	return protocolErr.Code == ports.ErrVersionMismatch \|\| protocolErr.Text == legacyMalformedHelloSignal
934	-	}
935			884
936		type localDaemonDialer struct {	885		type localDaemonDialer struct {
937		dir string	886		dir string

@@ -1011,17 +960,6 @@ func createDetachedLocalSession(ctx context.Context, name string) error {
1011		}	960		}
1012		}	961		}
1013			962
1014	-	func waitForDaemonStop(ctx context.Context) error {
1015	-	timer := time.NewTimer(daemonRestartSettle)
1016	-	defer timer.Stop()
1017	-	select {
1018	-	case <-ctx.Done():
1019	-	return ctx.Err()
1020	-	case <-timer.C:
1021	-	return nil
1022	-	}
1023	-	}
1024	-
1025		func requestDaemonStop(ctx context.Context) error {	963		func requestDaemonStop(ctx context.Context) error {
1026		ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)	964		ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)
1027		defer cancel()	965		defer cancel()

M internal/app/run_test.go +0 -126


@@ -23,7 +23,6 @@ import (
23	23		"github.com/bnema/vev/internal/ports"
24	24		portsmocks "github.com/bnema/vev/internal/ports/mocks"
25	25		"github.com/bnema/vev/internal/usecase/client"
26		-	"github.com/bnema/vev/internal/usecase/confirm"
27	26		"github.com/bnema/vev/internal/usecase/daemon"
28	27		"github.com/bnema/vev/pkg/kv"
29	28		"github.com/bnema/vev/pkg/safedir"

@@ -754,131 +753,6 @@ func captureStdout(t *testing.T, fn func()) string {
754	753		return string(out)
755	754		}
756	755
757		-	func TestRunLocalAttachPromptsAndRestartsOnProtocolMismatch(t *testing.T) {
758		-	var prompts bytes.Buffer
759		-	answers := strings.NewReader("y\n")
760		-	attachCalls := 0
761		-	killCalls := 0
762		-
763		-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
764		-	confirmer: confirm.NewConfirmer(answers, &prompts),
765		-	attach: func(context.Context, uint8, string) error {
766		-	attachCalls++
767		-	if attachCalls == 1 {
768		-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
769		-	}
770		-	return nil
771		-	},
772		-	killDaemon: func(context.Context) error {
773		-	killCalls++
774		-	return nil
775		-	},
776		-	})
777		-	if err != nil {
778		-	t.Fatalf("runLocalAttachWithRecovery returned error: %v", err)
779		-	}
780		-	if attachCalls != 2 {
781		-	t.Fatalf("attach calls = %d, want 2", attachCalls)
782		-	}
783		-	if killCalls != 1 {
784		-	t.Fatalf("kill calls = %d, want 1", killCalls)
785		-	}
786		-	if got := prompts.String(); !strings.Contains(got, "Your vev version differs") \|\| !strings.Contains(got, "kill it") {
787		-	t.Fatalf("prompt = %q, want version/kill prompt", got)
788		-	}
789		-	}
790		-
791		-	func TestRunLocalAttachPropagatesPromptError(t *testing.T) {
792		-	readErr := errors.New("read failed")
793		-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
794		-	confirmer: confirm.NewConfirmer(errorReader{err: readErr}, &bytes.Buffer{}),
795		-	attach: func(context.Context, uint8, string) error {
796		-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
797		-	},
798		-	killDaemon: func(context.Context) error {
799		-	t.Fatal("killDaemon should not be called after prompt error")
800		-	return nil
801		-	},
802		-	})
803		-	if !errors.Is(err, readErr) {
804		-	t.Fatalf("error = %v, want wrapped %v", err, readErr)
805		-	}
806		-	}
807		-
808		-	func TestRunLocalAttachPropagatesKillError(t *testing.T) {
809		-	killErr := errors.New("kill failed")
810		-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
811		-	confirmer: confirm.NewConfirmer(strings.NewReader("y\n"), &bytes.Buffer{}),
812		-	attach: func(context.Context, uint8, string) error {
813		-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
814		-	},
815		-	killDaemon: func(context.Context) error { return killErr },
816		-	})
817		-	if !errors.Is(err, killErr) {
818		-	t.Fatalf("error = %v, want %v", err, killErr)
819		-	}
820		-	}
821		-
822		-	func TestRunLocalAttachSettlesBeforeRetry(t *testing.T) {
823		-	var order []string
824		-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
825		-	confirmer: confirm.NewConfirmer(strings.NewReader("y\n"), &bytes.Buffer{}),
826		-	attach: func(context.Context, uint8, string) error {
827		-	order = append(order, "attach")
828		-	if len(order) == 1 {
829		-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
830		-	}
831		-	return nil
832		-	},
833		-	killDaemon: func(context.Context) error {
834		-	order = append(order, "kill")
835		-	return nil
836		-	},
837		-	settleAfterKill: func(context.Context) error {
838		-	order = append(order, "settle")
839		-	return nil
840		-	},
841		-	})
842		-	if err != nil {
843		-	t.Fatalf("runLocalAttachWithRecovery returned error: %v", err)
844		-	}
845		-	if got, want := strings.Join(order, ","), "attach,kill,settle,attach"; got != want {
846		-	t.Fatalf("order = %s, want %s", got, want)
847		-	}
848		-	}
849		-
850		-	type errorReader struct{ err error }
851		-
852		-	func (r errorReader) Read([]byte) (int, error) { return 0, r.err }
853		-
854		-	func TestRunLocalAttachDeclineKeepsOriginalError(t *testing.T) {
855		-	answers := strings.NewReader("n\n")
856		-	wantErr := &client.ProtocolError{Code: ports.ErrInternal, Text: "malformed hello"}
857		-	attachCalls := 0
858		-	killCalls := 0
859		-
860		-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
861		-	confirmer: confirm.NewConfirmer(answers, &bytes.Buffer{}),
862		-	attach: func(context.Context, uint8, string) error {
863		-	attachCalls++
864		-	return wantErr
865		-	},
866		-	killDaemon: func(context.Context) error {
867		-	killCalls++
868		-	return nil
869		-	},
870		-	})
871		-	if !errors.Is(err, wantErr) {
872		-	t.Fatalf("error = %v, want original %v", err, wantErr)
873		-	}
874		-	if attachCalls != 1 {
875		-	t.Fatalf("attach calls = %d, want 1", attachCalls)
876		-	}
877		-	if killCalls != 0 {
878		-	t.Fatalf("kill calls = %d, want 0", killCalls)
879		-	}
880		-	}
881		-
882	756		func TestRunAttachRejectsNestedVEVBeforeDial(t *testing.T) {
883	757		called := false
884	758		err := runAttachWithDeps(context.Background(), ports.IntentEphemeral, "", "", "outer", nil, runAttachDeps{


@@ -23,7 +23,6 @@ import (
23		"github.com/bnema/vev/internal/ports"	23	"github.com/bnema/vev/internal/ports"
24		portsmocks "github.com/bnema/vev/internal/ports/mocks"	24	portsmocks "github.com/bnema/vev/internal/ports/mocks"
25		"github.com/bnema/vev/internal/usecase/client"	25	"github.com/bnema/vev/internal/usecase/client"
26	-	"github.com/bnema/vev/internal/usecase/confirm"
27		"github.com/bnema/vev/internal/usecase/daemon"	26	"github.com/bnema/vev/internal/usecase/daemon"
28		"github.com/bnema/vev/pkg/kv"	27	"github.com/bnema/vev/pkg/kv"
29		"github.com/bnema/vev/pkg/safedir"	28	"github.com/bnema/vev/pkg/safedir"

@@ -754,131 +753,6 @@ func captureStdout(t *testing.T, fn func()) string {
754		return string(out)	753	return string(out)
755		}	754	}
756			755
757	-	func TestRunLocalAttachPromptsAndRestartsOnProtocolMismatch(t *testing.T) {
758	-	var prompts bytes.Buffer
759	-	answers := strings.NewReader("y\n")
760	-	attachCalls := 0
761	-	killCalls := 0
762	-
763	-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
764	-	confirmer: confirm.NewConfirmer(answers, &prompts),
765	-	attach: func(context.Context, uint8, string) error {
766	-	attachCalls++
767	-	if attachCalls == 1 {
768	-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
769	-	}
770	-	return nil
771	-	},
772	-	killDaemon: func(context.Context) error {
773	-	killCalls++
774	-	return nil
775	-	},
776	-	})
777	-	if err != nil {
778	-	t.Fatalf("runLocalAttachWithRecovery returned error: %v", err)
779	-	}
780	-	if attachCalls != 2 {
781	-	t.Fatalf("attach calls = %d, want 2", attachCalls)
782	-	}
783	-	if killCalls != 1 {
784	-	t.Fatalf("kill calls = %d, want 1", killCalls)
785	-	}
786	-	if got := prompts.String(); !strings.Contains(got, "Your vev version differs") \|\| !strings.Contains(got, "kill it") {
787	-	t.Fatalf("prompt = %q, want version/kill prompt", got)
788	-	}
789	-	}
790	-
791	-	func TestRunLocalAttachPropagatesPromptError(t *testing.T) {
792	-	readErr := errors.New("read failed")
793	-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
794	-	confirmer: confirm.NewConfirmer(errorReader{err: readErr}, &bytes.Buffer{}),
795	-	attach: func(context.Context, uint8, string) error {
796	-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
797	-	},
798	-	killDaemon: func(context.Context) error {
799	-	t.Fatal("killDaemon should not be called after prompt error")
800	-	return nil
801	-	},
802	-	})
803	-	if !errors.Is(err, readErr) {
804	-	t.Fatalf("error = %v, want wrapped %v", err, readErr)
805	-	}
806	-	}
807	-
808	-	func TestRunLocalAttachPropagatesKillError(t *testing.T) {
809	-	killErr := errors.New("kill failed")
810	-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
811	-	confirmer: confirm.NewConfirmer(strings.NewReader("y\n"), &bytes.Buffer{}),
812	-	attach: func(context.Context, uint8, string) error {
813	-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
814	-	},
815	-	killDaemon: func(context.Context) error { return killErr },
816	-	})
817	-	if !errors.Is(err, killErr) {
818	-	t.Fatalf("error = %v, want %v", err, killErr)
819	-	}
820	-	}
821	-
822	-	func TestRunLocalAttachSettlesBeforeRetry(t *testing.T) {
823	-	var order []string
824	-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
825	-	confirmer: confirm.NewConfirmer(strings.NewReader("y\n"), &bytes.Buffer{}),
826	-	attach: func(context.Context, uint8, string) error {
827	-	order = append(order, "attach")
828	-	if len(order) == 1 {
829	-	return &client.ProtocolError{Code: ports.ErrVersionMismatch, Text: "protocol version mismatch"}
830	-	}
831	-	return nil
832	-	},
833	-	killDaemon: func(context.Context) error {
834	-	order = append(order, "kill")
835	-	return nil
836	-	},
837	-	settleAfterKill: func(context.Context) error {
838	-	order = append(order, "settle")
839	-	return nil
840	-	},
841	-	})
842	-	if err != nil {
843	-	t.Fatalf("runLocalAttachWithRecovery returned error: %v", err)
844	-	}
845	-	if got, want := strings.Join(order, ","), "attach,kill,settle,attach"; got != want {
846	-	t.Fatalf("order = %s, want %s", got, want)
847	-	}
848	-	}
849	-
850	-	type errorReader struct{ err error }
851	-
852	-	func (r errorReader) Read([]byte) (int, error) { return 0, r.err }
853	-
854	-	func TestRunLocalAttachDeclineKeepsOriginalError(t *testing.T) {
855	-	answers := strings.NewReader("n\n")
856	-	wantErr := &client.ProtocolError{Code: ports.ErrInternal, Text: "malformed hello"}
857	-	attachCalls := 0
858	-	killCalls := 0
859	-
860	-	err := runLocalAttachWithRecovery(context.Background(), ports.IntentEphemeral, "", attachRecoveryDeps{
861	-	confirmer: confirm.NewConfirmer(answers, &bytes.Buffer{}),
862	-	attach: func(context.Context, uint8, string) error {
863	-	attachCalls++
864	-	return wantErr
865	-	},
866	-	killDaemon: func(context.Context) error {
867	-	killCalls++
868	-	return nil
869	-	},
870	-	})
871	-	if !errors.Is(err, wantErr) {
872	-	t.Fatalf("error = %v, want original %v", err, wantErr)
873	-	}
874	-	if attachCalls != 1 {
875	-	t.Fatalf("attach calls = %d, want 1", attachCalls)
876	-	}
877	-	if killCalls != 0 {
878	-	t.Fatalf("kill calls = %d, want 0", killCalls)
879	-	}
880	-	}
881	-
882		func TestRunAttachRejectsNestedVEVBeforeDial(t *testing.T) {	756	func TestRunAttachRejectsNestedVEVBeforeDial(t *testing.T) {
883		called := false	757	called := false
884		err := runAttachWithDeps(context.Background(), ports.IntentEphemeral, "", "", "outer", nil, runAttachDeps{	758	err := runAttachWithDeps(context.Background(), ports.IntentEphemeral, "", "", "outer", nil, runAttachDeps{

M internal/ports/frame.go +6 -0

@@ -1,5 +1,11 @@
1	1		package ports
2	2
	3	+	import "time"
	4	+
	5	+	// HandshakeTimeout bounds every transport handshake from connect through the
	6	+	// first committed publication.
	7	+	const HandshakeTimeout = 15 * time.Second
	8	+
3	9		// ProtocolVersion is the current vev IPC wire protocol version.
4	10		const ProtocolVersion uint16 = 23
5	11

@@ -1,5 +1,11 @@
1	package ports	1		package ports
2		2
		3	+	import "time"
		4	+
		5	+	// HandshakeTimeout bounds every transport handshake from connect through the
		6	+	// first committed publication.
		7	+	const HandshakeTimeout = 15 * time.Second
		8	+
3	// ProtocolVersion is the current vev IPC wire protocol version.	9		// ProtocolVersion is the current vev IPC wire protocol version.
4	const ProtocolVersion uint16 = 23	10		const ProtocolVersion uint16 = 23
5		11

M internal/usecase/client/client.go +99 -81


@@ -129,9 +129,9 @@ const stdinBufSize = 4096
129	129		// goroutine, decoupling the input/resize pumps from transport back-pressure.
130	130		const sendQueueDepth = 64
131	131
132		-	// preWelcomeTimeout bounds each blocking operation before the daemon has
133		-	// accepted the attach. Mosh uses the same 15-second startup budget.
134		-	const preWelcomeTimeout = 15 * time.Second
	132	+	// preWelcomeTimeout remains the focused-test name for the shared handshake
	133	+	// budget, which now also covers connect and initial publication.
	134	+	const preWelcomeTimeout = ports.HandshakeTimeout
135	135
136	136		// remoteHostLearnerShutdownTimeout bounds how long Run waits for async remote
137	137		// host learning after terminal restoration. Learning is best-effort; a stalled

@@ -313,8 +313,11 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
313	313		}
314	314
315	315		for {
316		-	transport, err := r.dialer.Dial(ctx)
	316	+	handshakeCtx, timedOut, finishHandshake := newHandshakeContext(ctx, r.clock)
	317	+	transport, err := boundedDial(handshakeCtx, r.dialer)
317	318		if err != nil {
	319	+	err = handshakeContextError(ctx, timedOut, err)
	320	+	finishHandshake()
318	321		if resumeToken == 0 \|\| ctx.Err() != nil {
319	322		if clearErr := reconnect.clear(); clearErr != nil {
320	323		return errors.Join(err, clearErr)

@@ -340,11 +343,18 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
340	343		continue
341	344		}
342	345		ms.dialed = true
	346	+	if transport == nil {
	347	+	finishHandshake()
	348	+	return errors.New("vev: dialer returned nil transport")
	349	+	}
	350	+	stopHandshakeTransport := watchHandshakeTransport(handshakeCtx, transport)
343	351		connection, err := NewSessionConnection(transport, SessionTarget{
344	352		Intent: attemptRequest.Intent,
345	353		SessionName: attemptRequest.SessionName,
346	354		})
347	355		if err != nil {
	356	+	stopHandshakeTransport()
	357	+	finishHandshake()
348	358		_ = transport.Close()
349	359		return err
350	360		}

@@ -354,22 +364,28 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error)
354	364		linkEvents = reporter.LinkEvents()
355	365		}
356	366		result := (&attachAttempt{
357		-	runner: r,
358		-	connection: connection,
359		-	remote: remote,
360		-	request: attemptRequest,
361		-	resumeToken: resumeToken,
362		-	clientID: processClientID,
363		-	milestones: &ms,
364		-	themeState: themeState,
365		-	enterRaw: enterRaw,
366		-	reconnect: reconnect,
367		-	linkEvents: linkEvents,
368		-	rememberRemoteHost: rememberRemoteHost,
	367	+	runner: r,
	368	+	connection: connection,
	369	+	remote: remote,
	370	+	handshakeCtx: handshakeCtx,
	371	+	handshakeTimedOut: timedOut,
	372	+	finishHandshake: finishHandshake,
	373	+	stopHandshakeTransport: stopHandshakeTransport,
	374	+	request: attemptRequest,
	375	+	resumeToken: resumeToken,
	376	+	clientID: processClientID,
	377	+	milestones: &ms,
	378	+	themeState: themeState,
	379	+	enterRaw: enterRaw,
	380	+	reconnect: reconnect,
	381	+	linkEvents: linkEvents,
	382	+	rememberRemoteHost: rememberRemoteHost,
369	383		terminalInput: func() *terminalInputPump {
370	384		return input
371	385		},
372	386		}).run(ctx)
	387	+	stopHandshakeTransport()
	388	+	finishHandshake()
373	389		if result.welcomed {
374	390		backoff = defaultReconnectBackoff.initial
375	391		}

@@ -533,20 +549,24 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time
533	549		}
534	550
535	551		type attachAttempt struct {
536		-	runner *Runner
537		-	connection *SessionConnection
538		-	remote bool
539		-	transport ports.Transport
540		-	request AttachRequest
541		-	resumeToken uint64
542		-	clientID [16]byte
543		-	milestones *milestones
544		-	themeState *terminalThemeState
545		-	enterRaw func() error
546		-	reconnect *reconnectUI
547		-	linkEvents <-chan ports.LinkEvent
548		-	rememberRemoteHost func()
549		-	terminalInput func() *terminalInputPump
	552	+	runner *Runner
	553	+	connection *SessionConnection
	554	+	remote bool
	555	+	transport ports.Transport
	556	+	handshakeCtx context.Context
	557	+	handshakeTimedOut <-chan struct{}
	558	+	finishHandshake func()
	559	+	stopHandshakeTransport func()
	560	+	request AttachRequest
	561	+	resumeToken uint64
	562	+	clientID [16]byte
	563	+	milestones *milestones
	564	+	themeState *terminalThemeState
	565	+	enterRaw func() error
	566	+	reconnect *reconnectUI
	567	+	linkEvents <-chan ports.LinkEvent
	568	+	rememberRemoteHost func()
	569	+	terminalInput func() *terminalInputPump
550	570		}
551	571
552	572		type attachResult struct {

@@ -571,47 +591,6 @@ func (e *AttachTargetError) Error() string {
571	591		return fmt.Sprintf("client: attach target %q/%q", e.Target.Endpoint, e.Target.Session)
572	592		}
573	593
574		-	// boundedPreWelcome runs one blocking handshake operation. Transport methods
575		-	// do not accept a context, so cancellation and expiry close this attempt's
576		-	// transport to interrupt the call, then wait for its goroutine to exit.
577		-	func boundedPreWelcome(ctx context.Context, clk ports.Clock, transport ports.Transport, operation func() error) (bool, error) {
578		-	if err := ctx.Err(); err != nil {
579		-	_ = transport.Close()
580		-	return true, err
581		-	}
582		-
583		-	timer := clk.NewTimer(preWelcomeTimeout)
584		-	defer timer.Stop()
585		-	completed := make(chan error, 1)
586		-	go func() { completed <- operation() }()
587		-
588		-	select {
589		-	case err := <-completed:
590		-	// Prefer a cancellation/expiry observed concurrently with completion.
591		-	// That event owns the transport lifetime even though the operation has
592		-	// already returned by the time we observe it here.
593		-	if ctx.Err() != nil {
594		-	_ = transport.Close()
595		-	return true, ctx.Err()
596		-	}
597		-	select {
598		-	case <-timer.C():
599		-	_ = transport.Close()
600		-	return true, context.DeadlineExceeded
601		-	default:
602		-	return false, err
603		-	}
604		-	case <-ctx.Done():
605		-	_ = transport.Close()
606		-	<-completed
607		-	return true, ctx.Err()
608		-	case <-timer.C():
609		-	_ = transport.Close()
610		-	<-completed
611		-	return true, context.DeadlineExceeded
612		-	}
613		-	}
614		-
615	594		// paletteSlot validates the scanner's signed slot before narrowing it for the
616	595		// wire accumulator. Keep this boundary even though the in-tree Scanner also
617	596		// validates slots: callbacks are an interface boundary.

@@ -656,6 +635,41 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
656	635		clipboard := a.runner.clipboard
657	636		log := a.runner.logger
658	637		observer := a.runner.runtimeObserver
	638	+	handshakeCtx := a.handshakeCtx
	639	+	handshakeTimedOut := a.handshakeTimedOut
	640	+	finishHandshake := a.finishHandshake
	641	+	stopHandshakeTransport := a.stopHandshakeTransport
	642	+	if handshakeCtx == nil {
	643	+	handshakeCtx, handshakeTimedOut, finishHandshake = newHandshakeContext(ctx, clk)
	644	+	stopHandshakeTransport = watchHandshakeTransport(handshakeCtx, transport)
	645	+	}
	646	+	handshakeFinished := false
	647	+	endHandshake := func() {
	648	+	if handshakeFinished {
	649	+	return
	650	+	}
	651	+	handshakeFinished = true
	652	+	if stopHandshakeTransport != nil {
	653	+	stopHandshakeTransport()
	654	+	}
	655	+	if finishHandshake != nil {
	656	+	finishHandshake()
	657	+	}
	658	+	}
	659	+	defer endHandshake()
	660	+	handshakeFailure := func(stage string, err error) attachResult {
	661	+	if handshakeTimedOut != nil {
	662	+	err = handshakeContextError(ctx, handshakeTimedOut, err)
	663	+	}
	664	+	return attachResult{transportClosed: handshakeCtx.Err() != nil, err: fmt.Errorf("vev: %s: %w", stage, err)}
	665	+	}
	666	+	sendHandshake := func(operation func() error) error {
	667	+	err := boundedHandshakeOperation(handshakeCtx, transport, operation)
	668	+	if err != nil {
	669	+	return handshakeContextError(ctx, handshakeTimedOut, err)
	670	+	}
	671	+	return nil
	672	+	}
659	673		// 1. Handshake: send Hello with our size and TERM.
660	674		size, err := term.Size()
661	675		if err != nil {

@@ -682,22 +696,21 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
682	696		MaxOutputInFlight: requestedOutputWindow(transport),
683	697		Env: os.Environ(),
684	698		}
685		-	if closed, err := boundedPreWelcome(ctx, clk, transport, func() error {
	699	+	if err := sendHandshake(func() error {
686	700		return transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)})
687	701		}); err != nil {
688		-	return attachResult{transportClosed: closed, err: fmt.Errorf("vev: sending hello: %w", err)}
	702	+	return handshakeFailure("sending hello", err)
689	703		}
690	704		ms.helloSent = true
691	705
692	706		// 2. Await Welcome or a typed rejection.
693	707		var reply ports.Frame
694		-	closed, err := boundedPreWelcome(ctx, clk, transport, func() error {
	708	+	if err := boundedHandshakeOperation(handshakeCtx, transport, func() error {
695	709		var recvErr error
696	710		reply, recvErr = transport.Recv()
697	711		return recvErr
698		-	})
699		-	if err != nil {
700		-	return attachResult{transportClosed: closed, err: fmt.Errorf("vev: awaiting welcome: %w", err)}
	712	+	}); err != nil {
	713	+	return handshakeFailure("awaiting welcome", err)
701	714		}
702	715		result := func(welcomed bool, err error) attachResult {
703	716		return attachResult{resumeToken: resumeToken, sessionName: name, welcomed: welcomed, err: err}

@@ -726,6 +739,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
726	739		default:
727	740		return result(false, fmt.Errorf("vev: unexpected reply type %d before welcome", reply.Type))
728	741		}
	742	+	if err := handshakeCtx.Err(); err != nil {
	743	+	return handshakeFailure("validating welcome", err)
	744	+	}
729	745		welcomedResult := func(err error) attachResult { return result(true, err) }
730	746
731	747		// 3. Enter raw mode after Welcome; Run owns restoration.

@@ -787,7 +803,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
787	803		// keeps the generation publication ordered before the first query and
788	804		// avoids leaving a queued Theme behind an immediate detach.
789	805		if !senderStarted {
790		-	if err := transport.Send(frame); err != nil {
	806	+	if err := sendHandshake(func() error { return transport.Send(frame) }); err != nil {
791	807		return fmt.Errorf("sending initial theme: %w", err)
792	808		}
793	809		return nil

@@ -876,7 +892,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
876	892		if reconnect.remote {
877	893		// Before the asynchronous sender starts, preserve transport ordering by
878	894		// requesting the reset synchronously after the initial Theme publication.
879		-	closed, resetErr := boundedPreWelcome(ctx, clk, transport, func() error {
	895	+	resetErr := sendHandshake(func() error {
880	896		payload, err := ports.MarshalResize(ports.Resize{Size: size})
881	897		if err != nil {
882	898		return err

@@ -884,8 +900,9 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
884	900		return transport.Send(ports.Frame{Type: ports.MsgResize, Payload: payload})
885	901		})
886	902		if resetErr != nil {
887		-	resetResult := welcomedResult(fmt.Errorf("requesting reconnect reconciliation: %w", resetErr))
888		-	resetResult.transportClosed = closed
	903	+	resetErr = handshakeContextError(ctx, handshakeTimedOut, resetErr)
	904	+	resetResult := welcomedResult(fmt.Errorf("vev: requesting reconnect reconciliation: %w", resetErr))
	905	+	resetResult.transportClosed = handshakeCtx.Err() != nil
889	906		return resetResult
890	907		}
891	908		awaitingReconnectReset = true

@@ -1082,6 +1099,7 @@ func (a *attachAttempt) run(ctx context.Context) attachResult {
1082	1099		if !ms.firstOutput {
1083	1100		ms.firstOutput = true
1084	1101		log.Debug("received first output")
	1102	+	endHandshake()
1085	1103		}
1086	1104		case ports.MsgAttachTarget:
1087	1105		target, derr := ports.UnmarshalAttachTarget(r.frame.Payload)

 // goroutine, decoupling the input/resize pumps from transport back-pressure.
 const sendQueueDepth = 64
-// preWelcomeTimeout bounds each blocking operation before the daemon has
+// preWelcomeTimeout remains the focused-test name for the shared handshake
-// accepted the attach. Mosh uses the same 15-second startup budget.
+// budget, which now also covers connect and initial publication.
-const preWelcomeTimeout = 15 * time.Second
+const preWelcomeTimeout = ports.HandshakeTimeout
 // remoteHostLearnerShutdownTimeout bounds how long Run waits for async remote
 // host learning after terminal restoration. Learning is best-effort; a stalled
 	}
 	for {
-		transport, err := r.dialer.Dial(ctx)
+		handshakeCtx, timedOut, finishHandshake := newHandshakeContext(ctx, r.clock)
+		transport, err := boundedDial(handshakeCtx, r.dialer)
 		if err != nil {
+			err = handshakeContextError(ctx, timedOut, err)
+			finishHandshake()
 			if resumeToken == 0 || ctx.Err() != nil {
 				if clearErr := reconnect.clear(); clearErr != nil {
 					return errors.Join(err, clearErr)
 			continue
 		}
 		ms.dialed = true
+		if transport == nil {
+			finishHandshake()
+			return errors.New("vev: dialer returned nil transport")
+		}
+		stopHandshakeTransport := watchHandshakeTransport(handshakeCtx, transport)
 		connection, err := NewSessionConnection(transport, SessionTarget{
 			Intent:      attemptRequest.Intent,
 			SessionName: attemptRequest.SessionName,
 		})
 		if err != nil {
+			stopHandshakeTransport()
+			finishHandshake()
 			_ = transport.Close()
 			return err
 		}
 			linkEvents = reporter.LinkEvents()
 		}
 		result := (&attachAttempt{
-			runner:             r,
+			runner:                 r,
-			connection:         connection,
+			connection:             connection,
-			remote:             remote,
+			remote:                 remote,
-			request:            attemptRequest,
+			handshakeCtx:           handshakeCtx,
-			resumeToken:        resumeToken,
+			handshakeTimedOut:      timedOut,
-			clientID:           processClientID,
+			finishHandshake:        finishHandshake,
-			milestones:         &ms,
+			stopHandshakeTransport: stopHandshakeTransport,
-			themeState:         themeState,
+			request:                attemptRequest,
-			enterRaw:           enterRaw,
+			resumeToken:            resumeToken,
-			reconnect:          reconnect,
+			clientID:               processClientID,
-			linkEvents:         linkEvents,
+			milestones:             &ms,
-			rememberRemoteHost: rememberRemoteHost,
+			themeState:             themeState,
+			enterRaw:               enterRaw,
+			reconnect:              reconnect,
+			linkEvents:             linkEvents,
+			rememberRemoteHost:     rememberRemoteHost,
 			terminalInput: func() *terminalInputPump {
 				return input
 			},
 		}).run(ctx)
+		stopHandshakeTransport()
+		finishHandshake()
 		if result.welcomed {
 			backoff = defaultReconnectBackoff.initial
 		}
 }
 type attachAttempt struct {
-	runner             *Runner
+	runner                 *Runner
-	connection         *SessionConnection
+	connection             *SessionConnection
-	remote             bool
+	remote                 bool
-	transport          ports.Transport
+	transport              ports.Transport
-	request            AttachRequest
+	handshakeCtx           context.Context
-	resumeToken        uint64
+	handshakeTimedOut      <-chan struct{}
-	clientID           [16]byte
+	finishHandshake        func()
-	milestones         *milestones
+	stopHandshakeTransport func()
-	themeState         *terminalThemeState
+	request                AttachRequest
-	enterRaw           func() error
+	resumeToken            uint64
-	reconnect          *reconnectUI
+	clientID               [16]byte
-	linkEvents         <-chan ports.LinkEvent
+	milestones             *milestones
-	rememberRemoteHost func()
+	themeState             *terminalThemeState
-	terminalInput      func() *terminalInputPump
+	enterRaw               func() error
+	reconnect              *reconnectUI
+	linkEvents             <-chan ports.LinkEvent
+	rememberRemoteHost     func()
+	terminalInput          func() *terminalInputPump
 }
 type attachResult struct {
 	return fmt.Sprintf("client: attach target %q/%q", e.Target.Endpoint, e.Target.Session)
 }
-// boundedPreWelcome runs one blocking handshake operation. Transport methods
-// do not accept a context, so cancellation and expiry close this attempt's
-// transport to interrupt the call, then wait for its goroutine to exit.
-func boundedPreWelcome(ctx context.Context, clk ports.Clock, transport ports.Transport, operation func() error) (bool, error) {
-	if err := ctx.Err(); err != nil {
-		_ = transport.Close()
-		return true, err
-	}
-	timer := clk.NewTimer(preWelcomeTimeout)
-	defer timer.Stop()
-	completed := make(chan error, 1)
-	go func() { completed <- operation() }()
-	select {
-	case err := <-completed:
-		// Prefer a cancellation/expiry observed concurrently with completion.
-		// That event owns the transport lifetime even though the operation has
-		// already returned by the time we observe it here.
-		if ctx.Err() != nil {
-			_ = transport.Close()
-			return true, ctx.Err()
-		}
-		select {
-		case <-timer.C():
-			_ = transport.Close()
-			return true, context.DeadlineExceeded
-		default:
-			return false, err
-		}
-	case <-ctx.Done():
-		_ = transport.Close()
-		<-completed
-		return true, ctx.Err()
-	case <-timer.C():
-		_ = transport.Close()
-		<-completed
-		return true, context.DeadlineExceeded
-	}
-}
 // paletteSlot validates the scanner's signed slot before narrowing it for the
 // wire accumulator. Keep this boundary even though the in-tree Scanner also
 // validates slots: callbacks are an interface boundary.
 	clipboard := a.runner.clipboard
 	log := a.runner.logger
 	observer := a.runner.runtimeObserver
+	handshakeCtx := a.handshakeCtx
+	handshakeTimedOut := a.handshakeTimedOut
+	finishHandshake := a.finishHandshake
+	stopHandshakeTransport := a.stopHandshakeTransport
+	if handshakeCtx == nil {
+		handshakeCtx, handshakeTimedOut, finishHandshake = newHandshakeContext(ctx, clk)
+		stopHandshakeTransport = watchHandshakeTransport(handshakeCtx, transport)
+	}
+	handshakeFinished := false
+	endHandshake := func() {
+		if handshakeFinished {
+			return
+		}
+		handshakeFinished = true
+		if stopHandshakeTransport != nil {
+			stopHandshakeTransport()
+		}
+		if finishHandshake != nil {
+			finishHandshake()
+		}
+	}
+	defer endHandshake()
+	handshakeFailure := func(stage string, err error) attachResult {
+		if handshakeTimedOut != nil {
+			err = handshakeContextError(ctx, handshakeTimedOut, err)
+		}
+		return attachResult{transportClosed: handshakeCtx.Err() != nil, err: fmt.Errorf("vev: %s: %w", stage, err)}
+	}
+	sendHandshake := func(operation func() error) error {
+		err := boundedHandshakeOperation(handshakeCtx, transport, operation)
+		if err != nil {
+			return handshakeContextError(ctx, handshakeTimedOut, err)
+		}
+		return nil
+	}
 	// 1. Handshake: send Hello with our size and TERM.
 	size, err := term.Size()
 	if err != nil {
 		MaxOutputInFlight: requestedOutputWindow(transport),
 		Env:               os.Environ(),
 	}
-	if closed, err := boundedPreWelcome(ctx, clk, transport, func() error {
+	if err := sendHandshake(func() error {
 		return transport.Send(ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(hello)})
 	}); err != nil {
-		return attachResult{transportClosed: closed, err: fmt.Errorf("vev: sending hello: %w", err)}
+		return handshakeFailure("sending hello", err)
 	}
 	ms.helloSent = true
 	// 2. Await Welcome or a typed rejection.
 	var reply ports.Frame
-	closed, err := boundedPreWelcome(ctx, clk, transport, func() error {
+	if err := boundedHandshakeOperation(handshakeCtx, transport, func() error {
 		var recvErr error
 		reply, recvErr = transport.Recv()
 		return recvErr
-	})
+	}); err != nil {
-	if err != nil {
+		return handshakeFailure("awaiting welcome", err)
-		return attachResult{transportClosed: closed, err: fmt.Errorf("vev: awaiting welcome: %w", err)}
 	}
 	result := func(welcomed bool, err error) attachResult {
 		return attachResult{resumeToken: resumeToken, sessionName: name, welcomed: welcomed, err: err}
 	default:
 		return result(false, fmt.Errorf("vev: unexpected reply type %d before welcome", reply.Type))
 	}
+	if err := handshakeCtx.Err(); err != nil {
+		return handshakeFailure("validating welcome", err)
+	}
 	welcomedResult := func(err error) attachResult { return result(true, err) }
 	// 3. Enter raw mode after Welcome; Run owns restoration.
 		// keeps the generation publication ordered before the first query and
 		// avoids leaving a queued Theme behind an immediate detach.
 		if !senderStarted {
-			if err := transport.Send(frame); err != nil {
+			if err := sendHandshake(func() error { return transport.Send(frame) }); err != nil {
 				return fmt.Errorf("sending initial theme: %w", err)
 			}
 			return nil
 		if reconnect.remote {
 			// Before the asynchronous sender starts, preserve transport ordering by
 			// requesting the reset synchronously after the initial Theme publication.
-			closed, resetErr := boundedPreWelcome(ctx, clk, transport, func() error {
+			resetErr := sendHandshake(func() error {
 				payload, err := ports.MarshalResize(ports.Resize{Size: size})
 				if err != nil {
 					return err
 				return transport.Send(ports.Frame{Type: ports.MsgResize, Payload: payload})
 			})
 			if resetErr != nil {
-				resetResult := welcomedResult(fmt.Errorf("requesting reconnect reconciliation: %w", resetErr))
+				resetErr = handshakeContextError(ctx, handshakeTimedOut, resetErr)
-				resetResult.transportClosed = closed
+				resetResult := welcomedResult(fmt.Errorf("vev: requesting reconnect reconciliation: %w", resetErr))
+				resetResult.transportClosed = handshakeCtx.Err() != nil
 				return resetResult
 			}
 			awaitingReconnectReset = true
 				if !ms.firstOutput {
 					ms.firstOutput = true
 					log.Debug("received first output")
+					endHandshake()
 				}
 			case ports.MsgAttachTarget:
 				target, derr := ports.UnmarshalAttachTarget(r.frame.Payload)

M internal/usecase/client/client_test.go +28 -19


@@ -182,7 +182,7 @@ func attachTestDependencies(transport ports.Transport, terminal ports.Terminal,
182	182		return testDependencies(transportDialer{transport: transport}, terminal, clock, nil, nil)
183	183		}
184	184
185		-	const testPreWelcomeTimeout = 15 * time.Second
	185	+	const testPreWelcomeTimeout = ports.HandshakeTimeout
186	186
187	187		func newHandshakeClock(t testing.T, timerCount int) (portsmocks.MockClock, <-chan chan time.Time) {
188	188		t.Helper()

@@ -199,6 +199,26 @@ func newHandshakeClock(t testing.T, timerCount int) (portsmocks.MockClock, <-c
199	199		return clk, created
200	200		}
201	201
	202	+	func TestRunBoundsBlockedDial(t *testing.T) {
	203	+	clk, createdTimers := newHandshakeClock(t, 1)
	204	+	started := make(chan struct{})
	205	+	dialer := portsmocks.NewMockDialer(t)
	206	+	dialer.EXPECT().Dial(mock.Anything).RunAndReturn(func(ctx context.Context) (ports.Transport, error) {
	207	+	close(started)
	208	+	<-ctx.Done()
	209	+	return nil, ctx.Err()
	210	+	}).Once()
	211	+
	212	+	errCh := make(chan error, 1)
	213	+	go func() {
	214	+	errCh <- runTestClient(context.Background(), testDependencies(dialer, nil, clk, nil, nil), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main"})
	215	+	}()
	216	+	timerC := <-createdTimers
	217	+	<-started
	218	+	timerC <- time.Time{}
	219	+	require.ErrorIs(t, <-errCh, context.DeadlineExceeded)
	220	+	}
	221	+
202	222		func TestRunBoundsPreWelcomeOperations(t *testing.T) {
203	223		tests := []struct {
204	224		name string

@@ -216,11 +236,7 @@ func TestRunBoundsPreWelcomeOperations(t *testing.T) {
216	236		t.Run(tt.name, func(t *testing.T) {
217	237		ctx, cancel := context.WithCancel(context.Background())
218	238		defer cancel()
219		-	timerCount := 1
220		-	if tt.phase == "recv" {
221		-	timerCount = 2
222		-	}
223		-	clk, createdTimers := newHandshakeClock(t, timerCount)
	239	+	clk, createdTimers := newHandshakeClock(t, 1)
224	240
225	241		term := portsmocks.NewMockTerminal(t)
226	242		term.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()

@@ -228,9 +244,10 @@ func TestRunBoundsPreWelcomeOperations(t *testing.T) {
228	244
229	245		started := make(chan struct{})
230	246		closed := make(chan struct{})
	247	+	var closeOnce sync.Once
231	248		operationExited := make(chan struct{})
232	249		tr := portsmocks.NewMockTransport(t)
233		-	tr.EXPECT().Close().Run(func() { close(closed) }).Return(nil).Once()
	250	+	tr.EXPECT().Close().Run(func() { closeOnce.Do(func() { close(closed) }) }).Return(nil).Maybe()
234	251
235	252		block := func() error {
236	253		close(started)

@@ -260,21 +277,13 @@ func TestRunBoundsPreWelcomeOperations(t *testing.T) {
260	277		select {
261	278		case timerC = <-createdTimers:
262	279		case <-time.After(time.Second):
263		-	t.Fatal("pre-Welcome timer was not created")
	280	+	t.Fatal("handshake timer was not created")
264	281		}
265	282		select {
266	283		case <-started:
267	284		case <-time.After(time.Second):
268		-	t.Fatal("pre-Welcome operation did not start")
269		-	}
270		-	if tt.phase == "recv" {
271		-	select {
272		-	case timerC = <-createdTimers:
273		-	case <-time.After(time.Second):
274		-	t.Fatal("Welcome timer was not created")
275		-	}
	285	+	t.Fatal("handshake operation did not start")
276	286		}
277		-
278	287		if tt.end == "cancel" {
279	288		cancel()
280	289		} else {

@@ -285,12 +294,12 @@ func TestRunBoundsPreWelcomeOperations(t *testing.T) {
285	294		case err := <-errCh:
286	295		require.ErrorIs(t, err, tt.want)
287	296		case <-time.After(time.Second):
288		-	t.Fatal("Run did not return after pre-Welcome cancellation")
	297	+	t.Fatal("Run did not return after handshake cancellation")
289	298		}
290	299		select {
291	300		case <-operationExited:
292	301		case <-time.After(time.Second):
293		-	t.Fatal("pre-Welcome operation leaked after Run returned")
	302	+	t.Fatal("handshake operation leaked after Run returned")
294	303		}
295	304		tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
296	305		})

 	return testDependencies(transportDialer{transport: transport}, terminal, clock, nil, nil)
 }
-const testPreWelcomeTimeout = 15 * time.Second
+const testPreWelcomeTimeout = ports.HandshakeTimeout
 func newHandshakeClock(t *testing.T, timerCount int) (*portsmocks.MockClock, <-chan chan time.Time) {
 	t.Helper()
 	return clk, created
 }
+func TestRunBoundsBlockedDial(t *testing.T) {
+	clk, createdTimers := newHandshakeClock(t, 1)
+	started := make(chan struct{})
+	dialer := portsmocks.NewMockDialer(t)
+	dialer.EXPECT().Dial(mock.Anything).RunAndReturn(func(ctx context.Context) (ports.Transport, error) {
+		close(started)
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}).Once()
+	errCh := make(chan error, 1)
+	go func() {
+		errCh <- runTestClient(context.Background(), testDependencies(dialer, nil, clk, nil, nil), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main"})
+	}()
+	timerC := <-createdTimers
+	<-started
+	timerC <- time.Time{}
+	require.ErrorIs(t, <-errCh, context.DeadlineExceeded)
+}
 func TestRunBoundsPreWelcomeOperations(t *testing.T) {
 	tests := []struct {
 		name  string
 		t.Run(tt.name, func(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
-			timerCount := 1
+			clk, createdTimers := newHandshakeClock(t, 1)
-			if tt.phase == "recv" {
-				timerCount = 2
-			}
-			clk, createdTimers := newHandshakeClock(t, timerCount)
 			term := portsmocks.NewMockTerminal(t)
 			term.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
 			started := make(chan struct{})
 			closed := make(chan struct{})
+			var closeOnce sync.Once
 			operationExited := make(chan struct{})
 			tr := portsmocks.NewMockTransport(t)
-			tr.EXPECT().Close().Run(func() { close(closed) }).Return(nil).Once()
+			tr.EXPECT().Close().Run(func() { closeOnce.Do(func() { close(closed) }) }).Return(nil).Maybe()
 			block := func() error {
 				close(started)
 			select {
 			case timerC = <-createdTimers:
 			case <-time.After(time.Second):
-				t.Fatal("pre-Welcome timer was not created")
+				t.Fatal("handshake timer was not created")
 			}
 			select {
 			case <-started:
 			case <-time.After(time.Second):
-				t.Fatal("pre-Welcome operation did not start")
+				t.Fatal("handshake operation did not start")
-			}
-			if tt.phase == "recv" {
-				select {
-				case timerC = <-createdTimers:
-				case <-time.After(time.Second):
-					t.Fatal("Welcome timer was not created")
-				}
 			}
 			if tt.end == "cancel" {
 				cancel()
 			} else {
 			case err := <-errCh:
 				require.ErrorIs(t, err, tt.want)
 			case <-time.After(time.Second):
-				t.Fatal("Run did not return after pre-Welcome cancellation")
+				t.Fatal("Run did not return after handshake cancellation")
 			}
 			select {
 			case <-operationExited:
 			case <-time.After(time.Second):
-				t.Fatal("pre-Welcome operation leaked after Run returned")
+				t.Fatal("handshake operation leaked after Run returned")
 			}
 			tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
 		})

A internal/usecase/client/handshake.go +139 -0

@@ -0,0 +1,139 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/vev/internal/ports"
	10	+	)
	11	+
	12	+	var errHandshakeTimeout = errors.New("handshake timed out")
	13	+
	14	+	type systemClock struct{}
	15	+	type systemTimer struct{ *time.Timer }
	16	+
	17	+	func (systemClock) Now() time.Time { return time.Now() }
	18	+	func (systemClock) NewTimer(delay time.Duration) ports.Timer {
	19	+	return systemTimer{Timer: time.NewTimer(delay)}
	20	+	}
	21	+	func (t systemTimer) C() <-chan time.Time { return t.Timer.C }
	22	+	func (t systemTimer) Reset(d time.Duration) bool { return t.Timer.Reset(d) }
	23	+	func (t systemTimer) Stop() bool { return t.Timer.Stop() }
	24	+
	25	+	// newHandshakeContext owns one deadline for the complete outbound handshake.
	26	+	// The caller must stop it before entering the long-lived connection loop.
	27	+	func newHandshakeContext(parent context.Context, clock ports.Clock) (context.Context, <-chan struct{}, func()) {
	28	+	if parent == nil {
	29	+	parent = context.Background()
	30	+	}
	31	+	ctx, cancel := context.WithCancel(parent)
	32	+	timedOut := make(chan struct{})
	33	+	if clock == nil {
	34	+	clock = systemClock{}
	35	+	}
	36	+	timer := clock.NewTimer(ports.HandshakeTimeout)
	37	+	if timer == nil {
	38	+	timer = systemClock{}.NewTimer(ports.HandshakeTimeout)
	39	+	}
	40	+	stop := make(chan struct{})
	41	+	done := make(chan struct{})
	42	+	go func() {
	43	+	defer close(done)
	44	+	select {
	45	+	case <-timer.C():
	46	+	close(timedOut)
	47	+	cancel()
	48	+	case <-stop:
	49	+	}
	50	+	}()
	51	+	var once sync.Once
	52	+	finish := func() {
	53	+	once.Do(func() {
	54	+	close(stop)
	55	+	<-done
	56	+	timer.Stop()
	57	+	cancel()
	58	+	})
	59	+	}
	60	+	return ctx, timedOut, finish
	61	+	}
	62	+
	63	+	// watchHandshakeTransport closes a transport when the handshake context ends.
	64	+	// Transport.Close is required to interrupt blocked Send and Recv operations.
	65	+	func watchHandshakeTransport(ctx context.Context, transport ports.Transport) func() {
	66	+	stop := make(chan struct{})
	67	+	done := make(chan struct{})
	68	+	go func() {
	69	+	defer close(done)
	70	+	select {
	71	+	case <-ctx.Done():
	72	+	_ = transport.Close()
	73	+	case <-stop:
	74	+	}
	75	+	}()
	76	+	var once sync.Once
	77	+	return func() {
	78	+	once.Do(func() {
	79	+	close(stop)
	80	+	<-done
	81	+	})
	82	+	}
	83	+	}
	84	+
	85	+	func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
	86	+	if err := ctx.Err(); err != nil {
	87	+	_ = transport.Close()
	88	+	return err
	89	+	}
	90	+	completed := make(chan error, 1)
	91	+	go func() { completed <- operation() }()
	92	+	select {
	93	+	case err := <-completed:
	94	+	if ctx.Err() != nil {
	95	+	_ = transport.Close()
	96	+	return ctx.Err()
	97	+	}
	98	+	return err
	99	+	case <-ctx.Done():
	100	+	_ = transport.Close()
	101	+	<-completed
	102	+	return ctx.Err()
	103	+	}
	104	+	}
	105	+
	106	+	func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, error) {
	107	+	result := make(chan struct {
	108	+	transport ports.Transport
	109	+	err error
	110	+	}, 1)
	111	+	go func() {
	112	+	transport, err := dialer.Dial(ctx)
	113	+	if ctx.Err() != nil && transport != nil {
	114	+	_ = transport.Close()
	115	+	}
	116	+	result <- struct {
	117	+	transport ports.Transport
	118	+	err error
	119	+	}{transport: transport, err: err}
	120	+	}()
	121	+	select {
	122	+	case result := <-result:
	123	+	return result.transport, result.err
	124	+	case <-ctx.Done():
	125	+	return nil, ctx.Err()
	126	+	}
	127	+	}
	128	+
	129	+	func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fallback error) error {
	130	+	select {
	131	+	case <-timedOut:
	132	+	return errors.Join(errHandshakeTimeout, context.DeadlineExceeded)
	133	+	default:
	134	+	}
	135	+	if err := parent.Err(); err != nil {
	136	+	return err
	137	+	}
	138	+	return fallback
	139	+	}

@@ -0,0 +1,139 @@
	1	+	package client
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+	"time"
	8	+
	9	+	"github.com/bnema/vev/internal/ports"
	10	+	)
	11	+
	12	+	var errHandshakeTimeout = errors.New("handshake timed out")
	13	+
	14	+	type systemClock struct{}
	15	+	type systemTimer struct{ *time.Timer }
	16	+
	17	+	func (systemClock) Now() time.Time { return time.Now() }
	18	+	func (systemClock) NewTimer(delay time.Duration) ports.Timer {
	19	+	return systemTimer{Timer: time.NewTimer(delay)}
	20	+	}
	21	+	func (t systemTimer) C() <-chan time.Time { return t.Timer.C }
	22	+	func (t systemTimer) Reset(d time.Duration) bool { return t.Timer.Reset(d) }
	23	+	func (t systemTimer) Stop() bool { return t.Timer.Stop() }
	24	+
	25	+	// newHandshakeContext owns one deadline for the complete outbound handshake.
	26	+	// The caller must stop it before entering the long-lived connection loop.
	27	+	func newHandshakeContext(parent context.Context, clock ports.Clock) (context.Context, <-chan struct{}, func()) {
	28	+	if parent == nil {
	29	+	parent = context.Background()
	30	+	}
	31	+	ctx, cancel := context.WithCancel(parent)
	32	+	timedOut := make(chan struct{})
	33	+	if clock == nil {
	34	+	clock = systemClock{}
	35	+	}
	36	+	timer := clock.NewTimer(ports.HandshakeTimeout)
	37	+	if timer == nil {
	38	+	timer = systemClock{}.NewTimer(ports.HandshakeTimeout)
	39	+	}
	40	+	stop := make(chan struct{})
	41	+	done := make(chan struct{})
	42	+	go func() {
	43	+	defer close(done)
	44	+	select {
	45	+	case <-timer.C():
	46	+	close(timedOut)
	47	+	cancel()
	48	+	case <-stop:
	49	+	}
	50	+	}()
	51	+	var once sync.Once
	52	+	finish := func() {
	53	+	once.Do(func() {
	54	+	close(stop)
	55	+	<-done
	56	+	timer.Stop()
	57	+	cancel()
	58	+	})
	59	+	}
	60	+	return ctx, timedOut, finish
	61	+	}
	62	+
	63	+	// watchHandshakeTransport closes a transport when the handshake context ends.
	64	+	// Transport.Close is required to interrupt blocked Send and Recv operations.
	65	+	func watchHandshakeTransport(ctx context.Context, transport ports.Transport) func() {
	66	+	stop := make(chan struct{})
	67	+	done := make(chan struct{})
	68	+	go func() {
	69	+	defer close(done)
	70	+	select {
	71	+	case <-ctx.Done():
	72	+	_ = transport.Close()
	73	+	case <-stop:
	74	+	}
	75	+	}()
	76	+	var once sync.Once
	77	+	return func() {
	78	+	once.Do(func() {
	79	+	close(stop)
	80	+	<-done
	81	+	})
	82	+	}
	83	+	}
	84	+
	85	+	func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
	86	+	if err := ctx.Err(); err != nil {
	87	+	_ = transport.Close()
	88	+	return err
	89	+	}
	90	+	completed := make(chan error, 1)
	91	+	go func() { completed <- operation() }()
	92	+	select {
	93	+	case err := <-completed:
	94	+	if ctx.Err() != nil {
	95	+	_ = transport.Close()
	96	+	return ctx.Err()
	97	+	}
	98	+	return err
	99	+	case <-ctx.Done():
	100	+	_ = transport.Close()
	101	+	<-completed
	102	+	return ctx.Err()
	103	+	}
	104	+	}
	105	+
	106	+	func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, error) {
	107	+	result := make(chan struct {
	108	+	transport ports.Transport
	109	+	err error
	110	+	}, 1)
	111	+	go func() {
	112	+	transport, err := dialer.Dial(ctx)
	113	+	if ctx.Err() != nil && transport != nil {
	114	+	_ = transport.Close()
	115	+	}
	116	+	result <- struct {
	117	+	transport ports.Transport
	118	+	err error
	119	+	}{transport: transport, err: err}
	120	+	}()
	121	+	select {
	122	+	case result := <-result:
	123	+	return result.transport, result.err
	124	+	case <-ctx.Done():
	125	+	return nil, ctx.Err()
	126	+	}
	127	+	}
	128	+
	129	+	func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fallback error) error {
	130	+	select {
	131	+	case <-timedOut:
	132	+	return errors.Join(errHandshakeTimeout, context.DeadlineExceeded)
	133	+	default:
	134	+	}
	135	+	if err := parent.Err(); err != nil {
	136	+	return err
	137	+	}
	138	+	return fallback
	139	+	}

M internal/usecase/client/palette_attach_generation_test.go +37 -17


@@ -20,22 +20,33 @@ import (
20	20
21	21		// attachPaletteClock deliberately exposes each independently-created timer.
22	22		// Tests fire only the deadline they intend to exercise; no wall clock is used.
23		-	type attachPaletteClock struct{ timers chan *attachPaletteTimer }
	23	+	type attachPaletteClock struct {
	24	+	timers chan *attachPaletteTimer
	25	+	handshakeTimers chan *attachPaletteTimer
	26	+	}
24	27
25	28		type attachPaletteTimer struct {
26	29		ch chan time.Time
27	30		stopped chan struct{}
28	31		stopOnce sync.Once
	32	+	duration time.Duration
29	33		}
30	34
31	35		func newAttachPaletteClock() *attachPaletteClock {
32		-	return &attachPaletteClock{timers: make(chan *attachPaletteTimer, 16)}
	36	+	return &attachPaletteClock{
	37	+	timers: make(chan *attachPaletteTimer, 16),
	38	+	handshakeTimers: make(chan *attachPaletteTimer, 1),
	39	+	}
33	40		}
34	41
35	42		func (*attachPaletteClock) Now() time.Time { return time.Time{} }
36		-	func (c *attachPaletteClock) NewTimer(time.Duration) ports.Timer {
37		-	timer := &attachPaletteTimer{ch: make(chan time.Time, 1), stopped: make(chan struct{})}
38		-	c.timers <- timer
	43	+	func (c *attachPaletteClock) NewTimer(delay time.Duration) ports.Timer {
	44	+	timer := &attachPaletteTimer{ch: make(chan time.Time, 1), stopped: make(chan struct{}), duration: delay}
	45	+	if delay == ports.HandshakeTimeout {
	46	+	c.handshakeTimers <- timer
	47	+	} else {
	48	+	c.timers <- timer
	49	+	}
39	50		return timer
40	51		}
41	52		func (t *attachPaletteTimer) C() <-chan time.Time { return t.ch }

@@ -160,11 +171,12 @@ func (t *attachPaletteTransport) snapshots() []ports.Theme {
160	171		}
161	172
162	173		type attachPaletteHarness struct {
163		-	clock *attachPaletteClock
164		-	reader *attachPaletteReader
165		-	writer *attachPaletteWriter
166		-	transport *attachPaletteTransport
167		-	done chan attachResult
	174	+	clock *attachPaletteClock
	175	+	handshakeTimer *attachPaletteTimer
	176	+	reader *attachPaletteReader
	177	+	writer *attachPaletteWriter
	178	+	transport *attachPaletteTransport
	179	+	done chan attachResult
168	180		}
169	181
170	182		func startAttachPaletteHarness(state terminalThemeState) attachPaletteHarness {

@@ -196,13 +208,15 @@ func startAttachPaletteHarnessWithInputAndWriteError(state *terminalThemeState,
196	208		return input
197	209		},
198	210		}
199		-	h := &attachPaletteHarness{clock: clock, reader: reader, writer: writer, transport: transport, done: make(chan attachResult, 1)}
	211	+	h := &attachPaletteHarness{
	212	+	clock: clock,
	213	+	reader: reader,
	214	+	writer: writer,
	215	+	transport: transport,
	216	+	done: make(chan attachResult, 1),
	217	+	}
200	218		go func() { h.done <- attempt.run(context.Background()) }()
201		-	// The handshake's two bounded-pre-welcome timers are unrelated to palette
202		-	// acquisition. Remove them in this harness so every exposed timer is a
203		-	// generation deadline (or a timer deliberately created by input handling).
204		-	<-clock.timers
205		-	<-clock.timers
	219	+	h.handshakeTimer = <-clock.handshakeTimers
206	220		return h
207	221		}
208	222

@@ -221,7 +235,13 @@ func (h attachPaletteHarness) nextTimer(t testing.T) *attachPaletteTimer {
221	235		func (h attachPaletteHarness) detach(t testing.T) {
222	236		t.Helper()
223	237		h.transport.detached <- ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach})}
224		-	require.NoError(t, (<-h.done).err)
	238	+	result := <-h.done
	239	+	require.NoError(t, result.err)
	240	+	select {
	241	+	case <-h.handshakeTimer.stopped:
	242	+	default:
	243	+	t.Fatal("handshake timer was not canceled after initial publication")
	244	+	}
225	245		}
226	246
227	247		func paletteReply(foreground, background string, slot uint8, color string) string {

 // attachPaletteClock deliberately exposes each independently-created timer.
 // Tests fire only the deadline they intend to exercise; no wall clock is used.
-type attachPaletteClock struct{ timers chan *attachPaletteTimer }
+type attachPaletteClock struct {
+	timers          chan *attachPaletteTimer
+	handshakeTimers chan *attachPaletteTimer
+}
 type attachPaletteTimer struct {
 	ch       chan time.Time
 	stopped  chan struct{}
 	stopOnce sync.Once
+	duration time.Duration
 }
 func newAttachPaletteClock() *attachPaletteClock {
-	return &attachPaletteClock{timers: make(chan *attachPaletteTimer, 16)}
+	return &attachPaletteClock{
+		timers:          make(chan *attachPaletteTimer, 16),
+		handshakeTimers: make(chan *attachPaletteTimer, 1),
+	}
 }
 func (*attachPaletteClock) Now() time.Time { return time.Time{} }
-func (c *attachPaletteClock) NewTimer(time.Duration) ports.Timer {
+func (c *attachPaletteClock) NewTimer(delay time.Duration) ports.Timer {
-	timer := &attachPaletteTimer{ch: make(chan time.Time, 1), stopped: make(chan struct{})}
+	timer := &attachPaletteTimer{ch: make(chan time.Time, 1), stopped: make(chan struct{}), duration: delay}
-	c.timers <- timer
+	if delay == ports.HandshakeTimeout {
+		c.handshakeTimers <- timer
+	} else {
+		c.timers <- timer
+	}
 	return timer
 }
 func (t *attachPaletteTimer) C() <-chan time.Time    { return t.ch }
 }
 type attachPaletteHarness struct {
-	clock     *attachPaletteClock
+	clock          *attachPaletteClock
-	reader    *attachPaletteReader
+	handshakeTimer *attachPaletteTimer
-	writer    *attachPaletteWriter
+	reader         *attachPaletteReader
-	transport *attachPaletteTransport
+	writer         *attachPaletteWriter
-	done      chan attachResult
+	transport      *attachPaletteTransport
+	done           chan attachResult
 }
 func startAttachPaletteHarness(state *terminalThemeState) *attachPaletteHarness {
 			return input
 		},
 	}
-	h := &attachPaletteHarness{clock: clock, reader: reader, writer: writer, transport: transport, done: make(chan attachResult, 1)}
+	h := &attachPaletteHarness{
+		clock:     clock,
+		reader:    reader,
+		writer:    writer,
+		transport: transport,
+		done:      make(chan attachResult, 1),
+	}
 	go func() { h.done <- attempt.run(context.Background()) }()
-	// The handshake's two bounded-pre-welcome timers are unrelated to palette
+	h.handshakeTimer = <-clock.handshakeTimers
-	// acquisition. Remove them in this harness so every exposed timer is a
-	// generation deadline (or a timer deliberately created by input handling).
-	<-clock.timers
-	<-clock.timers
 	return h
 }
 func (h *attachPaletteHarness) detach(t *testing.T) {
 	t.Helper()
 	h.transport.detached <- ports.Frame{Type: ports.MsgDetached, Payload: ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach})}
-	require.NoError(t, (<-h.done).err)
+	result := <-h.done
+	require.NoError(t, result.err)
+	select {
+	case <-h.handshakeTimer.stopped:
+	default:
+		t.Fatal("handshake timer was not canceled after initial publication")
+	}
 }
 func paletteReply(foreground, background string, slot uint8, color string) string {

M internal/usecase/client/reconnect_toast_test.go +3 -5


@@ -936,7 +936,7 @@ func TestPreSenderReconnectResetIsBounded(t *testing.T) {
936	936		trigger: func(_ context.CancelFunc, timer *reconnectResetTimer) {
937	937		timer.ch <- time.Time{}
938	938		},
939		-	wantErr: context.DeadlineExceeded,
	939	+	wantErr: errHandshakeTimeout,
940	940		},
941	941		{
942	942		name: "cancellation",

@@ -961,15 +961,13 @@ func TestPreSenderReconnectResetIsBounded(t *testing.T) {
961	961		resultCh := make(chan attachResult, 1)
962	962		go func() { resultCh <- attempt.run(ctx) }()
963	963
964		-	<-clock.preWelcome // Hello Send
965		-	<-clock.preWelcome // Welcome Recv
966		-	resetTimer := <-clock.preWelcome
	964	+	handshakeTimer := <-clock.preWelcome
967	965		select {
968	966		case <-tr.entered:
969	967		case <-time.After(time.Second):
970	968		t.Fatal("timed out waiting for synchronous reconnect reset Send")
971	969		}
972		-	tt.trigger(cancel, resetTimer)
	970	+	tt.trigger(cancel, handshakeTimer)
973	971
974	972		result := requireAttachResult(t, resultCh)
975	973		require.ErrorIs(t, result.err, tt.wantErr)

 			trigger: func(_ context.CancelFunc, timer *reconnectResetTimer) {
 				timer.ch <- time.Time{}
 			},
-			wantErr: context.DeadlineExceeded,
+			wantErr: errHandshakeTimeout,
 		},
 		{
 			name: "cancellation",
 			resultCh := make(chan attachResult, 1)
 			go func() { resultCh <- attempt.run(ctx) }()
-			<-clock.preWelcome // Hello Send
+			handshakeTimer := <-clock.preWelcome
-			<-clock.preWelcome // Welcome Recv
-			resetTimer := <-clock.preWelcome
 			select {
 			case <-tr.entered:
 			case <-time.After(time.Second):
 				t.Fatal("timed out waiting for synchronous reconnect reset Send")
 			}
-			tt.trigger(cancel, resetTimer)
+			tt.trigger(cancel, handshakeTimer)
 			result := requireAttachResult(t, resultCh)
 			require.ErrorIs(t, result.err, tt.wantErr)

M internal/usecase/daemon/daemon.go +87 -28


@@ -945,32 +945,50 @@ func (d Daemon) sessionsSnapshotLocked() []session {
945	945		}
946	946
947	947		// handleConn reads the first frame off a fresh connection and routes it. A
948		-	// context watcher closes the transport on serve-context cancel so a handler
949		-	// parked in Recv (mid-handshake or between input frames) unwinds on shutdown.
	948	+	// context watcher closes the transport on serve-context cancel or handshake
	949	+	// timeout so blocked transport operations unwind.
950	950		func (d *Daemon) handleConn(tr ports.Transport) {
951		-	stop := context.AfterFunc(d.hardCtx, func() { _ = tr.Close() })
952		-	defer stop()
953		-
954		-	first, err := tr.Recv()
	951	+	handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(d.hardCtx)
	952	+	stopTransport := watchHandshakeTransport(handshakeCtx, tr)
	953	+	defer finishHandshake()
	954	+	defer stopTransport()
	955	+
	956	+	var first ports.Frame
	957	+	err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
	958	+	var recvErr error
	959	+	first, recvErr = tr.Recv()
	960	+	return recvErr
	961	+	})
955	962		if err != nil {
	963	+	err = handshakeContextError(d.hardCtx, timedOut, err)
956	964		d.log.Warn("connection closed before hello", "err", err)
957	965		_ = tr.Close()
958	966		return
959	967		}
960	968		switch first.Type {
961	969		case ports.MsgList:
	970	+	stopTransport()
	971	+	finishHandshake()
962	972		d.handleList(tr)
963	973		case ports.MsgCommand:
	974	+	stopTransport()
	975	+	finishHandshake()
964	976		if err := d.handleCommand(tr, first); err != nil {
965	977		d.log.Warn("command handler failed", "err", err)
966	978		}
967	979		case ports.MsgKill:
	980	+	stopTransport()
	981	+	finishHandshake()
968	982		d.handleKill(tr, first)
969	983		case ports.MsgHello:
970		-	d.handleHello(tr, first)
	984	+	d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, first)
971	985		default:
972	986		d.log.Warn("hello rejected", "err", "expected hello", "type", first.Type)
973		-	_ = tr.Send(frameError(ports.ErrInternal, "expected hello"))
	987	+	_ = boundedHandshakeOperation(handshakeCtx, tr, func() error {
	988	+	return tr.Send(frameError(ports.ErrInternal, "expected hello"))
	989	+	})
	990	+	stopTransport()
	991	+	finishHandshake()
974	992		_ = tr.Close()
975	993		}
976	994		}

@@ -1060,24 +1078,46 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1060	1078		}
1061	1079		}
1062	1080
1063		-	// handleHello runs the attach handshake: version check, intent routing,
1064		-	// Welcome, guaranteed first paint, then the per-connection input loop.
	1081	+	// handleHello runs the attach handshake for direct package callers. Accepted
	1082	+	// connections use handleHelloWithContext so the deadline also covers the first
	1083	+	// frame read in handleConn.
1065	1084		func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
	1085	+	handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(context.Background())
	1086	+	stopTransport := watchHandshakeTransport(handshakeCtx, tr)
	1087	+	defer finishHandshake()
	1088	+	defer stopTransport()
	1089	+	d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, f)
	1090	+	}
	1091	+
	1092	+	// handleHelloWithContext runs the attach handshake: version check, intent
	1093	+	// routing, Welcome, guaranteed first paint, then the per-connection input loop.
	1094	+	func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <-chan struct{}, stopHandshakeTransport, finishHandshake func(), tr ports.Transport, f ports.Frame) {
	1095	+	if err := handshakeContextError(context.Background(), timedOut, handshakeCtx.Err()); err != nil {
	1096	+	_ = tr.Close()
	1097	+	return
	1098	+	}
	1099	+	sendHandshake := func(frame ports.Frame) error {
	1100	+	err := boundedHandshakeOperation(handshakeCtx, tr, func() error { return tr.Send(frame) })
	1101	+	if err != nil {
	1102	+	return handshakeContextError(context.Background(), timedOut, err)
	1103	+	}
	1104	+	return nil
	1105	+	}
1066	1106		h, err := ports.UnmarshalHello(f.Payload)
1067	1107		if err != nil {
1068	1108		if version, ok := ports.PeekHelloVersion(f.Payload); ok && version != ports.ProtocolVersion {
1069	1109		d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", version, "expected", ports.ProtocolVersion)
1070		-	_ = tr.Send(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
	1110	+	_ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1071	1111		} else {
1072	1112		d.log.Warn("hello rejected", "err", err)
1073		-	_ = tr.Send(frameError(ports.ErrInternal, "malformed hello"))
	1113	+	_ = sendHandshake(frameError(ports.ErrInternal, "malformed hello"))
1074	1114		}
1075	1115		_ = tr.Close()
1076	1116		return
1077	1117		}
1078	1118		if h.Version != ports.ProtocolVersion {
1079	1119		d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", h.Version, "expected", ports.ProtocolVersion, "intent", h.Intent, "session", h.Name)
1080		-	_ = tr.Send(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
	1120	+	_ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1081	1121		_ = tr.Close()
1082	1122		return
1083	1123		}

@@ -1086,14 +1126,25 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1086	1126		if rerr != nil {
1087	1127		d.log.Warn("hello rejected", "err", rerr, "intent", h.Intent, "session", h.Name)
1088	1128		if pe, ok := errors.AsType[*protoErr](rerr); ok {
1089		-	_ = tr.Send(frameError(pe.code, pe.text))
	1129	+	_ = sendHandshake(frameError(pe.code, pe.text))
1090	1130		} else {
1091		-	_ = tr.Send(frameError(ports.ErrInternal, rerr.Error()))
	1131	+	_ = sendHandshake(frameError(ports.ErrInternal, rerr.Error()))
1092	1132		}
1093	1133		_ = tr.Close()
1094	1134		return
1095	1135		}
1096	1136
	1137	+	if err := handshakeCtx.Err(); err != nil {
	1138	+	return
	1139	+	}
	1140	+	failAttachment := func() {
	1141	+	if !d.abortResumeClaim(ac) {
	1142	+	// A fresh attachment has no credential to park. Explicit teardown
	1143	+	// removes the unpublished attachment instead of creating a resumable
	1144	+	// entry for a handshake that never committed Welcome.
	1145	+	d.clientGone(sess, ac, tr, true)
	1146	+	}
	1147	+	}
1097	1148		rc := sess.renderCoordinator()
1098	1149		lease := (*attachmentLease)(nil)
1099	1150		if rc != nil {

@@ -1112,18 +1163,16 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1112	1163		}
1113	1164		return
1114	1165		}
1115		-	if err := ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, ac), welcomeTicket); err != nil {
	1166	+	if err := handshakeCtx.Err(); err != nil {
1116	1167		welcomeTicket.End()
1117		-	if !d.abortResumeClaim(ac) {
1118		-	d.clientGone(sess, ac, tr, false)
1119		-	}
	1168	+	failAttachment()
1120	1169		return
1121	1170		}
1122		-	if !d.commitResumeClaim(ac) {
	1171	+	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
	1172	+	return ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, ac), welcomeTicket)
	1173	+	}); err != nil {
1123	1174		welcomeTicket.End()
1124		-	if !d.abortResumeClaim(ac) {
1125		-	d.clientGone(sess, ac, tr, false)
1126		-	}
	1175	+	failAttachment()
1127	1176		return
1128	1177		}
1129	1178		// Release Welcome's effect before discovering post-handshake authority so a

@@ -1131,28 +1180,38 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1131	1180		welcomeTicket.End()
1132	1181		postWelcomeToken, postWelcomeTicket, admitted := ac.beginCurrentAttachmentEffect(sess, tr)
1133	1182		if !admitted {
1134		-	d.clientGone(sess, ac, tr, false)
	1183	+	failAttachment()
1135	1184		return
1136	1185		}
1137	1186		postWelcomeLease := postWelcomeToken.lease
1138	1187		if postWelcomeToken.ac == nil {
1139	1188		postWelcomeTicket.End()
1140		-	d.clientGone(sess, ac, tr, false)
	1189	+	failAttachment()
1141	1190		return
1142	1191		}
1143	1192		if postWelcomeLease != nil && (rc == nil \|\| !rc.markAttachmentReady(postWelcomeLease)) {
1144	1193		postWelcomeTicket.End()
1145	1194		// The attachment was detached while Welcome was in flight; never let
1146	1195		// this stale handshake emit an Output frame.
1147		-	d.clientGone(sess, ac, tr, false)
	1196	+	failAttachment()
1148	1197		return
1149	1198		}
1150	1199		postWelcomeTicket.End()
1151	1200		paintToken := sess.attachmentToken(ac, tr)
1152		-	if !d.firstPaintForTransition(paintToken) {
1153		-	d.clientGone(sess, ac, tr, false)
	1201	+	painted := false
	1202	+	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
	1203	+	painted = d.firstPaintForTransition(paintToken)
	1204	+	return nil
	1205	+	}); err != nil \|\| !painted \|\| handshakeCtx.Err() != nil {
	1206	+	failAttachment()
	1207	+	return
	1208	+	}
	1209	+	if !d.commitResumeClaim(ac) {
	1210	+	failAttachment()
1154	1211		return
1155	1212		}
	1213	+	stopHandshakeTransport()
	1214	+	finishHandshake()
1156	1215		d.runConnLoop(ac)
1157	1216		_ = tr.Close()
1158	1217		}


@@ -945,32 +945,50 @@ func (d Daemon) sessionsSnapshotLocked() []session {
945		}	945		}
946			946
947		// handleConn reads the first frame off a fresh connection and routes it. A	947		// handleConn reads the first frame off a fresh connection and routes it. A
948	-	// context watcher closes the transport on serve-context cancel so a handler	948	+	// context watcher closes the transport on serve-context cancel or handshake
949	-	// parked in Recv (mid-handshake or between input frames) unwinds on shutdown.	949	+	// timeout so blocked transport operations unwind.
950		func (d *Daemon) handleConn(tr ports.Transport) {	950		func (d *Daemon) handleConn(tr ports.Transport) {
951	-	stop := context.AfterFunc(d.hardCtx, func() { _ = tr.Close() })	951	+	handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(d.hardCtx)
952	-	defer stop()	952	+	stopTransport := watchHandshakeTransport(handshakeCtx, tr)
953	-		953	+	defer finishHandshake()
954	-	first, err := tr.Recv()	954	+	defer stopTransport()
			955	+
			956	+	var first ports.Frame
			957	+	err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
			958	+	var recvErr error
			959	+	first, recvErr = tr.Recv()
			960	+	return recvErr
			961	+	})
955		if err != nil {	962		if err != nil {
			963	+	err = handshakeContextError(d.hardCtx, timedOut, err)
956		d.log.Warn("connection closed before hello", "err", err)	964		d.log.Warn("connection closed before hello", "err", err)
957		_ = tr.Close()	965		_ = tr.Close()
958		return	966		return
959		}	967		}
960		switch first.Type {	968		switch first.Type {
961		case ports.MsgList:	969		case ports.MsgList:
			970	+	stopTransport()
			971	+	finishHandshake()
962		d.handleList(tr)	972		d.handleList(tr)
963		case ports.MsgCommand:	973		case ports.MsgCommand:
			974	+	stopTransport()
			975	+	finishHandshake()
964		if err := d.handleCommand(tr, first); err != nil {	976		if err := d.handleCommand(tr, first); err != nil {
965		d.log.Warn("command handler failed", "err", err)	977		d.log.Warn("command handler failed", "err", err)
966		}	978		}
967		case ports.MsgKill:	979		case ports.MsgKill:
			980	+	stopTransport()
			981	+	finishHandshake()
968		d.handleKill(tr, first)	982		d.handleKill(tr, first)
969		case ports.MsgHello:	983		case ports.MsgHello:
970	-	d.handleHello(tr, first)	984	+	d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, first)
971		default:	985		default:
972		d.log.Warn("hello rejected", "err", "expected hello", "type", first.Type)	986		d.log.Warn("hello rejected", "err", "expected hello", "type", first.Type)
973	-	_ = tr.Send(frameError(ports.ErrInternal, "expected hello"))	987	+	_ = boundedHandshakeOperation(handshakeCtx, tr, func() error {
			988	+	return tr.Send(frameError(ports.ErrInternal, "expected hello"))
			989	+	})
			990	+	stopTransport()
			991	+	finishHandshake()
974		_ = tr.Close()	992		_ = tr.Close()
975		}	993		}
976		}	994		}

@@ -1060,24 +1078,46 @@ func (d *Daemon) handleKill(tr ports.Transport, f ports.Frame) {
1060		}	1078		}
1061		}	1079		}
1062			1080
1063	-	// handleHello runs the attach handshake: version check, intent routing,	1081	+	// handleHello runs the attach handshake for direct package callers. Accepted
1064	-	// Welcome, guaranteed first paint, then the per-connection input loop.	1082	+	// connections use handleHelloWithContext so the deadline also covers the first
			1083	+	// frame read in handleConn.
1065		func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {	1084		func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
			1085	+	handshakeCtx, timedOut, finishHandshake := d.newHandshakeContext(context.Background())
			1086	+	stopTransport := watchHandshakeTransport(handshakeCtx, tr)
			1087	+	defer finishHandshake()
			1088	+	defer stopTransport()
			1089	+	d.handleHelloWithContext(handshakeCtx, timedOut, stopTransport, finishHandshake, tr, f)
			1090	+	}
			1091	+
			1092	+	// handleHelloWithContext runs the attach handshake: version check, intent
			1093	+	// routing, Welcome, guaranteed first paint, then the per-connection input loop.
			1094	+	func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <-chan struct{}, stopHandshakeTransport, finishHandshake func(), tr ports.Transport, f ports.Frame) {
			1095	+	if err := handshakeContextError(context.Background(), timedOut, handshakeCtx.Err()); err != nil {
			1096	+	_ = tr.Close()
			1097	+	return
			1098	+	}
			1099	+	sendHandshake := func(frame ports.Frame) error {
			1100	+	err := boundedHandshakeOperation(handshakeCtx, tr, func() error { return tr.Send(frame) })
			1101	+	if err != nil {
			1102	+	return handshakeContextError(context.Background(), timedOut, err)
			1103	+	}
			1104	+	return nil
			1105	+	}
1066		h, err := ports.UnmarshalHello(f.Payload)	1106		h, err := ports.UnmarshalHello(f.Payload)
1067		if err != nil {	1107		if err != nil {
1068		if version, ok := ports.PeekHelloVersion(f.Payload); ok && version != ports.ProtocolVersion {	1108		if version, ok := ports.PeekHelloVersion(f.Payload); ok && version != ports.ProtocolVersion {
1069		d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", version, "expected", ports.ProtocolVersion)	1109		d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", version, "expected", ports.ProtocolVersion)
1070	-	_ = tr.Send(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))	1110	+	_ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1071		} else {	1111		} else {
1072		d.log.Warn("hello rejected", "err", err)	1112		d.log.Warn("hello rejected", "err", err)
1073	-	_ = tr.Send(frameError(ports.ErrInternal, "malformed hello"))	1113	+	_ = sendHandshake(frameError(ports.ErrInternal, "malformed hello"))
1074		}	1114		}
1075		_ = tr.Close()	1115		_ = tr.Close()
1076		return	1116		return
1077		}	1117		}
1078		if h.Version != ports.ProtocolVersion {	1118		if h.Version != ports.ProtocolVersion {
1079		d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", h.Version, "expected", ports.ProtocolVersion, "intent", h.Intent, "session", h.Name)	1119		d.log.Warn("hello rejected", "err", "protocol version mismatch", "version", h.Version, "expected", ports.ProtocolVersion, "intent", h.Intent, "session", h.Name)
1080	-	_ = tr.Send(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))	1120	+	_ = sendHandshake(frameError(ports.ErrVersionMismatch, "protocol version mismatch"))
1081		_ = tr.Close()	1121		_ = tr.Close()
1082		return	1122		return
1083		}	1123		}

@@ -1086,14 +1126,25 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1086		if rerr != nil {	1126		if rerr != nil {
1087		d.log.Warn("hello rejected", "err", rerr, "intent", h.Intent, "session", h.Name)	1127		d.log.Warn("hello rejected", "err", rerr, "intent", h.Intent, "session", h.Name)
1088		if pe, ok := errors.AsType[*protoErr](rerr); ok {	1128		if pe, ok := errors.AsType[*protoErr](rerr); ok {
1089	-	_ = tr.Send(frameError(pe.code, pe.text))	1129	+	_ = sendHandshake(frameError(pe.code, pe.text))
1090		} else {	1130		} else {
1091	-	_ = tr.Send(frameError(ports.ErrInternal, rerr.Error()))	1131	+	_ = sendHandshake(frameError(ports.ErrInternal, rerr.Error()))
1092		}	1132		}
1093		_ = tr.Close()	1133		_ = tr.Close()
1094		return	1134		return
1095		}	1135		}
1096			1136
			1137	+	if err := handshakeCtx.Err(); err != nil {
			1138	+	return
			1139	+	}
			1140	+	failAttachment := func() {
			1141	+	if !d.abortResumeClaim(ac) {
			1142	+	// A fresh attachment has no credential to park. Explicit teardown
			1143	+	// removes the unpublished attachment instead of creating a resumable
			1144	+	// entry for a handshake that never committed Welcome.
			1145	+	d.clientGone(sess, ac, tr, true)
			1146	+	}
			1147	+	}
1097		rc := sess.renderCoordinator()	1148		rc := sess.renderCoordinator()
1098		lease := (*attachmentLease)(nil)	1149		lease := (*attachmentLease)(nil)
1099		if rc != nil {	1150		if rc != nil {

@@ -1112,18 +1163,16 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1112		}	1163		}
1113		return	1164		return
1114		}	1165		}
1115	-	if err := ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, ac), welcomeTicket); err != nil {	1166	+	if err := handshakeCtx.Err(); err != nil {
1116		welcomeTicket.End()	1167		welcomeTicket.End()
1117	-	if !d.abortResumeClaim(ac) {	1168	+	failAttachment()
1118	-	d.clientGone(sess, ac, tr, false)
1119	-	}
1120		return	1169		return
1121		}	1170		}
1122	-	if !d.commitResumeClaim(ac) {	1171	+	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
			1172	+	return ac.sendExpectedTransportForAttachment(expected, frameWelcome(sess, ac), welcomeTicket)
			1173	+	}); err != nil {
1123		welcomeTicket.End()	1174		welcomeTicket.End()
1124	-	if !d.abortResumeClaim(ac) {	1175	+	failAttachment()
1125	-	d.clientGone(sess, ac, tr, false)
1126	-	}
1127		return	1176		return
1128		}	1177		}
1129		// Release Welcome's effect before discovering post-handshake authority so a	1178		// Release Welcome's effect before discovering post-handshake authority so a

@@ -1131,28 +1180,38 @@ func (d *Daemon) handleHello(tr ports.Transport, f ports.Frame) {
1131		welcomeTicket.End()	1180		welcomeTicket.End()
1132		postWelcomeToken, postWelcomeTicket, admitted := ac.beginCurrentAttachmentEffect(sess, tr)	1181		postWelcomeToken, postWelcomeTicket, admitted := ac.beginCurrentAttachmentEffect(sess, tr)
1133		if !admitted {	1182		if !admitted {
1134	-	d.clientGone(sess, ac, tr, false)	1183	+	failAttachment()
1135		return	1184		return
1136		}	1185		}
1137		postWelcomeLease := postWelcomeToken.lease	1186		postWelcomeLease := postWelcomeToken.lease
1138		if postWelcomeToken.ac == nil {	1187		if postWelcomeToken.ac == nil {
1139		postWelcomeTicket.End()	1188		postWelcomeTicket.End()
1140	-	d.clientGone(sess, ac, tr, false)	1189	+	failAttachment()
1141		return	1190		return
1142		}	1191		}
1143		if postWelcomeLease != nil && (rc == nil \|\| !rc.markAttachmentReady(postWelcomeLease)) {	1192		if postWelcomeLease != nil && (rc == nil \|\| !rc.markAttachmentReady(postWelcomeLease)) {
1144		postWelcomeTicket.End()	1193		postWelcomeTicket.End()
1145		// The attachment was detached while Welcome was in flight; never let	1194		// The attachment was detached while Welcome was in flight; never let
1146		// this stale handshake emit an Output frame.	1195		// this stale handshake emit an Output frame.
1147	-	d.clientGone(sess, ac, tr, false)	1196	+	failAttachment()
1148		return	1197		return
1149		}	1198		}
1150		postWelcomeTicket.End()	1199		postWelcomeTicket.End()
1151		paintToken := sess.attachmentToken(ac, tr)	1200		paintToken := sess.attachmentToken(ac, tr)
1152	-	if !d.firstPaintForTransition(paintToken) {	1201	+	painted := false
1153	-	d.clientGone(sess, ac, tr, false)	1202	+	if err := boundedHandshakeOperation(handshakeCtx, tr, func() error {
			1203	+	painted = d.firstPaintForTransition(paintToken)
			1204	+	return nil
			1205	+	}); err != nil \|\| !painted \|\| handshakeCtx.Err() != nil {
			1206	+	failAttachment()
			1207	+	return
			1208	+	}
			1209	+	if !d.commitResumeClaim(ac) {
			1210	+	failAttachment()
1154		return	1211		return
1155		}	1212		}
			1213	+	stopHandshakeTransport()
			1214	+	finishHandshake()
1156		d.runConnLoop(ac)	1215		d.runConnLoop(ac)
1157		_ = tr.Close()	1216		_ = tr.Close()
1158		}	1217		}

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


@@ -664,6 +664,7 @@ func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) {
664	664		}()
665	665
666	666		awaitTestCompletion(t, tr.welcomeEntered, "timed out waiting for Welcome send")
	667	+	handshakeTimer := awaitHandshakeTimer(t, clock)
667	668		welcome := awaitFrame(t, tr.sends, ports.MsgWelcome)
668	669		require.Equal(t, ports.MsgWelcome, welcome.Type)
669	670		sess := firstSession(d)

@@ -694,6 +695,7 @@ func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) {
694	695		require.Zero(t, first.Base, "the first post-Welcome frame is full")
695	696		tr.finish()
696	697		awaitTestCompletion(t, done, "timed out waiting for Welcome handler completion")
	698	+	awaitTestCompletion(t, handshakeTimer.stopped, "handshake timer was not canceled after publication")
697	699		requireNoCoordinatorOutputFrame(t, tr.sends)
698	700		}
699	701


@@ -664,6 +664,7 @@ func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) {
664	}()	664		}()
665		665
666	awaitTestCompletion(t, tr.welcomeEntered, "timed out waiting for Welcome send")	666		awaitTestCompletion(t, tr.welcomeEntered, "timed out waiting for Welcome send")
		667	+	handshakeTimer := awaitHandshakeTimer(t, clock)
667	welcome := awaitFrame(t, tr.sends, ports.MsgWelcome)	668		welcome := awaitFrame(t, tr.sends, ports.MsgWelcome)
668	require.Equal(t, ports.MsgWelcome, welcome.Type)	669		require.Equal(t, ports.MsgWelcome, welcome.Type)
669	sess := firstSession(d)	670		sess := firstSession(d)

@@ -694,6 +695,7 @@ func TestHandleHelloDefersFreshOutputUntilWelcome(t *testing.T) {
694	require.Zero(t, first.Base, "the first post-Welcome frame is full")	695		require.Zero(t, first.Base, "the first post-Welcome frame is full")
695	tr.finish()	696		tr.finish()
696	awaitTestCompletion(t, done, "timed out waiting for Welcome handler completion")	697		awaitTestCompletion(t, done, "timed out waiting for Welcome handler completion")
		698	+	awaitTestCompletion(t, handshakeTimer.stopped, "handshake timer was not canceled after publication")
697	requireNoCoordinatorOutputFrame(t, tr.sends)	699		requireNoCoordinatorOutputFrame(t, tr.sends)
698	}	700		}
699		701

A internal/usecase/daemon/handshake.go +109 -0

@@ -0,0 +1,109 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	)
	10	+
	11	+	var errHandshakeTimeout = errors.New("handshake timed out")
	12	+
	13	+	// newHandshakeContext owns one deadline for the complete inbound handshake.
	14	+	// The caller must stop it before entering the long-lived connection loop.
	15	+	func (d *Daemon) newHandshakeContext(parent context.Context) (context.Context, <-chan struct{}, func()) {
	16	+	if parent == nil {
	17	+	parent = context.Background()
	18	+	}
	19	+	ctx, cancel := context.WithCancel(parent)
	20	+	timedOut := make(chan struct{})
	21	+	clock := ports.Clock(systemClock{})
	22	+	if d != nil && d.clock != nil {
	23	+	clock = d.clock
	24	+	}
	25	+	timer := clock.NewTimer(ports.HandshakeTimeout)
	26	+	if timer == nil {
	27	+	timer = systemClock{}.NewTimer(ports.HandshakeTimeout)
	28	+	}
	29	+	stop := make(chan struct{})
	30	+	done := make(chan struct{})
	31	+	go func() {
	32	+	defer close(done)
	33	+	select {
	34	+	case <-timer.C():
	35	+	close(timedOut)
	36	+	cancel()
	37	+	case <-stop:
	38	+	}
	39	+	}()
	40	+	var once sync.Once
	41	+	finish := func() {
	42	+	once.Do(func() {
	43	+	close(stop)
	44	+	<-done
	45	+	timer.Stop()
	46	+	cancel()
	47	+	})
	48	+	}
	49	+	return ctx, timedOut, finish
	50	+	}
	51	+
	52	+	// watchHandshakeTransport closes a transport when the handshake context ends.
	53	+	// Transport.Close is required to interrupt blocked Send and Recv operations.
	54	+	func watchHandshakeTransport(ctx context.Context, transport ports.Transport) func() {
	55	+	stop := make(chan struct{})
	56	+	done := make(chan struct{})
	57	+	go func() {
	58	+	defer close(done)
	59	+	select {
	60	+	case <-ctx.Done():
	61	+	_ = transport.Close()
	62	+	case <-stop:
	63	+	}
	64	+	}()
	65	+	var once sync.Once
	66	+	return func() {
	67	+	once.Do(func() {
	68	+	close(stop)
	69	+	<-done
	70	+	})
	71	+	}
	72	+	}
	73	+
	74	+	// boundedHandshakeOperation waits for one transport operation while the shared
	75	+	// handshake context remains live. Closing the transport is the interruption
	76	+	// mechanism because ports.Transport methods do not accept contexts.
	77	+	func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
	78	+	if err := ctx.Err(); err != nil {
	79	+	_ = transport.Close()
	80	+	return err
	81	+	}
	82	+	completed := make(chan error, 1)
	83	+	go func() { completed <- operation() }()
	84	+	select {
	85	+	case err := <-completed:
	86	+	if ctx.Err() != nil {
	87	+	_ = transport.Close()
	88	+	<-completed
	89	+	return ctx.Err()
	90	+	}
	91	+	return err
	92	+	case <-ctx.Done():
	93	+	_ = transport.Close()
	94	+	<-completed
	95	+	return ctx.Err()
	96	+	}
	97	+	}
	98	+
	99	+	func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fallback error) error {
	100	+	select {
	101	+	case <-timedOut:
	102	+	return errors.Join(errHandshakeTimeout, context.DeadlineExceeded)
	103	+	default:
	104	+	}
	105	+	if err := parent.Err(); err != nil {
	106	+	return err
	107	+	}
	108	+	return fallback
	109	+	}

@@ -0,0 +1,109 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"sync"
	7	+
	8	+	"github.com/bnema/vev/internal/ports"
	9	+	)
	10	+
	11	+	var errHandshakeTimeout = errors.New("handshake timed out")
	12	+
	13	+	// newHandshakeContext owns one deadline for the complete inbound handshake.
	14	+	// The caller must stop it before entering the long-lived connection loop.
	15	+	func (d *Daemon) newHandshakeContext(parent context.Context) (context.Context, <-chan struct{}, func()) {
	16	+	if parent == nil {
	17	+	parent = context.Background()
	18	+	}
	19	+	ctx, cancel := context.WithCancel(parent)
	20	+	timedOut := make(chan struct{})
	21	+	clock := ports.Clock(systemClock{})
	22	+	if d != nil && d.clock != nil {
	23	+	clock = d.clock
	24	+	}
	25	+	timer := clock.NewTimer(ports.HandshakeTimeout)
	26	+	if timer == nil {
	27	+	timer = systemClock{}.NewTimer(ports.HandshakeTimeout)
	28	+	}
	29	+	stop := make(chan struct{})
	30	+	done := make(chan struct{})
	31	+	go func() {
	32	+	defer close(done)
	33	+	select {
	34	+	case <-timer.C():
	35	+	close(timedOut)
	36	+	cancel()
	37	+	case <-stop:
	38	+	}
	39	+	}()
	40	+	var once sync.Once
	41	+	finish := func() {
	42	+	once.Do(func() {
	43	+	close(stop)
	44	+	<-done
	45	+	timer.Stop()
	46	+	cancel()
	47	+	})
	48	+	}
	49	+	return ctx, timedOut, finish
	50	+	}
	51	+
	52	+	// watchHandshakeTransport closes a transport when the handshake context ends.
	53	+	// Transport.Close is required to interrupt blocked Send and Recv operations.
	54	+	func watchHandshakeTransport(ctx context.Context, transport ports.Transport) func() {
	55	+	stop := make(chan struct{})
	56	+	done := make(chan struct{})
	57	+	go func() {
	58	+	defer close(done)
	59	+	select {
	60	+	case <-ctx.Done():
	61	+	_ = transport.Close()
	62	+	case <-stop:
	63	+	}
	64	+	}()
	65	+	var once sync.Once
	66	+	return func() {
	67	+	once.Do(func() {
	68	+	close(stop)
	69	+	<-done
	70	+	})
	71	+	}
	72	+	}
	73	+
	74	+	// boundedHandshakeOperation waits for one transport operation while the shared
	75	+	// handshake context remains live. Closing the transport is the interruption
	76	+	// mechanism because ports.Transport methods do not accept contexts.
	77	+	func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, operation func() error) error {
	78	+	if err := ctx.Err(); err != nil {
	79	+	_ = transport.Close()
	80	+	return err
	81	+	}
	82	+	completed := make(chan error, 1)
	83	+	go func() { completed <- operation() }()
	84	+	select {
	85	+	case err := <-completed:
	86	+	if ctx.Err() != nil {
	87	+	_ = transport.Close()
	88	+	<-completed
	89	+	return ctx.Err()
	90	+	}
	91	+	return err
	92	+	case <-ctx.Done():
	93	+	_ = transport.Close()
	94	+	<-completed
	95	+	return ctx.Err()
	96	+	}
	97	+	}
	98	+
	99	+	func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fallback error) error {
	100	+	select {
	101	+	case <-timedOut:
	102	+	return errors.Join(errHandshakeTimeout, context.DeadlineExceeded)
	103	+	default:
	104	+	}
	105	+	if err := parent.Err(); err != nil {
	106	+	return err
	107	+	}
	108	+	return fallback
	109	+	}

A internal/usecase/daemon/handshake_test.go +142 -0

@@ -0,0 +1,142 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"io"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	)
	14	+
	15	+	type handshakeBlockingTransport struct {
	16	+	blockWelcome bool
	17	+	welcome chan struct{}
	18	+	closed chan struct{}
	19	+	closeOnce sync.Once
	20	+	}
	21	+
	22	+	func newHandshakeBlockingTransport(blockWelcome bool) *handshakeBlockingTransport {
	23	+	return &handshakeBlockingTransport{
	24	+	blockWelcome: blockWelcome,
	25	+	welcome: make(chan struct{}),
	26	+	closed: make(chan struct{}),
	27	+	}
	28	+	}
	29	+
	30	+	func (t *handshakeBlockingTransport) Send(frame ports.Frame) error {
	31	+	if t.blockWelcome && frame.Type == ports.MsgWelcome {
	32	+	close(t.welcome)
	33	+	<-t.closed
	34	+	return io.ErrClosedPipe
	35	+	}
	36	+	return nil
	37	+	}
	38	+
	39	+	func (t *handshakeBlockingTransport) Recv() (ports.Frame, error) {
	40	+	<-t.closed
	41	+	return ports.Frame{}, io.ErrClosedPipe
	42	+	}
	43	+
	44	+	func (t *handshakeBlockingTransport) Close() error {
	45	+	t.closeOnce.Do(func() { close(t.closed) })
	46	+	return nil
	47	+	}
	48	+
	49	+	func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) {
	50	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	51	+	d := newTestDaemon(t, nil, clock)
	52	+	tr := newHandshakeBlockingTransport(false)
	53	+	done := make(chan struct{})
	54	+	go func() {
	55	+	d.handleConn(tr)
	56	+	close(done)
	57	+	}()
	58	+
	59	+	timer := <-clock.timers
	60	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	61	+	timer.ch <- time.Time{}
	62	+	awaitTestCompletion(t, done, "handshake receive did not stop after timeout")
	63	+	requireClosedHandshakeTransport(t, tr)
	64	+	d.mu.Lock()
	65	+	require.Empty(t, d.sessions)
	66	+	require.Empty(t, d.parked)
	67	+	d.mu.Unlock()
	68	+	}
	69	+
	70	+	func TestFailedResumeHandshakeRestoresParkedCredential(t *testing.T) {
	71	+	pty, release := newBlockingPTY(t)
	72	+	defer release()
	73	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	74	+	oldTransport := newHandshakeBlockingTransport(false)
	75	+	sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "resume-timeout", 0), oldTransport)
	76	+	require.NoError(t, err)
	77	+	token := ac.resumeToken
	78	+	d.clientGone(sess, ac, oldTransport, false)
	79	+	require.Empty(t, sess.snapshotAttachments())
	80	+
	81	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	82	+	d.clock = clock
	83	+	resumeTransport := newHandshakeBlockingTransport(true)
	84	+	done := make(chan struct{})
	85	+	go func() {
	86	+	d.handleHello(resumeTransport, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(helloResumeCapable(ports.IntentResume, sess.name, token))})
	87	+	close(done)
	88	+	}()
	89	+
	90	+	timer := <-clock.timers
	91	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	92	+	<-resumeTransport.welcome
	93	+	timer.ch <- time.Time{}
	94	+	awaitTestCompletion(t, done, "failed resume handshake did not finish")
	95	+	requireClosedHandshakeTransport(t, resumeTransport)
	96	+	d.mu.Lock()
	97	+	parked := d.parked[token]
	98	+	d.mu.Unlock()
	99	+	require.NotNil(t, parked, "failed pre-claim resume must restore the parked credential")
	100	+	require.Same(t, ac, parked.ac)
	101	+	require.Equal(t, token, ac.resumeToken)
	102	+	require.True(t, ac.parked)
	103	+	require.Empty(t, sess.snapshotAttachments())
	104	+	}
	105	+
	106	+	func TestHandshakeTimeoutClosesBlockedWelcomeSend(t *testing.T) {
	107	+	pty, release := newBlockingPTY(t)
	108	+	defer release()
	109	+	clock := &signalClock{timers: make(chan *signalTimer, 4)}
	110	+	d := newTestDaemon(t, newFactory(t, pty), clock)
	111	+	tr := newHandshakeBlockingTransport(true)
	112	+	done := make(chan struct{})
	113	+	go func() {
	114	+	d.handleHello(tr, mustHello(ports.IntentNew, "timeout-send", domain.Size{Cols: 80, Rows: 24}))
	115	+	close(done)
	116	+	}()
	117	+
	118	+	timer := <-clock.timers
	119	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	120	+	<-tr.welcome
	121	+	timer.ch <- time.Time{}
	122	+	awaitTestCompletion(t, done, "handshake send did not stop after timeout")
	123	+	requireClosedHandshakeTransport(t, tr)
	124	+	d.mu.Lock()
	125	+	require.Len(t, d.sessions, 1)
	126	+	var sess *session
	127	+	for _, entry := range d.sessions {
	128	+	sess, _ = localSession(entry)
	129	+	}
	130	+	require.Empty(t, sess.snapshotAttachments())
	131	+	require.Empty(t, d.parked)
	132	+	d.mu.Unlock()
	133	+	}
	134	+
	135	+	func requireClosedHandshakeTransport(t testing.T, tr handshakeBlockingTransport) {
	136	+	t.Helper()
	137	+	select {
	138	+	case <-tr.closed:
	139	+	default:
	140	+	t.Fatal("handshake timeout did not close transport")
	141	+	}
	142	+	}

@@ -0,0 +1,142 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"io"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	)
	14	+
	15	+	type handshakeBlockingTransport struct {
	16	+	blockWelcome bool
	17	+	welcome chan struct{}
	18	+	closed chan struct{}
	19	+	closeOnce sync.Once
	20	+	}
	21	+
	22	+	func newHandshakeBlockingTransport(blockWelcome bool) *handshakeBlockingTransport {
	23	+	return &handshakeBlockingTransport{
	24	+	blockWelcome: blockWelcome,
	25	+	welcome: make(chan struct{}),
	26	+	closed: make(chan struct{}),
	27	+	}
	28	+	}
	29	+
	30	+	func (t *handshakeBlockingTransport) Send(frame ports.Frame) error {
	31	+	if t.blockWelcome && frame.Type == ports.MsgWelcome {
	32	+	close(t.welcome)
	33	+	<-t.closed
	34	+	return io.ErrClosedPipe
	35	+	}
	36	+	return nil
	37	+	}
	38	+
	39	+	func (t *handshakeBlockingTransport) Recv() (ports.Frame, error) {
	40	+	<-t.closed
	41	+	return ports.Frame{}, io.ErrClosedPipe
	42	+	}
	43	+
	44	+	func (t *handshakeBlockingTransport) Close() error {
	45	+	t.closeOnce.Do(func() { close(t.closed) })
	46	+	return nil
	47	+	}
	48	+
	49	+	func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) {
	50	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	51	+	d := newTestDaemon(t, nil, clock)
	52	+	tr := newHandshakeBlockingTransport(false)
	53	+	done := make(chan struct{})
	54	+	go func() {
	55	+	d.handleConn(tr)
	56	+	close(done)
	57	+	}()
	58	+
	59	+	timer := <-clock.timers
	60	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	61	+	timer.ch <- time.Time{}
	62	+	awaitTestCompletion(t, done, "handshake receive did not stop after timeout")
	63	+	requireClosedHandshakeTransport(t, tr)
	64	+	d.mu.Lock()
	65	+	require.Empty(t, d.sessions)
	66	+	require.Empty(t, d.parked)
	67	+	d.mu.Unlock()
	68	+	}
	69	+
	70	+	func TestFailedResumeHandshakeRestoresParkedCredential(t *testing.T) {
	71	+	pty, release := newBlockingPTY(t)
	72	+	defer release()
	73	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	74	+	oldTransport := newHandshakeBlockingTransport(false)
	75	+	sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "resume-timeout", 0), oldTransport)
	76	+	require.NoError(t, err)
	77	+	token := ac.resumeToken
	78	+	d.clientGone(sess, ac, oldTransport, false)
	79	+	require.Empty(t, sess.snapshotAttachments())
	80	+
	81	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	82	+	d.clock = clock
	83	+	resumeTransport := newHandshakeBlockingTransport(true)
	84	+	done := make(chan struct{})
	85	+	go func() {
	86	+	d.handleHello(resumeTransport, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(helloResumeCapable(ports.IntentResume, sess.name, token))})
	87	+	close(done)
	88	+	}()
	89	+
	90	+	timer := <-clock.timers
	91	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	92	+	<-resumeTransport.welcome
	93	+	timer.ch <- time.Time{}
	94	+	awaitTestCompletion(t, done, "failed resume handshake did not finish")
	95	+	requireClosedHandshakeTransport(t, resumeTransport)
	96	+	d.mu.Lock()
	97	+	parked := d.parked[token]
	98	+	d.mu.Unlock()
	99	+	require.NotNil(t, parked, "failed pre-claim resume must restore the parked credential")
	100	+	require.Same(t, ac, parked.ac)
	101	+	require.Equal(t, token, ac.resumeToken)
	102	+	require.True(t, ac.parked)
	103	+	require.Empty(t, sess.snapshotAttachments())
	104	+	}
	105	+
	106	+	func TestHandshakeTimeoutClosesBlockedWelcomeSend(t *testing.T) {
	107	+	pty, release := newBlockingPTY(t)
	108	+	defer release()
	109	+	clock := &signalClock{timers: make(chan *signalTimer, 4)}
	110	+	d := newTestDaemon(t, newFactory(t, pty), clock)
	111	+	tr := newHandshakeBlockingTransport(true)
	112	+	done := make(chan struct{})
	113	+	go func() {
	114	+	d.handleHello(tr, mustHello(ports.IntentNew, "timeout-send", domain.Size{Cols: 80, Rows: 24}))
	115	+	close(done)
	116	+	}()
	117	+
	118	+	timer := <-clock.timers
	119	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	120	+	<-tr.welcome
	121	+	timer.ch <- time.Time{}
	122	+	awaitTestCompletion(t, done, "handshake send did not stop after timeout")
	123	+	requireClosedHandshakeTransport(t, tr)
	124	+	d.mu.Lock()
	125	+	require.Len(t, d.sessions, 1)
	126	+	var sess *session
	127	+	for _, entry := range d.sessions {
	128	+	sess, _ = localSession(entry)
	129	+	}
	130	+	require.Empty(t, sess.snapshotAttachments())
	131	+	require.Empty(t, d.parked)
	132	+	d.mu.Unlock()
	133	+	}
	134	+
	135	+	func requireClosedHandshakeTransport(t testing.T, tr handshakeBlockingTransport) {
	136	+	t.Helper()
	137	+	select {
	138	+	case <-tr.closed:
	139	+	default:
	140	+	t.Fatal("handshake timeout did not close transport")
	141	+	}
	142	+	}

M internal/usecase/daemon/render_coordinator_test.go +30 -7


@@ -23,25 +23,30 @@ import (
23	23		// coordinatorMockClock uses generated mocks while retaining deterministic timer
24	24		// channels and recorded deadlines for coordinator scheduling assertions.
25	25		type coordinatorMockClock struct {
26		-	clock *portsmocks.MockClock
27		-	timers chan *coordinatorMockTimer
	26	+	clock *portsmocks.MockClock
	27	+	timers chan *coordinatorMockTimer
	28	+	handshakeTimers chan *coordinatorMockTimer
28	29		}
29	30
30	31		type coordinatorMockTimer struct {
31	32		mock *portsmocks.MockTimer
32	33		ch chan time.Time
33	34		duration time.Duration
	35	+	stopped chan struct{}
	36	+	stopOnce sync.Once
34	37		}
35	38
36	39		func newCoordinatorMockClock(t testing.T, capacity int) coordinatorMockClock {
37	40		t.Helper()
38	41		clk := &coordinatorMockClock{
39		-	clock: portsmocks.NewMockClock(t),
40		-	timers: make(chan *coordinatorMockTimer, capacity),
	42	+	clock: portsmocks.NewMockClock(t),
	43	+	timers: make(chan *coordinatorMockTimer, capacity),
	44	+	handshakeTimers: make(chan *coordinatorMockTimer, 1),
41	45		}
42	46		clk.clock.EXPECT().Now().Return(time.Time{}).Maybe()
43	47		clk.clock.EXPECT().NewTimer(mock.MatchedBy(func(d time.Duration) bool {
44		-	return d == urgentRenderDeadline \|\|
	48	+	return d == ports.HandshakeTimeout \|\|
	49	+	d == urgentRenderDeadline \|\|
45	50		(d >= minOutputRenderDeadline && d <= maxOutputRenderDeadline) \|\|
46	51		d == maxSyncUpdateDuration \|\|
47	52		d == time.Second \|\|

@@ -50,14 +55,21 @@ func newCoordinatorMockClock(t testing.T, capacity int) coordinatorMockClock {
50	55		timer := &coordinatorMockTimer{
51	56		mock: portsmocks.NewMockTimer(t),
52	57		duration: d,
	58	+	stopped: make(chan struct{}),
53	59		}
54	60		timer.ch = make(chan time.Time, 1)
55	61		// A normal deadline reads C both in its worker and in the inert-clock
56	62		// guard; a watchdog reads it in its worker. The generated expectation
57	63		// makes every channel access observable without a hand-written Timer.
58	64		timer.mock.EXPECT().C().Maybe().Return((<-chan time.Time)(timer.ch))
59		-	timer.mock.EXPECT().Stop().Maybe().Return(true)
60		-	clk.timers <- timer
	65	+	timer.mock.EXPECT().Stop().Run(func() {
	66	+	timer.stopOnce.Do(func() { close(timer.stopped) })
	67	+	}).Maybe().Return(true)
	68	+	if d == ports.HandshakeTimeout {
	69	+	clk.handshakeTimers <- timer
	70	+	} else {
	71	+	clk.timers <- timer
	72	+	}
61	73		return timer.mock
62	74		}).Maybe()
63	75		return clk

@@ -219,6 +231,17 @@ func captureResizeCallbackDone(t testing.T, rc renderCoordinator) <-chan struc
219	231		return rc.resizeLane.token.done
220	232		}
221	233
	234	+	func awaitHandshakeTimer(t testing.T, clk coordinatorMockClock) *coordinatorMockTimer {
	235	+	t.Helper()
	236	+	select {
	237	+	case tm := <-clk.handshakeTimers:
	238	+	return tm
	239	+	default:
	240	+	t.Fatal("handshake did not synchronously arm a fake-clock timer")
	241	+	return nil
	242	+	}
	243	+	}
	244	+
222	245		func awaitCoordinatorScheduledTimer(t testing.T, clk coordinatorMockClock) *coordinatorMockTimer {
223	246		t.Helper()
224	247		select {

 // coordinatorMockClock uses generated mocks while retaining deterministic timer
 // channels and recorded deadlines for coordinator scheduling assertions.
 type coordinatorMockClock struct {
-	clock  *portsmocks.MockClock
+	clock           *portsmocks.MockClock
-	timers chan *coordinatorMockTimer
+	timers          chan *coordinatorMockTimer
+	handshakeTimers chan *coordinatorMockTimer
 }
 type coordinatorMockTimer struct {
 	mock     *portsmocks.MockTimer
 	ch       chan time.Time
 	duration time.Duration
+	stopped  chan struct{}
+	stopOnce sync.Once
 }
 func newCoordinatorMockClock(t *testing.T, capacity int) *coordinatorMockClock {
 	t.Helper()
 	clk := &coordinatorMockClock{
-		clock:  portsmocks.NewMockClock(t),
+		clock:           portsmocks.NewMockClock(t),
-		timers: make(chan *coordinatorMockTimer, capacity),
+		timers:          make(chan *coordinatorMockTimer, capacity),
+		handshakeTimers: make(chan *coordinatorMockTimer, 1),
 	}
 	clk.clock.EXPECT().Now().Return(time.Time{}).Maybe()
 	clk.clock.EXPECT().NewTimer(mock.MatchedBy(func(d time.Duration) bool {
-		return d == urgentRenderDeadline ||
+		return d == ports.HandshakeTimeout ||
+			d == urgentRenderDeadline ||
 			(d >= minOutputRenderDeadline && d <= maxOutputRenderDeadline) ||
 			d == maxSyncUpdateDuration ||
 			d == time.Second ||
 		timer := &coordinatorMockTimer{
 			mock:     portsmocks.NewMockTimer(t),
 			duration: d,
+			stopped:  make(chan struct{}),
 		}
 		timer.ch = make(chan time.Time, 1)
 		// A normal deadline reads C both in its worker and in the inert-clock
 		// guard; a watchdog reads it in its worker. The generated expectation
 		// makes every channel access observable without a hand-written Timer.
 		timer.mock.EXPECT().C().Maybe().Return((<-chan time.Time)(timer.ch))
-		timer.mock.EXPECT().Stop().Maybe().Return(true)
+		timer.mock.EXPECT().Stop().Run(func() {
-		clk.timers <- timer
+			timer.stopOnce.Do(func() { close(timer.stopped) })
+		}).Maybe().Return(true)
+		if d == ports.HandshakeTimeout {
+			clk.handshakeTimers <- timer
+		} else {
+			clk.timers <- timer
+		}
 		return timer.mock
 	}).Maybe()
 	return clk
 	return rc.resizeLane.token.done
 }
+func awaitHandshakeTimer(t *testing.T, clk *coordinatorMockClock) *coordinatorMockTimer {
+	t.Helper()
+	select {
+	case tm := <-clk.handshakeTimers:
+		return tm
+	default:
+		t.Fatal("handshake did not synchronously arm a fake-clock timer")
+		return nil
+	}
+}
 func awaitCoordinatorScheduledTimer(t *testing.T, clk *coordinatorMockClock) *coordinatorMockTimer {
 	t.Helper()
 	select {

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


@@ -258,6 +258,7 @@ func TestHandleHelloResumeDefersFreshOutputUntilWelcome(t *testing.T) {
258	258		}()
259	259
260	260		<-tr.welcomeEntered
	261	+	handshakeTimer := awaitHandshakeTimer(t, clock)
261	262		welcome := <-tr.sends
262	263		require.Equal(t, ports.MsgWelcome, welcome.Type)
263	264		sess.mu.Lock()

@@ -283,6 +284,7 @@ func TestHandleHelloResumeDefersFreshOutputUntilWelcome(t *testing.T) {
283	284		require.Zero(t, first.Base)
284	285		tr.finish()
285	286		<-done
	287	+	awaitTestCompletion(t, handshakeTimer.stopped, "handshake timer was not canceled after publication")
286	288		requireNoCoordinatorOutputFrame(t, tr.sends)
287	289		}
288	290


@@ -258,6 +258,7 @@ func TestHandleHelloResumeDefersFreshOutputUntilWelcome(t *testing.T) {
258	}()	258		}()
259		259
260	<-tr.welcomeEntered	260		<-tr.welcomeEntered
		261	+	handshakeTimer := awaitHandshakeTimer(t, clock)
261	welcome := <-tr.sends	262		welcome := <-tr.sends
262	require.Equal(t, ports.MsgWelcome, welcome.Type)	263		require.Equal(t, ports.MsgWelcome, welcome.Type)
263	sess.mu.Lock()	264		sess.mu.Lock()

@@ -283,6 +284,7 @@ func TestHandleHelloResumeDefersFreshOutputUntilWelcome(t *testing.T) {
283	require.Zero(t, first.Base)	284		require.Zero(t, first.Base)
284	tr.finish()	285		tr.finish()
285	<-done	286		<-done
		287	+	awaitTestCompletion(t, handshakeTimer.stopped, "handshake timer was not canceled after publication")
286	requireNoCoordinatorOutputFrame(t, tr.sends)	288		requireNoCoordinatorOutputFrame(t, tr.sends)
287	}	289		}
288		290