detached ca3d194

feat(client): add scoped attachment worker

commit
ca3d19416b077695f5800c4e050c401e91b75c50
author
brice · · 2026-09-17 09:40 UTC
parent
21cbe0b
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +1834 -2
Unified
UnifiedSplit
A internal/usecase/client/attachment_worker.go +819 -0
@@ -0,0 +1,819 @@
1+package client
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "sync"
9+ "sync/atomic"
10+ "time"
11+
12+ "github.com/bnema/vev/internal/domain"
13+ "github.com/bnema/vev/internal/ports"
14+)
15+
16+// Attachment worker mechanisms (Plan 001 P5.1b, offline and unactivated).
17+//
18+// This file extracts the pieces of the legacy attach attempt that the
19+// autonomous supervisor must own, without changing the legacy Runner and
20+// without wiring the supervisor into production. It reuses the existing
21+// terminal input pump (terminalInputPump) and foreground output gate
22+// (foregroundSendLease) rather than duplicating them.
23+//
24+// Ownership split:
25+//
26+// - The supervisor owns the terminal: raw mode, the terminal writer, the
27+// single input pump, the UI publication transaction, resize events, and the
28+// process lifecycle. None of those move into a worker.
29+// - One AttachmentWorker owns exactly one broker logical stream and the typed
30+// session I/O on it. It cannot enter or restore raw mode, choose an
31+// endpoint, restore a source route, retain an attachment in the cache, or
32+// write the terminal directly.
33+// - Every terminal-facing action a worker performs crosses the
34+// supervisor-granted AttachmentForeground. The supervisor grants exactly
35+// one foreground at a time; on cancellation or supersession it finalizes
36+// that grant (closing Done and refusing output, UI, resize, and new input),
37+// lets the woken worker Ack/Preserve the read it holds through a bounded
38+// join, then revokes input authority before closing the owned stream.
39+// - Every worker result carries the AttachmentToken (generation/attempt) the
40+// supervisor granted, so a late result from a canceled or superseded
41+// generation is discarded rather than applied.
42+
43+// Integration requirement: P5.1a currently owns a rival terminal reader. Before
44+// activation it must use this same pump for picker and attachment consumers;
45+// never start both readers. Runner and supervisor wiring remain unchanged.
46+//
47+// UI actions are deliberately not part of this slice. AttachmentForeground has
48+// no action admission seam and attachmentHostConfig has no Actions field:
49+// binding UI.bindForeground and admitting actions requires the P5 supervisor's
50+// automation loop, which does not exist yet, so exposing it would publish a
51+// generation no component can drive. The transactional UI output path
52+// (ports.UIOutputTransaction, reached through Output) stays independent and
53+// usable. When the P5 supervisor owns input automation, it must bind and revoke
54+// UI actions at the same foreground boundaries this host finalizes.
55+
56+// attachmentWorkerJoinTimeout bounds the supervisor's join of a cancelled
57+// worker during the final input window. A worker that honors neither
58+// cancellation nor a closed Done is abandoned rather than allowed to pin the
59+// supervisor; the supervisor then revokes the retained consumer and closes the
60+// stream anyway. It mirrors the attachment retirement join budget.
61+const attachmentWorkerJoinTimeout = 5 * time.Second
62+
63+// errAttachmentForegroundRevoked reports a worker action refused because the
64+// supervisor revoked its input/output authority.
65+var errAttachmentForegroundRevoked = errors.New("client: attachment foreground authority revoked")
66+
67+// AttachmentToken identifies one attachment worker run. Generation is the
68+// supervisor-assigned foreground generation. The host never allocates it while
69+// UI action binding is deferred, so the supervisor supplies it and the host
70+// stamps it onto the event and every publication; Attempt is the worker
71+// identity, not a UI generation. A zero Generation is never granted.
72+type AttachmentToken struct {
73+ Generation uint64
74+ Attempt uint64
75+}
76+
77+// IsZero reports whether the token was never assigned by a supervisor.
78+func (t AttachmentToken) IsZero() bool { return t.Generation == 0 }
79+
80+// String renders the token for diagnostics.
81+func (t AttachmentToken) String() string {
82+ return fmt.Sprintf("gen=%d attempt=%d", t.Generation, t.Attempt)
83+}
84+
85+// AttachmentEventKind classifies one worker's terminal outcome. Every kind
86+// returns the client to the picker: a worker event never terminates the process
87+// by itself. Only the supervisor's own state machine decides that.
88+type AttachmentEventKind uint8
89+
90+const (
91+ // AttachmentEventEnded is an orderly end: the logical stream reached EOF or
92+ // the worker detached without failure.
93+ AttachmentEventEnded AttachmentEventKind = iota + 1
94+ // AttachmentEventLost is a transport or logical-stream loss. The logical
95+ // stream's own Done/Err carries the stable cause.
96+ AttachmentEventLost
97+ // AttachmentEventFailed is a typed protocol or application failure on the
98+ // stream, distinct from a carriage loss.
99+ AttachmentEventFailed
100+)
101+
102+// String renders the event kind for diagnostics.
103+func (k AttachmentEventKind) String() string {
104+ switch k {
105+ case AttachmentEventEnded:
106+ return "ended"
107+ case AttachmentEventLost:
108+ return "lost"
109+ case AttachmentEventFailed:
110+ return "failed"
111+ default:
112+ return "unknown"
113+ }
114+}
115+
116+// AttachmentEvent is the typed result of one worker run. Token is always the
117+// token the supervisor granted: the host stamps it, so a worker cannot report
118+// under another generation.
119+type AttachmentEvent struct {
120+ Token AttachmentToken
121+ Kind AttachmentEventKind
122+ Err error
123+}
124+
125+// AttachmentInputEvent is one authorized terminal input delivery. Err carries a
126+// terminal read failure (for example io.EOF) alongside any final bytes.
127+type AttachmentInputEvent struct {
128+ Data []byte
129+ Err error
130+}
131+
132+// AttachmentForeground is the supervisor-granted handle through which a worker
133+// touches the terminal. It is the worker's only path to input, resize, output,
134+// UI publication, and the attached transition. Every method is refused once the
135+// supervisor revokes the grant, so a canceled or superseded generation cannot
136+// read input, resize, write output, publish UI, or mark itself attached.
137+type AttachmentForeground interface {
138+ // Token is the generation/attempt identity of this grant.
139+ Token() AttachmentToken
140+ // Stream is the supervisor-admitted logical stream this worker owns for its
141+ // lifetime. The supervisor retains close authority.
142+ Stream() ports.BrokerLogicalConnection
143+ // Input blocks until the next authorized terminal input delivery. It returns
144+ // false once authority is revoked or ctx ends.
145+ Input(ctx context.Context) (AttachmentInputEvent, bool)
146+ // AckInput commits the last Input delivery so the shared reader may accept
147+ // the next one. It decides that delivery: at most one of AckInput and
148+ // PreserveInput may decide a delivery, and a second decision for the same
149+ // delivery is rejected as a no-op. It is a no-op once authority is revoked.
150+ AckInput()
151+ // PreserveInput re-queues the undecided bytes of the last Input delivery so
152+ // the next authorized attempt can replay them exactly once. It decides that
153+ // delivery: a duplicate decision (a later AckInput or PreserveInput for the
154+ // same delivery, or a call with no outstanding delivery) is rejected as a
155+ // no-op instead of re-queuing different bytes. It stays authorized while the
156+ // supervisor finalizes this grant (after Done closes and before revocation),
157+ // so a cancelled worker can still save input; it is a no-op once authority is
158+ // revoked.
159+ PreserveInput(data []byte)
160+ // Resize blocks until the next authorized terminal resize. It returns false
161+ // once authority is revoked or ctx ends. A nil or closed resize source is
162+ // disabled, not an attachment end; Resize waits for cancellation in that case.
163+ Resize(ctx context.Context) (domain.Geometry, bool)
164+ // Output publishes context and writes and flushes bytes in one authorized
165+ // transaction. Missing UI or terminal returns ports.ErrUIUnavailable.
166+ Output(uiContext ports.UIContext, data []byte) error
167+ // MarkAttached records the attached transition. It is refused once authority
168+ // is revoked.
169+ MarkAttached() bool
170+ // Attached reports whether MarkAttached has succeeded.
171+ Attached() bool
172+ // Done is closed when the supervisor finalizes this grant. It stops new
173+ // input, resize, output, and UI publication and wakes the worker, while
174+ // AckInput and PreserveInput stay authorized for the read the worker already
175+ // holds until the supervisor revokes the grant.
176+ Done() <-chan struct{}
177+}
178+
179+// AttachmentWorker is the contract for one broker logical stream's typed
180+// session I/O. Run owns the stream returned by Stream (via the foreground) and
181+// must honor ctx cancellation: the supervisor closes the stream after revoking
182+// authority, which must unblock receive/send. Run must never enter or restore
183+// raw mode, choose an endpoint, restore a source route, or touch the attachment
184+// cache; none of those authorities are reachable from its arguments.
185+type AttachmentWorker interface {
186+ Run(ctx context.Context, fg AttachmentForeground) AttachmentEvent
187+}
188+
189+// attachmentAuthorityState is the explicit lifecycle of the single foreground
190+// slot. Finalization is a first-class state rather than a side effect of a
191+// closed Done channel: while finalizing, only Ack/Preserve remain authorized,
192+// every other action and a competing grant are refused.
193+type attachmentAuthorityState uint8
194+
195+const (
196+ // authorityIdle has no foreground owner.
197+ authorityIdle attachmentAuthorityState = iota
198+ // authorityActive grants the foreground every AttachmentForeground action.
199+ authorityActive
200+ // authorityFinalizing retains the same foreground only for Ack/Preserve
201+ // while the supervisor joins a cancelled worker. New input, resize, output,
202+ // UI publication, MarkAttached, and a competing grant are all refused.
203+ authorityFinalizing
204+)
205+
206+// attachmentAuthority owns the single supervisor foreground slot. It is safe for
207+// concurrent use: the supervisor finalizes and revokes from its own goroutine
208+// while a worker acts from its run goroutine.
209+type attachmentAuthority struct {
210+ mu sync.Mutex
211+ current *attachmentForeground
212+ state attachmentAuthorityState
213+}
214+
215+// grant installs fg as the sole foreground in the active state. It fails while
216+// another grant is live or while the shared input pump already has a consumer,
217+// which is what keeps exactly one foreground owner. The host never allocates a
218+// UI generation here: UI action binding is deferred to the P5 supervisor.
219+func (a *attachmentAuthority) grant(fg *attachmentForeground) bool {
220+ if a == nil || fg == nil {
221+ return false
222+ }
223+ a.mu.Lock()
224+ defer a.mu.Unlock()
225+ if a.current != nil {
226+ return false
227+ }
228+ if fg.input != nil {
229+ var ok bool
230+ fg.consumer, ok = fg.input.tryClaim()
231+ if !ok {
232+ return false
233+ }
234+ }
235+ a.current = fg
236+ a.state = authorityActive
237+ return true
238+}
239+
240+// beginFinalize moves fg into the finalizing state: the worker keeps the slot
241+// and its pump consumer for Ack/Preserve, but every other action is refused.
242+// It reports false when fg no longer holds the slot.
243+func (a *attachmentAuthority) beginFinalize(fg *attachmentForeground) bool {
244+ if a == nil || fg == nil {
245+ return false
246+ }
247+ a.mu.Lock()
248+ defer a.mu.Unlock()
249+ if a.current != fg {
250+ return false
251+ }
252+ a.state = authorityFinalizing
253+ return true
254+}
255+
256+// revoke clears fg if it is still the foreground, releasing the pump consumer
257+// and any future UI binding, and reports whether it was. It runs after the
258+// bounded join. Input release and clearing the slot are atomic with respect to
259+// the next grant.
260+func (a *attachmentAuthority) revoke(fg *attachmentForeground) bool {
261+ if a == nil || fg == nil {
262+ return false
263+ }
264+ a.mu.Lock()
265+ defer a.mu.Unlock()
266+ if a.current != fg {
267+ return false
268+ }
269+ if fg.input != nil && fg.consumer != 0 {
270+ fg.input.revoke(fg.consumer)
271+ }
272+ a.current = nil
273+ a.state = authorityIdle
274+ return true
275+}
276+
277+// foreground is the sole synchronized ownership accessor. It returns the
278+// owner in both the active and finalizing states.
279+func (a *attachmentAuthority) foreground() *attachmentForeground {
280+ if a == nil {
281+ return nil
282+ }
283+ a.mu.Lock()
284+ defer a.mu.Unlock()
285+ return a.current
286+}
287+
288+// actionAuthorized reports whether fg may take a terminal-facing action now: it
289+// must hold the slot and the slot must not be finalizing.
290+func (a *attachmentAuthority) actionAuthorized(fg *attachmentForeground) bool {
291+ if a == nil || fg == nil {
292+ return false
293+ }
294+ a.mu.Lock()
295+ defer a.mu.Unlock()
296+ return a.current == fg && a.state == authorityActive
297+}
298+
299+// act runs fn under the authority lock exactly when fg may still act, so an
300+// action or state transition (the attached marker) is atomic with respect to
301+// finalization and revocation.
302+func (a *attachmentAuthority) act(fg *attachmentForeground, fn func()) bool {
303+ if a == nil || fg == nil {
304+ return false
305+ }
306+ a.mu.Lock()
307+ defer a.mu.Unlock()
308+ if a.current != fg || a.state != authorityActive {
309+ return false
310+ }
311+ fn()
312+ return true
313+}
314+
315+// retain runs fn under the authority lock exactly when fg still owns the slot,
316+// including while finalizing. It is the Ack/Preserve path: those only commit or
317+// re-queue input the worker already held, so a cancelled generation may still
318+// save it during the final input window. Every other action uses act or
319+// actionAuthorized.
320+func (a *attachmentAuthority) retain(fg *attachmentForeground, fn func()) bool {
321+ if a == nil || fg == nil {
322+ return false
323+ }
324+ a.mu.Lock()
325+ defer a.mu.Unlock()
326+ if a.current != fg {
327+ return false
328+ }
329+ fn()
330+ return true
331+}
332+
333+// attachmentHostConfig supplies the supervisor-owned dependencies. Terminal,
334+// clock, and input belong to the supervisor; UI and OnAttached are optional
335+// seams. There is no Actions seam: UI action admission needs the P5
336+// supervisor's automation loop before it can be exposed (see the package
337+// integration requirement above).
338+type attachmentHostConfig struct {
339+ // Terminal is the controlling terminal the supervisor owns. Optional only
340+ // so a mechanism test can omit output entirely; production supplies it.
341+ Terminal ports.Terminal
342+ // Clock supplies the bounded-join timer. Use ports.Clock, not the wall
343+ // clock.
344+ Clock ports.Clock
345+ // Input is the supervisor-owned single terminal reader. The host claims one
346+ // consumer per foreground and revokes it after finalization. Optional.
347+ Input *terminalInputPump
348+ // UI is the supervisor-owned UI publication transaction. When nil it is
349+ // taken from Terminal when that implements ports.UIOutputTransaction.
350+ UI ports.UIOutputTransaction
351+ // OnAttached observes a successful MarkAttached. Optional.
352+ OnAttached func(AttachmentToken)
353+}
354+
355+// attachmentHost is the supervisor-side owner of the terminal foreground. It
356+// grants one worker generation at a time input/output authority over a single
357+// admitted logical stream, joins that worker within a bound, and discards late
358+// results. It owns no raw mode and restores no terminal: those stay with the
359+// supervisor.
360+type attachmentHost struct {
361+ term ports.Terminal
362+ clock ports.Clock
363+ input *terminalInputPump
364+ ui ports.UIOutputTransaction
365+ resizes <-chan domain.Geometry
366+ onAttach func(AttachmentToken)
367+ authority attachmentAuthority
368+}
369+
370+// newAttachmentHost builds the reusable foreground host. A nil clock falls back
371+// to the system clock; a nil terminal leaves output refused but still enforces
372+// authority.
373+func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost {
374+ clock := cfg.Clock
375+ if supervisorNil(clock) {
376+ clock = systemClock{}
377+ }
378+ ui := cfg.UI
379+ resizes := (<-chan domain.Geometry)(nil)
380+ if !supervisorNil(cfg.Terminal) {
381+ if supervisorNil(ui) {
382+ ui, _ = cfg.Terminal.(ports.UIOutputTransaction)
383+ }
384+ resizes = cfg.Terminal.ResizeEvents()
385+ }
386+ return &attachmentHost{
387+ term: cfg.Terminal,
388+ clock: clock,
389+ input: cfg.Input,
390+ ui: ui,
391+ resizes: resizes,
392+ onAttach: cfg.OnAttached,
393+ }
394+}
395+
396+// Begin grants one generation's foreground and starts its worker on a
397+// supervisor-joined goroutine. It never closes or otherwise mutates stream, so
398+// on false the caller retains sole ownership of it. It returns false when:
399+//
400+// - the host is nil;
401+// - the token is zero (no supervisor generation was assigned);
402+// - the worker or stream is nil, in any typed-nil shape;
403+// - another foreground is already live, or is finalizing a previous grant;
404+// - the shared input pump already has a consumer (a picker or rival
405+// foreground owns input).
406+//
407+// The granted token, event token, and published UI context always use the
408+// caller-supplied Generation. The caller must Wait or Cancel the returned run.
409+func (h *attachmentHost) Begin(ctx context.Context, token AttachmentToken, worker AttachmentWorker, stream ports.BrokerLogicalConnection) (*attachmentRun, bool) {
410+ if h == nil || token.IsZero() || supervisorNil(worker) || supervisorNil(stream) {
411+ return nil, false
412+ }
413+ if ctx == nil {
414+ ctx = context.Background()
415+ }
416+ workerCtx, cancel := context.WithCancel(ctx)
417+ fg := h.newForeground(token, stream)
418+ if !h.authority.grant(fg) {
419+ cancel()
420+ return nil, false
421+ }
422+ run := &attachmentRun{
423+ host: h,
424+ token: token,
425+ stream: stream,
426+ fg: fg,
427+ cancel: cancel,
428+ outcome: make(chan AttachmentEvent, 1),
429+ done: make(chan struct{}),
430+ retired: make(chan struct{}),
431+ }
432+ go func() {
433+ defer close(run.done)
434+ event := worker.Run(workerCtx, fg)
435+ // The host stamps the granted token: a worker can never report under a
436+ // generation it was not granted, and the settlement check is exact.
437+ event.Token = token
438+ run.outcome <- event
439+ }()
440+ return run, true
441+}
442+
443+// Run is Begin followed immediately by Wait, for callers that join in place.
444+// The second result is false when the run was superseded or cancelled, in which
445+// case the worker's late event is discarded.
446+func (h *attachmentHost) Run(ctx context.Context, token AttachmentToken, worker AttachmentWorker, stream ports.BrokerLogicalConnection) (AttachmentEvent, bool) {
447+ run, ok := h.Begin(ctx, token, worker, stream)
448+ if !ok {
449+ return AttachmentEvent{}, false
450+ }
451+ return run.Wait(ctx)
452+}
453+
454+// newForeground allocates a candidate; grant claims input before publishing it.
455+func (h *attachmentHost) newForeground(token AttachmentToken, stream ports.BrokerLogicalConnection) *attachmentForeground {
456+ return &attachmentForeground{
457+ authority: &h.authority,
458+ token: token,
459+ stream: stream,
460+ input: h.input,
461+ lease: newForegroundSendLease(),
462+ term: h.term,
463+ ui: h.ui,
464+ resizes: h.resizes,
465+ done: make(chan struct{}),
466+ decided: true,
467+ onAttached: h.onAttach,
468+ }
469+}
470+
471+// attachmentRun is one in-flight worker generation owned by the supervisor.
472+type attachmentRun struct {
473+ host *attachmentHost
474+ token AttachmentToken
475+ stream ports.BrokerLogicalConnection
476+ fg *attachmentForeground
477+ cancel context.CancelFunc
478+ // outcome is capacity one so a worker that settles after the supervisor has
479+ // already moved on can never block on its own return.
480+ outcome chan AttachmentEvent
481+ // done closes when the worker goroutine has returned.
482+ done chan struct{}
483+ joinOnce sync.Once
484+ stateMu sync.Mutex
485+ retired chan struct{}
486+ isRetired bool
487+}
488+
489+// Wait joins the run. It returns the worker's typed event with adopted=true
490+// only when the worker settled on its own and the supervisor had not cancelled;
491+// a run that was cancelled or superseded is joined and its late event discarded
492+// (adopted=false). The caller's ctx stops waiting for an outcome; retirement
493+// then uses the separate bounded worker-join budget.
494+func (r *attachmentRun) Wait(ctx context.Context) (AttachmentEvent, bool) {
495+ if ctx == nil {
496+ ctx = context.Background()
497+ }
498+ select {
499+ case event := <-r.outcome:
500+ // Settlement and external retirement compete at one state boundary.
501+ adopted := r.retire() && ctx.Err() == nil && event.Token == r.token
502+ r.cleanup()
503+ if adopted {
504+ return event, true
505+ }
506+ case <-r.retired:
507+ r.cleanup()
508+ case <-ctx.Done():
509+ r.Cancel()
510+ }
511+ return AttachmentEvent{}, false
512+}
513+
514+// retire is authoritative even when outcome and cancellation are both ready.
515+func (r *attachmentRun) retire() bool {
516+ r.stateMu.Lock()
517+ defer r.stateMu.Unlock()
518+ if r.isRetired {
519+ return false
520+ }
521+ r.isRetired = true
522+ close(r.retired)
523+ return true
524+}
525+
526+// Cancel marks retirement, runs the deterministic final input window, and
527+// returns once the worker is joined or the join bound expires and the retained
528+// consumer and stream are released.
529+func (r *attachmentRun) Cancel() {
530+ r.retire()
531+ r.cleanup()
532+}
533+
534+// cleanup is the ordered teardown. It refuses further output/UI/resize/new input
535+// and closes Done, wakes the worker, then joins within the bound while the pump
536+// consumer stays authorized for Ack/Preserve, so a worker woken by Done can
537+// still save the read it holds. Only after the join (or its timeout) does it
538+// revoke the consumer and foreground slot, then close the owned stream.
539+func (r *attachmentRun) cleanup() {
540+ r.joinOnce.Do(func() {
541+ r.fg.finalize()
542+ r.cancel()
543+ r.join()
544+ r.fg.finish()
545+ _ = r.stream.Close()
546+ })
547+}
548+
549+// join waits for the worker goroutine within the shared join bound. It allocates
550+// a timer only when the worker has not already returned.
551+func (r *attachmentRun) join() {
552+ select {
553+ case <-r.done:
554+ return
555+ default:
556+ }
557+ timer := r.host.clock.NewTimer(attachmentWorkerJoinTimeout)
558+ if supervisorNil(timer) {
559+ timer = systemClock{}.NewTimer(attachmentWorkerJoinTimeout)
560+ }
561+ defer stopSupervisorTimer(timer)
562+ select {
563+ case <-r.done:
564+ case <-timer.C():
565+ }
566+}
567+
568+// attachmentForeground is the supervisor-granted implementation of
569+// AttachmentForeground. It reuses the existing terminal input pump for input
570+// and the existing foreground send lease for output serialization.
571+type attachmentForeground struct {
572+ authority *attachmentAuthority
573+ token AttachmentToken
574+ stream ports.BrokerLogicalConnection
575+ input *terminalInputPump
576+ consumer uint64
577+ lease *foregroundSendLease
578+ term ports.Terminal
579+ ui ports.UIOutputTransaction
580+ resizes <-chan domain.Geometry
581+ done chan struct{}
582+ // finalizedOnce and finishedOnce split the single teardown into the
583+ // deterministic final input window: finalize closes Done and stops output
584+ // first, finish revokes the retained consumer only after the join.
585+ finalizedOnce sync.Once
586+ finishedOnce sync.Once
587+ // deliveryMu guards the at-most-once input decision. Input opens the window
588+ // by marking a freshly received delivery undecided; AckInput or PreserveInput
589+ // decides it at most once. decided starts true, so a decision with no
590+ // outstanding delivery is refused instead of silently committing nothing.
591+ deliveryMu sync.Mutex
592+ decided bool
593+ attached atomic.Bool
594+ onAttached func(AttachmentToken)
595+}
596+
597+// Token is the generation/attempt identity of this grant.
598+func (f *attachmentForeground) Token() AttachmentToken { return f.token }
599+
600+// Stream is the supervisor-admitted logical stream this worker owns.
601+func (f *attachmentForeground) Stream() ports.BrokerLogicalConnection { return f.stream }
602+
603+// Done is closed when the supervisor starts finalizing this grant. It stops
604+// Input and Resize and wakes a worker so it can still Ack/Preserve the read it
605+// holds; the output gate is stopped at the same time.
606+func (f *attachmentForeground) Done() <-chan struct{} { return f.done }
607+
608+// finalize opens the final input window exactly once: it moves the grant into
609+// the finalizing authority state, stops the output gate (waiting for any
610+// in-flight output transaction to end), and closes Done so Input/Resize refuse
611+// new work and the worker wakes. The pump consumer and foreground slot stay
612+// owned, so the worker may still Ack/Preserve the read it holds.
613+// Lock order is authority -> send lease; beginFinalize releases authority before
614+// waiting on the lease, so output never observes authority while it holds the
615+// lease.
616+func (f *attachmentForeground) finalize() {
617+ if f == nil {
618+ return
619+ }
620+ f.finalizedOnce.Do(func() {
621+ f.authority.beginFinalize(f)
622+ if f.lease != nil {
623+ f.lease.stop()
624+ }
625+ close(f.done)
626+ })
627+}
628+
629+// finish revokes the retained consumer and clears the foreground slot after the
630+// bounded join. It runs before the stream is closed, so undecided input either
631+// reached the pump during finalization or is dropped with the revocation.
632+func (f *attachmentForeground) finish() {
633+ if f == nil {
634+ return
635+ }
636+ f.finishedOnce.Do(func() { f.authority.revoke(f) })
637+}
638+
639+// beginDelivery opens the decision window for the delivery the worker just
640+// received: the next AckInput or PreserveInput decides it, and a further
641+// decision is refused until Input receives another delivery.
642+func (f *attachmentForeground) beginDelivery() {
643+ f.deliveryMu.Lock()
644+ f.decided = false
645+ f.deliveryMu.Unlock()
646+}
647+
648+// decideDelivery reports whether this is the first decision for the current
649+// delivery and marks it decided. A second Ack/Preserve for the same delivery,
650+// or a decision with no outstanding delivery, is refused.
651+func (f *attachmentForeground) decideDelivery() bool {
652+ f.deliveryMu.Lock()
653+ defer f.deliveryMu.Unlock()
654+ if f.decided {
655+ return false
656+ }
657+ f.decided = true
658+ return true
659+}
660+
661+// Input leases the next authorized terminal read. It waits on the pump's ready
662+// signal exactly like the attach scanner, so a shared single reader is never
663+// duplicated.
664+func (f *attachmentForeground) Input(ctx context.Context) (AttachmentInputEvent, bool) {
665+ if ctx == nil {
666+ ctx = context.Background()
667+ }
668+ for {
669+ if !f.authority.actionAuthorized(f) || f.input == nil {
670+ return AttachmentInputEvent{}, false
671+ }
672+ result, ok := f.input.take(ctx, f.consumer)
673+ if ok {
674+ f.beginDelivery()
675+ return AttachmentInputEvent{Data: result.data, Err: result.err}, true
676+ }
677+ if ctx.Err() != nil {
678+ return AttachmentInputEvent{}, false
679+ }
680+ select {
681+ case <-ctx.Done():
682+ return AttachmentInputEvent{}, false
683+ case <-f.done:
684+ return AttachmentInputEvent{}, false
685+ case <-f.input.ready:
686+ }
687+ }
688+}
689+
690+// AckInput commits the last delivery so the shared reader may publish the next.
691+// It decides that delivery, so a duplicate Ack/Preserve for it is a no-op.
692+func (f *attachmentForeground) AckInput() {
693+ if f == nil || f.input == nil {
694+ return
695+ }
696+ if !f.decideDelivery() {
697+ return
698+ }
699+ f.authority.retain(f, func() { f.input.ack(f.consumer) })
700+}
701+
702+// PreserveInput re-queues the undecided bytes of the last delivery for the next
703+// authorized attempt. It decides that delivery, so a duplicate Ack/Preserve for
704+// it is a no-op and can never re-queue different bytes. It stays authorized
705+// while the supervisor finalizes the grant.
706+func (f *attachmentForeground) PreserveInput(data []byte) {
707+ if f == nil || f.input == nil {
708+ return
709+ }
710+ if !f.decideDelivery() {
711+ return
712+ }
713+ f.authority.retain(f, func() {
714+ f.input.ack(f.consumer)
715+ f.input.preserveResidual(f.consumer, data)
716+ })
717+}
718+
719+// Resize waits for the next authorized terminal resize.
720+func (f *attachmentForeground) Resize(ctx context.Context) (domain.Geometry, bool) {
721+ if ctx == nil {
722+ ctx = context.Background()
723+ }
724+ if !f.authority.actionAuthorized(f) {
725+ return domain.Geometry{}, false
726+ }
727+ resizes := f.resizes
728+ for {
729+ select {
730+ case geometry, ok := <-resizes:
731+ if !ok {
732+ resizes = nil
733+ continue
734+ } // uiterm has no resize source
735+ if !f.authority.actionAuthorized(f) {
736+ return domain.Geometry{}, false
737+ }
738+ return geometry, true
739+ case <-ctx.Done():
740+ return domain.Geometry{}, false
741+ case <-f.done:
742+ return domain.Geometry{}, false
743+ }
744+ }
745+}
746+
747+// Output holds the same send lease across the entire UI observation boundary.
748+// It mirrors the legacy attach output path: a UI capture that is merely
749+// unavailable (ports.ErrUIUnavailable) never stops the frame, so the write and
750+// flush still proceed; any other publication failure aborts. EndOutput success
751+// follows the frame write and flush, so a failed publication, short write, or
752+// failed write or flush leaves EndOutput(false).
753+func (f *attachmentForeground) Output(uiContext ports.UIContext, data []byte) error {
754+ if f == nil {
755+ return errAttachmentForegroundRevoked
756+ }
757+ var outputErr error
758+ ok := f.lease.send(func() bool {
759+ if !f.authority.actionAuthorized(f) {
760+ return false
761+ }
762+ if supervisorNil(f.ui) || supervisorNil(f.term) {
763+ outputErr = ports.ErrUIUnavailable
764+ return true
765+ }
766+ writer := f.term.Out()
767+ if supervisorNil(writer) {
768+ outputErr = ports.ErrUIUnavailable
769+ return true
770+ }
771+ uiContext.Generation = f.token.Generation
772+ f.ui.BeginOutput(uiContext)
773+ success := false
774+ defer func() { f.ui.EndOutput(success) }()
775+ // An unavailable UI capture is tolerated, exactly like the legacy attach
776+ // path: the frame is still written and flushed. Any other publication
777+ // error aborts before the frame is written.
778+ if err := f.ui.PublishContext(uiContext); err != nil && !errors.Is(err, ports.ErrUIUnavailable) {
779+ outputErr = err
780+ return true
781+ }
782+ var n int
783+ n, outputErr = writer.Write(data)
784+ if outputErr == nil && n != len(data) {
785+ outputErr = io.ErrShortWrite
786+ }
787+ if outputErr != nil {
788+ return true
789+ }
790+ outputErr = f.term.Flush()
791+ success = outputErr == nil
792+ return true
793+ })
794+ if !ok {
795+ return errAttachmentForegroundRevoked
796+ }
797+ return outputErr
798+}
799+
800+// MarkAttached records the attached transition atomically with respect to
801+// finalization and revocation.
802+func (f *attachmentForeground) MarkAttached() bool {
803+ if f == nil {
804+ return false
805+ }
806+ set := f.authority.act(f, func() { f.attached.Store(true) })
807+ if !set {
808+ return false
809+ }
810+ if f.onAttached != nil {
811+ f.onAttached(f.token)
812+ }
813+ return true
814+}
815+
816+// Attached reports whether MarkAttached has succeeded.
817+func (f *attachmentForeground) Attached() bool {
818+ return f != nil && f.attached.Load()
819+}
@@ -0,0 +1,819 @@
1+package client
2+
3+import (
4+ "context"
5+ "errors"
6+ "fmt"
7+ "io"
8+ "sync"
9+ "sync/atomic"
10+ "time"
11+
12+ "github.com/bnema/vev/internal/domain"
13+ "github.com/bnema/vev/internal/ports"
14+)
15+
16+// Attachment worker mechanisms (Plan 001 P5.1b, offline and unactivated).
17+//
18+// This file extracts the pieces of the legacy attach attempt that the
19+// autonomous supervisor must own, without changing the legacy Runner and
20+// without wiring the supervisor into production. It reuses the existing
21+// terminal input pump (terminalInputPump) and foreground output gate
22+// (foregroundSendLease) rather than duplicating them.
23+//
24+// Ownership split:
25+//
26+// - The supervisor owns the terminal: raw mode, the terminal writer, the
27+// single input pump, the UI publication transaction, resize events, and the
28+// process lifecycle. None of those move into a worker.
29+// - One AttachmentWorker owns exactly one broker logical stream and the typed
30+// session I/O on it. It cannot enter or restore raw mode, choose an
31+// endpoint, restore a source route, retain an attachment in the cache, or
32+// write the terminal directly.
33+// - Every terminal-facing action a worker performs crosses the
34+// supervisor-granted AttachmentForeground. The supervisor grants exactly
35+// one foreground at a time; on cancellation or supersession it finalizes
36+// that grant (closing Done and refusing output, UI, resize, and new input),
37+// lets the woken worker Ack/Preserve the read it holds through a bounded
38+// join, then revokes input authority before closing the owned stream.
39+// - Every worker result carries the AttachmentToken (generation/attempt) the
40+// supervisor granted, so a late result from a canceled or superseded
41+// generation is discarded rather than applied.
42+
43+// Integration requirement: P5.1a currently owns a rival terminal reader. Before
44+// activation it must use this same pump for picker and attachment consumers;
45+// never start both readers. Runner and supervisor wiring remain unchanged.
46+//
47+// UI actions are deliberately not part of this slice. AttachmentForeground has
48+// no action admission seam and attachmentHostConfig has no Actions field:
49+// binding UI.bindForeground and admitting actions requires the P5 supervisor's
50+// automation loop, which does not exist yet, so exposing it would publish a
51+// generation no component can drive. The transactional UI output path
52+// (ports.UIOutputTransaction, reached through Output) stays independent and
53+// usable. When the P5 supervisor owns input automation, it must bind and revoke
54+// UI actions at the same foreground boundaries this host finalizes.
55+
56+// attachmentWorkerJoinTimeout bounds the supervisor's join of a cancelled
57+// worker during the final input window. A worker that honors neither
58+// cancellation nor a closed Done is abandoned rather than allowed to pin the
59+// supervisor; the supervisor then revokes the retained consumer and closes the
60+// stream anyway. It mirrors the attachment retirement join budget.
61+const attachmentWorkerJoinTimeout = 5 * time.Second
62+
63+// errAttachmentForegroundRevoked reports a worker action refused because the
64+// supervisor revoked its input/output authority.
65+var errAttachmentForegroundRevoked = errors.New("client: attachment foreground authority revoked")
66+
67+// AttachmentToken identifies one attachment worker run. Generation is the
68+// supervisor-assigned foreground generation. The host never allocates it while
69+// UI action binding is deferred, so the supervisor supplies it and the host
70+// stamps it onto the event and every publication; Attempt is the worker
71+// identity, not a UI generation. A zero Generation is never granted.
72+type AttachmentToken struct {
73+ Generation uint64
74+ Attempt uint64
75+}
76+
77+// IsZero reports whether the token was never assigned by a supervisor.
78+func (t AttachmentToken) IsZero() bool { return t.Generation == 0 }
79+
80+// String renders the token for diagnostics.
81+func (t AttachmentToken) String() string {
82+ return fmt.Sprintf("gen=%d attempt=%d", t.Generation, t.Attempt)
83+}
84+
85+// AttachmentEventKind classifies one worker's terminal outcome. Every kind
86+// returns the client to the picker: a worker event never terminates the process
87+// by itself. Only the supervisor's own state machine decides that.
88+type AttachmentEventKind uint8
89+
90+const (
91+ // AttachmentEventEnded is an orderly end: the logical stream reached EOF or
92+ // the worker detached without failure.
93+ AttachmentEventEnded AttachmentEventKind = iota + 1
94+ // AttachmentEventLost is a transport or logical-stream loss. The logical
95+ // stream's own Done/Err carries the stable cause.
96+ AttachmentEventLost
97+ // AttachmentEventFailed is a typed protocol or application failure on the
98+ // stream, distinct from a carriage loss.
99+ AttachmentEventFailed
100+)
101+
102+// String renders the event kind for diagnostics.
103+func (k AttachmentEventKind) String() string {
104+ switch k {
105+ case AttachmentEventEnded:
106+ return "ended"
107+ case AttachmentEventLost:
108+ return "lost"
109+ case AttachmentEventFailed:
110+ return "failed"
111+ default:
112+ return "unknown"
113+ }
114+}
115+
116+// AttachmentEvent is the typed result of one worker run. Token is always the
117+// token the supervisor granted: the host stamps it, so a worker cannot report
118+// under another generation.
119+type AttachmentEvent struct {
120+ Token AttachmentToken
121+ Kind AttachmentEventKind
122+ Err error
123+}
124+
125+// AttachmentInputEvent is one authorized terminal input delivery. Err carries a
126+// terminal read failure (for example io.EOF) alongside any final bytes.
127+type AttachmentInputEvent struct {
128+ Data []byte
129+ Err error
130+}
131+
132+// AttachmentForeground is the supervisor-granted handle through which a worker
133+// touches the terminal. It is the worker's only path to input, resize, output,
134+// UI publication, and the attached transition. Every method is refused once the
135+// supervisor revokes the grant, so a canceled or superseded generation cannot
136+// read input, resize, write output, publish UI, or mark itself attached.
137+type AttachmentForeground interface {
138+ // Token is the generation/attempt identity of this grant.
139+ Token() AttachmentToken
140+ // Stream is the supervisor-admitted logical stream this worker owns for its
141+ // lifetime. The supervisor retains close authority.
142+ Stream() ports.BrokerLogicalConnection
143+ // Input blocks until the next authorized terminal input delivery. It returns
144+ // false once authority is revoked or ctx ends.
145+ Input(ctx context.Context) (AttachmentInputEvent, bool)
146+ // AckInput commits the last Input delivery so the shared reader may accept
147+ // the next one. It decides that delivery: at most one of AckInput and
148+ // PreserveInput may decide a delivery, and a second decision for the same
149+ // delivery is rejected as a no-op. It is a no-op once authority is revoked.
150+ AckInput()
151+ // PreserveInput re-queues the undecided bytes of the last Input delivery so
152+ // the next authorized attempt can replay them exactly once. It decides that
153+ // delivery: a duplicate decision (a later AckInput or PreserveInput for the
154+ // same delivery, or a call with no outstanding delivery) is rejected as a
155+ // no-op instead of re-queuing different bytes. It stays authorized while the
156+ // supervisor finalizes this grant (after Done closes and before revocation),
157+ // so a cancelled worker can still save input; it is a no-op once authority is
158+ // revoked.
159+ PreserveInput(data []byte)
160+ // Resize blocks until the next authorized terminal resize. It returns false
161+ // once authority is revoked or ctx ends. A nil or closed resize source is
162+ // disabled, not an attachment end; Resize waits for cancellation in that case.
163+ Resize(ctx context.Context) (domain.Geometry, bool)
164+ // Output publishes context and writes and flushes bytes in one authorized
165+ // transaction. Missing UI or terminal returns ports.ErrUIUnavailable.
166+ Output(uiContext ports.UIContext, data []byte) error
167+ // MarkAttached records the attached transition. It is refused once authority
168+ // is revoked.
169+ MarkAttached() bool
170+ // Attached reports whether MarkAttached has succeeded.
171+ Attached() bool
172+ // Done is closed when the supervisor finalizes this grant. It stops new
173+ // input, resize, output, and UI publication and wakes the worker, while
174+ // AckInput and PreserveInput stay authorized for the read the worker already
175+ // holds until the supervisor revokes the grant.
176+ Done() <-chan struct{}
177+}
178+
179+// AttachmentWorker is the contract for one broker logical stream's typed
180+// session I/O. Run owns the stream returned by Stream (via the foreground) and
181+// must honor ctx cancellation: the supervisor closes the stream after revoking
182+// authority, which must unblock receive/send. Run must never enter or restore
183+// raw mode, choose an endpoint, restore a source route, or touch the attachment
184+// cache; none of those authorities are reachable from its arguments.
185+type AttachmentWorker interface {
186+ Run(ctx context.Context, fg AttachmentForeground) AttachmentEvent
187+}
188+
189+// attachmentAuthorityState is the explicit lifecycle of the single foreground
190+// slot. Finalization is a first-class state rather than a side effect of a
191+// closed Done channel: while finalizing, only Ack/Preserve remain authorized,
192+// every other action and a competing grant are refused.
193+type attachmentAuthorityState uint8
194+
195+const (
196+ // authorityIdle has no foreground owner.
197+ authorityIdle attachmentAuthorityState = iota
198+ // authorityActive grants the foreground every AttachmentForeground action.
199+ authorityActive
200+ // authorityFinalizing retains the same foreground only for Ack/Preserve
201+ // while the supervisor joins a cancelled worker. New input, resize, output,
202+ // UI publication, MarkAttached, and a competing grant are all refused.
203+ authorityFinalizing
204+)
205+
206+// attachmentAuthority owns the single supervisor foreground slot. It is safe for
207+// concurrent use: the supervisor finalizes and revokes from its own goroutine
208+// while a worker acts from its run goroutine.
209+type attachmentAuthority struct {
210+ mu sync.Mutex
211+ current *attachmentForeground
212+ state attachmentAuthorityState
213+}
214+
215+// grant installs fg as the sole foreground in the active state. It fails while
216+// another grant is live or while the shared input pump already has a consumer,
217+// which is what keeps exactly one foreground owner. The host never allocates a
218+// UI generation here: UI action binding is deferred to the P5 supervisor.
219+func (a *attachmentAuthority) grant(fg *attachmentForeground) bool {
220+ if a == nil || fg == nil {
221+ return false
222+ }
223+ a.mu.Lock()
224+ defer a.mu.Unlock()
225+ if a.current != nil {
226+ return false
227+ }
228+ if fg.input != nil {
229+ var ok bool
230+ fg.consumer, ok = fg.input.tryClaim()
231+ if !ok {
232+ return false
233+ }
234+ }
235+ a.current = fg
236+ a.state = authorityActive
237+ return true
238+}
239+
240+// beginFinalize moves fg into the finalizing state: the worker keeps the slot
241+// and its pump consumer for Ack/Preserve, but every other action is refused.
242+// It reports false when fg no longer holds the slot.
243+func (a *attachmentAuthority) beginFinalize(fg *attachmentForeground) bool {
244+ if a == nil || fg == nil {
245+ return false
246+ }
247+ a.mu.Lock()
248+ defer a.mu.Unlock()
249+ if a.current != fg {
250+ return false
251+ }
252+ a.state = authorityFinalizing
253+ return true
254+}
255+
256+// revoke clears fg if it is still the foreground, releasing the pump consumer
257+// and any future UI binding, and reports whether it was. It runs after the
258+// bounded join. Input release and clearing the slot are atomic with respect to
259+// the next grant.
260+func (a *attachmentAuthority) revoke(fg *attachmentForeground) bool {
261+ if a == nil || fg == nil {
262+ return false
263+ }
264+ a.mu.Lock()
265+ defer a.mu.Unlock()
266+ if a.current != fg {
267+ return false
268+ }
269+ if fg.input != nil && fg.consumer != 0 {
270+ fg.input.revoke(fg.consumer)
271+ }
272+ a.current = nil
273+ a.state = authorityIdle
274+ return true
275+}
276+
277+// foreground is the sole synchronized ownership accessor. It returns the
278+// owner in both the active and finalizing states.
279+func (a *attachmentAuthority) foreground() *attachmentForeground {
280+ if a == nil {
281+ return nil
282+ }
283+ a.mu.Lock()
284+ defer a.mu.Unlock()
285+ return a.current
286+}
287+
288+// actionAuthorized reports whether fg may take a terminal-facing action now: it
289+// must hold the slot and the slot must not be finalizing.
290+func (a *attachmentAuthority) actionAuthorized(fg *attachmentForeground) bool {
291+ if a == nil || fg == nil {
292+ return false
293+ }
294+ a.mu.Lock()
295+ defer a.mu.Unlock()
296+ return a.current == fg && a.state == authorityActive
297+}
298+
299+// act runs fn under the authority lock exactly when fg may still act, so an
300+// action or state transition (the attached marker) is atomic with respect to
301+// finalization and revocation.
302+func (a *attachmentAuthority) act(fg *attachmentForeground, fn func()) bool {
303+ if a == nil || fg == nil {
304+ return false
305+ }
306+ a.mu.Lock()
307+ defer a.mu.Unlock()
308+ if a.current != fg || a.state != authorityActive {
309+ return false
310+ }
311+ fn()
312+ return true
313+}
314+
315+// retain runs fn under the authority lock exactly when fg still owns the slot,
316+// including while finalizing. It is the Ack/Preserve path: those only commit or
317+// re-queue input the worker already held, so a cancelled generation may still
318+// save it during the final input window. Every other action uses act or
319+// actionAuthorized.
320+func (a *attachmentAuthority) retain(fg *attachmentForeground, fn func()) bool {
321+ if a == nil || fg == nil {
322+ return false
323+ }
324+ a.mu.Lock()
325+ defer a.mu.Unlock()
326+ if a.current != fg {
327+ return false
328+ }
329+ fn()
330+ return true
331+}
332+
333+// attachmentHostConfig supplies the supervisor-owned dependencies. Terminal,
334+// clock, and input belong to the supervisor; UI and OnAttached are optional
335+// seams. There is no Actions seam: UI action admission needs the P5
336+// supervisor's automation loop before it can be exposed (see the package
337+// integration requirement above).
338+type attachmentHostConfig struct {
339+ // Terminal is the controlling terminal the supervisor owns. Optional only
340+ // so a mechanism test can omit output entirely; production supplies it.
341+ Terminal ports.Terminal
342+ // Clock supplies the bounded-join timer. Use ports.Clock, not the wall
343+ // clock.
344+ Clock ports.Clock
345+ // Input is the supervisor-owned single terminal reader. The host claims one
346+ // consumer per foreground and revokes it after finalization. Optional.
347+ Input *terminalInputPump
348+ // UI is the supervisor-owned UI publication transaction. When nil it is
349+ // taken from Terminal when that implements ports.UIOutputTransaction.
350+ UI ports.UIOutputTransaction
351+ // OnAttached observes a successful MarkAttached. Optional.
352+ OnAttached func(AttachmentToken)
353+}
354+
355+// attachmentHost is the supervisor-side owner of the terminal foreground. It
356+// grants one worker generation at a time input/output authority over a single
357+// admitted logical stream, joins that worker within a bound, and discards late
358+// results. It owns no raw mode and restores no terminal: those stay with the
359+// supervisor.
360+type attachmentHost struct {
361+ term ports.Terminal
362+ clock ports.Clock
363+ input *terminalInputPump
364+ ui ports.UIOutputTransaction
365+ resizes <-chan domain.Geometry
366+ onAttach func(AttachmentToken)
367+ authority attachmentAuthority
368+}
369+
370+// newAttachmentHost builds the reusable foreground host. A nil clock falls back
371+// to the system clock; a nil terminal leaves output refused but still enforces
372+// authority.
373+func newAttachmentHost(cfg attachmentHostConfig) *attachmentHost {
374+ clock := cfg.Clock
375+ if supervisorNil(clock) {
376+ clock = systemClock{}
377+ }
378+ ui := cfg.UI
379+ resizes := (<-chan domain.Geometry)(nil)
380+ if !supervisorNil(cfg.Terminal) {
381+ if supervisorNil(ui) {
382+ ui, _ = cfg.Terminal.(ports.UIOutputTransaction)
383+ }
384+ resizes = cfg.Terminal.ResizeEvents()
385+ }
386+ return &attachmentHost{
387+ term: cfg.Terminal,
388+ clock: clock,
389+ input: cfg.Input,
390+ ui: ui,
391+ resizes: resizes,
392+ onAttach: cfg.OnAttached,
393+ }
394+}
395+
396+// Begin grants one generation's foreground and starts its worker on a
397+// supervisor-joined goroutine. It never closes or otherwise mutates stream, so
398+// on false the caller retains sole ownership of it. It returns false when:
399+//
400+// - the host is nil;
401+// - the token is zero (no supervisor generation was assigned);
402+// - the worker or stream is nil, in any typed-nil shape;
403+// - another foreground is already live, or is finalizing a previous grant;
404+// - the shared input pump already has a consumer (a picker or rival
405+// foreground owns input).
406+//
407+// The granted token, event token, and published UI context always use the
408+// caller-supplied Generation. The caller must Wait or Cancel the returned run.
409+func (h *attachmentHost) Begin(ctx context.Context, token AttachmentToken, worker AttachmentWorker, stream ports.BrokerLogicalConnection) (*attachmentRun, bool) {
410+ if h == nil || token.IsZero() || supervisorNil(worker) || supervisorNil(stream) {
411+ return nil, false
412+ }
413+ if ctx == nil {
414+ ctx = context.Background()
415+ }
416+ workerCtx, cancel := context.WithCancel(ctx)
417+ fg := h.newForeground(token, stream)
418+ if !h.authority.grant(fg) {
419+ cancel()
420+ return nil, false
421+ }
422+ run := &attachmentRun{
423+ host: h,
424+ token: token,
425+ stream: stream,
426+ fg: fg,
427+ cancel: cancel,
428+ outcome: make(chan AttachmentEvent, 1),
429+ done: make(chan struct{}),
430+ retired: make(chan struct{}),
431+ }
432+ go func() {
433+ defer close(run.done)
434+ event := worker.Run(workerCtx, fg)
435+ // The host stamps the granted token: a worker can never report under a
436+ // generation it was not granted, and the settlement check is exact.
437+ event.Token = token
438+ run.outcome <- event
439+ }()
440+ return run, true
441+}
442+
443+// Run is Begin followed immediately by Wait, for callers that join in place.
444+// The second result is false when the run was superseded or cancelled, in which
445+// case the worker's late event is discarded.
446+func (h *attachmentHost) Run(ctx context.Context, token AttachmentToken, worker AttachmentWorker, stream ports.BrokerLogicalConnection) (AttachmentEvent, bool) {
447+ run, ok := h.Begin(ctx, token, worker, stream)
448+ if !ok {
449+ return AttachmentEvent{}, false
450+ }
451+ return run.Wait(ctx)
452+}
453+
454+// newForeground allocates a candidate; grant claims input before publishing it.
455+func (h *attachmentHost) newForeground(token AttachmentToken, stream ports.BrokerLogicalConnection) *attachmentForeground {
456+ return &attachmentForeground{
457+ authority: &h.authority,
458+ token: token,
459+ stream: stream,
460+ input: h.input,
461+ lease: newForegroundSendLease(),
462+ term: h.term,
463+ ui: h.ui,
464+ resizes: h.resizes,
465+ done: make(chan struct{}),
466+ decided: true,
467+ onAttached: h.onAttach,
468+ }
469+}
470+
471+// attachmentRun is one in-flight worker generation owned by the supervisor.
472+type attachmentRun struct {
473+ host *attachmentHost
474+ token AttachmentToken
475+ stream ports.BrokerLogicalConnection
476+ fg *attachmentForeground
477+ cancel context.CancelFunc
478+ // outcome is capacity one so a worker that settles after the supervisor has
479+ // already moved on can never block on its own return.
480+ outcome chan AttachmentEvent
481+ // done closes when the worker goroutine has returned.
482+ done chan struct{}
483+ joinOnce sync.Once
484+ stateMu sync.Mutex
485+ retired chan struct{}
486+ isRetired bool
487+}
488+
489+// Wait joins the run. It returns the worker's typed event with adopted=true
490+// only when the worker settled on its own and the supervisor had not cancelled;
491+// a run that was cancelled or superseded is joined and its late event discarded
492+// (adopted=false). The caller's ctx stops waiting for an outcome; retirement
493+// then uses the separate bounded worker-join budget.
494+func (r *attachmentRun) Wait(ctx context.Context) (AttachmentEvent, bool) {
495+ if ctx == nil {
496+ ctx = context.Background()
497+ }
498+ select {
499+ case event := <-r.outcome:
500+ // Settlement and external retirement compete at one state boundary.
501+ adopted := r.retire() && ctx.Err() == nil && event.Token == r.token
502+ r.cleanup()
503+ if adopted {
504+ return event, true
505+ }
506+ case <-r.retired:
507+ r.cleanup()
508+ case <-ctx.Done():
509+ r.Cancel()
510+ }
511+ return AttachmentEvent{}, false
512+}
513+
514+// retire is authoritative even when outcome and cancellation are both ready.
515+func (r *attachmentRun) retire() bool {
516+ r.stateMu.Lock()
517+ defer r.stateMu.Unlock()
518+ if r.isRetired {
519+ return false
520+ }
521+ r.isRetired = true
522+ close(r.retired)
523+ return true
524+}
525+
526+// Cancel marks retirement, runs the deterministic final input window, and
527+// returns once the worker is joined or the join bound expires and the retained
528+// consumer and stream are released.
529+func (r *attachmentRun) Cancel() {
530+ r.retire()
531+ r.cleanup()
532+}
533+
534+// cleanup is the ordered teardown. It refuses further output/UI/resize/new input
535+// and closes Done, wakes the worker, then joins within the bound while the pump
536+// consumer stays authorized for Ack/Preserve, so a worker woken by Done can
537+// still save the read it holds. Only after the join (or its timeout) does it
538+// revoke the consumer and foreground slot, then close the owned stream.
539+func (r *attachmentRun) cleanup() {
540+ r.joinOnce.Do(func() {
541+ r.fg.finalize()
542+ r.cancel()
543+ r.join()
544+ r.fg.finish()
545+ _ = r.stream.Close()
546+ })
547+}
548+
549+// join waits for the worker goroutine within the shared join bound. It allocates
550+// a timer only when the worker has not already returned.
551+func (r *attachmentRun) join() {
552+ select {
553+ case <-r.done:
554+ return
555+ default:
556+ }
557+ timer := r.host.clock.NewTimer(attachmentWorkerJoinTimeout)
558+ if supervisorNil(timer) {
559+ timer = systemClock{}.NewTimer(attachmentWorkerJoinTimeout)
560+ }
561+ defer stopSupervisorTimer(timer)
562+ select {
563+ case <-r.done:
564+ case <-timer.C():
565+ }
566+}
567+
568+// attachmentForeground is the supervisor-granted implementation of
569+// AttachmentForeground. It reuses the existing terminal input pump for input
570+// and the existing foreground send lease for output serialization.
571+type attachmentForeground struct {
572+ authority *attachmentAuthority
573+ token AttachmentToken
574+ stream ports.BrokerLogicalConnection
575+ input *terminalInputPump
576+ consumer uint64
577+ lease *foregroundSendLease
578+ term ports.Terminal
579+ ui ports.UIOutputTransaction
580+ resizes <-chan domain.Geometry
581+ done chan struct{}
582+ // finalizedOnce and finishedOnce split the single teardown into the
583+ // deterministic final input window: finalize closes Done and stops output
584+ // first, finish revokes the retained consumer only after the join.
585+ finalizedOnce sync.Once
586+ finishedOnce sync.Once
587+ // deliveryMu guards the at-most-once input decision. Input opens the window
588+ // by marking a freshly received delivery undecided; AckInput or PreserveInput
589+ // decides it at most once. decided starts true, so a decision with no
590+ // outstanding delivery is refused instead of silently committing nothing.
591+ deliveryMu sync.Mutex
592+ decided bool
593+ attached atomic.Bool
594+ onAttached func(AttachmentToken)
595+}
596+
597+// Token is the generation/attempt identity of this grant.
598+func (f *attachmentForeground) Token() AttachmentToken { return f.token }
599+
600+// Stream is the supervisor-admitted logical stream this worker owns.
601+func (f *attachmentForeground) Stream() ports.BrokerLogicalConnection { return f.stream }
602+
603+// Done is closed when the supervisor starts finalizing this grant. It stops
604+// Input and Resize and wakes a worker so it can still Ack/Preserve the read it
605+// holds; the output gate is stopped at the same time.
606+func (f *attachmentForeground) Done() <-chan struct{} { return f.done }
607+
608+// finalize opens the final input window exactly once: it moves the grant into
609+// the finalizing authority state, stops the output gate (waiting for any
610+// in-flight output transaction to end), and closes Done so Input/Resize refuse
611+// new work and the worker wakes. The pump consumer and foreground slot stay
612+// owned, so the worker may still Ack/Preserve the read it holds.
613+// Lock order is authority -> send lease; beginFinalize releases authority before
614+// waiting on the lease, so output never observes authority while it holds the
615+// lease.
616+func (f *attachmentForeground) finalize() {
617+ if f == nil {
618+ return
619+ }
620+ f.finalizedOnce.Do(func() {
621+ f.authority.beginFinalize(f)
622+ if f.lease != nil {
623+ f.lease.stop()
624+ }
625+ close(f.done)
626+ })
627+}
628+
629+// finish revokes the retained consumer and clears the foreground slot after the
630+// bounded join. It runs before the stream is closed, so undecided input either
631+// reached the pump during finalization or is dropped with the revocation.
632+func (f *attachmentForeground) finish() {
633+ if f == nil {
634+ return
635+ }
636+ f.finishedOnce.Do(func() { f.authority.revoke(f) })
637+}
638+
639+// beginDelivery opens the decision window for the delivery the worker just
640+// received: the next AckInput or PreserveInput decides it, and a further
641+// decision is refused until Input receives another delivery.
642+func (f *attachmentForeground) beginDelivery() {
643+ f.deliveryMu.Lock()
644+ f.decided = false
645+ f.deliveryMu.Unlock()
646+}
647+
648+// decideDelivery reports whether this is the first decision for the current
649+// delivery and marks it decided. A second Ack/Preserve for the same delivery,
650+// or a decision with no outstanding delivery, is refused.
651+func (f *attachmentForeground) decideDelivery() bool {
652+ f.deliveryMu.Lock()
653+ defer f.deliveryMu.Unlock()
654+ if f.decided {
655+ return false
656+ }
657+ f.decided = true
658+ return true
659+}
660+
661+// Input leases the next authorized terminal read. It waits on the pump's ready
662+// signal exactly like the attach scanner, so a shared single reader is never
663+// duplicated.
664+func (f *attachmentForeground) Input(ctx context.Context) (AttachmentInputEvent, bool) {
665+ if ctx == nil {
666+ ctx = context.Background()
667+ }
668+ for {
669+ if !f.authority.actionAuthorized(f) || f.input == nil {
670+ return AttachmentInputEvent{}, false
671+ }
672+ result, ok := f.input.take(ctx, f.consumer)
673+ if ok {
674+ f.beginDelivery()
675+ return AttachmentInputEvent{Data: result.data, Err: result.err}, true
676+ }
677+ if ctx.Err() != nil {
678+ return AttachmentInputEvent{}, false
679+ }
680+ select {
681+ case <-ctx.Done():
682+ return AttachmentInputEvent{}, false
683+ case <-f.done:
684+ return AttachmentInputEvent{}, false
685+ case <-f.input.ready:
686+ }
687+ }
688+}
689+
690+// AckInput commits the last delivery so the shared reader may publish the next.
691+// It decides that delivery, so a duplicate Ack/Preserve for it is a no-op.
692+func (f *attachmentForeground) AckInput() {
693+ if f == nil || f.input == nil {
694+ return
695+ }
696+ if !f.decideDelivery() {
697+ return
698+ }
699+ f.authority.retain(f, func() { f.input.ack(f.consumer) })
700+}
701+
702+// PreserveInput re-queues the undecided bytes of the last delivery for the next
703+// authorized attempt. It decides that delivery, so a duplicate Ack/Preserve for
704+// it is a no-op and can never re-queue different bytes. It stays authorized
705+// while the supervisor finalizes the grant.
706+func (f *attachmentForeground) PreserveInput(data []byte) {
707+ if f == nil || f.input == nil {
708+ return
709+ }
710+ if !f.decideDelivery() {
711+ return
712+ }
713+ f.authority.retain(f, func() {
714+ f.input.ack(f.consumer)
715+ f.input.preserveResidual(f.consumer, data)
716+ })
717+}
718+
719+// Resize waits for the next authorized terminal resize.
720+func (f *attachmentForeground) Resize(ctx context.Context) (domain.Geometry, bool) {
721+ if ctx == nil {
722+ ctx = context.Background()
723+ }
724+ if !f.authority.actionAuthorized(f) {
725+ return domain.Geometry{}, false
726+ }
727+ resizes := f.resizes
728+ for {
729+ select {
730+ case geometry, ok := <-resizes:
731+ if !ok {
732+ resizes = nil
733+ continue
734+ } // uiterm has no resize source
735+ if !f.authority.actionAuthorized(f) {
736+ return domain.Geometry{}, false
737+ }
738+ return geometry, true
739+ case <-ctx.Done():
740+ return domain.Geometry{}, false
741+ case <-f.done:
742+ return domain.Geometry{}, false
743+ }
744+ }
745+}
746+
747+// Output holds the same send lease across the entire UI observation boundary.
748+// It mirrors the legacy attach output path: a UI capture that is merely
749+// unavailable (ports.ErrUIUnavailable) never stops the frame, so the write and
750+// flush still proceed; any other publication failure aborts. EndOutput success
751+// follows the frame write and flush, so a failed publication, short write, or
752+// failed write or flush leaves EndOutput(false).
753+func (f *attachmentForeground) Output(uiContext ports.UIContext, data []byte) error {
754+ if f == nil {
755+ return errAttachmentForegroundRevoked
756+ }
757+ var outputErr error
758+ ok := f.lease.send(func() bool {
759+ if !f.authority.actionAuthorized(f) {
760+ return false
761+ }
762+ if supervisorNil(f.ui) || supervisorNil(f.term) {
763+ outputErr = ports.ErrUIUnavailable
764+ return true
765+ }
766+ writer := f.term.Out()
767+ if supervisorNil(writer) {
768+ outputErr = ports.ErrUIUnavailable
769+ return true
770+ }
771+ uiContext.Generation = f.token.Generation
772+ f.ui.BeginOutput(uiContext)
773+ success := false
774+ defer func() { f.ui.EndOutput(success) }()
775+ // An unavailable UI capture is tolerated, exactly like the legacy attach
776+ // path: the frame is still written and flushed. Any other publication
777+ // error aborts before the frame is written.
778+ if err := f.ui.PublishContext(uiContext); err != nil && !errors.Is(err, ports.ErrUIUnavailable) {
779+ outputErr = err
780+ return true
781+ }
782+ var n int
783+ n, outputErr = writer.Write(data)
784+ if outputErr == nil && n != len(data) {
785+ outputErr = io.ErrShortWrite
786+ }
787+ if outputErr != nil {
788+ return true
789+ }
790+ outputErr = f.term.Flush()
791+ success = outputErr == nil
792+ return true
793+ })
794+ if !ok {
795+ return errAttachmentForegroundRevoked
796+ }
797+ return outputErr
798+}
799+
800+// MarkAttached records the attached transition atomically with respect to
801+// finalization and revocation.
802+func (f *attachmentForeground) MarkAttached() bool {
803+ if f == nil {
804+ return false
805+ }
806+ set := f.authority.act(f, func() { f.attached.Store(true) })
807+ if !set {
808+ return false
809+ }
810+ if f.onAttached != nil {
811+ f.onAttached(f.token)
812+ }
813+ return true
814+}
815+
816+// Attached reports whether MarkAttached has succeeded.
817+func (f *attachmentForeground) Attached() bool {
818+ return f != nil && f.attached.Load()
819+}
A internal/usecase/client/attachment_worker_test.go +1004 -0
@@ -0,0 +1,1004 @@
1+package client
2+
3+import (
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+// workerTestStream is one scripted broker logical stream. It records typed
21+// client messages and unblocks ReceiveServer when the supervisor closes it,
22+// exactly like a real logical stream's Close contract.
23+type workerTestStream struct {
24+ mu sync.Mutex
25+ sent []protocol.ClientMessage
26+ closed chan struct{}
27+ closeOnce sync.Once
28+ // closeHook, when set, runs during Close. It lets a test observe the
29+ // supervisor's revocation-before-close ordering.
30+ closeHook func()
31+}
32+
33+func newWorkerTestStream() *workerTestStream {
34+ return &workerTestStream{closed: make(chan struct{})}
35+}
36+
37+func (s *workerTestStream) SendClient(message protocol.ClientMessage) error {
38+ select {
39+ case <-s.closed:
40+ return errors.New("worker test stream is closed")
41+ default:
42+ }
43+ s.mu.Lock()
44+ s.sent = append(s.sent, message)
45+ s.mu.Unlock()
46+ return nil
47+}
48+
49+func (s *workerTestStream) ReceiveServer() (protocol.ServerMessage, error) {
50+ <-s.closed
51+ return nil, io.EOF
52+}
53+
54+func (s *workerTestStream) Capabilities() protocol.ConnectionCapabilities {
55+ return protocol.ConnectionCapabilities{}
56+}
57+
58+func (s *workerTestStream) LinkState() ports.LinkState { return 0 }
59+
60+func (s *workerTestStream) LinkEvents() <-chan ports.LinkEvent { return nil }
61+
62+func (s *workerTestStream) Close() error {
63+ s.closeOnce.Do(func() {
64+ close(s.closed)
65+ if s.closeHook != nil {
66+ s.closeHook()
67+ }
68+ })
69+ return nil
70+}
71+
72+func (s *workerTestStream) Done() <-chan struct{} { return s.closed }
73+
74+func (s *workerTestStream) Err() error { return nil }
75+
76+func (s *workerTestStream) messages() []protocol.ClientMessage {
77+ s.mu.Lock()
78+ defer s.mu.Unlock()
79+ return append([]protocol.ClientMessage(nil), s.sent...)
80+}
81+
82+func (s *workerTestStream) isClosed() bool {
83+ select {
84+ case <-s.closed:
85+ return true
86+ default:
87+ return false
88+ }
89+}
90+
91+// workerTestTerminal is the supervisor-owned terminal: it counts raw restores
92+// so a test can prove the worker mechanism never restores, captures written
93+// output, and doubles as the UI publication transaction.
94+type workerTestTerminal struct {
95+ mu sync.Mutex
96+ out bytes.Buffer
97+ flushes int
98+ restores int
99+ resizes chan domain.Geometry
100+ published []ports.UIContext
101+ steps []string
102+ successes []bool
103+ writeErr error
104+ shortWrite int
105+ flushErr error
106+ publishErr error
107+}
108+
109+func newWorkerTestTerminal() *workerTestTerminal {
110+ return &workerTestTerminal{resizes: make(chan domain.Geometry, 4)}
111+}
112+
113+func (t *workerTestTerminal) EnterRaw() (func() error, error) {
114+ return func() error {
115+ t.mu.Lock()
116+ t.restores++
117+ t.mu.Unlock()
118+ return nil
119+ }, nil
120+}
121+
122+func (t *workerTestTerminal) Geometry() (domain.Geometry, error) { return domain.Geometry{}, nil }
123+
124+func (t *workerTestTerminal) ResizeEvents() <-chan domain.Geometry { return t.resizes }
125+
126+func (t *workerTestTerminal) In() io.Reader { return nil }
127+
128+func (t *workerTestTerminal) Out() io.Writer { return &workerTestWriter{t} }
129+
130+func (t *workerTestTerminal) Flush() error {
131+ t.mu.Lock()
132+ t.flushes++
133+ t.steps = append(t.steps, "flush")
134+ t.mu.Unlock()
135+ return t.flushErr
136+}
137+
138+func (t *workerTestTerminal) BeginOutput(ports.UIContext) {
139+ t.mu.Lock()
140+ defer t.mu.Unlock()
141+ t.steps = append(t.steps, "begin")
142+}
143+
144+func (t *workerTestTerminal) EndOutput(success bool) {
145+ t.mu.Lock()
146+ defer t.mu.Unlock()
147+ t.steps = append(t.steps, "end")
148+ t.successes = append(t.successes, success)
149+}
150+
151+func (t *workerTestTerminal) PublishContext(context ports.UIContext) error {
152+ t.mu.Lock()
153+ defer t.mu.Unlock()
154+ t.published = append(t.published, context)
155+ t.steps = append(t.steps, "publish")
156+ return t.publishErr
157+}
158+
159+func (t *workerTestTerminal) written() string {
160+ t.mu.Lock()
161+ defer t.mu.Unlock()
162+ return t.out.String()
163+}
164+
165+func (t *workerTestTerminal) flushCount() int {
166+ t.mu.Lock()
167+ defer t.mu.Unlock()
168+ return t.flushes
169+}
170+
171+func (t *workerTestTerminal) restoreCount() int {
172+ t.mu.Lock()
173+ defer t.mu.Unlock()
174+ return t.restores
175+}
176+
177+func (t *workerTestTerminal) publications() []ports.UIContext {
178+ t.mu.Lock()
179+ defer t.mu.Unlock()
180+ return append([]ports.UIContext(nil), t.published...)
181+}
182+
183+type workerTestWriter struct{ term *workerTestTerminal }
184+
185+func (w *workerTestWriter) Write(data []byte) (int, error) {
186+ w.term.mu.Lock()
187+ defer w.term.mu.Unlock()
188+ w.term.steps = append(w.term.steps, "write")
189+ if w.term.writeErr != nil {
190+ return 0, w.term.writeErr
191+ }
192+ if w.term.shortWrite > 0 {
193+ n := min(w.term.shortWrite, len(data))
194+ _, _ = w.term.out.Write(data[:n])
195+ return n, nil
196+ }
197+ return w.term.out.Write(data)
198+}
199+
200+// fakeWorker adapts a function to the AttachmentWorker contract.
201+type fakeWorker struct {
202+ run func(ctx context.Context, fg AttachmentForeground) AttachmentEvent
203+}
204+
205+func (w *fakeWorker) Run(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
206+ return w.run(ctx, fg)
207+}
208+
209+// blockingWorker ignores input/output and parks until its context ends.
210+func blockingWorker() *fakeWorker {
211+ return &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
212+ <-ctx.Done()
213+ return AttachmentEvent{Kind: AttachmentEventEnded}
214+ }}
215+}
216+
217+func newWorkerTestHost(term *workerTestTerminal, input *terminalInputPump, onAttached func(AttachmentToken)) *attachmentHost {
218+ return newAttachmentHost(attachmentHostConfig{
219+ Terminal: term,
220+ Clock: newSupervisorTestClock(),
221+ Input: input,
222+ OnAttached: onAttached,
223+ })
224+}
225+
226+// TestAttachmentWorkerExclusiveForeground proves exactly one worker generation
227+// owns the terminal foreground at a time: a second Begin is refused while the
228+// first is live and succeeds once the first is cancelled.
229+func TestAttachmentWorkerExclusiveForeground(t *testing.T) {
230+ term := newWorkerTestTerminal()
231+ host := newWorkerTestHost(term, nil, nil)
232+
233+ run1, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, blockingWorker(), newWorkerTestStream())
234+ require.True(t, ok)
235+ require.NotNil(t, run1)
236+
237+ _, ok = host.Begin(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, blockingWorker(), newWorkerTestStream())
238+ require.False(t, ok, "second foreground must be refused while the first is live")
239+ require.Equal(t, AttachmentToken{Generation: 1, Attempt: 1}, host.authority.foreground().Token())
240+
241+ run1.Cancel()
242+ require.Nil(t, host.authority.foreground())
243+
244+ run2, ok := host.Begin(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, blockingWorker(), newWorkerTestStream())
245+ require.True(t, ok, "a fresh generation may own the foreground after cancellation")
246+ run2.Cancel()
247+ require.Nil(t, host.authority.foreground())
248+}
249+
250+// TestAttachmentWorkerZeroTokenRefused proves an unassigned token is never
251+// granted, so a worker can never run without a supervisor generation.
252+func TestAttachmentWorkerZeroTokenRefused(t *testing.T) {
253+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
254+ _, ok := host.Begin(context.Background(), AttachmentToken{}, blockingWorker(), newWorkerTestStream())
255+ require.False(t, ok)
256+ require.Nil(t, host.authority.foreground())
257+}
258+
259+// TestAttachmentWorkerCanceledGenerationCannotAct proves cancellation revokes
260+// input/output authority: a canceled generation cannot write, flush, consume or
261+// forward input, resize, publish UI, or mark itself attached, and its grant is
262+// closed.
263+func TestAttachmentWorkerCanceledGenerationCannotAct(t *testing.T) {
264+ term := newWorkerTestTerminal()
265+ host := newWorkerTestHost(term, nil, nil)
266+ stream := newWorkerTestStream()
267+
268+ fgReady := make(chan AttachmentForeground, 1)
269+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
270+ fgReady <- fg
271+ <-ctx.Done()
272+ return AttachmentEvent{Kind: AttachmentEventEnded}
273+ }}
274+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 4, Attempt: 2}, worker, stream)
275+ require.True(t, ok)
276+ fg := <-fgReady
277+ require.Equal(t, AttachmentToken{Generation: 4, Attempt: 2}, fg.Token())
278+
279+ run.Cancel()
280+
281+ select {
282+ case <-fg.Done():
283+ default:
284+ t.Fatal("grant Done must be closed after cancellation")
285+ }
286+ require.ErrorIs(t, fg.Output(ports.UIContext{}, []byte("late")), errAttachmentForegroundRevoked)
287+ require.ErrorIs(t, fg.Output(ports.UIContext{AttachmentHandle: "late"}, nil), errAttachmentForegroundRevoked)
288+ require.False(t, fg.MarkAttached())
289+ require.False(t, fg.Attached())
290+
291+ _, okInput := fg.Input(context.Background())
292+ require.False(t, okInput)
293+ _, okResize := fg.Resize(context.Background())
294+ require.False(t, okResize)
295+
296+ require.Empty(t, term.written(), "a canceled generation must not write the terminal")
297+ require.Empty(t, term.publications(), "a canceled generation must not publish UI")
298+ require.True(t, stream.isClosed(), "the supervisor closes the owned stream")
299+}
300+
301+// TestAttachmentWorkerConsumesAndForwardsInput proves an authorized worker
302+// consumes the shared terminal pump and forwards the bytes on its own stream.
303+func TestAttachmentWorkerConsumesAndForwardsInput(t *testing.T) {
304+ term := newWorkerTestTerminal()
305+ pr, pw := io.Pipe()
306+ pump := newTerminalInputPump(pr)
307+ pump.start()
308+ defer pump.stop()
309+ defer pw.Close()
310+
311+ host := newWorkerTestHost(term, pump, nil)
312+ stream := newWorkerTestStream()
313+
314+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
315+ event, ok := fg.Input(ctx)
316+ if !ok {
317+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("no authorized input")}
318+ }
319+ fg.AckInput()
320+ if err := fg.Stream().SendClient(protocol.Input{InputSeq: 1, Data: event.Data}); err != nil {
321+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
322+ }
323+ return AttachmentEvent{Kind: AttachmentEventEnded}
324+ }}
325+
326+ go func() { _, _ = pw.Write([]byte("hello")) }()
327+
328+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
329+ require.True(t, adopted)
330+ require.Equal(t, AttachmentEventEnded, event.Kind)
331+
332+ messages := stream.messages()
333+ require.Len(t, messages, 1)
334+ inputMessage, ok := messages[0].(protocol.Input)
335+ require.True(t, ok)
336+ require.Equal(t, []byte("hello"), inputMessage.Data)
337+}
338+
339+// TestAttachmentWorkerResizeAuthorized proves an authorized worker observes the
340+// supervisor's resize events and forwards them on its stream.
341+func TestAttachmentWorkerResizeAuthorized(t *testing.T) {
342+ term := newWorkerTestTerminal()
343+ host := newWorkerTestHost(term, nil, nil)
344+ stream := newWorkerTestStream()
345+ geometry := domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}
346+
347+ saw := make(chan domain.Geometry, 1)
348+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
349+ term.resizes <- geometry
350+ resize, ok := fg.Resize(ctx)
351+ if !ok {
352+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("no authorized resize")}
353+ }
354+ saw <- resize
355+ if err := fg.Stream().SendClient(protocol.Resize{Size: resize.Size, PixelWidth: resize.PixelWidth, PixelHeight: resize.PixelHeight}); err != nil {
356+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
357+ }
358+ return AttachmentEvent{Kind: AttachmentEventEnded}
359+ }}
360+
361+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
362+ require.True(t, adopted)
363+ require.Equal(t, AttachmentEventEnded, event.Kind)
364+ require.Equal(t, geometry, <-saw)
365+
366+ messages := stream.messages()
367+ require.Len(t, messages, 1)
368+ _, ok := messages[0].(protocol.Resize)
369+ require.True(t, ok)
370+}
371+
372+// TestAttachmentWorkerPublishesUIAuthorized proves an authorized worker may
373+// publish UI context and mark the attached transition, and that the grant's
374+// generation is stamped onto the publication.
375+func TestAttachmentWorkerPublishesUIAuthorized(t *testing.T) {
376+ term := newWorkerTestTerminal()
377+ attached := make(chan AttachmentToken, 1)
378+ host := newWorkerTestHost(term, nil, func(token AttachmentToken) { attached <- token })
379+ stream := newWorkerTestStream()
380+
381+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
382+ if err := fg.Output(ports.UIContext{}, nil); err != nil {
383+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
384+ }
385+ if !fg.MarkAttached() {
386+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("attach refused")}
387+ }
388+ if !fg.Attached() {
389+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("attach not recorded")}
390+ }
391+ return AttachmentEvent{Kind: AttachmentEventEnded}
392+ }}
393+
394+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 7, Attempt: 2}, worker, stream)
395+ require.True(t, adopted)
396+ require.Equal(t, AttachmentEventEnded, event.Kind)
397+
398+ publications := term.publications()
399+ require.Len(t, publications, 1)
400+ require.Equal(t, uint64(7), publications[0].Generation, "the granted generation is stamped")
401+ require.Equal(t, AttachmentToken{Generation: 7, Attempt: 2}, <-attached)
402+}
403+
404+// TestAttachmentWorkerEventReturnsToPickerWithoutTerminalRestore proves a
405+// worker end, stream loss, or typed failure returns a typed event stamped with
406+// the granted token while the worker mechanism never restores the terminal and
407+// never closes the process.
408+func TestAttachmentWorkerEventReturnsToPickerWithoutTerminalRestore(t *testing.T) {
409+ cases := []struct {
410+ name string
411+ kind AttachmentEventKind
412+ err error
413+ }{
414+ {name: "ended", kind: AttachmentEventEnded},
415+ {name: "lost", kind: AttachmentEventLost, err: errors.New("carriage lost")},
416+ {name: "failed", kind: AttachmentEventFailed, err: &ProtocolError{Code: 7, Text: "bad welcome"}},
417+ }
418+ for _, tc := range cases {
419+ t.Run(tc.name, func(t *testing.T) {
420+ term := newWorkerTestTerminal()
421+ host := newWorkerTestHost(term, nil, nil)
422+ stream := newWorkerTestStream()
423+ // The worker reports a spoofed token; the host must override it with
424+ // the granted generation/attempt.
425+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
426+ return AttachmentEvent{Token: AttachmentToken{Generation: 999}, Kind: tc.kind, Err: tc.err}
427+ }}
428+
429+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 3, Attempt: 5}, worker, stream)
430+ require.True(t, adopted)
431+ require.Equal(t, tc.kind, event.Kind)
432+ require.Equal(t, AttachmentToken{Generation: 3, Attempt: 5}, event.Token)
433+ if tc.err != nil {
434+ require.ErrorIs(t, event.Err, tc.err)
435+ } else {
436+ require.NoError(t, event.Err)
437+ }
438+ require.Equal(t, 0, term.restoreCount(), "the worker mechanism must not restore the terminal")
439+ require.True(t, stream.isClosed(), "the supervisor closes the stream after the worker settles")
440+ require.Nil(t, host.authority.foreground(), "the foreground is released after the worker settles")
441+ })
442+ }
443+}
444+
445+// TestAttachmentWorkerLateEventDiscardedOnCancellation proves a worker that
446+// settles after the supervisor cancelled is joined and its late event is
447+// discarded rather than returned as an adopted result.
448+func TestAttachmentWorkerLateEventDiscardedOnCancellation(t *testing.T) {
449+ term := newWorkerTestTerminal()
450+ host := newWorkerTestHost(term, nil, nil)
451+ stream := newWorkerTestStream()
452+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, blockingWorker(), stream)
453+ require.True(t, ok)
454+
455+ ctx, cancel := context.WithCancel(context.Background())
456+ cancel()
457+
458+ event, adopted := run.Wait(ctx)
459+ require.False(t, adopted)
460+ require.Equal(t, AttachmentEvent{}, event, "a late event from a canceled run is discarded")
461+ require.True(t, stream.isClosed())
462+ require.Nil(t, host.authority.foreground())
463+}
464+
465+// TestAttachmentWorkerPreservesResidualInputAcrossAttempts proves the reused
466+// input pump keeps undecided bytes for the next authorized generation instead of
467+// duplicating a reader.
468+func TestAttachmentWorkerPreservesResidualInputAcrossAttempts(t *testing.T) {
469+ term := newWorkerTestTerminal()
470+ pr, pw := io.Pipe()
471+ pump := newTerminalInputPump(pr)
472+ pump.start()
473+ defer pump.stop()
474+ defer pw.Close()
475+
476+ host := newWorkerTestHost(term, pump, nil)
477+
478+ observed := make(chan []byte, 2)
479+ first := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
480+ event, ok := fg.Input(ctx)
481+ if !ok {
482+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("first attempt saw no input")}
483+ }
484+ observed <- append([]byte(nil), event.Data...)
485+ // Leave the bytes undecided: the next attempt must see them again.
486+ fg.PreserveInput(event.Data)
487+ return AttachmentEvent{Kind: AttachmentEventEnded}
488+ }}
489+ second := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
490+ event, ok := fg.Input(ctx)
491+ if !ok {
492+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("second attempt saw no input")}
493+ }
494+ fg.AckInput()
495+ observed <- append([]byte(nil), event.Data...)
496+ return AttachmentEvent{Kind: AttachmentEventEnded}
497+ }}
498+
499+ go func() { _, _ = pw.Write([]byte("ab")) }()
500+
501+ _, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, first, newWorkerTestStream())
502+ require.True(t, adopted)
503+ _, adopted = host.Run(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, second, newWorkerTestStream())
504+ require.True(t, adopted)
505+
506+ require.Equal(t, []byte("ab"), <-observed)
507+ require.Equal(t, []byte("ab"), <-observed, "the replacement generation replays the preserved bytes")
508+ go func() { _, _ = pw.Write([]byte("fresh")) }()
509+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
510+ defer cancel()
511+ _, adopted = host.Run(ctx, AttachmentToken{Generation: 3, Attempt: 3}, second, newWorkerTestStream())
512+ require.True(t, adopted)
513+ require.Equal(t, []byte("fresh"), <-observed, "third attempt must not replay the original pending read")
514+
515+}
516+
517+// TestAttachmentWorkerRevokesBeforeClosingStream proves the supervisor revokes
518+// input/output authority before it closes the owned stream: at the instant of
519+// Close the foreground is already released and a late write is refused.
520+func TestAttachmentWorkerRevokesBeforeClosingStream(t *testing.T) {
521+ term := newWorkerTestTerminal()
522+ host := newWorkerTestHost(term, nil, nil)
523+ stream := newWorkerTestStream()
524+
525+ fgReady := make(chan AttachmentForeground, 1)
526+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
527+ fgReady <- fg
528+ <-ctx.Done()
529+ return AttachmentEvent{Kind: AttachmentEventEnded}
530+ }}
531+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
532+ require.True(t, ok)
533+ fg := <-fgReady
534+
535+ var closeWriteErr error
536+ stream.closeHook = func() {
537+ require.Nil(t, host.authority.foreground(), "authority must be revoked before the stream closes")
538+ closeWriteErr = fg.Output(ports.UIContext{}, []byte("late"))
539+ }
540+
541+ run.Cancel()
542+ require.ErrorIs(t, closeWriteErr, errAttachmentForegroundRevoked)
543+ require.Empty(t, term.written())
544+}
545+
546+// TestAttachmentWorkerConcurrentBeginSingleOwner proves concurrent Begin calls
547+// admit exactly one foreground owner under the race detector.
548+func TestAttachmentWorkerConcurrentBeginSingleOwner(t *testing.T) {
549+ term := newWorkerTestTerminal()
550+ host := newWorkerTestHost(term, nil, nil)
551+
552+ const contenders = 8
553+ var wg sync.WaitGroup
554+ var successes atomic.Int64
555+ runs := make(chan *attachmentRun, contenders)
556+ start := make(chan struct{})
557+ for i := 0; i < contenders; i++ {
558+ wg.Add(1)
559+ go func(i int) {
560+ defer wg.Done()
561+ <-start
562+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: uint64(i + 1), Attempt: 1}, blockingWorker(), newWorkerTestStream())
563+ if ok {
564+ successes.Add(1)
565+ runs <- run
566+ }
567+ }(i)
568+ }
569+ close(start)
570+ wg.Wait()
571+ close(runs)
572+
573+ require.Equal(t, int64(1), successes.Load(), "exactly one concurrent Begin may own the foreground")
574+ for run := range runs {
575+ run.Cancel()
576+ }
577+ require.Nil(t, host.authority.foreground())
578+}
579+
580+// TestAttachmentWorkerRepeatedAttemptsNoLeak proves repeated grant/cancel cycles
581+// release the foreground, close every stream, and never leave a live owner.
582+func TestAttachmentWorkerRepeatedAttemptsNoLeak(t *testing.T) {
583+ term := newWorkerTestTerminal()
584+ host := newWorkerTestHost(term, nil, nil)
585+
586+ const attempts = 64
587+ for i := 0; i < attempts; i++ {
588+ stream := newWorkerTestStream()
589+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
590+ if err := fg.Output(ports.UIContext{}, []byte("x")); err != nil {
591+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
592+ }
593+ return AttachmentEvent{Kind: AttachmentEventEnded}
594+ }}
595+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: uint64(i + 1), Attempt: 1}, worker, stream)
596+ require.True(t, adopted)
597+ require.Equal(t, AttachmentEventEnded, event.Kind)
598+ require.True(t, stream.isClosed())
599+ require.Nil(t, host.authority.foreground())
600+ }
601+ require.Equal(t, attempts, term.flushCount())
602+ require.Equal(t, 0, term.restoreCount(), "the mechanism never enters or restores raw mode")
603+}
604+
605+func TestAttachmentWorkerRetiredOutcomeNeverAdopted(t *testing.T) {
606+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
607+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
608+ require.True(t, ok)
609+ run.Cancel()
610+ for range 3 {
611+ event, adopted := run.Wait(context.Background())
612+ require.False(t, adopted)
613+ require.Zero(t, event)
614+ }
615+}
616+
617+func TestAttachmentWorkerBoundedJoin(t *testing.T) {
618+ clock := newSupervisorTestClock()
619+ pr, pw := io.Pipe()
620+ defer pr.Close()
621+ defer pw.Close()
622+ pump := newTerminalInputPump(pr)
623+ pump.start()
624+ defer pump.stop()
625+ host := newAttachmentHost(attachmentHostConfig{Clock: clock, Input: pump})
626+ unblock := make(chan struct{})
627+ defer close(unblock)
628+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, &fakeWorker{run: func(context.Context, AttachmentForeground) AttachmentEvent {
629+ <-unblock
630+ return AttachmentEvent{Kind: AttachmentEventEnded}
631+ }}, newWorkerTestStream())
632+ require.True(t, ok)
633+ cancelled := make(chan struct{})
634+ go func() { run.Cancel(); close(cancelled) }()
635+ timer := clock.awaitTimer(t)
636+ require.Equal(t, attachmentWorkerJoinTimeout, timer.delay)
637+ select {
638+ case <-run.retired:
639+ default:
640+ t.Fatal("retirement must precede join")
641+ }
642+ // During the final input window the grant is finalizing, not revoked: Done is
643+ // closed, output and new input are refused, the retained consumer still holds
644+ // the shared reader, and the stream is still open.
645+ require.Same(t, run.fg, host.authority.foreground())
646+ select {
647+ case <-run.fg.Done():
648+ default:
649+ t.Fatal("finalize must close Done before the join")
650+ }
651+ require.ErrorIs(t, run.fg.Output(ports.UIContext{}, []byte("late")), errAttachmentForegroundRevoked)
652+ _, ok = host.Begin(context.Background(), AttachmentToken{2, 1}, blockingWorker(), newWorkerTestStream())
653+ require.False(t, ok, "a finalizing grant blocks a competing Begin")
654+ _, claimed := pump.tryClaim()
655+ require.False(t, claimed, "the consumer is retained during the final input window")
656+ require.False(t, run.stream.(*workerTestStream).isClosed(), "the stream closes after the join")
657+ timer.fire()
658+ select {
659+ case <-cancelled:
660+ case <-time.After(time.Second):
661+ t.Fatal("uncooperative worker pinned cancel")
662+ }
663+ require.True(t, timer.stopped())
664+ // After the timeout the supervisor revokes anyway: slot freed, consumer
665+ // released, and the stream closed.
666+ require.Nil(t, host.authority.foreground())
667+ consumer, claimed := pump.tryClaim()
668+ require.True(t, claimed, "revocation releases the consumer after the join")
669+ pump.revoke(consumer)
670+ require.True(t, run.stream.(*workerTestStream).isClosed())
671+ _, adopted := run.Wait(context.Background())
672+ require.False(t, adopted)
673+}
674+
675+func TestAttachmentWorkerPumpClaimAndCancelRace(t *testing.T) {
676+ pr, pw := io.Pipe()
677+ defer pr.Close()
678+ defer pw.Close()
679+ pump := newTerminalInputPump(pr)
680+ pump.start()
681+ defer pump.stop()
682+ host := newWorkerTestHost(newWorkerTestTerminal(), pump, nil)
683+ rival := pump.claim()
684+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
685+ require.False(t, ok)
686+ require.Nil(t, host.authority.foreground())
687+ pump.revoke(rival)
688+ for i := range 100 {
689+ run, ok := host.Begin(context.Background(), AttachmentToken{uint64(i + 1), 1}, blockingWorker(), newWorkerTestStream())
690+ require.True(t, ok)
691+ var next *attachmentRun
692+ var wg sync.WaitGroup
693+ wg.Add(2)
694+ go func() { defer wg.Done(); run.Cancel() }()
695+ go func() {
696+ defer wg.Done()
697+ next, _ = host.Begin(context.Background(), AttachmentToken{uint64(i + 2), 2}, blockingWorker(), newWorkerTestStream())
698+ }()
699+ wg.Wait()
700+ if next != nil {
701+ next.Cancel()
702+ }
703+ consumer, claimed := pump.tryClaim()
704+ require.True(t, claimed)
705+ pump.revoke(consumer)
706+ }
707+}
708+
709+func TestAttachmentWorkerClosedResizeDisabled(t *testing.T) {
710+ term := newWorkerTestTerminal()
711+ close(term.resizes) // uiterm-like terminal: no physical resize events
712+ host := newWorkerTestHost(term, nil, nil)
713+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
714+ require.True(t, ok)
715+ result := make(chan bool, 1)
716+ go func() { _, ok := run.fg.Resize(context.Background()); result <- ok }()
717+ select {
718+ case <-result:
719+ t.Fatal("closed resize source terminated attachment")
720+ case <-time.After(10 * time.Millisecond):
721+ }
722+ require.True(t, run.fg.MarkAttached())
723+ run.Cancel()
724+ select {
725+ case ok := <-result:
726+ require.False(t, ok)
727+ case <-time.After(time.Second):
728+ t.Fatal("authority Done did not stop resize")
729+ }
730+}
731+
732+func TestAttachmentWorkerOutputTransaction(t *testing.T) {
733+ failure := errors.New("output failure")
734+ for _, tc := range []struct {
735+ name string
736+ writeErr, flushErr, publishErr error
737+ shortWrite int
738+ wantErr error
739+ steps []string
740+ }{
741+ {name: "success", steps: []string{"begin", "publish", "write", "flush", "end"}},
742+ {name: "short write", shortWrite: 2, wantErr: io.ErrShortWrite, steps: []string{"begin", "publish", "write", "end"}},
743+ {name: "write failure", writeErr: failure, wantErr: failure, steps: []string{"begin", "publish", "write", "end"}},
744+ {name: "flush failure", flushErr: failure, wantErr: failure, steps: []string{"begin", "publish", "write", "flush", "end"}},
745+ {name: "publish failure", publishErr: failure, wantErr: failure, steps: []string{"begin", "publish", "end"}},
746+ {name: "publish unavailable tolerated", publishErr: ports.ErrUIUnavailable, steps: []string{"begin", "publish", "write", "flush", "end"}},
747+ } {
748+ t.Run(tc.name, func(t *testing.T) {
749+ term := newWorkerTestTerminal()
750+ term.writeErr, term.flushErr, term.publishErr, term.shortWrite = tc.writeErr, tc.flushErr, tc.publishErr, tc.shortWrite
751+ host := newWorkerTestHost(term, nil, nil)
752+ run, ok := host.Begin(context.Background(), AttachmentToken{9, 2}, blockingWorker(), newWorkerTestStream())
753+ require.True(t, ok)
754+ defer run.Cancel()
755+ err := run.fg.Output(ports.UIContext{Generation: 999}, []byte("frame"))
756+ success := tc.wantErr == nil
757+ if success {
758+ require.NoError(t, err)
759+ } else {
760+ require.ErrorIs(t, err, tc.wantErr)
761+ }
762+ require.Equal(t, tc.steps, term.steps)
763+ require.Equal(t, []bool{success}, term.successes)
764+ require.Equal(t, uint64(9), term.published[0].Generation)
765+ })
766+ }
767+}
768+
769+// Hide the transaction methods to model a physical terminal without UI output.
770+type workerBareTerminal struct{ ports.Terminal }
771+
772+func TestAttachmentWorkerNilSeams(t *testing.T) {
773+ var nilTerm *workerTestTerminal
774+ var nilWorker *fakeWorker
775+ var nilStream *workerTestStream
776+ var nilClock *supervisorTestClock
777+ for _, cfg := range []attachmentHostConfig{
778+ {},
779+ {Terminal: nilTerm, UI: nilTerm, Clock: nilClock},
780+ {Terminal: workerBareTerminal{newWorkerTestTerminal()}, UI: nilTerm},
781+ } {
782+ host := newAttachmentHost(cfg)
783+ _, ok := host.Begin(nil, AttachmentToken{1, 1}, nilWorker, newWorkerTestStream())
784+ require.False(t, ok)
785+ _, ok = host.Begin(nil, AttachmentToken{1, 1}, blockingWorker(), nilStream)
786+ require.False(t, ok)
787+ run, ok := host.Begin(nil, AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
788+ require.True(t, ok)
789+ require.ErrorIs(t, run.fg.Output(ports.UIContext{}, nil), ports.ErrUIUnavailable)
790+ run.Cancel()
791+ require.ErrorIs(t, run.fg.Output(ports.UIContext{}, nil), errAttachmentForegroundRevoked)
792+ }
793+}
794+
795+// TestAttachmentWorkerGenerationIsSupervisorSupplied proves the host no longer
796+// allocates a UI generation: it grants and stamps the caller's generation, and
797+// the same value can be re-granted across sequential runs because the
798+// supervisor owns the counter. UI action binding stays deferred to the P5
799+// supervisor, so no UI admission is published here; the transactional output
800+// path still stamps the granted generation on every publication.
801+func TestAttachmentWorkerGenerationIsSupervisorSupplied(t *testing.T) {
802+ term := newWorkerTestTerminal()
803+ pump := newTerminalInputPump(nil)
804+ host := newWorkerTestHost(term, pump, nil)
805+ for i := uint64(1); i <= 3; i++ {
806+ run, ok := host.Begin(context.Background(), AttachmentToken{i, i + 10}, blockingWorker(), newWorkerTestStream())
807+ require.True(t, ok)
808+ require.Equal(t, AttachmentToken{i, i + 10}, run.fg.Token())
809+ require.NoError(t, run.fg.Output(ports.UIContext{Generation: 999}, nil))
810+ require.Equal(t, i, term.publications()[i-1].Generation)
811+ run.Cancel()
812+ }
813+}
814+
815+// TestAttachmentWorkerBeginRefusalsRetainStream proves every Begin refusal path
816+// leaves the caller's stream untouched: it is not closed, receives no message,
817+// no foreground is granted, and the caller keeps ownership.
818+func TestAttachmentWorkerBeginRefusalsRetainStream(t *testing.T) {
819+ t.Run("nil host", func(t *testing.T) {
820+ var host *attachmentHost
821+ stream := newWorkerTestStream()
822+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), stream)
823+ require.False(t, ok)
824+ require.Nil(t, run)
825+ require.False(t, stream.isClosed())
826+ require.Empty(t, stream.messages())
827+ })
828+ t.Run("zero token", func(t *testing.T) {
829+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
830+ stream := newWorkerTestStream()
831+ _, ok := host.Begin(context.Background(), AttachmentToken{}, blockingWorker(), stream)
832+ require.False(t, ok)
833+ require.False(t, stream.isClosed())
834+ require.Empty(t, stream.messages())
835+ require.Nil(t, host.authority.foreground())
836+ })
837+ t.Run("nil worker", func(t *testing.T) {
838+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
839+ stream := newWorkerTestStream()
840+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, nil, stream)
841+ require.False(t, ok)
842+ require.False(t, stream.isClosed())
843+ require.Nil(t, host.authority.foreground())
844+ })
845+ t.Run("nil stream", func(t *testing.T) {
846+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
847+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), nil)
848+ require.False(t, ok)
849+ require.Nil(t, host.authority.foreground())
850+ })
851+ t.Run("foreground already live", func(t *testing.T) {
852+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
853+ first, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
854+ require.True(t, ok)
855+ stream := newWorkerTestStream()
856+ _, ok = host.Begin(context.Background(), AttachmentToken{2, 1}, blockingWorker(), stream)
857+ require.False(t, ok)
858+ require.False(t, stream.isClosed(), "a refused Begin must not close the caller's stream")
859+ require.Empty(t, stream.messages())
860+ require.Same(t, first.fg, host.authority.foreground())
861+ first.Cancel()
862+ })
863+ t.Run("rival consumer owns input", func(t *testing.T) {
864+ pump := newTerminalInputPump(nil)
865+ host := newWorkerTestHost(newWorkerTestTerminal(), pump, nil)
866+ rival := pump.claim()
867+ defer pump.revoke(rival)
868+ stream := newWorkerTestStream()
869+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), stream)
870+ require.False(t, ok)
871+ require.False(t, stream.isClosed())
872+ require.Empty(t, stream.messages())
873+ require.Nil(t, host.authority.foreground())
874+ })
875+}
876+
877+// TestAttachmentWorkerFinalizePreservesUndecidedInput proves the deterministic
878+// final input window: cancellation wakes the worker through Done while the pump
879+// consumer is still authorized for Ack/Preserve only, so the undecided read
880+// survives to the next generation exactly once, and a duplicate PreserveInput
881+// for the same delivery can never overwrite it.
882+func TestAttachmentWorkerFinalizePreservesUndecidedInput(t *testing.T) {
883+ pr, pw := io.Pipe()
884+ defer pr.Close()
885+ defer pw.Close()
886+ pump := newTerminalInputPump(pr)
887+ pump.start()
888+ defer pump.stop()
889+ host := newAttachmentHost(attachmentHostConfig{Input: pump})
890+
891+ var mu sync.Mutex
892+ var refusals []string
893+ refuse := func(what string) {
894+ mu.Lock()
895+ refusals = append(refusals, what)
896+ mu.Unlock()
897+ }
898+ delivered := make(chan []byte, 1)
899+ preserved := make(chan struct{})
900+ first := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
901+ event, ok := fg.Input(ctx)
902+ if !ok {
903+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("first attempt saw no input")}
904+ }
905+ delivered <- append([]byte(nil), event.Data...)
906+ // The supervisor finalizes: Done wakes us while Ack/Preserve stay
907+ // authorized and every other action is refused.
908+ <-fg.Done()
909+ if _, ok := fg.Input(ctx); ok {
910+ refuse("new input")
911+ }
912+ if _, ok := fg.Resize(ctx); ok {
913+ refuse("resize")
914+ }
915+ if err := fg.Output(ports.UIContext{}, []byte("late")); !errors.Is(err, errAttachmentForegroundRevoked) {
916+ refuse("output")
917+ }
918+ fg.PreserveInput(event.Data)
919+ fg.PreserveInput([]byte("overwrite"))
920+ close(preserved)
921+ return AttachmentEvent{Kind: AttachmentEventEnded}
922+ }}
923+
924+ written := make(chan error, 1)
925+ go func() { _, err := pw.Write([]byte("ab")); written <- err }()
926+
927+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, first, newWorkerTestStream())
928+ require.True(t, ok)
929+ select {
930+ case data := <-delivered:
931+ require.Equal(t, []byte("ab"), data)
932+ case <-time.After(time.Second):
933+ t.Fatal("first attempt never received input")
934+ }
935+ run.Cancel()
936+ select {
937+ case <-preserved:
938+ case <-time.After(time.Second):
939+ t.Fatal("worker was not woken to preserve input")
940+ }
941+ mu.Lock()
942+ require.Empty(t, refusals, "finalization must refuse every action except Ack/Preserve")
943+ mu.Unlock()
944+ require.NoError(t, <-written)
945+ require.Nil(t, host.authority.foreground())
946+
947+ // The next generation replays the preserved bytes once, not the rejected
948+ // duplicate.
949+ observed := make(chan []byte, 2)
950+ second := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
951+ event, ok := fg.Input(ctx)
952+ if !ok {
953+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("second attempt saw no input")}
954+ }
955+ fg.AckInput()
956+ observed <- append([]byte(nil), event.Data...)
957+ return AttachmentEvent{Kind: AttachmentEventEnded}
958+ }}
959+ _, adopted := host.Run(context.Background(), AttachmentToken{2, 1}, second, newWorkerTestStream())
960+ require.True(t, adopted)
961+ require.Equal(t, []byte("ab"), <-observed)
962+
963+ // A third generation sees only fresh input: the preserved read was consumed
964+ // exactly once.
965+ freshWritten := make(chan error, 1)
966+ go func() { _, err := pw.Write([]byte("fresh")); freshWritten <- err }()
967+ _, adopted = host.Run(context.Background(), AttachmentToken{3, 1}, second, newWorkerTestStream())
968+ require.True(t, adopted)
969+ require.Equal(t, []byte("fresh"), <-observed)
970+ require.NoError(t, <-freshWritten)
971+}
972+
973+func TestAttachmentWorkerRetirementPrecedesWorkerCancellation(t *testing.T) {
974+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
975+ ready := make(chan *attachmentRun, 1)
976+ checked := make(chan bool, 1)
977+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
978+ run := <-ready
979+ <-ctx.Done()
980+ select {
981+ case <-run.retired:
982+ checked <- true
983+ default:
984+ checked <- false
985+ }
986+ return AttachmentEvent{Kind: AttachmentEventEnded}
987+ }}
988+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, worker, newWorkerTestStream())
989+ require.True(t, ok)
990+ ready <- run
991+ run.Cancel()
992+ require.True(t, <-checked)
993+ _, adopted := run.Wait(context.Background())
994+ require.False(t, adopted)
995+}
996+
997+func TestAttachmentWorkerBoundEventUsesGrantedGeneration(t *testing.T) {
998+ host := newAttachmentHost(attachmentHostConfig{Input: newTerminalInputPump(nil)})
999+ event, adopted := host.Run(context.Background(), AttachmentToken{99, 42}, &fakeWorker{run: func(context.Context, AttachmentForeground) AttachmentEvent {
1000+ return AttachmentEvent{Token: AttachmentToken{999, 999}, Kind: AttachmentEventEnded}
1001+ }}, newWorkerTestStream())
1002+ require.True(t, adopted)
1003+ require.Equal(t, AttachmentToken{99, 42}, event.Token)
1004+}
@@ -0,0 +1,1004 @@
1+package client
2+
3+import (
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+// workerTestStream is one scripted broker logical stream. It records typed
21+// client messages and unblocks ReceiveServer when the supervisor closes it,
22+// exactly like a real logical stream's Close contract.
23+type workerTestStream struct {
24+ mu sync.Mutex
25+ sent []protocol.ClientMessage
26+ closed chan struct{}
27+ closeOnce sync.Once
28+ // closeHook, when set, runs during Close. It lets a test observe the
29+ // supervisor's revocation-before-close ordering.
30+ closeHook func()
31+}
32+
33+func newWorkerTestStream() *workerTestStream {
34+ return &workerTestStream{closed: make(chan struct{})}
35+}
36+
37+func (s *workerTestStream) SendClient(message protocol.ClientMessage) error {
38+ select {
39+ case <-s.closed:
40+ return errors.New("worker test stream is closed")
41+ default:
42+ }
43+ s.mu.Lock()
44+ s.sent = append(s.sent, message)
45+ s.mu.Unlock()
46+ return nil
47+}
48+
49+func (s *workerTestStream) ReceiveServer() (protocol.ServerMessage, error) {
50+ <-s.closed
51+ return nil, io.EOF
52+}
53+
54+func (s *workerTestStream) Capabilities() protocol.ConnectionCapabilities {
55+ return protocol.ConnectionCapabilities{}
56+}
57+
58+func (s *workerTestStream) LinkState() ports.LinkState { return 0 }
59+
60+func (s *workerTestStream) LinkEvents() <-chan ports.LinkEvent { return nil }
61+
62+func (s *workerTestStream) Close() error {
63+ s.closeOnce.Do(func() {
64+ close(s.closed)
65+ if s.closeHook != nil {
66+ s.closeHook()
67+ }
68+ })
69+ return nil
70+}
71+
72+func (s *workerTestStream) Done() <-chan struct{} { return s.closed }
73+
74+func (s *workerTestStream) Err() error { return nil }
75+
76+func (s *workerTestStream) messages() []protocol.ClientMessage {
77+ s.mu.Lock()
78+ defer s.mu.Unlock()
79+ return append([]protocol.ClientMessage(nil), s.sent...)
80+}
81+
82+func (s *workerTestStream) isClosed() bool {
83+ select {
84+ case <-s.closed:
85+ return true
86+ default:
87+ return false
88+ }
89+}
90+
91+// workerTestTerminal is the supervisor-owned terminal: it counts raw restores
92+// so a test can prove the worker mechanism never restores, captures written
93+// output, and doubles as the UI publication transaction.
94+type workerTestTerminal struct {
95+ mu sync.Mutex
96+ out bytes.Buffer
97+ flushes int
98+ restores int
99+ resizes chan domain.Geometry
100+ published []ports.UIContext
101+ steps []string
102+ successes []bool
103+ writeErr error
104+ shortWrite int
105+ flushErr error
106+ publishErr error
107+}
108+
109+func newWorkerTestTerminal() *workerTestTerminal {
110+ return &workerTestTerminal{resizes: make(chan domain.Geometry, 4)}
111+}
112+
113+func (t *workerTestTerminal) EnterRaw() (func() error, error) {
114+ return func() error {
115+ t.mu.Lock()
116+ t.restores++
117+ t.mu.Unlock()
118+ return nil
119+ }, nil
120+}
121+
122+func (t *workerTestTerminal) Geometry() (domain.Geometry, error) { return domain.Geometry{}, nil }
123+
124+func (t *workerTestTerminal) ResizeEvents() <-chan domain.Geometry { return t.resizes }
125+
126+func (t *workerTestTerminal) In() io.Reader { return nil }
127+
128+func (t *workerTestTerminal) Out() io.Writer { return &workerTestWriter{t} }
129+
130+func (t *workerTestTerminal) Flush() error {
131+ t.mu.Lock()
132+ t.flushes++
133+ t.steps = append(t.steps, "flush")
134+ t.mu.Unlock()
135+ return t.flushErr
136+}
137+
138+func (t *workerTestTerminal) BeginOutput(ports.UIContext) {
139+ t.mu.Lock()
140+ defer t.mu.Unlock()
141+ t.steps = append(t.steps, "begin")
142+}
143+
144+func (t *workerTestTerminal) EndOutput(success bool) {
145+ t.mu.Lock()
146+ defer t.mu.Unlock()
147+ t.steps = append(t.steps, "end")
148+ t.successes = append(t.successes, success)
149+}
150+
151+func (t *workerTestTerminal) PublishContext(context ports.UIContext) error {
152+ t.mu.Lock()
153+ defer t.mu.Unlock()
154+ t.published = append(t.published, context)
155+ t.steps = append(t.steps, "publish")
156+ return t.publishErr
157+}
158+
159+func (t *workerTestTerminal) written() string {
160+ t.mu.Lock()
161+ defer t.mu.Unlock()
162+ return t.out.String()
163+}
164+
165+func (t *workerTestTerminal) flushCount() int {
166+ t.mu.Lock()
167+ defer t.mu.Unlock()
168+ return t.flushes
169+}
170+
171+func (t *workerTestTerminal) restoreCount() int {
172+ t.mu.Lock()
173+ defer t.mu.Unlock()
174+ return t.restores
175+}
176+
177+func (t *workerTestTerminal) publications() []ports.UIContext {
178+ t.mu.Lock()
179+ defer t.mu.Unlock()
180+ return append([]ports.UIContext(nil), t.published...)
181+}
182+
183+type workerTestWriter struct{ term *workerTestTerminal }
184+
185+func (w *workerTestWriter) Write(data []byte) (int, error) {
186+ w.term.mu.Lock()
187+ defer w.term.mu.Unlock()
188+ w.term.steps = append(w.term.steps, "write")
189+ if w.term.writeErr != nil {
190+ return 0, w.term.writeErr
191+ }
192+ if w.term.shortWrite > 0 {
193+ n := min(w.term.shortWrite, len(data))
194+ _, _ = w.term.out.Write(data[:n])
195+ return n, nil
196+ }
197+ return w.term.out.Write(data)
198+}
199+
200+// fakeWorker adapts a function to the AttachmentWorker contract.
201+type fakeWorker struct {
202+ run func(ctx context.Context, fg AttachmentForeground) AttachmentEvent
203+}
204+
205+func (w *fakeWorker) Run(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
206+ return w.run(ctx, fg)
207+}
208+
209+// blockingWorker ignores input/output and parks until its context ends.
210+func blockingWorker() *fakeWorker {
211+ return &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
212+ <-ctx.Done()
213+ return AttachmentEvent{Kind: AttachmentEventEnded}
214+ }}
215+}
216+
217+func newWorkerTestHost(term *workerTestTerminal, input *terminalInputPump, onAttached func(AttachmentToken)) *attachmentHost {
218+ return newAttachmentHost(attachmentHostConfig{
219+ Terminal: term,
220+ Clock: newSupervisorTestClock(),
221+ Input: input,
222+ OnAttached: onAttached,
223+ })
224+}
225+
226+// TestAttachmentWorkerExclusiveForeground proves exactly one worker generation
227+// owns the terminal foreground at a time: a second Begin is refused while the
228+// first is live and succeeds once the first is cancelled.
229+func TestAttachmentWorkerExclusiveForeground(t *testing.T) {
230+ term := newWorkerTestTerminal()
231+ host := newWorkerTestHost(term, nil, nil)
232+
233+ run1, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, blockingWorker(), newWorkerTestStream())
234+ require.True(t, ok)
235+ require.NotNil(t, run1)
236+
237+ _, ok = host.Begin(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, blockingWorker(), newWorkerTestStream())
238+ require.False(t, ok, "second foreground must be refused while the first is live")
239+ require.Equal(t, AttachmentToken{Generation: 1, Attempt: 1}, host.authority.foreground().Token())
240+
241+ run1.Cancel()
242+ require.Nil(t, host.authority.foreground())
243+
244+ run2, ok := host.Begin(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, blockingWorker(), newWorkerTestStream())
245+ require.True(t, ok, "a fresh generation may own the foreground after cancellation")
246+ run2.Cancel()
247+ require.Nil(t, host.authority.foreground())
248+}
249+
250+// TestAttachmentWorkerZeroTokenRefused proves an unassigned token is never
251+// granted, so a worker can never run without a supervisor generation.
252+func TestAttachmentWorkerZeroTokenRefused(t *testing.T) {
253+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
254+ _, ok := host.Begin(context.Background(), AttachmentToken{}, blockingWorker(), newWorkerTestStream())
255+ require.False(t, ok)
256+ require.Nil(t, host.authority.foreground())
257+}
258+
259+// TestAttachmentWorkerCanceledGenerationCannotAct proves cancellation revokes
260+// input/output authority: a canceled generation cannot write, flush, consume or
261+// forward input, resize, publish UI, or mark itself attached, and its grant is
262+// closed.
263+func TestAttachmentWorkerCanceledGenerationCannotAct(t *testing.T) {
264+ term := newWorkerTestTerminal()
265+ host := newWorkerTestHost(term, nil, nil)
266+ stream := newWorkerTestStream()
267+
268+ fgReady := make(chan AttachmentForeground, 1)
269+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
270+ fgReady <- fg
271+ <-ctx.Done()
272+ return AttachmentEvent{Kind: AttachmentEventEnded}
273+ }}
274+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 4, Attempt: 2}, worker, stream)
275+ require.True(t, ok)
276+ fg := <-fgReady
277+ require.Equal(t, AttachmentToken{Generation: 4, Attempt: 2}, fg.Token())
278+
279+ run.Cancel()
280+
281+ select {
282+ case <-fg.Done():
283+ default:
284+ t.Fatal("grant Done must be closed after cancellation")
285+ }
286+ require.ErrorIs(t, fg.Output(ports.UIContext{}, []byte("late")), errAttachmentForegroundRevoked)
287+ require.ErrorIs(t, fg.Output(ports.UIContext{AttachmentHandle: "late"}, nil), errAttachmentForegroundRevoked)
288+ require.False(t, fg.MarkAttached())
289+ require.False(t, fg.Attached())
290+
291+ _, okInput := fg.Input(context.Background())
292+ require.False(t, okInput)
293+ _, okResize := fg.Resize(context.Background())
294+ require.False(t, okResize)
295+
296+ require.Empty(t, term.written(), "a canceled generation must not write the terminal")
297+ require.Empty(t, term.publications(), "a canceled generation must not publish UI")
298+ require.True(t, stream.isClosed(), "the supervisor closes the owned stream")
299+}
300+
301+// TestAttachmentWorkerConsumesAndForwardsInput proves an authorized worker
302+// consumes the shared terminal pump and forwards the bytes on its own stream.
303+func TestAttachmentWorkerConsumesAndForwardsInput(t *testing.T) {
304+ term := newWorkerTestTerminal()
305+ pr, pw := io.Pipe()
306+ pump := newTerminalInputPump(pr)
307+ pump.start()
308+ defer pump.stop()
309+ defer pw.Close()
310+
311+ host := newWorkerTestHost(term, pump, nil)
312+ stream := newWorkerTestStream()
313+
314+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
315+ event, ok := fg.Input(ctx)
316+ if !ok {
317+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("no authorized input")}
318+ }
319+ fg.AckInput()
320+ if err := fg.Stream().SendClient(protocol.Input{InputSeq: 1, Data: event.Data}); err != nil {
321+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
322+ }
323+ return AttachmentEvent{Kind: AttachmentEventEnded}
324+ }}
325+
326+ go func() { _, _ = pw.Write([]byte("hello")) }()
327+
328+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
329+ require.True(t, adopted)
330+ require.Equal(t, AttachmentEventEnded, event.Kind)
331+
332+ messages := stream.messages()
333+ require.Len(t, messages, 1)
334+ inputMessage, ok := messages[0].(protocol.Input)
335+ require.True(t, ok)
336+ require.Equal(t, []byte("hello"), inputMessage.Data)
337+}
338+
339+// TestAttachmentWorkerResizeAuthorized proves an authorized worker observes the
340+// supervisor's resize events and forwards them on its stream.
341+func TestAttachmentWorkerResizeAuthorized(t *testing.T) {
342+ term := newWorkerTestTerminal()
343+ host := newWorkerTestHost(term, nil, nil)
344+ stream := newWorkerTestStream()
345+ geometry := domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}
346+
347+ saw := make(chan domain.Geometry, 1)
348+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
349+ term.resizes <- geometry
350+ resize, ok := fg.Resize(ctx)
351+ if !ok {
352+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("no authorized resize")}
353+ }
354+ saw <- resize
355+ if err := fg.Stream().SendClient(protocol.Resize{Size: resize.Size, PixelWidth: resize.PixelWidth, PixelHeight: resize.PixelHeight}); err != nil {
356+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
357+ }
358+ return AttachmentEvent{Kind: AttachmentEventEnded}
359+ }}
360+
361+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
362+ require.True(t, adopted)
363+ require.Equal(t, AttachmentEventEnded, event.Kind)
364+ require.Equal(t, geometry, <-saw)
365+
366+ messages := stream.messages()
367+ require.Len(t, messages, 1)
368+ _, ok := messages[0].(protocol.Resize)
369+ require.True(t, ok)
370+}
371+
372+// TestAttachmentWorkerPublishesUIAuthorized proves an authorized worker may
373+// publish UI context and mark the attached transition, and that the grant's
374+// generation is stamped onto the publication.
375+func TestAttachmentWorkerPublishesUIAuthorized(t *testing.T) {
376+ term := newWorkerTestTerminal()
377+ attached := make(chan AttachmentToken, 1)
378+ host := newWorkerTestHost(term, nil, func(token AttachmentToken) { attached <- token })
379+ stream := newWorkerTestStream()
380+
381+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
382+ if err := fg.Output(ports.UIContext{}, nil); err != nil {
383+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
384+ }
385+ if !fg.MarkAttached() {
386+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("attach refused")}
387+ }
388+ if !fg.Attached() {
389+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("attach not recorded")}
390+ }
391+ return AttachmentEvent{Kind: AttachmentEventEnded}
392+ }}
393+
394+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 7, Attempt: 2}, worker, stream)
395+ require.True(t, adopted)
396+ require.Equal(t, AttachmentEventEnded, event.Kind)
397+
398+ publications := term.publications()
399+ require.Len(t, publications, 1)
400+ require.Equal(t, uint64(7), publications[0].Generation, "the granted generation is stamped")
401+ require.Equal(t, AttachmentToken{Generation: 7, Attempt: 2}, <-attached)
402+}
403+
404+// TestAttachmentWorkerEventReturnsToPickerWithoutTerminalRestore proves a
405+// worker end, stream loss, or typed failure returns a typed event stamped with
406+// the granted token while the worker mechanism never restores the terminal and
407+// never closes the process.
408+func TestAttachmentWorkerEventReturnsToPickerWithoutTerminalRestore(t *testing.T) {
409+ cases := []struct {
410+ name string
411+ kind AttachmentEventKind
412+ err error
413+ }{
414+ {name: "ended", kind: AttachmentEventEnded},
415+ {name: "lost", kind: AttachmentEventLost, err: errors.New("carriage lost")},
416+ {name: "failed", kind: AttachmentEventFailed, err: &ProtocolError{Code: 7, Text: "bad welcome"}},
417+ }
418+ for _, tc := range cases {
419+ t.Run(tc.name, func(t *testing.T) {
420+ term := newWorkerTestTerminal()
421+ host := newWorkerTestHost(term, nil, nil)
422+ stream := newWorkerTestStream()
423+ // The worker reports a spoofed token; the host must override it with
424+ // the granted generation/attempt.
425+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
426+ return AttachmentEvent{Token: AttachmentToken{Generation: 999}, Kind: tc.kind, Err: tc.err}
427+ }}
428+
429+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: 3, Attempt: 5}, worker, stream)
430+ require.True(t, adopted)
431+ require.Equal(t, tc.kind, event.Kind)
432+ require.Equal(t, AttachmentToken{Generation: 3, Attempt: 5}, event.Token)
433+ if tc.err != nil {
434+ require.ErrorIs(t, event.Err, tc.err)
435+ } else {
436+ require.NoError(t, event.Err)
437+ }
438+ require.Equal(t, 0, term.restoreCount(), "the worker mechanism must not restore the terminal")
439+ require.True(t, stream.isClosed(), "the supervisor closes the stream after the worker settles")
440+ require.Nil(t, host.authority.foreground(), "the foreground is released after the worker settles")
441+ })
442+ }
443+}
444+
445+// TestAttachmentWorkerLateEventDiscardedOnCancellation proves a worker that
446+// settles after the supervisor cancelled is joined and its late event is
447+// discarded rather than returned as an adopted result.
448+func TestAttachmentWorkerLateEventDiscardedOnCancellation(t *testing.T) {
449+ term := newWorkerTestTerminal()
450+ host := newWorkerTestHost(term, nil, nil)
451+ stream := newWorkerTestStream()
452+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, blockingWorker(), stream)
453+ require.True(t, ok)
454+
455+ ctx, cancel := context.WithCancel(context.Background())
456+ cancel()
457+
458+ event, adopted := run.Wait(ctx)
459+ require.False(t, adopted)
460+ require.Equal(t, AttachmentEvent{}, event, "a late event from a canceled run is discarded")
461+ require.True(t, stream.isClosed())
462+ require.Nil(t, host.authority.foreground())
463+}
464+
465+// TestAttachmentWorkerPreservesResidualInputAcrossAttempts proves the reused
466+// input pump keeps undecided bytes for the next authorized generation instead of
467+// duplicating a reader.
468+func TestAttachmentWorkerPreservesResidualInputAcrossAttempts(t *testing.T) {
469+ term := newWorkerTestTerminal()
470+ pr, pw := io.Pipe()
471+ pump := newTerminalInputPump(pr)
472+ pump.start()
473+ defer pump.stop()
474+ defer pw.Close()
475+
476+ host := newWorkerTestHost(term, pump, nil)
477+
478+ observed := make(chan []byte, 2)
479+ first := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
480+ event, ok := fg.Input(ctx)
481+ if !ok {
482+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("first attempt saw no input")}
483+ }
484+ observed <- append([]byte(nil), event.Data...)
485+ // Leave the bytes undecided: the next attempt must see them again.
486+ fg.PreserveInput(event.Data)
487+ return AttachmentEvent{Kind: AttachmentEventEnded}
488+ }}
489+ second := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
490+ event, ok := fg.Input(ctx)
491+ if !ok {
492+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("second attempt saw no input")}
493+ }
494+ fg.AckInput()
495+ observed <- append([]byte(nil), event.Data...)
496+ return AttachmentEvent{Kind: AttachmentEventEnded}
497+ }}
498+
499+ go func() { _, _ = pw.Write([]byte("ab")) }()
500+
501+ _, adopted := host.Run(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, first, newWorkerTestStream())
502+ require.True(t, adopted)
503+ _, adopted = host.Run(context.Background(), AttachmentToken{Generation: 2, Attempt: 1}, second, newWorkerTestStream())
504+ require.True(t, adopted)
505+
506+ require.Equal(t, []byte("ab"), <-observed)
507+ require.Equal(t, []byte("ab"), <-observed, "the replacement generation replays the preserved bytes")
508+ go func() { _, _ = pw.Write([]byte("fresh")) }()
509+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
510+ defer cancel()
511+ _, adopted = host.Run(ctx, AttachmentToken{Generation: 3, Attempt: 3}, second, newWorkerTestStream())
512+ require.True(t, adopted)
513+ require.Equal(t, []byte("fresh"), <-observed, "third attempt must not replay the original pending read")
514+
515+}
516+
517+// TestAttachmentWorkerRevokesBeforeClosingStream proves the supervisor revokes
518+// input/output authority before it closes the owned stream: at the instant of
519+// Close the foreground is already released and a late write is refused.
520+func TestAttachmentWorkerRevokesBeforeClosingStream(t *testing.T) {
521+ term := newWorkerTestTerminal()
522+ host := newWorkerTestHost(term, nil, nil)
523+ stream := newWorkerTestStream()
524+
525+ fgReady := make(chan AttachmentForeground, 1)
526+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
527+ fgReady <- fg
528+ <-ctx.Done()
529+ return AttachmentEvent{Kind: AttachmentEventEnded}
530+ }}
531+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: 1, Attempt: 1}, worker, stream)
532+ require.True(t, ok)
533+ fg := <-fgReady
534+
535+ var closeWriteErr error
536+ stream.closeHook = func() {
537+ require.Nil(t, host.authority.foreground(), "authority must be revoked before the stream closes")
538+ closeWriteErr = fg.Output(ports.UIContext{}, []byte("late"))
539+ }
540+
541+ run.Cancel()
542+ require.ErrorIs(t, closeWriteErr, errAttachmentForegroundRevoked)
543+ require.Empty(t, term.written())
544+}
545+
546+// TestAttachmentWorkerConcurrentBeginSingleOwner proves concurrent Begin calls
547+// admit exactly one foreground owner under the race detector.
548+func TestAttachmentWorkerConcurrentBeginSingleOwner(t *testing.T) {
549+ term := newWorkerTestTerminal()
550+ host := newWorkerTestHost(term, nil, nil)
551+
552+ const contenders = 8
553+ var wg sync.WaitGroup
554+ var successes atomic.Int64
555+ runs := make(chan *attachmentRun, contenders)
556+ start := make(chan struct{})
557+ for i := 0; i < contenders; i++ {
558+ wg.Add(1)
559+ go func(i int) {
560+ defer wg.Done()
561+ <-start
562+ run, ok := host.Begin(context.Background(), AttachmentToken{Generation: uint64(i + 1), Attempt: 1}, blockingWorker(), newWorkerTestStream())
563+ if ok {
564+ successes.Add(1)
565+ runs <- run
566+ }
567+ }(i)
568+ }
569+ close(start)
570+ wg.Wait()
571+ close(runs)
572+
573+ require.Equal(t, int64(1), successes.Load(), "exactly one concurrent Begin may own the foreground")
574+ for run := range runs {
575+ run.Cancel()
576+ }
577+ require.Nil(t, host.authority.foreground())
578+}
579+
580+// TestAttachmentWorkerRepeatedAttemptsNoLeak proves repeated grant/cancel cycles
581+// release the foreground, close every stream, and never leave a live owner.
582+func TestAttachmentWorkerRepeatedAttemptsNoLeak(t *testing.T) {
583+ term := newWorkerTestTerminal()
584+ host := newWorkerTestHost(term, nil, nil)
585+
586+ const attempts = 64
587+ for i := 0; i < attempts; i++ {
588+ stream := newWorkerTestStream()
589+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
590+ if err := fg.Output(ports.UIContext{}, []byte("x")); err != nil {
591+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: err}
592+ }
593+ return AttachmentEvent{Kind: AttachmentEventEnded}
594+ }}
595+ event, adopted := host.Run(context.Background(), AttachmentToken{Generation: uint64(i + 1), Attempt: 1}, worker, stream)
596+ require.True(t, adopted)
597+ require.Equal(t, AttachmentEventEnded, event.Kind)
598+ require.True(t, stream.isClosed())
599+ require.Nil(t, host.authority.foreground())
600+ }
601+ require.Equal(t, attempts, term.flushCount())
602+ require.Equal(t, 0, term.restoreCount(), "the mechanism never enters or restores raw mode")
603+}
604+
605+func TestAttachmentWorkerRetiredOutcomeNeverAdopted(t *testing.T) {
606+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
607+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
608+ require.True(t, ok)
609+ run.Cancel()
610+ for range 3 {
611+ event, adopted := run.Wait(context.Background())
612+ require.False(t, adopted)
613+ require.Zero(t, event)
614+ }
615+}
616+
617+func TestAttachmentWorkerBoundedJoin(t *testing.T) {
618+ clock := newSupervisorTestClock()
619+ pr, pw := io.Pipe()
620+ defer pr.Close()
621+ defer pw.Close()
622+ pump := newTerminalInputPump(pr)
623+ pump.start()
624+ defer pump.stop()
625+ host := newAttachmentHost(attachmentHostConfig{Clock: clock, Input: pump})
626+ unblock := make(chan struct{})
627+ defer close(unblock)
628+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, &fakeWorker{run: func(context.Context, AttachmentForeground) AttachmentEvent {
629+ <-unblock
630+ return AttachmentEvent{Kind: AttachmentEventEnded}
631+ }}, newWorkerTestStream())
632+ require.True(t, ok)
633+ cancelled := make(chan struct{})
634+ go func() { run.Cancel(); close(cancelled) }()
635+ timer := clock.awaitTimer(t)
636+ require.Equal(t, attachmentWorkerJoinTimeout, timer.delay)
637+ select {
638+ case <-run.retired:
639+ default:
640+ t.Fatal("retirement must precede join")
641+ }
642+ // During the final input window the grant is finalizing, not revoked: Done is
643+ // closed, output and new input are refused, the retained consumer still holds
644+ // the shared reader, and the stream is still open.
645+ require.Same(t, run.fg, host.authority.foreground())
646+ select {
647+ case <-run.fg.Done():
648+ default:
649+ t.Fatal("finalize must close Done before the join")
650+ }
651+ require.ErrorIs(t, run.fg.Output(ports.UIContext{}, []byte("late")), errAttachmentForegroundRevoked)
652+ _, ok = host.Begin(context.Background(), AttachmentToken{2, 1}, blockingWorker(), newWorkerTestStream())
653+ require.False(t, ok, "a finalizing grant blocks a competing Begin")
654+ _, claimed := pump.tryClaim()
655+ require.False(t, claimed, "the consumer is retained during the final input window")
656+ require.False(t, run.stream.(*workerTestStream).isClosed(), "the stream closes after the join")
657+ timer.fire()
658+ select {
659+ case <-cancelled:
660+ case <-time.After(time.Second):
661+ t.Fatal("uncooperative worker pinned cancel")
662+ }
663+ require.True(t, timer.stopped())
664+ // After the timeout the supervisor revokes anyway: slot freed, consumer
665+ // released, and the stream closed.
666+ require.Nil(t, host.authority.foreground())
667+ consumer, claimed := pump.tryClaim()
668+ require.True(t, claimed, "revocation releases the consumer after the join")
669+ pump.revoke(consumer)
670+ require.True(t, run.stream.(*workerTestStream).isClosed())
671+ _, adopted := run.Wait(context.Background())
672+ require.False(t, adopted)
673+}
674+
675+func TestAttachmentWorkerPumpClaimAndCancelRace(t *testing.T) {
676+ pr, pw := io.Pipe()
677+ defer pr.Close()
678+ defer pw.Close()
679+ pump := newTerminalInputPump(pr)
680+ pump.start()
681+ defer pump.stop()
682+ host := newWorkerTestHost(newWorkerTestTerminal(), pump, nil)
683+ rival := pump.claim()
684+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
685+ require.False(t, ok)
686+ require.Nil(t, host.authority.foreground())
687+ pump.revoke(rival)
688+ for i := range 100 {
689+ run, ok := host.Begin(context.Background(), AttachmentToken{uint64(i + 1), 1}, blockingWorker(), newWorkerTestStream())
690+ require.True(t, ok)
691+ var next *attachmentRun
692+ var wg sync.WaitGroup
693+ wg.Add(2)
694+ go func() { defer wg.Done(); run.Cancel() }()
695+ go func() {
696+ defer wg.Done()
697+ next, _ = host.Begin(context.Background(), AttachmentToken{uint64(i + 2), 2}, blockingWorker(), newWorkerTestStream())
698+ }()
699+ wg.Wait()
700+ if next != nil {
701+ next.Cancel()
702+ }
703+ consumer, claimed := pump.tryClaim()
704+ require.True(t, claimed)
705+ pump.revoke(consumer)
706+ }
707+}
708+
709+func TestAttachmentWorkerClosedResizeDisabled(t *testing.T) {
710+ term := newWorkerTestTerminal()
711+ close(term.resizes) // uiterm-like terminal: no physical resize events
712+ host := newWorkerTestHost(term, nil, nil)
713+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
714+ require.True(t, ok)
715+ result := make(chan bool, 1)
716+ go func() { _, ok := run.fg.Resize(context.Background()); result <- ok }()
717+ select {
718+ case <-result:
719+ t.Fatal("closed resize source terminated attachment")
720+ case <-time.After(10 * time.Millisecond):
721+ }
722+ require.True(t, run.fg.MarkAttached())
723+ run.Cancel()
724+ select {
725+ case ok := <-result:
726+ require.False(t, ok)
727+ case <-time.After(time.Second):
728+ t.Fatal("authority Done did not stop resize")
729+ }
730+}
731+
732+func TestAttachmentWorkerOutputTransaction(t *testing.T) {
733+ failure := errors.New("output failure")
734+ for _, tc := range []struct {
735+ name string
736+ writeErr, flushErr, publishErr error
737+ shortWrite int
738+ wantErr error
739+ steps []string
740+ }{
741+ {name: "success", steps: []string{"begin", "publish", "write", "flush", "end"}},
742+ {name: "short write", shortWrite: 2, wantErr: io.ErrShortWrite, steps: []string{"begin", "publish", "write", "end"}},
743+ {name: "write failure", writeErr: failure, wantErr: failure, steps: []string{"begin", "publish", "write", "end"}},
744+ {name: "flush failure", flushErr: failure, wantErr: failure, steps: []string{"begin", "publish", "write", "flush", "end"}},
745+ {name: "publish failure", publishErr: failure, wantErr: failure, steps: []string{"begin", "publish", "end"}},
746+ {name: "publish unavailable tolerated", publishErr: ports.ErrUIUnavailable, steps: []string{"begin", "publish", "write", "flush", "end"}},
747+ } {
748+ t.Run(tc.name, func(t *testing.T) {
749+ term := newWorkerTestTerminal()
750+ term.writeErr, term.flushErr, term.publishErr, term.shortWrite = tc.writeErr, tc.flushErr, tc.publishErr, tc.shortWrite
751+ host := newWorkerTestHost(term, nil, nil)
752+ run, ok := host.Begin(context.Background(), AttachmentToken{9, 2}, blockingWorker(), newWorkerTestStream())
753+ require.True(t, ok)
754+ defer run.Cancel()
755+ err := run.fg.Output(ports.UIContext{Generation: 999}, []byte("frame"))
756+ success := tc.wantErr == nil
757+ if success {
758+ require.NoError(t, err)
759+ } else {
760+ require.ErrorIs(t, err, tc.wantErr)
761+ }
762+ require.Equal(t, tc.steps, term.steps)
763+ require.Equal(t, []bool{success}, term.successes)
764+ require.Equal(t, uint64(9), term.published[0].Generation)
765+ })
766+ }
767+}
768+
769+// Hide the transaction methods to model a physical terminal without UI output.
770+type workerBareTerminal struct{ ports.Terminal }
771+
772+func TestAttachmentWorkerNilSeams(t *testing.T) {
773+ var nilTerm *workerTestTerminal
774+ var nilWorker *fakeWorker
775+ var nilStream *workerTestStream
776+ var nilClock *supervisorTestClock
777+ for _, cfg := range []attachmentHostConfig{
778+ {},
779+ {Terminal: nilTerm, UI: nilTerm, Clock: nilClock},
780+ {Terminal: workerBareTerminal{newWorkerTestTerminal()}, UI: nilTerm},
781+ } {
782+ host := newAttachmentHost(cfg)
783+ _, ok := host.Begin(nil, AttachmentToken{1, 1}, nilWorker, newWorkerTestStream())
784+ require.False(t, ok)
785+ _, ok = host.Begin(nil, AttachmentToken{1, 1}, blockingWorker(), nilStream)
786+ require.False(t, ok)
787+ run, ok := host.Begin(nil, AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
788+ require.True(t, ok)
789+ require.ErrorIs(t, run.fg.Output(ports.UIContext{}, nil), ports.ErrUIUnavailable)
790+ run.Cancel()
791+ require.ErrorIs(t, run.fg.Output(ports.UIContext{}, nil), errAttachmentForegroundRevoked)
792+ }
793+}
794+
795+// TestAttachmentWorkerGenerationIsSupervisorSupplied proves the host no longer
796+// allocates a UI generation: it grants and stamps the caller's generation, and
797+// the same value can be re-granted across sequential runs because the
798+// supervisor owns the counter. UI action binding stays deferred to the P5
799+// supervisor, so no UI admission is published here; the transactional output
800+// path still stamps the granted generation on every publication.
801+func TestAttachmentWorkerGenerationIsSupervisorSupplied(t *testing.T) {
802+ term := newWorkerTestTerminal()
803+ pump := newTerminalInputPump(nil)
804+ host := newWorkerTestHost(term, pump, nil)
805+ for i := uint64(1); i <= 3; i++ {
806+ run, ok := host.Begin(context.Background(), AttachmentToken{i, i + 10}, blockingWorker(), newWorkerTestStream())
807+ require.True(t, ok)
808+ require.Equal(t, AttachmentToken{i, i + 10}, run.fg.Token())
809+ require.NoError(t, run.fg.Output(ports.UIContext{Generation: 999}, nil))
810+ require.Equal(t, i, term.publications()[i-1].Generation)
811+ run.Cancel()
812+ }
813+}
814+
815+// TestAttachmentWorkerBeginRefusalsRetainStream proves every Begin refusal path
816+// leaves the caller's stream untouched: it is not closed, receives no message,
817+// no foreground is granted, and the caller keeps ownership.
818+func TestAttachmentWorkerBeginRefusalsRetainStream(t *testing.T) {
819+ t.Run("nil host", func(t *testing.T) {
820+ var host *attachmentHost
821+ stream := newWorkerTestStream()
822+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), stream)
823+ require.False(t, ok)
824+ require.Nil(t, run)
825+ require.False(t, stream.isClosed())
826+ require.Empty(t, stream.messages())
827+ })
828+ t.Run("zero token", func(t *testing.T) {
829+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
830+ stream := newWorkerTestStream()
831+ _, ok := host.Begin(context.Background(), AttachmentToken{}, blockingWorker(), stream)
832+ require.False(t, ok)
833+ require.False(t, stream.isClosed())
834+ require.Empty(t, stream.messages())
835+ require.Nil(t, host.authority.foreground())
836+ })
837+ t.Run("nil worker", func(t *testing.T) {
838+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
839+ stream := newWorkerTestStream()
840+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, nil, stream)
841+ require.False(t, ok)
842+ require.False(t, stream.isClosed())
843+ require.Nil(t, host.authority.foreground())
844+ })
845+ t.Run("nil stream", func(t *testing.T) {
846+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
847+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), nil)
848+ require.False(t, ok)
849+ require.Nil(t, host.authority.foreground())
850+ })
851+ t.Run("foreground already live", func(t *testing.T) {
852+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
853+ first, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), newWorkerTestStream())
854+ require.True(t, ok)
855+ stream := newWorkerTestStream()
856+ _, ok = host.Begin(context.Background(), AttachmentToken{2, 1}, blockingWorker(), stream)
857+ require.False(t, ok)
858+ require.False(t, stream.isClosed(), "a refused Begin must not close the caller's stream")
859+ require.Empty(t, stream.messages())
860+ require.Same(t, first.fg, host.authority.foreground())
861+ first.Cancel()
862+ })
863+ t.Run("rival consumer owns input", func(t *testing.T) {
864+ pump := newTerminalInputPump(nil)
865+ host := newWorkerTestHost(newWorkerTestTerminal(), pump, nil)
866+ rival := pump.claim()
867+ defer pump.revoke(rival)
868+ stream := newWorkerTestStream()
869+ _, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, blockingWorker(), stream)
870+ require.False(t, ok)
871+ require.False(t, stream.isClosed())
872+ require.Empty(t, stream.messages())
873+ require.Nil(t, host.authority.foreground())
874+ })
875+}
876+
877+// TestAttachmentWorkerFinalizePreservesUndecidedInput proves the deterministic
878+// final input window: cancellation wakes the worker through Done while the pump
879+// consumer is still authorized for Ack/Preserve only, so the undecided read
880+// survives to the next generation exactly once, and a duplicate PreserveInput
881+// for the same delivery can never overwrite it.
882+func TestAttachmentWorkerFinalizePreservesUndecidedInput(t *testing.T) {
883+ pr, pw := io.Pipe()
884+ defer pr.Close()
885+ defer pw.Close()
886+ pump := newTerminalInputPump(pr)
887+ pump.start()
888+ defer pump.stop()
889+ host := newAttachmentHost(attachmentHostConfig{Input: pump})
890+
891+ var mu sync.Mutex
892+ var refusals []string
893+ refuse := func(what string) {
894+ mu.Lock()
895+ refusals = append(refusals, what)
896+ mu.Unlock()
897+ }
898+ delivered := make(chan []byte, 1)
899+ preserved := make(chan struct{})
900+ first := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
901+ event, ok := fg.Input(ctx)
902+ if !ok {
903+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("first attempt saw no input")}
904+ }
905+ delivered <- append([]byte(nil), event.Data...)
906+ // The supervisor finalizes: Done wakes us while Ack/Preserve stay
907+ // authorized and every other action is refused.
908+ <-fg.Done()
909+ if _, ok := fg.Input(ctx); ok {
910+ refuse("new input")
911+ }
912+ if _, ok := fg.Resize(ctx); ok {
913+ refuse("resize")
914+ }
915+ if err := fg.Output(ports.UIContext{}, []byte("late")); !errors.Is(err, errAttachmentForegroundRevoked) {
916+ refuse("output")
917+ }
918+ fg.PreserveInput(event.Data)
919+ fg.PreserveInput([]byte("overwrite"))
920+ close(preserved)
921+ return AttachmentEvent{Kind: AttachmentEventEnded}
922+ }}
923+
924+ written := make(chan error, 1)
925+ go func() { _, err := pw.Write([]byte("ab")); written <- err }()
926+
927+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, first, newWorkerTestStream())
928+ require.True(t, ok)
929+ select {
930+ case data := <-delivered:
931+ require.Equal(t, []byte("ab"), data)
932+ case <-time.After(time.Second):
933+ t.Fatal("first attempt never received input")
934+ }
935+ run.Cancel()
936+ select {
937+ case <-preserved:
938+ case <-time.After(time.Second):
939+ t.Fatal("worker was not woken to preserve input")
940+ }
941+ mu.Lock()
942+ require.Empty(t, refusals, "finalization must refuse every action except Ack/Preserve")
943+ mu.Unlock()
944+ require.NoError(t, <-written)
945+ require.Nil(t, host.authority.foreground())
946+
947+ // The next generation replays the preserved bytes once, not the rejected
948+ // duplicate.
949+ observed := make(chan []byte, 2)
950+ second := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
951+ event, ok := fg.Input(ctx)
952+ if !ok {
953+ return AttachmentEvent{Kind: AttachmentEventFailed, Err: errors.New("second attempt saw no input")}
954+ }
955+ fg.AckInput()
956+ observed <- append([]byte(nil), event.Data...)
957+ return AttachmentEvent{Kind: AttachmentEventEnded}
958+ }}
959+ _, adopted := host.Run(context.Background(), AttachmentToken{2, 1}, second, newWorkerTestStream())
960+ require.True(t, adopted)
961+ require.Equal(t, []byte("ab"), <-observed)
962+
963+ // A third generation sees only fresh input: the preserved read was consumed
964+ // exactly once.
965+ freshWritten := make(chan error, 1)
966+ go func() { _, err := pw.Write([]byte("fresh")); freshWritten <- err }()
967+ _, adopted = host.Run(context.Background(), AttachmentToken{3, 1}, second, newWorkerTestStream())
968+ require.True(t, adopted)
969+ require.Equal(t, []byte("fresh"), <-observed)
970+ require.NoError(t, <-freshWritten)
971+}
972+
973+func TestAttachmentWorkerRetirementPrecedesWorkerCancellation(t *testing.T) {
974+ host := newWorkerTestHost(newWorkerTestTerminal(), nil, nil)
975+ ready := make(chan *attachmentRun, 1)
976+ checked := make(chan bool, 1)
977+ worker := &fakeWorker{run: func(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
978+ run := <-ready
979+ <-ctx.Done()
980+ select {
981+ case <-run.retired:
982+ checked <- true
983+ default:
984+ checked <- false
985+ }
986+ return AttachmentEvent{Kind: AttachmentEventEnded}
987+ }}
988+ run, ok := host.Begin(context.Background(), AttachmentToken{1, 1}, worker, newWorkerTestStream())
989+ require.True(t, ok)
990+ ready <- run
991+ run.Cancel()
992+ require.True(t, <-checked)
993+ _, adopted := run.Wait(context.Background())
994+ require.False(t, adopted)
995+}
996+
997+func TestAttachmentWorkerBoundEventUsesGrantedGeneration(t *testing.T) {
998+ host := newAttachmentHost(attachmentHostConfig{Input: newTerminalInputPump(nil)})
999+ event, adopted := host.Run(context.Background(), AttachmentToken{99, 42}, &fakeWorker{run: func(context.Context, AttachmentForeground) AttachmentEvent {
1000+ return AttachmentEvent{Token: AttachmentToken{999, 999}, Kind: AttachmentEventEnded}
1001+ }}, newWorkerTestStream())
1002+ require.True(t, adopted)
1003+ require.Equal(t, AttachmentToken{99, 42}, event.Token)
1004+}
M internal/usecase/client/client.go +11 -2
@@ -3655,14 +3655,23 @@ func newTerminalInputPump(in io.Reader) *terminalInputPump {
36553655 // claim grants one attach scanner exclusive permission to dequeue terminal
36563656 // input. The reader itself remains lifecycle-owned across reconnects.
36573657 func (p *terminalInputPump) claim() uint64 {
3658+ consumer, ok := p.tryClaim()
3659+ if !ok {
3660+ panic("terminal input consumer already claimed")
3661+ }
3662+ return consumer
3663+}
3664+
3665+// tryClaim lets competing foreground owners decline admission without panicking.
3666+func (p *terminalInputPump) tryClaim() (uint64, bool) {
36583667 p.mu.Lock()
36593668 defer p.mu.Unlock()
36603669 if p.consumer != 0 {
3661- panic("terminal input consumer already claimed")
3670+ return 0, false
36623671 }
36633672 p.nextID++
36643673 p.consumer = p.nextID
3665- return p.consumer
3674+ return p.consumer, true
36663675 }
36673676
36683677 // revoke invalidates an attempt before its replacement is allowed to claim
@@ -3655,14 +3655,23 @@ func newTerminalInputPump(in io.Reader) *terminalInputPump {
3655 // claim grants one attach scanner exclusive permission to dequeue terminal3655 // claim grants one attach scanner exclusive permission to dequeue terminal
3656 // input. The reader itself remains lifecycle-owned across reconnects.3656 // input. The reader itself remains lifecycle-owned across reconnects.
3657 func (p *terminalInputPump) claim() uint64 {3657 func (p *terminalInputPump) claim() uint64 {
3658+ consumer, ok := p.tryClaim()
3659+ if !ok {
3660+ panic("terminal input consumer already claimed")
3661+ }
3662+ return consumer
3663+}
3664+
3665+// tryClaim lets competing foreground owners decline admission without panicking.
3666+func (p *terminalInputPump) tryClaim() (uint64, bool) {
3658 p.mu.Lock()3667 p.mu.Lock()
3659 defer p.mu.Unlock()3668 defer p.mu.Unlock()
3660 if p.consumer != 0 {3669 if p.consumer != 0 {
3661- panic("terminal input consumer already claimed")3670+ return 0, false
3662 }3671 }
3663 p.nextID++3672 p.nextID++
3664 p.consumer = p.nextID3673 p.consumer = p.nextID
3665- return p.consumer3674+ return p.consumer, true
3666 }3675 }
3667 3676
3668 // revoke invalidates an attempt before its replacement is allowed to claim3677 // revoke invalidates an attempt before its replacement is allowed to claim

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close