dea913c fix(phase4): integrate handshake validation and command deadlines · vev

M internal/app/cmd.go +39 -15


@@ -10,9 +10,11 @@ import (
10	10		"strings"
11	11		"text/tabwriter"
12	12
	13	+	"github.com/bnema/vev/internal/adapters/clock"
13	14		"github.com/bnema/vev/internal/adapters/ipc"
14	15		"github.com/bnema/vev/internal/ports"
15	16		commandusecase "github.com/bnema/vev/internal/usecase/command"
	17	+	"github.com/bnema/vev/internal/usecase/daemon"
16	18		)
17	19
18	20		type cmdInvocation struct {

@@ -126,10 +128,11 @@ type cmdDeps struct {
126	128		stdout io.Writer
127	129		getenv func(string) string
128	130		dial func(context.Context, string) (ports.Transport, error)
	131	+	clock ports.Clock
129	132		}
130	133
131	134		func runCmd(ctx context.Context, invocation cmdInvocation) error {
132		-	return runCmdWithDeps(ctx, invocation, cmdDeps{stdout: os.Stdout, getenv: os.Getenv, dial: realDial})
	135	+	return runCmdWithDeps(ctx, invocation, cmdDeps{stdout: os.Stdout, getenv: os.Getenv, dial: realDial, clock: clock.New()})
133	136		}
134	137
135	138		func runCmdWithDeps(ctx context.Context, invocation cmdInvocation, deps cmdDeps) error {

@@ -157,11 +160,9 @@ func runCmdWithDeps(ctx context.Context, invocation cmdInvocation, deps cmdDeps)
157	160		return usagef("--self requires running inside a vev pane")
158	161		}
159	162
160		-	payload, err := ports.MarshalCommandRequest(request)
161		-	if errors.Is(err, ports.ErrTooManyCommandArgs) {
	163	+	if _, err := ports.MarshalCommandRequest(request); errors.Is(err, ports.ErrTooManyCommandArgs) {
162	164		return usagef("too many command arguments")
163		-	}
164		-	if err != nil {
	165	+	} else if err != nil {
165	166		return fmt.Errorf("encoding command: %w", err)
166	167		}
167	168

@@ -171,20 +172,43 @@ func runCmdWithDeps(ctx context.Context, invocation cmdInvocation, deps cmdDeps)
171	172		}
172	173		defer func() { _ = transport.Close() }()
173	174
174		-	frame := ports.Frame{Type: ports.MsgCommand, Payload: payload}
175		-	if err := transport.Send(frame); err != nil {
176		-	return &exitCoded{code: 3, err: fmt.Errorf("sending command: %w", err)}
177		-	}
178		-	reply, err := transport.Recv()
	175	+	tracker := daemon.NewCommandRequestTracker()
	176	+	const generation = uint64(1)
	177	+	requestID, outcome := tracker.Publish(generation)
	178	+	request.RequestID = requestID
	179	+	payload, err := ports.MarshalCommandRequest(request)
179	180		if err != nil {
180		-	return &exitCoded{code: 3, err: fmt.Errorf("reading command reply: %w", err)}
	181	+	tracker.Remove(requestID, generation)
	182	+	return fmt.Errorf("encoding command: %w", err)
	183	+	}
	184	+	if err := transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
	185	+	tracker.Remove(requestID, generation)
	186	+	return &exitCoded{code: 3, err: fmt.Errorf("sending command: %w", err)}
181	187		}
182		-	if reply.Type != ports.MsgCommandResult {
183		-	return fmt.Errorf("unexpected command reply type %d", reply.Type)
	188	+	go func() {
	189	+	reply, recvErr := transport.Recv()
	190	+	if recvErr != nil {
	191	+	tracker.Fail(requestID, generation, fmt.Errorf("reading command reply: %w", recvErr))
	192	+	return
	193	+	}
	194	+	if reply.Type != ports.MsgCommandResult {
	195	+	tracker.Fail(requestID, generation, fmt.Errorf("unexpected command reply type %d", reply.Type))
	196	+	return
	197	+	}
	198	+	result, decodeErr := ports.UnmarshalCommandResult(reply.Payload)
	199	+	if decodeErr != nil {
	200	+	tracker.Fail(requestID, generation, fmt.Errorf("decoding command reply: %w", decodeErr))
	201	+	return
	202	+	}
	203	+	tracker.Complete(generation, result)
	204	+	}()
	205	+	commandClock := deps.clock
	206	+	if commandClock == nil {
	207	+	commandClock = clock.New()
184	208		}
185		-	result, err := ports.UnmarshalCommandResult(reply.Payload)
	209	+	result, err := tracker.Wait(ctx, commandClock, requestID, generation, outcome)
186	210		if err != nil {
187		-	return fmt.Errorf("decoding command reply: %w", err)
	211	+	return &exitCoded{code: 3, err: err}
188	212		}
189	213		if !result.OK {
190	214		if result.Code == ports.ErrInvalidCommandArgs {

 	"strings"
 	"text/tabwriter"
+	"github.com/bnema/vev/internal/adapters/clock"
 	"github.com/bnema/vev/internal/adapters/ipc"
 	"github.com/bnema/vev/internal/ports"
 	commandusecase "github.com/bnema/vev/internal/usecase/command"
+	"github.com/bnema/vev/internal/usecase/daemon"
 )
 type cmdInvocation struct {
 	stdout io.Writer
 	getenv func(string) string
 	dial   func(context.Context, string) (ports.Transport, error)
+	clock  ports.Clock
 }
 func runCmd(ctx context.Context, invocation cmdInvocation) error {
-	return runCmdWithDeps(ctx, invocation, cmdDeps{stdout: os.Stdout, getenv: os.Getenv, dial: realDial})
+	return runCmdWithDeps(ctx, invocation, cmdDeps{stdout: os.Stdout, getenv: os.Getenv, dial: realDial, clock: clock.New()})
 }
 func runCmdWithDeps(ctx context.Context, invocation cmdInvocation, deps cmdDeps) error {
 		return usagef("--self requires running inside a vev pane")
 	}
-	payload, err := ports.MarshalCommandRequest(request)
+	if _, err := ports.MarshalCommandRequest(request); errors.Is(err, ports.ErrTooManyCommandArgs) {
-	if errors.Is(err, ports.ErrTooManyCommandArgs) {
 		return usagef("too many command arguments")
-	}
+	} else if err != nil {
-	if err != nil {
 		return fmt.Errorf("encoding command: %w", err)
 	}
 	}
 	defer func() { _ = transport.Close() }()
-	frame := ports.Frame{Type: ports.MsgCommand, Payload: payload}
+	tracker := daemon.NewCommandRequestTracker()
-	if err := transport.Send(frame); err != nil {
+	const generation = uint64(1)
-		return &exitCoded{code: 3, err: fmt.Errorf("sending command: %w", err)}
+	requestID, outcome := tracker.Publish(generation)
-	}
+	request.RequestID = requestID
-	reply, err := transport.Recv()
+	payload, err := ports.MarshalCommandRequest(request)
 	if err != nil {
-		return &exitCoded{code: 3, err: fmt.Errorf("reading command reply: %w", err)}
+		tracker.Remove(requestID, generation)
+		return fmt.Errorf("encoding command: %w", err)
+	}
+	if err := transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
+		tracker.Remove(requestID, generation)
+		return &exitCoded{code: 3, err: fmt.Errorf("sending command: %w", err)}
 	}
-	if reply.Type != ports.MsgCommandResult {
+	go func() {
-		return fmt.Errorf("unexpected command reply type %d", reply.Type)
+		reply, recvErr := transport.Recv()
+		if recvErr != nil {
+			tracker.Fail(requestID, generation, fmt.Errorf("reading command reply: %w", recvErr))
+			return
+		}
+		if reply.Type != ports.MsgCommandResult {
+			tracker.Fail(requestID, generation, fmt.Errorf("unexpected command reply type %d", reply.Type))
+			return
+		}
+		result, decodeErr := ports.UnmarshalCommandResult(reply.Payload)
+		if decodeErr != nil {
+			tracker.Fail(requestID, generation, fmt.Errorf("decoding command reply: %w", decodeErr))
+			return
+		}
+		tracker.Complete(generation, result)
+	}()
+	commandClock := deps.clock
+	if commandClock == nil {
+		commandClock = clock.New()
 	}
-	result, err := ports.UnmarshalCommandResult(reply.Payload)
+	result, err := tracker.Wait(ctx, commandClock, requestID, generation, outcome)
 	if err != nil {
-		return fmt.Errorf("decoding command reply: %w", err)
+		return &exitCoded{code: 3, err: err}
 	}
 	if !result.OK {
 		if result.Code == ports.ErrInvalidCommandArgs {

M internal/app/cmd_test.go +69 -5


@@ -9,8 +9,10 @@ import (
9	9		"strconv"
10	10		"strings"
11	11		"testing"
	12	+	"time"
12	13
13	14		"github.com/bnema/vev/internal/ports"
	15	+	"github.com/bnema/vev/internal/usecase/daemon"
14	16		"github.com/stretchr/testify/require"
15	17		)
16	18

@@ -287,6 +289,50 @@ func TestRunCmdRejectsTooManyArgumentsBeforeDialing(t *testing.T) {
287	289		}
288	290		}
289	291
	292	+	func TestRunCmdTimesOutPendingRequest(t *testing.T) {
	293	+	timer := &cmdTestTimer{ch: make(chan time.Time, 1)}
	294	+	clock := &cmdTestClock{timer: timer}
	295	+	recvStarted := make(chan struct{})
	296	+	recvGate := make(chan struct{})
	297	+	transport := &cmdTestTransport{recvStarted: recvStarted, recvGate: recvGate}
	298	+	done := make(chan error, 1)
	299	+	go func() {
	300	+	done <- runCmdWithDeps(context.Background(), cmdInvocation{slug: "split-right"}, cmdDeps{
	301	+	stdout: io.Discard,
	302	+	getenv: func(string) string { return "" },
	303	+	dial: func(context.Context, string) (ports.Transport, error) { return transport, nil },
	304	+	clock: clock,
	305	+	})
	306	+	}()
	307	+	<-recvStarted
	308	+	timer.ch <- time.Time{}
	309	+	err := <-done
	310	+	close(recvGate)
	311	+	if ExitCode(err) != 3 \|\| !errors.Is(err, daemon.ErrCommandRequestTimeout) {
	312	+	t.Fatalf("timeout error=%v code=%d", err, ExitCode(err))
	313	+	}
	314	+	request, decodeErr := ports.UnmarshalCommandRequest(transport.sent[0].Payload)
	315	+	if decodeErr != nil \|\| request.RequestID == 0 {
	316	+	t.Fatalf("request=%+v decode=%v, want non-zero request ID", request, decodeErr)
	317	+	}
	318	+	}
	319	+
	320	+	type cmdTestTimer struct{ ch chan time.Time }
	321	+
	322	+	func (t *cmdTestTimer) C() <-chan time.Time { return t.ch }
	323	+	func (t *cmdTestTimer) Reset(time.Duration) bool { return false }
	324	+	func (t *cmdTestTimer) Stop() bool { return true }
	325	+
	326	+	type cmdTestClock struct{ timer *cmdTestTimer }
	327	+
	328	+	func (c *cmdTestClock) Now() time.Time { return time.Time{} }
	329	+	func (c *cmdTestClock) NewTimer(delay time.Duration) ports.Timer {
	330	+	if delay != daemon.CommandRequestTimeout {
	331	+	panic("unexpected command timeout")
	332	+	}
	333	+	return c.timer
	334	+	}
	335	+
290	336		func TestRunCmdDoesNotAutostartAndClassifiesDialFailure(t *testing.T) {
291	337		dials := 0
292	338		dialErr := errors.New("unreachable")

@@ -349,11 +395,13 @@ func TestRunCmdClassifiesDaemonCommandErrors(t *testing.T) {
349	395		}
350	396
351	397		type cmdTestTransport struct {
352		-	sent []ports.Frame
353		-	recv ports.Frame
354		-	recvErr error
355		-	recvCalls int
356		-	closeCalls int
	398	+	sent []ports.Frame
	399	+	recv ports.Frame
	400	+	recvErr error
	401	+	recvCalls int
	402	+	closeCalls int
	403	+	recvStarted chan struct{}
	404	+	recvGate <-chan struct{}
357	405		}
358	406
359	407		func (t *cmdTestTransport) Send(frame ports.Frame) error {

@@ -363,6 +411,22 @@ func (t *cmdTestTransport) Send(frame ports.Frame) error {
363	411
364	412		func (t *cmdTestTransport) Recv() (ports.Frame, error) {
365	413		t.recvCalls++
	414	+	if t.recvStarted != nil {
	415	+	close(t.recvStarted)
	416	+	}
	417	+	if t.recvGate != nil {
	418	+	<-t.recvGate
	419	+	}
	420	+	if t.recv.Type == ports.MsgCommandResult && len(t.sent) != 0 {
	421	+	result, err := ports.UnmarshalCommandResult(t.recv.Payload)
	422	+	if err == nil && result.RequestID == 0 {
	423	+	request, err := ports.UnmarshalCommandRequest(t.sent[0].Payload)
	424	+	if err == nil {
	425	+	result.RequestID = request.RequestID
	426	+	return ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(result)}, t.recvErr
	427	+	}
	428	+	}
	429	+	}
366	430		return t.recv, t.recvErr
367	431		}
368	432


@@ -9,8 +9,10 @@ import (
9		"strconv"	9		"strconv"
10		"strings"	10		"strings"
11		"testing"	11		"testing"
			12	+	"time"
12			13
13		"github.com/bnema/vev/internal/ports"	14		"github.com/bnema/vev/internal/ports"
			15	+	"github.com/bnema/vev/internal/usecase/daemon"
14		"github.com/stretchr/testify/require"	16		"github.com/stretchr/testify/require"
15		)	17		)
16			18

@@ -287,6 +289,50 @@ func TestRunCmdRejectsTooManyArgumentsBeforeDialing(t *testing.T) {
287		}	289		}
288		}	290		}
289			291
			292	+	func TestRunCmdTimesOutPendingRequest(t *testing.T) {
			293	+	timer := &cmdTestTimer{ch: make(chan time.Time, 1)}
			294	+	clock := &cmdTestClock{timer: timer}
			295	+	recvStarted := make(chan struct{})
			296	+	recvGate := make(chan struct{})
			297	+	transport := &cmdTestTransport{recvStarted: recvStarted, recvGate: recvGate}
			298	+	done := make(chan error, 1)
			299	+	go func() {
			300	+	done <- runCmdWithDeps(context.Background(), cmdInvocation{slug: "split-right"}, cmdDeps{
			301	+	stdout: io.Discard,
			302	+	getenv: func(string) string { return "" },
			303	+	dial: func(context.Context, string) (ports.Transport, error) { return transport, nil },
			304	+	clock: clock,
			305	+	})
			306	+	}()
			307	+	<-recvStarted
			308	+	timer.ch <- time.Time{}
			309	+	err := <-done
			310	+	close(recvGate)
			311	+	if ExitCode(err) != 3 \|\| !errors.Is(err, daemon.ErrCommandRequestTimeout) {
			312	+	t.Fatalf("timeout error=%v code=%d", err, ExitCode(err))
			313	+	}
			314	+	request, decodeErr := ports.UnmarshalCommandRequest(transport.sent[0].Payload)
			315	+	if decodeErr != nil \|\| request.RequestID == 0 {
			316	+	t.Fatalf("request=%+v decode=%v, want non-zero request ID", request, decodeErr)
			317	+	}
			318	+	}
			319	+
			320	+	type cmdTestTimer struct{ ch chan time.Time }
			321	+
			322	+	func (t *cmdTestTimer) C() <-chan time.Time { return t.ch }
			323	+	func (t *cmdTestTimer) Reset(time.Duration) bool { return false }
			324	+	func (t *cmdTestTimer) Stop() bool { return true }
			325	+
			326	+	type cmdTestClock struct{ timer *cmdTestTimer }
			327	+
			328	+	func (c *cmdTestClock) Now() time.Time { return time.Time{} }
			329	+	func (c *cmdTestClock) NewTimer(delay time.Duration) ports.Timer {
			330	+	if delay != daemon.CommandRequestTimeout {
			331	+	panic("unexpected command timeout")
			332	+	}
			333	+	return c.timer
			334	+	}
			335	+
290		func TestRunCmdDoesNotAutostartAndClassifiesDialFailure(t *testing.T) {	336		func TestRunCmdDoesNotAutostartAndClassifiesDialFailure(t *testing.T) {
291		dials := 0	337		dials := 0
292		dialErr := errors.New("unreachable")	338		dialErr := errors.New("unreachable")

@@ -349,11 +395,13 @@ func TestRunCmdClassifiesDaemonCommandErrors(t *testing.T) {
349		}	395		}
350			396
351		type cmdTestTransport struct {	397		type cmdTestTransport struct {
352	-	sent []ports.Frame	398	+	sent []ports.Frame
353	-	recv ports.Frame	399	+	recv ports.Frame
354	-	recvErr error	400	+	recvErr error
355	-	recvCalls int	401	+	recvCalls int
356	-	closeCalls int	402	+	closeCalls int
			403	+	recvStarted chan struct{}
			404	+	recvGate <-chan struct{}
357		}	405		}
358			406
359		func (t *cmdTestTransport) Send(frame ports.Frame) error {	407		func (t *cmdTestTransport) Send(frame ports.Frame) error {

@@ -363,6 +411,22 @@ func (t *cmdTestTransport) Send(frame ports.Frame) error {
363			411
364		func (t *cmdTestTransport) Recv() (ports.Frame, error) {	412		func (t *cmdTestTransport) Recv() (ports.Frame, error) {
365		t.recvCalls++	413		t.recvCalls++
			414	+	if t.recvStarted != nil {
			415	+	close(t.recvStarted)
			416	+	}
			417	+	if t.recvGate != nil {
			418	+	<-t.recvGate
			419	+	}
			420	+	if t.recv.Type == ports.MsgCommandResult && len(t.sent) != 0 {
			421	+	result, err := ports.UnmarshalCommandResult(t.recv.Payload)
			422	+	if err == nil && result.RequestID == 0 {
			423	+	request, err := ports.UnmarshalCommandRequest(t.sent[0].Payload)
			424	+	if err == nil {
			425	+	result.RequestID = request.RequestID
			426	+	return ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(result)}, t.recvErr
			427	+	}
			428	+	}
			429	+	}
366		return t.recv, t.recvErr	430		return t.recv, t.recvErr
367		}	431		}
368			432

M internal/app/run.go +14 -1


@@ -795,6 +795,16 @@ func remoteTransportModeFromEnv(value string) (ports.RemoteTransportMode, error)
795	795		}
796	796		}
797	797
	798	+	func validateRemoteAttachHandoff(target ports.AttachTarget) error {
	799	+	if err := ports.ValidateAttachTarget(target); err != nil {
	800	+	return err
	801	+	}
	802	+	if err := domain.ValidateRemoteHostTarget(target.Endpoint); err != nil {
	803	+	return err
	804	+	}
	805	+	return domain.ValidateSessionName(target.Session)
	806	+	}
	807	+
798	808		// remoteDiscoveryDaemonOption constructs the daemon-owned discovery ports from
799	809		// the same validated transport selection used by direct remote attach.
800	810		func remoteDiscoveryDaemonOption(stateDir string, observer ports.SerializedRuntimeObserver, transport string) (daemon.Option, error) {

@@ -854,9 +864,12 @@ func runAttachWithDeps(ctx context.Context, intent uint8, name, remoteTarget, ac
854	864		if !errors.As(err, &handoff) {
855	865		return err
856	866		}
857		-	if handoff == nil \|\| ports.ValidateAttachTarget(handoff.Target) != nil {
	867	+	if handoff == nil {
858	868		return fmt.Errorf("vev: invalid remote attach handoff")
859	869		}
	870	+	if err := validateRemoteAttachHandoff(handoff.Target); err != nil {
	871	+	return fmt.Errorf("vev: invalid remote attach handoff: %w", err)
	872	+	}
860	873		remoteTarget = handoff.Target.Endpoint
861	874		name = handoff.Target.Session
862	875		intent = handoff.Target.Intent

 	}
 }
+func validateRemoteAttachHandoff(target ports.AttachTarget) error {
+	if err := ports.ValidateAttachTarget(target); err != nil {
+		return err
+	}
+	if err := domain.ValidateRemoteHostTarget(target.Endpoint); err != nil {
+		return err
+	}
+	return domain.ValidateSessionName(target.Session)
+}
 // remoteDiscoveryDaemonOption constructs the daemon-owned discovery ports from
 // the same validated transport selection used by direct remote attach.
 func remoteDiscoveryDaemonOption(stateDir string, observer ports.SerializedRuntimeObserver, transport string) (daemon.Option, error) {
 			if !errors.As(err, &handoff) {
 				return err
 			}
-			if handoff == nil || ports.ValidateAttachTarget(handoff.Target) != nil {
+			if handoff == nil {
 				return fmt.Errorf("vev: invalid remote attach handoff")
 			}
+			if err := validateRemoteAttachHandoff(handoff.Target); err != nil {
+				return fmt.Errorf("vev: invalid remote attach handoff: %w", err)
+			}
 			remoteTarget = handoff.Target.Endpoint
 			name = handoff.Target.Session
 			intent = handoff.Target.Intent

M internal/app/run_test.go +30 -4


@@ -502,13 +502,13 @@ func TestParseArgs(t *testing.T) {
502	502		t.Fatalf("parseArgs(%q) = %+v, want error", tt.args, got)
503	503		}
504	504		if tt.nonUsageErr {
505		-	var ue *usageError
506		-	if errors.As(err, &ue) {
	505	+	_, usage := errors.AsType[*usageError](err)
	506	+	if usage {
507	507		t.Fatalf("parseArgs(%q) error = *usageError, want non-usage error", tt.args)
508	508		}
509	509		} else {
510		-	var ue *usageError
511		-	if !errors.As(err, &ue) {
	510	+	_, usage := errors.AsType[*usageError](err)
	511	+	if !usage {
512	512		t.Fatalf("parseArgs(%q) error = %T, want *usageError", tt.args, err)
513	513		}
514	514		}

@@ -889,6 +889,32 @@ func TestRunAttachWithDepsRemotePickerHandoffReopensDirectConnection(t *testing.
889	889		require.Equal(t, 2, calls, "a picker handoff must close the old connection and open a fresh direct connection")
890	890		}
891	891
	892	+	func TestRunAttachWithDepsRejectsInvalidHandoffBeforeDialing(t *testing.T) {
	893	+	tests := []struct {
	894	+	name string
	895	+	target ports.AttachTarget
	896	+	}{
	897	+	{name: "invalid host", target: ports.AttachTarget{Endpoint: "remote host", Session: "work", Intent: ports.IntentAttach}},
	898	+	{name: "invalid session", target: ports.AttachTarget{Endpoint: "remote.example", Session: "bad name", Intent: ports.IntentAttach}},
	899	+	}
	900	+	for _, tt := range tests {
	901	+	t.Run(tt.name, func(t *testing.T) {
	902	+	factory := portsmocks.NewMockRemoteDialerFactory(t)
	903	+	factory.EXPECT().DialerForRemote("remote.example", "work", ports.RemoteTransportUDP, mock.Anything).Return(namedDialer{name: "remote"}, nil).Once()
	904	+	calls := 0
	905	+	err := runAttachWithDeps(context.Background(), ports.IntentAttach, "work", "remote.example", "", nil, runAttachDeps{
	906	+	remoteDialerFactory: factory,
	907	+	runClient: func(_ context.Context, _ client.Dependencies, _ client.AttachRequest) error {
	908	+	calls++
	909	+	return &client.AttachTargetError{Target: tt.target}
	910	+	},
	911	+	})
	912	+	require.ErrorContains(t, err, "invalid remote attach handoff")
	913	+	require.Equal(t, 1, calls)
	914	+	})
	915	+	}
	916	+	}
	917	+
892	918		func TestRunAttachWithDepsRejectsInvalidRemoteTransportBeforeDialing(t *testing.T) {
893	919		factory := portsmocks.NewMockRemoteDialerFactory(t)
894	920		runClientCalled := false


@@ -502,13 +502,13 @@ func TestParseArgs(t *testing.T) {
502		t.Fatalf("parseArgs(%q) = %+v, want error", tt.args, got)	502		t.Fatalf("parseArgs(%q) = %+v, want error", tt.args, got)
503		}	503		}
504		if tt.nonUsageErr {	504		if tt.nonUsageErr {
505	-	var ue *usageError	505	+	_, usage := errors.AsType[*usageError](err)
506	-	if errors.As(err, &ue) {	506	+	if usage {
507		t.Fatalf("parseArgs(%q) error = *usageError, want non-usage error", tt.args)	507		t.Fatalf("parseArgs(%q) error = *usageError, want non-usage error", tt.args)
508		}	508		}
509		} else {	509		} else {
510	-	var ue *usageError	510	+	_, usage := errors.AsType[*usageError](err)
511	-	if !errors.As(err, &ue) {	511	+	if !usage {
512		t.Fatalf("parseArgs(%q) error = %T, want *usageError", tt.args, err)	512		t.Fatalf("parseArgs(%q) error = %T, want *usageError", tt.args, err)
513		}	513		}
514		}	514		}

@@ -889,6 +889,32 @@ func TestRunAttachWithDepsRemotePickerHandoffReopensDirectConnection(t *testing.
889		require.Equal(t, 2, calls, "a picker handoff must close the old connection and open a fresh direct connection")	889		require.Equal(t, 2, calls, "a picker handoff must close the old connection and open a fresh direct connection")
890		}	890		}
891			891
			892	+	func TestRunAttachWithDepsRejectsInvalidHandoffBeforeDialing(t *testing.T) {
			893	+	tests := []struct {
			894	+	name string
			895	+	target ports.AttachTarget
			896	+	}{
			897	+	{name: "invalid host", target: ports.AttachTarget{Endpoint: "remote host", Session: "work", Intent: ports.IntentAttach}},
			898	+	{name: "invalid session", target: ports.AttachTarget{Endpoint: "remote.example", Session: "bad name", Intent: ports.IntentAttach}},
			899	+	}
			900	+	for _, tt := range tests {
			901	+	t.Run(tt.name, func(t *testing.T) {
			902	+	factory := portsmocks.NewMockRemoteDialerFactory(t)
			903	+	factory.EXPECT().DialerForRemote("remote.example", "work", ports.RemoteTransportUDP, mock.Anything).Return(namedDialer{name: "remote"}, nil).Once()
			904	+	calls := 0
			905	+	err := runAttachWithDeps(context.Background(), ports.IntentAttach, "work", "remote.example", "", nil, runAttachDeps{
			906	+	remoteDialerFactory: factory,
			907	+	runClient: func(_ context.Context, _ client.Dependencies, _ client.AttachRequest) error {
			908	+	calls++
			909	+	return &client.AttachTargetError{Target: tt.target}
			910	+	},
			911	+	})
			912	+	require.ErrorContains(t, err, "invalid remote attach handoff")
			913	+	require.Equal(t, 1, calls)
			914	+	})
			915	+	}
			916	+	}
			917	+
892		func TestRunAttachWithDepsRejectsInvalidRemoteTransportBeforeDialing(t *testing.T) {	918		func TestRunAttachWithDepsRejectsInvalidRemoteTransportBeforeDialing(t *testing.T) {
893		factory := portsmocks.NewMockRemoteDialerFactory(t)	919		factory := portsmocks.NewMockRemoteDialerFactory(t)
894		runClientCalled := false	920		runClientCalled := false

M internal/usecase/daemon/attachment_commands.go +165 -70


@@ -9,31 +9,174 @@ import (
9	9		"github.com/bnema/vev/internal/ports"
10	10		)
11	11
12		-	const attachedCommandTimeout = 10 * time.Second
	12	+	const CommandRequestTimeout = 10 * time.Second
13	13
14	14		var (
15		-	errAttachedCommandTimeout = errors.New("attached command timed out")
16		-	errAttachedCommandUnavailable = errors.New("attached command connection is unavailable")
	15	+	ErrCommandRequestTimeout = errors.New("command request timed out")
	16	+	ErrCommandRequestUnavailable = errors.New("command request connection is unavailable")
17	17		)
18	18
19		-	type attachedCommandOutcome struct {
20		-	result ports.CommandResult
21		-	err error
	19	+	// CommandRequestOutcome is the result delivered for one tracked request.
	20	+	type CommandRequestOutcome struct {
	21	+	Result ports.CommandResult
	22	+	Err error
22	23		}
23	24
24		-	type attachedCommandPending struct {
	25	+	// CommandRequestTracker correlates request results for one connection. The
	26	+	// generation is part of every pending entry so a reused request ID from an
	27	+	// older connection cannot complete a newer request.
	28	+	type CommandRequestTracker struct {
	29	+	mu sync.Mutex
	30	+	next uint64
	31	+	pending map[uint64]commandRequestPending
	32	+	}
	33	+
	34	+	type commandRequestPending struct {
25	35		generation uint64
26		-	outcome chan attachedCommandOutcome
	36	+	outcome chan CommandRequestOutcome
	37	+	}
	38	+
	39	+	// NewCommandRequestTracker constructs an empty per-connection tracker.
	40	+	func NewCommandRequestTracker() *CommandRequestTracker {
	41	+	return &CommandRequestTracker{}
	42	+	}
	43	+
	44	+	// Publish allocates and records a request before its bytes are sent.
	45	+	func (t *CommandRequestTracker) Publish(generation uint64) (uint64, <-chan CommandRequestOutcome) {
	46	+	t.mu.Lock()
	47	+	defer t.mu.Unlock()
	48	+	if t.pending == nil {
	49	+	t.pending = make(map[uint64]commandRequestPending)
	50	+	}
	51	+	t.next++
	52	+	if t.next == 0 {
	53	+	t.next++
	54	+	}
	55	+	outcome := t.trackLocked(t.next, generation)
	56	+	return t.next, outcome
	57	+	}
	58	+
	59	+	// Track records an existing wire request ID, as used by an inbound one-shot
	60	+	// command connection. It rejects duplicate IDs on that connection.
	61	+	func (t *CommandRequestTracker) Track(requestID, generation uint64) (<-chan CommandRequestOutcome, bool) {
	62	+	t.mu.Lock()
	63	+	defer t.mu.Unlock()
	64	+	if t.pending == nil {
	65	+	t.pending = make(map[uint64]commandRequestPending)
	66	+	}
	67	+	if _, exists := t.pending[requestID]; exists {
	68	+	return nil, false
	69	+	}
	70	+	if requestID > t.next {
	71	+	t.next = requestID
	72	+	}
	73	+	outcome := t.trackLocked(requestID, generation)
	74	+	return outcome, true
	75	+	}
	76	+
	77	+	func (t *CommandRequestTracker) trackLocked(requestID, generation uint64) <-chan CommandRequestOutcome {
	78	+	outcome := make(chan CommandRequestOutcome, 1)
	79	+	t.pending[requestID] = commandRequestPending{generation: generation, outcome: outcome}
	80	+	return outcome
	81	+	}
	82	+
	83	+	// Wait waits without holding the tracker or connection sender lock. Timeout
	84	+	// and cancellation remove only the matching request.
	85	+	func (t *CommandRequestTracker) Wait(ctx context.Context, clock ports.Clock, requestID, generation uint64, outcome <-chan CommandRequestOutcome) (ports.CommandResult, error) {
	86	+	if t == nil \|\| clock == nil {
	87	+	return ports.CommandResult{}, ErrCommandRequestUnavailable
	88	+	}
	89	+	if ctx == nil {
	90	+	ctx = context.Background()
	91	+	}
	92	+	if err := ctx.Err(); err != nil {
	93	+	t.Remove(requestID, generation)
	94	+	return ports.CommandResult{}, err
	95	+	}
	96	+	timer := clock.NewTimer(CommandRequestTimeout)
	97	+	if timer == nil {
	98	+	t.Remove(requestID, generation)
	99	+	return ports.CommandResult{}, ErrCommandRequestTimeout
	100	+	}
	101	+	defer timer.Stop()
	102	+	select {
	103	+	case completed := <-outcome:
	104	+	return completed.Result, completed.Err
	105	+	case <-ctx.Done():
	106	+	t.Remove(requestID, generation)
	107	+	return ports.CommandResult{}, ctx.Err()
	108	+	case <-timer.C():
	109	+	t.Remove(requestID, generation)
	110	+	return ports.CommandResult{}, ErrCommandRequestTimeout
	111	+	}
	112	+	}
	113	+
	114	+	// Remove abandons only the exact request generation.
	115	+	func (t *CommandRequestTracker) Remove(requestID, generation uint64) {
	116	+	if t == nil {
	117	+	return
	118	+	}
	119	+	t.mu.Lock()
	120	+	pending, ok := t.pending[requestID]
	121	+	if ok && pending.generation == generation {
	122	+	delete(t.pending, requestID)
	123	+	}
	124	+	t.mu.Unlock()
	125	+	}
	126	+
	127	+	// Complete accepts only the exact request generation. Unknown, late, and old
	128	+	// generation results are safe no-ops.
	129	+	func (t *CommandRequestTracker) Complete(generation uint64, result ports.CommandResult) {
	130	+	t.finish(generation, result.RequestID, CommandRequestOutcome{Result: result})
	131	+	}
	132	+
	133	+	// Fail completes the exact request with a transport or decode failure.
	134	+	func (t *CommandRequestTracker) Fail(requestID, generation uint64, err error) {
	135	+	t.finish(generation, requestID, CommandRequestOutcome{Err: err})
27	136		}
28	137
29		-	// attachedCommandTracker owns command correlation for one attachment
30		-	// connection. Its lock protects only request publication state; command waits
31		-	// happen after the attachment sender lock is released.
	138	+	func (t *CommandRequestTracker) finish(generation, requestID uint64, outcome CommandRequestOutcome) {
	139	+	if t == nil {
	140	+	return
	141	+	}
	142	+	t.mu.Lock()
	143	+	pending, ok := t.pending[requestID]
	144	+	if !ok \|\| pending.generation != generation {
	145	+	t.mu.Unlock()
	146	+	return
	147	+	}
	148	+	delete(t.pending, requestID)
	149	+	t.mu.Unlock()
	150	+	pending.outcome <- outcome
	151	+	}
	152	+
	153	+	// PendingCount is intended for deterministic lifecycle tests.
	154	+	func (t *CommandRequestTracker) PendingCount() int {
	155	+	if t == nil {
	156	+	return 0
	157	+	}
	158	+	t.mu.Lock()
	159	+	defer t.mu.Unlock()
	160	+	return len(t.pending)
	161	+	}
	162	+
	163	+	// attachedCommandTracker adds ordering for interactive commands while reusing
	164	+	// the same generic per-connection request tracker.
32	165		type attachedCommandTracker struct {
33	166		commandMu sync.Mutex
34		-	mu sync.Mutex
35		-	next uint64
36		-	pending map[uint64]attachedCommandPending
	167	+	CommandRequestTracker
	168	+	}
	169	+
	170	+	const attachedCommandTimeout = CommandRequestTimeout
	171	+
	172	+	var (
	173	+	errAttachedCommandTimeout = ErrCommandRequestTimeout
	174	+	errAttachedCommandUnavailable = ErrCommandRequestUnavailable
	175	+	)
	176	+
	177	+	type attachedCommandOutcome struct {
	178	+	result ports.CommandResult
	179	+	err error
37	180		}
38	181
39	182		// sendCommand publishes one attached command and waits for its exact result.

@@ -63,20 +206,9 @@ func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, sl
63	206		ac.sendMu.Unlock()
64	207		return ports.CommandResult{}, errAttachedCommandUnavailable
65	208		}
66		-	ac.commands.mu.Lock()
67		-	if ac.commands.pending == nil {
68		-	ac.commands.pending = make(map[uint64]attachedCommandPending)
69		-	}
70		-	ac.commands.next++
71		-	if ac.commands.next == 0 {
72		-	ac.commands.next++
73		-	}
74		-	requestID := ac.commands.next
75		-	outcome := make(chan attachedCommandOutcome, 1)
76		-	ac.commands.pending[requestID] = attachedCommandPending{generation: generation, outcome: outcome}
77		-	ac.commands.mu.Unlock()
	209	+	requestID, outcome := ac.commands.Publish(generation)
78	210		if !ac.transportSnapshotCurrent(expected) \|\| ac.connectionGeneration.Load() != generation {
79		-	ac.removeAttachedCommand(requestID, generation)
	211	+	ac.commands.Remove(requestID, generation)
80	212		ac.sendMu.Unlock()
81	213		return ports.CommandResult{}, errAttachedCommandUnavailable
82	214		}

@@ -86,70 +218,33 @@ func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, sl
86	218		Slug: slug, Args: append([]string(nil), args...),
87	219		})
88	220		if err != nil {
89		-	ac.removeAttachedCommand(requestID, generation)
	221	+	ac.commands.Remove(requestID, generation)
90	222		ac.sendMu.Unlock()
91	223		return ports.CommandResult{}, err
92	224		}
93	225		if err := expected.transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
94		-	ac.removeAttachedCommand(requestID, generation)
	226	+	ac.commands.Remove(requestID, generation)
95	227		ac.sendMu.Unlock()
96	228		return ports.CommandResult{}, err
97	229		}
98	230		ac.sendMu.Unlock()
99	231
100		-	timer := clock.NewTimer(attachedCommandTimeout)
101		-	if timer == nil {
102		-	ac.removeAttachedCommand(requestID, generation)
103		-	return ports.CommandResult{}, errAttachedCommandTimeout
104		-	}
105		-	defer timer.Stop()
106		-	select {
107		-	case completed := <-outcome:
108		-	return completed.result, completed.err
109		-	case <-ctx.Done():
110		-	ac.removeAttachedCommand(requestID, generation)
111		-	return ports.CommandResult{}, ctx.Err()
112		-	case <-timer.C():
113		-	ac.removeAttachedCommand(requestID, generation)
114		-	return ports.CommandResult{}, errAttachedCommandTimeout
115		-	}
116		-	}
117		-
118		-	func (ac *attachedClient) removeAttachedCommand(requestID, generation uint64) {
119		-	if ac == nil {
120		-	return
121		-	}
122		-	ac.commands.mu.Lock()
123		-	pending, ok := ac.commands.pending[requestID]
124		-	if ok && pending.generation == generation {
125		-	delete(ac.commands.pending, requestID)
126		-	}
127		-	ac.commands.mu.Unlock()
	232	+	return ac.commands.Wait(ctx, clock, requestID, generation, outcome)
128	233		}
129	234
130	235		// completeCommandResult accepts only a result for the exact request and
131	236		// connection generation. Unknown, late, and old-generation results are safe
132	237		// no-ops.
133	238		func (ac *attachedClient) completeCommandResult(generation uint64, result ports.CommandResult) {
134		-	if ac == nil \|\| result.RequestID == 0 \|\| ac.connectionGeneration.Load() != generation {
135		-	return
136		-	}
137		-	ac.commands.mu.Lock()
138		-	pending, ok := ac.commands.pending[result.RequestID]
139		-	if !ok \|\| pending.generation != generation {
140		-	ac.commands.mu.Unlock()
	239	+	if ac == nil \|\| ac.connectionGeneration.Load() != generation {
141	240		return
142	241		}
143		-	delete(ac.commands.pending, result.RequestID)
144		-	ac.commands.mu.Unlock()
145		-	pending.outcome <- attachedCommandOutcome{result: result}
	242	+	ac.commands.Complete(generation, result)
146	243		}
147	244
148	245		func (ac *attachedClient) pendingAttachedCommandCount() int {
149	246		if ac == nil {
150	247		return 0
151	248		}
152		-	ac.commands.mu.Lock()
153		-	defer ac.commands.mu.Unlock()
154		-	return len(ac.commands.pending)
	249	+	return ac.commands.PendingCount()
155	250		}


@@ -9,31 +9,174 @@ import (
9		"github.com/bnema/vev/internal/ports"	9		"github.com/bnema/vev/internal/ports"
10		)	10		)
11			11
12	-	const attachedCommandTimeout = 10 * time.Second	12	+	const CommandRequestTimeout = 10 * time.Second
13			13
14		var (	14		var (
15	-	errAttachedCommandTimeout = errors.New("attached command timed out")	15	+	ErrCommandRequestTimeout = errors.New("command request timed out")
16	-	errAttachedCommandUnavailable = errors.New("attached command connection is unavailable")	16	+	ErrCommandRequestUnavailable = errors.New("command request connection is unavailable")
17		)	17		)
18			18
19	-	type attachedCommandOutcome struct {	19	+	// CommandRequestOutcome is the result delivered for one tracked request.
20	-	result ports.CommandResult	20	+	type CommandRequestOutcome struct {
21	-	err error	21	+	Result ports.CommandResult
			22	+	Err error
22		}	23		}
23			24
24	-	type attachedCommandPending struct {	25	+	// CommandRequestTracker correlates request results for one connection. The
			26	+	// generation is part of every pending entry so a reused request ID from an
			27	+	// older connection cannot complete a newer request.
			28	+	type CommandRequestTracker struct {
			29	+	mu sync.Mutex
			30	+	next uint64
			31	+	pending map[uint64]commandRequestPending
			32	+	}
			33	+
			34	+	type commandRequestPending struct {
25		generation uint64	35		generation uint64
26	-	outcome chan attachedCommandOutcome	36	+	outcome chan CommandRequestOutcome
			37	+	}
			38	+
			39	+	// NewCommandRequestTracker constructs an empty per-connection tracker.
			40	+	func NewCommandRequestTracker() *CommandRequestTracker {
			41	+	return &CommandRequestTracker{}
			42	+	}
			43	+
			44	+	// Publish allocates and records a request before its bytes are sent.
			45	+	func (t *CommandRequestTracker) Publish(generation uint64) (uint64, <-chan CommandRequestOutcome) {
			46	+	t.mu.Lock()
			47	+	defer t.mu.Unlock()
			48	+	if t.pending == nil {
			49	+	t.pending = make(map[uint64]commandRequestPending)
			50	+	}
			51	+	t.next++
			52	+	if t.next == 0 {
			53	+	t.next++
			54	+	}
			55	+	outcome := t.trackLocked(t.next, generation)
			56	+	return t.next, outcome
			57	+	}
			58	+
			59	+	// Track records an existing wire request ID, as used by an inbound one-shot
			60	+	// command connection. It rejects duplicate IDs on that connection.
			61	+	func (t *CommandRequestTracker) Track(requestID, generation uint64) (<-chan CommandRequestOutcome, bool) {
			62	+	t.mu.Lock()
			63	+	defer t.mu.Unlock()
			64	+	if t.pending == nil {
			65	+	t.pending = make(map[uint64]commandRequestPending)
			66	+	}
			67	+	if _, exists := t.pending[requestID]; exists {
			68	+	return nil, false
			69	+	}
			70	+	if requestID > t.next {
			71	+	t.next = requestID
			72	+	}
			73	+	outcome := t.trackLocked(requestID, generation)
			74	+	return outcome, true
			75	+	}
			76	+
			77	+	func (t *CommandRequestTracker) trackLocked(requestID, generation uint64) <-chan CommandRequestOutcome {
			78	+	outcome := make(chan CommandRequestOutcome, 1)
			79	+	t.pending[requestID] = commandRequestPending{generation: generation, outcome: outcome}
			80	+	return outcome
			81	+	}
			82	+
			83	+	// Wait waits without holding the tracker or connection sender lock. Timeout
			84	+	// and cancellation remove only the matching request.
			85	+	func (t *CommandRequestTracker) Wait(ctx context.Context, clock ports.Clock, requestID, generation uint64, outcome <-chan CommandRequestOutcome) (ports.CommandResult, error) {
			86	+	if t == nil \|\| clock == nil {
			87	+	return ports.CommandResult{}, ErrCommandRequestUnavailable
			88	+	}
			89	+	if ctx == nil {
			90	+	ctx = context.Background()
			91	+	}
			92	+	if err := ctx.Err(); err != nil {
			93	+	t.Remove(requestID, generation)
			94	+	return ports.CommandResult{}, err
			95	+	}
			96	+	timer := clock.NewTimer(CommandRequestTimeout)
			97	+	if timer == nil {
			98	+	t.Remove(requestID, generation)
			99	+	return ports.CommandResult{}, ErrCommandRequestTimeout
			100	+	}
			101	+	defer timer.Stop()
			102	+	select {
			103	+	case completed := <-outcome:
			104	+	return completed.Result, completed.Err
			105	+	case <-ctx.Done():
			106	+	t.Remove(requestID, generation)
			107	+	return ports.CommandResult{}, ctx.Err()
			108	+	case <-timer.C():
			109	+	t.Remove(requestID, generation)
			110	+	return ports.CommandResult{}, ErrCommandRequestTimeout
			111	+	}
			112	+	}
			113	+
			114	+	// Remove abandons only the exact request generation.
			115	+	func (t *CommandRequestTracker) Remove(requestID, generation uint64) {
			116	+	if t == nil {
			117	+	return
			118	+	}
			119	+	t.mu.Lock()
			120	+	pending, ok := t.pending[requestID]
			121	+	if ok && pending.generation == generation {
			122	+	delete(t.pending, requestID)
			123	+	}
			124	+	t.mu.Unlock()
			125	+	}
			126	+
			127	+	// Complete accepts only the exact request generation. Unknown, late, and old
			128	+	// generation results are safe no-ops.
			129	+	func (t *CommandRequestTracker) Complete(generation uint64, result ports.CommandResult) {
			130	+	t.finish(generation, result.RequestID, CommandRequestOutcome{Result: result})
			131	+	}
			132	+
			133	+	// Fail completes the exact request with a transport or decode failure.
			134	+	func (t *CommandRequestTracker) Fail(requestID, generation uint64, err error) {
			135	+	t.finish(generation, requestID, CommandRequestOutcome{Err: err})
27		}	136		}
28			137
29	-	// attachedCommandTracker owns command correlation for one attachment	138	+	func (t *CommandRequestTracker) finish(generation, requestID uint64, outcome CommandRequestOutcome) {
30	-	// connection. Its lock protects only request publication state; command waits	139	+	if t == nil {
31	-	// happen after the attachment sender lock is released.	140	+	return
			141	+	}
			142	+	t.mu.Lock()
			143	+	pending, ok := t.pending[requestID]
			144	+	if !ok \|\| pending.generation != generation {
			145	+	t.mu.Unlock()
			146	+	return
			147	+	}
			148	+	delete(t.pending, requestID)
			149	+	t.mu.Unlock()
			150	+	pending.outcome <- outcome
			151	+	}
			152	+
			153	+	// PendingCount is intended for deterministic lifecycle tests.
			154	+	func (t *CommandRequestTracker) PendingCount() int {
			155	+	if t == nil {
			156	+	return 0
			157	+	}
			158	+	t.mu.Lock()
			159	+	defer t.mu.Unlock()
			160	+	return len(t.pending)
			161	+	}
			162	+
			163	+	// attachedCommandTracker adds ordering for interactive commands while reusing
			164	+	// the same generic per-connection request tracker.
32		type attachedCommandTracker struct {	165		type attachedCommandTracker struct {
33		commandMu sync.Mutex	166		commandMu sync.Mutex
34	-	mu sync.Mutex	167	+	CommandRequestTracker
35	-	next uint64	168	+	}
36	-	pending map[uint64]attachedCommandPending	169	+
			170	+	const attachedCommandTimeout = CommandRequestTimeout
			171	+
			172	+	var (
			173	+	errAttachedCommandTimeout = ErrCommandRequestTimeout
			174	+	errAttachedCommandUnavailable = ErrCommandRequestUnavailable
			175	+	)
			176	+
			177	+	type attachedCommandOutcome struct {
			178	+	result ports.CommandResult
			179	+	err error
37		}	180		}
38			181
39		// sendCommand publishes one attached command and waits for its exact result.	182		// sendCommand publishes one attached command and waits for its exact result.

@@ -63,20 +206,9 @@ func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, sl
63		ac.sendMu.Unlock()	206		ac.sendMu.Unlock()
64		return ports.CommandResult{}, errAttachedCommandUnavailable	207		return ports.CommandResult{}, errAttachedCommandUnavailable
65		}	208		}
66	-	ac.commands.mu.Lock()	209	+	requestID, outcome := ac.commands.Publish(generation)
67	-	if ac.commands.pending == nil {
68	-	ac.commands.pending = make(map[uint64]attachedCommandPending)
69	-	}
70	-	ac.commands.next++
71	-	if ac.commands.next == 0 {
72	-	ac.commands.next++
73	-	}
74	-	requestID := ac.commands.next
75	-	outcome := make(chan attachedCommandOutcome, 1)
76	-	ac.commands.pending[requestID] = attachedCommandPending{generation: generation, outcome: outcome}
77	-	ac.commands.mu.Unlock()
78		if !ac.transportSnapshotCurrent(expected) \|\| ac.connectionGeneration.Load() != generation {	210		if !ac.transportSnapshotCurrent(expected) \|\| ac.connectionGeneration.Load() != generation {
79	-	ac.removeAttachedCommand(requestID, generation)	211	+	ac.commands.Remove(requestID, generation)
80		ac.sendMu.Unlock()	212		ac.sendMu.Unlock()
81		return ports.CommandResult{}, errAttachedCommandUnavailable	213		return ports.CommandResult{}, errAttachedCommandUnavailable
82		}	214		}

@@ -86,70 +218,33 @@ func (ac *attachedClient) sendCommand(ctx context.Context, clock ports.Clock, sl
86		Slug: slug, Args: append([]string(nil), args...),	218		Slug: slug, Args: append([]string(nil), args...),
87		})	219		})
88		if err != nil {	220		if err != nil {
89	-	ac.removeAttachedCommand(requestID, generation)	221	+	ac.commands.Remove(requestID, generation)
90		ac.sendMu.Unlock()	222		ac.sendMu.Unlock()
91		return ports.CommandResult{}, err	223		return ports.CommandResult{}, err
92		}	224		}
93		if err := expected.transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {	225		if err := expected.transport.Send(ports.Frame{Type: ports.MsgCommand, Payload: payload}); err != nil {
94	-	ac.removeAttachedCommand(requestID, generation)	226	+	ac.commands.Remove(requestID, generation)
95		ac.sendMu.Unlock()	227		ac.sendMu.Unlock()
96		return ports.CommandResult{}, err	228		return ports.CommandResult{}, err
97		}	229		}
98		ac.sendMu.Unlock()	230		ac.sendMu.Unlock()
99			231
100	-	timer := clock.NewTimer(attachedCommandTimeout)	232	+	return ac.commands.Wait(ctx, clock, requestID, generation, outcome)
101	-	if timer == nil {
102	-	ac.removeAttachedCommand(requestID, generation)
103	-	return ports.CommandResult{}, errAttachedCommandTimeout
104	-	}
105	-	defer timer.Stop()
106	-	select {
107	-	case completed := <-outcome:
108	-	return completed.result, completed.err
109	-	case <-ctx.Done():
110	-	ac.removeAttachedCommand(requestID, generation)
111	-	return ports.CommandResult{}, ctx.Err()
112	-	case <-timer.C():
113	-	ac.removeAttachedCommand(requestID, generation)
114	-	return ports.CommandResult{}, errAttachedCommandTimeout
115	-	}
116	-	}
117	-
118	-	func (ac *attachedClient) removeAttachedCommand(requestID, generation uint64) {
119	-	if ac == nil {
120	-	return
121	-	}
122	-	ac.commands.mu.Lock()
123	-	pending, ok := ac.commands.pending[requestID]
124	-	if ok && pending.generation == generation {
125	-	delete(ac.commands.pending, requestID)
126	-	}
127	-	ac.commands.mu.Unlock()
128		}	233		}
129			234
130		// completeCommandResult accepts only a result for the exact request and	235		// completeCommandResult accepts only a result for the exact request and
131		// connection generation. Unknown, late, and old-generation results are safe	236		// connection generation. Unknown, late, and old-generation results are safe
132		// no-ops.	237		// no-ops.
133		func (ac *attachedClient) completeCommandResult(generation uint64, result ports.CommandResult) {	238		func (ac *attachedClient) completeCommandResult(generation uint64, result ports.CommandResult) {
134	-	if ac == nil \|\| result.RequestID == 0 \|\| ac.connectionGeneration.Load() != generation {	239	+	if ac == nil \|\| ac.connectionGeneration.Load() != generation {
135	-	return
136	-	}
137	-	ac.commands.mu.Lock()
138	-	pending, ok := ac.commands.pending[result.RequestID]
139	-	if !ok \|\| pending.generation != generation {
140	-	ac.commands.mu.Unlock()
141		return	240		return
142		}	241		}
143	-	delete(ac.commands.pending, result.RequestID)	242	+	ac.commands.Complete(generation, result)
144	-	ac.commands.mu.Unlock()
145	-	pending.outcome <- attachedCommandOutcome{result: result}
146		}	243		}
147			244
148		func (ac *attachedClient) pendingAttachedCommandCount() int {	245		func (ac *attachedClient) pendingAttachedCommandCount() int {
149		if ac == nil {	246		if ac == nil {
150		return 0	247		return 0
151		}	248		}
152	-	ac.commands.mu.Lock()	249	+	return ac.commands.PendingCount()
153	-	defer ac.commands.mu.Unlock()
154	-	return len(ac.commands.pending)
155		}	250		}

