1package client
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "slices"
9 "time"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 "github.com/bnema/vev/internal/protocol"
14)
15
16// Real attachment worker.
17//
18// sessionAttachmentWorker drives the existing typed session protocol on the
19// broker logical stream the supervisor opened for one resolved request. It
20// sends Hello, awaits Welcome, then awaits and commits the initial full output
21// publication. Welcome alone is never attachment: the worker marks attached
22// only after the validated frame was written, flushed, and committed through
23// the foreground's UI output transaction. After attachment it keeps consuming
24// output until the stream ends.
25//
26// The worker owns no endpoint, route, raw mode, terminal writer, or second
27// reader. It reaches the terminal only through the supervisor-granted
28// AttachmentForeground. Input, geometry and lifecycle sends therefore retain
29// the foreground token and stop when the grant is superseded.
30
31// closedDetachTimeout bounds the final Detach{Closed} send when the client
32// process is ending, so a slow broker never delays terminal shutdown.
33const closedDetachTimeout = 300 * time.Millisecond
34
35// ErrClientClosed is the cancellation cause composition uses when the client
36// process is ending (SIGHUP/SIGTERM). The attachment then sends Detach{Closed}
37// so the daemon can apply ephemeral.close-on-exit instead of parking.
38var ErrClientClosed = errors.New("client: process closed")
39
40var (
41 // errAttachmentDeadline is the stable local sentinel of an expired
42 // attachment budget. The supervisor wraps it in a typed timeout.
43 errAttachmentDeadline = errors.New("client: attachment deadline elapsed")
44 // errAttachmentNoStream reports a foreground granted without a stream.
45 errAttachmentNoStream = errors.New("client: attachment stream is missing")
46 // errAttachmentPublication reports a publication frame that violates the
47 // output contract before it was written.
48 errAttachmentPublication = errors.New("client: invalid initial publication")
49)
50
51// AttachmentIdentityError reports a server commitment that disagrees with the
52// broker request resolved from the selected picker row.
53type AttachmentIdentityError struct {
54 Stage string
55 Want protocol.ExactSessionTarget
56 Got protocol.ExactSessionTarget
57 Name string
58}
59
60func (e *AttachmentIdentityError) Error() string {
61 if e.Name != "" {
62 return fmt.Sprintf("client: %s committed session name %q, want %q", e.Stage, e.Got.SessionName, e.Name)
63 }
64 return fmt.Sprintf("client: %s committed target %+v, want %+v", e.Stage, e.Got, e.Want)
65}
66
67// sessionAttachmentConfig supplies the typed-session inputs of one worker run.
68// The request is the exact request the supervisor opened the stream with; the
69// rest is client-local presentation data the worker mirrors into Hello.
70type sessionAttachmentConfig struct {
71 Request ports.BrokerOpenStreamRequest
72 ClientID [16]byte
73 Geometry domain.Geometry
74 TermEnv string
75 Cwd string
76 TrueColor bool
77 // PID is this client's process ID; Hello carries it on local attaches.
78 PID uint32
79 // Capabilities are the outer terminal's probed capabilities, declared in
80 // Hello.
81 Capabilities terminalCapabilities
82 SessionEnvironment SessionEnvironment
83 // BeforeAttached runs after the initial frame commits but before attached
84 // presentation becomes observable. A failure keeps the run unattached.
85 BeforeAttached func() error
86 // Clock bounds the move picker's lone-escape window and the terminal input
87 // timers (DECRQM ambiguity, paste framing, palette deadlines). A nil clock
88 // uses the system clock.
89 Clock ports.Clock
90 // Theme retains terminal-reported colors across attachments. When set,
91 // the worker queries the terminal palette and sends protocol.Theme; when
92 // nil, replies are still stripped from input but no query is written.
93 Theme *terminalThemeState
94 // Clipboard reads images on remote attachments only.
95 Clipboard ports.ClipboardReader
96 // Tab is the exact tab a picker tab row committed; its zero value
97 // attaches at session level.
98 Tab attachmentTab
99 // Tabs retains this client's committed tab per authority and session.
100 Tabs *sessionTabMemory
101}
102
103// sessionAttachmentWorker implements AttachmentWorker for one broker logical
104// stream. It holds no stream or terminal of its own: everything arrives through
105// Run's foreground for the duration of one run.
106type sessionAttachmentWorker struct {
107 cfg sessionAttachmentConfig
108}
109
110// newSessionAttachmentWorker builds the real typed-session attachment worker.
111func newSessionAttachmentWorker(cfg sessionAttachmentConfig) (*sessionAttachmentWorker, error) {
112 if err := cfg.SessionEnvironment.Validate(); err != nil {
113 return nil, fmt.Errorf("client: invalid session environment: %w", err)
114 }
115 if !slices.Equal(cfg.Request.Env, cfg.SessionEnvironment.Env) {
116 return nil, errors.New("client: session environment does not match request environment")
117 }
118 cfg.Request.Env = append([]string(nil), cfg.Request.Env...)
119 cfg.SessionEnvironment = cfg.SessionEnvironment.Clone()
120 if supervisorNil(cfg.Clock) {
121 cfg.Clock = systemClock{}
122 }
123 return &sessionAttachmentWorker{cfg: cfg}, nil
124}
125
126// serverReceive is one server message or the terminal receive error.
127type serverReceive struct {
128 message protocol.ServerMessage
129 err error
130}
131
132// startServerReceive owns the single reader of the logical stream. It is
133// bounded: it buffers one message and blocks on the next read, and the stream's
134// Close contract unblocks the read. The returned stop channel releases the
135// reader when the run ends, so a message the worker no longer reads can never
136// pin it.
137func startServerReceive(stream ports.BrokerLogicalConnection) (<-chan serverReceive, chan struct{}) {
138 out := make(chan serverReceive, 1)
139 stop := make(chan struct{})
140 go func() {
141 for {
142 message, err := stream.ReceiveServer()
143 select {
144 case out <- serverReceive{message: message, err: err}:
145 case <-stop:
146 return
147 }
148 if err != nil {
149 return
150 }
151 }
152 }()
153 return out, stop
154}
155
156// Run drives Hello -> Welcome -> committed initial publication -> attached
157// output until the stream ends or the context/deadline, the supervisor finalize
158// signal, or the stream itself stops it. Every return path is a typed event;
159// the supervisor stamps the granted token onto it.
160func (w *sessionAttachmentWorker) Run(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
161 token := fg.Token()
162 if ctx == nil {
163 ctx = context.Background()
164 }
165 stream := fg.Stream()
166 if supervisorNil(stream) {
167 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errAttachmentNoStream}
168 }
169 incoming, stopReceive := startServerReceive(stream)
170 defer close(stopReceive)
171
172 if err := w.sendHello(ctx, fg, stream); err != nil {
173 return w.settle(ctx, fg, stream, token, err)
174 }
175
176 welcome, err := w.awaitServer(ctx, fg, incoming)
177 if err != nil {
178 return w.settle(ctx, fg, stream, token, err)
179 }
180 switch message := welcome.(type) {
181 case protocol.Welcome:
182 // Welcome is a promise, not attachment: the daemon still owes the
183 // committed initial publication.
184 if err := w.validateWelcome(message); err != nil {
185 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
186 }
187 case protocol.ErrorMsg:
188 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: message.Code, Text: message.Text}}
189 default:
190 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: unexpected %T before welcome", welcome)}
191 }
192
193 state, event := w.awaitInitialPublication(ctx, fg, stream, token, incoming)
194 if event != nil {
195 return *event
196 }
197
198 return w.pumpAttached(ctx, fg, stream, token, state, incoming)
199}
200
201// awaitInitialPublication consumes server messages until the first valid full
202// output frame is committed. It returns the output state to continue with, or
203// a typed terminal event.
204func (w *sessionAttachmentWorker) awaitInitialPublication(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, incoming <-chan serverReceive) (outputApplyState, *AttachmentEvent) {
205 state := outputApplyState{}
206 for {
207 message, err := w.awaitServer(ctx, fg, incoming)
208 if err != nil {
209 event := w.settle(ctx, fg, stream, token, err)
210 return state, &event
211 }
212 switch typed := message.(type) {
213 case protocol.Output:
214 next, ok := state.next(typed)
215 if !ok {
216 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errAttachmentPublication}
217 return state, &event
218 }
219 if err := w.validatePublication(typed); err != nil {
220 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
221 return state, &event
222 }
223 state = next
224 w.noteCommitted(fg, state.context)
225 // The frame transaction is the pre-attach publication: it commits the first
226 // session bytes with the Connecting presentation and, through the foreground
227 // shape rule, a public generation of zero. Attached is published only after
228 // that write, flush, and commit succeeded.
229 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusConnecting), typed.Data); err != nil {
230 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing initial output: %w", err)}
231 return state, &event
232 }
233 if err := w.send(ctx, fg, stream, protocol.Ack{Epoch: state.epoch, State: state.state}); err != nil {
234 event := w.settle(ctx, fg, stream, token, err)
235 return state, &event
236 }
237 // Attachment is the committed publication. Complete the deadline
238 // boundary before presenting attached, so attached always implies the
239 // handshake budget can no longer cancel the run.
240 if w.cfg.BeforeAttached != nil {
241 if err := w.cfg.BeforeAttached(); err != nil {
242 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
243 return state, &event
244 }
245 }
246 // A refused marker means the grant was revoked underneath us, so the
247 // run ends as a failure.
248 if !fg.MarkAttached() {
249 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errAttachmentForegroundRevoked}
250 return state, &event
251 }
252 // The attached presentation is published last, once the committed frame and
253 // the attached marker are in place. It carries the real action generation
254 // with the same validated session identity and published boundary, and it
255 // writes no bytes: the screen it commits is exactly the frame that just
256 // drained.
257 if err := fg.PublishAttached(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached)); err != nil {
258 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing attached presentation: %w", err)}
259 return state, &event
260 }
261 return state, nil
262 case protocol.ErrorMsg:
263 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}
264 return state, &event
265 case protocol.Detached:
266 // The daemon ended the attempt before publishing anything: it
267 // never attached.
268 event := AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
269 return state, &event
270 default:
271 // Everything else the daemon may publish before the initial frame
272 // stays outside this slice's scope and is ignored.
273 }
274 }
275}
276
277type attachmentClientEvent struct {
278 message protocol.ClientMessage
279 input *AttachmentInputEvent
280 action AttachmentLifecycleActionKind
281}
282
283// attachmentOverlay returns the optional overlay seam of fg, or nil.
284func attachmentOverlay(fg AttachmentForeground) attachmentOverlayForeground {
285 overlay, _ := fg.(attachmentOverlayForeground)
286 return overlay
287}
288
289func (w *sessionAttachmentWorker) noteCommitted(fg AttachmentForeground, view protocol.ViewContext) {
290 w.cfg.Tabs.remember(requestAuthority(w.cfg.Request), view.Route.Target, view.TabID)
291 if overlay := attachmentOverlay(fg); overlay != nil {
292 overlay.noteCommitted(view.Route.Target, view.TabID)
293 }
294}
295
296var (
297 errDetachToPicker = errors.New("client: detach to picker")
298 errDetachAndExit = errors.New("client: detach and exit")
299)
300
301// pumpAttached publishes output and concurrently forwards the foreground's
302// authorized input, latest geometry and explicit lifecycle decision.
303func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, state outputApplyState, incoming <-chan serverReceive) AttachmentEvent {
304 pumpCtx, cancel := context.WithCancel(ctx)
305 defer cancel()
306 outgoing := make(chan attachmentClientEvent)
307 go pumpAttachmentInput(pumpCtx, fg, outgoing)
308 go pumpAttachmentGeometry(pumpCtx, fg, outgoing)
309 go pumpAttachmentLifecycle(pumpCtx, fg, outgoing)
310 outputResetRequested := false
311 overlay := attachmentOverlay(fg)
312 var repaint <-chan struct{}
313 var tabSelections <-chan domain.TabStableID
314 var routeUpdates <-chan protocol.RecentRouteSnapshot
315 var navigationReplies <-chan protocol.ClientMessage
316 if overlay != nil {
317 repaint = overlay.overlayRepaint()
318 tabSelections = overlay.tabSelections()
319 routeUpdates = overlay.routeSnapshots()
320 navigationReplies = overlay.navigationReplies()
321 }
322 var samePeerRequests uint64
323 var samePeerSwitch *samePeerSwitchPending
324 samePeerUI := attachmentSamePeerUI(fg)
325 picker := &attachmentMovePicker{worker: w, fg: fg, overlay: overlay, stream: stream, size: w.cfg.Geometry.Size, move: newMovePickerOverlay(w.cfg.TrueColor)}
326 defer picker.stopEscape()
327 input := newAttachmentInput(ctx, w, fg, stream, picker)
328 defer input.close()
329 if err := input.start(ctx); err != nil {
330 return w.settle(ctx, fg, stream, token, err)
331 }
332 // Hello declared the focus known at attach, for the first paint. The
333 // reporter re-sends the current focus, which the daemon ignores when
334 // unchanged, then every change. The pump records a focus report before it
335 // delivers the input read with it, so flushing focus first keeps the
336 // daemon's view in input order.
337 focus := &focusReporter{state: attachmentTerminalFocus(fg)}
338 sendFocus := func() error {
339 if message, ok := focus.next(); ok {
340 return w.send(ctx, fg, stream, message)
341 }
342 return nil
343 }
344 if err := sendFocus(); err != nil {
345 return w.settle(ctx, fg, stream, token, err)
346 }
347 for {
348 select {
349 case <-focus.changed:
350 if err := sendFocus(); err != nil {
351 return w.settle(ctx, fg, stream, token, err)
352 }
353 case <-input.wake:
354 if err := input.flush(ctx); err != nil {
355 return w.settle(ctx, fg, stream, token, err)
356 }
357 case <-input.markerC():
358 if err := input.markerExpired(ctx, state); err != nil {
359 return w.settle(ctx, fg, stream, token, err)
360 }
361 case <-input.drain.c():
362 if err := input.drainFired(ctx); err != nil {
363 return w.settle(ctx, fg, stream, token, err)
364 }
365 case <-input.completion.c():
366 if err := input.completionFired(ctx); err != nil {
367 return w.settle(ctx, fg, stream, token, err)
368 }
369 case <-repaint:
370 // An overlay released the terminal: the picker box is still on screen
371 // and suppressed frames were never written, so ask the daemon for an
372 // authoritative full repaint.
373 if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil {
374 return w.settle(ctx, fg, stream, token, err)
375 }
376 outputResetRequested = true
377 case tab := <-tabSelections:
378 // The picker overlay committed another tab of this session: switch
379 // the attachment's view in place instead of reconnecting.
380 if err := w.send(ctx, fg, stream, protocol.SelectTab{TabID: tab}); err != nil {
381 return w.settle(ctx, fg, stream, token, err)
382 }
383 case snapshot := <-routeUpdates:
384 // The supervisor's route ledger changed: the serving daemon renders
385 // the status-bar MRU, bells, and palette routes from it.
386 if err := w.send(ctx, fg, stream, snapshot); err != nil {
387 return w.settle(ctx, fg, stream, token, err)
388 }
389 case reply := <-navigationReplies:
390 if err := w.send(ctx, fg, stream, reply); err != nil {
391 return w.settle(ctx, fg, stream, token, err)
392 }
393 case <-picker.escape():
394 picker.escapeFired()
395 op, changed := picker.move.flush()
396 if err := picker.applyOp(ctx, state, op, changed); err != nil {
397 return w.settle(ctx, fg, stream, token, err)
398 }
399 case result := <-incoming:
400 if result.err != nil {
401 return w.settle(ctx, fg, stream, token, result.err)
402 }
403 switch typed := result.message.(type) {
404 case protocol.Output:
405 next, ok := state.next(typed)
406 if !ok {
407 continue
408 }
409 state = next
410 outputResetRequested = false
411 if samePeerSwitch != nil && state.context.Route.Target == samePeerSwitch.target {
412 // The in-place destination committed: its publication
413 // settles the action that caused the switch.
414 samePeerSwitch = nil
415 if samePeerUI != nil {
416 samePeerUI.uiSamePeerArrived()
417 }
418 }
419 w.noteCommitted(fg, state.context)
420 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil {
421 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing output: %w", err)}
422 }
423 if err := w.send(ctx, fg, stream, protocol.Ack{Epoch: state.epoch, State: state.state}); err != nil {
424 return w.settle(ctx, fg, stream, token, err)
425 }
426 case protocol.UIViewUpdate:
427 next, accepted, needsReset := state.nextView(typed)
428 if needsReset && !outputResetRequested {
429 if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil {
430 return w.settle(ctx, fg, stream, token, err)
431 }
432 outputResetRequested = true
433 }
434 if !accepted {
435 continue
436 }
437 state = next
438 w.noteCommitted(fg, state.context)
439 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), nil); err != nil {
440 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("vev: publishing view update: %w", err)}
441 }
442 case protocol.UIReceipt:
443 attachmentUIReceipt(fg, typed)
444 case protocol.PickerOffer:
445 if err := picker.offer(ctx, typed); err != nil {
446 return w.settle(ctx, fg, stream, token, err)
447 }
448 case protocol.PickerSnapshot:
449 if err := picker.snapshot(ctx, state, typed); err != nil {
450 return w.settle(ctx, fg, stream, token, err)
451 }
452 case protocol.PickerClosed:
453 picker.closed(typed.InteractionID)
454 case protocol.AttachTarget:
455 if typed.SamePeer && typed.ExactTarget != nil {
456 // An endpoint-empty offer on this very connection: confirm
457 // it and the daemon moves this attachment in place.
458 samePeerRequests++
459 request := protocol.SamePeerSwitchRequest{
460 RequestID: samePeerRequests, Target: *typed.ExactTarget,
461 PreferredTabID: w.cfg.Tabs.preferred(requestAuthority(w.cfg.Request), *typed.ExactTarget, typed.PreferredTabID),
462 }
463 if request.Validate() != nil {
464 continue
465 }
466 if samePeerUI != nil {
467 samePeerUI.uiFollowSamePeer(typed.CauseActionID)
468 }
469 samePeerSwitch = &samePeerSwitchPending{requestID: request.RequestID, target: request.Target}
470 if err := w.send(ctx, fg, stream, request); err != nil {
471 return w.settle(ctx, fg, stream, token, err)
472 }
473 continue
474 }
475 if overlay != nil {
476 overlay.requestNavigation(typed)
477 }
478 case protocol.SamePeerSwitchFailure:
479 // The daemon refused the confirmed switch and kept the source.
480 if samePeerSwitch != nil && typed.RequestID == samePeerSwitch.requestID {
481 samePeerSwitch = nil
482 if samePeerUI != nil {
483 samePeerUI.uiSamePeerFailed()
484 }
485 }
486 case protocol.RouteNavigationAction, protocol.RouteCreateSessionAction:
487 // The daemon asks the client to navigate: the supervisor
488 // resolves it through the broker catalogue.
489 if overlay != nil {
490 overlay.requestNavigation(typed)
491 }
492 case protocol.ErrorMsg:
493 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}
494 case protocol.Detached:
495 switch typed.Reason {
496 case protocol.ReasonDetachToPicker:
497 return attachmentLifecycleEnd(token, AttachmentDetachToPicker)
498 case protocol.ReasonDetach:
499 // The daemon confirmed the user's explicit detach request: the
500 // client exits cleanly instead of falling back to the picker.
501 return attachmentLifecycleEnd(token, AttachmentDetachAndExit)
502 }
503 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
504 }
505 case event := <-outgoing:
506 if event.input != nil && event.input.Err != nil {
507 fg.PreserveInput(event.input.Data)
508 return w.settle(ctx, fg, stream, token, event.input.Err)
509 }
510 if event.input != nil {
511 if err := sendFocus(); err != nil {
512 fg.PreserveInput(event.input.Data)
513 return w.settle(ctx, fg, stream, token, err)
514 }
515 }
516 if event.input != nil && event.input.actionID == 0 {
517 // A physical read: replies are stripped and ordinary bytes
518 // reach the overlay or the session inside read.
519 if err := input.read(ctx, state, event.input.Data); err != nil {
520 // Part of the read may already be delivered or held, so
521 // it is committed rather than replayed.
522 fg.AckInput()
523 return w.settle(ctx, fg, stream, token, err)
524 }
525 fg.AckInput()
526 continue
527 }
528 if event.input != nil {
529 // A driver batch: bytes held for disambiguation go first.
530 if err := input.flushHeld(ctx, state); err != nil {
531 fg.PreserveInput(event.input.Data)
532 return w.settle(ctx, fg, stream, token, err)
533 }
534 consumed, supervised, err := picker.consumeInput(ctx, state, *event.input)
535 if err != nil {
536 fg.PreserveInput(event.input.Data)
537 return w.settle(ctx, fg, stream, token, err)
538 }
539 if consumed {
540 // The overlay owned these bytes: nothing reaches the session.
541 // A move-picker action still fences through the daemon so it
542 // completes after the overlay applied it; the supervisor
543 // settles a navigation-picker action itself.
544 fg.AckInput()
545 if event.input.actionID != 0 && !supervised {
546 if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil {
547 return w.settle(ctx, fg, stream, token, err)
548 }
549 }
550 continue
551 }
552 event.message = protocol.Input{InputSeq: input.nextSeq(), ActionID: event.input.actionID, Data: append([]byte(nil), event.input.Data...)}
553 }
554 if resize, ok := event.message.(protocol.Resize); ok {
555 if err := picker.resize(ctx, state, resize.Size); err != nil {
556 return w.settle(ctx, fg, stream, token, err)
557 }
558 }
559 if err := w.send(ctx, fg, stream, event.message); err != nil {
560 if event.input != nil {
561 fg.PreserveInput(event.input.Data)
562 }
563 return w.settle(ctx, fg, stream, token, err)
564 }
565 if event.input != nil {
566 fg.AckInput()
567 if event.input.actionID != 0 {
568 if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil {
569 return w.settle(ctx, fg, stream, token, err)
570 }
571 }
572 }
573 if event.action != 0 {
574 return attachmentLifecycleEnd(token, event.action)
575 }
576 case <-ctx.Done():
577 return w.settle(ctx, fg, stream, token, ctx.Err())
578 case <-fg.Done():
579 return w.settle(ctx, fg, stream, token, errAttachmentForegroundRevoked)
580 }
581 }
582}
583
584// attachmentLifecycleEnd gives local and daemon-triggered lifecycle actions
585// the same settlement path: drain/revoke foreground before publishing picker.
586func attachmentLifecycleEnd(token AttachmentToken, action AttachmentLifecycleActionKind) AttachmentEvent {
587 err := errDetachToPicker
588 if action == AttachmentDetachAndExit {
589 err = errDetachAndExit
590 }
591 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: err}
592}
593
594type attachmentUIForeground interface {
595 actionableGeneration() uint64
596 uiReceipt(protocol.UIReceipt)
597}
598
599func attachmentActionableGeneration(fg AttachmentForeground, token AttachmentToken) uint64 {
600 if ui, ok := fg.(attachmentUIForeground); ok {
601 return ui.actionableGeneration()
602 }
603 return token.Generation
604}
605
606func attachmentUIReceipt(fg AttachmentForeground, receipt protocol.UIReceipt) {
607 if ui, ok := fg.(attachmentUIForeground); ok {
608 ui.uiReceipt(receipt)
609 }
610}
611
612// attachmentSamePeerForeground is the optional UI seam of an in-place
613// same-peer switch: the attachment survives, so the driver action that caused
614// the offer settles on the destination's first committed output.
615type attachmentSamePeerForeground interface {
616 uiFollowSamePeer(actionID uint64)
617 uiSamePeerArrived()
618 uiSamePeerFailed()
619}
620
621func attachmentSamePeerUI(fg AttachmentForeground) attachmentSamePeerForeground {
622 ui, _ := fg.(attachmentSamePeerForeground)
623 return ui
624}
625
626// samePeerSwitchPending is the one confirmed same-peer switch awaiting its
627// destination output or the daemon's refusal.
628type samePeerSwitchPending struct {
629 requestID uint64
630 target protocol.ExactSessionTarget
631}
632
633// pumpAttachmentInput forwards each authorized delivery to the attached loop.
634// The loop decides whether an overlay owns it or it becomes session Input, and
635// sequences only the deliveries that actually reach the session.
636func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
637 for {
638 event, ok := fg.Input(ctx)
639 if !ok {
640 return
641 }
642 if event.Err != nil {
643 select {
644 case out <- attachmentClientEvent{input: &event}:
645 case <-ctx.Done():
646 fg.PreserveInput(event.Data)
647 }
648 return
649 }
650 copyEvent := event
651 select {
652 case out <- attachmentClientEvent{input: ©Event}:
653 case <-ctx.Done():
654 fg.PreserveInput(event.Data)
655 return
656 }
657 }
658}
659
660func pumpAttachmentGeometry(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
661 for {
662 geometry, ok := fg.Resize(ctx)
663 if !ok {
664 return
665 }
666 geometry = geometry.NormalizePixels()
667 select {
668 case out <- attachmentClientEvent{message: protocol.Resize{Size: geometry.Size, PixelWidth: geometry.PixelWidth, PixelHeight: geometry.PixelHeight}}:
669 case <-ctx.Done():
670 return
671 }
672 }
673}
674
675func pumpAttachmentLifecycle(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
676 action, ok := fg.Lifecycle(ctx)
677 if !ok {
678 return
679 }
680 select {
681 case out <- attachmentClientEvent{message: protocol.Detach{}, action: action}:
682 case <-ctx.Done():
683 return
684 }
685}
686
687func (w *sessionAttachmentWorker) validateWelcome(welcome protocol.Welcome) error {
688 if welcome.CommittedIdentity != nil {
689 return w.validateIdentity("welcome", welcome.CommittedIdentity.Target)
690 }
691 if (w.cfg.Request.Admission == ports.BrokerAdmissionCreateNamed || w.cfg.Request.Admission == ports.BrokerAdmissionAttachNamed) && welcome.SessionName != w.cfg.Request.Name {
692 return &AttachmentIdentityError{Stage: "welcome", Got: protocol.ExactSessionTarget{SessionName: welcome.SessionName}, Name: w.cfg.Request.Name}
693 }
694 return nil
695}
696
697func (w *sessionAttachmentWorker) validatePublication(output protocol.Output) error {
698 if output.Context == nil {
699 return errAttachmentPublication
700 }
701 return w.validateIdentity("initial publication", output.Context.Route.Target)
702}
703
704func (w *sessionAttachmentWorker) validateIdentity(stage string, got protocol.ExactSessionTarget) error {
705 request := w.cfg.Request
706 switch request.Admission {
707 case ports.BrokerAdmissionExact:
708 if got != request.Target {
709 return &AttachmentIdentityError{Stage: stage, Want: request.Target, Got: got}
710 }
711 case ports.BrokerAdmissionCreateNamed, ports.BrokerAdmissionAttachNamed:
712 if got.SessionName != request.Name {
713 return &AttachmentIdentityError{Stage: stage, Got: got, Name: request.Name}
714 }
715 }
716 return nil
717}
718
719// sendHello sends the typed Hello derived from the exact opened request.
720func (w *sessionAttachmentWorker) sendHello(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection) error {
721 hello := w.hello(stream)
722 hello.TerminalFocus = attachmentTerminalFocus(fg).Focus()
723 return w.send(ctx, fg, stream, hello)
724}
725
726// hello builds the Hello the daemon expects for exactly one opened request. No
727// endpoint, route, or resume token is invented: an exact request attaches, a
728// named request creates, and an ephemeral request is ephemeral.
729func (w *sessionAttachmentWorker) hello(stream ports.BrokerLogicalConnection) protocol.Hello {
730 request := w.cfg.Request
731 geometry := w.cfg.Geometry.NormalizePixels()
732 var environmentPolicy protocol.EnvironmentPolicy
733 var cwd string
734 var env []string
735 switch w.cfg.SessionEnvironment.Provenance {
736 case SessionEnvironmentLocalCLI, SessionEnvironmentLocalPicker:
737 environmentPolicy = protocol.EnvironmentPolicyClientOwned
738 cwd = w.cfg.SessionEnvironment.Cwd
739 env = append([]string(nil), request.Env...)
740 case SessionEnvironmentRemote:
741 environmentPolicy = protocol.EnvironmentPolicyDaemonOwned
742 default:
743 environmentPolicy = protocol.EnvironmentPolicyDaemonOwned
744 }
745 hello := protocol.Hello{
746 Version: protocol.Version,
747 Intent: protocol.IntentAttach,
748 ClientID: w.cfg.ClientID,
749 Size: geometry.Size,
750 PixelWidth: geometry.PixelWidth,
751 PixelHeight: geometry.PixelHeight,
752 TermEnv: w.cfg.TermEnv,
753 Cwd: cwd,
754 TrueColor: w.cfg.TrueColor,
755 // A remote attach reaches the terminal through this client, so both
756 // declarations describe this client's outer terminal.
757 KittyDirectGraphics: w.cfg.Capabilities.KittyGraphics,
758 KittyKeyboard: w.cfg.Capabilities.KittyKeyboard,
759 MaxOutputInFlight: requestedOutputWindow(stream),
760 Env: env,
761 EnvironmentPolicy: environmentPolicy,
762 Remote: !request.Local,
763 }
764 if request.Local {
765 hello.ClientPID = w.cfg.PID
766 }
767 switch request.Admission {
768 case ports.BrokerAdmissionExact:
769 target := request.Target
770 hello.Intent = protocol.IntentAttach
771 hello.ExactTarget = &target
772 // protocol.ValidateHello requires Name to equal the exact target's
773 // session name, and the daemon's exact-target routing compares it too, so
774 // the name is carried rather than left empty. It is copied from the
775 // request's own validated target, never invented.
776 hello.Name = target.SessionName
777 switch {
778 case w.cfg.Tab.stopped != nil && !request.Local:
779 // A stopped remote tab is restored through its exact selector.
780 selector := *w.cfg.Tab.stopped
781 hello.SessionTarget = &selector
782 default:
783 hello.PreferredTabID = w.cfg.Tabs.preferred(requestAuthority(request), target, w.cfg.Tab.preferred)
784 }
785 case ports.BrokerAdmissionCreateNamed:
786 hello.Intent = protocol.IntentNew
787 hello.Name = request.Name
788 case ports.BrokerAdmissionAttachNamed:
789 // The daemon resolves the name, restoring it if needed.
790 hello.Intent = protocol.IntentAttach
791 hello.Name = request.Name
792 case ports.BrokerAdmissionCreateEphemeral:
793 hello.Intent = protocol.IntentEphemeral
794 }
795 return hello
796}
797
798// send sends one client message while the run is still live. If cancellation
799// wins a blocked send, closing the stream interrupts the carriage and the
800// worker joins the send before returning, so no per-send goroutine is leaked.
801func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, message protocol.ClientMessage) error {
802 sent := make(chan error, 1)
803 go func() { sent <- stream.SendClient(message) }()
804 select {
805 case err := <-sent:
806 return err
807 case <-ctx.Done():
808 _ = stream.Close()
809 <-sent
810 return ctx.Err()
811 case <-fg.Done():
812 _ = stream.Close()
813 <-sent
814 return errAttachmentForegroundRevoked
815 }
816}
817
818// awaitServer waits for the next server message while the run is live.
819func (w *sessionAttachmentWorker) awaitServer(ctx context.Context, fg AttachmentForeground, incoming <-chan serverReceive) (protocol.ServerMessage, error) {
820 select {
821 case result := <-incoming:
822 if result.err != nil {
823 return nil, result.err
824 }
825 return result.message, nil
826 case <-ctx.Done():
827 return nil, ctx.Err()
828 case <-fg.Done():
829 return nil, errAttachmentForegroundRevoked
830 }
831}
832
833// settle classifies a stopping cause into the correct typed event. A cancelled
834// context is the supervisor's own deadline/teardown, a published stream cause
835// is a loss, a bare EOF is an orderly end, and anything else is a protocol or
836// application failure on the stream.
837func (w *sessionAttachmentWorker) settle(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, err error) AttachmentEvent {
838 if ctxErr := ctx.Err(); ctxErr != nil {
839 if errors.Is(context.Cause(ctx), ErrClientClosed) {
840 w.sendClosed(stream)
841 }
842 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errors.Join(ctxErr, err)}
843 }
844 if cause := streamErr(stream); cause != nil {
845 return AttachmentEvent{Token: token, Kind: AttachmentEventLost, Err: cause}
846 }
847 if errors.Is(err, io.EOF) {
848 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: err}
849 }
850 if err != nil {
851 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
852 }
853 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
854}
855
856// sendClosed makes a best-effort, bounded attempt to tell the daemon that this
857// client process is ending. It bypasses send's cancellation because the run
858// context is already done; the daemon parks the attachment if it never arrives.
859func (w *sessionAttachmentWorker) sendClosed(stream ports.BrokerLogicalConnection) {
860 if supervisorNil(stream) {
861 return
862 }
863 sent := make(chan error, 1)
864 go func() { sent <- stream.SendClient(protocol.Detach{Closed: true}) }()
865 timer := w.cfg.Clock.NewTimer(closedDetachTimeout)
866 defer stopSupervisorTimer(timer)
867 select {
868 case <-sent:
869 case <-timer.C():
870 }
871}
872
873// streamErr reads the logical stream's stable terminal cause without treating a
874// nil stream as a failure.
875func streamErr(stream ports.BrokerLogicalConnection) error {
876 if supervisorNil(stream) {
877 return nil
878 }
879 return stream.Err()
880}