feat(daemon): let snatched clients reclaim sessions

commit: 09a2f38cd41a74ce749ed28a411a251f029d0ec8
author: brice · 2mo ago · 2026-07-27 23:48 UTC
parent: ddc019c
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +1174 -14

M internal/usecase/daemon/client.go +14 -12


@@ -74,8 +74,15 @@ type attachedClient struct {
74	74		// previousSession is guarded independently. It is retained through temporary
75	75		// setSession(nil) hand-offs and cleared only on terminal teardown.
76	76		previousSession Guarded[*session]
77		-	linkMu sync.Mutex
78		-	sendMu sync.Mutex
	77	+	// snatchedInputMu serializes the restricted input parser with its delayed
	78	+	// standalone-ESC callback. It is independent of routing and transport locks;
	79	+	// role activation and terminal close clear it only after releasing those locks.
	80	+	snatchedInputMu sync.Mutex
	81	+	snatchedInputPending []byte
	82	+	snatchedInputDrain bool
	83	+	snatchedInputESC pendingByteTimer
	84	+	linkMu sync.Mutex
	85	+	sendMu sync.Mutex
79	86		}
80	87
81	88		type cursorOut struct {

@@ -215,12 +222,14 @@ func (ac *attachedClient) revokeTransport(tr ports.Transport) ports.Transport {
215	222		return nil
216	223		}
217	224		ac.linkMu.Lock()
218		-	defer ac.linkMu.Unlock()
219	225		if ac.tr != tr {
	226	+	ac.linkMu.Unlock()
220	227		return nil
221	228		}
222	229		ac.tr = nil
223	230		ac.transportIncarnation++
	231	+	ac.linkMu.Unlock()
	232	+	ac.clearSnatchedInput()
224	233		return tr
225	234		}
226	235