M internal/usecase/daemon/attachment_commands_test.go +27 -0


@@ -118,6 +118,33 @@ func startAttachedCommand(t testing.T, ac attachedClient, clock *attachmentCom
118	118		return request, awaitAttachedCommandTimer(t, clock), done
119	119		}
120	120
	121	+	func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) {
	122	+	tracker := NewCommandRequestTracker()
	123	+	clock := newAttachmentCommandTestClock()
	124	+	requestID, outcome := tracker.Publish(4)
	125	+	waitDone := make(chan error, 1)
	126	+	go func() {
	127	+	_, err := tracker.Wait(context.Background(), clock, requestID, 4, outcome)
	128	+	waitDone <- err
	129	+	}()
	130	+	timer := awaitAttachedCommandTimer(t, clock)
	131	+	timer.fire()
	132	+	require.ErrorIs(t, <-waitDone, ErrCommandRequestTimeout)
	133	+	require.Zero(t, tracker.PendingCount())
	134	+
	135	+	tracker.Complete(4, ports.CommandResult{RequestID: requestID, OK: true})
	136	+	secondID, secondOutcome := tracker.Publish(8)
	137	+	tracker.Complete(4, ports.CommandResult{RequestID: secondID, OK: true})
	138	+	select {
	139	+	case <-secondOutcome:
	140	+	t.Fatal("old-generation result completed a newer request")
	141	+	default:
	142	+	}
	143	+	tracker.Complete(8, ports.CommandResult{RequestID: secondID, OK: true})
	144	+	result := <-secondOutcome
	145	+	require.True(t, result.Result.OK)
	146	+	}
	147	+
