1package client_test
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "io"
8 "log/slog"
9 "os"
10 "sync"
11 "sync/atomic"
12 "testing"
13 "time"
14
15 "github.com/stretchr/testify/mock"
16 "github.com/stretchr/testify/require"
17
18 "github.com/bnema/vev/internal/domain"
19 "github.com/bnema/vev/internal/ports"
20 portsmocks "github.com/bnema/vev/internal/ports/mocks"
21 "github.com/bnema/vev/internal/usecase/client"
22)
23
24func init() {
25 // Keep client diagnostics out of the test runner's stderr.
26 slog.SetDefault(slog.New(slog.NewTextHandler(io.Discard, nil)))
27}
28
29// realClock is a stdlib-backed ports.Clock for tests that exercise the stdin
30// pump end to end; the paste coalescer's flush timer runs on wall time here.
31type realClock struct{}
32
33func (realClock) Now() time.Time { return time.Now() }
34func (realClock) NewTimer(d time.Duration) ports.Timer {
35 return &realTimer{t: time.NewTimer(d)}
36}
37
38type realTimer struct{ t *time.Timer }
39
40func (r *realTimer) C() <-chan time.Time { return r.t.C }
41func (r *realTimer) Reset(d time.Duration) bool { return r.t.Reset(d) }
42func (r *realTimer) Stop() bool { return r.t.Stop() }
43
44type reconnectTestClock struct {
45 mu sync.Mutex
46 timers []*reconnectTestTimer
47}
48
49func (c *reconnectTestClock) Now() time.Time { return time.Time{} }
50func (c *reconnectTestClock) NewTimer(d time.Duration) ports.Timer {
51 t := &reconnectTestTimer{ch: make(chan time.Time, 1), duration: d}
52 c.mu.Lock()
53 c.timers = append(c.timers, t)
54 c.mu.Unlock()
55 return t
56}
57
58func (c *reconnectTestClock) fireReconnect(t *testing.T) {
59 t.Helper()
60 require.Eventually(t, func() bool {
61 c.mu.Lock()
62 defer c.mu.Unlock()
63 for _, timer := range c.timers {
64 timer.mu.Lock()
65 if timer.duration != 100*time.Millisecond {
66 timer.mu.Unlock()
67 continue
68 }
69 if !timer.stopped && !timer.fired {
70 timer.fired = true
71 select {
72 case timer.ch <- time.Time{}:
73 default:
74 timer.fired = false
75 timer.mu.Unlock()
76 continue
77 }
78 timer.mu.Unlock()
79 return true
80 }
81 timer.mu.Unlock()
82 }
83 return false
84 }, time.Second, time.Millisecond)
85}
86
87type reconnectTestTimer struct {
88 mu sync.Mutex
89 ch chan time.Time
90 duration time.Duration
91 stopped bool
92 fired bool
93}
94
95func (t *reconnectTestTimer) C() <-chan time.Time { return t.ch }
96func (t *reconnectTestTimer) Reset(d time.Duration) bool {
97 t.mu.Lock()
98 t.duration = d
99 t.fired = false
100 t.stopped = false
101 t.mu.Unlock()
102 return true
103}
104func (t *reconnectTestTimer) Stop() bool {
105 t.mu.Lock()
106 defer t.mu.Unlock()
107 wasStopped := t.stopped
108 t.stopped = true
109 return !wasStopped
110}
111
112// recvItem is one scripted Recv result.
113type recvItem struct {
114 f ports.Frame
115 err error
116}
117
118type markedDatagramTransport struct{ ports.Transport }
119
120func (markedDatagramTransport) DatagramTransport() {}
121
122// scriptRecv wires a MockTransport's Recv to yield the given items in order,
123// then block forever (simulating a live but idle connection). It returns a
124// cleanup that unblocks the parked reader.
125func scriptRecv(tr *portsmocks.MockTransport, items ...recvItem) func() {
126 ch := make(chan recvItem, len(items))
127 for _, it := range items {
128 ch <- it
129 }
130 done := make(chan struct{})
131 tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
132 select {
133 case it := <-ch:
134 return it.f, it.err
135 case <-done:
136 return ports.Frame{}, io.EOF
137 }
138 }).Maybe()
139 return func() { close(done) }
140}
141
142func frameOf(t ports.MsgType, payload []byte) ports.Frame {
143 return ports.Frame{Type: t, Payload: payload}
144}
145
146// blockingReader blocks on Read until closed, then returns EOF. Stands in
147// for a terminal's stdin that produces nothing during the test.
148type blockingReader struct{ ch chan struct{} }
149
150func newBlockingReader() *blockingReader { return &blockingReader{ch: make(chan struct{})} }
151
152func (b *blockingReader) Read(p []byte) (int, error) {
153 <-b.ch
154 return 0, io.EOF
155}
156func (b *blockingReader) unblock() { close(b.ch) }
157
158type oneShotBlockingReader struct {
159 data []byte
160 done chan struct{}
161 once sync.Once
162}
163
164func newOneShotBlockingReader(data []byte) *oneShotBlockingReader {
165 return &oneShotBlockingReader{data: data, done: make(chan struct{})}
166}
167
168func (r *oneShotBlockingReader) Read(p []byte) (int, error) {
169 read := false
170 r.once.Do(func() {
171 copy(p, r.data)
172 read = true
173 })
174 if read {
175 return len(r.data), nil
176 }
177 <-r.done
178 return 0, io.EOF
179}
180
181func (r *oneShotBlockingReader) unblock() { close(r.done) }
182
183type chunkedBlockingReader struct {
184 chunks [][]byte
185 done chan struct{}
186 mu sync.Mutex
187 next int
188}
189
190func newChunkedBlockingReader(chunks ...[]byte) *chunkedBlockingReader {
191 return &chunkedBlockingReader{chunks: chunks, done: make(chan struct{})}
192}
193
194func (r *chunkedBlockingReader) Read(p []byte) (int, error) {
195 r.mu.Lock()
196 defer r.mu.Unlock()
197 if r.next < len(r.chunks) {
198 chunk := r.chunks[r.next]
199 r.next++
200 return copy(p, chunk), nil
201 }
202 <-r.done
203 return 0, io.EOF
204}
205
206func (r *chunkedBlockingReader) unblock() { close(r.done) }
207
208func newHappyTerminal(t *testing.T, out *bytes.Buffer, restoreCount *atomic.Int32, resizeCh chan domain.Size) (*portsmocks.MockTerminal, *blockingReader) {
209 tm := portsmocks.NewMockTerminal(t)
210 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
211 tm.EXPECT().EnterRaw().Return(func() error {
212 restoreCount.Add(1)
213 return nil
214 }, nil).Once()
215 in := newBlockingReader()
216 tm.EXPECT().In().Return(in).Maybe()
217 tm.EXPECT().Out().Return(out).Maybe()
218 tm.EXPECT().Flush().Return(nil).Maybe()
219 tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
220 return tm, in
221}
222
223func isType(typ ports.MsgType) any {
224 return mock.MatchedBy(func(f ports.Frame) bool { return f.Type == typ })
225}
226
227// transportDialer adapts one already-open test transport to the Runner API.
228type transportDialer struct{ transport ports.Transport }
229
230func (d transportDialer) Dial(context.Context) (ports.Transport, error) {
231 return d.transport, nil
232}
233
234func runTestClient(ctx context.Context, deps client.Dependencies, request client.AttachRequest) error {
235 return client.NewRunner(deps).Run(ctx, request)
236}
237
238func testDependencies(dialer ports.Dialer, terminal ports.Terminal, clock ports.Clock, clipboard ports.ClipboardReader, observer ports.SerializedRuntimeObserver) client.Dependencies {
239 return client.Dependencies{
240 Dialer: dialer,
241 Terminal: terminal,
242 Clock: clock,
243 Clipboard: clipboard,
244 Logger: slog.New(slog.DiscardHandler),
245 RuntimeObserver: observer,
246 }
247}
248
249func attachTestDependencies(transport ports.Transport, terminal ports.Terminal, clock ports.Clock) client.Dependencies {
250 return testDependencies(transportDialer{transport: transport}, terminal, clock, nil, nil)
251}
252
253const testPreWelcomeTimeout = ports.HandshakeTimeout
254
255func newHandshakeClock(t *testing.T, timerCount int) (*portsmocks.MockClock, <-chan chan time.Time) {
256 t.Helper()
257 created := make(chan chan time.Time, timerCount)
258 clk := portsmocks.NewMockClock(t)
259 clk.EXPECT().NewTimer(testPreWelcomeTimeout).RunAndReturn(func(time.Duration) ports.Timer {
260 timerC := make(chan time.Time, 1)
261 timer := portsmocks.NewMockTimer(t)
262 timer.EXPECT().C().Return((<-chan time.Time)(timerC)).Maybe()
263 timer.EXPECT().Stop().Return(true).Once()
264 created <- timerC
265 return timer
266 }).Times(timerCount)
267 return clk, created
268}
269
270func TestRunBoundsBlockedDial(t *testing.T) {
271 clk, createdTimers := newHandshakeClock(t, 1)
272 started := make(chan struct{})
273 dialer := portsmocks.NewMockDialer(t)
274 dialer.EXPECT().Dial(mock.Anything).RunAndReturn(func(ctx context.Context) (ports.Transport, error) {
275 close(started)
276 <-ctx.Done()
277 return nil, ctx.Err()
278 }).Once()
279
280 errCh := make(chan error, 1)
281 go func() {
282 errCh <- runTestClient(context.Background(), testDependencies(dialer, nil, clk, nil, nil), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main"})
283 }()
284 timerC := <-createdTimers
285 <-started
286 timerC <- time.Time{}
287 require.ErrorIs(t, <-errCh, context.DeadlineExceeded)
288}
289
290func TestRunBoundsPreWelcomeOperations(t *testing.T) {
291 tests := []struct {
292 name string
293 phase string
294 end string
295 want error
296 }{
297 {name: "cancel while sending hello", phase: "send", end: "cancel", want: context.Canceled},
298 {name: "timeout while sending hello", phase: "send", end: "timeout", want: context.DeadlineExceeded},
299 {name: "cancel while receiving welcome", phase: "recv", end: "cancel", want: context.Canceled},
300 {name: "timeout while receiving welcome", phase: "recv", end: "timeout", want: context.DeadlineExceeded},
301 }
302
303 for _, tt := range tests {
304 t.Run(tt.name, func(t *testing.T) {
305 ctx, cancel := context.WithCancel(context.Background())
306 defer cancel()
307 clk, createdTimers := newHandshakeClock(t, 1)
308
309 term := portsmocks.NewMockTerminal(t)
310 term.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
311 // No EnterRaw expectation: it must not run before Welcome.
312
313 started := make(chan struct{})
314 closed := make(chan struct{})
315 var closeOnce sync.Once
316 operationExited := make(chan struct{})
317 tr := portsmocks.NewMockTransport(t)
318 tr.EXPECT().Close().Run(func() { closeOnce.Do(func() { close(closed) }) }).Return(nil).Maybe()
319
320 block := func() error {
321 close(started)
322 <-closed
323 close(operationExited)
324 return io.ErrClosedPipe
325 }
326 if tt.phase == "send" {
327 tr.EXPECT().Send(isType(ports.MsgHello)).RunAndReturn(func(ports.Frame) error {
328 return block()
329 }).Once()
330 } else {
331 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
332 tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
333 return ports.Frame{}, block()
334 }).Once()
335 }
336 dialer := portsmocks.NewMockDialer(t)
337 dialer.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
338
339 errCh := make(chan error, 1)
340 go func() {
341 errCh <- runTestClient(ctx, testDependencies(dialer, term, clk, nil, nil), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: false})
342 }()
343
344 var timerC chan time.Time
345 select {
346 case timerC = <-createdTimers:
347 case <-time.After(time.Second):
348 t.Fatal("handshake timer was not created")
349 }
350 select {
351 case <-started:
352 case <-time.After(time.Second):
353 t.Fatal("handshake operation did not start")
354 }
355 if tt.end == "cancel" {
356 cancel()
357 } else {
358 timerC <- time.Time{}
359 }
360
361 select {
362 case err := <-errCh:
363 require.ErrorIs(t, err, tt.want)
364 case <-time.After(time.Second):
365 t.Fatal("Run did not return after handshake cancellation")
366 }
367 select {
368 case <-operationExited:
369 case <-time.After(time.Second):
370 t.Fatal("handshake operation leaked after Run returned")
371 }
372 tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
373 })
374 }
375}
376
377func TestAttachHelloIncludesTrueColor(t *testing.T) {
378 t.Setenv("TERM", "xterm-256color")
379 t.Setenv("COLORTERM", "truecolor")
380
381 var out bytes.Buffer
382 var restoreCount atomic.Int32
383 resizeCh := make(chan domain.Size)
384 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
385 defer in.unblock()
386
387 gotHello := make(chan ports.Hello, 1)
388 tr := portsmocks.NewMockTransport(t)
389 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
390 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
391 tr.EXPECT().Send(isType(ports.MsgHello)).RunAndReturn(func(f ports.Frame) error {
392 hello, err := ports.UnmarshalHello(f.Payload)
393 require.NoError(t, err)
394 gotHello <- hello
395 return nil
396 }).Once()
397 unblock := scriptRecv(tr,
398 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
399 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
400 )
401 defer unblock()
402 tr.EXPECT().Close().Return(nil).Once()
403
404 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: ""})
405 require.NoError(t, err)
406
407 select {
408 case hello := <-gotHello:
409 require.Equal(t, "xterm-256color", hello.TermEnv)
410 require.True(t, hello.TrueColor)
411 require.Equal(t, uint8(8), hello.MaxOutputInFlight)
412 case <-time.After(2 * time.Second):
413 t.Fatal("hello frame was not sent")
414 }
415}
416
417func TestAttachPublishesCommittedRouteSnapshot(t *testing.T) {
418 term := newRunTerminal()
419 defer term.in.unblock()
420 lifecycle := domain.SessionLifecycleID{9, 8, 7}
421 tr := &recordingTransport{recvs: []recvItem{
422 {f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
423 SessionID: "s1",
424 SessionName: "main",
425 CommittedIdentity: &ports.CommittedRouteIdentity{
426 Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "main"},
427 },
428 }))},
429 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
430 }}
431
432 err := runTestClient(context.Background(), attachTestDependencies(tr, term, realClock{}), client.AttachRequest{
433 Intent: ports.IntentAttach,
434 SessionName: "main",
435 Origin: ports.RouteOriginLocal,
436 OriginKey: "local",
437 })
438 require.NoError(t, err)
439
440 var snapshot ports.RecentRouteSnapshot
441 var subscription ports.RouteAttentionSubscription
442 foundSnapshot := false
443 foundSubscription := false
444 for _, frame := range tr.Sends() {
445 switch frame.Type {
446 case ports.MsgRecentRouteSnapshot:
447 var err error
448 snapshot, err = ports.UnmarshalRecentRouteSnapshot(frame.Payload)
449 require.NoError(t, err)
450 foundSnapshot = true
451 case ports.MsgRouteAttentionSubscription:
452 var err error
453 subscription, err = ports.UnmarshalRouteAttentionSubscription(frame.Payload)
454 require.NoError(t, err)
455 foundSubscription = true
456 }
457 }
458 require.True(t, foundSnapshot, "successful Welcome must publish a route snapshot")
459 require.True(t, foundSubscription, "successful Welcome must publish a route attention subscription")
460 require.Empty(t, subscription.Targets, "the active route itself needs no status subscription")
461 require.NotZero(t, snapshot.Generation)
462 require.NotZero(t, snapshot.Active.Key)
463 require.Empty(t, snapshot.Entries, "the active route is metadata-only")
464}
465
466func TestLocalOnlyHandoffBetweenLocalSessionsKeepsClientRunning(t *testing.T) {
467 term := newRunTerminal()
468 defer term.in.unblock()
469
470 firstLifecycle := domain.SessionLifecycleID{1}
471 secondLifecycle := domain.SessionLifecycleID{2}
472 welcome := func(name string, lifecycle domain.SessionLifecycleID) ports.Frame {
473 return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
474 SessionID: name, SessionName: name, ResumeToken: 1, Capabilities: ports.CapabilityResume,
475 CommittedIdentity: &ports.CommittedRouteIdentity{
476 Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: name},
477 },
478 }))
479 }
480 secondTarget := ports.ExactSessionTarget{LifecycleID: secondLifecycle, SessionName: "second"}
481 first := &recordingTransport{recvs: []recvItem{
482 {f: welcome("first", firstLifecycle)},
483 {f: frameOf(ports.MsgAttachTarget, ports.MarshalAttachTarget(ports.AttachTarget{
484 Session: "second", Intent: ports.IntentAttach, ExactTarget: &secondTarget,
485 EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
486 }))},
487 }}
488 second := &recordingTransport{recvs: []recvItem{
489 {f: welcome("second", secondLifecycle)},
490 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
491 }}
492 dialer := &sequenceDialer{trs: []ports.Transport{first, second}}
493
494 err := runTestClient(context.Background(), testDependencies(dialer, term, realClock{}, nil, nil), client.AttachRequest{
495 Intent: ports.IntentAttach, SessionName: "first",
496 Origin: ports.RouteOriginLocal, OriginKey: "local",
497 EnvironmentPolicy: ports.EnvironmentPolicyClientOwned,
498 })
499
500 require.NoError(t, err)
501 hello := helloFromSend(t, second)
502 require.Equal(t, "second", hello.Name)
503 require.Equal(t, ports.EnvironmentPolicyClientOwned, hello.EnvironmentPolicy)
504 require.Nil(t, hello.RemoteTarget)
505}
506
507func TestAttachHelloPreservesCompleteAttachRequest(t *testing.T) {
508 term := newRunTerminal()
509 defer term.in.unblock()
510 lifecycle := domain.SessionLifecycleID{1, 2, 3}
511 target := &domain.RemoteSessionTarget{
512 Endpoint: "remote.example", DisplayOrigin: "remote.example", LifecycleID: lifecycle,
513 SessionName: "work", LiveTabID: "tab-1",
514 }
515 request := client.AttachRequest{
516 Intent: ports.IntentAttach, SessionName: "work", Remote: true,
517 RemoteTarget: target, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
518 NavigationCapabilities: ports.NavigationCapabilityHomePicker,
519 StartupOverlay: ports.StartupOverlayNone,
520 }
521 tr := &recordingTransport{recvs: []recvItem{
522 {f: welcomeFrame(0)},
523 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
524 }}
525
526 require.NoError(t, runTestClient(context.Background(), attachTestDependencies(tr, term, realClock{}), request))
527 hello := helloFromSend(t, tr)
528 require.Equal(t, request.Intent, hello.Intent)
529 require.Equal(t, request.SessionName, hello.Name)
530 require.Equal(t, target, hello.RemoteTarget)
531 require.Equal(t, request.EnvironmentPolicy, hello.EnvironmentPolicy)
532 require.Equal(t, request.NavigationCapabilities, hello.NavigationCapabilities)
533 require.Equal(t, request.StartupOverlay, hello.StartupOverlay)
534}
535
536func TestAcceptedHomeActionUsesCapturedRouteAfterRequestMetadataRebase(t *testing.T) {
537 term := newRunTerminal()
538 defer term.in.unblock()
539 localLifecycle := domain.SessionLifecycleID{1}
540 remoteLifecycle := domain.SessionLifecycleID{2}
541 welcome := func(name string, lifecycle domain.SessionLifecycleID) ports.Frame {
542 return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
543 SessionID: name, SessionName: name, ResumeToken: 1, Capabilities: ports.CapabilityResume,
544 CommittedIdentity: &ports.CommittedRouteIdentity{Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: name}},
545 }))
546 }
547 remoteTarget := domain.RemoteSessionTarget{
548 Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: remoteLifecycle,
549 SessionName: "work", LiveTabID: "tab-1",
550 }
551 local1 := &recordingTransport{recvs: []recvItem{
552 {f: welcome("local", localLifecycle)},
553 {f: frameOf(ports.MsgAttachTarget, ports.MarshalAttachTarget(ports.AttachTarget{
554 Endpoint: "remote", Session: "work", Intent: ports.IntentAttach,
555 RemoteTarget: &remoteTarget, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
556 }))},
557 }}
558 remote := &recordingTransport{recvs: []recvItem{
559 {f: welcome("work", remoteLifecycle)},
560 {f: frameOf(ports.MsgNavigationAction, ports.MarshalNavigationAction(ports.NavigationOpenHomePicker))},
561 }}
562 local2 := &recordingTransport{recvs: []recvItem{
563 {f: welcome("local", localLifecycle)},
564 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
565 }}
566 localDialer := &sequenceDialer{trs: []ports.Transport{local1, local2}}
567 remoteDialer := &sequenceDialer{trs: []ports.Transport{remote}}
568 deps := testDependencies(localDialer, term, realClock{}, nil, nil)
569 deps.AttachHandoff = func(ports.AttachTarget) (ports.Dialer, client.AttachRequest, error) {
570 target := ports.ExactSessionTarget{LifecycleID: remoteLifecycle, SessionName: "work"}
571 return remoteDialer, client.AttachRequest{
572 Intent: ports.IntentAttach, SessionName: "work", Remote: true,
573 Origin: ports.RouteOriginRemote, OriginKey: "remote", ExactTarget: &target,
574 EnvironmentPolicy: ports.EnvironmentPolicyClientOwned,
575 // Deliberately omit the copied Home capability. The accepted server
576 // action remains valid because Runner retained the concrete home route.
577 }, nil
578 }
579
580 require.NoError(t, runTestClient(context.Background(), deps, client.AttachRequest{
581 Intent: ports.IntentAttach, SessionName: "local", Origin: ports.RouteOriginLocal, OriginKey: "local",
582 }))
583 require.Equal(t, ports.StartupOverlaySessionPicker, helloFromSend(t, local2).StartupOverlay)
584}
585
586func TestRouteNavigationPreservesRemoteHomePickerAcrossLocalReturn(t *testing.T) {
587 term := newRunTerminal()
588 defer term.in.unblock()
589
590 localLifecycle := domain.SessionLifecycleID{1, 1, 1}
591 remoteLifecycle := domain.SessionLifecycleID{2, 2, 2}
592 newLifecycle := domain.SessionLifecycleID{3, 3, 3}
593 remoteNewLifecycle := domain.SessionLifecycleID{4, 4, 4}
594 welcome := func(name string, lifecycle domain.SessionLifecycleID, token uint64) ports.Frame {
595 return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
596 SessionID: name + "-id",
597 SessionName: name,
598 ResumeToken: token,
599 Capabilities: ports.CapabilityResume,
600 CommittedIdentity: &ports.CommittedRouteIdentity{
601 Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: name},
602 },
603 }))
604 }
605 remoteTarget := domain.RemoteSessionTarget{
606 Endpoint: "remote",
607 DisplayOrigin: "remote",
608 LifecycleID: remoteLifecycle,
609 SessionName: "remote-manual",
610 LiveTabID: "tab-1",
611 }
612 remoteHandoff := ports.AttachTarget{
613 Endpoint: "remote", Session: "remote-manual", Intent: ports.IntentAttach,
614 RemoteTarget: &remoteTarget, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
615 }
616
617 local1 := &recordingTransport{recvs: []recvItem{
618 {f: welcome("test", localLifecycle, 11)},
619 {f: frameOf(ports.MsgAttachTarget, ports.MarshalAttachTarget(remoteHandoff))},
620 }}
621 local2 := &recordingTransport{recvs: []recvItem{
622 {f: welcome("test", localLifecycle, 11)},
623 {f: frameOf(ports.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(ports.CommittedRouteIdentity{
624 Target: ports.ExactSessionTarget{LifecycleID: newLifecycle, SessionName: "new"},
625 }))},
626 {f: frameOf(ports.MsgNavigateRecentRoute, mustMarshalRouteAction(ports.RouteNavigationAction{SnapshotGeneration: 4, Key: 2, Generation: 2}))},
627 }}
628 local3 := &recordingTransport{recvs: []recvItem{
629 {f: welcome("test", localLifecycle, 11)},
630 {f: frameOf(ports.MsgNavigationAction, ports.MarshalNavigationAction(ports.NavigationBack))},
631 }}
632 remote1 := &recordingTransport{recvs: []recvItem{
633 {f: welcome("remote-manual", remoteLifecycle, 22)},
634 {f: frameOf(ports.MsgNavigateRecentRoute, mustMarshalRouteAction(ports.RouteNavigationAction{SnapshotGeneration: 2, Key: 1, Generation: 1}))},
635 }}
636 remote2 := &recordingTransport{recvs: []recvItem{
637 {f: welcome("remote-manual", remoteLifecycle, 22)},
638 {f: frameOf(ports.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(ports.CommittedRouteIdentity{
639 Target: ports.ExactSessionTarget{LifecycleID: remoteNewLifecycle, SessionName: "remote-new"},
640 }))},
641 {f: frameOf(ports.MsgRoutePosition, mustMarshalRoutePosition(ports.RoutePosition{
642 Target: ports.ExactSessionTarget{LifecycleID: remoteNewLifecycle, SessionName: "remote-new"}, ActiveTabID: "tab-2",
643 }))},
644 {f: frameOf(ports.MsgNavigationAction, ports.MarshalNavigationAction(ports.NavigationOpenHomePicker))},
645 }}
646 remote3 := &recordingTransport{recvs: []recvItem{
647 {f: welcome("remote-new", remoteNewLifecycle, 22)},
648 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
649 }}
650 localDialer := &sequenceDialer{trs: []ports.Transport{local1, local2, local3}}
651 remoteDialer := &sequenceDialer{trs: []ports.Transport{remote1, remote2, remote3}}
652
653 deps := testDependencies(localDialer, term, realClock{}, nil, nil)
654 deps.AttachHandoff = func(target ports.AttachTarget) (ports.Dialer, client.AttachRequest, error) {
655 require.Equal(t, remoteHandoff, target)
656 return remoteDialer, client.AttachRequest{
657 Intent: ports.IntentAttach, SessionName: target.Session, Remote: true,
658 Origin: ports.RouteOriginDiscovery, OriginKey: target.Endpoint,
659 RemoteTarget: &remoteTarget, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
660 }, nil
661 }
662
663 err := runTestClient(context.Background(), deps, client.AttachRequest{
664 Intent: ports.IntentAttach, SessionName: "test", Origin: ports.RouteOriginLocal, OriginKey: "local",
665 })
666 remoteHello := helloFromSend(t, remote2)
667 require.NoError(t, err)
668
669 require.Equal(t, ports.NavigationCapabilityHomePicker, remoteHello.NavigationCapabilities)
670 remoteNewHello := helloFromSend(t, remote3)
671 require.Equal(t, "remote-new", remoteNewHello.Name)
672 require.Equal(t, &ports.ExactSessionTarget{LifecycleID: remoteNewLifecycle, SessionName: "remote-new"}, remoteNewHello.ExactTarget)
673}
674
675func mustMarshalCommittedIdentity(identity ports.CommittedRouteIdentity) []byte {
676 payload, err := ports.MarshalCommittedRouteIdentity(identity)
677 if err != nil {
678 panic(err)
679 }
680 return payload
681}
682
683func mustMarshalRouteAction(action ports.RouteNavigationAction) []byte {
684 payload, err := ports.MarshalRouteNavigationAction(action)
685 if err != nil {
686 panic(err)
687 }
688 return payload
689}
690
691func mustMarshalRoutePosition(position ports.RoutePosition) []byte {
692 payload, err := ports.MarshalRoutePosition(position)
693 if err != nil {
694 panic(err)
695 }
696 return payload
697}
698
699func TestAttachTargetHandoffReturnsValidatedTargetAndClosesTransport(t *testing.T) {
700 var out bytes.Buffer
701 var restoreCount atomic.Int32
702 resizeCh := make(chan domain.Size)
703 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
704 defer in.unblock()
705
706 target := ports.AttachTarget{Endpoint: "remote.example", Session: "work", Intent: ports.IntentAttach}
707 tr := portsmocks.NewMockTransport(t)
708 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
709 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
710 unblock := scriptRecv(tr,
711 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "local"}))},
712 recvItem{f: frameOf(ports.MsgAttachTarget, ports.MarshalAttachTarget(target))},
713 )
714 defer unblock()
715 tr.EXPECT().Close().Return(nil).Once()
716
717 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "work"})
718 var handoff *client.AttachTargetError
719 require.ErrorAs(t, err, &handoff)
720 require.Equal(t, target, handoff.Target)
721 require.Equal(t, int32(1), restoreCount.Load(), "handoff must restore the terminal after closing the old connection")
722}
723
724func TestAttachHelloIncludesCompleteLocalEnvironment(t *testing.T) {
725 t.Setenv("VEV_CLIENT_ENV_TEST", "TOKEN=a=b=c")
726
727 var out bytes.Buffer
728 var restoreCount atomic.Int32
729 resizeCh := make(chan domain.Size)
730 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
731 defer in.unblock()
732
733 gotHello := make(chan ports.Hello, 1)
734 tr := portsmocks.NewMockTransport(t)
735 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
736 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
737 tr.EXPECT().Send(isType(ports.MsgHello)).RunAndReturn(func(f ports.Frame) error {
738 hello, err := ports.UnmarshalHello(f.Payload)
739 if err != nil {
740 return err
741 }
742 gotHello <- hello
743 return nil
744 }).Once()
745 unblock := scriptRecv(tr,
746 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
747 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
748 )
749 defer unblock()
750 tr.EXPECT().Close().Return(nil).Once()
751
752 require.NoError(t, runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral}))
753 require.Equal(t, os.Environ(), (<-gotHello).Env)
754}
755
756func TestAttachHelloRequestsSingleOutputForDatagramTransport(t *testing.T) {
757 var out bytes.Buffer
758 var restoreCount atomic.Int32
759 resizeCh := make(chan domain.Size)
760 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
761 defer in.unblock()
762
763 tr := portsmocks.NewMockTransport(t)
764 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
765 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
766 tr.EXPECT().Send(isType(ports.MsgHello)).RunAndReturn(func(f ports.Frame) error {
767 hello, err := ports.UnmarshalHello(f.Payload)
768 require.NoError(t, err)
769 require.Equal(t, uint8(1), hello.MaxOutputInFlight)
770 return nil
771 }).Once()
772 unblock := scriptRecv(tr,
773 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
774 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
775 )
776 defer unblock()
777 tr.EXPECT().Close().Return(nil).Once()
778
779 require.NoError(t, runTestClient(context.Background(), attachTestDependencies(markedDatagramTransport{tr}, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral}))
780}
781
782func TestAttachHappyPath(t *testing.T) {
783 var out bytes.Buffer
784 var restoreCount atomic.Int32
785 resizeCh := make(chan domain.Size)
786 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
787 defer in.unblock()
788
789 tr := portsmocks.NewMockTransport(t)
790 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
791 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
792 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
793 unblock := scriptRecv(tr,
794 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1", SessionName: "main"}))},
795 recvItem{f: frameOf(ports.MsgOutput, mustMarshalOutput(ports.Output{Epoch: 1, Size: domain.Size{Cols: 1, Rows: 1}, Data: []byte("hello world")}))},
796 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
797 )
798 defer unblock()
799 tr.EXPECT().Close().Return(nil).Once()
800
801 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: ""})
802 require.NoError(t, err)
803 require.Equal(t, "\x1b]10;?\x07\x1b]11;?\x07\x1b]4;0;?;1;?;2;?;3;?;4;?;5;?;6;?;7;?;8;?;9;?;10;?;11;?;12;?;13;?;14;?;15;?\x07\x1b[?2031$phello world", out.String())
804 require.Equal(t, int32(1), restoreCount.Load(), "restore must run exactly once")
805}
806
807func TestAttachVersionMismatch(t *testing.T) {
808 tm := portsmocks.NewMockTerminal(t)
809 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
810 // EnterRaw must NOT be called on the error-before-welcome path.
811
812 tr := portsmocks.NewMockTransport(t)
813 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
814 tr.EXPECT().Recv().Return(
815 frameOf(ports.MsgError, ports.MarshalErrorMsg(ports.ErrorMsg{Code: ports.ErrVersionMismatch, Text: "version mismatch"})),
816 nil,
817 ).Once()
818 tr.EXPECT().Close().Return(nil).Once()
819
820 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main"})
821 require.Error(t, err)
822 var pe *client.ProtocolError
823 require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
824 require.Equal(t, ports.ErrVersionMismatch, pe.Code)
825 tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
826}
827
828func TestAttachRestoredOnRecvErrorMidStream(t *testing.T) {
829 var out bytes.Buffer
830 var restoreCount atomic.Int32
831 resizeCh := make(chan domain.Size)
832 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
833 defer in.unblock()
834
835 tr := portsmocks.NewMockTransport(t)
836 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
837 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
838 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
839 boom := errors.New("connection reset")
840 unblock := scriptRecv(tr,
841 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
842 recvItem{err: boom},
843 )
844 defer unblock()
845 tr.EXPECT().Close().Return(nil).Once()
846
847 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: ""})
848 require.Error(t, err)
849 require.Equal(t, int32(1), restoreCount.Load(), "restore must run after mid-stream Recv error")
850}
851
852func TestAttachDaemonVanishedOnEOF(t *testing.T) {
853 var out bytes.Buffer
854 var restoreCount atomic.Int32
855 resizeCh := make(chan domain.Size)
856 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
857 defer in.unblock()
858
859 tr := portsmocks.NewMockTransport(t)
860 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
861 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
862 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
863 unblock := scriptRecv(tr,
864 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
865 recvItem{err: io.EOF},
866 )
867 defer unblock()
868 tr.EXPECT().Close().Return(nil).Once()
869
870 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: ""})
871 require.Error(t, err)
872 require.ErrorIs(t, err, io.EOF)
873 require.Equal(t, int32(1), restoreCount.Load())
874}
875
876func TestAttachStdinForwardsSGRMouseReportAsSingleFrame(t *testing.T) {
877 var out bytes.Buffer
878 var restoreCount atomic.Int32
879 resizeCh := make(chan domain.Size)
880 input := newOneShotBlockingReader([]byte("\x1b[<0;1;1M"))
881 defer input.unblock()
882
883 tm := portsmocks.NewMockTerminal(t)
884 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
885 tm.EXPECT().EnterRaw().Return(func() error { restoreCount.Add(1); return nil }, nil).Once()
886 tm.EXPECT().In().Return(input).Maybe()
887 tm.EXPECT().Out().Return(&out).Maybe()
888 tm.EXPECT().Flush().Return(nil).Maybe()
889 tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
890
891 gotInput := make(chan []byte, 2)
892 allowDetach := make(chan struct{})
893 tr := portsmocks.NewMockTransport(t)
894 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
895 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
896 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
897 tr.EXPECT().Send(isType(ports.MsgInput)).RunAndReturn(func(f ports.Frame) error {
898 in, err := ports.UnmarshalInput(f.Payload)
899 require.NoError(t, err)
900 gotInput <- in.Data
901 close(allowDetach)
902 return nil
903 }).Once()
904 welcome := frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))
905 detached := frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))
906 recvCh := make(chan recvItem, 1)
907 recvCh <- recvItem{f: welcome}
908 closed := make(chan struct{})
909 tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
910 select {
911 case it := <-recvCh:
912 return it.f, it.err
913 case <-allowDetach:
914 select {
915 case <-closed:
916 return ports.Frame{}, io.EOF
917 default:
918 close(closed)
919 return detached, nil
920 }
921 case <-closed:
922 return ports.Frame{}, io.EOF
923 }
924 }).Maybe()
925 tr.EXPECT().Close().Return(nil).Once()
926
927 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: ""})
928 require.NoError(t, err)
929 select {
930 case got := <-gotInput:
931 require.Equal(t, []byte("\x1b[<0;1;1M"), got, "SGR mouse report must arrive as one intact MsgInput frame")
932 case <-time.After(2 * time.Second):
933 t.Fatal("mouse report input was not sent")
934 }
935}
936
937func TestAttachStdinCoalescesSplitBracketedPaste(t *testing.T) {
938 var out bytes.Buffer
939 var restoreCount atomic.Int32
940 resizeCh := make(chan domain.Size)
941 paste := []byte("\x1b[200~hello\nworld\x1b[201~")
942 input := newChunkedBlockingReader([]byte("\x1b[200~hello\n"), []byte("world\x1b[201~"))
943 defer input.unblock()
944
945 tm := portsmocks.NewMockTerminal(t)
946 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
947 tm.EXPECT().EnterRaw().Return(func() error { restoreCount.Add(1); return nil }, nil).Once()
948 tm.EXPECT().In().Return(input).Maybe()
949 tm.EXPECT().Out().Return(&out).Maybe()
950 tm.EXPECT().Flush().Return(nil).Maybe()
951 tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
952
953 gotInput := make(chan []byte, 1)
954 allowDetach := make(chan struct{})
955 tr := portsmocks.NewMockTransport(t)
956 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
957 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
958 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
959 tr.EXPECT().Send(isType(ports.MsgInput)).RunAndReturn(func(f ports.Frame) error {
960 in, err := ports.UnmarshalInput(f.Payload)
961 require.NoError(t, err)
962 gotInput <- in.Data
963 close(allowDetach)
964 return nil
965 }).Once()
966 welcome := frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))
967 detached := frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))
968 recvCh := make(chan recvItem, 1)
969 recvCh <- recvItem{f: welcome}
970 closed := make(chan struct{})
971 tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
972 select {
973 case it := <-recvCh:
974 return it.f, it.err
975 case <-allowDetach:
976 select {
977 case <-closed:
978 return ports.Frame{}, io.EOF
979 default:
980 close(closed)
981 return detached, nil
982 }
983 case <-closed:
984 return ports.Frame{}, io.EOF
985 }
986 }).Maybe()
987 tr.EXPECT().Close().Return(nil).Once()
988
989 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: ""})
990 require.NoError(t, err)
991 select {
992 case got := <-gotInput:
993 require.Equal(t, paste, got, "split bracketed paste must arrive as one intact MsgInput frame")
994 case <-time.After(2 * time.Second):
995 t.Fatal("split bracketed paste input was not sent")
996 }
997}
998
999func TestAttachForwardsResize(t *testing.T) {
1000 var out bytes.Buffer
1001 var restoreCount atomic.Int32
1002 resizeCh := make(chan domain.Size)
1003 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
1004 defer in.unblock()
1005
1006 detachAfterResize := make(chan struct{})
1007 firstRecv := make(chan struct{})
1008 var firstRecvOnce sync.Once
1009
1010 tr := portsmocks.NewMockTransport(t)
1011 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1012 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1013 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
1014 // The resize frame is forwarded via the sender goroutine.
1015 gotResize := make(chan ports.Resize, 1)
1016 tr.EXPECT().Send(isType(ports.MsgResize)).RunAndReturn(func(f ports.Frame) error {
1017 r, _ := ports.UnmarshalResize(f.Payload)
1018 gotResize <- r
1019 close(detachAfterResize)
1020 return nil
1021 }).Once()
1022
1023 welcome := frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))
1024 detached := frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))
1025 recvCh := make(chan recvItem, 2)
1026 recvCh <- recvItem{f: welcome}
1027 done := make(chan struct{})
1028 tr.EXPECT().Recv().RunAndReturn(func() (ports.Frame, error) {
1029 firstRecvOnce.Do(func() { close(firstRecv) })
1030 select {
1031 case it := <-recvCh:
1032 return it.f, it.err
1033 case <-detachAfterResize:
1034 // Deliver the detach once the resize has been observed.
1035 select {
1036 case <-done:
1037 return ports.Frame{}, io.EOF
1038 default:
1039 close(done)
1040 return detached, nil
1041 }
1042 case <-done:
1043 return ports.Frame{}, io.EOF
1044 }
1045 }).Maybe()
1046 tr.EXPECT().Close().Return(nil).Once()
1047
1048 // Push a resize event once attach begins receiving daemon frames.
1049 go func() {
1050 <-firstRecv
1051 resizeCh <- domain.Size{Cols: 120, Rows: 40}
1052 }()
1053
1054 err := runTestClient(context.Background(), attachTestDependencies(tr, tm, realClock{}), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: ""})
1055 require.NoError(t, err)
1056 select {
1057 case r := <-gotResize:
1058 require.Equal(t, domain.Size{Cols: 120, Rows: 40}, r.Size)
1059 default:
1060 t.Fatal("resize was not forwarded")
1061 }
1062}
1063
1064func TestRunRestoresRawModeAfterAttachError(t *testing.T) {
1065 var out bytes.Buffer
1066 var restoreCount atomic.Int32
1067 resizeCh := make(chan domain.Size)
1068 tm, in := newHappyTerminal(t, &out, &restoreCount, resizeCh)
1069 defer in.unblock()
1070
1071 tr := portsmocks.NewMockTransport(t)
1072 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1073 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1074 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
1075 boom := errors.New("connection reset")
1076 unblock := scriptRecv(tr,
1077 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1078 recvItem{err: boom},
1079 )
1080 defer unblock()
1081 tr.EXPECT().Close().Return(nil).Once()
1082 d := portsmocks.NewMockDialer(t)
1083 d.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
1084
1085 err := runTestClient(context.Background(), testDependencies(d, tm, realClock{}, nil, nil), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: "", Remote: false})
1086 require.Error(t, err)
1087 require.Equal(t, int32(1), restoreCount.Load(), "Run must restore raw mode after attach attempt errors")
1088}
1089
1090func TestRunDoesNotEnterRawBeforePreWelcomeError(t *testing.T) {
1091 tm := portsmocks.NewMockTerminal(t)
1092 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
1093 // EnterRaw must NOT be called when the daemon rejects Hello before Welcome.
1094
1095 tr := portsmocks.NewMockTransport(t)
1096 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
1097 tr.EXPECT().Recv().Return(
1098 frameOf(ports.MsgError, ports.MarshalErrorMsg(ports.ErrorMsg{Code: ports.ErrVersionMismatch, Text: "version mismatch"})),
1099 nil,
1100 ).Once()
1101 tr.EXPECT().Close().Return(nil).Once()
1102 d := portsmocks.NewMockDialer(t)
1103 d.EXPECT().Dial(mock.Anything).Return(tr, nil).Once()
1104
1105 err := runTestClient(context.Background(), testDependencies(d, tm, realClock{}, nil, nil), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: false})
1106 require.Error(t, err)
1107 var pe *client.ProtocolError
1108 require.True(t, errors.As(err, &pe), "want *client.ProtocolError, got %T", err)
1109 tr.AssertNotCalled(t, "Send", isType(ports.MsgTheme))
1110}
1111
1112func TestRunPhaseASingleAttempt(t *testing.T) {
1113 dialErr := errors.New("dial failed")
1114 d := portsmocks.NewMockDialer(t)
1115 d.EXPECT().Dial(mock.Anything).Return(nil, dialErr).Once()
1116 tm := portsmocks.NewMockTerminal(t)
1117
1118 err := runTestClient(context.Background(), testDependencies(d, tm, realClock{}, nil, nil), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: "", Remote: false})
1119 require.ErrorIs(t, err, dialErr)
1120}
1121
1122type sequenceDialer struct {
1123 trs []ports.Transport
1124 errs []error
1125 calls atomic.Int32
1126}
1127
1128func (d *sequenceDialer) Dial(context.Context) (ports.Transport, error) {
1129 i := int(d.calls.Add(1)) - 1
1130 if i < len(d.errs) && d.errs[i] != nil {
1131 return nil, d.errs[i]
1132 }
1133 if i >= len(d.trs) {
1134 return nil, io.EOF
1135 }
1136 return d.trs[i], nil
1137}
1138
1139type recordingTransport struct {
1140 mu sync.Mutex
1141 recvs []recvItem
1142 sends []ports.Frame
1143 closed atomic.Int32
1144}
1145
1146func (t *recordingTransport) Send(f ports.Frame) error {
1147 t.mu.Lock()
1148 t.sends = append(t.sends, f)
1149 t.mu.Unlock()
1150 return nil
1151}
1152
1153func (t *recordingTransport) Sends() []ports.Frame {
1154 t.mu.Lock()
1155 defer t.mu.Unlock()
1156 return append([]ports.Frame(nil), t.sends...)
1157}
1158
1159func (t *recordingTransport) Recv() (ports.Frame, error) {
1160 if len(t.recvs) == 0 {
1161 return ports.Frame{}, io.EOF
1162 }
1163 it := t.recvs[0]
1164 t.recvs = t.recvs[1:]
1165 return it.f, it.err
1166}
1167
1168func (t *recordingTransport) Close() error {
1169 t.closed.Add(1)
1170 return nil
1171}
1172
1173type runTerminal struct {
1174 in *blockingReader
1175 out bytes.Buffer
1176 rawCount atomic.Int32
1177 restoreCount atomic.Int32
1178 resizeCh chan domain.Size
1179}
1180
1181func newRunTerminal() *runTerminal {
1182 return &runTerminal{in: newBlockingReader(), resizeCh: make(chan domain.Size)}
1183}
1184
1185func (t *runTerminal) EnterRaw() (func() error, error) {
1186 t.rawCount.Add(1)
1187 return func() error { t.restoreCount.Add(1); return nil }, nil
1188}
1189func (t *runTerminal) Size() (domain.Size, error) { return domain.Size{Cols: 80, Rows: 24}, nil }
1190func (t *runTerminal) ResizeEvents() <-chan domain.Size { return t.resizeCh }
1191func (t *runTerminal) In() io.Reader { return t.in }
1192func (t *runTerminal) Out() io.Writer { return &t.out }
1193func (t *runTerminal) Flush() error { return nil }
1194
1195func helloFromSend(t *testing.T, tr *recordingTransport) ports.Hello {
1196 t.Helper()
1197 sends := tr.Sends()
1198 require.NotEmpty(t, sends)
1199 require.Equal(t, ports.MsgHello, sends[0].Type)
1200 h, err := ports.UnmarshalHello(sends[0].Payload)
1201 require.NoError(t, err)
1202 return h
1203}
1204
1205func welcomeFrame(token uint64) ports.Frame {
1206 return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1", SessionName: "main", ResumeToken: token, Capabilities: ports.CapabilityResume}))
1207}
1208
1209func TestRunReconnectsWithRotatedTokenAndSameClientID(t *testing.T) {
1210 term := newRunTerminal()
1211 defer term.in.unblock()
1212 tr1 := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {err: io.EOF}}}
1213 tr2 := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(22)}, {err: io.EOF}}}
1214 tr3 := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(33)}, {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))}}}
1215 d := &sequenceDialer{trs: []ports.Transport{tr1, tr2, tr3}}
1216 clock := &reconnectTestClock{}
1217 result := make(chan error, 1)
1218 go func() {
1219 result <- runTestClient(context.Background(), testDependencies(d, term, clock, nil, nil), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: false})
1220 }()
1221 clock.fireReconnect(t)
1222 clock.fireReconnect(t)
1223 require.NoError(t, <-result)
1224 require.Equal(t, int32(3), d.calls.Load())
1225 require.Equal(t, int32(1), term.rawCount.Load())
1226 require.Equal(t, int32(1), term.restoreCount.Load())
1227
1228 h1 := helloFromSend(t, tr1)
1229 h2 := helloFromSend(t, tr2)
1230 h3 := helloFromSend(t, tr3)
1231 require.Equal(t, ports.IntentAttach, h1.Intent)
1232 require.Zero(t, h1.ResumeToken)
1233 require.Equal(t, ports.IntentResume, h2.Intent)
1234 require.Equal(t, uint64(11), h2.ResumeToken)
1235 require.Equal(t, ports.IntentResume, h3.Intent)
1236 require.Equal(t, uint64(22), h3.ResumeToken)
1237 require.Equal(t, h1.ClientID, h2.ClientID)
1238 require.Equal(t, h1.ClientID, h3.ClientID)
1239 require.Contains(t, term.out.String(), "reconnecting…")
1240 require.Contains(t, term.out.String(), "\x1b[2K")
1241}
1242
1243func TestReconnectRestoresLastPublishedRouteTab(t *testing.T) {
1244 term := newRunTerminal()
1245 defer term.in.unblock()
1246 lifecycle := domain.SessionLifecycleID{7, 8, 9}
1247 target := ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: "work"}
1248 welcome := func(token uint64) ports.Frame {
1249 return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
1250 SessionID: "work", SessionName: "work", ResumeToken: token, Capabilities: ports.CapabilityResume,
1251 CommittedIdentity: &ports.CommittedRouteIdentity{Target: target},
1252 }))
1253 }
1254 tr1 := &recordingTransport{recvs: []recvItem{
1255 {f: welcome(11)},
1256 {f: frameOf(ports.MsgRoutePosition, mustMarshalRoutePosition(ports.RoutePosition{Target: target, ActiveTabID: "tab-2"}))},
1257 {err: io.EOF},
1258 }}
1259 tr2 := &recordingTransport{recvs: []recvItem{
1260 {f: welcome(22)},
1261 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1262 }}
1263 dialer := &sequenceDialer{trs: []ports.Transport{tr1, tr2}}
1264 clock := &reconnectTestClock{}
1265 result := make(chan error, 1)
1266 go func() {
1267 result <- runTestClient(context.Background(), testDependencies(dialer, term, clock, nil, nil), client.AttachRequest{
1268 Intent: ports.IntentAttach, SessionName: "work", Origin: ports.RouteOriginLocal, OriginKey: "local",
1269 })
1270 }()
1271 clock.fireReconnect(t)
1272 require.NoError(t, <-result)
1273
1274 reconnectHello := helloFromSend(t, tr2)
1275 require.Equal(t, domain.TabStableID("tab-2"), reconnectHello.PreferredTabID)
1276}
1277
1278func TestRemoteReconnectDropsStalePickerTargetAfterDaemonSessionSwitch(t *testing.T) {
1279 term := newRunTerminal()
1280 defer term.in.unblock()
1281 oldLifecycle := domain.SessionLifecycleID{1, 2, 3}
1282 newLifecycle := domain.SessionLifecycleID{4, 5, 6}
1283 remoteTarget := &domain.RemoteSessionTarget{
1284 Endpoint: "remote", DisplayOrigin: "remote", LifecycleID: oldLifecycle,
1285 SessionName: "old", LiveTabID: "tab-1",
1286 }
1287 welcome := func(name string, lifecycle domain.SessionLifecycleID, token uint64) ports.Frame {
1288 return frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{
1289 SessionID: name, SessionName: name, ResumeToken: token, Capabilities: ports.CapabilityResume,
1290 CommittedIdentity: &ports.CommittedRouteIdentity{
1291 Target: ports.ExactSessionTarget{LifecycleID: lifecycle, SessionName: name},
1292 },
1293 }))
1294 }
1295 tr1 := &recordingTransport{recvs: []recvItem{
1296 {f: welcome("old", oldLifecycle, 11)},
1297 {f: frameOf(ports.MsgCommittedRouteIdentity, mustMarshalCommittedIdentity(ports.CommittedRouteIdentity{
1298 Target: ports.ExactSessionTarget{LifecycleID: newLifecycle, SessionName: "new"},
1299 }))},
1300 {err: io.EOF},
1301 }}
1302 tr2 := &recordingTransport{recvs: []recvItem{
1303 {f: welcome("new", newLifecycle, 22)},
1304 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1305 }}
1306 dialer := &sequenceDialer{trs: []ports.Transport{tr1, tr2}}
1307 clock := &reconnectTestClock{}
1308 result := make(chan error, 1)
1309 go func() {
1310 result <- runTestClient(context.Background(), testDependencies(dialer, term, clock, nil, nil), client.AttachRequest{
1311 Intent: ports.IntentAttach, SessionName: "old", Remote: true,
1312 Origin: ports.RouteOriginDiscovery, OriginKey: "remote",
1313 RemoteTarget: remoteTarget, EnvironmentPolicy: ports.EnvironmentPolicyDaemonOwned,
1314 })
1315 }()
1316 clock.fireReconnect(t)
1317 require.NoError(t, <-result)
1318
1319 reconnectHello := helloFromSend(t, tr2)
1320 require.Equal(t, ports.IntentResume, reconnectHello.Intent)
1321 require.Equal(t, "new", reconnectHello.Name)
1322 require.Equal(t, &ports.ExactSessionTarget{LifecycleID: newLifecycle, SessionName: "new"}, reconnectHello.ExactTarget)
1323 require.Nil(t, reconnectHello.RemoteTarget)
1324}
1325
1326func TestAttachRememberRemoteHost(t *testing.T) {
1327 tests := []struct {
1328 name string
1329 request client.AttachRequest
1330 remember func() error
1331 recv []recvItem
1332 preWelcomeReject bool
1333 wantCalled bool
1334 wantErr bool
1335 }{
1336 {
1337 name: "remote success",
1338 request: client.AttachRequest{
1339 Intent: ports.IntentAttach,
1340 SessionName: "main",
1341 Remote: true,
1342 },
1343 remember: func() error { return nil },
1344 recv: []recvItem{
1345 {f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1346 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1347 },
1348 wantCalled: true,
1349 },
1350 {
1351 name: "callback failure does not fail attach",
1352 request: client.AttachRequest{
1353 Intent: ports.IntentAttach,
1354 SessionName: "main",
1355 Remote: true,
1356 },
1357 remember: func() error { return errors.New("disk full") },
1358 recv: []recvItem{
1359 {f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1360 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1361 },
1362 wantCalled: true,
1363 },
1364 {
1365 name: "local attach does not call callback",
1366 request: client.AttachRequest{
1367 Intent: ports.IntentAttach,
1368 SessionName: "main",
1369 Remote: false,
1370 },
1371 remember: func() error {
1372 t.Fatal("remember callback must not run for local attach")
1373 return nil
1374 },
1375 recv: []recvItem{
1376 {f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1377 {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1378 },
1379 },
1380 {
1381 name: "rejected before welcome does not call callback",
1382 request: client.AttachRequest{
1383 Intent: ports.IntentAttach,
1384 SessionName: "main",
1385 Remote: true,
1386 },
1387 remember: func() error {
1388 t.Fatal("remember callback must not run before welcome")
1389 return nil
1390 },
1391 recv: []recvItem{
1392 {f: frameOf(ports.MsgError, ports.MarshalErrorMsg(ports.ErrorMsg{Code: ports.ErrVersionMismatch, Text: "version mismatch"}))},
1393 },
1394 preWelcomeReject: true,
1395 wantErr: true,
1396 },
1397 {
1398 name: "post-welcome error still uses happy terminal fixture",
1399 request: client.AttachRequest{
1400 Intent: ports.IntentAttach,
1401 SessionName: "main",
1402 Remote: true,
1403 },
1404 remember: func() error { return nil },
1405 recv: []recvItem{
1406 {f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1407 {err: errors.New("connection reset")},
1408 },
1409 wantCalled: true,
1410 wantErr: true,
1411 },
1412 }
1413
1414 for _, tt := range tests {
1415 t.Run(tt.name, func(t *testing.T) {
1416 var tm *portsmocks.MockTerminal
1417 var unblockIn func()
1418 if tt.preWelcomeReject {
1419 tm = portsmocks.NewMockTerminal(t)
1420 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
1421 // EnterRaw must NOT be called when the daemon rejects Hello before Welcome.
1422 } else {
1423 var out bytes.Buffer
1424 var restoreCount atomic.Int32
1425 resizeCh := make(chan domain.Size)
1426 var in *blockingReader
1427 tm, in = newHappyTerminal(t, &out, &restoreCount, resizeCh)
1428 unblockIn = in.unblock
1429 }
1430 if unblockIn != nil {
1431 defer unblockIn()
1432 }
1433
1434 var called atomic.Bool
1435 learnerDone := make(chan struct{})
1436 rememberHook := func() error {
1437 called.Store(true)
1438 defer close(learnerDone)
1439 if tt.remember != nil {
1440 return tt.remember()
1441 }
1442 return nil
1443 }
1444
1445 tr := portsmocks.NewMockTransport(t)
1446 if !tt.preWelcomeReject {
1447 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1448 }
1449 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
1450 if tt.preWelcomeReject {
1451 tr.EXPECT().Recv().Return(tt.recv[0].f, tt.recv[0].err).Once()
1452 } else {
1453 unblock := scriptRecv(tr, tt.recv...)
1454 defer unblock()
1455 }
1456 tr.EXPECT().Close().Return(nil).Once()
1457
1458 deps := attachTestDependencies(tr, tm, realClock{})
1459 learner := portsmocks.NewMockRemoteHostLearner(t)
1460 if tt.wantCalled {
1461 learner.EXPECT().RememberRemoteHost().RunAndReturn(rememberHook).Once()
1462 }
1463 deps.RemoteHostLearner = learner
1464
1465 err := runTestClient(context.Background(), deps, tt.request)
1466 if tt.wantCalled {
1467 select {
1468 case <-learnerDone:
1469 case <-time.After(2 * time.Second):
1470 t.Fatal("remote host learner did not complete")
1471 }
1472 }
1473 if tt.wantErr {
1474 require.Error(t, err)
1475 } else {
1476 require.NoError(t, err)
1477 }
1478 require.Equal(t, tt.wantCalled, called.Load())
1479 })
1480 }
1481}
1482
1483func TestRunRememberRemoteHostAtMostOnceAcrossReconnects(t *testing.T) {
1484 term := newRunTerminal()
1485 defer term.in.unblock()
1486
1487 var rememberCalls atomic.Int32
1488 learnerDone := make(chan struct{})
1489
1490 tr1 := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {err: io.EOF}}}
1491 tr2 := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(22)}, {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))}}}
1492 d := &sequenceDialer{trs: []ports.Transport{tr1, tr2}}
1493
1494 deps := testDependencies(d, term, realClock{}, nil, nil)
1495 learner := portsmocks.NewMockRemoteHostLearner(t)
1496 learner.EXPECT().RememberRemoteHost().RunAndReturn(func() error {
1497 rememberCalls.Add(1)
1498 close(learnerDone)
1499 return nil
1500 }).Once()
1501 deps.RemoteHostLearner = learner
1502 err := runTestClient(context.Background(), deps, client.AttachRequest{
1503 Intent: ports.IntentAttach,
1504 SessionName: "main",
1505 Remote: true,
1506 })
1507 require.NoError(t, err)
1508 select {
1509 case <-learnerDone:
1510 case <-time.After(2 * time.Second):
1511 t.Fatal("remote host learner did not complete")
1512 }
1513 require.Equal(t, int32(1), rememberCalls.Load())
1514 require.Equal(t, int32(2), d.calls.Load())
1515}
1516
1517func TestAttachRememberRemoteHostDoesNotBlockAfterWelcome(t *testing.T) {
1518 var out bytes.Buffer
1519 var restoreCount atomic.Int32
1520 resizeCh := make(chan domain.Size)
1521 tm := portsmocks.NewMockTerminal(t)
1522 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
1523 enteredRaw := make(chan struct{})
1524 tm.EXPECT().EnterRaw().Run(func() { close(enteredRaw) }).Return(func() error {
1525 restoreCount.Add(1)
1526 return nil
1527 }, nil).Once()
1528 in := newBlockingReader()
1529 tm.EXPECT().In().Return(in).Maybe()
1530 tm.EXPECT().Out().Return(&out).Maybe()
1531 tm.EXPECT().Flush().Return(nil).Maybe()
1532 tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
1533 defer in.unblock()
1534
1535 learnerStarted := make(chan struct{})
1536 releaseLearner := make(chan struct{})
1537 learnerDone := make(chan struct{})
1538
1539 tr := portsmocks.NewMockTransport(t)
1540 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1541 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
1542 unblock := scriptRecv(tr,
1543 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1544 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1545 )
1546 defer unblock()
1547 tr.EXPECT().Close().Return(nil).Once()
1548
1549 deps := attachTestDependencies(tr, tm, realClock{})
1550 learner := portsmocks.NewMockRemoteHostLearner(t)
1551 learner.EXPECT().RememberRemoteHost().RunAndReturn(func() error {
1552 close(learnerStarted)
1553 <-releaseLearner
1554 close(learnerDone)
1555 return nil
1556 }).Once()
1557 deps.RemoteHostLearner = learner
1558
1559 errCh := make(chan error, 1)
1560 go func() {
1561 errCh <- runTestClient(context.Background(), deps, client.AttachRequest{
1562 Intent: ports.IntentAttach,
1563 SessionName: "main",
1564 Remote: true,
1565 })
1566 }()
1567
1568 select {
1569 case <-learnerStarted:
1570 case <-time.After(2 * time.Second):
1571 t.Fatal("remote host learner did not start")
1572 }
1573 select {
1574 case <-enteredRaw:
1575 case <-time.After(2 * time.Second):
1576 t.Fatal("attach did not enter raw mode while remote host learner was blocked")
1577 }
1578 select {
1579 case err := <-errCh:
1580 t.Fatalf("run returned before remote host learner completed: %v", err)
1581 default:
1582 }
1583 close(releaseLearner)
1584 select {
1585 case <-learnerDone:
1586 case <-time.After(2 * time.Second):
1587 t.Fatal("remote host learner did not complete after release")
1588 }
1589 select {
1590 case err := <-errCh:
1591 require.NoError(t, err)
1592 case <-time.After(2 * time.Second):
1593 t.Fatal("run did not return after remote host learner completed")
1594 }
1595}
1596
1597func TestRunRestoresTerminalBeforeWaitingForLearner(t *testing.T) {
1598 var out bytes.Buffer
1599 var restoreCount atomic.Int32
1600 restored := make(chan struct{})
1601 resizeCh := make(chan domain.Size)
1602 tm := portsmocks.NewMockTerminal(t)
1603 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
1604 tm.EXPECT().EnterRaw().Return(func() error {
1605 restoreCount.Add(1)
1606 close(restored)
1607 return nil
1608 }, nil).Once()
1609 in := newBlockingReader()
1610 tm.EXPECT().In().Return(in).Maybe()
1611 tm.EXPECT().Out().Return(&out).Maybe()
1612 tm.EXPECT().Flush().Return(nil).Maybe()
1613 tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
1614 defer in.unblock()
1615
1616 learnerStarted := make(chan struct{})
1617 releaseLearner := make(chan struct{})
1618 learnerDone := make(chan struct{})
1619
1620 tr := portsmocks.NewMockTransport(t)
1621 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1622 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
1623 unblock := scriptRecv(tr,
1624 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1625 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1626 )
1627 defer unblock()
1628 tr.EXPECT().Close().Return(nil).Once()
1629
1630 deps := attachTestDependencies(tr, tm, realClock{})
1631 learner := portsmocks.NewMockRemoteHostLearner(t)
1632 learner.EXPECT().RememberRemoteHost().RunAndReturn(func() error {
1633 close(learnerStarted)
1634 <-releaseLearner
1635 close(learnerDone)
1636 return nil
1637 }).Once()
1638 deps.RemoteHostLearner = learner
1639
1640 errCh := make(chan error, 1)
1641 go func() {
1642 errCh <- runTestClient(context.Background(), deps, client.AttachRequest{
1643 Intent: ports.IntentAttach,
1644 SessionName: "main",
1645 Remote: true,
1646 })
1647 }()
1648
1649 select {
1650 case <-learnerStarted:
1651 case <-time.After(2 * time.Second):
1652 t.Fatal("remote host learner did not start")
1653 }
1654 select {
1655 case <-restored:
1656 case <-time.After(2 * time.Second):
1657 t.Fatal("terminal was not restored while remote host learner was blocked")
1658 }
1659 require.Equal(t, int32(1), restoreCount.Load())
1660 select {
1661 case err := <-errCh:
1662 t.Fatalf("run returned before remote host learner completed: %v", err)
1663 default:
1664 }
1665 close(releaseLearner)
1666 select {
1667 case <-learnerDone:
1668 case <-time.After(2 * time.Second):
1669 t.Fatal("remote host learner did not complete after release")
1670 }
1671 select {
1672 case err := <-errCh:
1673 require.NoError(t, err)
1674 case <-time.After(2 * time.Second):
1675 t.Fatal("run did not return after remote host learner completed")
1676 }
1677}
1678
1679func TestRunReturnsWhenRemoteHostLearnerStalls(t *testing.T) {
1680 var out bytes.Buffer
1681 var restoreCount atomic.Int32
1682 restored := make(chan struct{})
1683 resizeCh := make(chan domain.Size)
1684 tm := portsmocks.NewMockTerminal(t)
1685 tm.EXPECT().Size().Return(domain.Size{Cols: 80, Rows: 24}, nil).Once()
1686 tm.EXPECT().EnterRaw().Return(func() error {
1687 restoreCount.Add(1)
1688 close(restored)
1689 return nil
1690 }, nil).Once()
1691 in := newBlockingReader()
1692 tm.EXPECT().In().Return(in).Maybe()
1693 tm.EXPECT().Out().Return(&out).Maybe()
1694 tm.EXPECT().Flush().Return(nil).Maybe()
1695 tm.EXPECT().ResizeEvents().Return(resizeCh).Maybe()
1696 defer in.unblock()
1697
1698 learnerStarted := make(chan struct{})
1699 blockLearner := make(chan struct{})
1700 defer close(blockLearner)
1701
1702 shutdownTimerC := make(chan time.Time, 1)
1703 shutdownTimerCreated := make(chan struct{})
1704 shutdownTimer := portsmocks.NewMockTimer(t)
1705 shutdownTimer.EXPECT().C().Return((<-chan time.Time)(shutdownTimerC)).Once()
1706 shutdownTimer.EXPECT().Stop().Return(true).Once()
1707 clock := portsmocks.NewMockClock(t)
1708 clock.EXPECT().Now().Return(time.Now()).Maybe()
1709 clock.EXPECT().NewTimer(mock.Anything).RunAndReturn(func(d time.Duration) ports.Timer {
1710 if d == time.Second {
1711 close(shutdownTimerCreated)
1712 return shutdownTimer
1713 }
1714 return realClock{}.NewTimer(d)
1715 }).Maybe()
1716
1717 tr := portsmocks.NewMockTransport(t)
1718 tr.EXPECT().Send(isType(ports.MsgTheme)).Return(nil).Maybe()
1719 tr.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
1720 unblock := scriptRecv(tr,
1721 recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
1722 recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
1723 )
1724 defer unblock()
1725 tr.EXPECT().Close().Return(nil).Once()
1726
1727 deps := attachTestDependencies(tr, tm, clock)
1728 learner := portsmocks.NewMockRemoteHostLearner(t)
1729 learner.EXPECT().RememberRemoteHost().RunAndReturn(func() error {
1730 close(learnerStarted)
1731 <-blockLearner
1732 return nil
1733 }).Once()
1734 deps.RemoteHostLearner = learner
1735
1736 errCh := make(chan error, 1)
1737 go func() {
1738 errCh <- runTestClient(context.Background(), deps, client.AttachRequest{
1739 Intent: ports.IntentAttach,
1740 SessionName: "main",
1741 Remote: true,
1742 })
1743 }()
1744
1745 select {
1746 case <-learnerStarted:
1747 case <-time.After(2 * time.Second):
1748 t.Fatal("remote host learner did not start")
1749 }
1750 select {
1751 case <-restored:
1752 case <-time.After(2 * time.Second):
1753 t.Fatal("terminal was not restored while remote host learner was stalled")
1754 }
1755 require.Equal(t, int32(1), restoreCount.Load())
1756
1757 select {
1758 case <-shutdownTimerCreated:
1759 case <-time.After(2 * time.Second):
1760 t.Fatal("remote host learner shutdown timer was not created")
1761 }
1762 shutdownTimerC <- time.Now()
1763 select {
1764 case err := <-errCh:
1765 require.NoError(t, err)
1766 case <-time.After(2 * time.Second):
1767 t.Fatal("run did not return after remote host learner shutdown timer fired")
1768 }
1769}
1770
1771func TestRunDoesNotRetryTerminalDetachedError(t *testing.T) {
1772 term := newRunTerminal()
1773 defer term.in.unblock()
1774 tr := &recordingTransport{recvs: []recvItem{{f: welcomeFrame(11)}, {f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonSessionKilled}))}}}
1775 d := &sequenceDialer{trs: []ports.Transport{tr}}
1776
1777 err := runTestClient(context.Background(), testDependencies(d, term, realClock{}, nil, nil), client.AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: false})
1778 require.Error(t, err)
1779 var de *client.DetachedError
1780 require.True(t, errors.As(err, &de))
1781 require.Equal(t, int32(1), d.calls.Load())
1782 require.Equal(t, int32(1), term.restoreCount.Load())
1783}