@@ -819,8 +828,8 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr
819	828		}
820	829		switch f.Type {
821	830		case ports.MsgInput:
822		-	if in, err := ports.UnmarshalInput(f.Payload); err == nil {
823		-	d.handleSnatchedInput(token, in)
	831	+	if in, err := ports.UnmarshalInput(f.Payload); err == nil && d.handleSnatchedInput(token, in) {
	832	+	return true
824	833		}
825	834		case ports.MsgResize:
826	835		if resize, err := ports.UnmarshalResize(f.Payload); err == nil && resize.Size.Valid() {

@@ -875,13 +884,6 @@ func (d *Daemon) handleSnatchedClientFrame(token attachmentRoleToken, f ports.Fr
875	884		return false
876	885		}
877	886
878		-	// handleSnatchedInput is the narrow seam for the future strict snatched-action
879		-	// parser. Ordinary terminal input is deliberately ignored and never enters the
880		-	// key router or PTY path.
881		-	func (d *Daemon) handleSnatchedInput(token attachmentRoleToken, _ ports.Input) {
882		-	_ = token
883		-	}
884		-
885	887		// clientGone detaches ac if it is still the session's current client. The
886	888		// session remains registered and headless after the client is gone.
887	889		func (d Daemon) clientGone(sess session, ac *attachedClient, failed ports.Transport, explicit bool) {

 	// previousSession is guarded independently. It is retained through temporary
 	// setSession(nil) hand-offs and cleared only on terminal teardown.
 	previousSession Guarded[*session]
-	linkMu          sync.Mutex
+	// snatchedInputMu serializes the restricted input parser with its delayed
-	sendMu          sync.Mutex
+	// standalone-ESC callback. It is independent of routing and transport locks;
+	// role activation and terminal close clear it only after releasing those locks.
+	snatchedInputMu      sync.Mutex
+	snatchedInputPending []byte
+	snatchedInputDrain   bool
+	snatchedInputESC     pendingByteTimer
+	linkMu               sync.Mutex
+	sendMu               sync.Mutex
 }
 type cursorOut struct {
 		return nil
 	}
 	ac.linkMu.Lock()
-	defer ac.linkMu.Unlock()
 	if ac.tr != tr {
+		ac.linkMu.Unlock()
 		return nil
 	}
 	ac.tr = nil
 	ac.transportIncarnation++
+	ac.linkMu.Unlock()
+	ac.clearSnatchedInput()
 	return tr
 }
 	}
 	switch f.Type {
 	case ports.MsgInput:
-		if in, err := ports.UnmarshalInput(f.Payload); err == nil {
+		if in, err := ports.UnmarshalInput(f.Payload); err == nil && d.handleSnatchedInput(token, in) {
-			d.handleSnatchedInput(token, in)
+			return true
 		}
 	case ports.MsgResize:
 		if resize, err := ports.UnmarshalResize(f.Payload); err == nil && resize.Size.Valid() {
 	return false
 }
-// handleSnatchedInput is the narrow seam for the future strict snatched-action
-// parser. Ordinary terminal input is deliberately ignored and never enters the
-// key router or PTY path.
-func (d *Daemon) handleSnatchedInput(token attachmentRoleToken, _ ports.Input) {
-	_ = token
-}
 // clientGone detaches ac if it is still the session's current client. The
 // session remains registered and headless after the client is gone.
 func (d *Daemon) clientGone(sess *session, ac *attachedClient, failed ports.Transport, explicit bool) {

M internal/usecase/daemon/daemon.go +9 -0


@@ -115,6 +115,12 @@ type Daemon struct {
115	115		// afterSnatchOverlayFamily is a deterministic structural lock seam. It runs
116	116		// after one overlay-family mutex is released and before the next is taken.
117	117		afterSnatchOverlayFamily func(string)
	118	+	// afterSnatchedEscapeAccepted observes a standalone ESC after parser state
	119	+	// is consumed and before its captured role and transport are revalidated.
	120	+	afterSnatchedEscapeAccepted func()
	121	+	// afterSnatchedEscapeAttempt reports whether that callback acquired its exact
	122	+	// role capability after revalidation.
	123	+	afterSnatchedEscapeAttempt func(bool)
118	124		// afterActiveFrameDispatch is a deterministic test seam after the connection
119	125		// loop's first role check and before a decoded active frame takes effect.
120	126		afterActiveFrameDispatch func(attachmentRoleToken)

@@ -134,6 +140,9 @@ type Daemon struct {
134	140		// afterDisplacedCleanupStarted observes deferred displaced cleanup before it
135	141		// synchronizes with the attachment gate.
136	142		afterDisplacedCleanupStarted func()
	143	+	// beforeReclaimFirstPaint observes a committed reclaim after displaced cleanup
	144	+	// is queued and before its generation-bound reset first paint is admitted.
	145	+	beforeReclaimFirstPaint func(attachmentRoleToken)
137	146		// afterRoleEffectAdmitted is a deterministic test seam after a frame/paint
138	147		// reserves its exact capability and before its first observable mutation.
139	148		afterRoleEffectAdmitted func(attachmentRoleToken)


@@ -115,6 +115,12 @@ type Daemon struct {
115	// afterSnatchOverlayFamily is a deterministic structural lock seam. It runs	115		// afterSnatchOverlayFamily is a deterministic structural lock seam. It runs
116	// after one overlay-family mutex is released and before the next is taken.	116		// after one overlay-family mutex is released and before the next is taken.
117	afterSnatchOverlayFamily func(string)	117		afterSnatchOverlayFamily func(string)
		118	+	// afterSnatchedEscapeAccepted observes a standalone ESC after parser state
		119	+	// is consumed and before its captured role and transport are revalidated.
		120	+	afterSnatchedEscapeAccepted func()
		121	+	// afterSnatchedEscapeAttempt reports whether that callback acquired its exact
		122	+	// role capability after revalidation.
		123	+	afterSnatchedEscapeAttempt func(bool)
118	// afterActiveFrameDispatch is a deterministic test seam after the connection	124		// afterActiveFrameDispatch is a deterministic test seam after the connection
119	// loop's first role check and before a decoded active frame takes effect.	125		// loop's first role check and before a decoded active frame takes effect.
120	afterActiveFrameDispatch func(attachmentRoleToken)	126		afterActiveFrameDispatch func(attachmentRoleToken)

@@ -134,6 +140,9 @@ type Daemon struct {
134	// afterDisplacedCleanupStarted observes deferred displaced cleanup before it	140		// afterDisplacedCleanupStarted observes deferred displaced cleanup before it
135	// synchronizes with the attachment gate.	141		// synchronizes with the attachment gate.
136	afterDisplacedCleanupStarted func()	142		afterDisplacedCleanupStarted func()
		143	+	// beforeReclaimFirstPaint observes a committed reclaim after displaced cleanup
		144	+	// is queued and before its generation-bound reset first paint is admitted.
		145	+	beforeReclaimFirstPaint func(attachmentRoleToken)
137	// afterRoleEffectAdmitted is a deterministic test seam after a frame/paint	146		// afterRoleEffectAdmitted is a deterministic test seam after a frame/paint
138	// reserves its exact capability and before its first observable mutation.	147		// reserves its exact capability and before its first observable mutation.
139	afterRoleEffectAdmitted func(attachmentRoleToken)	148		afterRoleEffectAdmitted func(attachmentRoleToken)

M internal/usecase/daemon/snatched_client.go +74 -2


@@ -76,6 +76,10 @@ type attachmentTransitionRequest struct {
76	76		preflighted bool
77	77		roleEffectsFrozen bool
78	78		expectedTargetTransportInterrupted bool
	79	+	// activationBarrier is used when promoting an already connected snatched
	80	+	// attachment. Its role gate is deadline-drained before sendMu is acquired,
	81	+	// preserving the gate -> sendMu -> daemon/routing/session publication order.
	82	+	activationBarrier bool
79	83		}
80	84
81	85		func transitionSourceTokenMatchesRequest(token attachmentRoleToken, source *session, req attachmentTransitionRequest) bool {

@@ -235,7 +239,9 @@ func (d *Daemon) transitionAttachment(req attachmentTransitionRequest) (attachme
235	239		source = req.target
236	240		}
237	241		if source == nil \|\| (req.target == nil && req.createTargetLocked == nil) \|\| req.next == nil \|\| d.closing \|\|
238		-	d.sessions[source.id] != source \|\| req.target != nil && d.sessions[req.target.id] != req.target {
	242	+	d.sessions[source.id] != source \|\| req.target != nil && d.sessions[req.target.id] != req.target \|\|
	243	+	req.activationBarrier && (req.sourceToken == nil \|\| source != req.target \|\|
	244	+	req.expectedRole != attachmentSnatched \|\| req.targetRole != attachmentActive) {
239	245		d.mu.Unlock()
240	246		return attachmentTransitionResult{}, errAttachmentTransition
241	247		}

@@ -265,17 +271,47 @@ func (d *Daemon) transitionAttachment(req attachmentTransitionRequest) (attachme
265	271		interrupts := []roleTransportInterrupt{{
266	272		ac: req.expectedTargetCurrent, transport: req.expectedTargetTransport,
267	273		}}
268		-	frozen := freezeRoleEffectGatesInterruptingObserved(interrupts, func(ac *attachedClient) {
	274	+	var drainDeadline *roleEffectDrainDeadline
	275	+	var drainDone func() <-chan struct{}
	276	+	if req.activationBarrier {
	277	+	drainDeadline = newRoleEffectDrainDeadline(d.clock)
	278	+	drainDone = drainDeadline.Done
	279	+	}
	280	+	frozen := freezeRoleEffectGatesInterruptingObservedUntil(interrupts, drainDone, func(ac *attachedClient) {
269	281		if d.afterRoleEffectGateFrozen != nil {
270	282		d.afterRoleEffectGateFrozen(req.action, ac)
271	283		}
272	284		}, participants...)
	285	+	if drainDeadline != nil {
	286	+	drainDeadline.stop()
	287	+	}
273	288		defer frozen.unfreeze()
	289	+	if !frozen.acquired \|\| !frozen.drained {
	290	+	if req.activationBarrier {
	291	+	_ = req.next.closeCapturedTransport(req.expectedTransport.transport)
	292	+	return attachmentTransitionResult{}, errSendTimedOut
	293	+	}
	294	+	return attachmentTransitionResult{}, errAttachmentTransition
	295	+	}
274	296		req.expectedTargetTransportInterrupted = frozen.interrupted(req.expectedTargetCurrent, req.expectedTargetTransport)
275	297		if d.afterRoleEffectsFrozen != nil {
276	298		d.afterRoleEffectsFrozen()
277	299		}
278	300
	301	+	var releaseActivation func()
	302	+	if req.activationBarrier {
	303	+	var err error
	304	+	releaseActivation, err = d.acquireActivationBarrier(*req.sourceToken)
	305	+	if err != nil {
	306	+	return attachmentTransitionResult{}, err
	307	+	}
	308	+	defer func() {
	309	+	if releaseActivation != nil {
	310	+	releaseActivation()
	311	+	}
	312	+	}()
	313	+	}
	314	+
279	315		d.mu.Lock()
280	316		req.roleEffectsFrozen = true
281	317		if req.target == nil {

@@ -293,6 +329,24 @@ func (d *Daemon) transitionAttachment(req attachmentTransitionRequest) (attachme
293	329		}
294	330		result, err := d.transitionAttachmentLocked(req)
295	331		d.mu.Unlock()
	332	+	if err == nil && req.activationBarrier && result.published.role == attachmentActive {
	333	+	ac := result.published.ac
	334	+	if ac.output != nil {
	335	+	ac.output.rebase()
	336	+	}
	337	+	ac.captureFrames = nil
	338	+	d.applyHostThemeSendLocked(result.published.sess, ac, ac.getClientTheme(), false)
	339	+	// The first paint still requests a reset, but the old panel dependency
	340	+	// chain has already been rebased atomically under the activation barrier.
	341	+	result.published.rebase = false
	342	+	}
	343	+	if releaseActivation != nil {
	344	+	releaseActivation()
	345	+	releaseActivation = nil
	346	+	}
	347	+	if err == nil && result.published.role == attachmentActive {
	348	+	result.published.ac.clearSnatchedInput()
	349	+	}
296	350		return result, err
297	351		}
298	352

@@ -543,6 +597,22 @@ func (d *Daemon) sendSnatchedControl(token attachmentRoleToken, frame ports.Fram
543	597		return err
544	598		}
545	599
	600	+	// boundedSendSnatchedControl keeps a best-effort control acknowledgement bound
	601	+	// to the exact admitted role and link incarnation. On deadline it closes only
	602	+	// that captured link, releasing either the transport Send or a wait on sendMu.
	603	+	func (d *Daemon) boundedSendSnatchedControl(token attachmentRoleToken, frame ports.Frame) error {
	604	+	if token.ac == nil \|\| token.effect == nil \|\| token.transport.transport == nil {
	605	+	return errAttachmentTransition
	606	+	}
	607	+	_, err := d.boundedSendWith(token.transport.transport, func() error {
	608	+	return d.sendSnatchedControl(token, frame)
	609	+	})
	610	+	if errors.Is(err, errSendTimedOut) {
	611	+	_ = token.ac.closeCapturedTransport(token.transport.transport)
	612	+	}
	613	+	return err
	614	+	}
	615	+
546	616		// removeSnatchedAttachment removes exactly one role generation and transport
547	617		// from session routing. It never touches the active attachment or closes a
548	618		// transport, allowing callers that already interrupted a blocked send to avoid

@@ -569,6 +639,7 @@ func (d *Daemon) removeSnatchedAttachment(token attachmentRoleToken) bool {
569	639		return false
570	640		}
571	641
	642	+	token.ac.clearSnatchedInput()
572	643		d.unregisterPreview(token.ac)
573	644		token.ac.clearPreviousSession()
574	645		token.ac.setSession(nil)

@@ -620,6 +691,7 @@ func (d *Daemon) cleanupInterruptedSnatchedAttachment(token attachmentRoleToken)
620	691		}
621	692
622	693		_ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport))
	694	+	token.ac.clearSnatchedInput()
623	695		d.unregisterPreview(token.ac)
624	696		token.ac.clearPreviousSession()
625	697		token.ac.setSession(nil)


@@ -76,6 +76,10 @@ type attachmentTransitionRequest struct {
76		preflighted bool	76		preflighted bool
77		roleEffectsFrozen bool	77		roleEffectsFrozen bool
78		expectedTargetTransportInterrupted bool	78		expectedTargetTransportInterrupted bool
			79	+	// activationBarrier is used when promoting an already connected snatched
			80	+	// attachment. Its role gate is deadline-drained before sendMu is acquired,
			81	+	// preserving the gate -> sendMu -> daemon/routing/session publication order.
			82	+	activationBarrier bool
79		}	83		}
80			84
81		func transitionSourceTokenMatchesRequest(token attachmentRoleToken, source *session, req attachmentTransitionRequest) bool {	85		func transitionSourceTokenMatchesRequest(token attachmentRoleToken, source *session, req attachmentTransitionRequest) bool {

@@ -235,7 +239,9 @@ func (d *Daemon) transitionAttachment(req attachmentTransitionRequest) (attachme
235		source = req.target	239		source = req.target
236		}	240		}
237		if source == nil \|\| (req.target == nil && req.createTargetLocked == nil) \|\| req.next == nil \|\| d.closing \|\|	241		if source == nil \|\| (req.target == nil && req.createTargetLocked == nil) \|\| req.next == nil \|\| d.closing \|\|
238	-	d.sessions[source.id] != source \|\| req.target != nil && d.sessions[req.target.id] != req.target {	242	+	d.sessions[source.id] != source \|\| req.target != nil && d.sessions[req.target.id] != req.target \|\|
			243	+	req.activationBarrier && (req.sourceToken == nil \|\| source != req.target \|\|
			244	+	req.expectedRole != attachmentSnatched \|\| req.targetRole != attachmentActive) {
239		d.mu.Unlock()	245		d.mu.Unlock()
240		return attachmentTransitionResult{}, errAttachmentTransition	246		return attachmentTransitionResult{}, errAttachmentTransition
241		}	247		}

@@ -265,17 +271,47 @@ func (d *Daemon) transitionAttachment(req attachmentTransitionRequest) (attachme
265		interrupts := []roleTransportInterrupt{{	271		interrupts := []roleTransportInterrupt{{
266		ac: req.expectedTargetCurrent, transport: req.expectedTargetTransport,	272		ac: req.expectedTargetCurrent, transport: req.expectedTargetTransport,
267		}}	273		}}
268	-	frozen := freezeRoleEffectGatesInterruptingObserved(interrupts, func(ac *attachedClient) {	274	+	var drainDeadline *roleEffectDrainDeadline
			275	+	var drainDone func() <-chan struct{}
			276	+	if req.activationBarrier {
			277	+	drainDeadline = newRoleEffectDrainDeadline(d.clock)
			278	+	drainDone = drainDeadline.Done
			279	+	}
			280	+	frozen := freezeRoleEffectGatesInterruptingObservedUntil(interrupts, drainDone, func(ac *attachedClient) {
269		if d.afterRoleEffectGateFrozen != nil {	281		if d.afterRoleEffectGateFrozen != nil {
270		d.afterRoleEffectGateFrozen(req.action, ac)	282		d.afterRoleEffectGateFrozen(req.action, ac)
271		}	283		}
272		}, participants...)	284		}, participants...)
			285	+	if drainDeadline != nil {
			286	+	drainDeadline.stop()
			287	+	}
273		defer frozen.unfreeze()	288		defer frozen.unfreeze()
			289	+	if !frozen.acquired \|\| !frozen.drained {
			290	+	if req.activationBarrier {
			291	+	_ = req.next.closeCapturedTransport(req.expectedTransport.transport)
			292	+	return attachmentTransitionResult{}, errSendTimedOut
			293	+	}
			294	+	return attachmentTransitionResult{}, errAttachmentTransition
			295	+	}
274		req.expectedTargetTransportInterrupted = frozen.interrupted(req.expectedTargetCurrent, req.expectedTargetTransport)	296		req.expectedTargetTransportInterrupted = frozen.interrupted(req.expectedTargetCurrent, req.expectedTargetTransport)
275		if d.afterRoleEffectsFrozen != nil {	297		if d.afterRoleEffectsFrozen != nil {
276		d.afterRoleEffectsFrozen()	298		d.afterRoleEffectsFrozen()
277		}	299		}
278			300
			301	+	var releaseActivation func()
			302	+	if req.activationBarrier {
			303	+	var err error
			304	+	releaseActivation, err = d.acquireActivationBarrier(*req.sourceToken)
			305	+	if err != nil {
			306	+	return attachmentTransitionResult{}, err
			307	+	}
			308	+	defer func() {
			309	+	if releaseActivation != nil {
			310	+	releaseActivation()
			311	+	}
			312	+	}()
			313	+	}
			314	+
279		d.mu.Lock()	315		d.mu.Lock()
280		req.roleEffectsFrozen = true	316		req.roleEffectsFrozen = true
281		if req.target == nil {	317		if req.target == nil {

@@ -293,6 +329,24 @@ func (d *Daemon) transitionAttachment(req attachmentTransitionRequest) (attachme
293		}	329		}
294		result, err := d.transitionAttachmentLocked(req)	330		result, err := d.transitionAttachmentLocked(req)
295		d.mu.Unlock()	331		d.mu.Unlock()
			332	+	if err == nil && req.activationBarrier && result.published.role == attachmentActive {
			333	+	ac := result.published.ac
			334	+	if ac.output != nil {
			335	+	ac.output.rebase()
			336	+	}
			337	+	ac.captureFrames = nil
			338	+	d.applyHostThemeSendLocked(result.published.sess, ac, ac.getClientTheme(), false)
			339	+	// The first paint still requests a reset, but the old panel dependency
			340	+	// chain has already been rebased atomically under the activation barrier.
			341	+	result.published.rebase = false
			342	+	}
			343	+	if releaseActivation != nil {
			344	+	releaseActivation()
			345	+	releaseActivation = nil
			346	+	}
			347	+	if err == nil && result.published.role == attachmentActive {
			348	+	result.published.ac.clearSnatchedInput()
			349	+	}
296		return result, err	350		return result, err
297		}	351		}
298			352

@@ -543,6 +597,22 @@ func (d *Daemon) sendSnatchedControl(token attachmentRoleToken, frame ports.Fram
543		return err	597		return err
544		}	598		}
545			599
			600	+	// boundedSendSnatchedControl keeps a best-effort control acknowledgement bound
			601	+	// to the exact admitted role and link incarnation. On deadline it closes only
			602	+	// that captured link, releasing either the transport Send or a wait on sendMu.
			603	+	func (d *Daemon) boundedSendSnatchedControl(token attachmentRoleToken, frame ports.Frame) error {
			604	+	if token.ac == nil \|\| token.effect == nil \|\| token.transport.transport == nil {
			605	+	return errAttachmentTransition
			606	+	}
			607	+	_, err := d.boundedSendWith(token.transport.transport, func() error {
			608	+	return d.sendSnatchedControl(token, frame)
			609	+	})
			610	+	if errors.Is(err, errSendTimedOut) {
			611	+	_ = token.ac.closeCapturedTransport(token.transport.transport)
			612	+	}
			613	+	return err
			614	+	}
			615	+
546		// removeSnatchedAttachment removes exactly one role generation and transport	616		// removeSnatchedAttachment removes exactly one role generation and transport
547		// from session routing. It never touches the active attachment or closes a	617		// from session routing. It never touches the active attachment or closes a
548		// transport, allowing callers that already interrupted a blocked send to avoid	618		// transport, allowing callers that already interrupted a blocked send to avoid

@@ -569,6 +639,7 @@ func (d *Daemon) removeSnatchedAttachment(token attachmentRoleToken) bool {
569		return false	639		return false
570		}	640		}
571			641
			642	+	token.ac.clearSnatchedInput()
572		d.unregisterPreview(token.ac)	643		d.unregisterPreview(token.ac)
573		token.ac.clearPreviousSession()	644		token.ac.clearPreviousSession()
574		token.ac.setSession(nil)	645		token.ac.setSession(nil)

@@ -620,6 +691,7 @@ func (d *Daemon) cleanupInterruptedSnatchedAttachment(token attachmentRoleToken)
620		}	691		}
621			692
622		_ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport))	693		_ = token.ac.closeCapturedTransport(token.ac.revokeTransport(token.transport.transport))
			694	+	token.ac.clearSnatchedInput()
623		d.unregisterPreview(token.ac)	695		d.unregisterPreview(token.ac)
624		token.ac.clearPreviousSession()	696		token.ac.clearPreviousSession()
625		token.ac.setSession(nil)	697		token.ac.setSession(nil)

A internal/usecase/daemon/snatched_input.go +261 -0

@@ -0,0 +1,261 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"errors"
	5	+
	6	+	"github.com/bnema/vev/internal/ports"
	7	+	"github.com/bnema/vev/internal/usecase/keys"
	8	+	)
	9	+
	10	+	const snatchedInputSequenceLimit = 64
	11	+
	12	+	type snatchedInputAction uint8
	13	+
	14	+	const (
	15	+	snatchedInputNone snatchedInputAction = iota
	16	+	snatchedInputReclaim
	17	+	snatchedInputQuit
	18	+	)
	19	+
	20	+	// clearSnatchedInput invalidates every parser continuation before stopping its
	21	+	// timer. Callers must not hold daemon, routing, session, coordinator, send, or
	22	+	// transport locks because Timer.Stop is an external call.
	23	+	func (ac *attachedClient) clearSnatchedInput() {
	24	+	if ac == nil {
	25	+	return
	26	+	}
	27	+	ac.snatchedInputMu.Lock()
	28	+	ac.snatchedInputPending = nil
	29	+	ac.snatchedInputDrain = false
	30	+	pendingESC := ac.snatchedInputESC
	31	+	ac.snatchedInputESC = pendingByteTimer{}
	32	+	ac.snatchedInputMu.Unlock()
	33	+	pendingESC.stop()
	34	+	}
	35	+
	36	+	// handleSnatchedInput accepts actions only as exact one-byte input events.
	37	+	// Terminal escape sequences and arbitrary coalesced input are consumed as
	38	+	// indivisible events so a byte in a paste, sequence final, or frame tail can
	39	+	// never become an action.
	40	+	func (d *Daemon) handleSnatchedInput(token attachmentRoleToken, input ports.Input) bool {
	41	+	action := d.parseSnatchedInput(token, input.Data)
	42	+	switch action {
	43	+	case snatchedInputReclaim:
	44	+	d.reclaimSnatchedAttachment(token)
	45	+	case snatchedInputQuit:
	46	+	return d.quitSnatchedAttachment(token)
	47	+	}
	48	+	return false
	49	+	}
	50	+
	51	+	func (d *Daemon) parseSnatchedInput(token attachmentRoleToken, data []byte) snatchedInputAction {
	52	+	ac := token.ac
	53	+	if ac == nil \|\| len(data) == 0 {
	54	+	return snatchedInputNone
	55	+	}
	56	+
	57	+	ac.snatchedInputMu.Lock()
	58	+
	59	+	if ac.snatchedInputDrain {
	60	+	if snatchedSequenceFinal(data) >= 0 {
	61	+	ac.snatchedInputDrain = false
	62	+	}
	63	+	ac.snatchedInputMu.Unlock()
	64	+	return snatchedInputNone
	65	+	}
	66	+
	67	+	if len(ac.snatchedInputPending) != 0 {
	68	+	pendingESC := ac.snatchedInputESC
	69	+	ac.snatchedInputESC = pendingByteTimer{}
	70	+	combined := make([]byte, 0, len(ac.snatchedInputPending)+len(data))
	71	+	combined = append(combined, ac.snatchedInputPending...)
	72	+	combined = append(combined, data...)
	73	+	ac.snatchedInputPending = nil
	74	+	d.retainSnatchedSequenceLocked(ac, combined)
	75	+	ac.snatchedInputMu.Unlock()
	76	+	pendingESC.stop()
	77	+	return snatchedInputNone
	78	+	}
	79	+
	80	+	if len(data) == 1 {
	81	+	var action snatchedInputAction
	82	+	switch data[0] {
	83	+	case 'r', 'R':
	84	+	action = snatchedInputReclaim
	85	+	case 'q', 'Q':
	86	+	action = snatchedInputQuit
	87	+	case keys.ESC:
	88	+	ac.snatchedInputPending = append(ac.snatchedInputPending[:0], keys.ESC)
	89	+	ac.snatchedInputESC.retain(d.clock, keys.ESCDelay, func(timer ports.Timer) {
	90	+	d.fireSnatchedEscape(token, timer)
	91	+	})
	92	+	}
	93	+	ac.snatchedInputMu.Unlock()
	94	+	return action
	95	+	}
	96	+
	97	+	// A complete/coalesced event is always ignored. Only an incomplete CSI or
	98	+	// SS3 prefix is retained to prevent its future final byte from acting.
	99	+	d.retainSnatchedSequenceLocked(ac, data)
	100	+	ac.snatchedInputMu.Unlock()
	101	+	return snatchedInputNone
	102	+	}
	103	+
	104	+	func (d Daemon) retainSnatchedSequenceLocked(ac attachedClient, data []byte) {
	105	+	if !snatchedSequencePrefix(data) \|\| snatchedSequenceFinal(data[2:]) >= 0 {
	106	+	return
	107	+	}
	108	+	if len(data) > snatchedInputSequenceLimit {
	109	+	ac.snatchedInputDrain = true
	110	+	return
	111	+	}
	112	+	ac.snatchedInputPending = append(ac.snatchedInputPending[:0], data...)
	113	+	}
	114	+
	115	+	func snatchedSequencePrefix(data []byte) bool {
	116	+	return len(data) >= 2 && data[0] == keys.ESC && (data[1] == '[' \|\| data[1] == 'O')
	117	+	}
	118	+
	119	+	func snatchedSequenceFinal(data []byte) int {
	120	+	for i, b := range data {
	121	+	if b >= 0x40 && b <= 0x7e {
	122	+	return i
	123	+	}
	124	+	}
	125	+	return -1
	126	+	}
	127	+
	128	+	func (d *Daemon) fireSnatchedEscape(token attachmentRoleToken, timer ports.Timer) {
	129	+	ac := token.ac
	130	+	if ac == nil {
	131	+	return
	132	+	}
	133	+	ac.snatchedInputMu.Lock()
	134	+	if ac.snatchedInputESC.timer != timer \|\| len(ac.snatchedInputPending) != 1 \|\| ac.snatchedInputPending[0] != keys.ESC {
	135	+	ac.snatchedInputMu.Unlock()
	136	+	return
	137	+	}
	138	+	ac.snatchedInputPending = nil
	139	+	ac.snatchedInputESC.timer = nil
	140	+	ac.snatchedInputESC.done = nil
	141	+	ac.snatchedInputMu.Unlock()
	142	+
	143	+	if d.afterSnatchedEscapeAccepted != nil {
	144	+	d.afterSnatchedEscapeAccepted()
	145	+	}
	146	+	if !token.current() {
	147	+	if d.afterSnatchedEscapeAttempt != nil {
	148	+	d.afterSnatchedEscapeAttempt(false)
	149	+	}
	150	+	return
	151	+	}
	152	+	ticket, admitted := ac.beginRoleEffect(token)
	153	+	if d.afterSnatchedEscapeAttempt != nil {
	154	+	d.afterSnatchedEscapeAttempt(admitted)
	155	+	}
	156	+	if !admitted {
	157	+	return
	158	+	}
	159	+	token.effect = ticket
	160	+	defer token.endRoleEffect()
	161	+	d.quitSnatchedAttachment(token)
	162	+	}
	163	+
	164	+	// acquireActivationBarrier waits for the requester's serialized output path
	165	+	// without retaining role, daemon, routing, session, coordinator, or pane locks.
	166	+	// The caller has already frozen and drained the requester's role gate, so no new
	167	+	// role-bound sender can race this acquisition. A deadline closes only the exact
	168	+	// captured link, which releases a transport Send currently holding sendMu.
	169	+	func (d *Daemon) acquireActivationBarrier(token attachmentRoleToken) (func(), error) {
	170	+	if token.ac == nil \|\| token.transport.transport == nil \|\| !token.current() {
	171	+	return nil, errAttachmentTransition
	172	+	}
	173	+
	174	+	acquired := make(chan struct{})
	175	+	cancel := make(chan struct{})
	176	+	go func() {
	177	+	token.ac.sendMu.Lock()
	178	+	select {
	179	+	case acquired <- struct{}{}:
	180	+	// The receiver now owns the matching Unlock obligation.
	181	+	case <-cancel:
	182	+	token.ac.sendMu.Unlock()
	183	+	}
	184	+	}()
	185	+
	186	+	timer := d.clock.NewTimer(detachNotifyTimeout)
	187	+	select {
	188	+	case <-acquired:
	189	+	timer.Stop()
	190	+	if token.ac.roleGeneration.Load() != token.generation \|\|
	191	+	!token.ac.transportSnapshotCurrent(token.transport) {
	192	+	token.ac.sendMu.Unlock()
	193	+	return nil, errAttachmentTransition
	194	+	}
	195	+	return token.ac.sendMu.Unlock, nil
	196	+	case <-timer.C():
	197	+	close(cancel)
	198	+	_ = token.ac.closeCapturedTransport(token.transport.transport)
	199	+	return nil, errSendTimedOut
	200	+	}
	201	+	}
	202	+
	203	+	func (d *Daemon) reclaimSnatchedAttachment(token attachmentRoleToken) bool {
	204	+	if token.role != attachmentSnatched \|\| token.effect == nil {
	205	+	return false
	206	+	}
	207	+	token.effect.bindActionEnd(d, "reclaim-snatched")
	208	+	transition, err := d.transitionAttachment(attachmentTransitionRequest{
	209	+	source: token.sess, target: token.sess, next: token.ac,
	210	+	expectedRole: attachmentSnatched, targetRole: attachmentActive,
	211	+	expectedTransport: token.transport, sourceToken: &token,
	212	+	action: "reclaim-snatched", ready: true, activationBarrier: true,
	213	+	})
	214	+	if err != nil {
	215	+	if errors.Is(err, errSendTimedOut) {
	216	+	d.attachmentCleanupWg.Go(func() { d.dropSnatchedAttachment(token) })
	217	+	} else {
	218	+	d.showSnatchedUnavailable(token)
	219	+	}
	220	+	return false
	221	+	}
	222	+	d.deferAttachmentTransitionCleanups(transition)
	223	+	if d.beforeReclaimFirstPaint != nil {
	224	+	d.beforeReclaimFirstPaint(transition.published)
	225	+	}
	226	+	d.firstPaintForTransition(transition.published)
	227	+	return true
	228	+	}
	229	+
	230	+	// showSnatchedUnavailable is a best-effort precommit failure panel. Fresh role
	231	+	// admission and the generation-aware sender make it inert after a concurrent
	232	+	// reclaim, quit, transport replacement, or session teardown.
	233	+	func (d *Daemon) showSnatchedUnavailable(token attachmentRoleToken) {
	234	+	if !token.current() {
	235	+	return
	236	+	}
	237	+	fresh := token
	238	+	fresh.effect = nil
	239	+	ticket, admitted := token.ac.beginRoleEffect(fresh)
	240	+	if !admitted {
	241	+	return
	242	+	}
	243	+	fresh.effect = ticket
	244	+	defer fresh.endRoleEffect()
	245	+	if err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "Session is no longer available.", ticket); err != nil {
	246	+	fresh.endRoleEffect()
	247	+	d.dropSnatchedAttachment(token)
	248	+	}
	249	+	}
	250	+
	251	+	func (d *Daemon) quitSnatchedAttachment(token attachmentRoleToken) bool {
	252	+	if token.role != attachmentSnatched \|\| token.effect == nil {
	253	+	return false
	254	+	}
	255	+	token.effect.bindActionEnd(d, "quit-snatched")
	256	+	_ = d.boundedSendSnatchedControl(token, frameDetached(ports.ReasonDetach))
	257	+	// Terminal cleanup freezes the same role gate, so release this action's
	258	+	// admission before attempting the generation-bound removal.
	259	+	token.endRoleEffect()
	260	+	return d.dropSnatchedAttachment(token)
	261	+	}

@@ -0,0 +1,261 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"errors"
	5	+
	6	+	"github.com/bnema/vev/internal/ports"
	7	+	"github.com/bnema/vev/internal/usecase/keys"
	8	+	)
	9	+
	10	+	const snatchedInputSequenceLimit = 64
	11	+
	12	+	type snatchedInputAction uint8
	13	+
	14	+	const (
	15	+	snatchedInputNone snatchedInputAction = iota
	16	+	snatchedInputReclaim
	17	+	snatchedInputQuit
	18	+	)
	19	+
	20	+	// clearSnatchedInput invalidates every parser continuation before stopping its
	21	+	// timer. Callers must not hold daemon, routing, session, coordinator, send, or
	22	+	// transport locks because Timer.Stop is an external call.
	23	+	func (ac *attachedClient) clearSnatchedInput() {
	24	+	if ac == nil {
	25	+	return
	26	+	}
	27	+	ac.snatchedInputMu.Lock()
	28	+	ac.snatchedInputPending = nil
	29	+	ac.snatchedInputDrain = false
	30	+	pendingESC := ac.snatchedInputESC
	31	+	ac.snatchedInputESC = pendingByteTimer{}
	32	+	ac.snatchedInputMu.Unlock()
	33	+	pendingESC.stop()
	34	+	}
	35	+
	36	+	// handleSnatchedInput accepts actions only as exact one-byte input events.
	37	+	// Terminal escape sequences and arbitrary coalesced input are consumed as
	38	+	// indivisible events so a byte in a paste, sequence final, or frame tail can
	39	+	// never become an action.
	40	+	func (d *Daemon) handleSnatchedInput(token attachmentRoleToken, input ports.Input) bool {
	41	+	action := d.parseSnatchedInput(token, input.Data)
	42	+	switch action {
	43	+	case snatchedInputReclaim:
	44	+	d.reclaimSnatchedAttachment(token)
	45	+	case snatchedInputQuit:
	46	+	return d.quitSnatchedAttachment(token)
	47	+	}
	48	+	return false
	49	+	}
	50	+
	51	+	func (d *Daemon) parseSnatchedInput(token attachmentRoleToken, data []byte) snatchedInputAction {
	52	+	ac := token.ac
	53	+	if ac == nil \|\| len(data) == 0 {
	54	+	return snatchedInputNone
	55	+	}
	56	+
	57	+	ac.snatchedInputMu.Lock()
	58	+
	59	+	if ac.snatchedInputDrain {
	60	+	if snatchedSequenceFinal(data) >= 0 {
	61	+	ac.snatchedInputDrain = false
	62	+	}
	63	+	ac.snatchedInputMu.Unlock()
	64	+	return snatchedInputNone
	65	+	}
	66	+
	67	+	if len(ac.snatchedInputPending) != 0 {
	68	+	pendingESC := ac.snatchedInputESC
	69	+	ac.snatchedInputESC = pendingByteTimer{}
	70	+	combined := make([]byte, 0, len(ac.snatchedInputPending)+len(data))
	71	+	combined = append(combined, ac.snatchedInputPending...)
	72	+	combined = append(combined, data...)
	73	+	ac.snatchedInputPending = nil
	74	+	d.retainSnatchedSequenceLocked(ac, combined)
	75	+	ac.snatchedInputMu.Unlock()
	76	+	pendingESC.stop()
	77	+	return snatchedInputNone
	78	+	}
	79	+
	80	+	if len(data) == 1 {
	81	+	var action snatchedInputAction
	82	+	switch data[0] {
	83	+	case 'r', 'R':
	84	+	action = snatchedInputReclaim
	85	+	case 'q', 'Q':
	86	+	action = snatchedInputQuit
	87	+	case keys.ESC:
	88	+	ac.snatchedInputPending = append(ac.snatchedInputPending[:0], keys.ESC)
	89	+	ac.snatchedInputESC.retain(d.clock, keys.ESCDelay, func(timer ports.Timer) {
	90	+	d.fireSnatchedEscape(token, timer)
	91	+	})
	92	+	}
	93	+	ac.snatchedInputMu.Unlock()
	94	+	return action
	95	+	}
	96	+
	97	+	// A complete/coalesced event is always ignored. Only an incomplete CSI or
	98	+	// SS3 prefix is retained to prevent its future final byte from acting.
	99	+	d.retainSnatchedSequenceLocked(ac, data)
	100	+	ac.snatchedInputMu.Unlock()
	101	+	return snatchedInputNone
	102	+	}
	103	+
	104	+	func (d Daemon) retainSnatchedSequenceLocked(ac attachedClient, data []byte) {
	105	+	if !snatchedSequencePrefix(data) \|\| snatchedSequenceFinal(data[2:]) >= 0 {
	106	+	return
	107	+	}
	108	+	if len(data) > snatchedInputSequenceLimit {
	109	+	ac.snatchedInputDrain = true
	110	+	return
	111	+	}
	112	+	ac.snatchedInputPending = append(ac.snatchedInputPending[:0], data...)
	113	+	}
	114	+
	115	+	func snatchedSequencePrefix(data []byte) bool {
	116	+	return len(data) >= 2 && data[0] == keys.ESC && (data[1] == '[' \|\| data[1] == 'O')
	117	+	}
	118	+
	119	+	func snatchedSequenceFinal(data []byte) int {
	120	+	for i, b := range data {
	121	+	if b >= 0x40 && b <= 0x7e {
	122	+	return i
	123	+	}
	124	+	}
	125	+	return -1
	126	+	}
	127	+
	128	+	func (d *Daemon) fireSnatchedEscape(token attachmentRoleToken, timer ports.Timer) {
	129	+	ac := token.ac
	130	+	if ac == nil {
	131	+	return
	132	+	}
	133	+	ac.snatchedInputMu.Lock()
	134	+	if ac.snatchedInputESC.timer != timer \|\| len(ac.snatchedInputPending) != 1 \|\| ac.snatchedInputPending[0] != keys.ESC {
	135	+	ac.snatchedInputMu.Unlock()
	136	+	return
	137	+	}
	138	+	ac.snatchedInputPending = nil
	139	+	ac.snatchedInputESC.timer = nil
	140	+	ac.snatchedInputESC.done = nil
	141	+	ac.snatchedInputMu.Unlock()
	142	+
	143	+	if d.afterSnatchedEscapeAccepted != nil {
	144	+	d.afterSnatchedEscapeAccepted()
	145	+	}
	146	+	if !token.current() {
	147	+	if d.afterSnatchedEscapeAttempt != nil {
	148	+	d.afterSnatchedEscapeAttempt(false)
	149	+	}
	150	+	return
	151	+	}
	152	+	ticket, admitted := ac.beginRoleEffect(token)
	153	+	if d.afterSnatchedEscapeAttempt != nil {
	154	+	d.afterSnatchedEscapeAttempt(admitted)
	155	+	}
	156	+	if !admitted {
	157	+	return
	158	+	}
	159	+	token.effect = ticket
	160	+	defer token.endRoleEffect()
	161	+	d.quitSnatchedAttachment(token)
	162	+	}
	163	+
	164	+	// acquireActivationBarrier waits for the requester's serialized output path
	165	+	// without retaining role, daemon, routing, session, coordinator, or pane locks.
	166	+	// The caller has already frozen and drained the requester's role gate, so no new
	167	+	// role-bound sender can race this acquisition. A deadline closes only the exact
	168	+	// captured link, which releases a transport Send currently holding sendMu.
	169	+	func (d *Daemon) acquireActivationBarrier(token attachmentRoleToken) (func(), error) {
	170	+	if token.ac == nil \|\| token.transport.transport == nil \|\| !token.current() {
	171	+	return nil, errAttachmentTransition
	172	+	}
	173	+
	174	+	acquired := make(chan struct{})
	175	+	cancel := make(chan struct{})
	176	+	go func() {
	177	+	token.ac.sendMu.Lock()
	178	+	select {
	179	+	case acquired <- struct{}{}:
	180	+	// The receiver now owns the matching Unlock obligation.
	181	+	case <-cancel:
	182	+	token.ac.sendMu.Unlock()
	183	+	}
	184	+	}()
	185	+
	186	+	timer := d.clock.NewTimer(detachNotifyTimeout)
	187	+	select {
	188	+	case <-acquired:
	189	+	timer.Stop()
	190	+	if token.ac.roleGeneration.Load() != token.generation \|\|
	191	+	!token.ac.transportSnapshotCurrent(token.transport) {
	192	+	token.ac.sendMu.Unlock()
	193	+	return nil, errAttachmentTransition
	194	+	}
	195	+	return token.ac.sendMu.Unlock, nil
	196	+	case <-timer.C():
	197	+	close(cancel)
	198	+	_ = token.ac.closeCapturedTransport(token.transport.transport)
	199	+	return nil, errSendTimedOut
	200	+	}
	201	+	}
	202	+
	203	+	func (d *Daemon) reclaimSnatchedAttachment(token attachmentRoleToken) bool {
	204	+	if token.role != attachmentSnatched \|\| token.effect == nil {
	205	+	return false
	206	+	}
	207	+	token.effect.bindActionEnd(d, "reclaim-snatched")
	208	+	transition, err := d.transitionAttachment(attachmentTransitionRequest{
	209	+	source: token.sess, target: token.sess, next: token.ac,
	210	+	expectedRole: attachmentSnatched, targetRole: attachmentActive,
	211	+	expectedTransport: token.transport, sourceToken: &token,
	212	+	action: "reclaim-snatched", ready: true, activationBarrier: true,
	213	+	})
	214	+	if err != nil {
	215	+	if errors.Is(err, errSendTimedOut) {
	216	+	d.attachmentCleanupWg.Go(func() { d.dropSnatchedAttachment(token) })
	217	+	} else {
	218	+	d.showSnatchedUnavailable(token)
	219	+	}
	220	+	return false
	221	+	}
	222	+	d.deferAttachmentTransitionCleanups(transition)
	223	+	if d.beforeReclaimFirstPaint != nil {
	224	+	d.beforeReclaimFirstPaint(transition.published)
	225	+	}
	226	+	d.firstPaintForTransition(transition.published)
	227	+	return true
	228	+	}
	229	+
	230	+	// showSnatchedUnavailable is a best-effort precommit failure panel. Fresh role
	231	+	// admission and the generation-aware sender make it inert after a concurrent
	232	+	// reclaim, quit, transport replacement, or session teardown.
	233	+	func (d *Daemon) showSnatchedUnavailable(token attachmentRoleToken) {
	234	+	if !token.current() {
	235	+	return
	236	+	}
	237	+	fresh := token
	238	+	fresh.effect = nil
	239	+	ticket, admitted := token.ac.beginRoleEffect(fresh)
	240	+	if !admitted {
	241	+	return
	242	+	}
	243	+	fresh.effect = ticket
	244	+	defer fresh.endRoleEffect()
	245	+	if err := d.sendSnatchedPanel(token.ac, token.transport, token.generation, "Session is no longer available.", ticket); err != nil {
	246	+	fresh.endRoleEffect()
	247	+	d.dropSnatchedAttachment(token)
	248	+	}
	249	+	}
	250	+
	251	+	func (d *Daemon) quitSnatchedAttachment(token attachmentRoleToken) bool {
	252	+	if token.role != attachmentSnatched \|\| token.effect == nil {
	253	+	return false
	254	+	}
	255	+	token.effect.bindActionEnd(d, "quit-snatched")
	256	+	_ = d.boundedSendSnatchedControl(token, frameDetached(ports.ReasonDetach))
	257	+	// Terminal cleanup freezes the same role gate, so release this action's
	258	+	// admission before attempting the generation-bound removal.
	259	+	token.endRoleEffect()
	260	+	return d.dropSnatchedAttachment(token)
	261	+	}

A internal/usecase/daemon/snatched_input_test.go +365 -0

@@ -0,0 +1,365 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"io"
	6	+	"strings"
	7	+	"sync"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"github.com/bnema/vev/internal/domain"
	14	+	"github.com/bnema/vev/internal/ports"
	15	+	"github.com/bnema/vev/internal/usecase/keys"
	16	+	)
	17	+
	18	+	type snatchedInputTimer struct {
	19	+	ch chan time.Time
	20	+	}
	21	+
	22	+	func newSnatchedInputTimer() *snatchedInputTimer {
	23	+	return &snatchedInputTimer{ch: make(chan time.Time, 1)}
	24	+	}
	25	+
	26	+	func (t *snatchedInputTimer) C() <-chan time.Time { return t.ch }
	27	+	func (*snatchedInputTimer) Reset(time.Duration) bool { return false }
	28	+	func (*snatchedInputTimer) Stop() bool { return true }
	29	+	func (t *snatchedInputTimer) Fire() { t.ch <- time.Time{} }
	30	+
	31	+	type snatchedInputClock struct {
	32	+	timers chan *snatchedInputTimer
	33	+	}
	34	+
	35	+	func newSnatchedInputClock() *snatchedInputClock {
	36	+	return &snatchedInputClock{timers: make(chan *snatchedInputTimer, 8)}
	37	+	}
	38	+
	39	+	func (*snatchedInputClock) Now() time.Time { return time.Time{} }
	40	+	func (c *snatchedInputClock) NewTimer(delay time.Duration) ports.Timer {
	41	+	if delay != keys.ESCDelay {
	42	+	return stubTimer{}
	43	+	}
	44	+	timer := newSnatchedInputTimer()
	45	+	c.timers <- timer
	46	+	return timer
	47	+	}
	48	+
	49	+	type snatchedCloseSignalTransport struct {
	50	+	mu sync.Mutex
	51	+	closed bool
	52	+	closedCh chan struct{}
	53	+	closeOnce sync.Once
	54	+	}
	55	+
	56	+	func newSnatchedCloseSignalTransport() *snatchedCloseSignalTransport {
	57	+	return &snatchedCloseSignalTransport{closedCh: make(chan struct{})}
	58	+	}
	59	+
	60	+	func (*snatchedCloseSignalTransport) Send(ports.Frame) error { return nil }
	61	+	func (*snatchedCloseSignalTransport) Recv() (ports.Frame, error) {
	62	+	return ports.Frame{}, errors.New("closed")
	63	+	}
	64	+	func (t *snatchedCloseSignalTransport) Close() error {
	65	+	t.mu.Lock()
	66	+	t.closed = true
	67	+	t.mu.Unlock()
	68	+	t.closeOnce.Do(func() { close(t.closedCh) })
	69	+	return nil
	70	+	}
	71	+	func (t *snatchedCloseSignalTransport) Closed() bool {
	72	+	t.mu.Lock()
	73	+	defer t.mu.Unlock()
	74	+	return t.closed
	75	+	}
	76	+
	77	+	type blockedSnatchedQuitTransport struct {
	78	+	recv chan ports.Frame
	79	+	sendStarted chan struct{}
	80	+	sendDone chan struct{}
	81	+	closed chan struct{}
	82	+	closeOnce sync.Once
	83	+	sendOnce sync.Once
	84	+	mu sync.Mutex
	85	+	isClosed bool
	86	+	}
	87	+
	88	+	func newBlockedSnatchedQuitTransport() *blockedSnatchedQuitTransport {
	89	+	return &blockedSnatchedQuitTransport{
	90	+	recv: make(chan ports.Frame, 1),
	91	+	sendStarted: make(chan struct{}),
	92	+	sendDone: make(chan struct{}),
	93	+	closed: make(chan struct{}),
	94	+	}
	95	+	}
	96	+
	97	+	func (t *blockedSnatchedQuitTransport) Send(frame ports.Frame) error {
	98	+	if frame.Type != ports.MsgDetached {
	99	+	return nil
	100	+	}
	101	+	t.sendOnce.Do(func() { close(t.sendStarted) })
	102	+	<-t.closed
	103	+	close(t.sendDone)
	104	+	return io.ErrClosedPipe
	105	+	}
	106	+
	107	+	func (t *blockedSnatchedQuitTransport) Recv() (ports.Frame, error) {
	108	+	select {
	109	+	case frame := <-t.recv:
	110	+	return frame, nil
	111	+	case <-t.closed:
	112	+	return ports.Frame{}, io.ErrClosedPipe
	113	+	}
	114	+	}
	115	+
	116	+	func (t *blockedSnatchedQuitTransport) Close() error {
	117	+	t.mu.Lock()
	118	+	t.isClosed = true
	119	+	t.mu.Unlock()
	120	+	t.closeOnce.Do(func() { close(t.closed) })
	121	+	return nil
	122	+	}
	123	+
	124	+	func (t *blockedSnatchedQuitTransport) Closed() bool {
	125	+	t.mu.Lock()
	126	+	defer t.mu.Unlock()
	127	+	return t.isClosed
	128	+	}
	129	+
	130	+	func newSnatchedInputFixture(t testing.T, clock ports.Clock) (Daemon, session, attachedClient, attachedClient, snatchedCloseSignalTransport) {
	131	+	t.Helper()
	132	+	d := newTestDaemon(t, nil, clock)
	133	+	waitingTransport := newSnatchedCloseSignalTransport()
	134	+	waiting := &attachedClient{
	135	+	tr: waitingTransport, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24},
	136	+	}
	137	+	active := &attachedClient{
	138	+	tr: &closeTrackingTransport{}, output: newOutputStateStream(), size: waiting.size,
	139	+	}
	140	+	waiting.initOverlays()
	141	+	active.initOverlays()
	142	+	sess := &session{
	143	+	id: "strict-input", client: active, snatched: map[*attachedClient]struct{}{waiting: {}},
	144	+	}
	145	+	waiting.setSession(sess)
	146	+	active.setSession(sess)
	147	+	d.sessions[sess.id] = sess
	148	+	d.attachCoordinator(sess, nil, active, true)
	149	+	return d, sess, waiting, active, waitingTransport
	150	+	}
	151	+
	152	+	func sendSnatchedInput(d Daemon, sess session, ac *attachedClient, data string) bool {
	153	+	tr := ac.transport()
	154	+	token := sess.attachmentToken(ac, tr)
	155	+	return d.handleSnatchedClientFrame(token, ports.Frame{
	156	+	Type: ports.MsgInput,
	157	+	Payload: ports.MarshalInput(ports.Input{Data: []byte(data)}),
	158	+	})
	159	+	}
	160	+
	161	+	func TestSnatchedInputStrictActions(t *testing.T) {
	162	+	tests := []struct {
	163	+	name string
	164	+	chunks []string
	165	+	fireEscape bool
	166	+	wantRole attachmentRole
	167	+	wantClosed bool
	168	+	}{
	169	+	{name: "lowercase reclaim", chunks: []string{"r"}, wantRole: attachmentActive},
	170	+	{name: "uppercase reclaim", chunks: []string{"R"}, wantRole: attachmentActive},
	171	+	{name: "lowercase quit", chunks: []string{"q"}, wantRole: attachmentDetached, wantClosed: true},
	172	+	{name: "uppercase quit", chunks: []string{"Q"}, wantRole: attachmentDetached, wantClosed: true},
	173	+	{name: "standalone escape quits after delay", chunks: []string{"\x1b"}, fireEscape: true, wantRole: attachmentDetached, wantClosed: true},
	174	+	{name: "coalesced reclaim ignored", chunks: []string{"rr"}, wantRole: attachmentSnatched},
	175	+	{name: "paste containing quit ignored", chunks: []string{"paste q"}, wantRole: attachmentSnatched},
	176	+	{name: "complete CSI ignored", chunks: []string{"\x1b[r"}, wantRole: attachmentSnatched},
	177	+	{name: "complete SS3 ignored", chunks: []string{"\x1bOR"}, wantRole: attachmentSnatched},
	178	+	{name: "complete sequence tail ignored", chunks: []string{"\x1b[Aq"}, wantRole: attachmentSnatched},
	179	+	{name: "split CSI final reclaim ignored", chunks: []string{"\x1b[", "r"}, wantRole: attachmentSnatched},
	180	+	{name: "split SS3 final reclaim ignored", chunks: []string{"\x1bO", "R"}, wantRole: attachmentSnatched},
	181	+	{name: "split sequence frame tail ignored", chunks: []string{"\x1b[", "Aq"}, wantRole: attachmentSnatched},
	182	+	{name: "escape followed by action ignored as alt chord", chunks: []string{"\x1b", "q"}, wantRole: attachmentSnatched},
	183	+	{name: "CSI prefix remains inert", chunks: []string{"\x1b[", "12;"}, wantRole: attachmentSnatched},
	184	+	{name: "overflow drain consumes final and frame tail", chunks: []string{"\x1b[", strings.Repeat(";", 80), "rQ"}, wantRole: attachmentSnatched},
	185	+	}
	186	+
	187	+	for _, tt := range tests {
	188	+	t.Run(tt.name, func(t *testing.T) {
	189	+	clock := newSnatchedInputClock()
	190	+	d, sess, waiting, active, waitingTransport := newSnatchedInputFixture(t, clock)
	191	+	for _, chunk := range tt.chunks {
	192	+	sendSnatchedInput(d, sess, waiting, chunk)
	193	+	}
	194	+	if tt.fireEscape {
	195	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(waiting), "escape acted before its delay")
	196	+	require.False(t, waitingTransport.Closed(), "escape closed the transport before its delay")
	197	+	timer := awaitTestValue(t, clock.timers, "standalone escape did not create a timer")
	198	+	timer.Fire()
	199	+	awaitTestCompletion(t, waitingTransport.closedCh, "standalone escape did not quit")
	200	+	}
	201	+	d.attachmentCleanupWg.Wait()
	202	+
	203	+	require.Equal(t, tt.wantRole, sess.attachmentRole(waiting))
	204	+	require.Equal(t, tt.wantClosed, waitingTransport.Closed())
	205	+	if tt.wantRole == attachmentActive {
	206	+	require.Same(t, waiting, sess.client)
	207	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(active))
	208	+	} else {
	209	+	require.Same(t, active, sess.client)
	210	+	}
	211	+	})
	212	+	}
	213	+	}
	214	+
	215	+	func TestBlockedSnatchedQuitDeadlineRetiresOnlyExactWaiter(t *testing.T) {
	216	+	clock := &signalClock{timers: make(chan *signalTimer, 2)}
	217	+	waitingTransport := newBlockedSnatchedQuitTransport()
	218	+	defer waitingTransport.Close()
	219	+	activeTransport := &closeTrackingTransport{}
	220	+	d, sess, waiting, active := newAtomicReclaimFixture(t, clock, waitingTransport, activeTransport)
	221	+
	222	+	otherTransport := &closeTrackingTransport{}
	223	+	other := &attachedClient{tr: otherTransport, output: newOutputStateStream(), size: waiting.size}
	224	+	other.initOverlays()
	225	+	other.setSession(sess)
	226	+	sess.mu.Lock()
	227	+	sess.addSnatchedLocked(other)
	228	+	sess.mu.Unlock()
	229	+	otherToken := sess.attachmentToken(other, otherTransport)
	230	+	require.Equal(t, attachmentSnatched, otherToken.role)
	231	+
	232	+	ownerLease := sess.renderCoordinator().attachmentLease(active)
	233	+	require.NotNil(t, ownerLease)
	234	+	activeGeneration := active.roleGeneration.Load()
	235	+	otherGeneration := other.roleGeneration.Load()
	236	+
	237	+	loopDone := make(chan struct{})
	238	+	go func() {
	239	+	d.runConnLoop(waiting)
	240	+	close(loopDone)
	241	+	}()
	242	+	waitingTransport.recv <- ports.Frame{
	243	+	Type: ports.MsgInput,
	244	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'q'}}),
	245	+	}
	246	+	awaitTestCompletion(t, waitingTransport.sendStarted, "snatched quit did not attempt its detach acknowledgement")
	247	+	select {
	248	+	case <-loopDone:
	249	+	t.Fatal("snatched quit returned before its acknowledgement deadline")
	250	+	default:
	251	+	}
	252	+
	253	+	timer := awaitTestValue(t, clock.timers, "snatched quit did not install its acknowledgement deadline")
	254	+	require.Equal(t, detachNotifyTimeout, timer.duration)
	255	+	timer.ch <- time.Time{}
	256	+	awaitTestCompletion(t, waitingTransport.sendDone, "closing the captured link did not release its blocked send")
	257	+	awaitTestCompletion(t, loopDone, "snatched connection parser did not retire after the acknowledgement deadline")
	258	+
	259	+	require.True(t, waitingTransport.Closed())
	260	+	requireRoleGateRetired(t, waiting)
	261	+	require.Equal(t, attachmentDetached, sess.attachmentRole(waiting))
	262	+	require.Nil(t, waiting.currentSession())
	263	+	require.Nil(t, waiting.transport())
	264	+	require.Same(t, active, sess.client)
	265	+	require.Equal(t, attachmentActive, sess.attachmentRole(active))
	266	+	require.Same(t, ownerLease, sess.renderCoordinator().attachmentLease(active))
	267	+	require.Equal(t, activeGeneration, active.roleGeneration.Load())
	268	+	require.False(t, activeTransport.Closed())
	269	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(other))
	270	+	require.Equal(t, otherGeneration, other.roleGeneration.Load())
	271	+	require.Same(t, otherTransport, other.transport())
	272	+	require.False(t, otherTransport.Closed())
	273	+	}
	274	+
	275	+	func TestBlockedSnatchedQuitCleanupRejectsReclaimedTransportGeneration(t *testing.T) {
	276	+	clock := &signalClock{timers: make(chan *signalTimer, 2)}
	277	+	staleTransport := newBlockedSnatchedQuitTransport()
	278	+	defer staleTransport.Close()
	279	+	ownerTransport := &closeTrackingTransport{}
	280	+	d, sess, waiting, owner := newAtomicReclaimFixture(t, clock, staleTransport, ownerTransport)
	281	+
	282	+	ticketEnded := make(chan struct{})
	283	+	releaseQuit := make(chan struct{})
	284	+	d.afterActionRoleEffectEnded = func(action string) {
	285	+	if action == "quit-snatched" {
	286	+	close(ticketEnded)
	287	+	<-releaseQuit
	288	+	}
	289	+	}
	290	+	loopDone := make(chan struct{})
	291	+	go func() {
	292	+	d.runConnLoop(waiting)
	293	+	close(loopDone)
	294	+	}()
	295	+	staleTransport.recv <- ports.Frame{
	296	+	Type: ports.MsgInput,
	297	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'Q'}}),
	298	+	}
	299	+	awaitTestCompletion(t, staleTransport.sendStarted, "snatched quit did not block in its captured transport")
	300	+	timer := awaitTestValue(t, clock.timers, "snatched quit did not install its acknowledgement deadline")
	301	+	timer.ch <- time.Time{}
	302	+	awaitTestCompletion(t, staleTransport.sendDone, "deadline close did not retire the captured send")
	303	+	awaitTestCompletion(t, ticketEnded, "snatched quit did not end its role ticket before cleanup")
	304	+
	305	+	freshTransport := &closeTrackingTransport{}
	306	+	waiting.replaceTransport(freshTransport)
	307	+	reclaimed, err := d.transitionAttachment(attachmentTransitionRequest{
	308	+	target: sess, next: waiting, expectedRole: attachmentSnatched, targetRole: attachmentActive,
	309	+	expectedTransport: waiting.transportSnapshot(), ready: true,
	310	+	})
	311	+	require.NoError(t, err)
	312	+	d.deferAttachmentTransitionCleanups(reclaimed)
	313	+	freshGeneration := reclaimed.published.generation
	314	+	freshLease := reclaimed.published.lease
	315	+
	316	+	close(releaseQuit)
	317	+	awaitTestCompletion(t, loopDone, "stale quit parser did not retire after reclaim")
	318	+	d.attachmentCleanupWg.Wait()
	319	+
	320	+	require.True(t, staleTransport.Closed())
	321	+	requireRoleGateRetired(t, waiting)
	322	+	require.Same(t, waiting, sess.client)
	323	+	require.Equal(t, attachmentActive, sess.attachmentRole(waiting))
	324	+	require.Equal(t, freshGeneration, waiting.roleGeneration.Load())
	325	+	require.Same(t, freshTransport, waiting.transport())
	326	+	require.False(t, freshTransport.Closed(), "stale quit cleanup closed the reclaimed transport")
	327	+	require.Same(t, freshLease, sess.renderCoordinator().attachmentLease(waiting))
	328	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(owner))
	329	+	require.False(t, ownerTransport.Closed(), "stale quit cleanup affected the displaced owner")
	330	+	}
	331	+
	332	+	func TestSnatchedInputTimerCannotAffectNewRoleOrTransport(t *testing.T) {
	333	+	clock := newSnatchedInputClock()
	334	+	d, sess, waiting, _, oldTransport := newSnatchedInputFixture(t, clock)
	335	+
	336	+	sendSnatchedInput(d, sess, waiting, "\x1b")
	337	+	timer := awaitTestValue(t, clock.timers, "standalone escape did not create a timer")
	338	+	accepted := make(chan struct{})
	339	+	release := make(chan struct{})
	340	+	attempted := make(chan bool, 1)
	341	+	d.afterSnatchedEscapeAccepted = func() {
	342	+	close(accepted)
	343	+	<-release
	344	+	}
	345	+	d.afterSnatchedEscapeAttempt = func(admitted bool) { attempted <- admitted }
	346	+	timer.Fire()
	347	+	awaitTestCompletion(t, accepted, "escape timer callback did not consume parser state")
	348	+
	349	+	result, err := d.transitionAttachment(attachmentTransitionRequest{
	350	+	target: sess, next: waiting, expectedRole: attachmentSnatched, targetRole: attachmentActive,
	351	+	expectedTransport: waiting.transportSnapshot(), ready: true,
	352	+	})
	353	+	require.NoError(t, err)
	354	+	d.deferAttachmentTransitionCleanups(result)
	355	+	newTransport := newSnatchedCloseSignalTransport()
	356	+	waiting.replaceTransport(newTransport)
	357	+	close(release)
	358	+	require.False(t, awaitTestValue(t, attempted, "escape callback did not revalidate its stale role"))
	359	+	d.attachmentCleanupWg.Wait()
	360	+
	361	+	require.Equal(t, attachmentActive, sess.attachmentRole(waiting))
	362	+	require.Same(t, newTransport, waiting.transport())
	363	+	require.False(t, newTransport.Closed(), "stale timer closed the replacement transport")
	364	+	require.False(t, oldTransport.Closed(), "stale timer closed the old transport after role activation")
	365	+	}