121	148		func TestAttachedCommandWaitReleasesSenderLockAfterPublication(t *testing.T) {
122	149		transport := newAttachmentCommandTestTransport()
123	150		gate := make(chan struct{})


@@ -118,6 +118,33 @@ func startAttachedCommand(t testing.T, ac attachedClient, clock *attachmentCom
118	return request, awaitAttachedCommandTimer(t, clock), done	118		return request, awaitAttachedCommandTimer(t, clock), done
119	}	119		}
120		120
		121	+	func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) {
		122	+	tracker := NewCommandRequestTracker()
		123	+	clock := newAttachmentCommandTestClock()
		124	+	requestID, outcome := tracker.Publish(4)
		125	+	waitDone := make(chan error, 1)
		126	+	go func() {
		127	+	_, err := tracker.Wait(context.Background(), clock, requestID, 4, outcome)
		128	+	waitDone <- err
		129	+	}()
		130	+	timer := awaitAttachedCommandTimer(t, clock)
		131	+	timer.fire()
		132	+	require.ErrorIs(t, <-waitDone, ErrCommandRequestTimeout)
		133	+	require.Zero(t, tracker.PendingCount())
		134	+
		135	+	tracker.Complete(4, ports.CommandResult{RequestID: requestID, OK: true})
		136	+	secondID, secondOutcome := tracker.Publish(8)
		137	+	tracker.Complete(4, ports.CommandResult{RequestID: secondID, OK: true})
		138	+	select {
		139	+	case <-secondOutcome:
		140	+	t.Fatal("old-generation result completed a newer request")
		141	+	default:
		142	+	}
		143	+	tracker.Complete(8, ports.CommandResult{RequestID: secondID, OK: true})
		144	+	result := <-secondOutcome
		145	+	require.True(t, result.Result.OK)
		146	+	}
		147	+
