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