1package client
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "io"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "time"
12
13 "github.com/stretchr/testify/require"
14
15 "github.com/bnema/vev/internal/domain"
16 "github.com/bnema/vev/internal/ports"
17 "github.com/bnema/vev/internal/protocol"
18)
19
20// authorityToken reads the foreground token an attachmentAuthority currently
21// grants, or the zero token when nothing holds the slot. It exercises the same
22// foreground() accessor the supervisor uses, so tests observe the production
23// path instead of a test-only accessor.
24func authorityToken(a *attachmentAuthority) AttachmentToken {
25 fg := a.foreground()
26 if fg == nil {
27 return AttachmentToken{}
28 }
29 return fg.Token()
30}
31
32// workerTestStream is one scripted broker logical stream. It records typed
33// client messages and unblocks ReceiveServer when the supervisor closes it,
34// exactly like a real logical stream's Close contract.
35type workerTestStream struct {
36 mu sync.Mutex
37 sent []protocol.ClientMessage
38 closed chan struct{}
39 closeOnce sync.Once
40 // closeHook, when set, runs during Close. It lets a test observe the
41 // supervisor's revocation-before-close ordering.
42 closeHook func()
43}
44
45func newWorkerTestStream() *workerTestStream {
46 return &workerTestStream{closed: make(chan struct{})}
47}
48
49func (s *workerTestStream) SendClient(message protocol.ClientMessage) error {
50 select {
51 case <-s.closed:
52 return errors.New("worker test stream is closed")
53 default:
54 }
55 s.mu.Lock()
56 s.sent = append(s.sent, message)
57 s.mu.Unlock()
58 return nil
59}
60
61func (s *workerTestStream) ReceiveServer() (protocol.ServerMessage, error) {
62 <-s.closed
63 return nil, io.EOF
64}
65
66func (s *workerTestStream) Capabilities() protocol.ConnectionCapabilities {
67 return protocol.ConnectionCapabilities{}
68}
69
70func (s *workerTestStream) LinkState() ports.LinkState { return 0 }
71
72func (s *workerTestStream) LinkEvents() <-chan ports.LinkEvent { return nil }
73
74func (s *workerTestStream) Close() error {
75 s.closeOnce.Do(func() {
76 close(s.closed)
77 if s.closeHook != nil {
78 s.closeHook()
79 }
80 })
81 return nil
82}
83
84func (s *workerTestStream) Done() <-chan struct{} { return s.closed }
85
86func (s *workerTestStream) Err() error { return nil }
87
88func (s *workerTestStream) messages() []protocol.ClientMessage {
89 s.mu.Lock()
90 defer s.mu.Unlock()
91 return append([]protocol.ClientMessage(nil), s.sent...)
92}
93
94func (s *workerTestStream) isClosed() bool {
95 select {
96 case <-s.closed:
97 return true
98 default:
99 return false
100 }
101}
102
103// workerTestTerminal is the supervisor-owned terminal: it counts raw restores
104// so a test can prove the worker mechanism never restores, captures written
105// output, and doubles as the UI publication transaction.
106type workerTestTerminal struct {
107 mu sync.Mutex
108 out bytes.Buffer
109 flushes int
110 restores int
111 resizes chan domain.Geometry
112 published []ports.UIContext
113 steps []string
114 successes []bool
115 writeErr error
116 shortWrite int
117 flushErr error
118 publishErr error
119}
120
121func newWorkerTestTerminal() *workerTestTerminal {
122 return &workerTestTerminal{resizes: make(chan domain.Geometry, 4)}
123}
124
125func (t *workerTestTerminal) EnableKittyKeyboard(int) error { return nil }
126
127func (t *workerTestTerminal) EnterRaw() (func() error, error) {
128 return func() error {
129 t.mu.Lock()
130 t.restores++
131 t.mu.Unlock()
132 return nil
133 }, nil
134}
135
136func (t *workerTestTerminal) Geometry() (domain.Geometry, error) { return domain.Geometry{}, nil }
137
138func (t *workerTestTerminal) ResizeEvents() <-chan domain.Geometry { return t.resizes }
139
140func (t *workerTestTerminal) In() io.Reader { return nil }
141
142func (t *workerTestTerminal) Out() io.Writer { return &workerTestWriter{t} }
143
144func (t *workerTestTerminal) Flush() error {
145 t.mu.Lock()
146 t.flushes++
147 t.steps = append(t.steps, "flush")
148 t.mu.Unlock()
149 return t.flushErr
150}
151
152func (t *workerTestTerminal) BeginOutput(ports.UIContext) {
153 t.mu.Lock()
154 defer t.mu.Unlock()
155 t.steps = append(t.steps, "begin")
156}
157
158func (t *workerTestTerminal) EndOutput(success bool) {
159 t.mu.Lock()
160 defer t.mu.Unlock()
161 t.steps = append(t.steps, "end")
162 t.successes = append(t.successes, success)
163}
164
165func (t *workerTestTerminal) PublishContext(context ports.UIContext) error {
166 t.mu.Lock()
167 defer t.mu.Unlock()
168 t.published = append(t.published, context)
169 t.steps = append(t.steps, "publish")
170 return t.publishErr
171}
172
173func (t *workerTestTerminal) written() string {
174 t.mu.Lock()
175 defer t.mu.Unlock()
176 return t.out.String()
177}
178
179func (t *workerTestTerminal) flushCount() int {
180 t.mu.Lock()
181 defer t.mu.Unlock()
182 return t.flushes
183}
184
185func (t *workerTestTerminal) restoreCount() int {
186 t.mu.Lock()
187 defer t.mu.Unlock()
188 return t.restores
189}
190
191func (t *workerTestTerminal) publications() []ports.UIContext {
192 t.mu.Lock()
193 defer t.mu.Unlock()
194 return append([]ports.UIContext(nil), t.published...)
195}
196
197type workerTestWriter struct{ term *workerTestTerminal }
198
199func (w *workerTestWriter) Write(data []byte) (int, error) {
200 w.term.mu.Lock()
201 defer w.term.mu.Unlock()
202 w.term.steps = append(w.term.steps, "write")
203 if w.term.writeErr != nil {
204 return 0, w.term.writeErr
205 }
206 if w.term.shortWrite > 0 {
207 n := min(w.term.shortWrite, len(data))
208 _, _ = w.term.out.Write(data[:n])
209 return n, nil
210 }
211 return w.term.out.Write(data)
212}
213
214// fakeWorker adapts a function to the AttachmentWorker contract.
215type fakeWorker struct {
216 run func(ctx context.Context, fg AttachmentForeground) AttachmentEvent
217}
218
219func (w *fakeWorker) Run(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
220 return w.run(ctx, fg)
221}
222
223// blockingWorker ignores input/output and parks until its context ends.
224func blockingWorker() *fakeWorker {
225 return &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
226 <-ctx.Done()
227 return AttachmentEvent{Kind: AttachmentEventEnded}
228 }}
229}
230
231func newWorkerTestHost(term *workerTestTerminal, input *terminalInputPump, onAttached func(AttachmentToken)) *attachmentHost {
232 return newAttachmentHost(attachmentHostConfig{
233 Terminal: term,
234 Clock: newSupervisorTestClock(),
235 Input: input,
236 OnAttached: onAttached,
237 })
238}
239
240// TestAttachmentWorkerExclusiveForeground proves exactly one worker generation
241// owns the terminal foreground at a time: a second Begin is refused while the
242// first is live and succeeds once the first is cancelled.
243func TestAttachmentWorkerExclusiveForeground(t *testing.T) {
244 term := newWorkerTestTerminal()
245 host := newWorkerTestHost(term, nil, nil)
246
247 run1, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, blockingWorker(), newWorkerTestStream())
248 require.True(t, ok)
249 require.NotNil(t, run1)
250
251 _, ok = host.Begin(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, blockingWorker(), newWorkerTestStream())
252 require.False(t, ok, "second foreground must be refused while the first is live")
253 require.Equal(t, AttachmentToken{Generation: 1, Attempt: 1}, host.authority.foreground().Token())
254
255 run1.Cancel()
256 require.Nil(t, host.authority.foreground())
257
258 run2, ok := host.Begin(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, blockingWorker(), newWorkerTestStream())
259 require.True(t, ok, "a fresh generation may own the foreground after cancellation")
260 run2.Cancel()
261 require.Nil(t, host.authority.foreground())
262}
263
264// TestAttachmentWorkerZeroTokenRefused proves an unassigned token is never
265// granted, so a worker can never run without a supervisor generation.
266func TestAttachmentWorkerZeroTokenRefused(t *testing.T) {
267 host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
268 _, ok := host.Begin(context.Background(), AttachmentToken{}, blockingWorker(), newWorkerTestStream())
269 require.False(t, ok)
270 require.Nil(t, host.authority.foreground())
271}
272
273// TestAttachmentWorkerCanceledGenerationCannotAct proves cancellation revokes
274// input/output authority: a canceled generation cannot write, flush, consume or
275// forward input, resize, publish UI, or mark itself attached, and its grant is
276// closed.
277func TestAttachmentWorkerCanceledGenerationCannotAct(t *testing.T) {
278 term := newWorkerTestTerminal()
279 host := newWorkerTestHost(term, nil, nil)
280 stream := newWorkerTestStream()
281
282 fgReady := make(chan AttachmentForeground, 1)
283 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
284 fgReady <- fg
285 <-ctx.Done()
286 return AttachmentEvent{Kind: AttachmentEventEnded}
287 }}
288 run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 4, Attempt: 2}, worker, stream)
289 require.True(t, ok)
290 fg := <-fgReady
291 require.Equal(t, AttachmentToken{Generation: 4, Attempt: 2}, fg.Token())
292
293 run.Cancel()
294
295 select {
296 case <-fg.Done():
297 default:
298 t.Fatal("grant Done must be closed after cancellation")
299 }
300 require.ErrorIs(t, fg.Output(ports.UIContext{}, []byte("late")), errAttachmentForegroundRevoked)
301 require.ErrorIs(t, fg.Output(ports.UIContext{AttachmentHandle: "late"}, nil), errAttachmentForegroundRevoked)
302 require.False(t, fg.MarkAttached())
303 require.False(t, fg.Attached())
304
305 _, okInput := fg.Input(context.Background())
306 require.False(t, okInput)
307 _, okResize := fg.Resize(context.Background())
308 require.False(t, okResize)
309
310 require.Empty(t, term.written(), "a canceled generation must not write the terminal")
311 require.Empty(t, term.publications(), "a canceled generation must not publish UI")
312 require.True(t, stream.isClosed(), "the supervisor closes the owned stream")
313}
314
315// TestAttachmentWorkerConsumesAndForwardsInput proves an authorized worker
316// consumes the shared terminal pump and forwards the bytes on its own stream.
317func TestAttachmentWorkerConsumesAndForwardsInput(t *testing.T) {
318 term := newWorkerTestTerminal()
319 pr, pw := io.Pipe()
320 pump := newTerminalInputPump(pr)
321 pump.start()
322 defer pump.stop()
323 defer pw.Close()
324
325 host := newWorkerTestHost(term, pump, nil)
326 stream := newWorkerTestStream()
327
328 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
329 event, ok := fg.Input(ctx)
330 if !ok {
331 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("no authorized input")}
332 }
333 fg.AckInput()
334 if err := fg.Stream().SendClient(protocol.Input{InputSeq: 1, Data: event.Data}); err != nil {
335 return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
336 }
337 return AttachmentEvent{Kind: AttachmentEventEnded}
338 }}
339
340 go func() { _, _ = pw.Write([]byte("hello")) }()
341
342 event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
343 require.True(t, adopted)
344 require.Equal(t, AttachmentEventEnded, event.Kind)
345
346 messages := stream.messages()
347 require.Len(t, messages, 1)
348 inputMessage, ok := messages[0].(protocol.Input)
349 require.True(t, ok)
350 require.Equal(t, []byte("hello"), inputMessage.Data)
351}
352
353// TestAttachmentWorkerResizeAuthorized proves an authorized worker observes the
354// supervisor's resize events and forwards them on its stream.
355func TestAttachmentWorkerResizeAuthorized(t *testing.T) {
356 term := newWorkerTestTerminal()
357 host := newWorkerTestHost(term, nil, nil)
358 stream := newWorkerTestStream()
359 geometry := domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}
360
361 saw := make(chan domain.Geometry, 1)
362 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
363 term.resizes <- geometry
364 resize, ok := fg.Resize(ctx)
365 if !ok {
366 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("no authorized resize")}
367 }
368 saw <- resize
369 if err := fg.Stream().SendClient(protocol.Resize{Size: resize.Size, PixelWidth: resize.PixelWidth, PixelHeight: resize.PixelHeight}); err != nil {
370 return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
371 }
372 return AttachmentEvent{Kind: AttachmentEventEnded}
373 }}
374
375 event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
376 require.True(t, adopted)
377 require.Equal(t, AttachmentEventEnded, event.Kind)
378 require.Equal(t, geometry, <-saw)
379
380 messages := stream.messages()
381 require.Len(t, messages, 1)
382 _, ok := messages[0].(protocol.Resize)
383 require.True(t, ok)
384}
385
386// TestAttachmentWorkerPublishesUIAuthorized proves an authorized worker may
387// publish UI context and mark the attached transition, and that the grant's
388// generation is stamped onto the publication.
389func TestAttachmentWorkerPublishesUIAuthorized(t *testing.T) {
390 term := newWorkerTestTerminal()
391 attached := make(chan AttachmentToken, 1)
392 host := newWorkerTestHost(term, nil, func(token AttachmentToken) { attached <- token })
393 stream := newWorkerTestStream()
394
395 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
396 // The pre-attach frame commits the Connecting presentation with a public
397 // generation of zero, then the committed Attached presentation carries the
398 // granted action generation.
399 if err := fg.Output(ports.UIContext{Status: ports.UIStatusConnecting}, []byte("frame")); err != nil {
400 return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
401 }
402 if err := fg.Output(ports.UIContext{Status: ports.UIStatusAttached}, nil); err != nil {
403 return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
404 }
405 if !fg.MarkAttached() {
406 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("attach refused")}
407 }
408 if !fg.Attached() {
409 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("attach not recorded")}
410 }
411 return AttachmentEvent{Kind: AttachmentEventEnded}
412 }}
413
414 event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 7, Attempt: 2}, worker, stream)
415 require.True(t, adopted)
416 require.Equal(t, AttachmentEventEnded, event.Kind)
417
418 publications := term.publications()
419 require.Len(t, publications, 2)
420 require.Zero(t, publications[0].Generation, "a Connecting publication never carries an actionable generation")
421 require.Equal(t, ports.UIStatusConnecting, publications[0].Status)
422 require.Equal(t, uint64(7), publications[1].Generation, "the granted generation is stamped on the attached publication")
423 require.Equal(t, ports.UIStatusAttached, publications[1].Status)
424 require.Equal(t, AttachmentToken{Generation: 7, Attempt: 2}, <-attached)
425}
426
427// TestAttachmentWorkerEventReturnsToPickerWithoutTerminalRestore proves a
428// worker end, stream loss, or typed failure returns a typed event stamped with
429// the granted token while the worker mechanism never restores the terminal and
430// never closes the process.
431func TestAttachmentWorkerEventReturnsToPickerWithoutTerminalRestore(t *testing.T) {
432 cases := []struct {
433 name string
434 kind AttachmentEventKind
435 err error
436 }{
437 {name: "ended", kind: AttachmentEventEnded},
438 {name: "lost", kind: AttachmentEventLost, err: errors.New("carriage lost")},
439 {name: "failed", kind: AttachmentEventFailed, err: &ProtocolError{Code: 7, Text: "bad welcome"}},
440 }
441 for _, tc := range cases {
442 t.Run(tc.name, func(t *testing.T) {
443 term := newWorkerTestTerminal()
444 host := newWorkerTestHost(term, nil, nil)
445 stream := newWorkerTestStream()
446 // The worker reports a spoofed token; the host must override it with
447 // the granted generation/attempt.
448 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
449 return AttachmentEvent{Token: AttachmentToken{Generation: 999}, Kind: tc.kind, Err: tc.err}
450 }}
451
452 event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 3, Attempt: 5}, worker, stream)
453 require.True(t, adopted)
454 require.Equal(t, tc.kind, event.Kind)
455 require.Equal(t, AttachmentToken{Generation: 3, Attempt: 5}, event.Token)
456 if tc.err != nil {
457 require.ErrorIs(t, event.Err, tc.err)
458 } else {
459 require.NoError(t, event.Err)
460 }
461 require.Equal(t, 0, term.restoreCount(), "the worker mechanism must not restore the terminal")
462 require.True(t, stream.isClosed(), "the supervisor closes the stream after the worker settles")
463 require.Nil(t, host.authority.foreground(), "the foreground is released after the worker settles")
464 })
465 }
466}
467
468// TestAttachmentWorkerLateEventDiscardedOnCancellation proves a worker that
469// settles after the supervisor cancelled is joined and its late event is
470// discarded rather than returned as an adopted result.
471func TestAttachmentWorkerLateEventDiscardedOnCancellation(t *testing.T) {
472 term := newWorkerTestTerminal()
473 host := newWorkerTestHost(term, nil, nil)
474 stream := newWorkerTestStream()
475 run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, blockingWorker(), stream)
476 require.True(t, ok)
477
478 ctx, cancel := context.WithCancel(context.Background())
479 cancel()
480
481 event, adopted := run.Wait(ctx)
482 require.False(t, adopted)
483 require.Equal(t, AttachmentEvent{}, event, "a late event from a canceled run is discarded")
484 require.True(t, stream.isClosed())
485 require.Nil(t, host.authority.foreground())
486}
487
488// TestAttachmentWorkerPreservesResidualInputAcrossAttempts proves the reused
489// input pump keeps undecided bytes for the next authorized generation instead of
490// duplicating a reader.
491func TestAttachmentWorkerPreservesResidualInputAcrossAttempts(t *testing.T) {
492 term := newWorkerTestTerminal()
493 pr, pw := io.Pipe()
494 pump := newTerminalInputPump(pr)
495 pump.start()
496 defer pump.stop()
497 defer pw.Close()
498
499 host := newWorkerTestHost(term, pump, nil)
500
501 observed := make(chan []byte, 2)
502 first := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
503 event, ok := fg.Input(ctx)
504 if !ok {
505 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("first attempt saw no input")}
506 }
507 observed <- append([]byte(nil), event.Data...)
508 // Leave the bytes undecided: the next attempt must see them again.
509 fg.PreserveInput(event.Data)
510 return AttachmentEvent{Kind: AttachmentEventEnded}
511 }}
512 second := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
513 event, ok := fg.Input(ctx)
514 if !ok {
515 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("second attempt saw no input")}
516 }
517 fg.AckInput()
518 observed <- append([]byte(nil), event.Data...)
519 return AttachmentEvent{Kind: AttachmentEventEnded}
520 }}
521
522 go func() { _, _ = pw.Write([]byte("ab")) }()
523
524 _, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, first, newWorkerTestStream())
525 require.True(t, adopted)
526 _, adopted = host.Run(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, second, newWorkerTestStream())
527 require.True(t, adopted)
528
529 require.Equal(t, []byte("ab"), <-observed)
530 require.Equal(t, []byte("ab"), <-observed, "the replacement generation replays the preserved bytes")
531 go func() { _, _ = pw.Write([]byte("fresh")) }()
532 ctx, cancel := context.WithTimeout(context.Background(), time.Second)
533 defer cancel()
534 _, adopted = host.Run(ctx, AttachmentToken{Generation: 3, Attempt: 3}, second, newWorkerTestStream())
535 require.True(t, adopted)
536 require.Equal(t, []byte("fresh"), <-observed, "third attempt must not replay the original pending read")
537
538}
539
540// TestAttachmentWorkerRevokesBeforeClosingStream proves the supervisor revokes
541// input/output authority before it closes the owned stream: at the instant of
542// Close the foreground is already released and a late write is refused.
543func TestAttachmentWorkerRevokesBeforeClosingStream(t *testing.T) {
544 term := newWorkerTestTerminal()
545 host := newWorkerTestHost(term, nil, nil)
546 stream := newWorkerTestStream()
547
548 fgReady := make(chan AttachmentForeground, 1)
549 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
550 fgReady <- fg
551 <-ctx.Done()
552 return AttachmentEvent{Kind: AttachmentEventEnded}
553 }}
554 run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
555 require.True(t, ok)
556 fg := <-fgReady
557
558 var closeWriteErr error
559 stream.closeHook = func() {
560 require.Nil(t, host.authority.foreground(), "authority must be revoked before the stream closes")
561 closeWriteErr = fg.Output(ports.UIContext{}, []byte("late"))
562 }
563
564 run.Cancel()
565 require.ErrorIs(t, closeWriteErr, errAttachmentForegroundRevoked)
566 require.Empty(t, term.written())
567}
568
569// TestAttachmentWorkerConcurrentBeginSingleOwner proves concurrent Begin calls
570// admit exactly one foreground owner under the race detector.
571func TestAttachmentWorkerConcurrentBeginSingleOwner(t *testing.T) {
572 term := newWorkerTestTerminal()
573 host := newWorkerTestHost(term, nil, nil)
574
575 const contenders = 8
576 var wg sync.WaitGroup
577 var successes atomic.Int64
578 runs := make(chan *attachmentRun, contenders)
579 start := make(chan struct{})
580 for i := 0; i < contenders; i++ {
581 wg.Add(1)
582 go func(i int) {
583 defer wg.Done()
584 <-start
585 run, ok := host.Begin(context.Background(), AttachmentToken{Generation: uint64(i + 1), Attempt: 1}, blockingWorker(), newWorkerTestStream())
586 if ok {
587 successes.Add(1)
588 runs <- run
589 }
590 }(i)
591 }
592 close(start)
593 wg.Wait()
594 close(runs)
595
596 require.Equal(t, int64(1), successes.Load(), "exactly one concurrent Begin may own the foreground")
597 for run := range runs {
598 run.Cancel()
599 }
600 require.Nil(t, host.authority.foreground())
601}
602
603// TestAttachmentWorkerRepeatedAttemptsNoLeak proves repeated grant/cancel cycles
604// release the foreground, close every stream, and never leave a live owner.
605func TestAttachmentWorkerRepeatedAttemptsNoLeak(t *testing.T) {
606 term := newWorkerTestTerminal()
607 host := newWorkerTestHost(term, nil, nil)
608
609 const attempts = 64
610 for i := 0; i < attempts; i++ {
611 stream := newWorkerTestStream()
612 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
613 if err := fg.Output(ports.UIContext{}, []byte("x")); err != nil {
614 return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
615 }
616 return AttachmentEvent{Kind: AttachmentEventEnded}
617 }}
618 event, adopted := host.Run(context.Background(), AttachmentToken{Generation: uint64(i + 1), Attempt: 1}, worker, stream)
619 require.True(t, adopted)
620 require.Equal(t, AttachmentEventEnded, event.Kind)
621 require.True(t, stream.isClosed())
622 require.Nil(t, host.authority.foreground())
623 }
624 require.Equal(t, attempts, term.flushCount())
625 require.Equal(t, 0, term.restoreCount(), "the mechanism never enters or restores raw mode")
626}
627
628func TestAttachmentWorkerRetiredOutcomeNeverAdopted(t *testing.T) {
629 host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
630 run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
631 require.True(t, ok)
632 run.Cancel()
633 for range 3 {
634 event, adopted := run.Wait(context.Background())
635 require.False(t, adopted)
636 require.Zero(t, event)
637 }
638}
639
640func TestAttachmentWorkerBoundedJoin(t *testing.T) {
641 clock := newSupervisorTestClock()
642 pr, pw := io.Pipe()
643 defer pr.Close()
644 defer pw.Close()
645 pump := newTerminalInputPump(pr)
646 pump.start()
647 defer pump.stop()
648 host := newAttachmentHost(attachmentHostConfig{Clock: clock, Input: pump})
649 unblock := make(chan struct{})
650 defer close(unblock)
651 run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, &fakeWorker{run: func(context.Context, AttachmentForeground) AttachmentEvent {
652 <-unblock
653 return AttachmentEvent{Kind: AttachmentEventEnded}
654 }}, newWorkerTestStream())
655 require.True(t, ok)
656 cancelled := make(chan struct{})
657 go func() { run.Cancel(); close(cancelled) }()
658 timer := clock.awaitTimer(t)
659 require.Equal(t, attachmentWorkerJoinTimeout, timer.delay)
660 select {
661 case <-run.retired:
662 default:
663 t.Fatal("retirement must precede join")
664 }
665 // During the final input window the grant is finalizing, not revoked: Done is
666 // closed, output and new input are refused, the retained consumer still holds
667 // the shared reader, and the stream is still open.
668 require.Same(t, run.fg, host.authority.foreground())
669 select {
670 case <-run.fg.Done():
671 default:
672 t.Fatal("finalize must close Done before the join")
673 }
674 require.ErrorIs(t, run.fg.Output(ports.UIContext{}, []byte("late")), errAttachmentForegroundRevoked)
675 _, ok = host.Begin(context.Background(), AttachmentToken{2, 1}, blockingWorker(), newWorkerTestStream())
676 require.False(t, ok, "a finalizing grant blocks a competing Begin")
677 _, claimed := pump.tryClaim()
678 require.False(t, claimed, "the consumer is retained during the final input window")
679 require.False(t, run.stream.(*workerTestStream).isClosed(), "the stream closes after the join")
680 timer.fire()
681 select {
682 case <-cancelled:
683 case <-time.After(time.Second):
684 t.Fatal("uncooperative worker pinned cancel")
685 }
686 require.True(t, timer.stopped())
687 // After the timeout the supervisor revokes anyway: slot freed, consumer
688 // released, and the stream closed.
689 require.Nil(t, host.authority.foreground())
690 consumer, claimed := pump.tryClaim()
691 require.True(t, claimed, "revocation releases the consumer after the join")
692 pump.revoke(consumer)
693 require.True(t, run.stream.(*workerTestStream).isClosed())
694 _, adopted := run.Wait(context.Background())
695 require.False(t, adopted)
696}
697
698func TestAttachmentWorkerPumpClaimAndCancelRace(t *testing.T) {
699 pr, pw := io.Pipe()
700 defer pr.Close()
701 defer pw.Close()
702 pump := newTerminalInputPump(pr)
703 pump.start()
704 defer pump.stop()
705 host := newWorkerTestHost(newWorkerTestTerminal(), pump, nil)
706 rival, claimed := pump.tryClaim()
707 require.True(t, claimed)
708 _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
709 require.False(t, ok)
710 require.Nil(t, host.authority.foreground())
711 pump.revoke(rival)
712 for i := range 100 {
713 run, ok := host.Begin(context.Background(), AttachmentToken{uint64(i + 1), 1}, blockingWorker(), newWorkerTestStream())
714 require.True(t, ok)
715 var next *attachmentRun
716 var wg sync.WaitGroup
717 wg.Add(2)
718 go func() { defer wg.Done(); run.Cancel() }()
719 go func() {
720 defer wg.Done()
721 next, _ = host.Begin(context.Background(), AttachmentToken{uint64(i + 2), 2}, blockingWorker(), newWorkerTestStream())
722 }()
723 wg.Wait()
724 if next != nil {
725 next.Cancel()
726 }
727 consumer, claimed := pump.tryClaim()
728 require.True(t, claimed)
729 pump.revoke(consumer)
730 }
731}
732
733func TestAttachmentWorkerClosedResizeDisabled(t *testing.T) {
734 term := newWorkerTestTerminal()
735 close(term.resizes) // uiterm-like terminal: no physical resize events
736 host := newWorkerTestHost(term, nil, nil)
737 run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
738 require.True(t, ok)
739 result := make(chan bool, 1)
740 go func() { _, ok := run.fg.Resize(context.Background()); result <- ok }()
741 select {
742 case <-result:
743 t.Fatal("closed resize source terminated attachment")
744 case <-time.After(10 * time.Millisecond):
745 }
746 require.True(t, run.fg.MarkAttached())
747 run.Cancel()
748 select {
749 case ok := <-result:
750 require.False(t, ok)
751 case <-time.After(time.Second):
752 t.Fatal("authority Done did not stop resize")
753 }
754}
755
756func TestAttachmentWorkerOutputTransaction(t *testing.T) {
757 failure := errors.New("output failure")
758 for _, tc := range []struct {
759 name string
760 writeErr, flushErr, publishErr error
761 shortWrite int
762 wantErr error
763 steps []string
764 }{
765 {name: "success", steps: []string{"begin", "publish", "write", "flush", "end"}},
766 {name: "short write", shortWrite: 2, wantErr: io.ErrShortWrite, steps: []string{"begin", "publish", "write", "end"}},
767 {name: "write failure", writeErr: failure, wantErr: failure, steps: []string{"begin", "publish", "write", "end"}},
768 {name: "flush failure", flushErr: failure, wantErr: failure, steps: []string{"begin", "publish", "write", "flush", "end"}},
769 {name: "publish failure", publishErr: failure, wantErr: failure, steps: []string{"begin", "publish", "end"}},
770 {name: "publish unavailable tolerated", publishErr: ports.ErrUIUnavailable, steps: []string{"begin", "publish", "write", "flush", "end"}},
771 } {
772 t.Run(tc.name, func(t *testing.T) {
773 term := newWorkerTestTerminal()
774 term.writeErr, term.flushErr, term.publishErr, term.shortWrite = tc.writeErr, tc.flushErr, tc.publishErr, tc.shortWrite
775 host := newWorkerTestHost(term, nil, nil)
776 run, ok := host.Begin(context.Background(), AttachmentToken{9, 2}, blockingWorker(), newWorkerTestStream())
777 require.True(t, ok)
778 defer run.Cancel()
779 err := run.fg.Output(ports.UIContext{Generation: 999, Status: ports.UIStatusAttached}, []byte("frame"))
780 success := tc.wantErr == nil
781 if success {
782 require.NoError(t, err)
783 } else {
784 require.ErrorIs(t, err, tc.wantErr)
785 }
786 require.Equal(t, tc.steps, term.steps)
787 require.Equal(t, []bool{success}, term.successes)
788 require.Equal(t, uint64(9), term.published[0].Generation)
789 })
790 }
791}
792
793// Hide the transaction methods to model a physical terminal without UI output.
794type workerBareTerminal struct{ ports.Terminal }
795
796func TestAttachmentWorkerNilSeams(t *testing.T) {
797 var nilTerm *workerTestTerminal
798 var nilWorker *fakeWorker
799 var nilStream *workerTestStream
800 var nilClock *supervisorTestClock
801 for _, cfg := range []attachmentHostConfig{
802 {},
803 {Terminal: nilTerm, UI: nilTerm, Clock: nilClock},
804 } {
805 host := newAttachmentHost(cfg)
806 _, ok := host.Begin(nil, AttachmentToken{1, 1}, nilWorker, newWorkerTestStream())
807 require.False(t, ok)
808 _, ok = host.Begin(nil, AttachmentToken{1, 1}, blockingWorker(), nilStream)
809 require.False(t, ok)
810 run, ok := host.Begin(nil, AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
811 require.True(t, ok)
812 require.ErrorIs(t, run.fg.Output(ports.UIContext{}, nil), ports.ErrUIUnavailable)
813 run.Cancel()
814 require.ErrorIs(t, run.fg.Output(ports.UIContext{}, nil), errAttachmentForegroundRevoked)
815 }
816
817 term := newWorkerTestTerminal()
818 host := newAttachmentHost(attachmentHostConfig{Terminal: workerBareTerminal{term}, UI: nilTerm})
819 run, ok := host.Begin(nil, AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
820 require.True(t, ok)
821 require.NoError(t, run.fg.Output(ports.UIContext{}, []byte("frame")))
822 require.Equal(t, "frame", term.out.String())
823 run.Cancel()
824}
825
826// TestAttachmentWorkerGenerationIsSupervisorSupplied proves the host no longer
827// allocates a UI generation: it grants and stamps the caller's generation, and
828// the same value can be re-granted across sequential runs because the
829// supervisor owns the counter. UI action binding stays deferred to the P5
830// supervisor, so no UI admission is published here; the transactional output
831// path still stamps the granted generation on every publication.
832func TestAttachmentWorkerGenerationIsSupervisorSupplied(t *testing.T) {
833 term := newWorkerTestTerminal()
834 pump := newTerminalInputPump(nil)
835 host := newWorkerTestHost(term, pump, nil)
836 for i := uint64(1); i <= 3; i++ {
837 run, ok := host.Begin(context.Background(), AttachmentToken{i, i + 10}, blockingWorker(), newWorkerTestStream())
838 require.True(t, ok)
839 require.Equal(t, AttachmentToken{i, i + 10}, run.fg.Token())
840 require.NoError(t, run.fg.Output(ports.UIContext{Generation: 999, Status: ports.UIStatusAttached}, nil))
841 require.Equal(t, i, term.publications()[i-1].Generation)
842 run.Cancel()
843 }
844}
845
846// TestAttachmentWorkerBeginRefusalsRetainStream proves every Begin refusal path
847// leaves the caller's stream untouched: it is not closed, receives no message,
848// no foreground is granted, and the caller keeps ownership.
849func TestAttachmentWorkerBeginRefusalsRetainStream(t *testing.T) {
850 t.Run("nil host", func(t *testing.T) {
851 var host *attachmentHost
852 stream := newWorkerTestStream()
853 run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), stream)
854 require.False(t, ok)
855 require.Nil(t, run)
856 require.False(t, stream.isClosed())
857 require.Empty(t, stream.messages())
858 })
859 t.Run("zero token", func(t *testing.T) {
860 host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
861 stream := newWorkerTestStream()
862 _, ok := host.Begin(context.Background(), AttachmentToken{}, blockingWorker(), stream)
863 require.False(t, ok)
864 require.False(t, stream.isClosed())
865 require.Empty(t, stream.messages())
866 require.Nil(t, host.authority.foreground())
867 })
868 t.Run("nil worker", func(t *testing.T) {
869 host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
870 stream := newWorkerTestStream()
871 _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, nil, stream)
872 require.False(t, ok)
873 require.False(t, stream.isClosed())
874 require.Nil(t, host.authority.foreground())
875 })
876 t.Run("nil stream", func(t *testing.T) {
877 host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
878 _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), nil)
879 require.False(t, ok)
880 require.Nil(t, host.authority.foreground())
881 })
882 t.Run("foreground already live", func(t *testing.T) {
883 host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
884 first, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
885 require.True(t, ok)
886 stream := newWorkerTestStream()
887 _, ok = host.Begin(context.Background(), AttachmentToken{2, 1}, blockingWorker(), stream)
888 require.False(t, ok)
889 require.False(t, stream.isClosed(), "a refused Begin must not close the caller's stream")
890 require.Empty(t, stream.messages())
891 require.Same(t, first.fg, host.authority.foreground())
892 first.Cancel()
893 })
894 t.Run("rival consumer owns input", func(t *testing.T) {
895 pump := newTerminalInputPump(nil)
896 host := newWorkerTestHost(newWorkerTestTerminal(), pump, nil)
897 rival, claimed := pump.tryClaim()
898 require.True(t, claimed)
899 defer pump.revoke(rival)
900 stream := newWorkerTestStream()
901 _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), stream)
902 require.False(t, ok)
903 require.False(t, stream.isClosed())
904 require.Empty(t, stream.messages())
905 require.Nil(t, host.authority.foreground())
906 })
907}
908
909// TestAttachmentWorkerFinalizePreservesUndecidedInput proves the deterministic
910// final input window: cancellation wakes the worker through Done while the pump
911// consumer is still authorized for Ack/Preserve only, so the undecided read
912// survives to the next generation exactly once, and a duplicate PreserveInput
913// for the same delivery can never overwrite it.
914func TestAttachmentWorkerFinalizePreservesUndecidedInput(t *testing.T) {
915 pr, pw := io.Pipe()
916 defer pr.Close()
917 defer pw.Close()
918 pump := newTerminalInputPump(pr)
919 pump.start()
920 defer pump.stop()
921 host := newAttachmentHost(attachmentHostConfig{Input: pump})
922
923 var mu sync.Mutex
924 var refusals []string
925 refuse := func(what string) {
926 mu.Lock()
927 refusals = append(refusals, what)
928 mu.Unlock()
929 }
930 delivered := make(chan []byte, 1)
931 preserved := make(chan struct{})
932 first := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
933 event, ok := fg.Input(ctx)
934 if !ok {
935 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("first attempt saw no input")}
936 }
937 delivered <- append([]byte(nil), event.Data...)
938 // The supervisor finalizes: Done wakes us while Ack/Preserve stay
939 // authorized and every other action is refused.
940 <-fg.Done()
941 if _, ok := fg.Input(ctx); ok {
942 refuse("new input")
943 }
944 if _, ok := fg.Resize(ctx); ok {
945 refuse("resize")
946 }
947 if err := fg.Output(ports.UIContext{}, []byte("late")); !errors.Is(err, errAttachmentForegroundRevoked) {
948 refuse("output")
949 }
950 fg.PreserveInput(event.Data)
951 fg.PreserveInput([]byte("overwrite"))
952 close(preserved)
953 return AttachmentEvent{Kind: AttachmentEventEnded}
954 }}
955
956 written := make(chan error, 1)
957 go func() { _, err := pw.Write([]byte("ab")); written <- err }()
958
959 run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, first, newWorkerTestStream())
960 require.True(t, ok)
961 select {
962 case data := <-delivered:
963 require.Equal(t, []byte("ab"), data)
964 case <-time.After(time.Second):
965 t.Fatal("first attempt never received input")
966 }
967 run.Cancel()
968 select {
969 case <-preserved:
970 case <-time.After(time.Second):
971 t.Fatal("worker was not woken to preserve input")
972 }
973 mu.Lock()
974 require.Empty(t, refusals, "finalization must refuse every action except Ack/Preserve")
975 mu.Unlock()
976 require.NoError(t, <-written)
977 require.Nil(t, host.authority.foreground())
978
979 // The next generation replays the preserved bytes once, not the rejected
980 // duplicate.
981 observed := make(chan []byte, 2)
982 second := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
983 event, ok := fg.Input(ctx)
984 if !ok {
985 return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("second attempt saw no input")}
986 }
987 fg.AckInput()
988 observed <- append([]byte(nil), event.Data...)
989 return AttachmentEvent{Kind: AttachmentEventEnded}
990 }}
991 _, adopted := host.Run(context.Background(), AttachmentToken{2, 1}, second, newWorkerTestStream())
992 require.True(t, adopted)
993 require.Equal(t, []byte("ab"), <-observed)
994
995 // A third generation sees only fresh input: the preserved read was consumed
996 // exactly once.
997 freshWritten := make(chan error, 1)
998 go func() { _, err := pw.Write([]byte("fresh")); freshWritten <- err }()
999 _, adopted = host.Run(context.Background(), AttachmentToken{3, 1}, second, newWorkerTestStream())
1000 require.True(t, adopted)
1001 require.Equal(t, []byte("fresh"), <-observed)
1002 require.NoError(t, <-freshWritten)
1003}
1004
1005func TestAttachmentWorkerRetirementPrecedesWorkerCancellation(t *testing.T) {
1006 host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
1007 ready := make(chan *attachmentRun, 1)
1008 checked := make(chan bool, 1)
1009 worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
1010 run := <-ready
1011 <-ctx.Done()
1012 select {
1013 case <-run.retired:
1014 checked <- true
1015 default:
1016 checked <- false
1017 }
1018 return AttachmentEvent{Kind: AttachmentEventEnded}
1019 }}
1020 run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, worker, newWorkerTestStream())
1021 require.True(t, ok)
1022 ready <- run
1023 run.Cancel()
1024 require.True(t, <-checked)
1025 _, adopted := run.Wait(context.Background())
1026 require.False(t, adopted)
1027}
1028
1029func TestAttachmentWorkerBoundEventUsesGrantedGeneration(t *testing.T) {
1030 host := newAttachmentHost(attachmentHostConfig{Input: newTerminalInputPump(nil)})
1031 event, adopted := host.Run(context.Background(), AttachmentToken{99, 42}, &fakeWorker{run: func(context.Context, AttachmentForeground) AttachmentEvent {
1032 return AttachmentEvent{Token: AttachmentToken{999, 999}, Kind: AttachmentEventEnded}
1033 }}, newWorkerTestStream())
1034 require.True(t, adopted)
1035 require.Equal(t, AttachmentToken{99, 42}, event.Token)
1036}