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