1package client
2
3import (
4 "context"
5 "sync"
6 "testing"
7 "time"
8
9 "github.com/stretchr/testify/require"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol"
14)
15
16func TestP54LifecycleActionsAreDistinct(t *testing.T) {
17 for _, tc := range []struct {
18 name string
19 action AttachmentLifecycleActionKind
20 wantNotice LifecycleNoticeKind
21 wantExit bool
22 }{
23 {name: "detach to picker", action: AttachmentDetachToPicker, wantNotice: LifecycleNoticeDetachToPicker},
24 {name: "detach and exit", action: AttachmentDetachAndExit, wantNotice: LifecycleNoticeDetachAndExit, wantExit: true},
25 } {
26 t.Run(tc.name, func(t *testing.T) {
27 actions := make(chan AttachmentLifecycleAction, 1)
28 notices := make(chan LifecycleNotice, 4)
29 picker := newAttachTestPicker()
30 harness := startAttachHarnessConfig(t, picker, func(cfg *SupervisorConfig) {
31 cfg.LifecycleActions = actions
32 cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
33 })
34 stream := newSessionTestStream()
35 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
36 return stream, nil
37 })
38 picker.commit(sessionTestRequest(true))
39 awaitStreamHello(t, &sync.Mutex{}, &stream)
40 deliverReadyStream(t, stream)
41 awaitAttachedState(t, harness.sup)
42 actions <- AttachmentLifecycleAction{Token: authorityToken(&harness.sup.attachments.authority), Kind: tc.action}
43 require.Equal(t, tc.wantNotice, (<-notices).Kind)
44 if tc.wantExit {
45 require.NoError(t, harness.waitRun(t))
46 } else {
47 awaitPickerState(t, harness.sup)
48 require.True(t, picker.owns())
49 require.Len(t, harness.service.openedRequests(), 1, "detach must not auto-reattach")
50 }
51 })
52 }
53}
54
55func TestP54AttachmentOutcomesReturnToPicker(t *testing.T) {
56 for _, tc := range []struct {
57 name string
58 settle func(*sessionTestStream)
59 notice LifecycleNoticeKind
60 }{
61 // ReasonDetach (the explicit user detach) is intentionally excluded here:
62 // TestSupervisorDetachedReasonDeterminesExitOrPicker pins that it exits
63 // the client instead of returning to the picker.
64 {name: "server shutdown", settle: func(s *sessionTestStream) { s.deliver(protocol.Detached{Reason: protocol.ReasonServerShutdown}) }, notice: LifecycleNoticeSessionEnded},
65 {name: "session deletion", settle: func(s *sessionTestStream) { s.deliver(protocol.Detached{Reason: protocol.ReasonSessionKilled}) }, notice: LifecycleNoticeSessionEnded},
66 // A stream loss after attachment resumes the same session instead:
67 // TestSupervisorAttachmentLossResumesSameSession pins that path.
68 } {
69 t.Run(tc.name, func(t *testing.T) {
70 picker := newAttachTestPicker()
71 notices := make(chan LifecycleNotice, 4)
72 harness := startAttachHarness(t, picker)
73 harness.sup.cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
74 stream := newSessionTestStream()
75 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
76 return stream, nil
77 })
78 picker.commit(sessionTestRequest(true))
79 require.Eventually(t, func() bool { return len(stream.messages()) > 0 }, 5*time.Second, time.Millisecond)
80 deliverReadyStream(t, stream)
81 awaitAttachedState(t, harness.sup)
82 tc.settle(stream)
83 awaitPickerState(t, harness.sup)
84 require.Equal(t, tc.notice, (<-notices).Kind)
85 require.Len(t, harness.service.openedRequests(), 1)
86 })
87 }
88}
89
90func TestP54LocalRefusalIsNotReportedAsDestinationFailure(t *testing.T) {
91 picker := newAttachTestPicker()
92 notices := make(chan LifecycleNotice, 4)
93 harness := startAttachHarness(t, picker)
94 harness.sup.cfg.NotifyLifecycle = func(notice LifecycleNotice) { notices <- notice }
95
96 // The committed selection is refused locally, before any destination exists:
97 // reporting that as a destination failure would name the wrong cause.
98 picker.mu.Lock()
99 picker.resolveErr = pickerCatalogueError{Code: pickerCatalogueGone, Text: "the committed session is gone"}
100 picker.mu.Unlock()
101 picker.commit(sessionTestRequest(true))
102
103 awaitPickerState(t, harness.sup)
104 require.Equal(t, LifecycleNoticeSelectionUnavailable, (<-notices).Kind)
105 require.Empty(t, harness.service.openedRequests(), "a refused selection never dials a destination")
106}
107
108// TestP54ResolveInitialUsesRealInitialNavigationResolve proves the closed union
109// resolves the no-argument ephemeral creation through a real catalogue: the
110// resolution uses the same rules as the interactive picker and applies no
111// privileged initial semantics.
112func TestP54ResolveInitialUsesRealInitialNavigationResolve(t *testing.T) {
113 clock := newSupervisorTestClock()
114 catalogue, _ := pickerTestCatalogue(t)
115 require.True(t, catalogue.Apply(ports.BrokerSnapshot{Epoch: 3, Revision: 1, Daemons: []ports.BrokerDaemonObservation{pickerTestLocalObservation(clock.Now())}}))
116 navigation := InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
117 request, err := navigation.Resolve(catalogue.Snapshot())
118 require.NoError(t, err)
119 require.Equal(t, ports.BrokerAdmissionCreateEphemeral, request.Admission)
120 require.True(t, request.Local)
121 require.Empty(t, request.Name)
122 require.Equal(t, protocol.ExactSessionTarget{}, request.Target)
123}
124
125func TestP54InitialNavigationCreatesEphemeralThroughBroker(t *testing.T) {
126 picker := newAttachTestPicker()
127 navigation := InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
128 ctx, cancel := context.WithCancel(context.Background())
129 defer cancel()
130 clock := newSupervisorTestClock()
131 reader := newAttachTestReader()
132 terminal := &attachTestTerminal{in: reader}
133 service := newSupervisorTestService(ports.BrokerConnectionID{1})
134 service.publishSnapshot(localDaemonSnapshot(3, 1))
135 stream := newSessionTestStream()
136 service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
137 return stream, nil
138 })
139 sup := mustSupervisor(t, SupervisorConfig{Connector: newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) { return service, nil }), Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker, InitialNavigation: navigation})
140 done := make(chan error, 1)
141 go func() { done <- sup.Run(ctx) }()
142 require.Eventually(t, func() bool { return len(service.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
143 request := service.openedRequests()[0]
144 require.Equal(t, ports.BrokerAdmissionCreateEphemeral, request.Admission)
145 require.Equal(t, sessionTestRequest(true).Local, request.Local)
146 require.Zero(t, picker.resolveCount(), "initial navigation resolves through the union, never through the picker")
147 stream.deliver(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "creation failed"})
148 require.Eventually(t, func() bool { return sup.State().Presentation == PresentPicker }, 5*time.Second, time.Millisecond)
149 cancel()
150 require.ErrorIs(t, <-done, context.Canceled)
151}
152
153func TestP54InitialNavigationIsNotReplayedAfterReconnect(t *testing.T) {
154 ctx, cancel := context.WithCancel(context.Background())
155 defer cancel()
156 clock := newSupervisorTestClock()
157 reader := newAttachTestReader()
158 terminal := &attachTestTerminal{in: reader}
159 picker := newAttachTestPicker()
160 navigation := InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}
161 first := newSupervisorTestService(ports.BrokerConnectionID{1})
162 first.publishSnapshot(localDaemonSnapshot(1, 1))
163 second := newSupervisorTestService(ports.BrokerConnectionID{2})
164 stream := newSessionTestStream()
165 first.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
166 return stream, nil
167 })
168 connector := newSupervisorTestConnector(func(_ context.Context, call int) (ports.BrokerNavigator, error) {
169 if call == 1 {
170 return first, nil
171 }
172 return second, nil
173 })
174 sup := mustSupervisor(t, SupervisorConfig{Connector: connector, Terminal: terminal, Clock: clock, Jitter: func() float64 { return 0 }, Picker: picker, InitialNavigation: navigation})
175 done := make(chan error, 1)
176 go func() { done <- sup.Run(ctx) }()
177 require.Equal(t, 1, connector.awaitStart(t))
178 require.Eventually(t, func() bool { return len(first.openedRequests()) == 1 }, 5*time.Second, time.Millisecond)
179 stream.deliver(protocol.ErrorMsg{Code: protocol.ErrInternal, Text: "creation failed"})
180 require.Eventually(t, func() bool { return sup.State().Presentation == PresentPicker }, 5*time.Second, time.Millisecond)
181 first.lose(ports.BrokerError{Code: ports.BrokerErrorUnavailable})
182 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityRetryWait }, 5*time.Second, time.Millisecond)
183 for connector.callCount() < 2 {
184 clock.mu.Lock()
185 var timer *supervisorTestTimer
186 if len(clock.timers) != 0 {
187 timer = clock.timers[len(clock.timers)-1]
188 }
189 clock.mu.Unlock()
190 if timer != nil {
191 timer.fire()
192 }
193 }
194 second.publishSnapshot(localDaemonSnapshot(2, 1))
195 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
196 require.Empty(t, second.openedRequests(), "one-shot navigation must not replay on the replacement generation")
197 cancel()
198 require.ErrorIs(t, <-done, context.Canceled)
199}
200
201func TestP54SingleReaderDeliversAttachedByteAfterRealSupervisorHandoff(t *testing.T) {
202 picker := newAttachTestPicker()
203 harness := startAttachHarness(t, picker)
204 stream := newSessionTestStream()
205 harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
206 return stream, nil
207 })
208 picker.commit(sessionTestRequest(true))
209 require.Eventually(t, func() bool { return len(stream.messages()) > 0 }, 5*time.Second, time.Millisecond)
210 deliverReadyStream(t, stream)
211 awaitAttachedState(t, harness.sup)
212 harness.reader.push([]byte("x"))
213 require.Eventually(t, func() bool {
214 for _, message := range stream.messages() {
215 if input, ok := message.(protocol.Input); ok && string(input.Data) == "x" {
216 return true
217 }
218 }
219 return false
220 }, 5*time.Second, time.Millisecond, "one typed byte must survive the picker-to-attachment claim handoff")
221}
222
223func TestP54BrokerRestartDoesNotReattachOrReplayUnknownMutation(t *testing.T) {
224 ctx, cancel := context.WithCancel(context.Background())
225 defer cancel()
226 clock := newSupervisorTestClock()
227 reader := newSupervisorTestReader()
228 t.Cleanup(reader.unblock)
229 first := newSupervisorTestService(ports.BrokerConnectionID{1})
230 first.publish(1, 1)
231 second := newSupervisorTestService(ports.BrokerConnectionID{2})
232 connector := newSupervisorTestConnector(func(_ context.Context, call int) (ports.BrokerNavigator, error) {
233 if call == 1 {
234 return first, nil
235 }
236 return second, nil
237 })
238 notices := make(chan LifecycleNotice, 4)
239 sup := mustSupervisor(t, SupervisorConfig{Connector: connector, Terminal: newSupervisorTestTerminal(reader), Clock: clock, Jitter: func() float64 { return 0 }, NotifyLifecycle: func(n LifecycleNotice) { notices <- n }})
240 done := make(chan error, 1)
241 go func() { done <- sup.Run(ctx) }()
242 require.Equal(t, 1, connector.awaitStart(t))
243 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
244 first.lose(ports.BrokerError{Code: ports.BrokerErrorOutcomeUnknown, Text: "reply lost after mutation"})
245 require.Equal(t, LifecycleNoticeBrokerLost, (<-notices).Kind)
246 retry := clock.awaitTimer(t)
247 retry.fire()
248 require.Equal(t, 2, connector.awaitStart(t))
249 second.publish(2, 1)
250 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
251 require.Equal(t, LifecycleNoticeBrokerReconnected, (<-notices).Kind)
252 require.Empty(t, first.openedRequests())
253 require.Empty(t, second.openedRequests(), "restart must not auto-reattach or replay a mutation")
254 cancel()
255 require.ErrorIs(t, <-done, context.Canceled)
256}
257
258func TestP54ReadyMapPrunedAcrossClaims(t *testing.T) {
259 pump := newTerminalInputPump(nil)
260 for i := 0; i < 100; i++ {
261 consumer, ok := pump.tryClaim()
262 require.True(t, ok)
263 pump.revoke(consumer)
264 }
265 pump.readyMu.Lock()
266 defer pump.readyMu.Unlock()
267 require.Empty(t, pump.ready, "revoked monotonic claim IDs must not retain readiness channels")
268}
269
270// TestP54PickerCommittedReadIsNotReplayedToAttachment proves a picker read the
271// user already acted on is committed at the handoff and never re-delivered as
272// session input.
273func TestP54PickerCommittedReadIsNotReplayedToAttachment(t *testing.T) {
274 reader := newAttachTestReader()
275 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
276 lifetime := startTerminalInputLifetime(reader, consumer)
277 defer lifetime.stop()
278 reader.push([]byte("1\r"))
279 var committed []byte
280 require.Eventually(t, func() bool {
281 select {
282 case committed = <-consumer.reads:
283 return true
284 default:
285 return false
286 }
287 }, 5*time.Second, time.Millisecond)
288 require.Equal(t, []byte("1\r"), committed)
289 lifetime.releasePicker()
290 attachment, claimed := lifetime.pump.tryClaim()
291 require.True(t, claimed)
292 _, ok := lifetime.pump.take(context.Background(), attachment)
293 require.False(t, ok, "committed picker read must not cross the handoff")
294 lifetime.pump.revoke(attachment)
295}
296
297func TestP54AttachmentFinalizeDropsUnleasedPendingBeforePicker(t *testing.T) {
298 reader := newAttachTestReader()
299 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
300 lifetime := startTerminalInputLifetime(reader, consumer)
301 defer lifetime.stop()
302 lifetime.releasePicker()
303 host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump})
304 host.ownerBoundary = true
305 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
306 require.True(t, host.authority.grant(fg))
307 reader.push([]byte("ls\r"))
308 require.Eventually(t, func() bool {
309 lifetime.pump.mu.Lock()
310 defer lifetime.pump.mu.Unlock()
311 return lifetime.pump.pending != nil && lifetime.pump.delivering == 0
312 }, 5*time.Second, time.Millisecond)
313 fg.finish()
314 // The supervisor drops unclaimed session input as the terminal returns to
315 // the picker; finish itself stays lossless for a resume.
316 lifetime.pump.dropUnclaimed()
317 lifetime.acquirePicker()
318 assertNoPickerRead(t, consumer.reads)
319}
320
321func TestP54AttachmentFinalizeDropsPreservedResidualBeforePicker(t *testing.T) {
322 reader := newAttachTestReader()
323 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
324 lifetime := startTerminalInputLifetime(reader, consumer)
325 defer lifetime.stop()
326 lifetime.releasePicker()
327 host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump})
328 host.ownerBoundary = true
329 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
330 require.True(t, host.authority.grant(fg))
331 reader.push([]byte("\r"))
332 event, ok := fg.Input(context.Background())
333 require.True(t, ok)
334 fg.PreserveInput(event.Data)
335 fg.finish()
336 // The supervisor drops unclaimed session input as the terminal returns to
337 // the picker; finish itself stays lossless for a resume.
338 lifetime.pump.dropUnclaimed()
339 lifetime.acquirePicker()
340 assertNoPickerRead(t, consumer.reads)
341}
342
343func assertNoPickerRead(t *testing.T, reads <-chan []byte) {
344 t.Helper()
345 select {
346 case got := <-reads:
347 t.Fatalf("attachment bytes crossed into picker: %q", got)
348 case <-time.After(100 * time.Millisecond):
349 }
350}
351
352func TestP54UndecidedAttachmentReadIsDroppedBeforePicker(t *testing.T) {
353 reader := newAttachTestReader()
354 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
355 lifetime := startTerminalInputLifetime(reader, consumer)
356 defer lifetime.stop()
357 lifetime.releasePicker()
358 attachment, claimed := lifetime.pump.tryClaim()
359 require.True(t, claimed)
360 reader.push([]byte("ls\r"))
361 var delivery terminalReadResult
362 require.Eventually(t, func() bool {
363 var ok bool
364 delivery, ok = lifetime.pump.take(context.Background(), attachment)
365 return ok
366 }, 5*time.Second, time.Millisecond)
367 require.Equal(t, []byte("ls\r"), delivery.data)
368 lifetime.pump.dropOwned(attachment)
369 lifetime.pump.revoke(attachment)
370 lifetime.acquirePicker()
371 select {
372 case got := <-consumer.reads:
373 t.Fatalf("attachment bytes crossed into picker: %q", got)
374 case <-time.After(100 * time.Millisecond):
375 }
376}
377
378type recordingPickerConsumer struct{ reads chan []byte }
379
380func (c *recordingPickerConsumer) ConsumeTerminalRead(data []byte) bool {
381 c.reads <- append([]byte(nil), data...)
382 return true
383}
384
385func TestP54InputDecisionProtocol(t *testing.T) {
386 pump := newTerminalInputPump(nil)
387 <-pump.space
388 host := newAttachmentHost(attachmentHostConfig{Input: pump})
389 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
390 require.True(t, host.authority.grant(fg))
391 pump.mu.Lock()
392 pump.pending = &terminalReadResult{data: []byte("once")}
393 pump.mu.Unlock()
394 pump.signalReady(fg.consumer)
395 event, ok := fg.Input(context.Background())
396 require.True(t, ok)
397 require.Equal(t, []byte("once"), event.Data)
398 fg.PreserveInput(event.Data)
399 fg.AckInput() // refused: already decided
400 host.authority.revoke(fg)
401 next := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream())
402 require.True(t, host.authority.grant(next))
403 replayed, ok := next.Input(context.Background())
404 require.True(t, ok)
405 require.Equal(t, []byte("once"), replayed.Data)
406 next.AckInput()
407 next.PreserveInput([]byte("duplicate")) // refused: no outstanding delivery
408 require.Empty(t, pump.residual)
409}
410
411func TestP54GeometryIsPerAttachmentAndCarriesAcrossPicker(t *testing.T) {
412 resizes := make(chan domain.Geometry, 4)
413 host := newAttachmentHost(attachmentHostConfig{})
414 host.resizes = resizes
415 host.ownerBoundary = true
416 ctx, cancel := context.WithCancel(context.Background())
417 defer cancel()
418 host.startGeometry(ctx)
419 defer host.stopGeometry()
420 first := domain.Geometry{Size: domain.Size{Cols: 90, Rows: 30}}
421 latest := domain.Geometry{Size: domain.Size{Cols: 120, Rows: 40}}
422 resizes <- first
423 resizes <- latest
424 require.Eventually(t, func() bool { geometry, ok := host.latestGeometry(); return ok && geometry.Size == latest.Size }, 5*time.Second, time.Millisecond)
425 fg1 := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
426 require.True(t, host.authority.grant(fg1))
427 geometry, ok := fg1.Resize(context.Background())
428 require.True(t, ok)
429 require.Equal(t, latest.Size, geometry.Size)
430 host.authority.revoke(fg1)
431 fg2 := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream())
432 require.True(t, host.authority.grant(fg2))
433 geometry, ok = fg2.Resize(context.Background())
434 require.True(t, ok)
435 require.Equal(t, latest.Size, geometry.Size)
436 _, stale := fg1.Resize(context.Background())
437 require.False(t, stale)
438}
439
440func TestP54LifecycleSkipsStaleTokenAndDeliversLiveDetach(t *testing.T) {
441 actions := make(chan AttachmentLifecycleAction, 2)
442 host := newAttachmentHost(attachmentHostConfig{Actions: actions})
443 fg := host.newForeground(AttachmentToken{Generation: 2, Attempt: 3}, newSessionTestStream())
444 require.True(t, host.authority.grant(fg))
445 actions <- AttachmentLifecycleAction{Token: AttachmentToken{Generation: 1, Attempt: 9}, Kind: AttachmentDetachAndExit}
446 actions <- AttachmentLifecycleAction{Token: fg.token, Kind: AttachmentDetachToPicker}
447 action, ok := fg.Lifecycle(context.Background())
448 require.True(t, ok)
449 require.Equal(t, AttachmentDetachToPicker, action)
450 fg.finish()
451}
452
453func TestP54LifecycleInterleavingsRace(t *testing.T) {
454 for i := 0; i < 50; i++ {
455 pump := newTerminalInputPump(nil)
456 resizes := make(chan domain.Geometry, 1)
457 actions := make(chan AttachmentLifecycleAction, 1)
458 host := newAttachmentHost(attachmentHostConfig{Input: pump, Actions: actions})
459 host.resizes = resizes
460 ctx, cancel := context.WithCancel(context.Background())
461 host.startGeometry(ctx)
462 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
463 require.True(t, host.authority.grant(fg))
464 var wg sync.WaitGroup
465 wg.Add(3)
466 go func() { defer wg.Done(); resizes <- domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}} }()
467 go func() {
468 defer wg.Done()
469 actions <- AttachmentLifecycleAction{Token: fg.token, Kind: AttachmentDetachToPicker}
470 }()
471 go func() { defer wg.Done(); fg.finish() }()
472 wg.Wait()
473 cancel()
474 host.stopGeometry()
475 }
476}
477
478// TestResumeKeysSurviveFinish runs the production owner boundary: input kept
479// for a resume survives the attachment's finish and reaches the next
480// foreground. TestResumeHeldKeysNeverReachThePicker covers the picker side.
481func TestResumeKeysSurviveFinish(t *testing.T) {
482 reader := newAttachTestReader()
483 consumer := &recordingPickerConsumer{reads: make(chan []byte, 1)}
484 lifetime := startTerminalInputLifetime(reader, consumer)
485 defer lifetime.stop()
486 lifetime.releasePicker()
487 host := newAttachmentHost(attachmentHostConfig{Input: lifetime.pump})
488 host.ownerBoundary = true
489 fg := host.newForeground(AttachmentToken{Generation: 1}, newSessionTestStream())
490 require.True(t, host.authority.grant(fg))
491 reader.push([]byte("ls\r"))
492 event, ok := fg.Input(context.Background())
493 require.True(t, ok)
494 fg.PreserveInput(event.Data)
495 fg.finish()
496
497 next := host.newForeground(AttachmentToken{Generation: 2}, newSessionTestStream())
498 require.True(t, host.authority.grant(next))
499 replayed, ok := next.Input(context.Background())
500 require.True(t, ok)
501 require.Equal(t, []byte("ls\r"), replayed.Data)
502}