detached 113b229

refactor(client): extract terminal input pump

commit
113b2297b9c30216cb82771fb0fdeff0d579bbd8
author
brice · · 2026-09-19 19:30 UTC
parent
2bcc916
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +457 -449
Split
UnifiedSplit
M internal/usecase/client/client.go +4 -449
@@ -3568,455 +3568,6 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.C
35683568 }
35693569 }
35703570
3571-// terminalReadResult is copied by terminalInputPump before the next Read so
3572-// terminal input remains owned by one lifecycle reader across reconnects.
3573-type terminalInputSource uint8
3574-
3575-const (
3576- terminalInputHuman terminalInputSource = iota
3577- terminalInputReply
3578- terminalInputAutomation
3579-)
3580-
3581-type terminalReadResult struct {
3582- data []byte
3583- err error
3584- source terminalInputSource
3585- generation uint64
3586- actionID uint64
3587- endBatch bool
3588- // keys/text carry the decoded ui-driver op for picker interception.
3589- // Physical input leaves both empty; the picker consumer then decodes the
3590- // raw bytes itself.
3591- keys []string
3592- text string
3593-}
3594-
3595-// terminalAutomationRequest is admitted only by the foreground scanner owner.
3596-// Its reply is buffered so cancellation never holds up physical input.
3597-type terminalAutomationRequest struct {
3598- ctx context.Context
3599- owner *UI
3600- consumer uint64
3601- record terminalReadResult
3602- admitted chan bool
3603- dispatched chan bool
3604-}
3605-
3606-// terminalInputPump is the sole consumer of the caller-owned terminal reader.
3607-// It is started once after raw mode is entered and is reused by each attach
3608-// scanner. A bare io.Reader cannot be interrupted, so a Read may stay blocked
3609-// after Run exits; suspend drops any result received while inactive and never
3610-// closes caller-owned stdin. Reconnects and later runs never add another reader.
3611-type terminalInputPump struct {
3612- in io.Reader
3613- done chan struct{}
3614- automation chan terminalAutomationRequest
3615-
3616- mu sync.Mutex
3617- // residual holds scanner bytes that were read but not delivered when an
3618- // attempt is cancelled. It is consumed before the next terminal Read so a
3619- // replacement scanner neither loses a partial marker nor replays bytes
3620- // whose callbacks already ran.
3621- residual []byte
3622- pending *terminalReadResult
3623- consumer uint64
3624- delivering uint64
3625- deliveringResidual bool
3626- nextID uint64
3627- activation uint64
3628- closed bool
3629- active bool
3630- // readCause stores the cause the terminal reader ended with. The reader
3631- // enqueues its final result for a consumer to report, but the exit watcher
3632- // can win that race, so the cause is recorded here as well: a read failure
3633- // must never be published as an orderly EOF.
3634- readCause error
3635- readyMu sync.Mutex
3636- ready map[uint64]chan struct{}
3637- claimChanged chan struct{}
3638- space chan struct{}
3639- state chan struct{}
3640- exited chan struct{}
3641- afterRevoke func() // test synchronization hook
3642- startMu sync.Once
3643- stopMu sync.Once
3644-}
3645-
3646-func newTerminalInputPump(in io.Reader) *terminalInputPump {
3647- space := make(chan struct{}, 1)
3648- space <- struct{}{}
3649- return &terminalInputPump{
3650- in: in,
3651- done: make(chan struct{}),
3652- automation: make(chan terminalAutomationRequest),
3653- activation: 1,
3654- active: true,
3655- ready: make(map[uint64]chan struct{}),
3656- claimChanged: make(chan struct{}, 1),
3657- space: space,
3658- state: make(chan struct{}, 1),
3659- exited: make(chan struct{}),
3660- }
3661-}
3662-
3663-// claim grants one attach scanner exclusive permission to dequeue terminal
3664-// input. The reader itself remains lifecycle-owned across reconnects.
3665-func (p *terminalInputPump) claim() uint64 {
3666- consumer, ok := p.tryClaim()
3667- if !ok {
3668- panic("terminal input consumer already claimed")
3669- }
3670- return consumer
3671-}
3672-
3673-// tryClaim lets competing foreground owners decline admission without panicking.
3674-func (p *terminalInputPump) tryClaim() (uint64, bool) {
3675- p.mu.Lock()
3676- if p.consumer != 0 {
3677- p.mu.Unlock()
3678- return 0, false
3679- }
3680- p.nextID++
3681- p.consumer = p.nextID
3682- consumer := p.consumer
3683- p.readyMu.Lock()
3684- p.ready[consumer] = make(chan struct{}, 1)
3685- p.readyMu.Unlock()
3686- ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
3687- p.mu.Unlock()
3688- p.signalClaimChanged()
3689- if ready {
3690- p.signalReady(consumer)
3691- }
3692- return consumer, true
3693-}
3694-
3695-// revoke invalidates an attempt before its replacement is allowed to claim
3696-// input. Pending bytes are deliberately retained for that replacement.
3697-func (p *terminalInputPump) revoke(consumer uint64) {
3698- p.mu.Lock()
3699- if p.consumer == consumer {
3700- p.consumer = 0
3701- }
3702- if p.delivering == consumer {
3703- // Legacy attach-attempt replacement remains lossless. Autonomous
3704- // supervisor ownership uses ack-before-picker-handoff and
3705- // drop-before-attachment-handoff, so bytes cannot cross owner kinds.
3706- p.delivering = 0
3707- p.deliveringResidual = false
3708- }
3709- p.mu.Unlock()
3710- p.readyMu.Lock()
3711- delete(p.ready, consumer)
3712- p.readyMu.Unlock()
3713- p.signalClaimChanged()
3714- if p.afterRevoke != nil {
3715- p.afterRevoke()
3716- }
3717-}
3718-
3719-func (p *terminalInputPump) signalClaimChanged() {
3720- select {
3721- case p.claimChanged <- struct{}{}:
3722- default:
3723- }
3724-}
3725-
3726-func (p *terminalInputPump) readyFor(consumer uint64) <-chan struct{} {
3727- p.readyMu.Lock()
3728- defer p.readyMu.Unlock()
3729- return p.ready[consumer]
3730-}
3731-
3732-func (p *terminalInputPump) signalReady(consumer uint64) {
3733- if consumer == 0 {
3734- return
3735- }
3736- p.readyMu.Lock()
3737- ready := p.ready[consumer]
3738- p.readyMu.Unlock()
3739- if ready == nil {
3740- return
3741- }
3742- select {
3743- case ready <- struct{}{}:
3744- default:
3745- }
3746-}
3747-
3748-// enqueue retains at most one completed terminal read. The reader blocks
3749-// before publishing another result, keeping handoff buffering bounded.
3750-func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
3751- select {
3752- case <-p.done:
3753- return false
3754- case <-p.space:
3755- }
3756- p.mu.Lock()
3757- if p.closed || !p.active || p.activation != activation {
3758- p.mu.Unlock()
3759- p.space <- struct{}{}
3760- return false
3761- }
3762- p.pending = &result
3763- consumer := p.consumer
3764- p.mu.Unlock()
3765- p.signalReady(consumer)
3766- return true
3767-}
3768-
3769-// take leases a ready result only to the current, non-cancelled consumer.
3770-// The raw read remains pending until ack, so revocation or cancellation before
3771-// scanner delivery leaves the exact bytes available to the next attempt.
3772-func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
3773- p.mu.Lock()
3774- if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
3775- p.mu.Unlock()
3776- return terminalReadResult{}, false
3777- }
3778- if len(p.residual) != 0 {
3779- result := terminalReadResult{data: append([]byte(nil), p.residual...)}
3780- p.delivering = consumer
3781- p.deliveringResidual = true
3782- p.mu.Unlock()
3783- return result, true
3784- }
3785- if p.pending == nil {
3786- p.mu.Unlock()
3787- return terminalReadResult{}, false
3788- }
3789- result := *p.pending
3790- p.delivering = consumer
3791- p.deliveringResidual = false
3792- p.mu.Unlock()
3793- return result, true
3794-}
3795-
3796-// preserveResidual records the undecided marker prefix after callbacks for the
3797-// rest of its read have completed. It is intentionally bounded by the marker
3798-// scanner's fixed response prefixes and is replayed before later terminal
3799-// reads on a replacement attempt.
3800-func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
3801- if len(data) == 0 {
3802- return
3803- }
3804- p.mu.Lock()
3805- defer p.mu.Unlock()
3806- if p.consumer != consumer {
3807- return
3808- }
3809- p.residual = append(p.residual[:0], data...)
3810-}
3811-
3812-// ack commits scanner delivery of a leased read and lets the lifecycle reader
3813-// accept the next result. Only the consumer that holds the lease can ack it.
3814-// dropOwned disposes every undecided byte at an autonomous owner-class
3815-// boundary. Picker-queued bytes are dropped before attachment claim, and
3816-// attachment-queued bytes are dropped before picker claim. No delivery,
3817-// pending read, or preserved residual may cross between those owner classes.
3818-// revoke remains lossless for legacy attachment-to-attachment replacement.
3819-func (p *terminalInputPump) dropOwned(consumer uint64) {
3820- p.mu.Lock()
3821- if p.consumer != consumer {
3822- p.mu.Unlock()
3823- return
3824- }
3825- hadPending := p.pending != nil
3826- p.pending = nil
3827- p.residual = nil
3828- if p.delivering == consumer {
3829- p.delivering = 0
3830- p.deliveringResidual = false
3831- }
3832- p.mu.Unlock()
3833- if hadPending {
3834- select {
3835- case p.space <- struct{}{}:
3836- default:
3837- }
3838- }
3839-}
3840-
3841-func (p *terminalInputPump) ack(consumer uint64) {
3842- p.mu.Lock()
3843- if p.consumer != consumer || p.delivering != consumer {
3844- p.mu.Unlock()
3845- return
3846- }
3847- wasResidual := p.deliveringResidual
3848- if wasResidual {
3849- p.residual = nil
3850- } else {
3851- p.pending = nil
3852- }
3853- p.delivering = 0
3854- p.deliveringResidual = false
3855- more := len(p.residual) != 0 || p.pending != nil
3856- p.mu.Unlock()
3857- if wasResidual {
3858- if more {
3859- p.signalReady(consumer)
3860- }
3861- return
3862- }
3863- p.space <- struct{}{}
3864-}
3865-
3866-func (p *terminalInputPump) finish() {
3867- p.mu.Lock()
3868- p.closed = true
3869- p.mu.Unlock()
3870- p.signalClaimChanged()
3871- p.signalState()
3872-}
3873-
3874-func (p *terminalInputPump) signalState() {
3875- select {
3876- case p.state <- struct{}{}:
3877- default:
3878- }
3879-}
3880-
3881-func (p *terminalInputPump) waitForActive() (uint64, bool) {
3882- for {
3883- p.mu.Lock()
3884- active, closed, activation := p.active, p.closed, p.activation
3885- p.mu.Unlock()
3886- if closed {
3887- return 0, false
3888- }
3889- if active {
3890- return activation, true
3891- }
3892- select {
3893- case <-p.done:
3894- return 0, false
3895- case <-p.state:
3896- }
3897- }
3898-}
3899-
3900-func (p *terminalInputPump) start() {
3901- p.startMu.Do(func() {
3902- go func() {
3903- defer close(p.exited)
3904- buf := make([]byte, stdinBufSize)
3905- for {
3906- activation, ok := p.waitForActive()
3907- if !ok {
3908- return
3909- }
3910- n, err := p.in.Read(buf)
3911- result := terminalReadResult{err: err}
3912- if n > 0 {
3913- result.data = append([]byte(nil), buf[:n]...)
3914- }
3915- if !p.enqueue(result, activation) {
3916- if p.isClosed() {
3917- return
3918- }
3919- continue
3920- }
3921- if err != nil {
3922- p.setReadCause(err)
3923- p.finish()
3924- return
3925- }
3926- }
3927- }()
3928- })
3929-}
3930-
3931-func (p *terminalInputPump) hasExited() bool {
3932- select {
3933- case <-p.exited:
3934- return true
3935- default:
3936- return false
3937- }
3938-}
3939-
3940-// setReadCause records the cause the terminal reader ended with. An orderly EOF
3941-// is the absence of a failure and stays zero, so cause reports io.EOF for it.
3942-func (p *terminalInputPump) setReadCause(err error) {
3943- if err == nil || errors.Is(err, io.EOF) {
3944- return
3945- }
3946- p.mu.Lock()
3947- if p.readCause == nil {
3948- p.readCause = err
3949- }
3950- p.mu.Unlock()
3951-}
3952-
3953-// cause reports why the terminal read ended: the read failure when there was
3954-// one, otherwise an orderly EOF.
3955-func (p *terminalInputPump) cause() error {
3956- p.mu.Lock()
3957- defer p.mu.Unlock()
3958- if p.readCause != nil {
3959- return p.readCause
3960- }
3961- return io.EOF
3962-}
3963-
3964-func (p *terminalInputPump) isClosed() bool {
3965- p.mu.Lock()
3966- defer p.mu.Unlock()
3967- return p.closed
3968-}
3969-
3970-// suspend drops input completed after an attach ends and parks the lifecycle
3971-// reader before its next Read. A later Run resumes this same reader, avoiding
3972-// both input replay between runs and a competing read on caller-owned stdin.
3973-func (p *terminalInputPump) suspend() {
3974- p.mu.Lock()
3975- p.active = false
3976- discarded := p.pending != nil
3977- p.residual = nil
3978- p.pending = nil
3979- p.delivering = 0
3980- p.deliveringResidual = false
3981- p.mu.Unlock()
3982- if discarded {
3983- select {
3984- case p.space <- struct{}{}:
3985- default:
3986- }
3987- }
3988- p.signalState()
3989- p.signalClaimChanged()
3990-}
3991-
3992-func (p *terminalInputPump) resume() {
3993- p.mu.Lock()
3994- if !p.closed && !p.active {
3995- p.activation++
3996- p.active = true
3997- }
3998- p.mu.Unlock()
3999- p.signalState()
4000-}
4001-
4002-func (p *terminalInputPump) stop() {
4003- p.stopMu.Do(func() {
4004- // Mark closure while holding the same mutex that publishes a completed
4005- // Read. A Read which returns after stop therefore cannot publish even
4006- // when both done and space are selectable in enqueue.
4007- p.mu.Lock()
4008- p.closed = true
4009- p.mu.Unlock()
4010- close(p.done)
4011- p.signalClaimChanged()
4012- p.signalState()
4013- })
4014-}
4015-
4016-// stdinPump parses terminal color reports into generation-tagged events while
4017-// preserving all ordinary terminal input byte-for-byte. It never reads the
4018-// terminal directly, publishes a Theme, or writes terminal output;
4019-// attachAttempt owns those operations.
40203571 type foregroundSendLease struct {
40213572 mu sync.Mutex
40223573 active bool
@@ -4047,6 +3598,10 @@ func (l *foregroundSendLease) stop() {
40473598 l.mu.Unlock()
40483599 }
40493600
3601+// stdinPump parses terminal color reports into generation-tagged events while
3602+// preserving all ordinary terminal input byte-for-byte. It never reads the
3603+// terminal directly, publishes a Theme, or writes terminal output;
3604+// attachAttempt owns those operations.
40503605 type stdinPump struct {
40513606 ctx context.Context
40523607 cancel context.CancelFunc
@@ -3568,455 +3568,6 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.C
3568 }3568 }
3569 }3569 }
3570 3570
3571-// terminalReadResult is copied by terminalInputPump before the next Read so
3572-// terminal input remains owned by one lifecycle reader across reconnects.
3573-type terminalInputSource uint8
3574-
3575-const (
3576- terminalInputHuman terminalInputSource = iota
3577- terminalInputReply
3578- terminalInputAutomation
3579-)
3580-
3581-type terminalReadResult struct {
3582- data []byte
3583- err error
3584- source terminalInputSource
3585- generation uint64
3586- actionID uint64
3587- endBatch bool
3588- // keys/text carry the decoded ui-driver op for picker interception.
3589- // Physical input leaves both empty; the picker consumer then decodes the
3590- // raw bytes itself.
3591- keys []string
3592- text string
3593-}
3594-
3595-// terminalAutomationRequest is admitted only by the foreground scanner owner.
3596-// Its reply is buffered so cancellation never holds up physical input.
3597-type terminalAutomationRequest struct {
3598- ctx context.Context
3599- owner *UI
3600- consumer uint64
3601- record terminalReadResult
3602- admitted chan bool
3603- dispatched chan bool
3604-}
3605-
3606-// terminalInputPump is the sole consumer of the caller-owned terminal reader.
3607-// It is started once after raw mode is entered and is reused by each attach
3608-// scanner. A bare io.Reader cannot be interrupted, so a Read may stay blocked
3609-// after Run exits; suspend drops any result received while inactive and never
3610-// closes caller-owned stdin. Reconnects and later runs never add another reader.
3611-type terminalInputPump struct {
3612- in io.Reader
3613- done chan struct{}
3614- automation chan terminalAutomationRequest
3615-
3616- mu sync.Mutex
3617- // residual holds scanner bytes that were read but not delivered when an
3618- // attempt is cancelled. It is consumed before the next terminal Read so a
3619- // replacement scanner neither loses a partial marker nor replays bytes
3620- // whose callbacks already ran.
3621- residual []byte
3622- pending *terminalReadResult
3623- consumer uint64
3624- delivering uint64
3625- deliveringResidual bool
3626- nextID uint64
3627- activation uint64
3628- closed bool
3629- active bool
3630- // readCause stores the cause the terminal reader ended with. The reader
3631- // enqueues its final result for a consumer to report, but the exit watcher
3632- // can win that race, so the cause is recorded here as well: a read failure
3633- // must never be published as an orderly EOF.
3634- readCause error
3635- readyMu sync.Mutex
3636- ready map[uint64]chan struct{}
3637- claimChanged chan struct{}
3638- space chan struct{}
3639- state chan struct{}
3640- exited chan struct{}
3641- afterRevoke func() // test synchronization hook
3642- startMu sync.Once
3643- stopMu sync.Once
3644-}
3645-
3646-func newTerminalInputPump(in io.Reader) *terminalInputPump {
3647- space := make(chan struct{}, 1)
3648- space <- struct{}{}
3649- return &terminalInputPump{
3650- in: in,
3651- done: make(chan struct{}),
3652- automation: make(chan terminalAutomationRequest),
3653- activation: 1,
3654- active: true,
3655- ready: make(map[uint64]chan struct{}),
3656- claimChanged: make(chan struct{}, 1),
3657- space: space,
3658- state: make(chan struct{}, 1),
3659- exited: make(chan struct{}),
3660- }
3661-}
3662-
3663-// claim grants one attach scanner exclusive permission to dequeue terminal
3664-// input. The reader itself remains lifecycle-owned across reconnects.
3665-func (p *terminalInputPump) claim() uint64 {
3666- consumer, ok := p.tryClaim()
3667- if !ok {
3668- panic("terminal input consumer already claimed")
3669- }
3670- return consumer
3671-}
3672-
3673-// tryClaim lets competing foreground owners decline admission without panicking.
3674-func (p *terminalInputPump) tryClaim() (uint64, bool) {
3675- p.mu.Lock()
3676- if p.consumer != 0 {
3677- p.mu.Unlock()
3678- return 0, false
3679- }
3680- p.nextID++
3681- p.consumer = p.nextID
3682- consumer := p.consumer
3683- p.readyMu.Lock()
3684- p.ready[consumer] = make(chan struct{}, 1)
3685- p.readyMu.Unlock()
3686- ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
3687- p.mu.Unlock()
3688- p.signalClaimChanged()
3689- if ready {
3690- p.signalReady(consumer)
3691- }
3692- return consumer, true
3693-}
3694-
3695-// revoke invalidates an attempt before its replacement is allowed to claim
3696-// input. Pending bytes are deliberately retained for that replacement.
3697-func (p *terminalInputPump) revoke(consumer uint64) {
3698- p.mu.Lock()
3699- if p.consumer == consumer {
3700- p.consumer = 0
3701- }
3702- if p.delivering == consumer {
3703- // Legacy attach-attempt replacement remains lossless. Autonomous
3704- // supervisor ownership uses ack-before-picker-handoff and
3705- // drop-before-attachment-handoff, so bytes cannot cross owner kinds.
3706- p.delivering = 0
3707- p.deliveringResidual = false
3708- }
3709- p.mu.Unlock()
3710- p.readyMu.Lock()
3711- delete(p.ready, consumer)
3712- p.readyMu.Unlock()
3713- p.signalClaimChanged()
3714- if p.afterRevoke != nil {
3715- p.afterRevoke()
3716- }
3717-}
3718-
3719-func (p *terminalInputPump) signalClaimChanged() {
3720- select {
3721- case p.claimChanged <- struct{}{}:
3722- default:
3723- }
3724-}
3725-
3726-func (p *terminalInputPump) readyFor(consumer uint64) <-chan struct{} {
3727- p.readyMu.Lock()
3728- defer p.readyMu.Unlock()
3729- return p.ready[consumer]
3730-}
3731-
3732-func (p *terminalInputPump) signalReady(consumer uint64) {
3733- if consumer == 0 {
3734- return
3735- }
3736- p.readyMu.Lock()
3737- ready := p.ready[consumer]
3738- p.readyMu.Unlock()
3739- if ready == nil {
3740- return
3741- }
3742- select {
3743- case ready <- struct{}{}:
3744- default:
3745- }
3746-}
3747-
3748-// enqueue retains at most one completed terminal read. The reader blocks
3749-// before publishing another result, keeping handoff buffering bounded.
3750-func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
3751- select {
3752- case <-p.done:
3753- return false
3754- case <-p.space:
3755- }
3756- p.mu.Lock()
3757- if p.closed || !p.active || p.activation != activation {
3758- p.mu.Unlock()
3759- p.space <- struct{}{}
3760- return false
3761- }
3762- p.pending = &result
3763- consumer := p.consumer
3764- p.mu.Unlock()
3765- p.signalReady(consumer)
3766- return true
3767-}
3768-
3769-// take leases a ready result only to the current, non-cancelled consumer.
3770-// The raw read remains pending until ack, so revocation or cancellation before
3771-// scanner delivery leaves the exact bytes available to the next attempt.
3772-func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
3773- p.mu.Lock()
3774- if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
3775- p.mu.Unlock()
3776- return terminalReadResult{}, false
3777- }
3778- if len(p.residual) != 0 {
3779- result := terminalReadResult{data: append([]byte(nil), p.residual...)}
3780- p.delivering = consumer
3781- p.deliveringResidual = true
3782- p.mu.Unlock()
3783- return result, true
3784- }
3785- if p.pending == nil {
3786- p.mu.Unlock()
3787- return terminalReadResult{}, false
3788- }
3789- result := *p.pending
3790- p.delivering = consumer
3791- p.deliveringResidual = false
3792- p.mu.Unlock()
3793- return result, true
3794-}
3795-
3796-// preserveResidual records the undecided marker prefix after callbacks for the
3797-// rest of its read have completed. It is intentionally bounded by the marker
3798-// scanner's fixed response prefixes and is replayed before later terminal
3799-// reads on a replacement attempt.
3800-func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
3801- if len(data) == 0 {
3802- return
3803- }
3804- p.mu.Lock()
3805- defer p.mu.Unlock()
3806- if p.consumer != consumer {
3807- return
3808- }
3809- p.residual = append(p.residual[:0], data...)
3810-}
3811-
3812-// ack commits scanner delivery of a leased read and lets the lifecycle reader
3813-// accept the next result. Only the consumer that holds the lease can ack it.
3814-// dropOwned disposes every undecided byte at an autonomous owner-class
3815-// boundary. Picker-queued bytes are dropped before attachment claim, and
3816-// attachment-queued bytes are dropped before picker claim. No delivery,
3817-// pending read, or preserved residual may cross between those owner classes.
3818-// revoke remains lossless for legacy attachment-to-attachment replacement.
3819-func (p *terminalInputPump) dropOwned(consumer uint64) {
3820- p.mu.Lock()
3821- if p.consumer != consumer {
3822- p.mu.Unlock()
3823- return
3824- }
3825- hadPending := p.pending != nil
3826- p.pending = nil
3827- p.residual = nil
3828- if p.delivering == consumer {
3829- p.delivering = 0
3830- p.deliveringResidual = false
3831- }
3832- p.mu.Unlock()
3833- if hadPending {
3834- select {
3835- case p.space <- struct{}{}:
3836- default:
3837- }
3838- }
3839-}
3840-
3841-func (p *terminalInputPump) ack(consumer uint64) {
3842- p.mu.Lock()
3843- if p.consumer != consumer || p.delivering != consumer {
3844- p.mu.Unlock()
3845- return
3846- }
3847- wasResidual := p.deliveringResidual
3848- if wasResidual {
3849- p.residual = nil
3850- } else {
3851- p.pending = nil
3852- }
3853- p.delivering = 0
3854- p.deliveringResidual = false
3855- more := len(p.residual) != 0 || p.pending != nil
3856- p.mu.Unlock()
3857- if wasResidual {
3858- if more {
3859- p.signalReady(consumer)
3860- }
3861- return
3862- }
3863- p.space <- struct{}{}
3864-}
3865-
3866-func (p *terminalInputPump) finish() {
3867- p.mu.Lock()
3868- p.closed = true
3869- p.mu.Unlock()
3870- p.signalClaimChanged()
3871- p.signalState()
3872-}
3873-
3874-func (p *terminalInputPump) signalState() {
3875- select {
3876- case p.state <- struct{}{}:
3877- default:
3878- }
3879-}
3880-
3881-func (p *terminalInputPump) waitForActive() (uint64, bool) {
3882- for {
3883- p.mu.Lock()
3884- active, closed, activation := p.active, p.closed, p.activation
3885- p.mu.Unlock()
3886- if closed {
3887- return 0, false
3888- }
3889- if active {
3890- return activation, true
3891- }
3892- select {
3893- case <-p.done:
3894- return 0, false
3895- case <-p.state:
3896- }
3897- }
3898-}
3899-
3900-func (p *terminalInputPump) start() {
3901- p.startMu.Do(func() {
3902- go func() {
3903- defer close(p.exited)
3904- buf := make([]byte, stdinBufSize)
3905- for {
3906- activation, ok := p.waitForActive()
3907- if !ok {
3908- return
3909- }
3910- n, err := p.in.Read(buf)
3911- result := terminalReadResult{err: err}
3912- if n > 0 {
3913- result.data = append([]byte(nil), buf[:n]...)
3914- }
3915- if !p.enqueue(result, activation) {
3916- if p.isClosed() {
3917- return
3918- }
3919- continue
3920- }
3921- if err != nil {
3922- p.setReadCause(err)
3923- p.finish()
3924- return
3925- }
3926- }
3927- }()
3928- })
3929-}
3930-
3931-func (p *terminalInputPump) hasExited() bool {
3932- select {
3933- case <-p.exited:
3934- return true
3935- default:
3936- return false
3937- }
3938-}
3939-
3940-// setReadCause records the cause the terminal reader ended with. An orderly EOF
3941-// is the absence of a failure and stays zero, so cause reports io.EOF for it.
3942-func (p *terminalInputPump) setReadCause(err error) {
3943- if err == nil || errors.Is(err, io.EOF) {
3944- return
3945- }
3946- p.mu.Lock()
3947- if p.readCause == nil {
3948- p.readCause = err
3949- }
3950- p.mu.Unlock()
3951-}
3952-
3953-// cause reports why the terminal read ended: the read failure when there was
3954-// one, otherwise an orderly EOF.
3955-func (p *terminalInputPump) cause() error {
3956- p.mu.Lock()
3957- defer p.mu.Unlock()
3958- if p.readCause != nil {
3959- return p.readCause
3960- }
3961- return io.EOF
3962-}
3963-
3964-func (p *terminalInputPump) isClosed() bool {
3965- p.mu.Lock()
3966- defer p.mu.Unlock()
3967- return p.closed
3968-}
3969-
3970-// suspend drops input completed after an attach ends and parks the lifecycle
3971-// reader before its next Read. A later Run resumes this same reader, avoiding
3972-// both input replay between runs and a competing read on caller-owned stdin.
3973-func (p *terminalInputPump) suspend() {
3974- p.mu.Lock()
3975- p.active = false
3976- discarded := p.pending != nil
3977- p.residual = nil
3978- p.pending = nil
3979- p.delivering = 0
3980- p.deliveringResidual = false
3981- p.mu.Unlock()
3982- if discarded {
3983- select {
3984- case p.space <- struct{}{}:
3985- default:
3986- }
3987- }
3988- p.signalState()
3989- p.signalClaimChanged()
3990-}
3991-
3992-func (p *terminalInputPump) resume() {
3993- p.mu.Lock()
3994- if !p.closed && !p.active {
3995- p.activation++
3996- p.active = true
3997- }
3998- p.mu.Unlock()
3999- p.signalState()
4000-}
4001-
4002-func (p *terminalInputPump) stop() {
4003- p.stopMu.Do(func() {
4004- // Mark closure while holding the same mutex that publishes a completed
4005- // Read. A Read which returns after stop therefore cannot publish even
4006- // when both done and space are selectable in enqueue.
4007- p.mu.Lock()
4008- p.closed = true
4009- p.mu.Unlock()
4010- close(p.done)
4011- p.signalClaimChanged()
4012- p.signalState()
4013- })
4014-}
4015-
4016-// stdinPump parses terminal color reports into generation-tagged events while
4017-// preserving all ordinary terminal input byte-for-byte. It never reads the
4018-// terminal directly, publishes a Theme, or writes terminal output;
4019-// attachAttempt owns those operations.
4020 type foregroundSendLease struct {3571 type foregroundSendLease struct {
4021 mu sync.Mutex3572 mu sync.Mutex
4022 active bool3573 active bool
@@ -4047,6 +3598,10 @@ func (l *foregroundSendLease) stop() {
4047 l.mu.Unlock()3598 l.mu.Unlock()
4048 }3599 }
4049 3600
3601+// stdinPump parses terminal color reports into generation-tagged events while
3602+// preserving all ordinary terminal input byte-for-byte. It never reads the
3603+// terminal directly, publishes a Theme, or writes terminal output;
3604+// attachAttempt owns those operations.
4050 type stdinPump struct {3605 type stdinPump struct {
4051 ctx context.Context3606 ctx context.Context
4052 cancel context.CancelFunc3607 cancel context.CancelFunc
A internal/usecase/client/terminal_input_pump.go +453 -0
@@ -0,0 +1,453 @@
1+package client
2+
3+import (
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.
12+type terminalInputSource uint8
13+
14+const (
15+ terminalInputHuman terminalInputSource = iota
16+ terminalInputReply
17+ terminalInputAutomation
18+)
19+
20+type 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.
36+type 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.
50+type terminalInputPump struct {
51+ in io.Reader
52+ done chan struct{}
53+ automation chan terminalAutomationRequest
54+
55+ mu sync.Mutex
56+ // residual holds scanner bytes that were read but not delivered when an
57+ // attempt is cancelled. It is consumed before the next terminal Read so a
58+ // replacement scanner neither loses a partial marker nor replays bytes
59+ // whose callbacks already ran.
60+ residual []byte
61+ pending *terminalReadResult
62+ consumer uint64
63+ delivering uint64
64+ deliveringResidual bool
65+ nextID uint64
66+ activation uint64
67+ closed bool
68+ active bool
69+ // readCause stores the cause the terminal reader ended with. The reader
70+ // enqueues its final result for a consumer to report, but the exit watcher
71+ // can win that race, so the cause is recorded here as well: a read failure
72+ // must never be published as an orderly EOF.
73+ readCause error
74+ readyMu sync.Mutex
75+ ready map[uint64]chan struct{}
76+ claimChanged chan struct{}
77+ space chan struct{}
78+ state chan struct{}
79+ exited chan struct{}
80+ afterRevoke func() // test synchronization hook
81+ startMu sync.Once
82+ stopMu sync.Once
83+}
84+
85+func newTerminalInputPump(in io.Reader) *terminalInputPump {
86+ space := make(chan struct{}, 1)
87+ space <- struct{}{}
88+ return &terminalInputPump{
89+ in: in,
90+ done: make(chan struct{}),
91+ automation: make(chan terminalAutomationRequest),
92+ activation: 1,
93+ active: true,
94+ ready: make(map[uint64]chan struct{}),
95+ claimChanged: make(chan struct{}, 1),
96+ space: space,
97+ state: make(chan struct{}, 1),
98+ exited: make(chan struct{}),
99+ }
100+}
101+
102+// claim grants one attach scanner exclusive permission to dequeue terminal
103+// input. The reader itself remains lifecycle-owned across reconnects.
104+func (p *terminalInputPump) claim() uint64 {
105+ consumer, ok := p.tryClaim()
106+ if !ok {
107+ panic("terminal input consumer already claimed")
108+ }
109+ return consumer
110+}
111+
112+// tryClaim lets competing foreground owners decline admission without panicking.
113+func (p *terminalInputPump) tryClaim() (uint64, bool) {
114+ p.mu.Lock()
115+ if p.consumer != 0 {
116+ p.mu.Unlock()
117+ return 0, false
118+ }
119+ p.nextID++
120+ p.consumer = p.nextID
121+ consumer := p.consumer
122+ p.readyMu.Lock()
123+ p.ready[consumer] = make(chan struct{}, 1)
124+ p.readyMu.Unlock()
125+ ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
126+ p.mu.Unlock()
127+ p.signalClaimChanged()
128+ if ready {
129+ p.signalReady(consumer)
130+ }
131+ return consumer, true
132+}
133+
134+// revoke invalidates an attempt before its replacement is allowed to claim
135+// input. Pending bytes are deliberately retained for that replacement.
136+func (p *terminalInputPump) revoke(consumer uint64) {
137+ p.mu.Lock()
138+ if p.consumer == consumer {
139+ p.consumer = 0
140+ }
141+ if p.delivering == consumer {
142+ // Legacy attach-attempt replacement remains lossless. Autonomous
143+ // supervisor ownership uses ack-before-picker-handoff and
144+ // drop-before-attachment-handoff, so bytes cannot cross owner kinds.
145+ p.delivering = 0
146+ p.deliveringResidual = false
147+ }
148+ p.mu.Unlock()
149+ p.readyMu.Lock()
150+ delete(p.ready, consumer)
151+ p.readyMu.Unlock()
152+ p.signalClaimChanged()
153+ if p.afterRevoke != nil {
154+ p.afterRevoke()
155+ }
156+}
157+
158+func (p *terminalInputPump) signalClaimChanged() {
159+ select {
160+ case p.claimChanged <- struct{}{}:
161+ default:
162+ }
163+}
164+
165+func (p *terminalInputPump) readyFor(consumer uint64) <-chan struct{} {
166+ p.readyMu.Lock()
167+ defer p.readyMu.Unlock()
168+ return p.ready[consumer]
169+}
170+
171+func (p *terminalInputPump) signalReady(consumer uint64) {
172+ if consumer == 0 {
173+ return
174+ }
175+ p.readyMu.Lock()
176+ ready := p.ready[consumer]
177+ p.readyMu.Unlock()
178+ if ready == nil {
179+ return
180+ }
181+ select {
182+ case ready <- struct{}{}:
183+ default:
184+ }
185+}
186+
187+// enqueue retains at most one completed terminal read. The reader blocks
188+// before publishing another result, keeping handoff buffering bounded.
189+func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
190+ select {
191+ case <-p.done:
192+ return false
193+ case <-p.space:
194+ }
195+ p.mu.Lock()
196+ if p.closed || !p.active || p.activation != activation {
197+ p.mu.Unlock()
198+ p.space <- struct{}{}
199+ return false
200+ }
201+ p.pending = &result
202+ consumer := p.consumer
203+ p.mu.Unlock()
204+ p.signalReady(consumer)
205+ return true
206+}
207+
208+// take leases a ready result only to the current, non-cancelled consumer.
209+// The raw read remains pending until ack, so revocation or cancellation before
210+// scanner delivery leaves the exact bytes available to the next attempt.
211+func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
212+ p.mu.Lock()
213+ if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
214+ p.mu.Unlock()
215+ return terminalReadResult{}, false
216+ }
217+ if len(p.residual) != 0 {
218+ result := terminalReadResult{data: append([]byte(nil), p.residual...)}
219+ p.delivering = consumer
220+ p.deliveringResidual = true
221+ p.mu.Unlock()
222+ return result, true
223+ }
224+ if p.pending == nil {
225+ p.mu.Unlock()
226+ return terminalReadResult{}, false
227+ }
228+ result := *p.pending
229+ p.delivering = consumer
230+ p.deliveringResidual = false
231+ p.mu.Unlock()
232+ return result, true
233+}
234+
235+// preserveResidual records the undecided marker prefix after callbacks for the
236+// rest of its read have completed. It is intentionally bounded by the marker
237+// scanner's fixed response prefixes and is replayed before later terminal
238+// reads on a replacement attempt.
239+func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
240+ if len(data) == 0 {
241+ return
242+ }
243+ p.mu.Lock()
244+ defer p.mu.Unlock()
245+ if p.consumer != consumer {
246+ return
247+ }
248+ p.residual = append(p.residual[:0], data...)
249+}
250+
251+// ack commits scanner delivery of a leased read and lets the lifecycle reader
252+// accept the next result. Only the consumer that holds the lease can ack it.
253+// dropOwned disposes every undecided byte at an autonomous owner-class
254+// boundary. Picker-queued bytes are dropped before attachment claim, and
255+// attachment-queued bytes are dropped before picker claim. No delivery,
256+// pending read, or preserved residual may cross between those owner classes.
257+// revoke remains lossless for legacy attachment-to-attachment replacement.
258+func (p *terminalInputPump) dropOwned(consumer uint64) {
259+ p.mu.Lock()
260+ if p.consumer != consumer {
261+ p.mu.Unlock()
262+ return
263+ }
264+ hadPending := p.pending != nil
265+ p.pending = nil
266+ p.residual = nil
267+ if p.delivering == consumer {
268+ p.delivering = 0
269+ p.deliveringResidual = false
270+ }
271+ p.mu.Unlock()
272+ if hadPending {
273+ select {
274+ case p.space <- struct{}{}:
275+ default:
276+ }
277+ }
278+}
279+
280+func (p *terminalInputPump) ack(consumer uint64) {
281+ p.mu.Lock()
282+ if p.consumer != consumer || p.delivering != consumer {
283+ p.mu.Unlock()
284+ return
285+ }
286+ wasResidual := p.deliveringResidual
287+ if wasResidual {
288+ p.residual = nil
289+ } else {
290+ p.pending = nil
291+ }
292+ p.delivering = 0
293+ p.deliveringResidual = false
294+ more := len(p.residual) != 0 || p.pending != nil
295+ p.mu.Unlock()
296+ if wasResidual {
297+ if more {
298+ p.signalReady(consumer)
299+ }
300+ return
301+ }
302+ p.space <- struct{}{}
303+}
304+
305+func (p *terminalInputPump) finish() {
306+ p.mu.Lock()
307+ p.closed = true
308+ p.mu.Unlock()
309+ p.signalClaimChanged()
310+ p.signalState()
311+}
312+
313+func (p *terminalInputPump) signalState() {
314+ select {
315+ case p.state <- struct{}{}:
316+ default:
317+ }
318+}
319+
320+func (p *terminalInputPump) waitForActive() (uint64, bool) {
321+ for {
322+ p.mu.Lock()
323+ active, closed, activation := p.active, p.closed, p.activation
324+ p.mu.Unlock()
325+ if closed {
326+ return 0, false
327+ }
328+ if active {
329+ return activation, true
330+ }
331+ select {
332+ case <-p.done:
333+ return 0, false
334+ case <-p.state:
335+ }
336+ }
337+}
338+
339+func (p *terminalInputPump) start() {
340+ p.startMu.Do(func() {
341+ go func() {
342+ defer close(p.exited)
343+ buf := make([]byte, stdinBufSize)
344+ for {
345+ activation, ok := p.waitForActive()
346+ if !ok {
347+ return
348+ }
349+ n, err := p.in.Read(buf)
350+ result := terminalReadResult{err: err}
351+ if n > 0 {
352+ result.data = append([]byte(nil), buf[:n]...)
353+ }
354+ if !p.enqueue(result, activation) {
355+ if p.isClosed() {
356+ return
357+ }
358+ continue
359+ }
360+ if err != nil {
361+ p.setReadCause(err)
362+ p.finish()
363+ return
364+ }
365+ }
366+ }()
367+ })
368+}
369+
370+func (p *terminalInputPump) hasExited() bool {
371+ select {
372+ case <-p.exited:
373+ return true
374+ default:
375+ return false
376+ }
377+}
378+
379+// setReadCause records the cause the terminal reader ended with. An orderly EOF
380+// is the absence of a failure and stays zero, so cause reports io.EOF for it.
381+func (p *terminalInputPump) setReadCause(err error) {
382+ if err == nil || errors.Is(err, io.EOF) {
383+ return
384+ }
385+ p.mu.Lock()
386+ if p.readCause == nil {
387+ p.readCause = err
388+ }
389+ p.mu.Unlock()
390+}
391+
392+// cause reports why the terminal read ended: the read failure when there was
393+// one, otherwise an orderly EOF.
394+func (p *terminalInputPump) cause() error {
395+ p.mu.Lock()
396+ defer p.mu.Unlock()
397+ if p.readCause != nil {
398+ return p.readCause
399+ }
400+ return io.EOF
401+}
402+
403+func (p *terminalInputPump) isClosed() bool {
404+ p.mu.Lock()
405+ defer p.mu.Unlock()
406+ return p.closed
407+}
408+
409+// suspend drops input completed after an attach ends and parks the lifecycle
410+// reader before its next Read. A later Run resumes this same reader, avoiding
411+// both input replay between runs and a competing read on caller-owned stdin.
412+func (p *terminalInputPump) suspend() {
413+ p.mu.Lock()
414+ p.active = false
415+ discarded := p.pending != nil
416+ p.residual = nil
417+ p.pending = nil
418+ p.delivering = 0
419+ p.deliveringResidual = false
420+ p.mu.Unlock()
421+ if discarded {
422+ select {
423+ case p.space <- struct{}{}:
424+ default:
425+ }
426+ }
427+ p.signalState()
428+ p.signalClaimChanged()
429+}
430+
431+func (p *terminalInputPump) resume() {
432+ p.mu.Lock()
433+ if !p.closed && !p.active {
434+ p.activation++
435+ p.active = true
436+ }
437+ p.mu.Unlock()
438+ p.signalState()
439+}
440+
441+func (p *terminalInputPump) stop() {
442+ p.stopMu.Do(func() {
443+ // Mark closure while holding the same mutex that publishes a completed
444+ // Read. A Read which returns after stop therefore cannot publish even
445+ // when both done and space are selectable in enqueue.
446+ p.mu.Lock()
447+ p.closed = true
448+ p.mu.Unlock()
449+ close(p.done)
450+ p.signalClaimChanged()
451+ p.signalState()
452+ })
453+}
@@ -0,0 +1,453 @@
1+package client
2+
3+import (
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.
12+type terminalInputSource uint8
13+
14+const (
15+ terminalInputHuman terminalInputSource = iota
16+ terminalInputReply
17+ terminalInputAutomation
18+)
19+
20+type 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.
36+type 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.
50+type terminalInputPump struct {
51+ in io.Reader
52+ done chan struct{}
53+ automation chan terminalAutomationRequest
54+
55+ mu sync.Mutex
56+ // residual holds scanner bytes that were read but not delivered when an
57+ // attempt is cancelled. It is consumed before the next terminal Read so a
58+ // replacement scanner neither loses a partial marker nor replays bytes
59+ // whose callbacks already ran.
60+ residual []byte
61+ pending *terminalReadResult
62+ consumer uint64
63+ delivering uint64
64+ deliveringResidual bool
65+ nextID uint64
66+ activation uint64
67+ closed bool
68+ active bool
69+ // readCause stores the cause the terminal reader ended with. The reader
70+ // enqueues its final result for a consumer to report, but the exit watcher
71+ // can win that race, so the cause is recorded here as well: a read failure
72+ // must never be published as an orderly EOF.
73+ readCause error
74+ readyMu sync.Mutex
75+ ready map[uint64]chan struct{}
76+ claimChanged chan struct{}
77+ space chan struct{}
78+ state chan struct{}
79+ exited chan struct{}
80+ afterRevoke func() // test synchronization hook
81+ startMu sync.Once
82+ stopMu sync.Once
83+}
84+
85+func newTerminalInputPump(in io.Reader) *terminalInputPump {
86+ space := make(chan struct{}, 1)
87+ space <- struct{}{}
88+ return &terminalInputPump{
89+ in: in,
90+ done: make(chan struct{}),
91+ automation: make(chan terminalAutomationRequest),
92+ activation: 1,
93+ active: true,
94+ ready: make(map[uint64]chan struct{}),
95+ claimChanged: make(chan struct{}, 1),
96+ space: space,
97+ state: make(chan struct{}, 1),
98+ exited: make(chan struct{}),
99+ }
100+}
101+
102+// claim grants one attach scanner exclusive permission to dequeue terminal
103+// input. The reader itself remains lifecycle-owned across reconnects.
104+func (p *terminalInputPump) claim() uint64 {
105+ consumer, ok := p.tryClaim()
106+ if !ok {
107+ panic("terminal input consumer already claimed")
108+ }
109+ return consumer
110+}
111+
112+// tryClaim lets competing foreground owners decline admission without panicking.
113+func (p *terminalInputPump) tryClaim() (uint64, bool) {
114+ p.mu.Lock()
115+ if p.consumer != 0 {
116+ p.mu.Unlock()
117+ return 0, false
118+ }
119+ p.nextID++
120+ p.consumer = p.nextID
121+ consumer := p.consumer
122+ p.readyMu.Lock()
123+ p.ready[consumer] = make(chan struct{}, 1)
124+ p.readyMu.Unlock()
125+ ready := (len(p.residual) != 0 || p.pending != nil) && p.delivering == 0
126+ p.mu.Unlock()
127+ p.signalClaimChanged()
128+ if ready {
129+ p.signalReady(consumer)
130+ }
131+ return consumer, true
132+}
133+
134+// revoke invalidates an attempt before its replacement is allowed to claim
135+// input. Pending bytes are deliberately retained for that replacement.
136+func (p *terminalInputPump) revoke(consumer uint64) {
137+ p.mu.Lock()
138+ if p.consumer == consumer {
139+ p.consumer = 0
140+ }
141+ if p.delivering == consumer {
142+ // Legacy attach-attempt replacement remains lossless. Autonomous
143+ // supervisor ownership uses ack-before-picker-handoff and
144+ // drop-before-attachment-handoff, so bytes cannot cross owner kinds.
145+ p.delivering = 0
146+ p.deliveringResidual = false
147+ }
148+ p.mu.Unlock()
149+ p.readyMu.Lock()
150+ delete(p.ready, consumer)
151+ p.readyMu.Unlock()
152+ p.signalClaimChanged()
153+ if p.afterRevoke != nil {
154+ p.afterRevoke()
155+ }
156+}
157+
158+func (p *terminalInputPump) signalClaimChanged() {
159+ select {
160+ case p.claimChanged <- struct{}{}:
161+ default:
162+ }
163+}
164+
165+func (p *terminalInputPump) readyFor(consumer uint64) <-chan struct{} {
166+ p.readyMu.Lock()
167+ defer p.readyMu.Unlock()
168+ return p.ready[consumer]
169+}
170+
171+func (p *terminalInputPump) signalReady(consumer uint64) {
172+ if consumer == 0 {
173+ return
174+ }
175+ p.readyMu.Lock()
176+ ready := p.ready[consumer]
177+ p.readyMu.Unlock()
178+ if ready == nil {
179+ return
180+ }
181+ select {
182+ case ready <- struct{}{}:
183+ default:
184+ }
185+}
186+
187+// enqueue retains at most one completed terminal read. The reader blocks
188+// before publishing another result, keeping handoff buffering bounded.
189+func (p *terminalInputPump) enqueue(result terminalReadResult, activation uint64) bool {
190+ select {
191+ case <-p.done:
192+ return false
193+ case <-p.space:
194+ }
195+ p.mu.Lock()
196+ if p.closed || !p.active || p.activation != activation {
197+ p.mu.Unlock()
198+ p.space <- struct{}{}
199+ return false
200+ }
201+ p.pending = &result
202+ consumer := p.consumer
203+ p.mu.Unlock()
204+ p.signalReady(consumer)
205+ return true
206+}
207+
208+// take leases a ready result only to the current, non-cancelled consumer.
209+// The raw read remains pending until ack, so revocation or cancellation before
210+// scanner delivery leaves the exact bytes available to the next attempt.
211+func (p *terminalInputPump) take(ctx context.Context, consumer uint64) (terminalReadResult, bool) {
212+ p.mu.Lock()
213+ if ctx.Err() != nil || p.consumer != consumer || p.delivering != 0 {
214+ p.mu.Unlock()
215+ return terminalReadResult{}, false
216+ }
217+ if len(p.residual) != 0 {
218+ result := terminalReadResult{data: append([]byte(nil), p.residual...)}
219+ p.delivering = consumer
220+ p.deliveringResidual = true
221+ p.mu.Unlock()
222+ return result, true
223+ }
224+ if p.pending == nil {
225+ p.mu.Unlock()
226+ return terminalReadResult{}, false
227+ }
228+ result := *p.pending
229+ p.delivering = consumer
230+ p.deliveringResidual = false
231+ p.mu.Unlock()
232+ return result, true
233+}
234+
235+// preserveResidual records the undecided marker prefix after callbacks for the
236+// rest of its read have completed. It is intentionally bounded by the marker
237+// scanner's fixed response prefixes and is replayed before later terminal
238+// reads on a replacement attempt.
239+func (p *terminalInputPump) preserveResidual(consumer uint64, data []byte) {
240+ if len(data) == 0 {
241+ return
242+ }
243+ p.mu.Lock()
244+ defer p.mu.Unlock()
245+ if p.consumer != consumer {
246+ return
247+ }
248+ p.residual = append(p.residual[:0], data...)
249+}
250+
251+// ack commits scanner delivery of a leased read and lets the lifecycle reader
252+// accept the next result. Only the consumer that holds the lease can ack it.
253+// dropOwned disposes every undecided byte at an autonomous owner-class
254+// boundary. Picker-queued bytes are dropped before attachment claim, and
255+// attachment-queued bytes are dropped before picker claim. No delivery,
256+// pending read, or preserved residual may cross between those owner classes.
257+// revoke remains lossless for legacy attachment-to-attachment replacement.
258+func (p *terminalInputPump) dropOwned(consumer uint64) {
259+ p.mu.Lock()
260+ if p.consumer != consumer {
261+ p.mu.Unlock()
262+ return
263+ }
264+ hadPending := p.pending != nil
265+ p.pending = nil
266+ p.residual = nil
267+ if p.delivering == consumer {
268+ p.delivering = 0
269+ p.deliveringResidual = false
270+ }
271+ p.mu.Unlock()
272+ if hadPending {
273+ select {
274+ case p.space <- struct{}{}:
275+ default:
276+ }
277+ }
278+}
279+
280+func (p *terminalInputPump) ack(consumer uint64) {
281+ p.mu.Lock()
282+ if p.consumer != consumer || p.delivering != consumer {
283+ p.mu.Unlock()
284+ return
285+ }
286+ wasResidual := p.deliveringResidual
287+ if wasResidual {
288+ p.residual = nil
289+ } else {
290+ p.pending = nil
291+ }
292+ p.delivering = 0
293+ p.deliveringResidual = false
294+ more := len(p.residual) != 0 || p.pending != nil
295+ p.mu.Unlock()
296+ if wasResidual {
297+ if more {
298+ p.signalReady(consumer)
299+ }
300+ return
301+ }
302+ p.space <- struct{}{}
303+}
304+
305+func (p *terminalInputPump) finish() {
306+ p.mu.Lock()
307+ p.closed = true
308+ p.mu.Unlock()
309+ p.signalClaimChanged()
310+ p.signalState()
311+}
312+
313+func (p *terminalInputPump) signalState() {
314+ select {
315+ case p.state <- struct{}{}:
316+ default:
317+ }
318+}
319+
320+func (p *terminalInputPump) waitForActive() (uint64, bool) {
321+ for {
322+ p.mu.Lock()
323+ active, closed, activation := p.active, p.closed, p.activation
324+ p.mu.Unlock()
325+ if closed {
326+ return 0, false
327+ }
328+ if active {
329+ return activation, true
330+ }
331+ select {
332+ case <-p.done:
333+ return 0, false
334+ case <-p.state:
335+ }
336+ }
337+}
338+
339+func (p *terminalInputPump) start() {
340+ p.startMu.Do(func() {
341+ go func() {
342+ defer close(p.exited)
343+ buf := make([]byte, stdinBufSize)
344+ for {
345+ activation, ok := p.waitForActive()
346+ if !ok {
347+ return
348+ }
349+ n, err := p.in.Read(buf)
350+ result := terminalReadResult{err: err}
351+ if n > 0 {
352+ result.data = append([]byte(nil), buf[:n]...)
353+ }
354+ if !p.enqueue(result, activation) {
355+ if p.isClosed() {
356+ return
357+ }
358+ continue
359+ }
360+ if err != nil {
361+ p.setReadCause(err)
362+ p.finish()
363+ return
364+ }
365+ }
366+ }()
367+ })
368+}
369+
370+func (p *terminalInputPump) hasExited() bool {
371+ select {
372+ case <-p.exited:
373+ return true
374+ default:
375+ return false
376+ }
377+}
378+
379+// setReadCause records the cause the terminal reader ended with. An orderly EOF
380+// is the absence of a failure and stays zero, so cause reports io.EOF for it.
381+func (p *terminalInputPump) setReadCause(err error) {
382+ if err == nil || errors.Is(err, io.EOF) {
383+ return
384+ }
385+ p.mu.Lock()
386+ if p.readCause == nil {
387+ p.readCause = err
388+ }
389+ p.mu.Unlock()
390+}
391+
392+// cause reports why the terminal read ended: the read failure when there was
393+// one, otherwise an orderly EOF.
394+func (p *terminalInputPump) cause() error {
395+ p.mu.Lock()
396+ defer p.mu.Unlock()
397+ if p.readCause != nil {
398+ return p.readCause
399+ }
400+ return io.EOF
401+}
402+
403+func (p *terminalInputPump) isClosed() bool {
404+ p.mu.Lock()
405+ defer p.mu.Unlock()
406+ return p.closed
407+}
408+
409+// suspend drops input completed after an attach ends and parks the lifecycle
410+// reader before its next Read. A later Run resumes this same reader, avoiding
411+// both input replay between runs and a competing read on caller-owned stdin.
412+func (p *terminalInputPump) suspend() {
413+ p.mu.Lock()
414+ p.active = false
415+ discarded := p.pending != nil
416+ p.residual = nil
417+ p.pending = nil
418+ p.delivering = 0
419+ p.deliveringResidual = false
420+ p.mu.Unlock()
421+ if discarded {
422+ select {
423+ case p.space <- struct{}{}:
424+ default:
425+ }
426+ }
427+ p.signalState()
428+ p.signalClaimChanged()
429+}
430+
431+func (p *terminalInputPump) resume() {
432+ p.mu.Lock()
433+ if !p.closed && !p.active {
434+ p.activation++
435+ p.active = true
436+ }
437+ p.mu.Unlock()
438+ p.signalState()
439+}
440+
441+func (p *terminalInputPump) stop() {
442+ p.stopMu.Do(func() {
443+ // Mark closure while holding the same mutex that publishes a completed
444+ // Read. A Read which returns after stop therefore cannot publish even
445+ // when both done and space are selectable in enqueue.
446+ p.mu.Lock()
447+ p.closed = true
448+ p.mu.Unlock()
449+ close(p.done)
450+ p.signalClaimChanged()
451+ p.signalState()
452+ })
453+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close