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("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("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("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, raw ports.BrokerLogicalConnection, token AttachmentToken, state outputApplyState, incoming <-chan serverReceive) AttachmentEvent {
305 pumpCtx, cancel := context.WithCancel(ctx)
306 defer cancel()
307 // Every send while attached goes through the outbox so a stalled link
308 // never blocks key handling, overlays, or painting.
309 stream := newAttachmentOutbox(raw)
310 defer stream.stop()
311 outgoing := make(chan attachmentClientEvent)
312 go pumpAttachmentInput(pumpCtx, fg, outgoing)
313 go pumpAttachmentGeometry(pumpCtx, fg, outgoing)
314 go pumpAttachmentLifecycle(pumpCtx, fg, outgoing)
315 outputResetRequested := false
316 overlay := attachmentOverlay(fg)
317 var repaint <-chan struct{}
318 var tabSelections <-chan domain.TabStableID
319 var routeUpdates <-chan protocol.RecentRouteSnapshot
320 var navigationReplies <-chan protocol.ClientMessage
321 if overlay != nil {
322 repaint = overlay.overlayRepaint()
323 tabSelections = overlay.tabSelections()
324 routeUpdates = overlay.routeSnapshots()
325 navigationReplies = overlay.navigationReplies()
326 }
327 var samePeerRequests uint64
328 var samePeerSwitch *samePeerSwitchPending
329 samePeerUI := attachmentSamePeerUI(fg)
330 picker := &attachmentMovePicker{worker: w, fg: fg, overlay: overlay, stream: stream, size: w.cfg.Geometry.Size, move: newMovePickerOverlay(w.cfg.Color)}
331 defer picker.stopEscape()
332 input := newAttachmentInput(ctx, w, fg, stream, picker)
333 defer input.close()
334 if err := input.start(ctx); err != nil {
335 return w.settle(ctx, fg, stream, token, err)
336 }
337 // Hello declared the focus known at attach, for the first paint. The
338 // reporter re-sends the current focus, which the daemon ignores when
339 // unchanged, then every change. The pump records a focus report before it
340 // delivers the input read with it, so flushing focus first keeps the
341 // daemon's view in input order.
342 focus := &focusReporter{state: attachmentTerminalFocus(fg)}
343 sendFocus := func() error {
344 if message, ok := focus.next(); ok {
345 return w.send(ctx, fg, stream, message)
346 }
347 return nil
348 }
349 if err := sendFocus(); err != nil {
350 return w.settle(ctx, fg, stream, token, err)
351 }
352 // pending is the one undecided input delivery waiting on the outbox.
353 var pending *pendingInputAck
354 defer func() {
355 if pending == nil {
356 return
357 }
358 // Join the sender before deciding which input reached the carriage.
359 // A written automation action remains OutcomeUnknown if teardown
360 // prevents its UIFence receipt; acknowledging input is not proof
361 // that the daemon processed the action.
362 stream.join()
363 if unsent := stream.unsentInput(pending.from, pending.mark); len(unsent) != 0 {
364 fg.PreserveInput(unsent)
365 return
366 }
367 fg.AckInput()
368 }()
369 // ackWritten commits pending once the outbox wrote it, then fences its
370 // UI action after it.
371 ackWritten := func() error {
372 if pending == nil || stream.writtenCount() < pending.mark {
373 return nil
374 }
375 actionID := pending.actionID
376 pending = nil
377 fg.AckInput()
378 if actionID != 0 {
379 return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID})
380 }
381 return nil
382 }
383 // await makes the delivery just routed pending until the outbox wrote
384 // every message accepted since from.
385 await := func(from, actionID uint64) error {
386 mark := stream.accepted()
387 if mark == from {
388 // Nothing was queued (an overlay consumed the keys), so there is
389 // no write to wait for: an earlier stalled message must not hold
390 // the overlay's next key.
391 fg.AckInput()
392 if actionID != 0 {
393 return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID})
394 }
395 return nil
396 }
397 pending = &pendingInputAck{from: from, mark: mark, actionID: actionID}
398 return ackWritten()
399 }
400 for {
401 // Failure also closes progress. Settle it before rearming that
402 // channel, rather than spinning on an acknowledgement that cannot
403 // advance.
404 select {
405 case <-stream.Failed():
406 return w.settle(ctx, fg, stream, token, stream.failure())
407 default:
408 }
409 var progress <-chan struct{}
410 if pending != nil {
411 progress = stream.Progress()
412 if err := ackWritten(); err != nil {
413 return w.settle(ctx, fg, stream, token, err)
414 }
415 if pending == nil {
416 progress = nil
417 }
418 }
419 select {
420 case <-focus.changed:
421 if err := sendFocus(); err != nil {
422 return w.settle(ctx, fg, stream, token, err)
423 }
424 case <-input.wake:
425 if err := input.flush(ctx); err != nil {
426 return w.settle(ctx, fg, stream, token, err)
427 }
428 case <-input.markerC():
429 if err := input.markerExpired(ctx, state); err != nil {
430 return w.settle(ctx, fg, stream, token, err)
431 }
432 case <-input.drain.c():
433 if err := input.drainFired(ctx); err != nil {
434 return w.settle(ctx, fg, stream, token, err)
435 }
436 case <-input.completion.c():
437 if err := input.completionFired(ctx); err != nil {
438 return w.settle(ctx, fg, stream, token, err)
439 }
440 case <-repaint:
441 // An overlay released the terminal: the picker box is still on screen
442 // and suppressed frames were never written, so ask the daemon for an
443 // authoritative full repaint.
444 if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil {
445 return w.settle(ctx, fg, stream, token, err)
446 }
447 outputResetRequested = true
448 case tab := <-tabSelections:
449 // The picker overlay committed another tab of this session: switch
450 // the attachment's view in place instead of reconnecting.
451 if err := w.send(ctx, fg, stream, protocol.SelectTab{TabID: tab}); err != nil {
452 return w.settle(ctx, fg, stream, token, err)
453 }
454 case snapshot := <-routeUpdates:
455 // The supervisor's route ledger changed: the serving daemon renders
456 // the status-bar MRU, bells, and palette routes from it.
457 if err := w.send(ctx, fg, stream, snapshot); err != nil {
458 return w.settle(ctx, fg, stream, token, err)
459 }
460 case reply := <-navigationReplies:
461 if err := w.send(ctx, fg, stream, reply); err != nil {
462 return w.settle(ctx, fg, stream, token, err)
463 }
464 case <-picker.escape():
465 picker.escapeFired()
466 op, changed := picker.move.flush()
467 if err := picker.applyOp(ctx, state, op, changed); err != nil {
468 return w.settle(ctx, fg, stream, token, err)
469 }
470 case result := <-incoming:
471 if result.err != nil {
472 return w.settle(ctx, fg, stream, token, result.err)
473 }
474 switch typed := result.message.(type) {
475 case protocol.Output:
476 next, ok := state.next(typed)
477 if !ok {
478 continue
479 }
480 state = next
481 outputResetRequested = false
482 if samePeerSwitch != nil && state.context.Route.Target == samePeerSwitch.target {
483 // The in-place destination committed: its publication
484 // settles the action that caused the switch.
485 samePeerSwitch = nil
486 if samePeerUI != nil {
487 samePeerUI.uiSamePeerArrived()
488 }
489 }
490 w.noteCommitted(fg, state.context)
491 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil {
492 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing output: %w", err)}
493 }
494 if err := w.send(ctx, fg, stream, protocol.Ack{Epoch: state.epoch, State: state.state}); err != nil {
495 return w.settle(ctx, fg, stream, token, err)
496 }
497 case protocol.UIViewUpdate:
498 next, accepted, needsReset := state.nextView(typed)
499 if needsReset && !outputResetRequested {
500 if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil {
501 return w.settle(ctx, fg, stream, token, err)
502 }
503 outputResetRequested = true
504 }
505 if !accepted {
506 continue
507 }
508 state = next
509 w.noteCommitted(fg, state.context)
510 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), nil); err != nil {
511 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing view update: %w", err)}
512 }
513 case protocol.UIReceipt:
514 attachmentUIReceipt(fg, typed)
515 case protocol.PickerOffer:
516 if err := picker.offer(ctx, typed); err != nil {
517 return w.settle(ctx, fg, stream, token, err)
518 }
519 case protocol.PickerSnapshot:
520 if err := picker.snapshot(ctx, state, typed); err != nil {
521 return w.settle(ctx, fg, stream, token, err)
522 }
523 case protocol.PickerClosed:
524 picker.closed(typed.InteractionID)
525 case protocol.AttachTarget:
526 if typed.SamePeer && typed.ExactTarget != nil {
527 // An endpoint-empty offer on this very connection: confirm
528 // it and the daemon moves this attachment in place.
529 samePeerRequests++
530 request := protocol.SamePeerSwitchRequest{
531 RequestID: samePeerRequests, Target: *typed.ExactTarget,
532 PreferredTabID: w.cfg.Tabs.preferred(requestAuthority(w.cfg.Request), *typed.ExactTarget, typed.PreferredTabID),
533 }
534 if request.Validate() != nil {
535 continue
536 }
537 if samePeerUI != nil {
538 samePeerUI.uiFollowSamePeer(typed.CauseActionID)
539 }
540 samePeerSwitch = &samePeerSwitchPending{requestID: request.RequestID, target: request.Target}
541 if err := w.send(ctx, fg, stream, request); err != nil {
542 return w.settle(ctx, fg, stream, token, err)
543 }
544 continue
545 }
546 if overlay != nil {
547 overlay.requestNavigation(typed)
548 }
549 case protocol.SamePeerSwitchFailure:
550 // The daemon refused the confirmed switch and kept the source.
551 if samePeerSwitch != nil && typed.RequestID == samePeerSwitch.requestID {
552 samePeerSwitch = nil
553 if samePeerUI != nil {
554 samePeerUI.uiSamePeerFailed()
555 }
556 }
557 case protocol.RouteNavigationAction, protocol.RouteCreateSessionAction:
558 // The daemon asks the client to navigate: the supervisor
559 // resolves it through the broker catalogue.
560 if overlay != nil {
561 overlay.requestNavigation(typed)
562 }
563 case protocol.ErrorMsg:
564 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}
565 case protocol.Detached:
566 switch typed.Reason {
567 case protocol.ReasonDetachToPicker:
568 return attachmentLifecycleEnd(token, AttachmentDetachToPicker)
569 case protocol.ReasonDetach:
570 // The daemon confirmed the user's explicit detach request: the
571 // client exits cleanly instead of falling back to the picker.
572 return attachmentLifecycleEnd(token, AttachmentDetachAndExit)
573 }
574 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
575 }
576 case event := <-outgoing:
577 if event.input != nil && event.input.Err != nil {
578 fg.PreserveInput(event.input.Data)
579 return w.settle(ctx, fg, stream, token, event.input.Err)
580 }
581 if event.input != nil {
582 if err := sendFocus(); err != nil {
583 fg.PreserveInput(event.input.Data)
584 return w.settle(ctx, fg, stream, token, err)
585 }
586 }
587 if event.input != nil && event.input.actionID == 0 {
588 // A physical read: replies are stripped and ordinary bytes
589 // reach the overlay or the session inside read.
590 from := stream.accepted()
591 if err := input.read(ctx, state, event.input.Data); err != nil {
592 // Part of the read may already be delivered or held, so
593 // it is committed rather than replayed.
594 fg.AckInput()
595 return w.settle(ctx, fg, stream, token, err)
596 }
597 if err := await(from, 0); err != nil {
598 return w.settle(ctx, fg, stream, token, err)
599 }
600 continue
601 }
602 if event.input != nil {
603 // A driver batch: bytes held for disambiguation go first.
604 if err := input.flushHeld(ctx, state); err != nil {
605 fg.PreserveInput(event.input.Data)
606 return w.settle(ctx, fg, stream, token, err)
607 }
608 consumed, supervised, err := picker.consumeInput(ctx, state, *event.input)
609 if err != nil {
610 fg.PreserveInput(event.input.Data)
611 return w.settle(ctx, fg, stream, token, err)
612 }
613 if consumed {
614 // The overlay owned these bytes: nothing reaches the session.
615 // A move-picker action still fences through the daemon so it
616 // completes after the overlay applied it; the supervisor
617 // settles a navigation-picker action itself.
618 fg.AckInput()
619 if event.input.actionID != 0 && !supervised {
620 if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil {
621 return w.settle(ctx, fg, stream, token, err)
622 }
623 }
624 continue
625 }
626 event.message = protocol.Input{InputSeq: input.nextSeq(), ActionID: event.input.actionID, Data: append([]byte(nil), event.input.Data...)}
627 }
628 if resize, ok := event.message.(protocol.Resize); ok {
629 if err := picker.resize(ctx, state, resize.Size); err != nil {
630 return w.settle(ctx, fg, stream, token, err)
631 }
632 }
633 from := stream.accepted()
634 if err := w.send(ctx, fg, stream, event.message); err != nil {
635 if event.input != nil {
636 fg.PreserveInput(event.input.Data)
637 }
638 return w.settle(ctx, fg, stream, token, err)
639 }
640 if event.input != nil {
641 // The delivery stays undecided until the outbox wrote it, so
642 // a link failure preserves it for the next attempt exactly as
643 // a failed synchronous send did. The pump hands out no newer
644 // delivery meanwhile, which bounds queued input.
645 if err := await(from, event.input.actionID); err != nil {
646 return w.settle(ctx, fg, stream, token, err)
647 }
648 }
649 if event.action != 0 {
650 // The lifecycle message must reach the daemon before the
651 // supervisor closes the stream.
652 if err := w.drainOutbox(ctx, fg, stream); err != nil {
653 return w.settle(ctx, fg, stream, token, err)
654 }
655 return attachmentLifecycleEnd(token, event.action)
656 }
657 case <-progress:
658 if err := ackWritten(); err != nil {
659 return w.settle(ctx, fg, stream, token, err)
660 }
661 case <-stream.Failed():
662 return w.settle(ctx, fg, stream, token, stream.failure())
663 case <-ctx.Done():
664 return w.settle(ctx, fg, stream, token, ctx.Err())
665 case <-fg.Done():
666 return w.settle(ctx, fg, stream, token, errAttachmentForegroundRevoked)
667 }
668 }
669}
670
671// pendingInputAck is an input delivery whose messages the outbox accepted but
672// has not written yet: accepted messages numbered (from, mark].
673type pendingInputAck struct {
674 from uint64
675 mark uint64
676 actionID uint64
677}
678
679// attachmentLifecycleEnd gives local and daemon-triggered lifecycle actions
680// the same settlement path: drain/revoke foreground before publishing picker.
681func attachmentLifecycleEnd(token AttachmentToken, action AttachmentLifecycleActionKind) AttachmentEvent {
682 err := errDetachToPicker
683 if action == AttachmentDetachAndExit {
684 err = errDetachAndExit
685 }
686 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: err}
687}
688
689type attachmentUIForeground interface {
690 actionableGeneration() uint64
691 uiReceipt(protocol.UIReceipt)
692}
693
694func attachmentActionableGeneration(fg AttachmentForeground, token AttachmentToken) uint64 {
695 if ui, ok := fg.(attachmentUIForeground); ok {
696 return ui.actionableGeneration()
697 }
698 return token.Generation
699}
700
701func attachmentUIReceipt(fg AttachmentForeground, receipt protocol.UIReceipt) {
702 if ui, ok := fg.(attachmentUIForeground); ok {
703 ui.uiReceipt(receipt)
704 }
705}
706
707// attachmentSamePeerForeground is the optional UI seam of an in-place
708// same-peer switch: the attachment survives, so the driver action that caused
709// the offer settles on the destination's first committed output.
710type attachmentSamePeerForeground interface {
711 uiFollowSamePeer(actionID uint64)
712 uiSamePeerArrived()
713 uiSamePeerFailed()
714}
715
716func attachmentSamePeerUI(fg AttachmentForeground) attachmentSamePeerForeground {
717 ui, _ := fg.(attachmentSamePeerForeground)
718 return ui
719}
720
721// samePeerSwitchPending is the one confirmed same-peer switch awaiting its
722// destination output or the daemon's refusal.
723type samePeerSwitchPending struct {
724 requestID uint64
725 target protocol.ExactSessionTarget
726}
727
728// pumpAttachmentInput forwards each authorized delivery to the attached loop.
729// The loop decides whether an overlay owns it or it becomes session Input, and
730// sequences only the deliveries that actually reach the session.
731func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
732 for {
733 event, ok := fg.Input(ctx)
734 if !ok {
735 return
736 }
737 if event.Err != nil {
738 select {
739 case out <- attachmentClientEvent{input: &event}:
740 case <-ctx.Done():
741 fg.PreserveInput(event.Data)
742 }
743 return
744 }
745 copyEvent := event
746 select {
747 case out <- attachmentClientEvent{input: ©Event}:
748 case <-ctx.Done():
749 fg.PreserveInput(event.Data)
750 return
751 }
752 }
753}
754
755func pumpAttachmentGeometry(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
756 for {
757 geometry, ok := fg.Resize(ctx)
758 if !ok {
759 return
760 }
761 geometry = geometry.NormalizePixels()
762 select {
763 case out <- attachmentClientEvent{message: protocol.Resize{Size: geometry.Size, PixelWidth: geometry.PixelWidth, PixelHeight: geometry.PixelHeight}}:
764 case <-ctx.Done():
765 return
766 }
767 }
768}
769
770func pumpAttachmentLifecycle(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
771 action, ok := fg.Lifecycle(ctx)
772 if !ok {
773 return
774 }
775 select {
776 case out <- attachmentClientEvent{message: protocol.Detach{}, action: action}:
777 case <-ctx.Done():
778 return
779 }
780}
781
782func (w *sessionAttachmentWorker) validateWelcome(welcome protocol.Welcome) error {
783 if welcome.CommittedIdentity != nil {
784 return w.validateIdentity("welcome", welcome.CommittedIdentity.Target)
785 }
786 if (w.cfg.Request.Admission == ports.BrokerAdmissionCreateNamed || w.cfg.Request.Admission == ports.BrokerAdmissionAttachNamed) && welcome.SessionName != w.cfg.Request.Name {
787 return &AttachmentIdentityError{Stage: "welcome", Got: protocol.ExactSessionTarget{SessionName: welcome.SessionName}, Name: w.cfg.Request.Name}
788 }
789 return nil
790}
791
792func (w *sessionAttachmentWorker) validatePublication(output protocol.Output) error {
793 if output.Context == nil {
794 return errAttachmentPublication
795 }
796 return w.validateIdentity("initial publication", output.Context.Route.Target)
797}
798
799func (w *sessionAttachmentWorker) validateIdentity(stage string, got protocol.ExactSessionTarget) error {
800 request := w.cfg.Request
801 switch request.Admission {
802 case ports.BrokerAdmissionExact:
803 if got != request.Target {
804 return &AttachmentIdentityError{Stage: stage, Want: request.Target, Got: got}
805 }
806 case ports.BrokerAdmissionCreateNamed, ports.BrokerAdmissionAttachNamed:
807 if got.SessionName != request.Name {
808 return &AttachmentIdentityError{Stage: stage, Got: got, Name: request.Name}
809 }
810 }
811 return nil
812}
813
814// sendHello sends the typed Hello derived from the exact opened request.
815func (w *sessionAttachmentWorker) sendHello(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection) error {
816 hello := w.hello(stream)
817 hello.TerminalFocus = attachmentTerminalFocus(fg).Focus()
818 return w.send(ctx, fg, stream, hello)
819}
820
821// hello builds the Hello the daemon expects for exactly one opened request. No
822// endpoint, route, or resume token is invented: an exact request attaches, a
823// named request creates, and an ephemeral request is ephemeral.
824func (w *sessionAttachmentWorker) hello(stream ports.BrokerLogicalConnection) protocol.Hello {
825 request := w.cfg.Request
826 geometry := w.cfg.Geometry.NormalizePixels()
827 var environmentPolicy protocol.EnvironmentPolicy
828 var cwd string
829 var env []string
830 switch w.cfg.SessionEnvironment.Provenance {
831 case SessionEnvironmentLocalCLI, SessionEnvironmentLocalPicker:
832 environmentPolicy = protocol.EnvironmentPolicyClientOwned
833 cwd = w.cfg.SessionEnvironment.Cwd
834 env = append([]string(nil), request.Env...)
835 case SessionEnvironmentRemote:
836 environmentPolicy = protocol.EnvironmentPolicyDaemonOwned
837 default:
838 environmentPolicy = protocol.EnvironmentPolicyDaemonOwned
839 }
840 hello := protocol.Hello{
841 Version: protocol.Version,
842 Intent: protocol.IntentAttach,
843 ClientID: w.cfg.ClientID,
844 Size: geometry.Size,
845 PixelWidth: geometry.PixelWidth,
846 PixelHeight: geometry.PixelHeight,
847 TermEnv: w.cfg.TermEnv,
848 Cwd: cwd,
849 Color: w.cfg.Color,
850 // A remote attach reaches the terminal through this client, so both
851 // declarations describe this client's outer terminal.
852 KittyDirectGraphics: w.cfg.Capabilities.KittyGraphics,
853 KittyKeyboard: w.cfg.Capabilities.KittyKeyboard,
854 MaxOutputInFlight: requestedOutputWindow(stream),
855 Env: env,
856 EnvironmentPolicy: environmentPolicy,
857 Remote: !request.Local,
858 }
859 if request.Local {
860 hello.ClientPID = w.cfg.PID
861 }
862 switch request.Admission {
863 case ports.BrokerAdmissionExact:
864 target := request.Target
865 hello.Intent = protocol.IntentAttach
866 hello.ExactTarget = &target
867 // protocol.ValidateHello requires Name to equal the exact target's
868 // session name, and the daemon's exact-target routing compares it too, so
869 // the name is carried rather than left empty. It is copied from the
870 // request's own validated target, never invented.
871 hello.Name = target.SessionName
872 switch {
873 case w.cfg.Tab.stopped != nil && !request.Local:
874 // A stopped remote tab is restored through its exact selector.
875 selector := *w.cfg.Tab.stopped
876 hello.SessionTarget = &selector
877 default:
878 hello.PreferredTabID = w.cfg.Tabs.preferred(requestAuthority(request), target, w.cfg.Tab.preferred)
879 }
880 case ports.BrokerAdmissionCreateNamed:
881 hello.Intent = protocol.IntentNew
882 hello.Name = request.Name
883 case ports.BrokerAdmissionAttachNamed:
884 // The daemon resolves the name, restoring it if needed.
885 hello.Intent = protocol.IntentAttach
886 hello.Name = request.Name
887 case ports.BrokerAdmissionCreateEphemeral:
888 hello.Intent = protocol.IntentEphemeral
889 }
890 return hello
891}
892
893// send sends one client message while the run is still live. If cancellation
894// wins a blocked send, closing the stream interrupts the carriage and the
895// worker joins the send before returning, so no per-send goroutine is leaked.
896func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, message protocol.ClientMessage) error {
897 if outbox, ok := stream.(*attachmentOutbox); ok {
898 // Enqueue returns at once unless the outbox is full, and then still
899 // honours cancellation, so no helper goroutine is needed.
900 err := outbox.enqueue(ctx, fg.Done(), message)
901 if errors.Is(err, errAttachmentForegroundRevoked) || (ctx.Err() != nil && errors.Is(err, ctx.Err())) {
902 _ = outbox.Close()
903 }
904 return err
905 }
906 sent := make(chan error, 1)
907 go func() { sent <- stream.SendClient(message) }()
908 select {
909 case err := <-sent:
910 return err
911 case <-ctx.Done():
912 _ = stream.Close()
913 <-sent
914 return ctx.Err()
915 case <-fg.Done():
916 _ = stream.Close()
917 <-sent
918 return errAttachmentForegroundRevoked
919 }
920}
921
922// drainOutboxTimeout bounds how long a lifecycle exit waits for queued
923// messages on a stalled link. Past it the client leaves anyway; the daemon
924// parks the attachment when the detach never arrives.
925const drainOutboxTimeout = 2 * time.Second
926
927// drainOutbox waits until every queued message was written, the bound
928// elapses, or the run ends.
929func (w *sessionAttachmentWorker) drainOutbox(ctx context.Context, fg AttachmentForeground, outbox *attachmentOutbox) error {
930 timer := w.cfg.Clock.NewTimer(drainOutboxTimeout)
931 defer stopSupervisorTimer(timer)
932 select {
933 case <-timer.C():
934 return nil
935 case <-outbox.Idle():
936 return outbox.failure()
937 case <-outbox.Failed():
938 return outbox.failure()
939 case <-ctx.Done():
940 _ = outbox.Close()
941 return ctx.Err()
942 case <-fg.Done():
943 _ = outbox.Close()
944 return errAttachmentForegroundRevoked
945 }
946}
947
948// awaitServer waits for the next server message while the run is live.
949func (w *sessionAttachmentWorker) awaitServer(ctx context.Context, fg AttachmentForeground, incoming <-chan serverReceive) (protocol.ServerMessage, error) {
950 select {
951 case result := <-incoming:
952 if result.err != nil {
953 return nil, result.err
954 }
955 return result.message, nil
956 case <-ctx.Done():
957 return nil, ctx.Err()
958 case <-fg.Done():
959 return nil, errAttachmentForegroundRevoked
960 }
961}
962
963// settle classifies a stopping cause into the correct typed event. A cancelled
964// context is the supervisor's own deadline/teardown, a published stream cause
965// is a loss, a bare EOF is an orderly end, and anything else is a protocol or
966// application failure on the stream.
967func (w *sessionAttachmentWorker) settle(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, err error) AttachmentEvent {
968 if ctxErr := ctx.Err(); ctxErr != nil {
969 if errors.Is(context.Cause(ctx), ErrClientClosed) {
970 w.sendClosed(stream)
971 }
972 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errors.Join(ctxErr, err)}
973 }
974 if cause := streamErr(stream); cause != nil {
975 return AttachmentEvent{Token: token, Kind: AttachmentEventLost, Err: cause}
976 }
977 if errors.Is(err, io.EOF) {
978 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: err}
979 }
980 if err != nil {
981 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
982 }
983 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
984}
985
986// sendClosed makes a best-effort, bounded attempt to tell the daemon that this
987// client process is ending. It bypasses send's cancellation because the run
988// context is already done; the daemon parks the attachment if it never arrives.
989func (w *sessionAttachmentWorker) sendClosed(stream ports.BrokerLogicalConnection) {
990 if supervisorNil(stream) {
991 return
992 }
993 if outbox, ok := stream.(*attachmentOutbox); ok {
994 // Bypass the queue: the run is ending and its sender stops with it.
995 outbox.stop()
996 stream = outbox.BrokerLogicalConnection
997 }
998 sent := make(chan error, 1)
999 go func() { sent <- stream.SendClient(protocol.Detach{Closed: true}) }()
1000 timer := w.cfg.Clock.NewTimer(closedDetachTimeout)
1001 defer stopSupervisorTimer(timer)
1002 select {
1003 case <-sent:
1004 case <-timer.C():
1005 }
1006}
1007
1008// streamErr reads the logical stream's stable terminal cause without treating a
1009// nil stream as a failure.
1010func streamErr(stream ports.BrokerLogicalConnection) error {
1011 if supervisorNil(stream) {
1012 return nil
1013 }
1014 return stream.Err()
1015}