@@ -0,0 +1,365 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"io"
	6	+	"strings"
	7	+	"sync"
	8	+	"testing"
	9	+	"time"
	10	+
	11	+	"github.com/stretchr/testify/require"
	12	+
	13	+	"github.com/bnema/vev/internal/domain"
	14	+	"github.com/bnema/vev/internal/ports"
	15	+	"github.com/bnema/vev/internal/usecase/keys"
	16	+	)
	17	+
	18	+	type snatchedInputTimer struct {
	19	+	ch chan time.Time
	20	+	}
	21	+
	22	+	func newSnatchedInputTimer() *snatchedInputTimer {
	23	+	return &snatchedInputTimer{ch: make(chan time.Time, 1)}
	24	+	}
	25	+
	26	+	func (t *snatchedInputTimer) C() <-chan time.Time { return t.ch }
	27	+	func (*snatchedInputTimer) Reset(time.Duration) bool { return false }
	28	+	func (*snatchedInputTimer) Stop() bool { return true }
	29	+	func (t *snatchedInputTimer) Fire() { t.ch <- time.Time{} }
	30	+
	31	+	type snatchedInputClock struct {
	32	+	timers chan *snatchedInputTimer
	33	+	}
	34	+
	35	+	func newSnatchedInputClock() *snatchedInputClock {
	36	+	return &snatchedInputClock{timers: make(chan *snatchedInputTimer, 8)}
	37	+	}
	38	+
	39	+	func (*snatchedInputClock) Now() time.Time { return time.Time{} }
	40	+	func (c *snatchedInputClock) NewTimer(delay time.Duration) ports.Timer {
	41	+	if delay != keys.ESCDelay {
	42	+	return stubTimer{}
	43	+	}
	44	+	timer := newSnatchedInputTimer()
	45	+	c.timers <- timer
	46	+	return timer
	47	+	}
	48	+
	49	+	type snatchedCloseSignalTransport struct {
	50	+	mu sync.Mutex
	51	+	closed bool
	52	+	closedCh chan struct{}
	53	+	closeOnce sync.Once
	54	+	}
	55	+
	56	+	func newSnatchedCloseSignalTransport() *snatchedCloseSignalTransport {
	57	+	return &snatchedCloseSignalTransport{closedCh: make(chan struct{})}
	58	+	}
	59	+
	60	+	func (*snatchedCloseSignalTransport) Send(ports.Frame) error { return nil }
	61	+	func (*snatchedCloseSignalTransport) Recv() (ports.Frame, error) {
	62	+	return ports.Frame{}, errors.New("closed")
	63	+	}
	64	+	func (t *snatchedCloseSignalTransport) Close() error {
	65	+	t.mu.Lock()
	66	+	t.closed = true
	67	+	t.mu.Unlock()
	68	+	t.closeOnce.Do(func() { close(t.closedCh) })
	69	+	return nil
	70	+	}
	71	+	func (t *snatchedCloseSignalTransport) Closed() bool {
	72	+	t.mu.Lock()
	73	+	defer t.mu.Unlock()
	74	+	return t.closed
	75	+	}
	76	+
	77	+	type blockedSnatchedQuitTransport struct {
	78	+	recv chan ports.Frame
	79	+	sendStarted chan struct{}
	80	+	sendDone chan struct{}
	81	+	closed chan struct{}
	82	+	closeOnce sync.Once
	83	+	sendOnce sync.Once
	84	+	mu sync.Mutex
	85	+	isClosed bool
	86	+	}
	87	+
	88	+	func newBlockedSnatchedQuitTransport() *blockedSnatchedQuitTransport {
	89	+	return &blockedSnatchedQuitTransport{
	90	+	recv: make(chan ports.Frame, 1),
	91	+	sendStarted: make(chan struct{}),
	92	+	sendDone: make(chan struct{}),
	93	+	closed: make(chan struct{}),
	94	+	}
	95	+	}
	96	+
	97	+	func (t *blockedSnatchedQuitTransport) Send(frame ports.Frame) error {
	98	+	if frame.Type != ports.MsgDetached {
	99	+	return nil
	100	+	}
	101	+	t.sendOnce.Do(func() { close(t.sendStarted) })
	102	+	<-t.closed
	103	+	close(t.sendDone)
	104	+	return io.ErrClosedPipe
	105	+	}
	106	+
	107	+	func (t *blockedSnatchedQuitTransport) Recv() (ports.Frame, error) {
	108	+	select {
	109	+	case frame := <-t.recv:
	110	+	return frame, nil
	111	+	case <-t.closed:
	112	+	return ports.Frame{}, io.ErrClosedPipe
	113	+	}
	114	+	}
	115	+
	116	+	func (t *blockedSnatchedQuitTransport) Close() error {
	117	+	t.mu.Lock()
	118	+	t.isClosed = true
	119	+	t.mu.Unlock()
	120	+	t.closeOnce.Do(func() { close(t.closed) })
	121	+	return nil
	122	+	}
	123	+
	124	+	func (t *blockedSnatchedQuitTransport) Closed() bool {
	125	+	t.mu.Lock()
	126	+	defer t.mu.Unlock()
	127	+	return t.isClosed
	128	+	}
	129	+
	130	+	func newSnatchedInputFixture(t testing.T, clock ports.Clock) (Daemon, session, attachedClient, attachedClient, snatchedCloseSignalTransport) {
	131	+	t.Helper()
	132	+	d := newTestDaemon(t, nil, clock)
	133	+	waitingTransport := newSnatchedCloseSignalTransport()
	134	+	waiting := &attachedClient{
	135	+	tr: waitingTransport, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24},
	136	+	}
	137	+	active := &attachedClient{
	138	+	tr: &closeTrackingTransport{}, output: newOutputStateStream(), size: waiting.size,
	139	+	}
	140	+	waiting.initOverlays()
	141	+	active.initOverlays()
	142	+	sess := &session{
	143	+	id: "strict-input", client: active, snatched: map[*attachedClient]struct{}{waiting: {}},
	144	+	}
	145	+	waiting.setSession(sess)
	146	+	active.setSession(sess)
	147	+	d.sessions[sess.id] = sess
	148	+	d.attachCoordinator(sess, nil, active, true)
	149	+	return d, sess, waiting, active, waitingTransport
	150	+	}
	151	+
	152	+	func sendSnatchedInput(d Daemon, sess session, ac *attachedClient, data string) bool {
	153	+	tr := ac.transport()
	154	+	token := sess.attachmentToken(ac, tr)
	155	+	return d.handleSnatchedClientFrame(token, ports.Frame{
	156	+	Type: ports.MsgInput,
	157	+	Payload: ports.MarshalInput(ports.Input{Data: []byte(data)}),
	158	+	})
	159	+	}
	160	+
	161	+	func TestSnatchedInputStrictActions(t *testing.T) {
	162	+	tests := []struct {
	163	+	name string
	164	+	chunks []string
	165	+	fireEscape bool
	166	+	wantRole attachmentRole
	167	+	wantClosed bool
	168	+	}{
	169	+	{name: "lowercase reclaim", chunks: []string{"r"}, wantRole: attachmentActive},
	170	+	{name: "uppercase reclaim", chunks: []string{"R"}, wantRole: attachmentActive},
	171	+	{name: "lowercase quit", chunks: []string{"q"}, wantRole: attachmentDetached, wantClosed: true},
	172	+	{name: "uppercase quit", chunks: []string{"Q"}, wantRole: attachmentDetached, wantClosed: true},
	173	+	{name: "standalone escape quits after delay", chunks: []string{"\x1b"}, fireEscape: true, wantRole: attachmentDetached, wantClosed: true},
	174	+	{name: "coalesced reclaim ignored", chunks: []string{"rr"}, wantRole: attachmentSnatched},
	175	+	{name: "paste containing quit ignored", chunks: []string{"paste q"}, wantRole: attachmentSnatched},
	176	+	{name: "complete CSI ignored", chunks: []string{"\x1b[r"}, wantRole: attachmentSnatched},
	177	+	{name: "complete SS3 ignored", chunks: []string{"\x1bOR"}, wantRole: attachmentSnatched},
	178	+	{name: "complete sequence tail ignored", chunks: []string{"\x1b[Aq"}, wantRole: attachmentSnatched},
	179	+	{name: "split CSI final reclaim ignored", chunks: []string{"\x1b[", "r"}, wantRole: attachmentSnatched},
	180	+	{name: "split SS3 final reclaim ignored", chunks: []string{"\x1bO", "R"}, wantRole: attachmentSnatched},
	181	+	{name: "split sequence frame tail ignored", chunks: []string{"\x1b[", "Aq"}, wantRole: attachmentSnatched},
	182	+	{name: "escape followed by action ignored as alt chord", chunks: []string{"\x1b", "q"}, wantRole: attachmentSnatched},
	183	+	{name: "CSI prefix remains inert", chunks: []string{"\x1b[", "12;"}, wantRole: attachmentSnatched},
	184	+	{name: "overflow drain consumes final and frame tail", chunks: []string{"\x1b[", strings.Repeat(";", 80), "rQ"}, wantRole: attachmentSnatched},
	185	+	}
	186	+
	187	+	for _, tt := range tests {
	188	+	t.Run(tt.name, func(t *testing.T) {
	189	+	clock := newSnatchedInputClock()
	190	+	d, sess, waiting, active, waitingTransport := newSnatchedInputFixture(t, clock)
	191	+	for _, chunk := range tt.chunks {
	192	+	sendSnatchedInput(d, sess, waiting, chunk)
	193	+	}
	194	+	if tt.fireEscape {
	195	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(waiting), "escape acted before its delay")
	196	+	require.False(t, waitingTransport.Closed(), "escape closed the transport before its delay")
	197	+	timer := awaitTestValue(t, clock.timers, "standalone escape did not create a timer")
	198	+	timer.Fire()
	199	+	awaitTestCompletion(t, waitingTransport.closedCh, "standalone escape did not quit")
	200	+	}
	201	+	d.attachmentCleanupWg.Wait()
	202	+
	203	+	require.Equal(t, tt.wantRole, sess.attachmentRole(waiting))
	204	+	require.Equal(t, tt.wantClosed, waitingTransport.Closed())
	205	+	if tt.wantRole == attachmentActive {
	206	+	require.Same(t, waiting, sess.client)
	207	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(active))
	208	+	} else {
	209	+	require.Same(t, active, sess.client)
	210	+	}
	211	+	})
	212	+	}
	213	+	}
	214	+
	215	+	func TestBlockedSnatchedQuitDeadlineRetiresOnlyExactWaiter(t *testing.T) {
	216	+	clock := &signalClock{timers: make(chan *signalTimer, 2)}
	217	+	waitingTransport := newBlockedSnatchedQuitTransport()
	218	+	defer waitingTransport.Close()
	219	+	activeTransport := &closeTrackingTransport{}
	220	+	d, sess, waiting, active := newAtomicReclaimFixture(t, clock, waitingTransport, activeTransport)
	221	+
	222	+	otherTransport := &closeTrackingTransport{}
	223	+	other := &attachedClient{tr: otherTransport, output: newOutputStateStream(), size: waiting.size}
	224	+	other.initOverlays()
	225	+	other.setSession(sess)
	226	+	sess.mu.Lock()
	227	+	sess.addSnatchedLocked(other)
	228	+	sess.mu.Unlock()
	229	+	otherToken := sess.attachmentToken(other, otherTransport)
	230	+	require.Equal(t, attachmentSnatched, otherToken.role)
	231	+
	232	+	ownerLease := sess.renderCoordinator().attachmentLease(active)
	233	+	require.NotNil(t, ownerLease)
	234	+	activeGeneration := active.roleGeneration.Load()
	235	+	otherGeneration := other.roleGeneration.Load()
	236	+
	237	+	loopDone := make(chan struct{})
	238	+	go func() {
	239	+	d.runConnLoop(waiting)
	240	+	close(loopDone)
	241	+	}()
	242	+	waitingTransport.recv <- ports.Frame{
	243	+	Type: ports.MsgInput,
	244	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'q'}}),
	245	+	}
	246	+	awaitTestCompletion(t, waitingTransport.sendStarted, "snatched quit did not attempt its detach acknowledgement")
	247	+	select {
	248	+	case <-loopDone:
	249	+	t.Fatal("snatched quit returned before its acknowledgement deadline")
	250	+	default:
	251	+	}
	252	+
	253	+	timer := awaitTestValue(t, clock.timers, "snatched quit did not install its acknowledgement deadline")
	254	+	require.Equal(t, detachNotifyTimeout, timer.duration)
	255	+	timer.ch <- time.Time{}
	256	+	awaitTestCompletion(t, waitingTransport.sendDone, "closing the captured link did not release its blocked send")
	257	+	awaitTestCompletion(t, loopDone, "snatched connection parser did not retire after the acknowledgement deadline")
	258	+
	259	+	require.True(t, waitingTransport.Closed())
	260	+	requireRoleGateRetired(t, waiting)
	261	+	require.Equal(t, attachmentDetached, sess.attachmentRole(waiting))
	262	+	require.Nil(t, waiting.currentSession())
	263	+	require.Nil(t, waiting.transport())
	264	+	require.Same(t, active, sess.client)
	265	+	require.Equal(t, attachmentActive, sess.attachmentRole(active))
	266	+	require.Same(t, ownerLease, sess.renderCoordinator().attachmentLease(active))
	267	+	require.Equal(t, activeGeneration, active.roleGeneration.Load())
	268	+	require.False(t, activeTransport.Closed())
	269	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(other))
	270	+	require.Equal(t, otherGeneration, other.roleGeneration.Load())
	271	+	require.Same(t, otherTransport, other.transport())
	272	+	require.False(t, otherTransport.Closed())
	273	+	}
	274	+
	275	+	func TestBlockedSnatchedQuitCleanupRejectsReclaimedTransportGeneration(t *testing.T) {
	276	+	clock := &signalClock{timers: make(chan *signalTimer, 2)}
	277	+	staleTransport := newBlockedSnatchedQuitTransport()
	278	+	defer staleTransport.Close()
	279	+	ownerTransport := &closeTrackingTransport{}
	280	+	d, sess, waiting, owner := newAtomicReclaimFixture(t, clock, staleTransport, ownerTransport)
	281	+
	282	+	ticketEnded := make(chan struct{})
	283	+	releaseQuit := make(chan struct{})
	284	+	d.afterActionRoleEffectEnded = func(action string) {
	285	+	if action == "quit-snatched" {
	286	+	close(ticketEnded)
	287	+	<-releaseQuit
	288	+	}
	289	+	}
	290	+	loopDone := make(chan struct{})
	291	+	go func() {
	292	+	d.runConnLoop(waiting)
	293	+	close(loopDone)
	294	+	}()
	295	+	staleTransport.recv <- ports.Frame{
	296	+	Type: ports.MsgInput,
	297	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'Q'}}),
	298	+	}
	299	+	awaitTestCompletion(t, staleTransport.sendStarted, "snatched quit did not block in its captured transport")
	300	+	timer := awaitTestValue(t, clock.timers, "snatched quit did not install its acknowledgement deadline")
	301	+	timer.ch <- time.Time{}
	302	+	awaitTestCompletion(t, staleTransport.sendDone, "deadline close did not retire the captured send")
	303	+	awaitTestCompletion(t, ticketEnded, "snatched quit did not end its role ticket before cleanup")
	304	+
	305	+	freshTransport := &closeTrackingTransport{}
	306	+	waiting.replaceTransport(freshTransport)
	307	+	reclaimed, err := d.transitionAttachment(attachmentTransitionRequest{
	308	+	target: sess, next: waiting, expectedRole: attachmentSnatched, targetRole: attachmentActive,
	309	+	expectedTransport: waiting.transportSnapshot(), ready: true,
	310	+	})
	311	+	require.NoError(t, err)
	312	+	d.deferAttachmentTransitionCleanups(reclaimed)
	313	+	freshGeneration := reclaimed.published.generation
	314	+	freshLease := reclaimed.published.lease
	315	+
	316	+	close(releaseQuit)
	317	+	awaitTestCompletion(t, loopDone, "stale quit parser did not retire after reclaim")
	318	+	d.attachmentCleanupWg.Wait()
	319	+
	320	+	require.True(t, staleTransport.Closed())
	321	+	requireRoleGateRetired(t, waiting)
	322	+	require.Same(t, waiting, sess.client)
	323	+	require.Equal(t, attachmentActive, sess.attachmentRole(waiting))
	324	+	require.Equal(t, freshGeneration, waiting.roleGeneration.Load())
	325	+	require.Same(t, freshTransport, waiting.transport())
	326	+	require.False(t, freshTransport.Closed(), "stale quit cleanup closed the reclaimed transport")
	327	+	require.Same(t, freshLease, sess.renderCoordinator().attachmentLease(waiting))
	328	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(owner))
	329	+	require.False(t, ownerTransport.Closed(), "stale quit cleanup affected the displaced owner")
	330	+	}
	331	+
	332	+	func TestSnatchedInputTimerCannotAffectNewRoleOrTransport(t *testing.T) {
	333	+	clock := newSnatchedInputClock()
	334	+	d, sess, waiting, _, oldTransport := newSnatchedInputFixture(t, clock)
	335	+
	336	+	sendSnatchedInput(d, sess, waiting, "\x1b")
	337	+	timer := awaitTestValue(t, clock.timers, "standalone escape did not create a timer")
	338	+	accepted := make(chan struct{})
	339	+	release := make(chan struct{})
	340	+	attempted := make(chan bool, 1)
	341	+	d.afterSnatchedEscapeAccepted = func() {
	342	+	close(accepted)
	343	+	<-release
	344	+	}
	345	+	d.afterSnatchedEscapeAttempt = func(admitted bool) { attempted <- admitted }
	346	+	timer.Fire()
	347	+	awaitTestCompletion(t, accepted, "escape timer callback did not consume parser state")
	348	+
	349	+	result, err := d.transitionAttachment(attachmentTransitionRequest{
	350	+	target: sess, next: waiting, expectedRole: attachmentSnatched, targetRole: attachmentActive,
	351	+	expectedTransport: waiting.transportSnapshot(), ready: true,
	352	+	})
	353	+	require.NoError(t, err)
	354	+	d.deferAttachmentTransitionCleanups(result)
	355	+	newTransport := newSnatchedCloseSignalTransport()
	356	+	waiting.replaceTransport(newTransport)
	357	+	close(release)
	358	+	require.False(t, awaitTestValue(t, attempted, "escape callback did not revalidate its stale role"))
	359	+	d.attachmentCleanupWg.Wait()
	360	+
	361	+	require.Equal(t, attachmentActive, sess.attachmentRole(waiting))
	362	+	require.Same(t, newTransport, waiting.transport())
	363	+	require.False(t, newTransport.Closed(), "stale timer closed the replacement transport")
	364	+	require.False(t, oldTransport.Closed(), "stale timer closed the old transport after role activation")
	365	+	}