121	func TestAttachedCommandWaitReleasesSenderLockAfterPublication(t *testing.T) {	148		func TestAttachedCommandWaitReleasesSenderLockAfterPublication(t *testing.T) {
122	transport := newAttachmentCommandTestTransport()	149		transport := newAttachmentCommandTestTransport()
123	gate := make(chan struct{})	150		gate := make(chan struct{})

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


@@ -86,6 +86,13 @@ type attachedClient struct {
86	86		linkMu sync.Mutex
87	87		sendMu sync.Mutex
88	88		commands attachedCommandTracker
	89	+	// routeCreatedSession marks a session created by this attachment's route.
	90	+	// A handshake that never commits Welcome must tear down that exact empty
	91	+	// session, while an attachment routed to an existing session must not.
	92	+	routeCreatedSession bool
	93	+	// routeSessionPurge distinguishes a brand-new session from a restored
	94	+	// stopped session when failed-handshake cleanup removes the live instance.
	95	+	routeSessionPurge bool
89	96		// beforeAttachmentTokenValidation is a deterministic lifecycle-race seam
90	97		// used only by package tests.
91	98		beforeAttachmentTokenValidation func()


@@ -86,6 +86,13 @@ type attachedClient struct {
86	linkMu sync.Mutex	86		linkMu sync.Mutex
87	sendMu sync.Mutex	87		sendMu sync.Mutex
88	commands attachedCommandTracker	88		commands attachedCommandTracker
		89	+	// routeCreatedSession marks a session created by this attachment's route.
		90	+	// A handshake that never commits Welcome must tear down that exact empty
		91	+	// session, while an attachment routed to an existing session must not.
		92	+	routeCreatedSession bool
		93	+	// routeSessionPurge distinguishes a brand-new session from a restored
		94	+	// stopped session when failed-handshake cleanup removes the live instance.
		95	+	routeSessionPurge bool
89	// beforeAttachmentTokenValidation is a deterministic lifecycle-race seam	96		// beforeAttachmentTokenValidation is a deterministic lifecycle-race seam
90	// used only by package tests.	97		// used only by package tests.
91	beforeAttachmentTokenValidation func()	98		beforeAttachmentTokenValidation func()

M internal/usecase/daemon/control.go +41 -2


@@ -34,8 +34,47 @@ func (d *Daemon) handleCommand(tr ports.Transport, f ports.Frame) error {
34	34		Text: "attached command relay is not enabled",
35	35		})
36	36		}
37		-	result := d.dispatchCommand(d.serveCtx, request)
38		-	result.RequestID = request.RequestID
	37	+
	38	+	tracker := NewCommandRequestTracker()
	39	+	const generation = uint64(1)
	40	+	outcome, ok := tracker.Track(request.RequestID, generation)
	41	+	if !ok {
	42	+	return d.sendCommandResult(tr, ports.CommandResult{
	43	+	RequestID: request.RequestID,
	44	+	Code: ports.ErrInternal,
	45	+	Text: "duplicate command request id",
	46	+	})
	47	+	}
	48	+	commandCtx := d.serveCtx
	49	+	if commandCtx == nil {
	50	+	commandCtx = context.Background()
	51	+	}
	52	+	commandCtx, cancel := context.WithCancel(commandCtx)
	53	+	defer cancel()
	54	+	go func() {
	55	+	result := d.dispatchCommand(commandCtx, request)
	56	+	result.RequestID = request.RequestID
	57	+	tracker.Complete(generation, result)
	58	+	}()
	59	+	commandClock := d.clock
	60	+	if commandClock == nil {
	61	+	commandClock = systemClock{}
	62	+	}
	63	+	result, waitErr := tracker.Wait(context.Background(), commandClock, request.RequestID, generation, outcome)
	64	+	if waitErr != nil {
	65	+	if errors.Is(waitErr, ErrCommandRequestTimeout) {
	66	+	return d.sendCommandResult(tr, ports.CommandResult{
	67	+	RequestID: request.RequestID,
	68	+	Code: ports.ErrInternal,
	69	+	Text: waitErr.Error(),
	70	+	})
	71	+	}
	72	+	return d.sendCommandResult(tr, ports.CommandResult{
	73	+	RequestID: request.RequestID,
	74	+	Code: ports.ErrInternal,
	75	+	Text: waitErr.Error(),
	76	+	})
	77	+	}
39	78		return d.sendCommandResult(tr, result)
40	79		}
41	80


@@ -34,8 +34,47 @@ func (d *Daemon) handleCommand(tr ports.Transport, f ports.Frame) error {
34		Text: "attached command relay is not enabled",	34		Text: "attached command relay is not enabled",
35		})	35		})
36		}	36		}
37	-	result := d.dispatchCommand(d.serveCtx, request)	37	+
38	-	result.RequestID = request.RequestID	38	+	tracker := NewCommandRequestTracker()
			39	+	const generation = uint64(1)
			40	+	outcome, ok := tracker.Track(request.RequestID, generation)
			41	+	if !ok {
			42	+	return d.sendCommandResult(tr, ports.CommandResult{
			43	+	RequestID: request.RequestID,
			44	+	Code: ports.ErrInternal,
			45	+	Text: "duplicate command request id",
			46	+	})
			47	+	}
			48	+	commandCtx := d.serveCtx
			49	+	if commandCtx == nil {
			50	+	commandCtx = context.Background()
			51	+	}
			52	+	commandCtx, cancel := context.WithCancel(commandCtx)
			53	+	defer cancel()
			54	+	go func() {
			55	+	result := d.dispatchCommand(commandCtx, request)
			56	+	result.RequestID = request.RequestID
			57	+	tracker.Complete(generation, result)
			58	+	}()
			59	+	commandClock := d.clock
			60	+	if commandClock == nil {
			61	+	commandClock = systemClock{}
			62	+	}
			63	+	result, waitErr := tracker.Wait(context.Background(), commandClock, request.RequestID, generation, outcome)
			64	+	if waitErr != nil {
			65	+	if errors.Is(waitErr, ErrCommandRequestTimeout) {
			66	+	return d.sendCommandResult(tr, ports.CommandResult{
			67	+	RequestID: request.RequestID,
			68	+	Code: ports.ErrInternal,
			69	+	Text: waitErr.Error(),
			70	+	})
			71	+	}
			72	+	return d.sendCommandResult(tr, ports.CommandResult{
			73	+	RequestID: request.RequestID,
			74	+	Code: ports.ErrInternal,
			75	+	Text: waitErr.Error(),
			76	+	})
			77	+	}
