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 // EchoPredict selects predictive local echo on remote attachments.
103 EchoPredict domain.EchoPredictMode
104}
105
106// sessionAttachmentWorker implements AttachmentWorker for one broker logical
107// stream. It holds no stream or terminal of its own: everything arrives through
108// Run's foreground for the duration of one run.
109type sessionAttachmentWorker struct {
110 cfg sessionAttachmentConfig
111}
112
113// newSessionAttachmentWorker builds the real typed-session attachment worker.
114func newSessionAttachmentWorker(cfg sessionAttachmentConfig) (*sessionAttachmentWorker, error) {
115 if err := cfg.SessionEnvironment.Validate(); err != nil {
116 return nil, fmt.Errorf("client: invalid session environment: %w", err)
117 }
118 if !slices.Equal(cfg.Request.Env, cfg.SessionEnvironment.Env) {
119 return nil, errors.New("client: session environment does not match request environment")
120 }
121 cfg.Request.Env = append([]string(nil), cfg.Request.Env...)
122 cfg.SessionEnvironment = cfg.SessionEnvironment.Clone()
123 if supervisorNil(cfg.Clock) {
124 cfg.Clock = systemClock{}
125 }
126 return &sessionAttachmentWorker{cfg: cfg}, nil
127}
128
129// serverReceive is one server message or the terminal receive error.
130type serverReceive struct {
131 message protocol.ServerMessage
132 err error
133}
134
135// startServerReceive owns the single reader of the logical stream. It is
136// bounded: it buffers one message and blocks on the next read, and the stream's
137// Close contract unblocks the read. The returned stop channel releases the
138// reader when the run ends, so a message the worker no longer reads can never
139// pin it.
140func startServerReceive(stream ports.BrokerLogicalConnection) (<-chan serverReceive, chan struct{}) {
141 out := make(chan serverReceive, 1)
142 stop := make(chan struct{})
143 go func() {
144 for {
145 message, err := stream.ReceiveServer()
146 select {
147 case out <- serverReceive{message: message, err: err}:
148 case <-stop:
149 return
150 }
151 if err != nil {
152 return
153 }
154 }
155 }()
156 return out, stop
157}
158
159// Run drives Hello -> Welcome -> committed initial publication -> attached
160// output until the stream ends or the context/deadline, the supervisor finalize
161// signal, or the stream itself stops it. Every return path is a typed event;
162// the supervisor stamps the granted token onto it.
163func (w *sessionAttachmentWorker) Run(ctx context.Context, fg AttachmentForeground) AttachmentEvent {
164 token := fg.Token()
165 if ctx == nil {
166 ctx = context.Background()
167 }
168 stream := fg.Stream()
169 if supervisorNil(stream) {
170 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errAttachmentNoStream}
171 }
172 incoming, stopReceive := startServerReceive(stream)
173 defer close(stopReceive)
174
175 if err := w.sendHello(ctx, fg, stream); err != nil {
176 return w.settle(ctx, fg, stream, token, err)
177 }
178
179 welcome, err := w.awaitServer(ctx, fg, incoming)
180 if err != nil {
181 return w.settle(ctx, fg, stream, token, err)
182 }
183 switch message := welcome.(type) {
184 case protocol.Welcome:
185 // Welcome is a promise, not attachment: the daemon still owes the
186 // committed initial publication.
187 if err := w.validateWelcome(message); err != nil {
188 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
189 }
190 case protocol.ErrorMsg:
191 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: message.Code, Text: message.Text}}
192 default:
193 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("unexpected %T before welcome", welcome)}
194 }
195
196 state, initial, event := w.awaitInitialPublication(ctx, fg, stream, token, incoming)
197 if event != nil {
198 return *event
199 }
200
201 return w.pumpAttached(ctx, fg, stream, token, state, initial, incoming)
202}
203
204// awaitInitialPublication consumes server messages until the first valid full
205// output frame is committed. It returns the output state to continue with and
206// the committed frame, or a typed terminal event.
207func (w *sessionAttachmentWorker) awaitInitialPublication(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, incoming <-chan serverReceive) (outputApplyState, protocol.Output, *AttachmentEvent) {
208 state := outputApplyState{}
209 for {
210 message, err := w.awaitServer(ctx, fg, incoming)
211 if err != nil {
212 event := w.settle(ctx, fg, stream, token, err)
213 return state, protocol.Output{}, &event
214 }
215 switch typed := message.(type) {
216 case protocol.Output:
217 next, ok := state.next(typed)
218 if !ok {
219 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errAttachmentPublication}
220 return state, protocol.Output{}, &event
221 }
222 if err := w.validatePublication(typed); err != nil {
223 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
224 return state, protocol.Output{}, &event
225 }
226 state = next
227 w.noteCommitted(fg, state.context)
228 // The frame transaction is the pre-attach publication: it commits the first
229 // session bytes with the Connecting presentation and, through the foreground
230 // shape rule, a public generation of zero. Attached is published only after
231 // that write, flush, and commit succeeded.
232 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusConnecting), typed.Data); err != nil {
233 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing initial output: %w", err)}
234 return state, protocol.Output{}, &event
235 }
236 if err := w.send(ctx, fg, stream, protocol.Ack{Epoch: state.epoch, State: state.state}); err != nil {
237 event := w.settle(ctx, fg, stream, token, err)
238 return state, protocol.Output{}, &event
239 }
240 // Attachment is the committed publication. Complete the deadline
241 // boundary before presenting attached, so attached always implies the
242 // handshake budget can no longer cancel the run.
243 if w.cfg.BeforeAttached != nil {
244 if err := w.cfg.BeforeAttached(); err != nil {
245 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
246 return state, protocol.Output{}, &event
247 }
248 }
249 // A refused marker means the grant was revoked underneath us, so the
250 // run ends as a failure.
251 if !fg.MarkAttached() {
252 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errAttachmentForegroundRevoked}
253 return state, protocol.Output{}, &event
254 }
255 // The attached presentation is published last, once the committed frame and
256 // the attached marker are in place. It carries the real action generation
257 // with the same validated session identity and published boundary, and it
258 // writes no bytes: the screen it commits is exactly the frame that just
259 // drained.
260 if err := fg.PublishAttached(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached)); err != nil {
261 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing attached presentation: %w", err)}
262 return state, protocol.Output{}, &event
263 }
264 return state, typed, nil
265 case protocol.ErrorMsg:
266 event := AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}
267 return state, protocol.Output{}, &event
268 case protocol.Detached:
269 // The daemon ended the attempt before publishing anything: it
270 // never attached.
271 event := AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
272 return state, protocol.Output{}, &event
273 default:
274 // Everything else the daemon may publish before the initial frame
275 // stays outside this slice's scope and is ignored.
276 }
277 }
278}
279
280type attachmentClientEvent struct {
281 message protocol.ClientMessage
282 input *AttachmentInputEvent
283 action AttachmentLifecycleActionKind
284}
285
286// attachmentOverlay returns the optional overlay seam of fg, or nil.
287func attachmentOverlay(fg AttachmentForeground) attachmentOverlayForeground {
288 overlay, _ := fg.(attachmentOverlayForeground)
289 return overlay
290}
291
292func (w *sessionAttachmentWorker) noteCommitted(fg AttachmentForeground, view protocol.ViewContext) {
293 w.cfg.Tabs.remember(requestAuthority(w.cfg.Request), view.Route.Target, view.TabID)
294 if overlay := attachmentOverlay(fg); overlay != nil {
295 overlay.noteCommitted(view.Route.Target, view.TabID)
296 }
297}
298
299var (
300 errDetachToPicker = errors.New("client: detach to picker")
301 errDetachAndExit = errors.New("client: detach and exit")
302)
303
304// pumpAttached publishes output and concurrently forwards the foreground's
305// authorized input, latest geometry and explicit lifecycle decision.
306func (w *sessionAttachmentWorker) pumpAttached(ctx context.Context, fg AttachmentForeground, raw ports.BrokerLogicalConnection, token AttachmentToken, state outputApplyState, initial protocol.Output, incoming <-chan serverReceive) AttachmentEvent {
307 pumpCtx, cancel := context.WithCancel(ctx)
308 defer cancel()
309 // Every send while attached goes through the outbox so a stalled link
310 // never blocks key handling, overlays, or painting.
311 stream := newAttachmentOutbox(raw)
312 defer stream.stop()
313 outgoing := make(chan attachmentClientEvent)
314 go pumpAttachmentInput(pumpCtx, fg, outgoing)
315 go pumpAttachmentGeometry(pumpCtx, fg, outgoing)
316 go pumpAttachmentLifecycle(pumpCtx, fg, outgoing)
317 outputResetRequested := false
318 overlay := attachmentOverlay(fg)
319 var repaint <-chan struct{}
320 var tabSelections <-chan domain.TabStableID
321 var routeUpdates <-chan protocol.RecentRouteSnapshot
322 var navigationReplies <-chan protocol.ClientMessage
323 var inPlaceSwitches <-chan inPlaceSwitch
324 if overlay != nil {
325 repaint = overlay.overlayRepaint()
326 tabSelections = overlay.tabSelections()
327 routeUpdates = overlay.routeSnapshots()
328 navigationReplies = overlay.navigationReplies()
329 inPlaceSwitches = overlay.inPlaceSwitches()
330 }
331 var samePeerRequests uint64
332 var samePeerSwitch *samePeerSwitchPending
333 // repaintOwed is set while a picker in-place switch replaced the closing
334 // picker's repaint: if the source is kept after all, the worker still owes
335 // the daemon that full repaint over the picker box.
336 repaintOwed := false
337 reportInPlace := func(seq uint64, outcome inPlaceOutcome) {
338 if overlay != nil {
339 overlay.reportInPlace(seq, outcome)
340 }
341 }
342 // startSamePeer asks the daemon to move this attachment to target. seq is
343 // the supervisor's choice, or zero for a daemon offer. A switch still in
344 // flight is superseded: the newest navigation wins.
345 newSamePeerRequest := func(target protocol.ExactSessionTarget, tab domain.TabStableID) (protocol.SamePeerSwitchRequest, bool) {
346 request := protocol.SamePeerSwitchRequest{
347 RequestID: samePeerRequests + 1, Target: target,
348 PreferredTabID: w.cfg.Tabs.preferred(requestAuthority(w.cfg.Request), target, tab),
349 }
350 return request, request.Validate() == nil
351 }
352 startSamePeer := func(request protocol.SamePeerSwitchRequest, seq uint64) error {
353 samePeerRequests = request.RequestID
354 if seq != 0 {
355 repaintOwed = true
356 }
357 if samePeerSwitch != nil {
358 reportInPlace(samePeerSwitch.seq, inPlaceSuperseded)
359 }
360 samePeerSwitch = &samePeerSwitchPending{requestID: request.RequestID, target: request.Target, seq: seq}
361 return w.send(ctx, fg, stream, request)
362 }
363 samePeerUI := attachmentSamePeerUI(fg)
364 picker := &attachmentMovePicker{worker: w, fg: fg, overlay: overlay, stream: stream, size: w.cfg.Geometry.Size, move: newMovePickerOverlay(w.cfg.Color)}
365 defer picker.stopEscape()
366 echo := newAttachmentEcho(w, fg)
367 defer echo.close()
368 echo.seed(initial)
369 input := newAttachmentInput(ctx, w, fg, stream, picker, echo)
370 defer input.close()
371 if err := input.start(ctx); err != nil {
372 return w.settle(ctx, fg, stream, token, err)
373 }
374 // Hello declared the focus known at attach, for the first paint. The
375 // reporter re-sends the current focus, which the daemon ignores when
376 // unchanged, then every change. The pump records a focus report before it
377 // delivers the input read with it, so flushing focus first keeps the
378 // daemon's view in input order.
379 focus := &focusReporter{state: attachmentTerminalFocus(fg)}
380 sendFocus := func() error {
381 if message, ok := focus.next(); ok {
382 return w.send(ctx, fg, stream, message)
383 }
384 return nil
385 }
386 if err := sendFocus(); err != nil {
387 return w.settle(ctx, fg, stream, token, err)
388 }
389 // pending is the one undecided input delivery waiting on the outbox.
390 var pending *pendingInputAck
391 defer func() {
392 if pending == nil {
393 return
394 }
395 // Join the sender before deciding which input reached the carriage.
396 // A written automation action remains OutcomeUnknown if teardown
397 // prevents its UIFence receipt; acknowledging input is not proof
398 // that the daemon processed the action.
399 stream.join()
400 if unsent := stream.unsentInput(pending.from, pending.mark); len(unsent) != 0 {
401 fg.PreserveInput(unsent)
402 return
403 }
404 fg.AckInput()
405 }()
406 // ackWritten commits pending once the outbox wrote it, then fences its
407 // UI action after it.
408 ackWritten := func() error {
409 if pending == nil || stream.writtenCount() < pending.mark {
410 return nil
411 }
412 actionID := pending.actionID
413 pending = nil
414 fg.AckInput()
415 if actionID != 0 {
416 return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID})
417 }
418 return nil
419 }
420 // await makes the delivery just routed pending until the outbox wrote
421 // every message accepted since from.
422 await := func(from, actionID uint64) error {
423 mark := stream.accepted()
424 if mark == from {
425 // Nothing was queued (an overlay consumed the keys), so there is
426 // no write to wait for: an earlier stalled message must not hold
427 // the overlay's next key.
428 fg.AckInput()
429 if actionID != 0 {
430 return w.send(ctx, fg, stream, protocol.UIFence{ActionID: actionID})
431 }
432 return nil
433 }
434 pending = &pendingInputAck{from: from, mark: mark, actionID: actionID}
435 return ackWritten()
436 }
437 for {
438 // Failure also closes progress. Settle it before rearming that
439 // channel, rather than spinning on an acknowledgement that cannot
440 // advance.
441 select {
442 case <-stream.Failed():
443 return w.settle(ctx, fg, stream, token, stream.failure())
444 default:
445 }
446 var progress <-chan struct{}
447 if pending != nil {
448 progress = stream.Progress()
449 if err := ackWritten(); err != nil {
450 return w.settle(ctx, fg, stream, token, err)
451 }
452 if pending == nil {
453 progress = nil
454 }
455 }
456 select {
457 case <-focus.changed:
458 if err := sendFocus(); err != nil {
459 return w.settle(ctx, fg, stream, token, err)
460 }
461 case <-input.wake:
462 if err := input.flush(ctx); err != nil {
463 return w.settle(ctx, fg, stream, token, err)
464 }
465 case <-input.markerC():
466 if err := input.markerExpired(ctx, state); err != nil {
467 return w.settle(ctx, fg, stream, token, err)
468 }
469 case <-input.drain.c():
470 if err := input.drainFired(ctx); err != nil {
471 return w.settle(ctx, fg, stream, token, err)
472 }
473 case <-input.completion.c():
474 if err := input.completionFired(ctx); err != nil {
475 return w.settle(ctx, fg, stream, token, err)
476 }
477 case <-echo.tickC():
478 if err := echo.ticked(); err != nil {
479 return w.settle(ctx, fg, stream, token, err)
480 }
481 case <-repaint:
482 // An overlay released the terminal: the picker box is still on screen
483 // and suppressed frames were never written, so ask the daemon for an
484 // authoritative full repaint.
485 if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil {
486 return w.settle(ctx, fg, stream, token, err)
487 }
488 outputResetRequested = true
489 case tab := <-tabSelections:
490 // The picker overlay committed another tab of this session: switch
491 // the attachment's view in place instead of reconnecting.
492 if err := w.send(ctx, fg, stream, protocol.SelectTab{TabID: tab}); err != nil {
493 return w.settle(ctx, fg, stream, token, err)
494 }
495 case snapshot := <-routeUpdates:
496 // The supervisor's route ledger changed: the serving daemon renders
497 // the status-bar MRU, bells, and palette routes from it.
498 if err := w.send(ctx, fg, stream, snapshot); err != nil {
499 return w.settle(ctx, fg, stream, token, err)
500 }
501 case reply := <-navigationReplies:
502 if err := w.send(ctx, fg, stream, reply); err != nil {
503 return w.settle(ctx, fg, stream, token, err)
504 }
505 case next := <-inPlaceSwitches:
506 // The supervisor committed another session on this daemon: ask it
507 // to move this attachment there instead of reconnecting.
508 request, ok := newSamePeerRequest(next.target, next.tab)
509 if !ok {
510 reportInPlace(next.seq, inPlaceRefused)
511 continue
512 }
513 if err := startSamePeer(request, next.seq); err != nil {
514 return w.settle(ctx, fg, stream, token, err)
515 }
516 case <-picker.escape():
517 picker.escapeFired()
518 op, changed := picker.move.flush()
519 if err := picker.applyOp(ctx, state, op, changed); err != nil {
520 return w.settle(ctx, fg, stream, token, err)
521 }
522 case result := <-incoming:
523 if result.err != nil {
524 return w.settle(ctx, fg, stream, token, result.err)
525 }
526 switch typed := result.message.(type) {
527 case protocol.Output:
528 next, ok := state.next(typed)
529 if !ok {
530 continue
531 }
532 state = next
533 outputResetRequested = false
534 if samePeerSwitch != nil && state.context.Route.Target == samePeerSwitch.target {
535 // The in-place destination committed: its publication
536 // settles the action that caused the switch.
537 reportInPlace(samePeerSwitch.seq, inPlaceArrived)
538 samePeerSwitch = nil
539 repaintOwed = false
540 if samePeerUI != nil {
541 samePeerUI.uiSamePeerArrived()
542 }
543 }
544 w.noteCommitted(fg, state.context)
545 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), typed.Data); err != nil {
546 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing output: %w", err)}
547 }
548 // The guesses go on top of the frame just written.
549 if err := echo.output(typed); err != nil {
550 return w.settle(ctx, fg, stream, token, err)
551 }
552 if err := w.send(ctx, fg, stream, protocol.Ack{Epoch: state.epoch, State: state.state}); err != nil {
553 return w.settle(ctx, fg, stream, token, err)
554 }
555 case protocol.UIViewUpdate:
556 next, accepted, needsReset := state.nextView(typed)
557 if needsReset && !outputResetRequested {
558 if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil {
559 return w.settle(ctx, fg, stream, token, err)
560 }
561 outputResetRequested = true
562 }
563 if !accepted {
564 continue
565 }
566 state = next
567 w.noteCommitted(fg, state.context)
568 if err := fg.Output(state.uiContext(ports.UIContext{Generation: attachmentActionableGeneration(fg, token)}, ports.UIStatusAttached), nil); err != nil {
569 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: fmt.Errorf("publishing view update: %w", err)}
570 }
571 case protocol.UIReceipt:
572 attachmentUIReceipt(fg, typed)
573 case protocol.PickerOffer:
574 if err := picker.offer(ctx, typed); err != nil {
575 return w.settle(ctx, fg, stream, token, err)
576 }
577 case protocol.PickerSnapshot:
578 if err := picker.snapshot(ctx, state, typed); err != nil {
579 return w.settle(ctx, fg, stream, token, err)
580 }
581 case protocol.PickerClosed:
582 picker.closed(typed.InteractionID)
583 case protocol.AttachTarget:
584 if typed.SamePeer && typed.ExactTarget != nil {
585 // An endpoint-empty offer on this very connection: confirm
586 // it and the daemon moves this attachment in place.
587 request, ok := newSamePeerRequest(*typed.ExactTarget, typed.PreferredTabID)
588 if !ok {
589 continue
590 }
591 if samePeerUI != nil {
592 samePeerUI.uiFollowSamePeer(typed.CauseActionID)
593 }
594 if err := startSamePeer(request, 0); err != nil {
595 return w.settle(ctx, fg, stream, token, err)
596 }
597 continue
598 }
599 if overlay != nil {
600 overlay.requestNavigation(typed)
601 }
602 case protocol.SamePeerSwitchFailure:
603 // The daemon refused the confirmed switch and kept the source.
604 if samePeerSwitch != nil && typed.RequestID == samePeerSwitch.requestID {
605 if samePeerSwitch.seq != 0 {
606 // The supervisor falls back to a fresh attachment; the
607 // action it follows settles there.
608 reportInPlace(samePeerSwitch.seq, inPlaceRefused)
609 } else {
610 if samePeerUI != nil {
611 samePeerUI.uiSamePeerFailed()
612 }
613 if repaintOwed {
614 // No swap follows: the source stays, under what is
615 // left of the picker box.
616 repaintOwed = false
617 if err := w.send(ctx, fg, stream, protocol.OutputResetRequest{}); err != nil {
618 return w.settle(ctx, fg, stream, token, err)
619 }
620 outputResetRequested = true
621 }
622 }
623 samePeerSwitch = nil
624 }
625 case protocol.RouteNavigationAction, protocol.RouteCreateSessionAction:
626 // The daemon asks the client to navigate: the supervisor
627 // resolves it through the broker catalogue.
628 if overlay != nil {
629 overlay.requestNavigation(typed)
630 }
631 case protocol.ErrorMsg:
632 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: &ProtocolError{Code: typed.Code, Text: typed.Text}}
633 case protocol.Detached:
634 switch typed.Reason {
635 case protocol.ReasonDetachToPicker:
636 return attachmentLifecycleEnd(token, AttachmentDetachToPicker)
637 case protocol.ReasonDetach:
638 // The daemon confirmed the user's explicit detach request: the
639 // client exits cleanly instead of falling back to the picker.
640 return attachmentLifecycleEnd(token, AttachmentDetachAndExit)
641 }
642 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
643 }
644 case event := <-outgoing:
645 if event.input != nil && event.input.Err != nil {
646 fg.PreserveInput(event.input.Data)
647 return w.settle(ctx, fg, stream, token, event.input.Err)
648 }
649 if event.input != nil {
650 if err := sendFocus(); err != nil {
651 fg.PreserveInput(event.input.Data)
652 return w.settle(ctx, fg, stream, token, err)
653 }
654 }
655 if event.input != nil && event.input.actionID == 0 {
656 // A physical read: replies are stripped and ordinary bytes
657 // reach the overlay or the session inside read.
658 from := stream.accepted()
659 if err := input.read(ctx, state, event.input.Data); err != nil {
660 // Part of the read may already be delivered or held, so
661 // it is committed rather than replayed.
662 fg.AckInput()
663 return w.settle(ctx, fg, stream, token, err)
664 }
665 if err := await(from, 0); err != nil {
666 return w.settle(ctx, fg, stream, token, err)
667 }
668 continue
669 }
670 if event.input != nil {
671 // A driver batch: bytes held for disambiguation go first.
672 if err := input.flushHeld(ctx, state); err != nil {
673 fg.PreserveInput(event.input.Data)
674 return w.settle(ctx, fg, stream, token, err)
675 }
676 consumed, supervised, err := picker.consumeInput(ctx, state, *event.input)
677 if err != nil {
678 fg.PreserveInput(event.input.Data)
679 return w.settle(ctx, fg, stream, token, err)
680 }
681 if consumed {
682 // The overlay owned these bytes: nothing reaches the session.
683 // A move-picker action still fences through the daemon so it
684 // completes after the overlay applied it; the supervisor
685 // settles a navigation-picker action itself.
686 fg.AckInput()
687 if event.input.actionID != 0 && !supervised {
688 if err := w.send(ctx, fg, stream, protocol.UIFence{ActionID: event.input.actionID}); err != nil {
689 return w.settle(ctx, fg, stream, token, err)
690 }
691 }
692 continue
693 }
694 event.message = protocol.Input{InputSeq: input.nextSeq(), ActionID: event.input.actionID, Data: append([]byte(nil), event.input.Data...)}
695 }
696 if resize, ok := event.message.(protocol.Resize); ok {
697 if err := echo.resized(resize.Size); err != nil {
698 return w.settle(ctx, fg, stream, token, err)
699 }
700 if err := picker.resize(ctx, state, resize.Size); err != nil {
701 return w.settle(ctx, fg, stream, token, err)
702 }
703 }
704 from := stream.accepted()
705 if err := w.send(ctx, fg, stream, event.message); err != nil {
706 if event.input != nil {
707 fg.PreserveInput(event.input.Data)
708 }
709 return w.settle(ctx, fg, stream, token, err)
710 }
711 if event.input != nil {
712 // The delivery stays undecided until the outbox wrote it, so
713 // a link failure preserves it for the next attempt exactly as
714 // a failed synchronous send did. The pump hands out no newer
715 // delivery meanwhile, which bounds queued input.
716 if err := await(from, event.input.actionID); err != nil {
717 return w.settle(ctx, fg, stream, token, err)
718 }
719 if message, ok := event.message.(protocol.Input); ok {
720 if err := echo.input(message.InputSeq, message.Data); err != nil {
721 return w.settle(ctx, fg, stream, token, err)
722 }
723 }
724 }
725 if event.action != 0 {
726 // The lifecycle message must reach the daemon before the
727 // supervisor closes the stream.
728 if err := w.drainOutbox(ctx, fg, stream); err != nil {
729 return w.settle(ctx, fg, stream, token, err)
730 }
731 return attachmentLifecycleEnd(token, event.action)
732 }
733 case <-progress:
734 if err := ackWritten(); err != nil {
735 return w.settle(ctx, fg, stream, token, err)
736 }
737 case <-stream.Failed():
738 return w.settle(ctx, fg, stream, token, stream.failure())
739 case <-ctx.Done():
740 return w.settle(ctx, fg, stream, token, ctx.Err())
741 case <-fg.Done():
742 return w.settle(ctx, fg, stream, token, errAttachmentForegroundRevoked)
743 }
744 }
745}
746
747// pendingInputAck is an input delivery whose messages the outbox accepted but
748// has not written yet: accepted messages numbered (from, mark].
749type pendingInputAck struct {
750 from uint64
751 mark uint64
752 actionID uint64
753}
754
755// attachmentLifecycleEnd gives local and daemon-triggered lifecycle actions
756// the same settlement path: drain/revoke foreground before publishing picker.
757func attachmentLifecycleEnd(token AttachmentToken, action AttachmentLifecycleActionKind) AttachmentEvent {
758 err := errDetachToPicker
759 if action == AttachmentDetachAndExit {
760 err = errDetachAndExit
761 }
762 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: err}
763}
764
765type attachmentUIForeground interface {
766 actionableGeneration() uint64
767 uiReceipt(protocol.UIReceipt)
768}
769
770func attachmentActionableGeneration(fg AttachmentForeground, token AttachmentToken) uint64 {
771 if ui, ok := fg.(attachmentUIForeground); ok {
772 return ui.actionableGeneration()
773 }
774 return token.Generation
775}
776
777func attachmentUIReceipt(fg AttachmentForeground, receipt protocol.UIReceipt) {
778 if ui, ok := fg.(attachmentUIForeground); ok {
779 ui.uiReceipt(receipt)
780 }
781}
782
783// attachmentSamePeerForeground is the optional UI seam of an in-place
784// same-peer switch: the attachment survives, so the driver action that caused
785// the offer settles on the destination's first committed output.
786type attachmentSamePeerForeground interface {
787 uiFollowSamePeer(actionID uint64)
788 uiSamePeerArrived()
789 uiSamePeerFailed()
790}
791
792func attachmentSamePeerUI(fg AttachmentForeground) attachmentSamePeerForeground {
793 ui, _ := fg.(attachmentSamePeerForeground)
794 return ui
795}
796
797// samePeerSwitchPending is the one confirmed same-peer switch awaiting its
798// destination output or the daemon's refusal.
799type samePeerSwitchPending struct {
800 requestID uint64
801 target protocol.ExactSessionTarget
802 // seq is the supervisor's in-place choice this switch carries out, zero
803 // for a daemon offer. A nonzero seq reports its outcome to the supervisor.
804 seq uint64
805}
806
807// pumpAttachmentInput forwards each authorized delivery to the attached loop.
808// The loop decides whether an overlay owns it or it becomes session Input, and
809// sequences only the deliveries that actually reach the session.
810func pumpAttachmentInput(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
811 for {
812 event, ok := fg.Input(ctx)
813 if !ok {
814 return
815 }
816 if event.Err != nil {
817 select {
818 case out <- attachmentClientEvent{input: &event}:
819 case <-ctx.Done():
820 fg.PreserveInput(event.Data)
821 }
822 return
823 }
824 copyEvent := event
825 select {
826 case out <- attachmentClientEvent{input: ©Event}:
827 case <-ctx.Done():
828 fg.PreserveInput(event.Data)
829 return
830 }
831 }
832}
833
834func pumpAttachmentGeometry(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
835 for {
836 geometry, ok := fg.Resize(ctx)
837 if !ok {
838 return
839 }
840 geometry = geometry.NormalizePixels()
841 select {
842 case out <- attachmentClientEvent{message: protocol.Resize{Size: geometry.Size, PixelWidth: geometry.PixelWidth, PixelHeight: geometry.PixelHeight}}:
843 case <-ctx.Done():
844 return
845 }
846 }
847}
848
849func pumpAttachmentLifecycle(ctx context.Context, fg AttachmentForeground, out chan<- attachmentClientEvent) {
850 action, ok := fg.Lifecycle(ctx)
851 if !ok {
852 return
853 }
854 select {
855 case out <- attachmentClientEvent{message: protocol.Detach{}, action: action}:
856 case <-ctx.Done():
857 return
858 }
859}
860
861func (w *sessionAttachmentWorker) validateWelcome(welcome protocol.Welcome) error {
862 if welcome.CommittedIdentity != nil {
863 return w.validateIdentity("welcome", welcome.CommittedIdentity.Target)
864 }
865 if (w.cfg.Request.Admission == ports.BrokerAdmissionCreateNamed || w.cfg.Request.Admission == ports.BrokerAdmissionAttachNamed) && welcome.SessionName != w.cfg.Request.Name {
866 return &AttachmentIdentityError{Stage: "welcome", Got: protocol.ExactSessionTarget{SessionName: welcome.SessionName}, Name: w.cfg.Request.Name}
867 }
868 return nil
869}
870
871func (w *sessionAttachmentWorker) validatePublication(output protocol.Output) error {
872 if output.Context == nil {
873 return errAttachmentPublication
874 }
875 return w.validateIdentity("initial publication", output.Context.Route.Target)
876}
877
878func (w *sessionAttachmentWorker) validateIdentity(stage string, got protocol.ExactSessionTarget) error {
879 request := w.cfg.Request
880 switch request.Admission {
881 case ports.BrokerAdmissionExact:
882 if got != request.Target {
883 return &AttachmentIdentityError{Stage: stage, Want: request.Target, Got: got}
884 }
885 case ports.BrokerAdmissionCreateNamed, ports.BrokerAdmissionAttachNamed:
886 if got.SessionName != request.Name {
887 return &AttachmentIdentityError{Stage: stage, Got: got, Name: request.Name}
888 }
889 }
890 return nil
891}
892
893// sendHello sends the typed Hello derived from the exact opened request.
894func (w *sessionAttachmentWorker) sendHello(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection) error {
895 hello := w.hello(stream)
896 hello.TerminalFocus = attachmentTerminalFocus(fg).Focus()
897 return w.send(ctx, fg, stream, hello)
898}
899
900// hello builds the Hello the daemon expects for exactly one opened request. No
901// endpoint, route, or resume token is invented: an exact request attaches, a
902// named request creates, and an ephemeral request is ephemeral.
903func (w *sessionAttachmentWorker) hello(stream ports.BrokerLogicalConnection) protocol.Hello {
904 request := w.cfg.Request
905 geometry := w.cfg.Geometry.NormalizePixels()
906 var environmentPolicy protocol.EnvironmentPolicy
907 var cwd string
908 var env []string
909 switch w.cfg.SessionEnvironment.Provenance {
910 case SessionEnvironmentLocalCLI, SessionEnvironmentLocalPicker:
911 environmentPolicy = protocol.EnvironmentPolicyClientOwned
912 cwd = w.cfg.SessionEnvironment.Cwd
913 env = append([]string(nil), request.Env...)
914 case SessionEnvironmentRemote:
915 environmentPolicy = protocol.EnvironmentPolicyDaemonOwned
916 default:
917 environmentPolicy = protocol.EnvironmentPolicyDaemonOwned
918 }
919 hello := protocol.Hello{
920 Version: protocol.Version,
921 Intent: protocol.IntentAttach,
922 ClientID: w.cfg.ClientID,
923 Size: geometry.Size,
924 PixelWidth: geometry.PixelWidth,
925 PixelHeight: geometry.PixelHeight,
926 TermEnv: w.cfg.TermEnv,
927 Cwd: cwd,
928 Color: w.cfg.Color,
929 // A remote attach reaches the terminal through this client, so both
930 // declarations describe this client's outer terminal.
931 KittyDirectGraphics: w.cfg.Capabilities.KittyGraphics,
932 KittyKeyboard: w.cfg.Capabilities.KittyKeyboard,
933 MaxOutputInFlight: requestedOutputWindow(stream),
934 Env: env,
935 EnvironmentPolicy: environmentPolicy,
936 Remote: !request.Local,
937 }
938 if request.Local {
939 hello.ClientPID = w.cfg.PID
940 }
941 switch request.Admission {
942 case ports.BrokerAdmissionExact:
943 target := request.Target
944 hello.Intent = protocol.IntentAttach
945 hello.ExactTarget = &target
946 // protocol.ValidateHello requires Name to equal the exact target's
947 // session name, and the daemon's exact-target routing compares it too, so
948 // the name is carried rather than left empty. It is copied from the
949 // request's own validated target, never invented.
950 hello.Name = target.SessionName
951 switch {
952 case w.cfg.Tab.stopped != nil && !request.Local:
953 // A stopped remote tab is restored through its exact selector.
954 selector := *w.cfg.Tab.stopped
955 hello.SessionTarget = &selector
956 default:
957 hello.PreferredTabID = w.cfg.Tabs.preferred(requestAuthority(request), target, w.cfg.Tab.preferred)
958 }
959 case ports.BrokerAdmissionCreateNamed:
960 hello.Intent = protocol.IntentNew
961 hello.Name = request.Name
962 case ports.BrokerAdmissionAttachNamed:
963 // The daemon resolves the name, restoring it if needed.
964 hello.Intent = protocol.IntentAttach
965 hello.Name = request.Name
966 case ports.BrokerAdmissionCreateEphemeral:
967 hello.Intent = protocol.IntentEphemeral
968 }
969 return hello
970}
971
972// send sends one client message while the run is still live. If cancellation
973// wins a blocked send, closing the stream interrupts the carriage and the
974// worker joins the send before returning, so no per-send goroutine is leaked.
975func (w *sessionAttachmentWorker) send(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, message protocol.ClientMessage) error {
976 if outbox, ok := stream.(*attachmentOutbox); ok {
977 // Enqueue returns at once unless the outbox is full, and then still
978 // honours cancellation, so no helper goroutine is needed.
979 err := outbox.enqueue(ctx, fg.Done(), message)
980 if errors.Is(err, errAttachmentForegroundRevoked) || (ctx.Err() != nil && errors.Is(err, ctx.Err())) {
981 _ = outbox.Close()
982 }
983 return err
984 }
985 sent := make(chan error, 1)
986 go func() { sent <- stream.SendClient(message) }()
987 select {
988 case err := <-sent:
989 return err
990 case <-ctx.Done():
991 _ = stream.Close()
992 <-sent
993 return ctx.Err()
994 case <-fg.Done():
995 _ = stream.Close()
996 <-sent
997 return errAttachmentForegroundRevoked
998 }
999}
1000
1001// drainOutboxTimeout bounds how long a lifecycle exit waits for queued
1002// messages on a stalled link. Past it the client leaves anyway; the daemon
1003// parks the attachment when the detach never arrives.
1004const drainOutboxTimeout = 2 * time.Second
1005
1006// drainOutbox waits until every queued message was written, the bound
1007// elapses, or the run ends.
1008func (w *sessionAttachmentWorker) drainOutbox(ctx context.Context, fg AttachmentForeground, outbox *attachmentOutbox) error {
1009 timer := w.cfg.Clock.NewTimer(drainOutboxTimeout)
1010 defer stopSupervisorTimer(timer)
1011 select {
1012 case <-timer.C():
1013 return nil
1014 case <-outbox.Idle():
1015 return outbox.failure()
1016 case <-outbox.Failed():
1017 return outbox.failure()
1018 case <-ctx.Done():
1019 _ = outbox.Close()
1020 return ctx.Err()
1021 case <-fg.Done():
1022 _ = outbox.Close()
1023 return errAttachmentForegroundRevoked
1024 }
1025}
1026
1027// awaitServer waits for the next server message while the run is live.
1028func (w *sessionAttachmentWorker) awaitServer(ctx context.Context, fg AttachmentForeground, incoming <-chan serverReceive) (protocol.ServerMessage, error) {
1029 select {
1030 case result := <-incoming:
1031 if result.err != nil {
1032 return nil, result.err
1033 }
1034 return result.message, nil
1035 case <-ctx.Done():
1036 return nil, ctx.Err()
1037 case <-fg.Done():
1038 return nil, errAttachmentForegroundRevoked
1039 }
1040}
1041
1042// settle classifies a stopping cause into the correct typed event. A cancelled
1043// context is the supervisor's own deadline/teardown, a published stream cause
1044// is a loss, a bare EOF is an orderly end, and anything else is a protocol or
1045// application failure on the stream.
1046func (w *sessionAttachmentWorker) settle(ctx context.Context, fg AttachmentForeground, stream ports.BrokerLogicalConnection, token AttachmentToken, err error) AttachmentEvent {
1047 if ctxErr := ctx.Err(); ctxErr != nil {
1048 if errors.Is(context.Cause(ctx), ErrClientClosed) {
1049 w.sendClosed(stream)
1050 }
1051 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: errors.Join(ctxErr, err)}
1052 }
1053 if cause := streamErr(stream); cause != nil {
1054 return AttachmentEvent{Token: token, Kind: AttachmentEventLost, Err: cause}
1055 }
1056 if errors.Is(err, io.EOF) {
1057 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded, Err: err}
1058 }
1059 if err != nil {
1060 return AttachmentEvent{Token: token, Kind: AttachmentEventFailed, Err: err}
1061 }
1062 return AttachmentEvent{Token: token, Kind: AttachmentEventEnded}
1063}
1064
1065// sendClosed makes a best-effort, bounded attempt to tell the daemon that this
1066// client process is ending. It bypasses send's cancellation because the run
1067// context is already done; the daemon parks the attachment if it never arrives.
1068func (w *sessionAttachmentWorker) sendClosed(stream ports.BrokerLogicalConnection) {
1069 if supervisorNil(stream) {
1070 return
1071 }
1072 if outbox, ok := stream.(*attachmentOutbox); ok {
1073 // Bypass the queue: the run is ending and its sender stops with it.
1074 outbox.stop()
1075 stream = outbox.BrokerLogicalConnection
1076 }
1077 sent := make(chan error, 1)
1078 go func() { sent <- stream.SendClient(protocol.Detach{Closed: true}) }()
1079 timer := w.cfg.Clock.NewTimer(closedDetachTimeout)
1080 defer stopSupervisorTimer(timer)
1081 select {
1082 case <-sent:
1083 case <-timer.C():
1084 }
1085}
1086
1087// streamErr reads the logical stream's stable terminal cause without treating a
1088// nil stream as a failure.
1089func streamErr(stream ports.BrokerLogicalConnection) error {
1090 if supervisorNil(stream) {
1091 return nil
1092 }
1093 return stream.Err()
1094}