A internal/usecase/daemon/snatched_reclaim_test.go +443 -0

@@ -0,0 +1,443 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	themeui "github.com/bnema/vev/internal/usecase/theme"
	14	+	"github.com/bnema/vev/pkg/renderer"
	15	+	)
	16	+
	17	+	type failingReclaimTransport struct {
	18	+	mu sync.Mutex
	19	+	closed bool
	20	+	}
	21	+
	22	+	func (*failingReclaimTransport) Send(ports.Frame) error { return errors.New("send failed") }
	23	+	func (*failingReclaimTransport) Recv() (ports.Frame, error) {
	24	+	return ports.Frame{}, errors.New("closed")
	25	+	}
	26	+	func (t *failingReclaimTransport) Close() error {
	27	+	t.mu.Lock()
	28	+	t.closed = true
	29	+	t.mu.Unlock()
	30	+	return nil
	31	+	}
	32	+	func (t *failingReclaimTransport) Closed() bool {
	33	+	t.mu.Lock()
	34	+	defer t.mu.Unlock()
	35	+	return t.closed
	36	+	}
	37	+
	38	+	func newAtomicReclaimFixture(t testing.T, clock ports.Clock, waitingTransport, activeTransport ports.Transport) (Daemon, session, attachedClient, *attachedClient) {
	39	+	t.Helper()
	40	+	d := newTestDaemon(t, nil, clock)
	41	+	waiting := &attachedClient{tr: waitingTransport, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}}
	42	+	active := &attachedClient{tr: activeTransport, output: newOutputStateStream(), size: waiting.size}
	43	+	waiting.initOverlays()
	44	+	active.initOverlays()
	45	+	sess := &session{
	46	+	id: "atomic-reclaim", client: active,
	47	+	snatched: map[*attachedClient]struct{}{waiting: {}},
	48	+	}
	49	+	waiting.setSession(sess)
	50	+	active.setSession(sess)
	51	+	d.sessions[sess.id] = sess
	52	+	d.attachCoordinator(sess, nil, active, true)
	53	+	return d, sess, waiting, active
	54	+	}
	55	+
	56	+	func requireSingleOwnerAndLease(t testing.T, sess session, clients ...*attachedClient) {
	57	+	t.Helper()
	58	+	sess.mu.Lock()
	59	+	owner := sess.client
	60	+	snatched := make(map[*attachedClient]struct{}, len(sess.snatched))
	61	+	for ac := range sess.snatched {
	62	+	snatched[ac] = struct{}{}
	63	+	}
	64	+	sess.mu.Unlock()
	65	+	require.NotNil(t, owner)
	66	+
	67	+	activeCount := 0
	68	+	for _, ac := range clients {
	69	+	if ac == owner {
	70	+	activeCount++
	71	+	continue
	72	+	}
	73	+	_, waiting := snatched[ac]
	74	+	require.True(t, waiting, "every non-owner must remain snatched")
	75	+	}
	76	+	require.Equal(t, 1, activeCount)
	77	+
	78	+	rc := sess.renderCoordinator()
	79	+	rc.mu.Lock()
	80	+	lease := rc.lease
	81	+	rc.mu.Unlock()
	82	+	require.NotNil(t, lease)
	83	+	require.True(t, lease.active)
	84	+	require.Same(t, owner, lease.attachment)
	85	+	}
	86	+
	87	+	func TestReclaimSnatchedSwapsOwnerLeaseRebasesAndAppliesTheme(t *testing.T) {
	88	+	d, sess, requester, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	89	+	requesterTransport := &closeTrackingTransport{}
	90	+	requester.replaceTransport(requesterTransport)
	91	+	d.attachCoordinator(sess, nil, requester, true)
	92	+
	93	+	active := &attachedClient{
	94	+	tr: &closeTrackingTransport{},
	95	+	output: newOutputStateStream(),
	96	+	size: requester.size,
	97	+	}
	98	+	active.initOverlays()
	99	+	displaced, err := d.transitionAttachment(attachmentTransitionRequest{
	100	+	target: sess, next: active,
	101	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	102	+	expectedTransport: active.transportSnapshot(), ready: true,
	103	+	})
	104	+	require.NoError(t, err)
	105	+	d.deferAttachmentTransitionCleanups(displaced)
	106	+	d.attachmentCleanupWg.Wait()
	107	+
	108	+	requester.sendMu.Lock()
	109	+	beforeReclaimState := requester.output.next
	110	+	requester.sendMu.Unlock()
	111	+	requesterTheme := themeui.Theme{
	112	+	Foreground: renderer.RGB{R: 11, G: 22, B: 33},
	113	+	Background: renderer.RGB{R: 44, G: 55, B: 66},
	114	+	HasFG: true, HasBG: true, Known: true, TrueColor: true,
	115	+	}
	116	+	requester.setClientTheme(requesterTheme)
	117	+
	118	+	token := sess.attachmentToken(requester, requesterTransport)
	119	+	require.Equal(t, attachmentSnatched, token.role)
	120	+	require.False(t, d.handleSnatchedClientFrame(token, ports.Frame{
	121	+	Type: ports.MsgInput,
	122	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	123	+	}))
	124	+
	125	+	sess.mu.Lock()
	126	+	require.Same(t, requester, sess.client)
	127	+	_, activeIsSnatched := sess.snatched[active]
	128	+	_, requesterIsSnatched := sess.snatched[requester]
	129	+	sess.mu.Unlock()
	130	+	require.True(t, activeIsSnatched)
	131	+	require.False(t, requesterIsSnatched)
	132	+	rc := sess.renderCoordinator()
	133	+	lease := rc.attachmentLease(requester)
	134	+	require.NotNil(t, lease)
	135	+	require.Nil(t, rc.attachmentLease(active))
	136	+	require.True(t, lease.active)
	137	+
	138	+	requester.sendMu.Lock()
	139	+	require.Greater(t, requester.output.next, beforeReclaimState, "reclaim must emit a reset first paint")
	140	+	require.GreaterOrEqual(t, requester.output.acked, beforeReclaimState, "reclaim must rebase the old panel output chain")
	141	+	requester.sendMu.Unlock()
	142	+	assertSessionDefaultColors(t, sess, requesterTheme.Foreground, requesterTheme.Background)
	143	+
	144	+	d.attachmentCleanupWg.Wait()
	145	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(active))
	146	+	require.False(t, active.transport().(*closeTrackingTransport).Closed())
	147	+	require.Equal(t, domain.Size{Cols: 80, Rows: 24}, requester.size)
	148	+	}
	149	+
	150	+	func TestFailedReclaimBeforeCommitPreservesOwnerAndShowsUnavailable(t *testing.T) {
	151	+	waitingTransport := &closeTrackingTransport{}
	152	+	activeTransport := &closeTrackingTransport{}
	153	+	d, sess, waiting, active := newAtomicReclaimFixture(t, stubClock{}, waitingTransport, activeTransport)
	154	+	token := sess.attachmentToken(waiting, waitingTransport)
	155	+
	156	+	d.mu.Lock()
	157	+	delete(d.sessions, sess.id)
	158	+	d.mu.Unlock()
	159	+	d.handleSnatchedClientFrame(token, ports.Frame{
	160	+	Type: ports.MsgInput,
	161	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	162	+	})
	163	+
	164	+	requireSingleOwnerAndLease(t, sess, waiting, active)
	165	+	require.Same(t, active, sess.client)
	166	+	require.False(t, waitingTransport.Closed())
	167	+	frames := waitingTransport.Sends()
	168	+	require.Len(t, frames, 1)
	169	+	output, err := ports.UnmarshalOutput(frames[0].Payload)
	170	+	require.NoError(t, err)
	171	+	require.Contains(t, string(output.Data), "Session is no longer available.")
	172	+	}
	173	+
	174	+	func TestPostcommitPanelFailureRetiresOnlyDisplacedClient(t *testing.T) {
	175	+	waitingTransport := &closeTrackingTransport{}
	176	+	activeTransport := &failingReclaimTransport{}
	177	+	d, sess, waiting, active := newAtomicReclaimFixture(t, stubClock{}, waitingTransport, activeTransport)
	178	+	token := sess.attachmentToken(waiting, waitingTransport)
	179	+
	180	+	d.handleSnatchedClientFrame(token, ports.Frame{
	181	+	Type: ports.MsgInput,
	182	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	183	+	})
	184	+	d.attachmentCleanupWg.Wait()
	185	+
	186	+	require.Same(t, waiting, sess.client)
	187	+	require.Equal(t, attachmentActive, sess.attachmentRole(waiting))
	188	+	require.Equal(t, attachmentDetached, sess.attachmentRole(active))
	189	+	require.NotNil(t, sess.renderCoordinator().attachmentLease(waiting))
	190	+	require.Nil(t, sess.renderCoordinator().attachmentLease(active))
	191	+	require.False(t, waitingTransport.Closed())
	192	+	require.True(t, activeTransport.Closed())
	193	+	}
	194	+
	195	+	func TestBlockedRequesterActivationDeadlinePreservesOwnerLease(t *testing.T) {
	196	+	clock := &signalClock{timers: make(chan *signalTimer, 2)}
	197	+	waitingTransport := newSnatchedCloseSignalTransport()
	198	+	activeTransport := &closeTrackingTransport{}
	199	+	d, sess, waiting, active := newAtomicReclaimFixture(t, clock, waitingTransport, activeTransport)
	200	+
	201	+	waiting.sendMu.Lock()
	202	+	locked := true
	203	+	defer func() {
	204	+	if locked {
	205	+	waiting.sendMu.Unlock()
	206	+	}
	207	+	}()
	208	+	done := make(chan struct{})
	209	+	go func() {
	210	+	token := sess.attachmentToken(waiting, waitingTransport)
	211	+	d.handleSnatchedClientFrame(token, ports.Frame{
	212	+	Type: ports.MsgInput,
	213	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	214	+	})
	215	+	close(done)
	216	+	}()
	217	+
	218	+	timer := awaitTestValue(t, clock.timers, "reclaim did not install its activation deadline")
	219	+	require.Equal(t, detachNotifyTimeout, timer.duration)
	220	+	timer.ch <- time.Time{}
	221	+	awaitTestCompletion(t, done, "deadline did not release blocked reclaim")
	222	+	require.Same(t, active, sess.client)
	223	+	require.Same(t, active, sess.renderCoordinator().attachmentLease(active).attachment)
	224	+	require.True(t, waitingTransport.Closed())
	225	+
	226	+	waiting.sendMu.Unlock()
	227	+	locked = false
	228	+	d.attachmentCleanupWg.Wait()
	229	+	require.Equal(t, attachmentDetached, sess.attachmentRole(waiting))
	230	+	require.Equal(t, attachmentActive, sess.attachmentRole(active))
	231	+	}
	232	+
	233	+	func TestQuitInvalidatesPausedSnatchedFinalizerWithoutLatePanel(t *testing.T) {
	234	+	waitingTransport := &closeTrackingTransport{}
	235	+	activeTransport := &closeTrackingTransport{}
	236	+	d, sess, waiting, active := newAtomicReclaimFixture(t, stubClock{}, waitingTransport, activeTransport)
	237	+
	238	+	finalizerStarted := make(chan struct{})
	239	+	releaseFinalizer := make(chan struct{})
	240	+	d.afterDisplacedCleanupStarted = func() {
	241	+	close(finalizerStarted)
	242	+	<-releaseFinalizer
	243	+	}
	244	+	waitingToken := sess.attachmentToken(waiting, waitingTransport)
	245	+	d.handleSnatchedClientFrame(waitingToken, ports.Frame{
	246	+	Type: ports.MsgInput,
	247	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	248	+	})
	249	+	awaitTestCompletion(t, finalizerStarted, "displaced finalizer did not pause")
	250	+
	251	+	activeToken := sess.attachmentToken(active, activeTransport)
	252	+	require.Equal(t, attachmentSnatched, activeToken.role)
	253	+	require.True(t, d.handleSnatchedClientFrame(activeToken, ports.Frame{
	254	+	Type: ports.MsgInput,
	255	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'q'}}),
	256	+	}))
	257	+	close(releaseFinalizer)
	258	+	d.attachmentCleanupWg.Wait()
	259	+
	260	+	require.Equal(t, attachmentDetached, sess.attachmentRole(active))
	261	+	frames := activeTransport.Sends()
	262	+	require.Len(t, frames, 1, "stale finalizer emitted output after quit")
	263	+	require.Equal(t, ports.MsgDetached, frames[0].Type)
	264	+	}
	265	+
	266	+	func TestReclaimOrdersDisplacedPanelAfterAdmittedOldPaint(t *testing.T) {
	267	+	d, sess, requester, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	268	+	requesterTransport := &closeTrackingTransport{}
	269	+	requester.replaceTransport(requesterTransport)
	270	+	d.attachCoordinator(sess, nil, requester, true)
	271	+
	272	+	activeTransport := &closeTrackingTransport{}
	273	+	active := &attachedClient{tr: activeTransport, output: newOutputStateStream(), size: requester.size}
	274	+	active.initOverlays()
	275	+	initial, err := d.transitionAttachment(attachmentTransitionRequest{
	276	+	target: sess, next: active,
	277	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	278	+	expectedTransport: active.transportSnapshot(), ready: true,
	279	+	})
	280	+	require.NoError(t, err)
	281	+	d.deferAttachmentTransitionCleanups(initial)
	282	+	d.attachmentCleanupWg.Wait()
	283	+	baselineActiveFrames := len(activeTransport.Sends())
	284	+
	285	+	paintAdmitted := make(chan struct{})
	286	+	releasePaint := make(chan struct{})
	287	+	d.afterRoleEffectAdmitted = func(token attachmentRoleToken) {
	288	+	if token.ac == active && token.role == attachmentActive {
	289	+	close(paintAdmitted)
	290	+	<-releasePaint
	291	+	}
	292	+	}
	293	+	activeToken := sess.attachmentToken(active, activeTransport)
	294	+	paintDone := make(chan bool, 1)
	295	+	go func() { paintDone <- d.firstPaintForTransition(activeToken) }()
	296	+	awaitTestCompletion(t, paintAdmitted, "old active paint was not admitted")
	297	+
	298	+	reclaimFrozen := make(chan struct{})
	299	+	var frozenOnce sync.Once
	300	+	d.afterRoleEffectGateFrozen = func(action string, _ *attachedClient) {
	301	+	if action == "reclaim-snatched" {
	302	+	frozenOnce.Do(func() { close(reclaimFrozen) })
	303	+	}
	304	+	}
	305	+	reclaimDone := make(chan struct{})
	306	+	go func() {
	307	+	token := sess.attachmentToken(requester, requesterTransport)
	308	+	d.handleSnatchedClientFrame(token, ports.Frame{
	309	+	Type: ports.MsgInput,
	310	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	311	+	})
	312	+	close(reclaimDone)
	313	+	}()
	314	+	awaitTestCompletion(t, reclaimFrozen, "reclaim did not freeze role effects behind old paint")
	315	+	close(releasePaint)
	316	+	require.True(t, awaitTestValue(t, paintDone, "old paint did not finish"))
	317	+	awaitTestCompletion(t, reclaimDone, "reclaim did not finish after old paint")
	318	+	d.attachmentCleanupWg.Wait()
	319	+
	320	+	frames := activeTransport.Sends()[baselineActiveFrames:]
	321	+	require.NotEmpty(t, frames)
	322	+	last := frames[len(frames)-1]
	323	+	require.Equal(t, ports.MsgOutput, last.Type)
	324	+	output, err := ports.UnmarshalOutput(last.Payload)
	325	+	require.NoError(t, err)
	326	+	require.Contains(t, string(output.Data), "Session snatched", "old paint overtook the displaced reset panel")
	327	+	requireSingleOwnerAndLease(t, sess, requester, active)
	328	+	}
	329	+
	330	+	func TestLaterReclaimMakesPausedFirstPaintStale(t *testing.T) {
	331	+	d, sess, first, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	332	+	firstTransport := &closeTrackingTransport{}
	333	+	first.replaceTransport(firstTransport)
	334	+	d.attachCoordinator(sess, nil, first, true)
	335	+
	336	+	secondTransport := &closeTrackingTransport{}
	337	+	second := &attachedClient{tr: secondTransport, output: newOutputStateStream(), size: first.size}
	338	+	second.initOverlays()
	339	+	initial, err := d.transitionAttachment(attachmentTransitionRequest{
	340	+	target: sess, next: second,
	341	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	342	+	expectedTransport: second.transportSnapshot(), ready: true,
	343	+	})
	344	+	require.NoError(t, err)
	345	+	d.deferAttachmentTransitionCleanups(initial)
	346	+	d.attachmentCleanupWg.Wait()
	347	+	baselineFirstFrames := len(firstTransport.Sends())
	348	+
	349	+	firstPaintPaused := make(chan struct{})
	350	+	releaseFirstPaint := make(chan struct{})
	351	+	d.beforeReclaimFirstPaint = func(token attachmentRoleToken) {
	352	+	if token.ac != first {
	353	+	return
	354	+	}
	355	+	close(firstPaintPaused)
	356	+	<-releaseFirstPaint
	357	+	}
	358	+	firstDone := make(chan struct{})
	359	+	go func() {
	360	+	token := sess.attachmentToken(first, firstTransport)
	361	+	d.handleSnatchedClientFrame(token, ports.Frame{
	362	+	Type: ports.MsgInput,
	363	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	364	+	})
	365	+	close(firstDone)
	366	+	}()
	367	+	awaitTestCompletion(t, firstPaintPaused, "first reclaim did not pause before first paint")
	368	+	require.Same(t, first, sess.client, "first reclaim was not committed before its paint")
	369	+
	370	+	secondToken := sess.attachmentToken(second, secondTransport)
	371	+	require.Equal(t, attachmentSnatched, secondToken.role)
	372	+	d.handleSnatchedClientFrame(secondToken, ports.Frame{
	373	+	Type: ports.MsgInput,
	374	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	375	+	})
	376	+	require.Same(t, second, sess.client)
	377	+	close(releaseFirstPaint)
	378	+	awaitTestCompletion(t, firstDone, "stale first paint did not return")
	379	+	d.attachmentCleanupWg.Wait()
	380	+
	381	+	newFirstFrames := firstTransport.Sends()[baselineFirstFrames:]
	382	+	require.Len(t, newFirstFrames, 1, "stale first paint emitted after a later reclaim")
	383	+	require.Equal(t, ports.MsgOutput, newFirstFrames[0].Type)
	384	+	output, err := ports.UnmarshalOutput(newFirstFrames[0].Payload)
	385	+	require.NoError(t, err)
	386	+	require.Contains(t, string(output.Data), "Session snatched")
	387	+	requireSingleOwnerAndLease(t, sess, first, second)
	388	+	}
	389	+
	390	+	func TestAnyWaiterCanReclaimAndConcurrentRequestsKeepSingleOwnerLease(t *testing.T) {
	391	+	d, sess, first, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	392	+	firstTransport := &closeTrackingTransport{}
	393	+	first.replaceTransport(firstTransport)
	394	+	d.attachCoordinator(sess, nil, first, true)
	395	+
	396	+	secondTransport := &closeTrackingTransport{}
	397	+	second := &attachedClient{tr: secondTransport, output: newOutputStateStream(), size: first.size}
	398	+	second.initOverlays()
	399	+	toSecond, err := d.transitionAttachment(attachmentTransitionRequest{
	400	+	target: sess, next: second,
	401	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	402	+	expectedTransport: second.transportSnapshot(), ready: true,
	403	+	})
	404	+	require.NoError(t, err)
	405	+	d.deferAttachmentTransitionCleanups(toSecond)
	406	+	d.attachmentCleanupWg.Wait()
	407	+
	408	+	thirdTransport := &closeTrackingTransport{}
	409	+	third := &attachedClient{tr: thirdTransport, output: newOutputStateStream(), size: first.size}
	410	+	third.initOverlays()
	411	+	toThird, err := d.transitionAttachment(attachmentTransitionRequest{
	412	+	target: sess, next: third,
	413	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	414	+	expectedTransport: third.transportSnapshot(), ready: true,
	415	+	})
	416	+	require.NoError(t, err)
	417	+	d.deferAttachmentTransitionCleanups(toThird)
	418	+	d.attachmentCleanupWg.Wait()
	419	+	requireSingleOwnerAndLease(t, sess, first, second, third)
	420	+
	421	+	firstToken := sess.attachmentToken(first, firstTransport)
	422	+	secondToken := sess.attachmentToken(second, secondTransport)
	423	+	require.Equal(t, attachmentSnatched, firstToken.role)
	424	+	require.Equal(t, attachmentSnatched, secondToken.role)
	425	+	start := make(chan struct{})
	426	+	done := make(chan struct{}, 2)
	427	+	for _, token := range []attachmentRoleToken{firstToken, secondToken} {
	428	+	go func() {
	429	+	<-start
	430	+	d.handleSnatchedClientFrame(token, ports.Frame{
	431	+	Type: ports.MsgInput,
	432	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	433	+	})
	434	+	done <- struct{}{}
	435	+	}()
	436	+	}
	437	+	close(start)
	438	+	awaitTestCompletion(t, done, "first concurrent reclaim did not return")
	439	+	awaitTestCompletion(t, done, "second concurrent reclaim did not return")
	440	+	d.attachmentCleanupWg.Wait()
	441	+
	442	+	requireSingleOwnerAndLease(t, sess, first, second, third)
	443	+	}

