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