39		return d.sendCommandResult(tr, result)	78		return d.sendCommandResult(tr, result)
40		}	79		}
41			80

M internal/usecase/daemon/control_test.go +25 -0


@@ -124,6 +124,31 @@ func (s *actionRunnerSpy) Run(request daemonActionRequest) error {
124	124		return s.err
125	125		}
126	126
	127	+	func TestHandleCommandTimesOutWithRequestGeneration(t *testing.T) {
	128	+	factory := &controlPTYFactory{entered: make(chan struct{}), release: make(chan struct{})}
	129	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	130	+	d := newTestDaemon(t, factory, clock)
	131	+	addControlSession(d, "work", "t_work", "p_work")
	132	+	tr, sends, releaseConn := newConn(t, commandFrame(t, ports.CommandRequest{
	133	+	RequestID: 17, Slug: "new-tab", TargetSession: "work",
	134	+	}))
	135	+	defer releaseConn()
	136	+	done := make(chan error, 1)
	137	+	go func() {
	138	+	done <- d.handleCommand(tr, commandFrame(t, ports.CommandRequest{RequestID: 17, Slug: "new-tab", TargetSession: "work"}))
	139	+	}()
	140	+	<-factory.entered
	141	+	timer := <-clock.timers
	142	+	require.Equal(t, CommandRequestTimeout, timer.duration)
	143	+	timer.ch <- time.Time{}
	144	+	result := awaitCommandResult(t, sends)
	145	+	require.Equal(t, uint64(17), result.RequestID)
	146	+	require.Equal(t, ports.ErrInternal, result.Code)
	147	+	require.Equal(t, ErrCommandRequestTimeout.Error(), result.Text)
	148	+	require.NoError(t, <-done)
	149	+	close(factory.release)
	150	+	}
	151	+