@@ -0,0 +1,443 @@
	1	+	package daemon
	2	+
	3	+	import (
	4	+	"errors"
	5	+	"sync"
	6	+	"testing"
	7	+	"time"
	8	+
	9	+	"github.com/stretchr/testify/require"
	10	+
	11	+	"github.com/bnema/vev/internal/domain"
	12	+	"github.com/bnema/vev/internal/ports"
	13	+	themeui "github.com/bnema/vev/internal/usecase/theme"
	14	+	"github.com/bnema/vev/pkg/renderer"
	15	+	)
	16	+
	17	+	type failingReclaimTransport struct {
	18	+	mu sync.Mutex
	19	+	closed bool
	20	+	}
	21	+
	22	+	func (*failingReclaimTransport) Send(ports.Frame) error { return errors.New("send failed") }
	23	+	func (*failingReclaimTransport) Recv() (ports.Frame, error) {
	24	+	return ports.Frame{}, errors.New("closed")
	25	+	}
	26	+	func (t *failingReclaimTransport) Close() error {
	27	+	t.mu.Lock()
	28	+	t.closed = true
	29	+	t.mu.Unlock()
	30	+	return nil
	31	+	}
	32	+	func (t *failingReclaimTransport) Closed() bool {
	33	+	t.mu.Lock()
	34	+	defer t.mu.Unlock()
	35	+	return t.closed
	36	+	}
	37	+
	38	+	func newAtomicReclaimFixture(t testing.T, clock ports.Clock, waitingTransport, activeTransport ports.Transport) (Daemon, session, attachedClient, *attachedClient) {
	39	+	t.Helper()
	40	+	d := newTestDaemon(t, nil, clock)
	41	+	waiting := &attachedClient{tr: waitingTransport, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}}
	42	+	active := &attachedClient{tr: activeTransport, output: newOutputStateStream(), size: waiting.size}
	43	+	waiting.initOverlays()
	44	+	active.initOverlays()
	45	+	sess := &session{
	46	+	id: "atomic-reclaim", client: active,
	47	+	snatched: map[*attachedClient]struct{}{waiting: {}},
	48	+	}
	49	+	waiting.setSession(sess)
	50	+	active.setSession(sess)
	51	+	d.sessions[sess.id] = sess
	52	+	d.attachCoordinator(sess, nil, active, true)
	53	+	return d, sess, waiting, active
	54	+	}
	55	+
	56	+	func requireSingleOwnerAndLease(t testing.T, sess session, clients ...*attachedClient) {
	57	+	t.Helper()
	58	+	sess.mu.Lock()
	59	+	owner := sess.client
	60	+	snatched := make(map[*attachedClient]struct{}, len(sess.snatched))
	61	+	for ac := range sess.snatched {
	62	+	snatched[ac] = struct{}{}
	63	+	}
	64	+	sess.mu.Unlock()
	65	+	require.NotNil(t, owner)
	66	+
	67	+	activeCount := 0
	68	+	for _, ac := range clients {
	69	+	if ac == owner {
	70	+	activeCount++
	71	+	continue
	72	+	}
	73	+	_, waiting := snatched[ac]
	74	+	require.True(t, waiting, "every non-owner must remain snatched")
	75	+	}
	76	+	require.Equal(t, 1, activeCount)
	77	+
	78	+	rc := sess.renderCoordinator()
	79	+	rc.mu.Lock()
	80	+	lease := rc.lease
	81	+	rc.mu.Unlock()
	82	+	require.NotNil(t, lease)
	83	+	require.True(t, lease.active)
	84	+	require.Same(t, owner, lease.attachment)
	85	+	}
	86	+
	87	+	func TestReclaimSnatchedSwapsOwnerLeaseRebasesAndAppliesTheme(t *testing.T) {
	88	+	d, sess, requester, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	89	+	requesterTransport := &closeTrackingTransport{}
	90	+	requester.replaceTransport(requesterTransport)
	91	+	d.attachCoordinator(sess, nil, requester, true)
	92	+
	93	+	active := &attachedClient{
	94	+	tr: &closeTrackingTransport{},
	95	+	output: newOutputStateStream(),
	96	+	size: requester.size,
	97	+	}
	98	+	active.initOverlays()
	99	+	displaced, err := d.transitionAttachment(attachmentTransitionRequest{
	100	+	target: sess, next: active,
	101	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	102	+	expectedTransport: active.transportSnapshot(), ready: true,
	103	+	})
	104	+	require.NoError(t, err)
	105	+	d.deferAttachmentTransitionCleanups(displaced)
	106	+	d.attachmentCleanupWg.Wait()
	107	+
	108	+	requester.sendMu.Lock()
	109	+	beforeReclaimState := requester.output.next
	110	+	requester.sendMu.Unlock()
	111	+	requesterTheme := themeui.Theme{
	112	+	Foreground: renderer.RGB{R: 11, G: 22, B: 33},
	113	+	Background: renderer.RGB{R: 44, G: 55, B: 66},
	114	+	HasFG: true, HasBG: true, Known: true, TrueColor: true,
	115	+	}
	116	+	requester.setClientTheme(requesterTheme)
	117	+
	118	+	token := sess.attachmentToken(requester, requesterTransport)
	119	+	require.Equal(t, attachmentSnatched, token.role)
	120	+	require.False(t, d.handleSnatchedClientFrame(token, ports.Frame{
	121	+	Type: ports.MsgInput,
	122	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	123	+	}))
	124	+
	125	+	sess.mu.Lock()
	126	+	require.Same(t, requester, sess.client)
	127	+	_, activeIsSnatched := sess.snatched[active]
	128	+	_, requesterIsSnatched := sess.snatched[requester]
	129	+	sess.mu.Unlock()
	130	+	require.True(t, activeIsSnatched)
	131	+	require.False(t, requesterIsSnatched)
	132	+	rc := sess.renderCoordinator()
	133	+	lease := rc.attachmentLease(requester)
	134	+	require.NotNil(t, lease)
	135	+	require.Nil(t, rc.attachmentLease(active))
	136	+	require.True(t, lease.active)
	137	+
	138	+	requester.sendMu.Lock()
	139	+	require.Greater(t, requester.output.next, beforeReclaimState, "reclaim must emit a reset first paint")
	140	+	require.GreaterOrEqual(t, requester.output.acked, beforeReclaimState, "reclaim must rebase the old panel output chain")
	141	+	requester.sendMu.Unlock()
	142	+	assertSessionDefaultColors(t, sess, requesterTheme.Foreground, requesterTheme.Background)
	143	+
	144	+	d.attachmentCleanupWg.Wait()
	145	+	require.Equal(t, attachmentSnatched, sess.attachmentRole(active))
	146	+	require.False(t, active.transport().(*closeTrackingTransport).Closed())
	147	+	require.Equal(t, domain.Size{Cols: 80, Rows: 24}, requester.size)
	148	+	}
	149	+
	150	+	func TestFailedReclaimBeforeCommitPreservesOwnerAndShowsUnavailable(t *testing.T) {
	151	+	waitingTransport := &closeTrackingTransport{}
	152	+	activeTransport := &closeTrackingTransport{}
	153	+	d, sess, waiting, active := newAtomicReclaimFixture(t, stubClock{}, waitingTransport, activeTransport)
	154	+	token := sess.attachmentToken(waiting, waitingTransport)
	155	+
	156	+	d.mu.Lock()
	157	+	delete(d.sessions, sess.id)
	158	+	d.mu.Unlock()
	159	+	d.handleSnatchedClientFrame(token, ports.Frame{
	160	+	Type: ports.MsgInput,
	161	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	162	+	})
	163	+
	164	+	requireSingleOwnerAndLease(t, sess, waiting, active)
	165	+	require.Same(t, active, sess.client)
	166	+	require.False(t, waitingTransport.Closed())
	167	+	frames := waitingTransport.Sends()
	168	+	require.Len(t, frames, 1)
	169	+	output, err := ports.UnmarshalOutput(frames[0].Payload)
	170	+	require.NoError(t, err)
	171	+	require.Contains(t, string(output.Data), "Session is no longer available.")
	172	+	}
	173	+
	174	+	func TestPostcommitPanelFailureRetiresOnlyDisplacedClient(t *testing.T) {
	175	+	waitingTransport := &closeTrackingTransport{}
	176	+	activeTransport := &failingReclaimTransport{}
	177	+	d, sess, waiting, active := newAtomicReclaimFixture(t, stubClock{}, waitingTransport, activeTransport)
	178	+	token := sess.attachmentToken(waiting, waitingTransport)
	179	+
	180	+	d.handleSnatchedClientFrame(token, ports.Frame{
	181	+	Type: ports.MsgInput,
	182	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	183	+	})
	184	+	d.attachmentCleanupWg.Wait()
	185	+
	186	+	require.Same(t, waiting, sess.client)
	187	+	require.Equal(t, attachmentActive, sess.attachmentRole(waiting))
	188	+	require.Equal(t, attachmentDetached, sess.attachmentRole(active))
	189	+	require.NotNil(t, sess.renderCoordinator().attachmentLease(waiting))
	190	+	require.Nil(t, sess.renderCoordinator().attachmentLease(active))
	191	+	require.False(t, waitingTransport.Closed())
	192	+	require.True(t, activeTransport.Closed())
	193	+	}
	194	+
	195	+	func TestBlockedRequesterActivationDeadlinePreservesOwnerLease(t *testing.T) {
	196	+	clock := &signalClock{timers: make(chan *signalTimer, 2)}
	197	+	waitingTransport := newSnatchedCloseSignalTransport()
	198	+	activeTransport := &closeTrackingTransport{}
	199	+	d, sess, waiting, active := newAtomicReclaimFixture(t, clock, waitingTransport, activeTransport)
	200	+
	201	+	waiting.sendMu.Lock()
	202	+	locked := true
	203	+	defer func() {
	204	+	if locked {
	205	+	waiting.sendMu.Unlock()
	206	+	}
	207	+	}()
	208	+	done := make(chan struct{})
	209	+	go func() {
	210	+	token := sess.attachmentToken(waiting, waitingTransport)
	211	+	d.handleSnatchedClientFrame(token, ports.Frame{
	212	+	Type: ports.MsgInput,
	213	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	214	+	})
	215	+	close(done)
	216	+	}()
	217	+
	218	+	timer := awaitTestValue(t, clock.timers, "reclaim did not install its activation deadline")
	219	+	require.Equal(t, detachNotifyTimeout, timer.duration)
	220	+	timer.ch <- time.Time{}
	221	+	awaitTestCompletion(t, done, "deadline did not release blocked reclaim")
	222	+	require.Same(t, active, sess.client)
	223	+	require.Same(t, active, sess.renderCoordinator().attachmentLease(active).attachment)
	224	+	require.True(t, waitingTransport.Closed())
	225	+
	226	+	waiting.sendMu.Unlock()
	227	+	locked = false
	228	+	d.attachmentCleanupWg.Wait()
	229	+	require.Equal(t, attachmentDetached, sess.attachmentRole(waiting))
	230	+	require.Equal(t, attachmentActive, sess.attachmentRole(active))
	231	+	}
	232	+
	233	+	func TestQuitInvalidatesPausedSnatchedFinalizerWithoutLatePanel(t *testing.T) {
	234	+	waitingTransport := &closeTrackingTransport{}
	235	+	activeTransport := &closeTrackingTransport{}
	236	+	d, sess, waiting, active := newAtomicReclaimFixture(t, stubClock{}, waitingTransport, activeTransport)
	237	+
	238	+	finalizerStarted := make(chan struct{})
	239	+	releaseFinalizer := make(chan struct{})
	240	+	d.afterDisplacedCleanupStarted = func() {
	241	+	close(finalizerStarted)
	242	+	<-releaseFinalizer
	243	+	}
	244	+	waitingToken := sess.attachmentToken(waiting, waitingTransport)
	245	+	d.handleSnatchedClientFrame(waitingToken, ports.Frame{
	246	+	Type: ports.MsgInput,
	247	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	248	+	})
	249	+	awaitTestCompletion(t, finalizerStarted, "displaced finalizer did not pause")
	250	+
	251	+	activeToken := sess.attachmentToken(active, activeTransport)
	252	+	require.Equal(t, attachmentSnatched, activeToken.role)
	253	+	require.True(t, d.handleSnatchedClientFrame(activeToken, ports.Frame{
	254	+	Type: ports.MsgInput,
	255	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'q'}}),
	256	+	}))
	257	+	close(releaseFinalizer)
	258	+	d.attachmentCleanupWg.Wait()
	259	+
	260	+	require.Equal(t, attachmentDetached, sess.attachmentRole(active))
	261	+	frames := activeTransport.Sends()
	262	+	require.Len(t, frames, 1, "stale finalizer emitted output after quit")
	263	+	require.Equal(t, ports.MsgDetached, frames[0].Type)
	264	+	}
	265	+
	266	+	func TestReclaimOrdersDisplacedPanelAfterAdmittedOldPaint(t *testing.T) {
	267	+	d, sess, requester, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	268	+	requesterTransport := &closeTrackingTransport{}
	269	+	requester.replaceTransport(requesterTransport)
	270	+	d.attachCoordinator(sess, nil, requester, true)
	271	+
	272	+	activeTransport := &closeTrackingTransport{}
	273	+	active := &attachedClient{tr: activeTransport, output: newOutputStateStream(), size: requester.size}
	274	+	active.initOverlays()
	275	+	initial, err := d.transitionAttachment(attachmentTransitionRequest{
	276	+	target: sess, next: active,
	277	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	278	+	expectedTransport: active.transportSnapshot(), ready: true,
	279	+	})
	280	+	require.NoError(t, err)
	281	+	d.deferAttachmentTransitionCleanups(initial)
	282	+	d.attachmentCleanupWg.Wait()
	283	+	baselineActiveFrames := len(activeTransport.Sends())
	284	+
	285	+	paintAdmitted := make(chan struct{})
	286	+	releasePaint := make(chan struct{})
	287	+	d.afterRoleEffectAdmitted = func(token attachmentRoleToken) {
	288	+	if token.ac == active && token.role == attachmentActive {
	289	+	close(paintAdmitted)
	290	+	<-releasePaint
	291	+	}
	292	+	}
	293	+	activeToken := sess.attachmentToken(active, activeTransport)
	294	+	paintDone := make(chan bool, 1)
	295	+	go func() { paintDone <- d.firstPaintForTransition(activeToken) }()
	296	+	awaitTestCompletion(t, paintAdmitted, "old active paint was not admitted")
	297	+
	298	+	reclaimFrozen := make(chan struct{})
	299	+	var frozenOnce sync.Once
	300	+	d.afterRoleEffectGateFrozen = func(action string, _ *attachedClient) {
	301	+	if action == "reclaim-snatched" {
	302	+	frozenOnce.Do(func() { close(reclaimFrozen) })
	303	+	}
	304	+	}
	305	+	reclaimDone := make(chan struct{})
	306	+	go func() {
	307	+	token := sess.attachmentToken(requester, requesterTransport)
	308	+	d.handleSnatchedClientFrame(token, ports.Frame{
	309	+	Type: ports.MsgInput,
	310	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	311	+	})
	312	+	close(reclaimDone)
	313	+	}()
	314	+	awaitTestCompletion(t, reclaimFrozen, "reclaim did not freeze role effects behind old paint")
	315	+	close(releasePaint)
	316	+	require.True(t, awaitTestValue(t, paintDone, "old paint did not finish"))
	317	+	awaitTestCompletion(t, reclaimDone, "reclaim did not finish after old paint")
	318	+	d.attachmentCleanupWg.Wait()
	319	+
	320	+	frames := activeTransport.Sends()[baselineActiveFrames:]
	321	+	require.NotEmpty(t, frames)
	322	+	last := frames[len(frames)-1]
	323	+	require.Equal(t, ports.MsgOutput, last.Type)
	324	+	output, err := ports.UnmarshalOutput(last.Payload)
	325	+	require.NoError(t, err)
	326	+	require.Contains(t, string(output.Data), "Session snatched", "old paint overtook the displaced reset panel")
	327	+	requireSingleOwnerAndLease(t, sess, requester, active)
	328	+	}
	329	+
	330	+	func TestLaterReclaimMakesPausedFirstPaintStale(t *testing.T) {
	331	+	d, sess, first, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	332	+	firstTransport := &closeTrackingTransport{}
	333	+	first.replaceTransport(firstTransport)
	334	+	d.attachCoordinator(sess, nil, first, true)
	335	+
	336	+	secondTransport := &closeTrackingTransport{}
	337	+	second := &attachedClient{tr: secondTransport, output: newOutputStateStream(), size: first.size}
	338	+	second.initOverlays()
	339	+	initial, err := d.transitionAttachment(attachmentTransitionRequest{
	340	+	target: sess, next: second,
	341	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	342	+	expectedTransport: second.transportSnapshot(), ready: true,
	343	+	})
	344	+	require.NoError(t, err)
	345	+	d.deferAttachmentTransitionCleanups(initial)
	346	+	d.attachmentCleanupWg.Wait()
	347	+	baselineFirstFrames := len(firstTransport.Sends())
	348	+
	349	+	firstPaintPaused := make(chan struct{})
	350	+	releaseFirstPaint := make(chan struct{})
	351	+	d.beforeReclaimFirstPaint = func(token attachmentRoleToken) {
	352	+	if token.ac != first {
	353	+	return
	354	+	}
	355	+	close(firstPaintPaused)
	356	+	<-releaseFirstPaint
	357	+	}
	358	+	firstDone := make(chan struct{})
	359	+	go func() {
	360	+	token := sess.attachmentToken(first, firstTransport)
	361	+	d.handleSnatchedClientFrame(token, ports.Frame{
	362	+	Type: ports.MsgInput,
	363	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	364	+	})
	365	+	close(firstDone)
	366	+	}()
	367	+	awaitTestCompletion(t, firstPaintPaused, "first reclaim did not pause before first paint")
	368	+	require.Same(t, first, sess.client, "first reclaim was not committed before its paint")
	369	+
	370	+	secondToken := sess.attachmentToken(second, secondTransport)
	371	+	require.Equal(t, attachmentSnatched, secondToken.role)
	372	+	d.handleSnatchedClientFrame(secondToken, ports.Frame{
	373	+	Type: ports.MsgInput,
	374	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	375	+	})
	376	+	require.Same(t, second, sess.client)
	377	+	close(releaseFirstPaint)
	378	+	awaitTestCompletion(t, firstDone, "stale first paint did not return")
	379	+	d.attachmentCleanupWg.Wait()
	380	+
	381	+	newFirstFrames := firstTransport.Sends()[baselineFirstFrames:]
	382	+	require.Len(t, newFirstFrames, 1, "stale first paint emitted after a later reclaim")
	383	+	require.Equal(t, ports.MsgOutput, newFirstFrames[0].Type)
	384	+	output, err := ports.UnmarshalOutput(newFirstFrames[0].Payload)
	385	+	require.NoError(t, err)
	386	+	require.Contains(t, string(output.Data), "Session snatched")
	387	+	requireSingleOwnerAndLease(t, sess, first, second)
	388	+	}
	389	+
	390	+	func TestAnyWaiterCanReclaimAndConcurrentRequestsKeepSingleOwnerLease(t *testing.T) {
	391	+	d, sess, first, _ := newManualSessionWithPTYs(t, &transactionalResizePTY{})
	392	+	firstTransport := &closeTrackingTransport{}
	393	+	first.replaceTransport(firstTransport)
	394	+	d.attachCoordinator(sess, nil, first, true)
	395	+
	396	+	secondTransport := &closeTrackingTransport{}
	397	+	second := &attachedClient{tr: secondTransport, output: newOutputStateStream(), size: first.size}
	398	+	second.initOverlays()
	399	+	toSecond, err := d.transitionAttachment(attachmentTransitionRequest{
	400	+	target: sess, next: second,
	401	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	402	+	expectedTransport: second.transportSnapshot(), ready: true,
	403	+	})
	404	+	require.NoError(t, err)
	405	+	d.deferAttachmentTransitionCleanups(toSecond)
	406	+	d.attachmentCleanupWg.Wait()
	407	+
	408	+	thirdTransport := &closeTrackingTransport{}
	409	+	third := &attachedClient{tr: thirdTransport, output: newOutputStateStream(), size: first.size}
	410	+	third.initOverlays()
	411	+	toThird, err := d.transitionAttachment(attachmentTransitionRequest{
	412	+	target: sess, next: third,
	413	+	expectedRole: attachmentDetached, targetRole: attachmentActive,
	414	+	expectedTransport: third.transportSnapshot(), ready: true,
	415	+	})
	416	+	require.NoError(t, err)
	417	+	d.deferAttachmentTransitionCleanups(toThird)
	418	+	d.attachmentCleanupWg.Wait()
	419	+	requireSingleOwnerAndLease(t, sess, first, second, third)
	420	+
	421	+	firstToken := sess.attachmentToken(first, firstTransport)
	422	+	secondToken := sess.attachmentToken(second, secondTransport)
	423	+	require.Equal(t, attachmentSnatched, firstToken.role)
	424	+	require.Equal(t, attachmentSnatched, secondToken.role)
	425	+	start := make(chan struct{})
	426	+	done := make(chan struct{}, 2)
	427	+	for _, token := range []attachmentRoleToken{firstToken, secondToken} {
	428	+	go func() {
	429	+	<-start
	430	+	d.handleSnatchedClientFrame(token, ports.Frame{
	431	+	Type: ports.MsgInput,
	432	+	Payload: ports.MarshalInput(ports.Input{Data: []byte{'r'}}),
	433	+	})
	434	+	done <- struct{}{}
	435	+	}()
	436	+	}
	437	+	close(start)
	438	+	awaitTestCompletion(t, done, "first concurrent reclaim did not return")
	439	+	awaitTestCompletion(t, done, "second concurrent reclaim did not return")
	440	+	d.attachmentCleanupWg.Wait()
	441	+
	442	+	requireSingleOwnerAndLease(t, sess, first, second, third)
	443	+	}

