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