127	152		func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {
128	153		d := newTestDaemon(t, nil, stubClock{})
129	154		var logs bytes.Buffer


@@ -124,6 +124,31 @@ func (s *actionRunnerSpy) Run(request daemonActionRequest) error {
124	return s.err	124		return s.err
125	}	125		}
126		126
		127	+	func TestHandleCommandTimesOutWithRequestGeneration(t *testing.T) {
		128	+	factory := &controlPTYFactory{entered: make(chan struct{}), release: make(chan struct{})}
		129	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
		130	+	d := newTestDaemon(t, factory, clock)
		131	+	addControlSession(d, "work", "t_work", "p_work")
		132	+	tr, sends, releaseConn := newConn(t, commandFrame(t, ports.CommandRequest{
		133	+	RequestID: 17, Slug: "new-tab", TargetSession: "work",
		134	+	}))
		135	+	defer releaseConn()
		136	+	done := make(chan error, 1)
		137	+	go func() {
		138	+	done <- d.handleCommand(tr, commandFrame(t, ports.CommandRequest{RequestID: 17, Slug: "new-tab", TargetSession: "work"}))
		139	+	}()
		140	+	<-factory.entered
		141	+	timer := <-clock.timers
		142	+	require.Equal(t, CommandRequestTimeout, timer.duration)
		143	+	timer.ch <- time.Time{}
		144	+	result := awaitCommandResult(t, sends)
		145	+	require.Equal(t, uint64(17), result.RequestID)
		146	+	require.Equal(t, ports.ErrInternal, result.Code)
		147	+	require.Equal(t, ErrCommandRequestTimeout.Error(), result.Text)
		148	+	require.NoError(t, <-done)
		149	+	close(factory.release)
		150	+	}
		151	+