M internal/usecase/daemon/theme_resolve.go +8 -0


@@ -12,6 +12,14 @@ func (d Daemon) applyHostTheme(sess session, ac *attachedClient, t theme.Theme
12	12		return d.applyHostThemeLocked(sess, ac, t, clearUnknownScheme)
13	13		}
14	14
	15	+	// applyHostThemeSendLocked applies a theme when the caller already owns
	16	+	// ac.sendMu, preserving the usual sendMu -> themeMu lock order.
	17	+	func (d Daemon) applyHostThemeSendLocked(sess session, ac *attachedClient, t theme.Theme, clearUnknownScheme bool) bool {
	18	+	sess.themeMu.Lock()
	19	+	defer sess.themeMu.Unlock()
	20	+	return d.applyHostThemeLocked(sess, ac, t, clearUnknownScheme)
	21	+	}
	22	+
15	23		func (d Daemon) applyHostThemeLocked(sess session, ac *attachedClient, t theme.Theme, clearUnknownScheme bool) bool {
16	24		// Resolve while applying, never while composing. The theme mutex only
17	25		// publishes the completed value and is not held across session/tab/pane


@@ -12,6 +12,14 @@ func (d Daemon) applyHostTheme(sess session, ac *attachedClient, t theme.Theme
12	return d.applyHostThemeLocked(sess, ac, t, clearUnknownScheme)	12		return d.applyHostThemeLocked(sess, ac, t, clearUnknownScheme)
13	}	13		}
14		14
		15	+	// applyHostThemeSendLocked applies a theme when the caller already owns
		16	+	// ac.sendMu, preserving the usual sendMu -> themeMu lock order.
		17	+	func (d Daemon) applyHostThemeSendLocked(sess session, ac *attachedClient, t theme.Theme, clearUnknownScheme bool) bool {
		18	+	sess.themeMu.Lock()
		19	+	defer sess.themeMu.Unlock()
		20	+	return d.applyHostThemeLocked(sess, ac, t, clearUnknownScheme)
		21	+	}
		22	+
15	func (d Daemon) applyHostThemeLocked(sess session, ac *attachedClient, t theme.Theme, clearUnknownScheme bool) bool {	23		func (d Daemon) applyHostThemeLocked(sess session, ac *attachedClient, t theme.Theme, clearUnknownScheme bool) bool {
16	// Resolve while applying, never while composing. The theme mutex only	24		// Resolve while applying, never while composing. The theme mutex only
17	// publishes the completed value and is not held across session/tab/pane	25		// publishes the completed value and is not held across session/tab/pane