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 p.mu.Lock()
109 if p.consumer != 0 {
110 p.mu.Unlock()
111 return 0, false
112 }
113 p.nextID++
114 p.consumer = p.nextID
115 consumer := p.consumer
116 p.readyMu.Lock()
117 p.ready[consumer] = make(chan struct{}, 1)
118 p.readyMu.Unlock()
119 ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
120 p.mu.Unlock()
121 p.signalClaimChanged()
122 if ready {
123 p.signalReady(consumer)
124 }
125 return consumer, true
126}
127
128// revoke invalidates an attempt before its replacement is allowed to claim
129// input. Pending bytes are deliberately retained for that replacement.
130func (p *terminalInputPump) revoke(consumer uint64) {
131 p.mu.Lock()
132 if p.consumer == consumer {
133 p.consumer = 0
134 }
135 if p.delivering == consumer {
136 // Legacy attach-attempt replacement remains lossless. Autonomous
137 // supervisor ownership uses ack-before-picker-handoff and
138 // drop-before-attachment-handoff, so bytes cannot cross owner kinds.
139 p.delivering = 0
140 p.deliveringResidual = false
141 }
142 p.mu.Unlock()
143 p.readyMu.Lock()
144 delete(p.ready, consumer)
145 p.readyMu.Unlock()
146 p.signalClaimChanged()
147 if p.afterRevoke != nil {
148 p.afterRevoke()
149 }
150}
151
152func (p *terminalInputPump) signalClaimChanged() {
153 select {
154 case p.claimChanged <- struct{}{}:
155 default:
156 }
157}
158
159func (p *terminalInputPump) readyFor(consumer uint64) <-chan struct{} {
160 p.readyMu.Lock()
161 defer p.readyMu.Unlock()
162 return p.ready[consumer]
163}
164
165func (p *terminalInputPump) signalReady(consumer uint64) {
166 if consumer == 0 {
167 return
168 }
169 p.readyMu.Lock()
170 ready := p.ready[consumer]
171 p.readyMu.Unlock()
172 if ready == nil {
173 return
174 }
175 select {
176 case ready <- struct{}{}:
177 default:
178 }
179}
180
181// enqueue retains at most one completed terminal read. The reader blocks
182// before publishing another result, keeping handoff buffering bounded.
183func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
184 select {
185 case <-p.done:
186 return false
187 case <-p.space:
188 }
189 p.mu.Lock()
190 if p.closed || !p.active || p.activation != activation {
191 p.mu.Unlock()
192 p.space <- struct{}{}
193 return false
194 }
195 p.pending = &result
196 consumer := p.consumer
197 p.mu.Unlock()
198 p.signalReady(consumer)
199 return true
200}
201
202// take leases a ready result only to the current, non-cancelled consumer.
203// The raw read remains pending until ack, so revocation or cancellation before
204// scanner delivery leaves the exact bytes available to the next attempt.
205func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
206 p.mu.Lock()
207 if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
208 p.mu.Unlock()
209 return terminalReadResult{}, false
210 }
211 if len(p.residual) != 0 {
212 result := terminalReadResult{data: append([]byte(nil), p.residual...)}
213 p.delivering = consumer
214 p.deliveringResidual = true
215 p.mu.Unlock()
216 return result, true
217 }
218 if p.pending == nil {
219 p.mu.Unlock()
220 return terminalReadResult{}, false
221 }
222 result := *p.pending
223 p.delivering = consumer
224 p.deliveringResidual = false
225 p.mu.Unlock()
226 return result, true
227}
228
229// preserveResidual records the undecided marker prefix after callbacks for the
230// rest of its read have completed. It is intentionally bounded by the marker
231// scanner's fixed response prefixes and is replayed before later terminal
232// reads on a replacement attempt.
233func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
234 if len(data) == 0 {
235 return
236 }
237 p.mu.Lock()
238 defer p.mu.Unlock()
239 if p.consumer != consumer {
240 return
241 }
242 p.residual = append(p.residual[:0], data...)
243}
244
245// ack commits scanner delivery of a leased read and lets the lifecycle reader
246// accept the next result. Only the consumer that holds the lease can ack it.
247// dropOwned disposes every undecided byte at an autonomous owner-class
248// boundary. Picker-queued bytes are dropped before attachment claim, and
249// attachment-queued bytes are dropped before picker claim. No delivery,
250// pending read, or preserved residual may cross between those owner classes.
251// revoke remains lossless for attachment-to-attachment replacement.
252func (p *terminalInputPump) dropOwned(consumer uint64) {
253 p.mu.Lock()
254 if p.consumer != consumer {
255 p.mu.Unlock()
256 return
257 }
258 hadPending := p.pending != nil
259 p.pending = nil
260 p.residual = nil
261 if p.delivering == consumer {
262 p.delivering = 0
263 p.deliveringResidual = false
264 }
265 p.mu.Unlock()
266 if hadPending {
267 select {
268 case p.space <- struct{}{}:
269 default:
270 }
271 }
272}
273
274func (p *terminalInputPump) ack(consumer uint64) {
275 p.mu.Lock()
276 if p.consumer != consumer || p.delivering != consumer {
277 p.mu.Unlock()
278 return
279 }
280 wasResidual := p.deliveringResidual
281 if wasResidual {
282 p.residual = nil
283 } else {
284 p.pending = nil
285 }
286 p.delivering = 0
287 p.deliveringResidual = false
288 more := len(p.residual) != 0 || p.pending != nil
289 p.mu.Unlock()
290 if wasResidual {
291 if more {
292 p.signalReady(consumer)
293 }
294 return
295 }
296 p.space <- struct{}{}
297}
298
299func (p *terminalInputPump) finish() {
300 p.mu.Lock()
301 p.closed = true
302 p.mu.Unlock()
303 p.signalClaimChanged()
304 p.signalState()
305}
306
307func (p *terminalInputPump) signalState() {
308 select {
309 case p.state <- struct{}{}:
310 default:
311 }
312}
313
314func (p *terminalInputPump) waitForActive() (uint64, bool) {
315 for {
316 p.mu.Lock()
317 active, closed, activation := p.active, p.closed, p.activation
318 p.mu.Unlock()
319 if closed {
320 return 0, false
321 }
322 if active {
323 return activation, true
324 }
325 select {
326 case <-p.done:
327 return 0, false
328 case <-p.state:
329 }
330 }
331}
332
333func (p *terminalInputPump) start() {
334 p.startMu.Do(func() {
335 go func() {
336 defer close(p.exited)
337 buf := make([]byte, stdinBufSize)
338 for {
339 activation, ok := p.waitForActive()
340 if !ok {
341 return
342 }
343 n, err := p.in.Read(buf)
344 result := terminalReadResult{err: err}
345 if n > 0 {
346 if data := stripTerminalFocusReports(p.focus, buf[:n]); len(data) != 0 {
347 result.data = append([]byte(nil), data...)
348 } else if err == nil {
349 // A read that only carried focus reports delivers nothing.
350 continue
351 }
352 }
353 if !p.enqueue(result, activation) {
354 if p.isClosed() {
355 return
356 }
357 continue
358 }
359 if err != nil {
360 p.setReadCause(err)
361 p.finish()
362 return
363 }
364 }
365 }()
366 })
367}
368
369// setReadCause records the cause the terminal reader ended with. An orderly EOF
370// is the absence of a failure and stays zero, so cause reports io.EOF for it.
371func (p *terminalInputPump) setReadCause(err error) {
372 if err == nil || errors.Is(err, io.EOF) {
373 return
374 }
375 p.mu.Lock()
376 if p.readCause == nil {
377 p.readCause = err
378 }
379 p.mu.Unlock()
380}
381
382// cause reports why the terminal read ended: the read failure when there was
383// one, otherwise an orderly EOF.
384func (p *terminalInputPump) cause() error {
385 p.mu.Lock()
386 defer p.mu.Unlock()
387 if p.readCause != nil {
388 return p.readCause
389 }
390 return io.EOF
391}
392
393func (p *terminalInputPump) isClosed() bool {
394 p.mu.Lock()
395 defer p.mu.Unlock()
396 return p.closed
397}
398
399// suspend drops input completed after an attach ends and parks the lifecycle
400// reader before its next Read. A later Run resumes this same reader, avoiding
401// both input replay between runs and a competing read on caller-owned stdin.
402func (p *terminalInputPump) suspend() {
403 p.mu.Lock()
404 p.active = false
405 discarded := p.pending != nil
406 p.residual = nil
407 p.pending = nil
408 p.delivering = 0
409 p.deliveringResidual = false
410 p.mu.Unlock()
411 if discarded {
412 select {
413 case p.space <- struct{}{}:
414 default:
415 }
416 }
417 p.signalState()
418 p.signalClaimChanged()
419}
420
421func (p *terminalInputPump) stop() {
422 p.stopMu.Do(func() {
423 // Mark closure while holding the same mutex that publishes a completed
424 // Read. A Read which returns after stop therefore cannot publish even
425 // when both done and space are selectable in enqueue.
426 p.mu.Lock()
427 p.closed = true
428 p.mu.Unlock()
429 close(p.done)
430 p.signalClaimChanged()
431 p.signalState()
432 })
433}