127	func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {	152		func TestHandleCommandAttachedRejectionReturnsSendFailure(t *testing.T) {
128	d := newTestDaemon(t, nil, stubClock{})	153		d := newTestDaemon(t, nil, stubClock{})
129	var logs bytes.Buffer	154		var logs bytes.Buffer

M internal/usecase/daemon/daemon.go +39 -14


@@ -1122,7 +1122,7 @@ func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <
1122	1122		return
1123	1123		}
1124	1124
1125		-	sess, ac, rerr := d.route(h, tr)
	1125	+	sess, ac, rerr := d.routeWithContext(handshakeCtx, h, tr)
1126	1126		if rerr != nil {
1127	1127		d.log.Warn("hello rejected", "err", rerr, "intent", h.Intent, "session", h.Name)
1128	1128		if pe, ok := errors.AsType[*protoErr](rerr); ok {

@@ -1134,17 +1134,13 @@ func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut <
1134	1134		return
1135	1135		}
1136	1136
	1137	+	failAttachment := func() {
	1138	+	d.failHandshakeAttachment(sess, ac, tr)
	1139	+	}
1137	1140		if err := handshakeCtx.Err(); err != nil {
	1141	+	failAttachment()
1138	1142		return
1139	1143		}
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		-	}
1148	1144		rc := sess.renderCoordinator()
1149	1145		lease := (*attachmentLease)(nil)
1150	1146		if rc != nil {

@@ -1309,8 +1305,23 @@ func (d *Daemon) waitForTargetRestore(ctx context.Context, name string) error {
1309	1305		}
1310	1306
1311	1307		// route resolves a Hello to a session and a freshly attached client, creating
1312		-	// the session for ephemeral/new intents.
	1308	+	// the session for ephemeral/new intents. Direct package callers retain the
	1309	+	// daemon lifetime context; inbound handshakes use routeWithContext.
1313	1310		func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedClient, error) {
	1311	+	ctx := d.serveCtx
	1312	+	if ctx == nil {
	1313	+	ctx = context.Background()
	1314	+	}
	1315	+	return d.routeWithContext(ctx, h, tr)
	1316	+	}
	1317	+
	1318	+	func (d Daemon) routeWithContext(ctx context.Context, h ports.Hello, tr ports.Transport) (session, *attachedClient, error) {
	1319	+	if ctx == nil {
	1320	+	ctx = context.Background()
	1321	+	}
	1322	+	if err := ctx.Err(); err != nil {
	1323	+	return nil, nil, err
	1324	+	}
1314	1325		sz := h.Size
1315	1326		if !sz.Valid() {
1316	1327		return nil, nil, &protoErr{ports.ErrInternal, "invalid terminal size"}

@@ -1355,10 +1366,6 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
1355	1366		}
1356	1367		}
1357	1368		if h.Intent == ports.IntentResume \|\| h.Intent == ports.IntentAttach {
1358		-	ctx := d.serveCtx
1359		-	if ctx == nil {
1360		-	ctx = context.Background()
1361		-	}
1362	1369		if err := d.waitForTargetRestore(ctx, h.Name); err != nil {
1363	1370		return nil, nil, err
1364	1371		}

@@ -1376,6 +1383,7 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
1376	1383		case ports.IntentResume:
1377	1384		// Resume miss/expiry falls back to normal attach semantics.
1378	1385		sess := d.findByNameLocked(h.Name)
	1386	+	created := false
1379	1387		if sess == nil {
1380	1388		stopped, ok := d.stopped[h.Name]
1381	1389		if !ok \|\| stopped.purging {

@@ -1389,8 +1397,12 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
1389	1397		d.mu.Unlock()
1390	1398		return nil, nil, err
1391	1399		}
	1400	+	created = true
1392	1401		}
1393	1402		ac, err := d.finishAttach(sess, tr, sz, term, h)
	1403	+	if err == nil && created {
	1404	+	ac.routeCreatedSession = true
	1405	+	}
1394	1406		return sess, ac, err
1395	1407
1396	1408		case ports.IntentEphemeral:

@@ -1401,6 +1413,10 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
1401	1413		return nil, nil, err
1402	1414		}
1403	1415		ac, err := d.finishAttach(sess, tr, sz, term, h)
	1416	+	if err == nil {
	1417	+	ac.routeCreatedSession = true
	1418	+	ac.routeSessionPurge = true
	1419	+	}
1404	1420		return sess, ac, err
1405	1421
1406	1422		case ports.IntentNew:

@@ -1422,10 +1438,15 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
1422	1438		return nil, nil, err
1423	1439		}
1424	1440		ac, err := d.finishAttach(sess, tr, sz, term, h)
	1441	+	if err == nil {
	1442	+	ac.routeCreatedSession = true
	1443	+	ac.routeSessionPurge = true
	1444	+	}
1425	1445		return sess, ac, err
1426	1446
1427	1447		case ports.IntentAttach:
1428	1448		sess := d.findByNameLocked(h.Name)
	1449	+	created := false
1429	1450		if sess == nil {
1430	1451		stopped, ok := d.stopped[h.Name]
1431	1452		if !ok \|\| stopped.purging {

@@ -1439,8 +1460,12 @@ func (d Daemon) route(h ports.Hello, tr ports.Transport) (session, *attachedCl
1439	1460		d.mu.Unlock()
1440	1461		return nil, nil, err
1441	1462		}
	1463	+	created = true
1442	1464		}
1443	1465		ac, err := d.finishAttach(sess, tr, sz, term, h)
	1466	+	if err == nil && created {
	1467	+	ac.routeCreatedSession = true
	1468	+	}
1444	1469		return sess, ac, err
1445	1470
1446	1471		default:

 		return
 	}
-	sess, ac, rerr := d.route(h, tr)
+	sess, ac, rerr := d.routeWithContext(handshakeCtx, h, tr)
 	if rerr != nil {
 		d.log.Warn("hello rejected", "err", rerr, "intent", h.Intent, "session", h.Name)
 		if pe, ok := errors.AsType[*protoErr](rerr); ok {
 		return
 	}
+	failAttachment := func() {
+		d.failHandshakeAttachment(sess, ac, tr)
+	}
 	if err := handshakeCtx.Err(); err != nil {
+		failAttachment()
 		return
 	}
-	failAttachment := func() {
-		if !d.abortResumeClaim(ac) {
-			// A fresh attachment has no credential to park. Explicit teardown
-			// removes the unpublished attachment instead of creating a resumable
-			// entry for a handshake that never committed Welcome.
-			d.clientGone(sess, ac, tr, true)
-		}
-	}
 	rc := sess.renderCoordinator()
 	lease := (*attachmentLease)(nil)
 	if rc != nil {
 }
 // route resolves a Hello to a session and a freshly attached client, creating
-// the session for ephemeral/new intents.
+// the session for ephemeral/new intents. Direct package callers retain the
+// daemon lifetime context; inbound handshakes use routeWithContext.
 func (d *Daemon) route(h ports.Hello, tr ports.Transport) (*session, *attachedClient, error) {
+	ctx := d.serveCtx
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	return d.routeWithContext(ctx, h, tr)
+}
+func (d *Daemon) routeWithContext(ctx context.Context, h ports.Hello, tr ports.Transport) (*session, *attachedClient, error) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, nil, err
+	}
 	sz := h.Size
 	if !sz.Valid() {
 		return nil, nil, &protoErr{ports.ErrInternal, "invalid terminal size"}
 		}
 	}
 	if h.Intent == ports.IntentResume || h.Intent == ports.IntentAttach {
-		ctx := d.serveCtx
-		if ctx == nil {
-			ctx = context.Background()
-		}
 		if err := d.waitForTargetRestore(ctx, h.Name); err != nil {
 			return nil, nil, err
 		}
 	case ports.IntentResume:
 		// Resume miss/expiry falls back to normal attach semantics.
 		sess := d.findByNameLocked(h.Name)
+		created := false
 		if sess == nil {
 			stopped, ok := d.stopped[h.Name]
 			if !ok || stopped.purging {
 				d.mu.Unlock()
 				return nil, nil, err
 			}
+			created = true
 		}
 		ac, err := d.finishAttach(sess, tr, sz, term, h)
+		if err == nil && created {
+			ac.routeCreatedSession = true
+		}
 		return sess, ac, err
 	case ports.IntentEphemeral:
 			return nil, nil, err
 		}
 		ac, err := d.finishAttach(sess, tr, sz, term, h)
+		if err == nil {
+			ac.routeCreatedSession = true
+			ac.routeSessionPurge = true
+		}
 		return sess, ac, err
 	case ports.IntentNew:
 			return nil, nil, err
 		}
 		ac, err := d.finishAttach(sess, tr, sz, term, h)
+		if err == nil {
+			ac.routeCreatedSession = true
+			ac.routeSessionPurge = true
+		}
 		return sess, ac, err
 	case ports.IntentAttach:
 		sess := d.findByNameLocked(h.Name)
+		created := false
 		if sess == nil {
 			stopped, ok := d.stopped[h.Name]
 			if !ok || stopped.purging {
 				d.mu.Unlock()
 				return nil, nil, err
 			}
+			created = true
 		}
 		ac, err := d.finishAttach(sess, tr, sz, term, h)
+		if err == nil && created {
+			ac.routeCreatedSession = true
+		}
 		return sess, ac, err
 	default:

M internal/usecase/daemon/handshake.go +35 -0


@@ -107,3 +107,38 @@ func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fal
107	107		}
108	108		return fallback
109	109		}
	110	+
	111	+	// failHandshakeAttachment synchronously retires the exact connection admitted
	112	+	// by route. A fresh route-owned session is purged only after its attachment is
	113	+	// removed; existing sessions and their unrelated attachments are preserved.
	114	+	func (d Daemon) failHandshakeAttachment(sess session, ac *attachedClient, tr ports.Transport) {
	115	+	if sess == nil \|\| ac == nil {
	116	+	if tr != nil {
	117	+	_ = tr.Close()
	118	+	}
	119	+	return
	120	+	}
	121	+	if d.abortResumeClaim(ac) {
	122	+	// abortResumeClaim closes the replacement transport and restores the
	123	+	// original parked credential. Closing the exact handshake link again is
	124	+	// harmless for transports and makes this boundary synchronous.
	125	+	if tr != nil {
	126	+	_ = tr.Close()
	127	+	}
	128	+	return
	129	+	}
	130	+	d.clientGone(sess, ac, tr, true)
	131	+	if tr != nil {
	132	+	_ = tr.Close()
	133	+	}
	134	+	if ac.routeCreatedSession {
	135	+	sess.mu.Lock()
	136	+	empty := len(sess.attachments) == 0
	137	+	sess.mu.Unlock()
	138	+	if empty {
	139	+	// killSession revalidates the attachment set under its publication
	140	+	// locks before removing the session.
	141	+	_ = d.killSession(sess, ports.ReasonSessionKilled, ac.routeSessionPurge)
	142	+	}
	143	+	}
	144	+	}


@@ -107,3 +107,38 @@ func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fal
107	}	107		}
108	return fallback	108		return fallback
109	}	109		}
		110	+
		111	+	// failHandshakeAttachment synchronously retires the exact connection admitted
		112	+	// by route. A fresh route-owned session is purged only after its attachment is
		113	+	// removed; existing sessions and their unrelated attachments are preserved.
		114	+	func (d Daemon) failHandshakeAttachment(sess session, ac *attachedClient, tr ports.Transport) {
		115	+	if sess == nil \|\| ac == nil {
		116	+	if tr != nil {
		117	+	_ = tr.Close()
		118	+	}
		119	+	return
		120	+	}
		121	+	if d.abortResumeClaim(ac) {
		122	+	// abortResumeClaim closes the replacement transport and restores the
		123	+	// original parked credential. Closing the exact handshake link again is
		124	+	// harmless for transports and makes this boundary synchronous.
		125	+	if tr != nil {
		126	+	_ = tr.Close()
		127	+	}
		128	+	return
		129	+	}
		130	+	d.clientGone(sess, ac, tr, true)
		131	+	if tr != nil {
		132	+	_ = tr.Close()
		133	+	}
		134	+	if ac.routeCreatedSession {
		135	+	sess.mu.Lock()
		136	+	empty := len(sess.attachments) == 0
		137	+	sess.mu.Unlock()
		138	+	if empty {
		139	+	// killSession revalidates the attachment set under its publication
		140	+	// locks before removing the session.
		141	+	_ = d.killSession(sess, ports.ReasonSessionKilled, ac.routeSessionPurge)
		142	+	}
		143	+	}
		144	+	}

M internal/usecase/daemon/handshake_test.go +108 -6

@@ -1,6 +1,7 @@
1	1		package daemon
2	2
3	3		import (
	4	+	"context"
4	5		"io"
5	6		"sync"
6	7		"testing"

@@ -122,12 +123,7 @@ func TestHandshakeTimeoutClosesBlockedWelcomeSend(t *testing.T) {
122	123		awaitTestCompletion(t, done, "handshake send did not stop after timeout")
123	124		requireClosedHandshakeTransport(t, tr)
124	125		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())
	126	+	require.Empty(t, d.sessions, "a timed-out newly-created handshake must not leave an empty session")
131	127		require.Empty(t, d.parked)
132	128		d.mu.Unlock()
133	129		}

