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