1package client
2
3import (
4 "context"
5 "errors"
6 "io"
7 "sync"
8)
9
10// terminalReadResult is copied by terminalInputPump before the next Read so
11// terminal input remains owned by one lifecycle reader across reconnects.
12type terminalInputSource uint8
13
14const (
15 terminalInputHuman terminalInputSource = iota
16 terminalInputReply
17 terminalInputAutomation
18)
19
20type terminalReadResult struct {
21 data []byte
22 err error
23 source terminalInputSource
24 generation uint64
25 actionID uint64
26 endBatch bool
27 // keys/text carry the decoded ui-driver op for picker interception.
28 // Physical input leaves both empty; the picker consumer then decodes the
29 // raw bytes itself.
30 keys []string
31 text string
32}
33
34// terminalAutomationRequest is admitted only by the foreground scanner owner.
35// Its reply is buffered so cancellation never holds up physical input.
36type terminalAutomationRequest struct {
37 ctx context.Context
38 owner *UI
39 consumer uint64
40 record terminalReadResult
41 admitted chan bool
42 dispatched chan bool
43}
44
45// terminalInputPump is the sole consumer of the caller-owned terminal reader.
46// It is started once after raw mode is entered and is reused by each attach
47// scanner. A bare io.Reader cannot be interrupted, so a Read may stay blocked
48// after Run exits; suspend drops any result received while inactive and never
49// closes caller-owned stdin. Reconnects and later runs never add another reader.
50type terminalInputPump struct {
51 in io.Reader
52 // focus is the terminal focus decoded from this reader. The pump strips
53 // focus reports before any consumer sees the bytes.
54 focus *TerminalFocusState
55 // keys rewrites kitty key events to the encoding consumers read and
56 // tracks a held Ctrl, once the terminal reports every key event.
57 keys *terminalKeys
58 done chan struct{}
59 automation chan terminalAutomationRequest
60
61 mu sync.Mutex
62 // residual holds scanner bytes that were read but not delivered when an
63 // attempt is cancelled. It is consumed before the next terminal Read so a
64 // replacement scanner neither loses a partial marker nor replays bytes
65 // whose callbacks already ran.
66 residual []byte
67 pending *terminalReadResult
68 consumer uint64
69 delivering uint64
70 deliveringResidual bool
71 nextID uint64
72 activation uint64
73 closed bool
74 active bool
75 // readCause stores the cause the terminal reader ended with. The reader
76 // enqueues its final result for a consumer to report, but the exit watcher
77 // can win that race, so the cause is recorded here as well: a read failure
78 // must never be published as an orderly EOF.
79 readCause error
80 readyMu sync.Mutex
81 ready map[uint64]chan struct{}
82 claimChanged chan struct{}
83 space chan struct{}
84 state chan struct{}
85 exited chan struct{}
86 afterRevoke func() // test synchronization hook
87 startMu sync.Once
88 stopMu sync.Once
89}
90
91func newTerminalInputPump(in io.Reader) *terminalInputPump {
92 space := make(chan struct{}, 1)
93 space <- struct{}{}
94 return &terminalInputPump{
95 in: in,
96 focus: newTerminalFocusState(),
97 keys: newTerminalKeys(),
98 done: make(chan struct{}),
99 automation: make(chan terminalAutomationRequest),
100 activation: 1,
101 active: true,
102 ready: make(map[uint64]chan struct{}),
103 claimChanged: make(chan struct{}, 1),
104 space: space,
105 state: make(chan struct{}, 1),
106 exited: make(chan struct{}),
107 }
108}
109
110// tryClaim lets competing foreground owners decline admission without panicking.
111func (p *terminalInputPump) tryClaim() (uint64, bool) {
112 consumer, _, ok := p.claimWithBacklog()
113 return consumer, ok
114}
115
116// claimBacklog describes deliveries already waiting at claim. They are taken
117// in this order: the residual first, then the pending read.
118type claimBacklog struct {
119 residual, pending bool
120}
121
122// claimWithBacklog claims the pump and reports, atomically with the claim,
123// which deliveries were already waiting. They were read for an earlier owner,
124// not typed for this one.
125func (p *terminalInputPump) claimWithBacklog() (uint64, claimBacklog, bool) {
126 p.mu.Lock()
127 if p.consumer != 0 {
128 p.mu.Unlock()
129 return 0, claimBacklog{}, false
130 }
131 backlog := claimBacklog{residual: len(p.residual) != 0, pending: p.pending != nil}
132 p.nextID++
133 p.consumer = p.nextID
134 consumer := p.consumer
135 p.readyMu.Lock()
136 p.ready[consumer] = make(chan struct{}, 1)
137 p.readyMu.Unlock()
138 ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
139 p.mu.Unlock()
140 p.signalClaimChanged()
141 if ready {
142 p.signalReady(consumer)
143 }
144 return consumer, backlog, true
145}
146
147// revoke invalidates an attempt before its replacement is allowed to claim
148// input. Pending bytes are deliberately retained for that replacement.
149func (p *terminalInputPump) revoke(consumer uint64) {
150 p.mu.Lock()
151 if p.consumer == consumer {
152 p.consumer = 0
153 }
154 if p.delivering == consumer {
155 // Legacy attach-attempt replacement remains lossless. Autonomous
156 // supervisor ownership uses ack-before-picker-handoff and
157 // drop-before-attachment-handoff, so bytes cannot cross owner kinds.
158 p.delivering = 0
159 p.deliveringResidual = false
160 }
161 p.mu.Unlock()
162 p.readyMu.Lock()
163 delete(p.ready, consumer)
164 p.readyMu.Unlock()
165 p.signalClaimChanged()
166 if p.afterRevoke != nil {
167 p.afterRevoke()
168 }
169}
170
171func (p *terminalInputPump) signalClaimChanged() {
172 select {
173 case p.claimChanged <- struct{}{}:
174 default:
175 }
176}
177
178func (p *terminalInputPump) readyFor(consumer uint64) <-chan struct{} {
179 p.readyMu.Lock()
180 defer p.readyMu.Unlock()
181 return p.ready[consumer]
182}
183
184func (p *terminalInputPump) signalReady(consumer uint64) {
185 if consumer == 0 {
186 return
187 }
188 p.readyMu.Lock()
189 ready := p.ready[consumer]
190 p.readyMu.Unlock()
191 if ready == nil {
192 return
193 }
194 select {
195 case ready <- struct{}{}:
196 default:
197 }
198}
199
200// enqueue retains at most one completed terminal read. The reader blocks
201// before publishing another result, keeping handoff buffering bounded.
202func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
203 select {
204 case <-p.done:
205 return false
206 case <-p.space:
207 }
208 p.mu.Lock()
209 if p.closed || !p.active || p.activation != activation {
210 p.mu.Unlock()
211 p.space <- struct{}{}
212 return false
213 }
214 p.pending = &result
215 consumer := p.consumer
216 p.mu.Unlock()
217 p.signalReady(consumer)
218 return true
219}
220
221// take leases a ready result only to the current, non-cancelled consumer.
222// The raw read remains pending until ack, so revocation or cancellation before
223// scanner delivery leaves the exact bytes available to the next attempt.
224func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
225 p.mu.Lock()
226 if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
227 p.mu.Unlock()
228 return terminalReadResult{}, false
229 }
230 if len(p.residual) != 0 {
231 result := terminalReadResult{data: append([]byte(nil), p.residual...)}
232 p.delivering = consumer
233 p.deliveringResidual = true
234 p.mu.Unlock()
235 return result, true
236 }
237 if p.pending == nil {
238 p.mu.Unlock()
239 return terminalReadResult{}, false
240 }
241 result := *p.pending
242 p.delivering = consumer
243 p.deliveringResidual = false
244 p.mu.Unlock()
245 return result, true
246}
247
248// preserveResidual records the undecided marker prefix after callbacks for the
249// rest of its read have completed. It is intentionally bounded by the marker
250// scanner's fixed response prefixes and is replayed before later terminal
251// reads on a replacement attempt.
252func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
253 if len(data) == 0 {
254 return
255 }
256 p.mu.Lock()
257 defer p.mu.Unlock()
258 if p.consumer != consumer {
259 return
260 }
261 p.residual = append(p.residual[:0], data...)
262}
263
264// appendResidual appends bytes the consumer took but did not deliver, so the
265// next owner replays them after any earlier residual. It reports false when
266// the bound would be exceeded; the bytes are then dropped.
267func (p *terminalInputPump) appendResidual(consumer uint64, data []byte, limit int) bool {
268 p.mu.Lock()
269 defer p.mu.Unlock()
270 if p.consumer != consumer || len(p.residual)+len(data) > limit {
271 return false
272 }
273 p.residual = append(p.residual, data...)
274 return true
275}
276
277// ack commits scanner delivery of a leased read and lets the lifecycle reader
278// accept the next result. Only the consumer that holds the lease can ack it.
279// dropOwned disposes every undecided byte at an autonomous owner-class
280// boundary. Picker-queued bytes are dropped before attachment claim, and
281// attachment-queued bytes are dropped before picker claim. No delivery,
282// pending read, or preserved residual may cross between those owner classes.
283// revoke remains lossless for attachment-to-attachment replacement.
284func (p *terminalInputPump) dropOwned(consumer uint64) {
285 p.mu.Lock()
286 if p.consumer != consumer {
287 p.mu.Unlock()
288 return
289 }
290 hadPending := p.pending != nil
291 p.pending = nil
292 p.residual = nil
293 if p.delivering == consumer {
294 p.delivering = 0
295 p.deliveringResidual = false
296 }
297 p.mu.Unlock()
298 if hadPending {
299 select {
300 case p.space <- struct{}{}:
301 default:
302 }
303 }
304}
305
306// dropUnclaimed disposes pending and preserved bytes while no consumer holds
307// the claim. The supervisor calls it when the terminal goes back to the
308// picker, so session input kept for a resume never reaches the picker.
309func (p *terminalInputPump) dropUnclaimed() {
310 p.mu.Lock()
311 if p.consumer != 0 {
312 p.mu.Unlock()
313 return
314 }
315 hadPending := p.pending != nil
316 p.pending = nil
317 p.residual = nil
318 p.mu.Unlock()
319 if hadPending {
320 select {
321 case p.space <- struct{}{}:
322 default:
323 }
324 }
325}
326
327func (p *terminalInputPump) ack(consumer uint64) {
328 p.mu.Lock()
329 if p.consumer != consumer || p.delivering != consumer {
330 p.mu.Unlock()
331 return
332 }
333 wasResidual := p.deliveringResidual
334 if wasResidual {
335 p.residual = nil
336 } else {
337 p.pending = nil
338 }
339 p.delivering = 0
340 p.deliveringResidual = false
341 more := len(p.residual) != 0 || p.pending != nil
342 p.mu.Unlock()
343 if wasResidual {
344 if more {
345 p.signalReady(consumer)
346 }
347 return
348 }
349 p.space <- struct{}{}
350}
351
352func (p *terminalInputPump) finish() {
353 p.mu.Lock()
354 p.closed = true
355 p.mu.Unlock()
356 p.signalClaimChanged()
357 p.signalState()
358}
359
360func (p *terminalInputPump) signalState() {
361 select {
362 case p.state <- struct{}{}:
363 default:
364 }
365}
366
367func (p *terminalInputPump) waitForActive() (uint64, bool) {
368 for {
369 p.mu.Lock()
370 active, closed, activation := p.active, p.closed, p.activation
371 p.mu.Unlock()
372 if closed {
373 return 0, false
374 }
375 if active {
376 return activation, true
377 }
378 select {
379 case <-p.done:
380 return 0, false
381 case <-p.state:
382 }
383 }
384}
385
386func (p *terminalInputPump) start() {
387 p.startMu.Do(func() {
388 go func() {
389 defer close(p.exited)
390 buf := make([]byte, stdinBufSize)
391 for {
392 activation, ok := p.waitForActive()
393 if !ok {
394 return
395 }
396 n, err := p.in.Read(buf)
397 result := terminalReadResult{err: err}
398 if n > 0 {
399 data := stripTerminalFocusReports(p.focus, buf[:n])
400 if data = p.keys.normalize(data, p.focus.Focus()); len(data) != 0 {
401 result.data = append([]byte(nil), data...)
402 } else if err == nil {
403 // A read that only carried focus reports or dropped key
404 // events delivers nothing.
405 continue
406 }
407 }
408 if !p.enqueue(result, activation) {
409 if p.isClosed() {
410 return
411 }
412 continue
413 }
414 if err != nil {
415 p.setReadCause(err)
416 p.finish()
417 return
418 }
419 }
420 }()
421 })
422}
423
424// setReadCause records the cause the terminal reader ended with. An orderly EOF
425// is the absence of a failure and stays zero, so cause reports io.EOF for it.
426func (p *terminalInputPump) setReadCause(err error) {
427 if err == nil || errors.Is(err, io.EOF) {
428 return
429 }
430 p.mu.Lock()
431 if p.readCause == nil {
432 p.readCause = err
433 }
434 p.mu.Unlock()
435}
436
437// cause reports why the terminal read ended: the read failure when there was
438// one, otherwise an orderly EOF.
439func (p *terminalInputPump) cause() error {
440 p.mu.Lock()
441 defer p.mu.Unlock()
442 if p.readCause != nil {
443 return p.readCause
444 }
445 return io.EOF
446}
447
448func (p *terminalInputPump) isClosed() bool {
449 p.mu.Lock()
450 defer p.mu.Unlock()
451 return p.closed
452}
453
454// suspend drops input completed after an attach ends and parks the lifecycle
455// reader before its next Read. A later Run resumes this same reader, avoiding
456// both input replay between runs and a competing read on caller-owned stdin.
457func (p *terminalInputPump) suspend() {
458 p.mu.Lock()
459 p.active = false
460 discarded := p.pending != nil
461 p.residual = nil
462 p.pending = nil
463 p.delivering = 0
464 p.deliveringResidual = false
465 p.mu.Unlock()
466 if discarded {
467 select {
468 case p.space <- struct{}{}:
469 default:
470 }
471 }
472 p.signalState()
473 p.signalClaimChanged()
474}
475
476func (p *terminalInputPump) stop() {
477 p.stopMu.Do(func() {
478 // Mark closure while holding the same mutex that publishes a completed
479 // Read. A Read which returns after stop therefore cannot publish even
480 // when both done and space are selectable in enqueue.
481 p.mu.Lock()
482 p.closed = true
483 p.mu.Unlock()
484 close(p.done)
485 p.signalClaimChanged()
486 p.signalState()
487 })
488}