@@ -140,3 +136,109 @@ func requireClosedHandshakeTransport(t testing.T, tr handshakeBlockingTranspor
140	136		t.Fatal("handshake timeout did not close transport")
141	137		}
142	138		}
	139	+
	140	+	func TestHandshakeTimeoutCancelsRouteRestoreWait(t *testing.T) {
	141	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	142	+	d := newTestDaemon(t, nil, clock)
	143	+	restoreDone := make(chan struct{})
	144	+	d.mu.Lock()
	145	+	d.stopped["restoring"] = stoppedSession{name: "restoring", restoreDone: restoreDone}
	146	+	d.mu.Unlock()
	147	+
	148	+	tr := newHandshakeBlockingTransport(false)
	149	+	done := make(chan struct{})
	150	+	go func() {
	151	+	d.handleHello(tr, mustHello(ports.IntentAttach, "restoring", domain.Size{Cols: 80, Rows: 24}))
	152	+	close(done)
	153	+	}()
	154	+
	155	+	timer := <-clock.timers
	156	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	157	+	timer.ch <- time.Time{}
	158	+	awaitTestCompletion(t, done, "route restore wait did not stop after handshake timeout")
	159	+	requireClosedHandshakeTransport(t, tr)
	160	+	d.mu.Lock()
	161	+	require.Empty(t, d.sessions)
	162	+	require.Empty(t, d.parked)
	163	+	d.mu.Unlock()
	164	+	}
	165	+
	166	+	func TestHandshakeTimeoutRemovesRestoredEmptySession(t *testing.T) {
	167	+	pty, release := newBlockingPTY(t)
	168	+	defer release()
	169	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	170	+	d.mu.Lock()
	171	+	d.stopped["restored"] = stoppedSession{name: "restored", cwd: "/tmp"}
	172	+	d.mu.Unlock()
	173	+
	174	+	clock := &signalClock{timers: make(chan *signalTimer, 4)}
	175	+	d.clock = clock
	176	+	tr := newHandshakeBlockingTransport(true)
	177	+	done := make(chan struct{})
	178	+	go func() {
	179	+	d.handleHello(tr, mustHello(ports.IntentAttach, "restored", domain.Size{Cols: 80, Rows: 24}))
	180	+	close(done)
	181	+	}()
	182	+
	183	+	timer := <-clock.timers
	184	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	185	+	<-tr.welcome
	186	+	timer.ch <- time.Time{}
	187	+	awaitTestCompletion(t, done, "timed-out restored attachment cleanup did not finish")
	188	+	requireClosedHandshakeTransport(t, tr)
	189	+	d.mu.Lock()
	190	+	require.Empty(t, d.sessions)
	191	+	_, retained := d.stopped["restored"]
	192	+	d.mu.Unlock()
	193	+	require.True(t, retained, "failed attach must retain the stopped session authority")
	194	+	}
	195	+
	196	+	func TestHandshakeTimeoutPreservesUnrelatedAttachment(t *testing.T) {
	197	+	pty, release := newBlockingPTY(t)
	198	+	defer release()
	199	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	200	+	oldTransport := &closeTrackingTransport{}
	201	+	sess, old, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "shared", Size: domain.Size{Cols: 80, Rows: 24}}, oldTransport)
	202	+	require.NoError(t, err)
	203	+
	204	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
	205	+	d.clock = clock
	206	+	tr := newHandshakeBlockingTransport(true)
	207	+	done := make(chan struct{})
	208	+	go func() {
	209	+	d.handleHello(tr, mustHello(ports.IntentAttach, "shared", domain.Size{Cols: 80, Rows: 24}))
	210	+	close(done)
	211	+	}()
	212	+
	213	+	timer := <-clock.timers
	214	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
	215	+	<-tr.welcome
	216	+	timer.ch <- time.Time{}
	217	+	awaitTestCompletion(t, done, "timed-out attachment cleanup did not finish")
	218	+	requireClosedHandshakeTransport(t, tr)
	219	+	require.False(t, oldTransport.Closed(), "cleanup closed an unrelated attachment transport")
	220	+	require.Same(t, sess, old.currentAttachmentSession())
	221	+	require.Equal(t, []*attachedClient{old}, sess.snapshotAttachments())
	222	+	d.mu.Lock()
	223	+	require.Len(t, d.sessions, 1)
	224	+	require.Empty(t, d.parked)
	225	+	d.mu.Unlock()
	226	+	_ = d.killSession(sess, ports.ReasonServerShutdown, false)
	227	+	}
	228	+
	229	+	func TestSuccessfulRouteOutlivesStoppedHandshakeContext(t *testing.T) {
	230	+	pty, release := newBlockingPTY(t)
	231	+	defer release()
	232	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
	233	+	ctx, cancel := context.WithCancel(context.Background())
	234	+	sess, ac, err := d.routeWithContext(ctx, ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "lifetime", Size: domain.Size{Cols: 80, Rows: 24}}, &closeTrackingTransport{})
	235	+	require.NoError(t, err)
	236	+	cancel()
	237	+	select {
	238	+	case <-sess.ctx.Done():
	239	+	t.Fatal("session lifetime was parented to the handshake context")
	240	+	default:
	241	+	}
	242	+	require.NotNil(t, ac)
	243	+	_ = d.killSession(sess, ports.ReasonServerShutdown, false)
	244	+	}

@@ -1,6 +1,7 @@
1		package daemon	1		package daemon
2			2
3		import (	3		import (
			4	+	"context"
4		"io"	5		"io"
5		"sync"	6		"sync"
6		"testing"	7		"testing"

@@ -122,12 +123,7 @@ func TestHandshakeTimeoutClosesBlockedWelcomeSend(t *testing.T) {
122		awaitTestCompletion(t, done, "handshake send did not stop after timeout")	123		awaitTestCompletion(t, done, "handshake send did not stop after timeout")
123		requireClosedHandshakeTransport(t, tr)	124		requireClosedHandshakeTransport(t, tr)
124		d.mu.Lock()	125		d.mu.Lock()
125	-	require.Len(t, d.sessions, 1)	126	+	require.Empty(t, d.sessions, "a timed-out newly-created handshake must not leave an empty session")
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)	127		require.Empty(t, d.parked)
132		d.mu.Unlock()	128		d.mu.Unlock()
133		}	129		}

@@ -140,3 +136,109 @@ func requireClosedHandshakeTransport(t testing.T, tr handshakeBlockingTranspor
140		t.Fatal("handshake timeout did not close transport")	136		t.Fatal("handshake timeout did not close transport")
141		}	137		}
142		}	138		}
			139	+
			140	+	func TestHandshakeTimeoutCancelsRouteRestoreWait(t *testing.T) {
			141	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
			142	+	d := newTestDaemon(t, nil, clock)
			143	+	restoreDone := make(chan struct{})
			144	+	d.mu.Lock()
			145	+	d.stopped["restoring"] = stoppedSession{name: "restoring", restoreDone: restoreDone}
			146	+	d.mu.Unlock()
			147	+
			148	+	tr := newHandshakeBlockingTransport(false)
			149	+	done := make(chan struct{})
			150	+	go func() {
			151	+	d.handleHello(tr, mustHello(ports.IntentAttach, "restoring", domain.Size{Cols: 80, Rows: 24}))
			152	+	close(done)
			153	+	}()
			154	+
			155	+	timer := <-clock.timers
			156	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
			157	+	timer.ch <- time.Time{}
			158	+	awaitTestCompletion(t, done, "route restore wait did not stop after handshake timeout")
			159	+	requireClosedHandshakeTransport(t, tr)
			160	+	d.mu.Lock()
			161	+	require.Empty(t, d.sessions)
			162	+	require.Empty(t, d.parked)
			163	+	d.mu.Unlock()
			164	+	}
			165	+
			166	+	func TestHandshakeTimeoutRemovesRestoredEmptySession(t *testing.T) {
			167	+	pty, release := newBlockingPTY(t)
			168	+	defer release()
			169	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
			170	+	d.mu.Lock()
			171	+	d.stopped["restored"] = stoppedSession{name: "restored", cwd: "/tmp"}
			172	+	d.mu.Unlock()
			173	+
			174	+	clock := &signalClock{timers: make(chan *signalTimer, 4)}
			175	+	d.clock = clock
			176	+	tr := newHandshakeBlockingTransport(true)
			177	+	done := make(chan struct{})
			178	+	go func() {
			179	+	d.handleHello(tr, mustHello(ports.IntentAttach, "restored", domain.Size{Cols: 80, Rows: 24}))
			180	+	close(done)
			181	+	}()
			182	+
			183	+	timer := <-clock.timers
			184	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
			185	+	<-tr.welcome
			186	+	timer.ch <- time.Time{}
			187	+	awaitTestCompletion(t, done, "timed-out restored attachment cleanup did not finish")
			188	+	requireClosedHandshakeTransport(t, tr)
			189	+	d.mu.Lock()
			190	+	require.Empty(t, d.sessions)
			191	+	_, retained := d.stopped["restored"]
			192	+	d.mu.Unlock()
			193	+	require.True(t, retained, "failed attach must retain the stopped session authority")
			194	+	}
			195	+
			196	+	func TestHandshakeTimeoutPreservesUnrelatedAttachment(t *testing.T) {
			197	+	pty, release := newBlockingPTY(t)
			198	+	defer release()
			199	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
			200	+	oldTransport := &closeTrackingTransport{}
			201	+	sess, old, err := d.route(ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "shared", Size: domain.Size{Cols: 80, Rows: 24}}, oldTransport)
			202	+	require.NoError(t, err)
			203	+
			204	+	clock := &signalClock{timers: make(chan *signalTimer, 1)}
			205	+	d.clock = clock
			206	+	tr := newHandshakeBlockingTransport(true)
			207	+	done := make(chan struct{})
			208	+	go func() {
			209	+	d.handleHello(tr, mustHello(ports.IntentAttach, "shared", domain.Size{Cols: 80, Rows: 24}))
			210	+	close(done)
			211	+	}()
			212	+
			213	+	timer := <-clock.timers
			214	+	require.Equal(t, ports.HandshakeTimeout, timer.duration)
			215	+	<-tr.welcome
			216	+	timer.ch <- time.Time{}
			217	+	awaitTestCompletion(t, done, "timed-out attachment cleanup did not finish")
			218	+	requireClosedHandshakeTransport(t, tr)
			219	+	require.False(t, oldTransport.Closed(), "cleanup closed an unrelated attachment transport")
			220	+	require.Same(t, sess, old.currentAttachmentSession())
			221	+	require.Equal(t, []*attachedClient{old}, sess.snapshotAttachments())
			222	+	d.mu.Lock()
			223	+	require.Len(t, d.sessions, 1)
			224	+	require.Empty(t, d.parked)
			225	+	d.mu.Unlock()
			226	+	_ = d.killSession(sess, ports.ReasonServerShutdown, false)
			227	+	}
			228	+
			229	+	func TestSuccessfulRouteOutlivesStoppedHandshakeContext(t *testing.T) {
			230	+	pty, release := newBlockingPTY(t)
			231	+	defer release()
			232	+	d := newTestDaemon(t, newFactory(t, pty), stubClock{})
			233	+	ctx, cancel := context.WithCancel(context.Background())
			234	+	sess, ac, err := d.routeWithContext(ctx, ports.Hello{Version: ports.ProtocolVersion, Intent: ports.IntentNew, Name: "lifetime", Size: domain.Size{Cols: 80, Rows: 24}}, &closeTrackingTransport{})
			235	+	require.NoError(t, err)
			236	+	cancel()
			237	+	select {
			238	+	case <-sess.ctx.Done():
			239	+	t.Fatal("session lifetime was parented to the handshake context")
			240	+	default:
			241	+	}
			242	+	require.NotNil(t, ac)
			243	+	_ = d.killSession(sess, ports.ReasonServerShutdown, false)
			244	+	}