1package daemon
2
3import (
4 "context"
5 "errors"
6 "strings"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13
14 vt "github.com/bnema/vev-vt"
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/ports"
17 portsmocks "github.com/bnema/vev/internal/ports/mocks"
18 "github.com/bnema/vev/internal/usecase/layout"
19)
20
21type closeTrackingTransport struct {
22 mu sync.Mutex
23 closed bool
24 sends []ports.Frame
25}
26
27func (t *closeTrackingTransport) Send(f ports.Frame) error {
28 t.mu.Lock()
29 defer t.mu.Unlock()
30 t.sends = append(t.sends, f)
31 return nil
32}
33func (t *closeTrackingTransport) Recv() (ports.Frame, error) {
34 return ports.Frame{}, errors.New("closed")
35}
36func (t *closeTrackingTransport) Close() error {
37 t.mu.Lock()
38 t.closed = true
39 t.mu.Unlock()
40 return nil
41}
42func (t *closeTrackingTransport) Closed() bool {
43 t.mu.Lock()
44 defer t.mu.Unlock()
45 return t.closed
46}
47
48func (t *closeTrackingTransport) Sends() []ports.Frame {
49 t.mu.Lock()
50 defer t.mu.Unlock()
51 return append([]ports.Frame(nil), t.sends...)
52}
53
54type closeCountingBlockedTransport struct {
55 closeOnce sync.Once
56 closed chan struct{}
57 count int
58 mu sync.Mutex
59}
60
61func newCloseCountingBlockedTransport() *closeCountingBlockedTransport {
62 return &closeCountingBlockedTransport{closed: make(chan struct{})}
63}
64
65func (t *closeCountingBlockedTransport) Send(ports.Frame) error {
66 <-t.closed
67 return errors.New("closed")
68}
69
70func (t *closeCountingBlockedTransport) Recv() (ports.Frame, error) {
71 return ports.Frame{}, errors.New("closed")
72}
73
74func (t *closeCountingBlockedTransport) Close() error {
75 t.mu.Lock()
76 t.count++
77 t.mu.Unlock()
78 t.closeOnce.Do(func() { close(t.closed) })
79 return nil
80}
81
82func (t *closeCountingBlockedTransport) CloseCount() int {
83 t.mu.Lock()
84 defer t.mu.Unlock()
85 return t.count
86}
87
88func helloResumeCapable(intent uint8, name string, token uint64) ports.Hello {
89 return ports.Hello{
90 Version: ports.ProtocolVersion,
91 Intent: intent,
92 ClientID: [16]byte{1, 2, 3, 4},
93 ResumeToken: token,
94 Name: name,
95 Size: domain.Size{Cols: 80, Rows: 24},
96 TermEnv: "xterm-256color",
97 }
98}
99
100func TestTokenlessResumeIsRejectedBeforeNameRouting(t *testing.T) {
101 d := newTestDaemon(t, nil, stubClock{})
102 tr, _ := newCapturingTransport(t)
103 _, _, err := d.route(helloResumeCapable(ports.IntentResume, "work", 0), tr)
104 var protocol *protoErr
105 require.ErrorAs(t, err, &protocol)
106 require.Equal(t, ports.ErrNoSuchSession, protocol.code)
107 d.mu.Lock()
108 sessionCount := len(d.sessions)
109 d.mu.Unlock()
110 require.Zero(t, sessionCount)
111}
112
113func TestResumeRejectsMismatchedRemoteTargetBeforeOwnershipMutation(t *testing.T) {
114 pty, release := newBlockingPTY(t)
115 defer release()
116 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
117 oldTransport := &closeTrackingTransport{}
118 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport)
119 require.NoError(t, err)
120 token := ac.resumeToken
121
122 target := domain.RemoteSessionTarget{
123 LifecycleID: sess.incarnation, SessionName: sess.name, LiveTabID: "missing-tab",
124 }
125 resume := helloResumeCapable(ports.IntentResume, "work", token)
126 resume.RemoteTarget = &target
127 resume.EnvironmentPolicy = ports.EnvironmentPolicyDaemonOwned
128 _, _, err = d.route(resume, &closeTrackingTransport{})
129 var protocol *protoErr
130 require.ErrorAs(t, err, &protocol)
131 require.Equal(t, ports.ErrNoSuchTarget, protocol.code)
132 require.Same(t, oldTransport, ac.transport())
133 d.mu.Lock()
134 _, parked := d.parked[token]
135 d.mu.Unlock()
136 require.False(t, parked, "mismatched target must not park the live attachment")
137
138 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
139}
140
141func TestParkedResumeRejectsMismatchedRemoteTargetBeforeClaim(t *testing.T) {
142 pty, release := newBlockingPTY(t)
143 defer release()
144 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
145 oldTransport := &closeTrackingTransport{}
146 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport)
147 require.NoError(t, err)
148 token := ac.resumeToken
149 d.clientGone(sess, ac, oldTransport, false)
150
151 target := domain.RemoteSessionTarget{
152 LifecycleID: sess.incarnation, SessionName: sess.name, LiveTabID: "missing-tab",
153 }
154 resume := helloResumeCapable(ports.IntentResume, "work", token)
155 resume.RemoteTarget = &target
156 resume.EnvironmentPolicy = ports.EnvironmentPolicyDaemonOwned
157 _, _, err = d.route(resume, &closeTrackingTransport{})
158 var protocol *protoErr
159 require.ErrorAs(t, err, &protocol)
160 require.Equal(t, ports.ErrNoSuchTarget, protocol.code)
161 d.mu.Lock()
162 parked := d.parked[token]
163 claimed := parked != nil && parked.claimed
164 d.mu.Unlock()
165 require.NotNil(t, parked)
166 require.False(t, claimed, "mismatched target must not claim the parked attachment")
167
168 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
169}
170
171func TestFailedResumeRearmsParkExpiryAfterClaimTimerFires(t *testing.T) {
172 clock := &signalClock{timers: make(chan *signalTimer, 8)}
173 pty, release := newBlockingPTY(t)
174 defer release()
175 d := newTestDaemon(t, newFactory(t, pty), clock)
176
177 oldTransport := &closeTrackingTransport{}
178 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport)
179 require.NoError(t, err)
180 token := ac.resumeToken
181 d.clientGone(sess, ac, oldTransport, false)
182 parkTimer := <-clock.timers
183 d.mu.Lock()
184 parked := d.parked[token]
185 d.mu.Unlock()
186 require.NotNil(t, parked)
187
188 resumed, resumedAC, ok, err := d.resumeParked(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{}, defaultSize)
189 require.NoError(t, err)
190 require.True(t, ok)
191 require.Same(t, sess, resumed)
192 require.Same(t, ac, resumedAC)
193
194 // Drive the original fake timer after the claim. The old entry remains
195 // claimable only until abort; abort must install a fresh expiry rather than
196 // restoring a credential whose timer has already fired.
197 parkTimer.ch <- time.Time{}
198 d.expireParked(token, parked)
199 require.True(t, d.abortResumeClaim(ac))
200 rearmedTimer := <-clock.timers
201 require.NotSame(t, parkTimer, rearmedTimer)
202 d.mu.Lock()
203 rearmed := d.parked[token]
204 claimed := rearmed != nil && rearmed.claimed
205 d.mu.Unlock()
206 require.NotNil(t, rearmed)
207 require.False(t, claimed)
208 rearmedTimer.ch <- time.Time{}
209 d.expireParked(token, rearmed)
210 d.mu.Lock()
211 _, retained := d.parked[token]
212 d.mu.Unlock()
213 require.False(t, retained, "rearmed credential must still expire")
214
215 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, false))
216 d.sessWg.Wait()
217}
218
219func TestNamedLinkLossParks(t *testing.T) {
220 pty, release := newBlockingPTY(t)
221 defer release()
222 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
223
224 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
225 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
226 require.NoError(t, err)
227 token := ac.resumeToken
228 require.NotZero(t, token)
229
230 d.clientGone(sess, ac, ac.transport(), false)
231 require.Equal(t, 1, sessionCount(d), "named session survives parked link loss")
232 require.Empty(t, sess.snapshotAttachments())
233 d.mu.Lock()
234 _, parked := d.parked[token]
235 d.mu.Unlock()
236 require.True(t, parked, "named resume-capable link loss is parked")
237}
238
239func TestBoundedSendTimeoutCannotTargetResumedTransport(t *testing.T) {
240 pty, release := newBlockingPTY(t)
241 defer release()
242 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
243 oldTransport := &closeTrackingTransport{}
244 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "bounded-send-resume", 0), oldTransport)
245 require.NoError(t, err)
246 token := ac.resumeToken
247 d.clientGone(sess, ac, oldTransport, false)
248
249 clock := &signalClock{timers: make(chan *signalTimer, 1)}
250 d.clock = clock
251 ac.sendMu.Lock()
252 expected := ac.transportSnapshot()
253 orphanDone := make(chan struct{})
254 type boundedResult struct {
255 transport ports.Transport
256 err error
257 }
258 result := make(chan boundedResult, 1)
259 go func() {
260 transport, sendErr := d.boundedSendWithTimeout(time.Second, expected.transport, func() error {
261 defer close(orphanDone)
262 return ac.sendExpectedTransport(expected, frameDetached(ports.ReasonDetach))
263 })
264 result <- boundedResult{transport: transport, err: sendErr}
265 }()
266
267 // The worker is parked behind sendMu, so it has not observed a transport.
268 // Expire its deadline before resuming this same attachment with a new link.
269 timer := <-clock.timers
270 timer.ch <- time.Time{}
271 got := <-result
272 require.ErrorIs(t, got.err, errSendTimedOut)
273 require.Same(t, oldTransport, got.transport)
274 require.NoError(t, ac.closeCapturedTransport(got.transport))
275
276 newTransport := &closeTrackingTransport{}
277 d.mu.Lock()
278 resumedSess, resumedAC, ok, err := d.resumeParkedLocked(
279 helloResumeCapable(ports.IntentResume, sess.name, token),
280 newTransport,
281 domain.Size{Cols: 80, Rows: 24},
282 )
283 d.mu.Unlock()
284 require.NoError(t, err)
285 require.True(t, ok)
286 require.Same(t, sess, resumedSess)
287 require.Same(t, ac, resumedAC)
288
289 ac.sendMu.Unlock()
290 <-orphanDone
291 require.Empty(t, newTransport.Sends(), "orphaned send must not write to the resumed transport")
292 require.False(t, newTransport.Closed(), "orphaned send must not close the resumed transport")
293}
294
295func TestDetachedNotificationTimeoutClosesCapturedTransportOnce(t *testing.T) {
296 clock := &signalClock{timers: make(chan *signalTimer, 1)}
297 d := newTestDaemon(t, nil, clock)
298 tr := newCloseCountingBlockedTransport()
299 ac := &attachedClient{tr: tr, output: newOutputStateStream()}
300
301 d.notifyDetachedSnapshotAsync(detachedAttachmentSnapshot{
302 ac: ac,
303 transport: ac.transportSnapshot(),
304 }, ports.ReasonDetach)
305 timer := <-clock.timers
306 timer.ch <- time.Time{}
307 d.waitNotifies()
308
309 require.Equal(t, 1, tr.CloseCount(), "timeout revocation must close the captured transport at most once")
310 require.Nil(t, ac.transport(), "timeout cleanup must revoke the captured transport")
311}
312
313func TestHandleHelloResumeDefersFreshOutputUntilWelcome(t *testing.T) {
314 pty, releasePTY := newBlockingPTY(t)
315 defer releasePTY()
316 clock := newCoordinatorMockClock(t, 4)
317 d := newTestDaemon(t, newFactorySeq(t, pty), clock.clock)
318 oldTransport := &closeTrackingTransport{}
319 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "resume-welcome-gate", 0), oldTransport)
320 require.NoError(t, err)
321 token := ac.resumeToken
322 d.clientGone(sess, ac, oldTransport, false)
323
324 tr := newWelcomeBlockingTransport(t)
325 done := make(chan struct{})
326 resumeHello := helloResumeCapable(ports.IntentResume, sess.name, token)
327 go func() {
328 d.handleHello(tr.tr, ports.Frame{Type: ports.MsgHello, Payload: ports.MarshalHello(resumeHello)})
329 close(done)
330 }()
331
332 <-tr.welcomeEntered
333 handshakeTimer := awaitHandshakeTimer(t, clock)
334 welcome := <-tr.sends
335 require.Equal(t, ports.MsgWelcome, welcome.Type)
336 sess.mu.Lock()
337 resumed := sess.snapshotAttachmentsLocked()[0]
338 sess.mu.Unlock()
339 require.Same(t, ac, resumed)
340
341 d.invalidateRender(sess, resumed, true, "resume-welcome-gate-test")
342 timer := awaitLatestCoordinatorTimer(t, clock)
343 rc := sess.renderCoordinator()
344 rc.mu.Lock()
345 workerDone := rc.normalLane.token.done
346 rc.mu.Unlock()
347 require.NotNil(t, workerDone)
348 timer.ch <- time.Time{}
349 awaitTestCompletion(t, workerDone, "coordinator deadline worker did not complete")
350 requireNoCoordinatorOutputFrame(t, tr.sends)
351
352 tr.release()
353 output := awaitFrame(t, tr.sends, ports.MsgOutput)
354 first, err := ports.UnmarshalOutput(output.Payload)
355 require.NoError(t, err)
356 require.Zero(t, first.Base)
357 tr.finish()
358 <-done
359 awaitTestCompletion(t, handshakeTimer.stopped, "handshake timer was not canceled after publication")
360 requireNoCoordinatorOutputFrame(t, tr.sends)
361}
362
363func TestEphemeralLinkLossParksAndResumes(t *testing.T) {
364 pty, release := newBlockingPTY(t)
365 defer release()
366 d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
367
368 tr := &closeTrackingTransport{}
369 initial := helloResumeCapable(ports.IntentEphemeral, "", 0)
370 initial.Size = domain.Size{Cols: 120, Rows: 40}
371 sess, ac, err := d.route(initial, tr)
372 require.NoError(t, err)
373 require.True(t, sess.ephemeral)
374 token := ac.resumeToken
375 require.NotZero(t, token, "ephemeral sessions receive resume tokens")
376
377 peerHello := helloResumeCapable(ports.IntentAttach, sess.name, 0)
378 peerHello.ClientID = [16]byte{9}
379 peerHello.Size = domain.Size{Cols: 100, Rows: 30}
380 peer := &closeTrackingTransport{}
381 _, peerAC, err := d.route(peerHello, peer)
382 require.NoError(t, err)
383 require.Same(t, sess, peerAC.currentSession())
384 d.firstPaint(sess, peerAC)
385 tb := sess.tabs[0]
386 tb.mu.Lock()
387 beforeParkingSize := tb.size
388 tb.mu.Unlock()
389 require.Equal(t, domain.Size{Cols: 100, Rows: 28}, beforeParkingSize, "the later attachment claim should be current before parking")
390
391 d.clientGone(sess, ac, ac.transport(), false)
392 require.Equal(t, 1, sessionCount(d), "ephemeral link loss keeps session alive")
393 attachments := sess.snapshotAttachments()
394 var peerRegistered bool
395 for _, candidate := range attachments {
396 require.NotSame(t, ac, candidate, "parked attachment remained registered")
397 if candidate == peerAC {
398 peerRegistered = true
399 }
400 }
401 require.True(t, peerRegistered, "the live peer attachment was removed while parking")
402 d.mu.Lock()
403 _, parked := d.parked[token]
404 d.mu.Unlock()
405 require.True(t, parked, "ephemeral resume-capable link loss is parked")
406
407 resumeHello := helloResumeCapable(ports.IntentResume, sess.name, token)
408 resumeHello.Size = domain.Size{Cols: 80, Rows: 24}
409 newTr := &closeTrackingTransport{}
410 resumedSess, resumedAC, err := d.route(resumeHello, newTr)
411 require.NoError(t, err)
412 require.Same(t, sess, resumedSess)
413 require.Same(t, ac, resumedAC)
414 require.NotEqual(t, token, resumedAC.resumeToken, "resume rotates token")
415 require.Contains(t, sess.snapshotAttachments(), resumedAC)
416 d.firstPaint(sess, resumedAC)
417 require.Equal(t, domain.Size{Cols: 100, Rows: 30}, peerAC.sizeSnapshot())
418 tb.mu.Lock()
419 resumedSize := tb.size
420 tb.mu.Unlock()
421 require.Equal(t, domain.Size{Cols: 80, Rows: 22}, resumedSize, "resume must make its new size the latest geometry claim")
422}
423
424// TestResumeParkedTokenReplacedDuringWaitFailsClosed covers the lifecycle race
425// where a parked credential is consumed or replaced while IntentResume waits on
426// the attachment send lock. The nonzero token must fail closed instead of
427// falling through to ordinary attach/create routing.
428func TestResumeParkedTokenReplacedDuringWaitFailsClosed(t *testing.T) {
429 pty, release := newBlockingPTY(t)
430 defer release()
431 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
432
433 oldTr := &closeTrackingTransport{}
434 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
435 require.NoError(t, err)
436 token := ac.resumeToken
437 require.NotZero(t, token)
438 d.clientGone(sess, ac, oldTr, false)
439 require.Empty(t, sess.snapshotAttachments())
440 d.mu.Lock()
441 parked := d.parked[token]
442 sessionsBefore := len(d.sessions)
443 d.mu.Unlock()
444 require.NotNil(t, parked)
445
446 reachedLookup := make(chan struct{})
447 releaseLookup := make(chan struct{})
448 var releaseOnce sync.Once
449 t.Cleanup(func() { releaseOnce.Do(func() { close(releaseLookup) }) })
450 d.beforeResumeParkedSendMu = func() {
451 close(reachedLookup)
452 <-releaseLookup
453 }
454
455 resumeTr := &closeTrackingTransport{}
456 type routeResult struct {
457 sess *session
458 ac *attachedClient
459 err error
460 }
461 result := make(chan routeResult, 1)
462 go func() {
463 routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), resumeTr)
464 result <- routeResult{sess: routedSess, ac: routedAC, err: routeErr}
465 }()
466 awaitTestCompletion(t, reachedLookup, "resumeParked did not pause after the parked lookup")
467
468 d.mu.Lock()
469 require.Same(t, parked, d.parked[token], "fixture: token must still be parked at the seam")
470 d.removeParkedLocked(token, parked)
471 require.Nil(t, d.parked[token])
472 require.Same(t, sess, d.sessions[sess.id], "fixture: named session must remain registered")
473 d.mu.Unlock()
474
475 releaseOnce.Do(func() { close(releaseLookup) })
476 got := awaitTestValue(t, result, "IntentResume did not finish after parked token replacement")
477
478 require.Error(t, got.err)
479 require.Nil(t, got.sess)
480 require.Nil(t, got.ac)
481 var pe *protoErr
482 require.ErrorAs(t, got.err, &pe)
483 require.Equal(t, ports.ErrNoSuchSession, pe.code)
484 require.Contains(t, pe.Error(), "resume token is no longer valid")
485 require.Empty(t, sess.snapshotAttachments(), "lifecycle-race resume must not take over the named session")
486 require.Empty(t, resumeTr.Sends(), "failed resume must not complete a Welcome handshake")
487 d.mu.Lock()
488 _, stillParked := d.parked[token]
489 sessionsAfter := len(d.sessions)
490 d.mu.Unlock()
491 require.False(t, stillParked, "consumed token must stay invalid")
492 require.Equal(t, sessionsBefore, sessionsAfter, "lifecycle-race resume must not create a session")
493}
494
495// TestResumeLiveAttachmentParkedResumeRaceFailsClosed covers the live-recovery
496// path that parks a still-active credential and then loses a competing
497// resumeParked race before sendMu revalidation. The lifecycle sentinel must
498// surface as fail-closed ErrNoSuchSession, never as a raw ErrInternal leak to
499// handleHello.
500func TestResumeLiveAttachmentParkedResumeRaceFailsClosed(t *testing.T) {
501 pty, release := newBlockingPTY(t)
502 defer release()
503 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
504
505 oldTr := &closeTrackingTransport{}
506 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
507 require.NoError(t, err)
508 token := ac.resumeToken
509 require.NotZero(t, token)
510 require.Contains(t, sess.snapshotAttachments(), ac)
511 d.mu.Lock()
512 _, parkedAtStart := d.parked[token]
513 sessionsBefore := len(d.sessions)
514 d.mu.Unlock()
515 require.False(t, parkedAtStart, "fixture: live attachment must not be parked yet")
516
517 reachedLookup := make(chan struct{})
518 releaseLookup := make(chan struct{})
519 var releaseOnce sync.Once
520 t.Cleanup(func() { releaseOnce.Do(func() { close(releaseLookup) }) })
521 d.beforeResumeParkedSendMu = func() {
522 close(reachedLookup)
523 <-releaseLookup
524 }
525
526 resumeTr := &closeTrackingTransport{}
527 type routeResult struct {
528 sess *session
529 ac *attachedClient
530 err error
531 }
532 result := make(chan routeResult, 1)
533 go func() {
534 routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), resumeTr)
535 result <- routeResult{sess: routedSess, ac: routedAC, err: routeErr}
536 }()
537 awaitTestCompletion(t, reachedLookup, "live resume did not pause inside resumeParked after parking")
538
539 d.mu.Lock()
540 parked := d.parked[token]
541 require.NotNil(t, parked, "fixture: live recovery must have parked before the sendMu seam")
542 require.Same(t, ac, parked.ac)
543 d.removeParkedLocked(token, parked)
544 require.Nil(t, d.parked[token])
545 require.Same(t, sess, d.sessions[sess.id], "fixture: named session must remain registered")
546 d.mu.Unlock()
547
548 releaseOnce.Do(func() { close(releaseLookup) })
549 got := awaitTestValue(t, result, "live IntentResume did not finish after competing resume consumed the token")
550
551 require.Error(t, got.err)
552 require.Nil(t, got.sess)
553 require.Nil(t, got.ac)
554 require.False(t, errors.Is(got.err, errResumeTokenLifecycleRace), "raw lifecycle sentinel must not reach route callers")
555 var pe *protoErr
556 require.ErrorAs(t, got.err, &pe)
557 require.Equal(t, ports.ErrNoSuchSession, pe.code)
558 require.Contains(t, pe.Error(), "resume token is no longer valid")
559 require.Empty(t, sess.snapshotAttachments(), "losing live resume must not take over the named session")
560 require.Empty(t, resumeTr.Sends(), "failed live resume must not complete a Welcome handshake")
561 d.mu.Lock()
562 _, stillParked := d.parked[token]
563 sessionsAfter := len(d.sessions)
564 d.mu.Unlock()
565 require.False(t, stillParked, "consumed token must stay invalid")
566 require.Equal(t, sessionsBefore, sessionsAfter, "losing live resume must not create a session")
567}
568
569// TestResumeActiveTokenBeforeParkRecoversSameAttachment covers the laptop-sleep
570// reconnect race: IntentResume arrives with the live attachment's token before
571// transport teardown has parked it. Same client must recover; unknown and
572// mismatched credentials stay fail-closed.
573func TestResumeDuringTeardownBeforeParkRecoversSameAttachment(t *testing.T) {
574 pty, release := newBlockingPTY(t)
575 defer release()
576 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
577
578 oldTr := &closeTrackingTransport{}
579 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
580 require.NoError(t, err)
581 token := ac.resumeToken
582 require.NotZero(t, token)
583 require.Contains(t, sess.snapshotAttachments(), ac)
584
585 reachedGap := make(chan struct{})
586 releaseGap := make(chan struct{})
587 var releaseOnce sync.Once
588 t.Cleanup(func() { releaseOnce.Do(func() { close(releaseGap) }) })
589 d.afterClientGoneDetach = func() {
590 close(reachedGap)
591 <-releaseGap
592 }
593
594 goneDone := make(chan struct{})
595 go func() {
596 d.clientGone(sess, ac, oldTr, false)
597 close(goneDone)
598 }()
599 awaitTestCompletion(t, reachedGap, "teardown did not pause after detach before park")
600
601 require.Empty(t, sess.snapshotAttachments(), "fixture: detach must have cleared the live owner")
602 d.mu.Lock()
603 parkingInGap := d.parking[token]
604 _, parkedInGap := d.parked[token]
605 d.mu.Unlock()
606 require.NotNil(t, parkingInGap, "fixture: parking marker must precede detach publication")
607 require.Same(t, ac, parkingInGap.ac)
608 require.Same(t, sess, parkingInGap.sess)
609 require.False(t, parkedInGap, "fixture: park must not have published yet")
610
611 _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", 0xdecafbad), &closeTrackingTransport{})
612 require.Error(t, err)
613 var unknownErr *protoErr
614 require.ErrorAs(t, err, &unknownErr)
615 require.Equal(t, ports.ErrNoSuchSession, unknownErr.code)
616 require.Contains(t, unknownErr.Error(), "resume token is no longer valid")
617
618 wrongClient := helloResumeCapable(ports.IntentResume, "work", token)
619 wrongClient.ClientID = [16]byte{9, 9, 9, 9}
620 _, _, err = d.route(wrongClient, &closeTrackingTransport{})
621 require.Error(t, err)
622 var mismatchErr *protoErr
623 require.ErrorAs(t, err, &mismatchErr)
624 require.Equal(t, ports.ErrNoSuchSession, mismatchErr.code)
625 require.Contains(t, mismatchErr.Error(), "resume token is no longer valid")
626
627 newTr := &closeTrackingTransport{}
628 type routeResult struct {
629 sess *session
630 ac *attachedClient
631 err error
632 }
633 waiterArmed := make(chan struct{})
634 d.afterParkingWaitArmed = func() { close(waiterArmed) }
635 result := make(chan routeResult, 1)
636 go func() {
637 routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), newTr)
638 result <- routeResult{sess: routedSess, ac: routedAC, err: routeErr}
639 }()
640 awaitTestCompletion(t, waiterArmed, "resume waiter did not arm on parking marker before park")
641
642 releaseOnce.Do(func() { close(releaseGap) })
643 awaitTestCompletion(t, goneDone, "teardown did not finish after park release")
644 got := awaitTestValue(t, result, "IntentResume did not finish after teardown-before-park gap")
645
646 require.NoError(t, got.err, "same-client token must recover across detach-before-park")
647 require.Same(t, sess, got.sess)
648 require.Same(t, ac, got.ac)
649 require.Contains(t, sess.snapshotAttachments(), ac)
650 require.Same(t, newTr, ac.transport())
651 require.NotEqual(t, token, ac.resumeToken, "successful resume rotates the credential")
652 require.True(t, d.commitResumeClaim(ac), "successful resume must consume its parked credential")
653 require.True(t, oldTr.Closed(), "teardown/resume must retire the old transport")
654 require.False(t, newTr.Closed(), "rebound transport must survive")
655 d.mu.Lock()
656 _, oldTokenParked := d.parked[token]
657 _, stillParking := d.parking[token]
658 d.mu.Unlock()
659 require.False(t, oldTokenParked, "consumed credential must not remain resumeable")
660 require.False(t, stillParking, "parking marker must be consumed after park/resume")
661
662 d.clientGone(sess, ac, oldTr, false)
663 require.Contains(t, sess.snapshotAttachments(), ac, "stale old-link cleanup must not detach the rebound attachment")
664 require.Same(t, newTr, ac.transport())
665 require.False(t, newTr.Closed())
666}
667
668// TestConcurrentLiveResumesWaitParkingMarkerBeforePark covers two same-token
669// IntentResume handshakes where the winner of detachIfCurrentTransport pauses
670// before parkAttachment. The parking marker must already be published so the
671// loser waits instead of treating the credential as unknown across that gap.
672func TestConcurrentLiveResumesWaitParkingMarkerBeforePark(t *testing.T) {
673 pty, release := newBlockingPTY(t)
674 defer release()
675 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
676
677 oldTr := &closeTrackingTransport{}
678 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
679 require.NoError(t, err)
680 token := ac.resumeToken
681 require.NotZero(t, token)
682 require.Contains(t, sess.snapshotAttachments(), ac)
683
684 reachedGap := make(chan struct{})
685 releaseGap := make(chan struct{})
686 var releaseOnce sync.Once
687 t.Cleanup(func() { releaseOnce.Do(func() { close(releaseGap) }) })
688 d.afterResumeLiveDetach = func() {
689 close(reachedGap)
690 <-releaseGap
691 }
692
693 type routeResult struct {
694 sess *session
695 ac *attachedClient
696 err error
697 }
698 firstTr := &closeTrackingTransport{}
699 firstResult := make(chan routeResult, 1)
700 go func() {
701 routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), firstTr)
702 firstResult <- routeResult{sess: routedSess, ac: routedAC, err: routeErr}
703 }()
704 awaitTestCompletion(t, reachedGap, "first live resume did not pause after detach before park")
705
706 require.Empty(t, sess.snapshotAttachments(), "fixture: winning resume must have cleared the live owner")
707 d.mu.Lock()
708 parkingInGap := d.parking[token]
709 _, parkedInGap := d.parked[token]
710 d.mu.Unlock()
711 require.NotNil(t, parkingInGap, "fixture: parking marker must precede detach for concurrent resumes")
712 require.Same(t, ac, parkingInGap.ac)
713 require.False(t, parkedInGap, "fixture: park must not have published yet")
714
715 _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", 0xdecafbad), &closeTrackingTransport{})
716 require.Error(t, err)
717 var unknownErr *protoErr
718 require.ErrorAs(t, err, &unknownErr)
719 require.Equal(t, ports.ErrNoSuchSession, unknownErr.code)
720
721 secondTr := &closeTrackingTransport{}
722 secondWaiting := make(chan struct{})
723 d.afterParkingWaitArmed = func() {
724 close(secondWaiting)
725 }
726 secondResult := make(chan routeResult, 1)
727 go func() {
728 routedSess, routedAC, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), secondTr)
729 secondResult <- routeResult{sess: routedSess, ac: routedAC, err: routeErr}
730 }()
731 awaitTestCompletion(t, secondWaiting, "second live resume did not arm parking wait")
732
733 releaseOnce.Do(func() { close(releaseGap) })
734 first := awaitTestValue(t, firstResult, "first concurrent IntentResume did not finish")
735 second := awaitTestValue(t, secondResult, "second concurrent IntentResume did not finish")
736
737 var winner, loser routeResult
738 var winnerTr, loserTr *closeTrackingTransport
739 switch {
740 case first.err == nil && second.err != nil:
741 winner, loser = first, second
742 winnerTr, loserTr = firstTr, secondTr
743 case second.err == nil && first.err != nil:
744 winner, loser = second, first
745 winnerTr, loserTr = secondTr, firstTr
746 default:
747 t.Fatalf("expected exactly one concurrent resume to succeed; first=%v second=%v", first.err, second.err)
748 }
749
750 require.Same(t, sess, winner.sess)
751 require.Same(t, ac, winner.ac)
752 require.Contains(t, sess.snapshotAttachments(), ac)
753 require.Same(t, winnerTr, ac.transport())
754 require.NotEqual(t, token, ac.resumeToken, "successful resume rotates the credential")
755 require.True(t, oldTr.Closed(), "winning resume must retire the old transport")
756 require.False(t, winnerTr.Closed(), "winning rebound transport must survive")
757
758 var loserErr *protoErr
759 require.ErrorAs(t, loser.err, &loserErr)
760 require.Equal(t, ports.ErrNoSuchSession, loserErr.code)
761 require.Contains(t, loserErr.Error(), "resume token is no longer valid")
762 require.True(t, loserTr.Closed() || len(loserTr.Sends()) == 0, "losing resume must not complete Welcome")
763
764 require.True(t, d.commitResumeClaim(ac), "successful concurrent resume must consume its parked credential")
765 d.mu.Lock()
766 _, oldTokenParked := d.parked[token]
767 _, stillParking := d.parking[token]
768 d.mu.Unlock()
769 require.False(t, oldTokenParked, "consumed credential must not remain resumeable")
770 require.False(t, stillParking, "parking marker must be consumed after park/resume")
771
772 d.clientGone(sess, ac, oldTr, false)
773 require.Contains(t, sess.snapshotAttachments(), ac, "stale old-link cleanup must not detach the rebound attachment")
774 require.Same(t, winnerTr, ac.transport())
775 require.False(t, winnerTr.Closed())
776}
777
778func TestExplicitDetachClearsOrphanedSameAttachmentParkingMarker(t *testing.T) {
779 workPTY, releaseWork := newBlockingPTY(t)
780 defer releaseWork()
781 otherPTY, releaseOther := newBlockingPTY(t)
782 defer releaseOther()
783 // Separate PTYs keep the other session alive while work detaches.
784 d := newTestDaemon(t, newFactorySeq(t, workPTY, otherPTY), stubClock{})
785
786 oldTr := &closeTrackingTransport{}
787 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
788 require.NoError(t, err)
789 token := ac.resumeToken
790 require.NotZero(t, token)
791
792 otherTr := &closeTrackingTransport{}
793 otherSess, otherAC, err := d.route(helloResumeCapable(ports.IntentNew, "other", 0), otherTr)
794 require.NoError(t, err)
795 otherToken := d.markParkingInFlight(otherSess, otherAC)
796 require.NotZero(t, otherToken)
797 require.NotEqual(t, token, otherToken)
798
799 // Pre-mark as a same-attachment non-explicit teardown would before detach.
800 require.Equal(t, token, d.markParkingInFlight(sess, ac))
801 d.mu.Lock()
802 orphaned := d.parking[token]
803 otherMarker := d.parking[otherToken]
804 d.mu.Unlock()
805 require.NotNil(t, orphaned, "fixture: non-explicit teardown must publish the parking marker")
806 require.Same(t, ac, orphaned.ac)
807 require.NotNil(t, otherMarker, "fixture: other attachment marker must remain published")
808 require.Same(t, otherAC, otherMarker.ac)
809
810 waiterArmed := make(chan struct{})
811 d.afterParkingWaitArmed = func() { close(waiterArmed) }
812 waiterDone := make(chan bool, 1)
813 go func() {
814 waiterDone <- d.waitParkingInFlight(helloResumeCapable(ports.IntentResume, "work", token))
815 }()
816 awaitTestCompletion(t, waiterArmed, "parking waiter did not arm on the orphaned marker")
817
818 d.clientGone(sess, ac, oldTr, true)
819 require.Empty(t, sess.snapshotAttachments(), "explicit detach must clear the live owner")
820
821 waited := awaitTestValue(t, waiterDone, "parking waiter stayed blocked after explicit detach")
822 require.True(t, waited, "waiter must observe the same-attachment marker before it is cleared")
823
824 d.mu.Lock()
825 _, stillParking := d.parking[token]
826 otherAfter := d.parking[otherToken]
827 d.mu.Unlock()
828 require.False(t, stillParking, "explicit winner must clear the same attachment's orphaned marker")
829 require.NotNil(t, otherAfter, "explicit winner must not clear another attachment's marker")
830 require.Same(t, otherAC, otherAfter.ac)
831
832 _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", 0xdecafbad), &closeTrackingTransport{})
833 require.Error(t, err)
834 var unknownErr *protoErr
835 require.ErrorAs(t, err, &unknownErr)
836 require.Equal(t, ports.ErrNoSuchSession, unknownErr.code)
837 require.Contains(t, unknownErr.Error(), "resume token is no longer valid")
838
839 _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{})
840 require.Error(t, err)
841 var retiredErr *protoErr
842 require.ErrorAs(t, err, &retiredErr)
843 require.Equal(t, ports.ErrNoSuchSession, retiredErr.code)
844 require.Contains(t, retiredErr.Error(), "resume token is no longer valid")
845
846 // A later parked lifecycle for another attachment must survive a clear
847 // scoped to the retired attachment identity.
848 require.True(t, d.parkAttachment(otherSess, otherAC))
849 d.mu.Lock()
850 parkedOther := d.parked[otherToken]
851 _, otherStillParking := d.parking[otherToken]
852 d.mu.Unlock()
853 require.NotNil(t, parkedOther, "subsequent park must publish the other attachment")
854 require.False(t, otherStillParking, "park consumes the other attachment's in-flight marker")
855
856 d.clearParkingInFlight(otherToken, ac)
857 d.mu.Lock()
858 parkedAfterMismatchedClear := d.parked[otherToken]
859 d.mu.Unlock()
860 require.Same(t, parkedOther, parkedAfterMismatchedClear, "mismatched clear must not drop a subsequent parked lifecycle")
861}
862
863// TestLiveResumeRejectsLateMarkerAfterTerminalCleanupWins covers the race where
864// live resume has validated the same-client credential, then explicit detach or
865// session removal wins before markParkingInFlight. Late marker publication must
866// be rejected so IntentResume returns promptly ErrNoSuchSession with no parking
867// marker left to strand waiters. Direct parkAttachment after detach still works.
868func TestLiveResumeRejectsLateMarkerAfterTerminalCleanupWins(t *testing.T) {
869 cases := []struct {
870 name string
871 win func(t *testing.T, d *Daemon, sess *session, ac *attachedClient, oldTr *closeTrackingTransport, token uint64)
872 }{
873 {
874 name: "explicit_detach",
875 win: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient, oldTr *closeTrackingTransport, token uint64) {
876 t.Helper()
877 d.clientGone(sess, ac, oldTr, true)
878 require.Empty(t, sess.snapshotAttachments(), "explicit detach must clear the live owner before late mark")
879 },
880 },
881 {
882 name: "session_removal",
883 win: func(t *testing.T, d *Daemon, sess *session, ac *attachedClient, oldTr *closeTrackingTransport, token uint64) {
884 t.Helper()
885 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
886 d.mu.Lock()
887 registered := d.sessions[sess.id] == sess
888 d.mu.Unlock()
889 require.False(t, registered, "killSession must unregister before late mark")
890 },
891 },
892 }
893 for _, tc := range cases {
894 t.Run(tc.name, func(t *testing.T) {
895 pty, release := newBlockingPTY(t)
896 defer release()
897 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
898
899 oldTr := &closeTrackingTransport{}
900 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
901 require.NoError(t, err)
902 token := ac.resumeToken
903 require.NotZero(t, token)
904 require.Contains(t, sess.snapshotAttachments(), ac)
905
906 validated := make(chan struct{})
907 releaseMark := make(chan struct{})
908 var releaseOnce sync.Once
909 t.Cleanup(func() { releaseOnce.Do(func() { close(releaseMark) }) })
910 d.beforeMarkParkingInFlight = func() {
911 close(validated)
912 <-releaseMark
913 }
914
915 type routeResult struct {
916 err error
917 }
918 result := make(chan routeResult, 1)
919 go func() {
920 _, _, routeErr := d.route(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{})
921 result <- routeResult{err: routeErr}
922 }()
923 awaitTestCompletion(t, validated, "live resume did not pause after credential validation before mark")
924
925 tc.win(t, d, sess, ac, oldTr, token)
926
927 d.mu.Lock()
928 _, parkingBeforeRelease := d.parking[token]
929 d.mu.Unlock()
930 require.False(t, parkingBeforeRelease, "terminal cleanup must not leave a parking marker before late mark resumes")
931
932 releaseOnce.Do(func() { close(releaseMark) })
933 got := awaitTestValue(t, result, "live IntentResume stayed blocked after terminal cleanup won before mark")
934 require.Error(t, got.err)
935 var pe *protoErr
936 require.ErrorAs(t, got.err, &pe)
937 require.Equal(t, ports.ErrNoSuchSession, pe.code)
938 require.Contains(t, pe.Error(), "resume token is no longer valid")
939
940 d.mu.Lock()
941 _, stillParking := d.parking[token]
942 _, stillParked := d.parked[token]
943 d.mu.Unlock()
944 require.False(t, stillParking, "late markParkingInFlight must not recreate a parking marker after terminal cleanup")
945 require.False(t, stillParked, "terminal cleanup must not leave a parked credential")
946 })
947 }
948
949 t.Run("direct_park_after_detach_without_marker", func(t *testing.T) {
950 pty, release := newBlockingPTY(t)
951 defer release()
952 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
953
954 oldTr := &closeTrackingTransport{}
955 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
956 require.NoError(t, err)
957 token := ac.resumeToken
958 require.NotZero(t, token)
959
960 require.True(t, d.detachIfCurrentTransport(sess, ac, ac.transportSnapshot()))
961 require.Empty(t, sess.snapshotAttachments())
962 require.Zero(t, d.markParkingInFlight(sess, ac), "detached active attachment must not publish a late marker")
963
964 d.mu.Lock()
965 _, parkingBeforePark := d.parking[token]
966 d.mu.Unlock()
967 require.False(t, parkingBeforePark)
968
969 require.True(t, d.parkAttachment(sess, ac), "direct parkAttachment after detach must still park without recreating an in-flight marker")
970 d.mu.Lock()
971 parked := d.parked[token]
972 _, stillParking := d.parking[token]
973 d.mu.Unlock()
974 require.NotNil(t, parked)
975 require.Same(t, ac, parked.ac)
976 require.False(t, stillParking, "park must not leave an in-flight marker")
977 })
978}
979
980// TestMarkParkingInFlightRequiresExactLiveOwnership covers lock-safe rejection
981// when the attachment is no longer the exact registered owner, while
982// still advertising for both live roles.
983func TestStaleClientGoneAfterTransportCheckDoesNotDetachReboundAttachment(t *testing.T) {
984 pty, release := newBlockingPTY(t)
985 defer release()
986 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
987
988 oldTr := &closeTrackingTransport{}
989 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
990 require.NoError(t, err)
991 token := ac.resumeToken
992 require.NotZero(t, token)
993
994 passedPrecheck := make(chan struct{})
995 releaseDetach := make(chan struct{})
996 var releaseOnce sync.Once
997 t.Cleanup(func() { releaseOnce.Do(func() { close(releaseDetach) }) })
998 d.beforeClientGoneDetach = func() {
999 close(passedPrecheck)
1000 <-releaseDetach
1001 }
1002
1003 goneDone := make(chan struct{})
1004 go func() {
1005 d.clientGone(sess, ac, oldTr, false)
1006 close(goneDone)
1007 }()
1008 awaitTestCompletion(t, passedPrecheck, "stale clientGone did not reach the post-precheck seam")
1009
1010 newTr := &closeTrackingTransport{}
1011 resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), newTr)
1012 require.NoError(t, err)
1013 require.Same(t, sess, resumedSess)
1014 require.Same(t, ac, resumedAC)
1015 require.Same(t, newTr, ac.transport())
1016
1017 releaseOnce.Do(func() { close(releaseDetach) })
1018 awaitTestCompletion(t, goneDone, "stale clientGone did not finish after resume rebound")
1019
1020 require.Contains(t, sess.snapshotAttachments(), ac, "stale cleanup must not detach the rebound attachment")
1021 require.Same(t, newTr, ac.transport())
1022 require.False(t, newTr.Closed(), "rebound transport must survive stale old-link cleanup")
1023 require.True(t, oldTr.Closed(), "live resume retires the captured old transport")
1024}
1025
1026// TestResumeLiveAttachmentParkFailureRetiresOldTransport covers shutdown racing
1027// park after live detach: the old link must still be revoked/closed exactly once
1028// and the session must not keep a half-detached owner.
1029func TestResumeLiveAttachmentParkFailureRetiresOldTransport(t *testing.T) {
1030 pty, release := newBlockingPTY(t)
1031 defer release()
1032 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
1033
1034 oldTr := &closeTrackingTransport{}
1035 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
1036 require.NoError(t, err)
1037 token := ac.resumeToken
1038 require.NotZero(t, token)
1039
1040 d.afterDetachAttachmentEffectsFrozen = func() {
1041 d.mu.Lock()
1042 d.closing = true
1043 d.mu.Unlock()
1044 }
1045
1046 _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{})
1047 require.Error(t, err)
1048 var pe *protoErr
1049 require.ErrorAs(t, err, &pe)
1050 require.Equal(t, ports.ErrServerShutdown, pe.code)
1051 require.Empty(t, sess.snapshotAttachments(), "failed live park must leave the session without an owner")
1052 require.Nil(t, ac.transport(), "failed live park must revoke the captured old transport")
1053 require.True(t, oldTr.Closed(), "failed live park must close the captured old transport")
1054 d.mu.Lock()
1055 _, parked := d.parked[token]
1056 d.mu.Unlock()
1057 require.False(t, parked, "failed live park must not publish a resume credential")
1058}
1059
1060func TestResumeRebindsRotatesAndDoesNotOpenPTY(t *testing.T) {
1061 pty, release := newBlockingPTY(t)
1062 defer release()
1063 d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
1064
1065 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1066 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1067 require.NoError(t, err)
1068 oldToken := ac.resumeToken
1069 d.clientGone(sess, ac, ac.transport(), false)
1070
1071 tr2, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1072 resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", oldToken), tr2)
1073 require.NoError(t, err)
1074 require.Same(t, sess, resumedSess)
1075 require.Same(t, ac, resumedAC)
1076 require.NotEqual(t, oldToken, resumedAC.resumeToken, "resume rotates token")
1077 require.Contains(t, sess.snapshotAttachments(), resumedAC)
1078}
1079
1080func TestOutputAckLagAloneDoesNotForceFullStateRepaint(t *testing.T) {
1081 ac := &attachedClient{output: newOutputStateStream()}
1082 ac.output.next = 3
1083 epoch := ac.output.currentEpoch()
1084 ac.ackOutputState(epoch, 3)
1085 ac.ackOutputState(epoch, 2)
1086 ac.ackOutputState(epoch, 4)
1087 require.Equal(t, uint64(3), ac.output.acked, "stale or future ACKs must not move output state incorrectly")
1088
1089 ac.sendMu.Lock()
1090 ac.output.next = 5
1091 reset := false
1092 require.False(t, reset, "reliable output ack lag alone must not force dependency-free full repaint")
1093
1094 f := outputStateFrame(ac.output, []byte("incremental while reliable backlog drains"), reset, 0)
1095 ac.sendMu.Unlock()
1096 out, err := ports.UnmarshalOutput(f.Payload)
1097 require.NoError(t, err)
1098 require.Equal(t, uint64(5), out.Base, "output should remain incremental unless an explicit reset is requested")
1099 require.Equal(t, uint64(6), out.New)
1100
1101 ac.sendMu.Lock()
1102 reset = true
1103 require.True(t, reset, "explicit reset should still force full repaint")
1104 full := outputStateFrame(ac.output, []byte("explicit full repaint"), reset, 0)
1105 ac.sendMu.Unlock()
1106 fullOut, err := ports.UnmarshalOutput(full.Payload)
1107 require.NoError(t, err)
1108 require.Equal(t, uint64(0), fullOut.Base)
1109 require.Equal(t, uint64(7), fullOut.New)
1110}
1111
1112func TestResumeClientIDMismatchDoesNotConsumeParkedToken(t *testing.T) {
1113 pty, release := newBlockingPTY(t)
1114 defer release()
1115 d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
1116
1117 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1118 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1119 require.NoError(t, err)
1120 token := ac.resumeToken
1121 d.clientGone(sess, ac, ac.transport(), false)
1122
1123 wrongClient := helloResumeCapable(ports.IntentResume, "work", token)
1124 wrongClient.ClientID = [16]byte{9, 9, 9, 9}
1125 _, _, ok, err := d.resumeParked(wrongClient, &closeTrackingTransport{}, domain.Size{Cols: 80, Rows: 24})
1126 d.mu.Lock()
1127 _, stillParked := d.parked[token]
1128 d.mu.Unlock()
1129 require.Error(t, err)
1130 require.False(t, ok)
1131 require.True(t, stillParked, "mismatched client must not consume parked token")
1132 require.Equal(t, token, ac.resumeToken)
1133 require.True(t, ac.parked)
1134
1135 tr2, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1136 resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), tr2)
1137 require.NoError(t, err)
1138 require.Same(t, sess, resumedSess)
1139 require.Same(t, ac, resumedAC)
1140}
1141
1142func TestResumeCloseCapturedOldTransportDoesNotCloseReboundTransport(t *testing.T) {
1143 pty, release := newBlockingPTY(t)
1144 defer release()
1145 d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
1146
1147 oldTr := &closeTrackingTransport{}
1148 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
1149 require.NoError(t, err)
1150 token := ac.resumeToken
1151 require.True(t, sess.detachIfCurrent(ac))
1152 require.True(t, d.parkAttachment(sess, ac))
1153 generation := ac.connectionGeneration.Load()
1154
1155 newTr := &closeTrackingTransport{}
1156 resumedSess, resumedAC, ok, err := d.resumeParked(helloResumeCapable(ports.IntentResume, "work", token), newTr, domain.Size{Cols: 80, Rows: 24})
1157 require.NoError(t, err)
1158 require.True(t, ok)
1159 require.Same(t, sess, resumedSess)
1160 require.Same(t, ac, resumedAC)
1161 require.Greater(t, ac.connectionGeneration.Load(), generation, "resume must publish active ownership through the attachment transition")
1162
1163 _ = ac.closeCapturedTransport(oldTr)
1164 require.True(t, oldTr.Closed(), "old transport is closed")
1165 require.False(t, newTr.Closed(), "newly rebound transport is not closed by old cleanup")
1166 require.Same(t, newTr, ac.transport())
1167}
1168
1169func TestResumeRebasesFullOutputWindowBeforeFirstPaint(t *testing.T) {
1170 pty, release := newBlockingPTY(t)
1171 defer release()
1172 d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
1173 oldTr := &closeTrackingTransport{}
1174 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
1175 require.NoError(t, err)
1176 ac.output.next = maxUnackedOutputStates
1177 token := ac.resumeToken
1178 require.True(t, sess.detachIfCurrent(ac))
1179 require.True(t, d.parkAttachment(sess, ac))
1180
1181 newTr := &closeTrackingTransport{}
1182 resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), newTr)
1183 require.NoError(t, err)
1184 require.Same(t, ac, resumedAC)
1185 d.paint(resumedSess, resumedAC, true, nil)
1186
1187 sends := testFramesOfType(newTr.Sends(), ports.MsgOutput)
1188 require.Len(t, sends, 1)
1189 first, err := ports.UnmarshalOutput(sends[0].Payload)
1190 require.NoError(t, err)
1191 require.Zero(t, first.Base)
1192 require.Equal(t, uint64(1), first.New)
1193 resumedAC.ackOutputState(first.Epoch, first.New)
1194
1195 resumedSess.tabs[0].focusedPane().screen.Write([]byte("A"))
1196 d.paint(resumedSess, resumedAC, false, nil)
1197 sends = testFramesOfType(newTr.Sends(), ports.MsgOutput)
1198 require.Len(t, sends, 2)
1199 second, err := ports.UnmarshalOutput(sends[1].Payload)
1200 require.NoError(t, err)
1201 require.Equal(t, first.New, second.Base)
1202}
1203
1204func TestParkingReleasesPaneCapturesBeforeHeadlessCloseAndResume(t *testing.T) {
1205 pty, release := newBlockingPTY(t)
1206 defer release()
1207 d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
1208 oldTransport := &closeTrackingTransport{}
1209 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport)
1210 require.NoError(t, err)
1211
1212 tb := sess.tabs[0]
1213 survivor := tb.panes["pane-1"]
1214 closed := newPane("pane-2", nil, domain.Size{Cols: 40, Rows: 23})
1215 tb.mu.Lock()
1216 tb.tree = &layout.Tree{Root: &layout.Node{Kind: layout.Split, Dir: layout.Horizontal, Children: []*layout.Node{layout.NewLeaf("pane-1"), layout.NewLeaf("pane-2")}}, Focus: "pane-1"}
1217 tb.panes[closed.id] = closed
1218 tb.bumpLayoutGenerationLocked()
1219 tb.mu.Unlock()
1220 d.applyTabLayout(sess, tb)
1221 survivor.screen.Write([]byte("survivor"))
1222 closed.screen.Write([]byte("closed"))
1223 d.paint(sess, ac, true, nil)
1224
1225 ac.sendMu.Lock()
1226 require.Contains(t, ac.captureFrames, closed, "fixture must render and capture the pane before parking")
1227 ac.sendMu.Unlock()
1228 token := ac.resumeToken
1229 require.True(t, sess.detachIfCurrent(ac))
1230 require.True(t, d.parkAttachment(sess, ac))
1231
1232 // Headless close cannot find the parked attachment through sess.snapshotAttachments()[0].
1233 // Its capture must already have been released before the attachment parked.
1234 require.NoError(t, d.closePane(sess, tb, closed.id, nil, false))
1235 ac.sendMu.Lock()
1236 require.NotContains(t, ac.captureFrames, closed, "parked attachment must not retain a pane closed while headless")
1237 ac.sendMu.Unlock()
1238
1239 newTransport := &closeTrackingTransport{}
1240 resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, sess.name, token), newTransport)
1241 require.NoError(t, err)
1242 require.Same(t, sess, resumedSess)
1243 require.Same(t, ac, resumedAC)
1244 require.True(t, resumedSess.renderCoordinator().markAttachmentReady(resumedSess.renderCoordinator().attachmentLease(resumedAC)))
1245 d.firstPaint(resumedSess, resumedAC)
1246
1247 sends := testFramesOfType(newTransport.Sends(), ports.MsgOutput)
1248 require.Len(t, sends, 1)
1249 output, err := ports.UnmarshalOutput(sends[0].Payload)
1250 require.NoError(t, err)
1251 require.Zero(t, output.Base, "resume must start with a complete frame")
1252 terminal := vt.NewScreen(resumedAC.size.Cols, resumedAC.size.Rows)
1253 terminal.Write(output.Data)
1254 contents := strings.Join(frameRows(terminal.Frame), "\n")
1255 require.Contains(t, contents, "survivor", "resume first paint must contain current headless content")
1256 require.NotContains(t, contents, "closed", "resume first paint must not contain closed pane content")
1257}
1258
1259func TestExplicitDetachDoesNotPark(t *testing.T) {
1260 pty, release := newBlockingPTY(t)
1261 defer release()
1262 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
1263 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1264 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1265 require.NoError(t, err)
1266 d.clientGone(sess, ac, ac.transport(), true)
1267 d.mu.Lock()
1268 parked := len(d.parked)
1269 d.mu.Unlock()
1270 require.Zero(t, parked)
1271}
1272
1273func TestResumeParkUsesConfiguredGraceAndExpiresOnlyAfterGrace(t *testing.T) {
1274 clk := &signalClock{timers: make(chan *signalTimer, 8)}
1275 pty, release := newBlockingPTY(t)
1276 defer release()
1277 d := newTestDaemon(t, newFactory(t, pty), clk)
1278 WithResumeParkGrace(20 * time.Minute)(d)
1279 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1280 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1281 require.NoError(t, err)
1282 token := ac.resumeToken
1283
1284 d.clientGone(sess, ac, ac.transport(), false)
1285 timer := <-clk.timers
1286 require.Equal(t, 20*time.Minute, timer.duration)
1287 d.mu.Lock()
1288 _, parkedBeforeGrace := d.parked[token]
1289 d.mu.Unlock()
1290 require.True(t, parkedBeforeGrace, "parked attachment remains before configured grace timer fires")
1291
1292 timer.ch <- clk.Now()
1293 require.Eventually(t, func() bool {
1294 d.mu.Lock()
1295 defer d.mu.Unlock()
1296 _, ok := d.parked[token]
1297 return !ok
1298 }, 2*time.Second, 10*time.Millisecond)
1299}
1300
1301func TestParkExpiryAndShutdownCleanup(t *testing.T) {
1302 clk := &signalClock{timers: make(chan *signalTimer, 8)}
1303 pty, release := newBlockingPTY(t)
1304 defer release()
1305 d := newTestDaemon(t, newFactory(t, pty), clk)
1306 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1307 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1308 require.NoError(t, err)
1309 token := ac.resumeToken
1310 d.clientGone(sess, ac, ac.transport(), false)
1311 timer := <-clk.timers
1312 timer.ch <- clk.Now()
1313 require.Eventually(t, func() bool {
1314 d.mu.Lock()
1315 defer d.mu.Unlock()
1316 _, ok := d.parked[token]
1317 return !ok
1318 }, 2*time.Second, 10*time.Millisecond)
1319
1320 pty2, release2 := newBlockingPTY(t)
1321 defer release2()
1322 d2 := newTestDaemon(t, newFactory(t, pty2), stubClock{})
1323 tr2, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1324 sess2, ac2, err := d2.route(helloResumeCapable(ports.IntentNew, "other", 0), tr2)
1325 require.NoError(t, err)
1326 d2.clientGone(sess2, ac2, ac2.transport(), false)
1327 d2.shutdownAll(ports.ReasonServerShutdown)
1328 d2.mu.Lock()
1329 parked := len(d2.parked)
1330 d2.mu.Unlock()
1331 require.Zero(t, parked)
1332}
1333
1334func TestLiveParkAndResumeRetainsAttachmentWithoutDaemonHistory(t *testing.T) {
1335 pty, release := newBlockingPTY(t)
1336 defer release()
1337 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
1338 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1339 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1340 require.NoError(t, err)
1341
1342 d.clientGone(sess, ac, ac.transport(), false)
1343
1344 token := ac.resumeToken
1345 d.mu.Lock()
1346 parked := d.parked[token]
1347 d.mu.Unlock()
1348 require.NotNil(t, parked)
1349
1350 tr2, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1351 resumedSess, resumedAC, err := d.route(helloResumeCapable(ports.IntentResume, "work", token), tr2)
1352 require.NoError(t, err)
1353 require.Same(t, sess, resumedSess)
1354 require.Same(t, ac, resumedAC)
1355}
1356
1357func TestDiscardingParkedAttachmentDoesNotOwnPreviousSession(t *testing.T) {
1358 for _, tc := range []struct {
1359 name string
1360 discard func(*Daemon, uint64, *parkedAttachment) []parkedAttachmentRetirement
1361 }{
1362 {
1363 name: "expiry",
1364 discard: func(d *Daemon, token uint64, parked *parkedAttachment) []parkedAttachmentRetirement {
1365 d.removeParkedLocked(token, parked)
1366 return nil
1367 },
1368 },
1369 {
1370 name: "session purge",
1371 discard: func(d *Daemon, _ uint64, parked *parkedAttachment) []parkedAttachmentRetirement {
1372 return d.purgeParkedForSessionLocked(parked.sess)
1373 },
1374 },
1375 } {
1376 t.Run(tc.name, func(t *testing.T) {
1377 pty, release := newBlockingPTY(t)
1378 defer release()
1379 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
1380 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1381 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1382 require.NoError(t, err)
1383
1384 d.clientGone(sess, ac, ac.transport(), false)
1385
1386 d.mu.Lock()
1387 parked := d.parked[ac.resumeToken]
1388 require.NotNil(t, parked)
1389 retirements := tc.discard(d, ac.resumeToken, parked)
1390 d.mu.Unlock()
1391 d.finishParkedAttachmentRetirements(retirements)
1392
1393 })
1394 }
1395}
1396
1397func TestEphemeralParkExpiryKeepsSession(t *testing.T) {
1398 clk := &signalClock{timers: make(chan *signalTimer, 8)}
1399 pty, release := newBlockingPTY(t)
1400 defer release()
1401 d := newTestDaemon(t, newFactory(t, pty), clk)
1402
1403 tr := &closeTrackingTransport{}
1404 sess, ac, err := d.route(helloResumeCapable(ports.IntentEphemeral, "", 0), tr)
1405 require.NoError(t, err)
1406 token := ac.resumeToken
1407 require.NotZero(t, token)
1408
1409 d.clientGone(sess, ac, ac.transport(), false)
1410 timer := <-clk.timers
1411 timer.ch <- clk.Now()
1412
1413 require.Eventually(t, func() bool {
1414 d.mu.Lock()
1415 defer d.mu.Unlock()
1416 _, ok := d.parked[token]
1417 return !ok
1418 }, 2*time.Second, 10*time.Millisecond)
1419
1420 require.Equal(t, 1, sessionCount(d), "token expiry does not kill ephemeral session")
1421 sess.mu.Lock()
1422 require.Empty(t, sess.snapshotAttachmentsLocked())
1423 sess.mu.Unlock()
1424}
1425
1426func TestKilledSessionPurgesParkedResumeToken(t *testing.T) {
1427 pty, release := newBlockingPTY(t)
1428 defer release()
1429 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
1430 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1431 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1432 require.NoError(t, err)
1433 token := ac.resumeToken
1434 d.clientGone(sess, ac, ac.transport(), false)
1435
1436 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
1437 d.mu.Lock()
1438 _, parked := d.parked[token]
1439 d.mu.Unlock()
1440 _, _, ok, err := d.resumeParked(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{}, domain.Size{Cols: 80, Rows: 24})
1441 require.False(t, parked, "killSession purges parked token")
1442 require.NoError(t, err)
1443 require.False(t, ok, "killed session cannot be resumed")
1444}
1445
1446// TestKilledSessionUnblocksParkingWaiterFailsClosed covers session removal while
1447// a same-token IntentResume is waiting on an in-flight parking marker. killSession
1448// must purge only that session's markers so the waiter unblocks and fails closed,
1449// without touching another session's parking marker.
1450func TestKilledSessionUnblocksParkingWaiterFailsClosed(t *testing.T) {
1451 workPTY, releaseWork := newBlockingPTY(t)
1452 defer releaseWork()
1453 otherPTY, releaseOther := newBlockingPTY(t)
1454 defer releaseOther()
1455 // Separate PTYs are required: a shared mock PTY makes killSession(work)
1456 // close other's pane and cascade-remove the other session under -race,
1457 // falsely failing the other-marker / fail-closed assertions.
1458 d := newTestDaemon(t, newFactorySeq(t, workPTY, otherPTY), stubClock{})
1459
1460 oldTr := &closeTrackingTransport{}
1461 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
1462 require.NoError(t, err)
1463 token := ac.resumeToken
1464 require.NotZero(t, token)
1465
1466 otherTr := &closeTrackingTransport{}
1467 otherSess, otherAC, err := d.route(helloResumeCapable(ports.IntentNew, "other", 0), otherTr)
1468 require.NoError(t, err)
1469 otherToken := d.markParkingInFlight(otherSess, otherAC)
1470 require.NotZero(t, otherToken)
1471 require.NotEqual(t, token, otherToken)
1472
1473 require.Equal(t, token, d.markParkingInFlight(sess, ac))
1474 d.mu.Lock()
1475 parkingMarker := d.parking[token]
1476 otherMarker := d.parking[otherToken]
1477 d.mu.Unlock()
1478 require.NotNil(t, parkingMarker, "fixture: parking marker must be published before kill")
1479 require.Same(t, ac, parkingMarker.ac)
1480 require.NotNil(t, otherMarker, "fixture: other session marker must remain published")
1481 require.Same(t, otherAC, otherMarker.ac)
1482
1483 waiterArmed := make(chan struct{})
1484 d.afterParkingWaitArmed = func() { close(waiterArmed) }
1485 waiterDone := make(chan bool, 1)
1486 go func() {
1487 waiterDone <- d.waitParkingInFlight(helloResumeCapable(ports.IntentResume, "work", token))
1488 }()
1489 awaitTestCompletion(t, waiterArmed, "parking waiter did not arm before session kill")
1490
1491 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
1492
1493 waited := awaitTestValue(t, waiterDone, "parking waiter stayed blocked after session kill")
1494 require.True(t, waited, "waiter must observe the session marker before it is purged")
1495
1496 d.mu.Lock()
1497 _, stillParking := d.parking[token]
1498 otherAfter := d.parking[otherToken]
1499 otherRegistered := d.sessions[otherSess.id] == otherSess
1500 closing := d.closing
1501 d.mu.Unlock()
1502 require.False(t, stillParking, "killSession must purge the killed session's parking marker")
1503 require.NotNil(t, otherAfter, "killSession must not purge another session's parking marker")
1504 require.Same(t, otherAC, otherAfter.ac)
1505 require.True(t, otherRegistered, "other session must survive work kill when PTYs are isolated")
1506 require.False(t, closing, "daemon must not shut down while the other session remains")
1507
1508 _, _, err = d.route(helloResumeCapable(ports.IntentResume, "work", token), &closeTrackingTransport{})
1509 require.Error(t, err)
1510 var killedErr *protoErr
1511 require.ErrorAs(t, err, &killedErr)
1512 require.Equal(t, ports.ErrNoSuchSession, killedErr.code)
1513 require.Contains(t, killedErr.Error(), "resume token is no longer valid")
1514}
1515
1516func TestStaleClientGoneDoesNotDetachOrCloseFreshTransport(t *testing.T) {
1517 p, release := newBlockingPTY(t)
1518 defer release()
1519 d := newTestDaemon(t, newFactory(t, p), stubClock{})
1520
1521 oldTr := &closeTrackingTransport{}
1522 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTr)
1523 require.NoError(t, err)
1524 freshTr := &closeTrackingTransport{}
1525 ac.replaceTransport(freshTr)
1526
1527 d.clientGone(sess, ac, oldTr, false)
1528
1529 require.Contains(t, sess.snapshotAttachments(), ac, "stale connection must not detach current client")
1530 require.False(t, oldTr.Closed(), "stale transport is owned by its own loop/handler")
1531 require.False(t, freshTr.Closed(), "fresh resumed transport must not be closed by stale loop")
1532}
1533
1534func TestRawTerminalSideEffectsAreOutputStateNeutral(t *testing.T) {
1535 d := newTestDaemon(t, nil, stubClock{})
1536 tr := &closeTrackingTransport{}
1537 ac := &attachedClient{tr: tr, output: newOutputStateStream()}
1538 ac.initOverlays()
1539
1540 require.NoError(t, d.boundedSendOutputErr(ac, []byte("copy")))
1541 require.NoError(t, d.boundedSendOutputErr(ac, []byte("more")))
1542
1543 sends := tr.Sends()
1544 require.Len(t, sends, 2)
1545 first, err := ports.UnmarshalOutput(sends[0].Payload)
1546 require.NoError(t, err)
1547 second, err := ports.UnmarshalOutput(sends[1].Payload)
1548 require.NoError(t, err)
1549 require.Zero(t, first.Base)
1550 require.Zero(t, first.New)
1551 require.Zero(t, second.Base)
1552 require.Zero(t, second.New)
1553}
1554
1555func TestSequencedInputDoesNotPrematurelyEcho(t *testing.T) {
1556 p, release := newBlockingPTY(t)
1557 defer release()
1558 d := newTestDaemon(t, newFactory(t, p), stubClock{})
1559 tr := &closeTrackingTransport{}
1560 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), tr)
1561 require.NoError(t, err)
1562
1563 d.handleSequencedInput(sess, ac, 42, []byte("x"))
1564
1565 require.Zero(t, ac.echoAck.Load())
1566}
1567
1568func TestResumeParkedReplacesFuturePTYEnvironment(t *testing.T) {
1569 initialPTY, releaseInitial := newBlockingPTY(t)
1570 futurePTY, releaseFuture := newBlockingPTY(t)
1571 defer releaseInitial()
1572 defer releaseFuture()
1573 var commands []string
1574 var envs [][]string
1575 factory := portsmocks.NewMockPTYFactory(t)
1576 factory.EXPECT().Open(mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).RunAndReturn(
1577 func(_ context.Context, command string, _ []string, env []string, _ string, size domain.Size) (ports.PTY, error) {
1578 if size != (domain.Size{Cols: 80, Rows: 22}) {
1579 return newQuietPTY(), nil
1580 }
1581 commands = append(commands, command)
1582 envs = append(envs, append([]string(nil), env...))
1583 if len(commands) == 1 {
1584 return initialPTY, nil
1585 }
1586 return futurePTY, nil
1587 },
1588 ).Maybe()
1589 d := newTestDaemon(t, factory, stubClock{})
1590 tr, _, _ := newConn(t, mustHello(ports.IntentAttach, "unused", domain.Size{}))
1591 hello := helloResumeCapable(ports.IntentNew, "work", 0)
1592 hello.Env = []string{"SECRET=before", "SHELL=/bin/sh", "TERM=old", "VEV=old"}
1593 sess, ac, err := d.route(hello, tr)
1594 require.NoError(t, err)
1595 token := ac.resumeToken
1596 require.True(t, sess.detachIfCurrent(ac))
1597 require.True(t, d.parkAttachment(sess, ac))
1598
1599 resumeHello := helloResumeCapable(ports.IntentResume, "work", token)
1600 resumeHello.Env = []string{"SECRET=after", "PAIR=a=b", "SHELL=/usr/bin/fish", "TERM=old", "COLORTERM=old", "TERM_PROGRAM=old", "VEV=old"}
1601 resumeHello.TrueColor = true
1602 _, _, ok, err := d.resumeParked(resumeHello, &closeTrackingTransport{}, domain.Size{Cols: 80, Rows: 24})
1603 require.NoError(t, err)
1604 require.True(t, ok)
1605 require.NoError(t, d.createTab(sess, domain.Size{Cols: 80, Rows: 24}))
1606
1607 sess.mu.Lock()
1608 future := sess.tabs[1].focusedPane()
1609 futureTabID, futurePaneID := sess.tabs[1].stableID, future.stableID
1610 require.True(t, sess.terminal.TrueColor)
1611 require.Equal(t, resumeHello.Env, sess.env)
1612 sess.mu.Unlock()
1613 require.Equal(t, []string{"/bin/sh", "/usr/bin/fish"}, commands)
1614 require.Equal(t, []string{"SECRET=after", "PAIR=a=b", "SHELL=/usr/bin/fish", "TERM=xterm-direct", "COLORTERM=truecolor", "TERM_PROGRAM=vev", "VEV=session=work,tab=" + futureTabID + ",pane=" + futurePaneID}, envs[1])
1615}
1616
1617func TestResumeParkedDaemonOwnedPreservesSessionEnvironment(t *testing.T) {
1618 pty, release := newBlockingPTY(t)
1619 defer release()
1620 d := newTestDaemon(t, newFactory(t, pty), stubClock{})
1621 d.baseEnv = []string{"DAEMON=owned"}
1622 oldTransport := &closeTrackingTransport{}
1623 sess, ac, err := d.route(helloResumeCapable(ports.IntentNew, "work", 0), oldTransport)
1624 require.NoError(t, err)
1625 sess.mu.Lock()
1626 sess.env = copyEnvironment(d.baseEnv)
1627 sess.mu.Unlock()
1628 token := ac.resumeToken
1629 require.True(t, sess.detachIfCurrent(ac))
1630 require.True(t, d.parkAttachment(sess, ac))
1631
1632 resume := helloResumeCapable(ports.IntentResume, "work", token)
1633 resume.EnvironmentPolicy = ports.EnvironmentPolicyDaemonOwned
1634 resume.Env = []string{"UNTRUSTED=client"}
1635 _, _, ok, err := d.resumeParked(resume, &closeTrackingTransport{}, defaultSize)
1636 require.NoError(t, err)
1637 require.True(t, ok)
1638 sess.mu.Lock()
1639 require.Equal(t, []string{"DAEMON=owned"}, sess.env)
1640 sess.mu.Unlock()
1641
1642 require.NoError(t, d.killSession(sess, ports.ReasonSessionKilled, true))
1643}
1644
1645func TestResumeRenegotiatesOutputWindowOnReusedStream(t *testing.T) {
1646 pty, release := newBlockingPTY(t)
1647 defer release()
1648 d := newTestDaemon(t, newFactorySeq(t, pty), stubClock{})
1649 firstHello := helloResumeCapable(ports.IntentNew, "work", 0)
1650 firstHello.MaxOutputInFlight = 8
1651 sess, ac, err := d.route(firstHello, &closeTrackingTransport{})
1652 require.NoError(t, err)
1653 require.Equal(t, uint64(8), ac.output.maxOutstanding)
1654 stream := ac.output
1655 token := ac.resumeToken
1656 require.True(t, sess.detachIfCurrent(ac))
1657 require.True(t, d.parkAttachment(sess, ac))
1658
1659 resumeOne := helloResumeCapable(ports.IntentResume, "work", token)
1660 resumeOne.MaxOutputInFlight = 1
1661 _, resumed, err := d.route(resumeOne, &closeTrackingTransport{})
1662 require.NoError(t, err)
1663 require.Same(t, stream, resumed.output)
1664 require.Equal(t, uint64(1), resumed.output.maxOutstanding)
1665
1666 token = resumed.resumeToken
1667 require.True(t, sess.detachIfCurrent(resumed))
1668 require.True(t, d.parkAttachment(sess, resumed))
1669 resumeEight := helloResumeCapable(ports.IntentResume, "work", token)
1670 resumeEight.MaxOutputInFlight = 8
1671 _, resumed, err = d.route(resumeEight, &closeTrackingTransport{})
1672 require.NoError(t, err)
1673 require.Same(t, stream, resumed.output)
1674 require.Equal(t, uint64(8), resumed.output.maxOutstanding)
1675}