1package daemon
2
3import (
4 "errors"
5 "strconv"
6 "sync"
7 "sync/atomic"
8
9 renderer "github.com/bnema/vev-vt/ansi"
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/domain/terminalcap"
12 "github.com/bnema/vev/internal/protocol"
13 "github.com/bnema/vev/internal/usecase/colorprofile"
14)
15
16// attachmentOutput owns the terminal-output dependency chain and emitted
17// render state for one attached client. Each incremental output is rendered
18// from the preceding emitted frame and remains pipelineable without waiting for an
19// ACK. Cumulative ACKs advance the bounded paint window.
20// A reset frame starts a new epoch, clears the dependency chain, and advertises
21// state 0 as its base.
22//
23// Callers serialize access with attachedClient.sendMu. A prepared transaction
24// must be sent or committed as a no-op before the next transaction is prepared.
25type attachmentOutput struct {
26 renderer *renderer.Renderer
27 epoch uint64
28 next uint64
29 acked uint64
30 generation uint64
31 maxOutstanding uint64
32 outstandingAtomic atomic.Uint64
33 maxOutstandingAtomic atomic.Uint64
34 // unackedBytes charges Output data sent and not yet acknowledged, keyed by
35 // the state an ACK must reach to release it. Side-effect output carries no
36 // state of its own, so it is charged to the next state. Charges stay in
37 // non-decreasing state order and merge per state, so the slice holds at most
38 // maxOutstanding+1 entries. Guarded by the view lock; unackedBytesAtomic
39 // mirrors the total for lock-free capacity reads.
40 unackedBytes []unackedOutputBytes
41 unackedBytesAtomic atomic.Uint64
42 forceSnapshot bool
43 initialized bool
44 viewPublication uint64
45 lastViewContext protocol.ViewContext
46 lastCursor cursorOut
47 lastRoutePosition protocol.RoutePosition
48 graphicsOutput *graphicsOutputState
49 graphicsUnsupportedWarned atomic.Bool
50 attachment *attachedClient
51 stateMu sync.Mutex
52}
53
54// unackedOutputBytes is the Output data released once the client acknowledges
55// state.
56type unackedOutputBytes struct {
57 state uint64
58 bytes uint64
59}
60
61// chargeBytesLocked records n bytes released once state is acknowledged.
62// Caller holds the view lock.
63func (s *attachmentOutput) chargeBytesLocked(state uint64, n int) {
64 if n <= 0 {
65 return
66 }
67 if last := len(s.unackedBytes) - 1; last >= 0 && s.unackedBytes[last].state == state {
68 s.unackedBytes[last].bytes += uint64(n)
69 } else {
70 s.unackedBytes = append(s.unackedBytes, unackedOutputBytes{state: state, bytes: uint64(n)})
71 }
72 s.unackedBytesAtomic.Add(uint64(n))
73}
74
75// releaseBytesLocked drops every charge an acknowledgement of state covers.
76// Caller holds the view lock.
77func (s *attachmentOutput) releaseBytesLocked(state uint64) {
78 released, kept := uint64(0), 0
79 for _, charge := range s.unackedBytes {
80 if charge.state <= state {
81 released += charge.bytes
82 continue
83 }
84 s.unackedBytes[kept] = charge
85 kept++
86 }
87 clear(s.unackedBytes[kept:])
88 s.unackedBytes = s.unackedBytes[:kept]
89 s.unackedBytesAtomic.Add(-released)
90}
91
92// resetBytesLocked forgets every charge: a new epoch acknowledges nothing the
93// old one sent. Caller holds the view lock.
94func (s *attachmentOutput) resetBytesLocked() {
95 s.unackedBytes = s.unackedBytes[:0]
96 s.unackedBytesAtomic.Store(0)
97}
98
99// reconfigureAttachmentOutput applies a replacement terminal's declared
100// capabilities to attachment-owned output state. Caller holds d.mu.
101func (d *Daemon) reconfigureAttachmentOutput(sess *session, ac *attachedClient, h protocol.Hello) {
102 if d == nil || ac == nil || ac.output == nil {
103 return
104 }
105 ac.terminalCapabilities.KittyGraphics = h.KittyDirectGraphics
106 // The replacement terminal's color claim wins over whatever the previous
107 // link negotiated: its renderer must encode for the terminal now attached.
108 // Only the mode changes the encoding; the source is stored but never
109 // forces a renderer swap.
110 color := terminalcap.Resolve(h.Env, h.Color).Color
111 swap := color.Mode != ac.terminalCapabilities.Color.Mode
112 ac.terminalCapabilities.Color = color
113 if swap {
114 ac.output.setColorProfile(colorprofile.Profile(color))
115 }
116 // A replacement terminal declares its own keyboard protocol.
117 if ac.keys != nil {
118 ac.keys.SetKittyKeyboard(h.KittyKeyboard)
119 }
120 ac.output.graphicsUnsupportedWarned.Store(false)
121 // A replacement connection is a different outer terminal. Retire the old
122 // attachment-local state without writing cleanup to the new terminal, then
123 // allocate a fresh namespace for any replay it accepts.
124 d.discardAttachmentOutputLocked(ac)
125 if !h.KittyDirectGraphics {
126 // The parked transport has already been retired. Do not emit cleanup into
127 // a replacement connection that did not declare graphics support.
128 return
129 }
130 if namespace, fence := d.reserveGraphicsNamespaceLeaseLocked(graphicsNamespaceKey(sess, ac.clientID)); namespace != 0 {
131 ac.output.graphicsOutput = newGraphicsOutputStateWithLease(namespace, fence)
132 return
133 }
134 // Namespace exhaustion fails closed; ANSI output remains available.
135 ac.terminalCapabilities.KittyGraphics = false
136}
137
138// setColorProfile swaps in a renderer for profile. The next frame is a full
139// repaint (rebaseLocked forces a snapshot), so no emitted-state diff crosses
140// the profile change.
141func (s *attachmentOutput) setColorProfile(profile renderer.ColorProfile) {
142 if s == nil {
143 return
144 }
145 s.lockView()
146 defer s.unlockView()
147 s.renderer = renderer.NewWithColorProfile(renderer.Capabilities{}, profile)
148 s.rebaseLocked()
149}
150
151// lockView serializes an attachment's view publication with every output
152// state transition. It deliberately does not take sendMu: callers that hold
153// architecture locks can publish a view without reversing the paint path's
154// sendMu -> session lock order.
155func (s *attachmentOutput) lockView() {
156 if s == nil {
157 return
158 }
159 if s.attachment == nil {
160 s.stateMu.Lock()
161 return
162 }
163 s.attachment.viewMu.Lock()
164}
165
166func (s *attachmentOutput) unlockView() {
167 if s == nil {
168 return
169 }
170 if s.attachment == nil {
171 s.stateMu.Unlock()
172 return
173 }
174 s.attachment.viewMu.Unlock()
175}
176
177func newOutputStateStream(windowSize ...uint8) *attachmentOutput {
178 return newOutputStateStreamForProfile(renderer.ColorProfileTrueColor, windowSize...)
179}
180
181func newOutputStateStreamForCapabilities(capabilities terminalcap.Capabilities, windowSize ...uint8) *attachmentOutput {
182 return newOutputStateStreamForProfile(colorprofile.Profile(capabilities.Color), windowSize...)
183}
184
185func newOutputStateStreamForProfile(profile renderer.ColorProfile, windowSize ...uint8) *attachmentOutput {
186 window := uint8(maxUnackedOutputStates)
187 if len(windowSize) > 0 {
188 window = normalizeOutputWindow(windowSize[0])
189 }
190 stream := &attachmentOutput{
191 renderer: renderer.NewWithColorProfile(renderer.Capabilities{}, profile),
192 epoch: 1,
193 maxOutstanding: uint64(window),
194 }
195 stream.maxOutstandingAtomic.Store(uint64(window))
196 return stream
197}
198
199type cursorCandidate struct {
200 data []byte
201 next cursorOut
202}
203
204// preparedAttachmentOutput is one speculative terminal-output transaction.
205// ANSI renderer state and cursor state advance together only after the output
206// frame is accepted by the attachment's state stream.
207type preparedAttachmentOutput struct {
208 output *attachmentOutput
209 ansi *preparedOutput
210 graphics *preparedGraphicsOutput
211 cursor cursorCandidate
212 data []byte
213}
214
215func (o *attachmentOutput) prepareFrame(d *Daemon, state *capturedRenderState, frame renderer.Frame, damage []renderer.Damage, reset bool, desired cursorOut) (*preparedAttachmentOutput, error) {
216 ansi, err := o.prepare(frame, damage, reset)
217 if err != nil {
218 return nil, err
219 }
220 context := state.viewContext()
221 ansi.context = &context
222 ansi.viewRevision = state.view.revision
223 cursor := o.prepareCursorTail(desired, len(ansi.data) > 0)
224 graphicsReset := reset || o.forceSnapshot || !o.initialized
225 // Images share the frame's data limit. The unacknowledged byte window gates
226 // the next frame rather than this one: an upload must fit one frame, so
227 // clamping it here would drop large images and their deletes outright.
228 graphicsBudget := max(protocol.MaxOutputDataLen-len(ansi.data)-len(cursor.data), 0)
229 preparedGraphics, err := graphicsOutputDataWithDaemonLimit(d, state, o.attachment, graphicsReset, graphicsBudget)
230 if err != nil {
231 return nil, err
232 }
233 data := append([]byte(nil), ansi.data...)
234 if preparedGraphics != nil {
235 data = append(data, preparedGraphics.data...)
236 }
237 data = append(data, cursor.data...)
238 return &preparedAttachmentOutput{output: o, ansi: ansi, graphics: preparedGraphics, cursor: cursor, data: data}, nil
239}
240
241func (p *preparedAttachmentOutput) send(echoAck uint64, send func(protocol.Output) error) error {
242 if p == nil || p.ansi == nil {
243 return nil
244 }
245 sendOutput := send
246 if p.graphics != nil {
247 sendOutput = func(output protocol.Output) error {
248 p.graphics.markSendAttempted()
249 return send(output)
250 }
251 }
252 if err := p.ansi.send(p.data, echoAck, sendOutput); err != nil {
253 p.abortGraphics()
254 return err
255 }
256 p.commitIfSent()
257 return nil
258}
259
260func (p *preparedAttachmentOutput) publishNoBytes(confirm bool, send func(protocol.UIViewUpdate) error) error {
261 if p == nil || p.ansi == nil {
262 return nil
263 }
264 if err := p.ansi.publishNoBytes(confirm, send); err != nil {
265 p.abortGraphics()
266 return err
267 }
268 p.commitIfSent()
269 return nil
270}
271
272func (p *preparedAttachmentOutput) commitNoSend() {
273 if p == nil || p.ansi == nil {
274 return
275 }
276 p.ansi.commitNoSend()
277 p.commitIfSent()
278}
279
280func (p *preparedAttachmentOutput) commitIfSent() {
281 if p == nil || p.ansi == nil || !p.ansi.sent {
282 p.abortGraphics()
283 return
284 }
285 if p.graphics != nil {
286 p.graphics.commit()
287 }
288 if p.output != nil {
289 p.output.lastCursor = p.cursor.next
290 }
291}
292
293func (p *preparedAttachmentOutput) abort() {
294 if p != nil {
295 p.abortGraphics()
296 }
297}
298
299func (p *preparedAttachmentOutput) abortGraphics() {
300 if p != nil && p.graphics != nil {
301 p.graphics.abort()
302 }
303}
304
305func (p *preparedAttachmentOutput) sent() bool {
306 return p != nil && p.ansi != nil && p.ansi.sent
307}
308
309func (o *attachmentOutput) prepareCursorTail(desired cursorOut, force bool) cursorCandidate {
310 candidate := cursorCandidate{next: desired}
311 candidate.next.valid = true
312 prev := o.lastCursor
313 changed := force || !prev.valid || prev.hidden != desired.hidden || prev.row != desired.row || prev.col != desired.col || prev.style != desired.style || prev.hasStyle != desired.hasStyle
314 if !changed {
315 return candidate
316 }
317 if desired.hidden {
318 candidate.data = []byte("\x1b[?25l")
319 return candidate
320 }
321 candidate.data = append(candidate.data, "\x1b["...)
322 candidate.data = strconv.AppendInt(candidate.data, int64(desired.row+1), 10)
323 candidate.data = append(candidate.data, ';')
324 candidate.data = strconv.AppendInt(candidate.data, int64(desired.col+1), 10)
325 candidate.data = append(candidate.data, 'H')
326 if !prev.valid || prev.hidden || prev.style != desired.style || prev.hasStyle != desired.hasStyle {
327 candidate.data = append(candidate.data, "\x1b["...)
328 candidate.data = strconv.AppendInt(candidate.data, int64(desired.style), 10)
329 candidate.data = append(candidate.data, " q"...)
330 }
331 candidate.data = append(candidate.data, "\x1b[?25h"...)
332 return candidate
333}
334
335type preparedOutput struct {
336 stream *attachmentOutput
337 draw renderer.PreparedDraw
338 base uint64
339 next uint64
340 epoch uint64
341 generation uint64
342 connectionGeneration uint64
343 viewRevision uint64
344 context *protocol.ViewContext
345 // boundary is the exact accepted semantic publication, not mutable stream
346 // state that a concurrent view rebase could replace before receipt send.
347 boundary protocol.UIReceipt
348 size domain.Size
349 data []byte
350 reset bool
351 attempted bool
352 sent bool
353}
354
355func (s *attachmentOutput) prepare(frame renderer.Frame, damage []renderer.Damage, reset bool) (*preparedOutput, error) {
356 s.lockView()
357 defer s.unlockView()
358 if s.epoch == 0 {
359 s.epoch = 1
360 }
361 if reset && s.initialized {
362 s.rebaseLocked()
363 }
364 reset = reset || s.forceSnapshot || !s.initialized
365 if s.next == ^uint64(0) {
366 return nil, errors.New("output state number exhausted")
367 }
368 draw, err := s.renderer.Prepare(frame, damage, reset)
369 if err != nil {
370 return nil, err
371 }
372 s.initialized = true
373 generation, epoch, viewRevision := s.fenceLocked()
374 return &preparedOutput{
375 stream: s,
376 draw: draw,
377 base: s.next,
378 next: s.next + 1,
379 epoch: epoch,
380 generation: s.generation,
381 connectionGeneration: generation,
382 viewRevision: viewRevision,
383 size: domain.Size{Cols: frame.Width, Rows: frame.Height},
384 data: draw.Bytes(),
385 reset: reset,
386 }, nil
387}
388
389func (p *preparedOutput) send(data []byte, echoAck uint64, send func(protocol.Output) error) error {
390 if p == nil || p.stream == nil || p.attempted {
391 return nil
392 }
393 p.stream.lockView()
394 defer p.stream.unlockView()
395 p.attempted = true
396 if !p.currentLocked() {
397 return nil
398 }
399 if len(data) == 0 {
400 p.commit()
401 p.sent = true
402 return nil
403 }
404 base := p.base
405 if p.reset {
406 base = 0
407 }
408 var context *protocol.ViewContext
409 if p.context != nil {
410 if p.stream.viewPublication == ^uint64(0) {
411 return errors.New("view publication number exhausted")
412 }
413 candidate := *p.context
414 candidate.Publication = p.stream.viewPublication + 1
415 context = &candidate
416 }
417 out, err := marshalOutputState(data, p.epoch, base, p.next, echoAck, p.viewRevision, p.size, p.reset, context)
418 if err != nil {
419 p.stream.forceSnapshot = true
420 return err
421 }
422 if err := send(out); err != nil {
423 p.stream.forceSnapshot = true
424 return err
425 }
426 p.commit()
427 if context != nil {
428 p.stream.viewPublication = context.Publication
429 p.stream.lastViewContext = *context
430 p.boundary = protocol.UIReceipt{Epoch: p.epoch, State: p.next, ViewPublication: context.Publication, Outcome: protocol.UIReceiptProcessed}
431 }
432 p.stream.next = p.next
433 p.stream.chargeBytesLocked(p.next, len(data))
434 p.stream.publishOutstanding()
435 p.sent = true
436 return nil
437}
438
439func (p *preparedOutput) commitNoSend() {
440 if p == nil || p.stream == nil || p.attempted {
441 return
442 }
443 p.stream.lockView()
444 defer p.stream.unlockView()
445 p.attempted = true
446 if !p.currentLocked() || len(p.data) != 0 {
447 return
448 }
449 p.commit()
450 p.sent = true
451}
452
453// publishNoBytes commits a composition without advancing the terminal replay
454// chain. A fence can request a fresh semantic boundary even for an unchanged
455// view. Callers retain the ordinary attachment send/effect ownership.
456func (p *preparedOutput) publishNoBytes(confirm bool, send func(protocol.UIViewUpdate) error) error {
457 if p == nil || p.stream == nil || p.attempted {
458 return nil
459 }
460 p.stream.lockView()
461 defer p.stream.unlockView()
462 p.attempted = true
463 if !p.currentLocked() || len(p.data) != 0 {
464 return nil
465 }
466 if p.context != nil {
467 candidate := *p.context
468 candidate.Publication = p.stream.viewPublication
469 if confirm || candidate != p.stream.lastViewContext {
470 if p.stream.next == 0 {
471 return errors.New("view update requires committed output")
472 }
473 if candidate.Publication == ^uint64(0) {
474 return errors.New("view publication number exhausted")
475 }
476 candidate.Publication++
477 if err := candidate.Validate(); err != nil {
478 return err
479 }
480 if err := send(protocol.UIViewUpdate{Epoch: p.epoch, State: p.stream.next, Context: candidate}); err != nil {
481 return err
482 }
483 p.stream.viewPublication = candidate.Publication
484 p.stream.lastViewContext = candidate
485 p.boundary = protocol.UIReceipt{Epoch: p.epoch, State: p.stream.next, ViewPublication: candidate.Publication, Outcome: protocol.UIReceiptProcessed}
486 }
487 }
488 p.commit()
489 p.sent = true
490 return nil
491}
492
493func (p *preparedOutput) commit() {
494 p.draw.Commit()
495 p.stream.forceSnapshot = false
496}
497
498func (s *attachmentOutput) currentEpoch() uint64 {
499 s.lockView()
500 defer s.unlockView()
501 return s.currentEpochLocked()
502}
503
504func (s *attachmentOutput) currentEpochLocked() uint64 {
505 if s.epoch == 0 {
506 s.epoch = 1
507 }
508 return s.epoch
509}
510
511func (s *attachmentOutput) fenceLocked() (uint64, uint64, uint64) {
512 if s == nil {
513 return 0, 1, 0
514 }
515 generation, revision := uint64(0), uint64(0)
516 if s.attachment != nil {
517 generation = s.attachment.lifecycle.generationValue()
518 revision = s.attachment.view.revision
519 }
520 return generation, s.currentEpochLocked(), revision
521}
522
523func (s *attachmentOutput) preparedCurrentLocked(generation, epoch, viewRevision, base uint64) bool {
524 if s == nil || s.currentEpochLocked() != epoch || s.generation != generation || s.next != base {
525 return false
526 }
527 if s.attachment == nil {
528 return true
529 }
530 return s.attachment.view.revision == viewRevision
531}
532
533func (p *preparedOutput) currentLocked() bool {
534 return p != nil && p.stream != nil && p.stream.preparedCurrentLocked(p.generation, p.epoch, p.viewRevision, p.base) &&
535 (p.stream.attachment == nil || p.stream.attachment.lifecycle.generationValue() == p.connectionGeneration)
536}
537
538func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision uint64, size domain.Size, full bool, context *protocol.ViewContext) (protocol.Output, error) {
539 output := protocol.Output{
540 Epoch: epoch, Base: base, New: next, Echo: echoAck, ViewRevision: viewRevision,
541 Size: size, Full: full, Data: data, Context: context,
542 }
543 if err := protocol.ValidateOutput(output); err != nil {
544 return protocol.Output{}, err
545 }
546 return output, nil
547}
548
549// sideEffectLocked requires the attachment view lock for both output creation
550// and connection emission. Keeping that lock across the send prevents a view
551// publication from overtaking output fenced just before it.
552func (s *attachmentOutput) sideEffectLocked(data []byte, echoAck uint64) (protocol.Output, error) {
553 _, epoch, viewRevision := s.fenceLocked()
554 size := domain.Size{Cols: 1, Rows: 1}
555 if s != nil && s.attachment != nil {
556 size = s.attachment.sizeSnapshot()
557 }
558 return marshalOutputState(data, epoch, 0, 0, echoAck, viewRevision, size, false, nil)
559}
560
561// sideEffectSentLocked charges a side effect that reached the transport. It has
562// no state of its own, so the ACK of the next state frame releases it: every
563// state frame is sent after it on the same ordered stream. Caller holds the view
564// lock that built output, so the epoch cannot have moved.
565func (s *attachmentOutput) sideEffectSentLocked(output protocol.Output) {
566 if s == nil || output.Epoch != s.currentEpochLocked() {
567 return
568 }
569 s.chargeBytesLocked(s.next+1, len(output.Data))
570}
571
572func (s *attachmentOutput) ack(epoch, state uint64) bool {
573 if s == nil {
574 return false
575 }
576 s.lockView()
577 defer s.unlockView()
578 if epoch != s.currentEpochLocked() || state > s.next || state <= s.acked {
579 return false
580 }
581 s.acked = state
582 s.releaseBytesLocked(state)
583 s.publishOutstanding()
584 return true
585}
586
587func (s *attachmentOutput) rebase() {
588 if s == nil {
589 return
590 }
591 s.lockView()
592 defer s.unlockView()
593 s.rebaseLocked()
594}
595
596// rebaseAttachment retires all output state coupled to the attachment's
597// current session or transport while preserving independent client state.
598func (s *attachmentOutput) rebaseAttachment() {
599 if s == nil {
600 return
601 }
602 s.rebase()
603 s.lastRoutePosition = protocol.RoutePosition{}
604}
605
606// rebaseLocked is called while the attachment view lock is held. Publishing
607// the next view revision takes this path before making that revision visible.
608func (s *attachmentOutput) rebaseLocked() {
609 if s == nil {
610 return
611 }
612 epoch := s.currentEpochLocked()
613 if epoch == ^uint64(0) {
614 return
615 }
616 s.epoch = epoch + 1
617 s.next = 0
618 s.acked = 0
619 s.resetBytesLocked()
620 s.generation++
621 s.publishOutstanding()
622 s.maxOutstandingAtomic.Store(s.maxOutstanding)
623 s.renderer.Reset()
624 s.forceSnapshot = true
625 s.initialized = false
626}
627
628func (s *attachmentOutput) publishOutstanding() {
629 if s == nil {
630 return
631 }
632 s.outstandingAtomic.Store(s.next - s.acked)
633}
634
635// syncCapacityLocked refreshes the lock-free readiness snapshot after a
636// caller mutates the legacy fields under attachedClient.sendMu.
637func (s *attachmentOutput) syncCapacityLocked() {
638 if s == nil {
639 return
640 }
641 s.lockView()
642 defer s.unlockView()
643 s.maxOutstandingAtomic.Store(s.maxOutstanding)
644 s.publishOutstanding()
645}
646
647func (s *attachmentOutput) setWindow(window uint8) {
648 if s == nil {
649 return
650 }
651 s.maxOutstanding = uint64(normalizeOutputWindow(window))
652 s.maxOutstandingAtomic.Store(s.maxOutstanding)
653}
654
655// atCapacity reports whether the next state frame must wait for an ACK: the
656// frame window is full, or unacknowledged Output data reached
657// protocol.MaxOutputWindowBytes while a state frame is in flight. With no state
658// frame in flight the next frame is always allowed, since only its ACK can
659// release bytes charged by side effects.
660func (s *attachmentOutput) atCapacity() bool {
661 if s == nil {
662 return false
663 }
664 maxOutstanding := s.maxOutstandingAtomic.Load()
665 if maxOutstanding == 0 {
666 maxOutstanding = s.maxOutstanding
667 }
668 outstanding := s.outstandingAtomic.Load()
669 return outstanding >= maxOutstanding ||
670 (outstanding > 0 && s.unackedBytesAtomic.Load() >= protocol.MaxOutputWindowBytes)
671}
672
673// committedState reports the latest committed output state number. It takes
674// only the view lock, so barrier snapshots can read it without joining the
675// sendMu-ordered paint path.
676func (s *attachmentOutput) committedState() uint64 {
677 if s == nil {
678 return 0
679 }
680 s.lockView()
681 defer s.unlockView()
682 return s.next
683}