be54d4f refactor: deepen daemon render publication and client attachment lifecycle (#298) · vev

M internal/usecase/client/supervisor.go +3 -14


@@ -451,21 +451,10 @@ type Supervisor struct {
451	451		// runs, so the picker overlay over a live attachment keeps folding broker
452	452		// publications. It is only touched from the run goroutine.
453	453		readySub ports.BrokerSubscription
454		-	// pendingSwap is the request the picker overlay committed to another
455		-	// target while an attachment was live. It is only touched from the run
456		-	// goroutine and consumed by runResolvedAttachment.
	454	+	// pendingSwap is the request the picker overlay or a daemon navigation
	455	+	// committed to another target while an attachment was live. It is only
	456	+	// touched from the run goroutine and consumed by settleAttachment.
457	457		pendingSwap *pickerAttachmentTarget
458		-	// resume is the exact target a lost attachment reconnects to, set by
459		-	// settleAttachment and consumed by runResolvedAttachment. It is only
460		-	// touched from the run goroutine.
461		-	resume *pickerAttachmentTarget
462		-	// resuming is true while runResolvedAttachment reconnects a lost
463		-	// attachment; transport-class failures of those attempts are recorded in
464		-	// resumeErr instead of being presented, so a retry stays on Connecting.
465		-	// resumeErr also carries the loss that started the resume. Both are only
466		-	// touched from the run goroutine.
467		-	resuming bool
468		-	resumeErr error
469	458		// kills runs the picker's `x` operations off the run goroutine.
470	459		kills pickerKills
471	460		// routes is the client route ledger published to the serving daemon;


@@ -451,21 +451,10 @@ type Supervisor struct {
451		// runs, so the picker overlay over a live attachment keeps folding broker	451		// runs, so the picker overlay over a live attachment keeps folding broker
452		// publications. It is only touched from the run goroutine.	452		// publications. It is only touched from the run goroutine.
453		readySub ports.BrokerSubscription	453		readySub ports.BrokerSubscription
454	-	// pendingSwap is the request the picker overlay committed to another	454	+	// pendingSwap is the request the picker overlay or a daemon navigation
455	-	// target while an attachment was live. It is only touched from the run	455	+	// committed to another target while an attachment was live. It is only
456	-	// goroutine and consumed by runResolvedAttachment.	456	+	// touched from the run goroutine and consumed by settleAttachment.
457		pendingSwap *pickerAttachmentTarget	457		pendingSwap *pickerAttachmentTarget
458	-	// resume is the exact target a lost attachment reconnects to, set by
459	-	// settleAttachment and consumed by runResolvedAttachment. It is only
460	-	// touched from the run goroutine.
461	-	resume *pickerAttachmentTarget
462	-	// resuming is true while runResolvedAttachment reconnects a lost
463	-	// attachment; transport-class failures of those attempts are recorded in
464	-	// resumeErr instead of being presented, so a retry stays on Connecting.
465	-	// resumeErr also carries the loss that started the resume. Both are only
466	-	// touched from the run goroutine.
467	-	resuming bool
468	-	resumeErr error
469		// kills runs the picker's `x` operations off the run goroutine.	458		// kills runs the picker's `x` operations off the run goroutine.
470		kills pickerKills	459		kills pickerKills
471		// routes is the client route ledger published to the serving daemon;	460		// routes is the client route ledger published to the serving daemon;

M internal/usecase/client/supervisor_attach.go +141 -123


@@ -238,6 +238,40 @@ type attachmentSettlement struct {
238	238		adopted bool
239	239		}
240	240
	241	+	// attachmentOutcomeKind is what one attachment attempt tells the resume loop
	242	+	// to do next.
	243	+	type attachmentOutcomeKind uint8
	244	+
	245	+	const (
	246	+	// attachmentEnded returns to the picker; any failure is already presented.
	247	+	attachmentEnded attachmentOutcomeKind = iota
	248	+	// attachmentTerminated ends the whole run (cancellation, terminal EOF, or
	249	+	// detach-and-exit); cause is the run's termination error, possibly nil.
	250	+	attachmentTerminated
	251	+	// attachmentSwap attaches to target, which the picker overlay or a daemon
	252	+	// navigation committed while this attachment was live.
	253	+	attachmentSwap
	254	+	// attachmentResume reconnects to target after a live attachment was lost
	255	+	// with cause.
	256	+	attachmentResume
	257	+	// attachmentRetry repeats a resume attempt that hit the transport-class
	258	+	// failure cause before attaching.
	259	+	attachmentRetry
	260	+	)
	261	+
	262	+	// attachmentOutcome is the complete result of one attachment attempt.
	263	+	type attachmentOutcome struct {
	264	+	kind attachmentOutcomeKind
	265	+	target pickerAttachmentTarget
	266	+	cause error
	267	+	}
	268	+
	269	+	func attachmentEndedOutcome() attachmentOutcome { return attachmentOutcome{kind: attachmentEnded} }
	270	+
	271	+	func attachmentTerminatedOutcome(err error) attachmentOutcome {
	272	+	return attachmentOutcome{kind: attachmentTerminated, cause: err}
	273	+	}
	274	+
241	275		// resolveCommittedStreamRequest turns exactly one committed catalogue key into
242	276		// the exact broker stream request the user committed. The stream ID is
243	277		// allocated by the broker service — the single allocator on its own connection

@@ -556,93 +590,81 @@ func (s Supervisor) runResolvedAttachment(ctx context.Context, input terminalI
556	590		picker.SetOwnsInput(true)
557	591		input.acquirePicker()
558	592		}()
559		-	defer func() { s.resuming, s.resumeErr, s.resume = false, nil, nil }()
	593	+	// resuming is true while the loop reconnects a lost attachment: attempts
	594	+	// then return transport-class failures as attachmentRetry instead of
	595	+	// presenting them, so a retry stays on Connecting.
	596	+	resuming := false
560	597		resumes := 0
561	598		for {
562		-	s.resumeErr = nil
563	599		attemptStart := s.cfg.Clock.Now()
564		-	terminated, termErr, swap := s.attachOnce(ctx, input, service, target, localProvenance)
565		-	if terminated {
566		-	return terminated, termErr
567		-	}
568		-	if swap == nil {
569		-	resume := s.resume
570		-	s.resume = nil
571		-	switch {
572		-	case resume != nil:
573		-	// A live attachment was lost. One that stayed up long enough
574		-	// starts a fresh resume budget; one that drops right after
575		-	// attaching keeps counting, so a flapping session gives up.
576		-	if s.cfg.Clock.Now().Sub(attemptStart) >= resumeStableAttachment {
577		-	resumes = 0
578		-	}
579		-	case s.resuming && s.resumeErr != nil:
580		-	// A resume attempt hit a transport-class failure before
581		-	// attaching (the route or destination is still down): try the
582		-	// same target again.
583		-	retry := target
584		-	resume = &retry
585		-	default:
586		-	// No resume, or the attempt already presented a final outcome
587		-	// (session gone, ended, refused).
588		-	return false, nil
589		-	}
590		-	lastErr := s.resumeErr
591		-	resumes++
592		-	if resumes > maxAttachmentResumes {
593		-	s.logger.Warn("client_attachment_resume_exhausted", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempts", resumes-1, "error", lastErr)
594		-	s.resuming = false
595		-	s.transition(supervisorEvent{kind: supervisorAttachEnded, err: lastErr})
596		-	s.notifyLifecycle(LifecycleNoticeDestinationFailed)
597		-	s.notifyAttachment(lastErr)
598		-	return false, nil
599		-	}
600		-	s.resuming = true
601		-	if terminated, termErr, brokerLost := s.waitResume(ctx, input, service, resumes); terminated \|\| brokerLost {
602		-	return terminated, termErr
603		-	}
	600	+	outcome := s.attachResolved(ctx, input, service, target, localProvenance, resuming)
	601	+	switch outcome.kind {
	602	+	case attachmentTerminated:
	603	+	return true, outcome.cause
	604	+	case attachmentEnded:
	605	+	// The attempt already presented a final outcome (session gone,
	606	+	// ended, refused).
	607	+	return false, nil
	608	+	case attachmentSwap:
	609	+	resumes = 0
	610	+	resuming = false
	611	+	// A commit from the picker overlay to another target: the previous
	612	+	// attachment detached cleanly, and the swap goes through Connecting
	613	+	// exactly like any committed selection. The stream identity is
	614	+	// allocated now, not at commit: streams opened while the source was
	615	+	// live (route publications) may have moved the connection's
	616	+	// anti-replay window past an identity reserved earlier.
604	617		stream, err := service.NextStreamID()
605	618		if err != nil {
606		-	s.resuming = false
607	619		s.reportAttachmentFailure(err)
608	620		return false, nil
609	621		}
610		-	resume.request.Stream = stream
611		-	s.logger.Info("client_attachment_resume", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempt", resumes)
612		-	target = *resume
	622	+	target, localProvenance = outcome.target, SessionEnvironmentLocalPicker
	623	+	target.request.Stream = stream
613	624		continue
	625	+	case attachmentResume:
	626	+	// A live attachment was lost. One that stayed up long enough
	627	+	// starts a fresh resume budget; one that drops right after
	628	+	// attaching keeps counting, so a flapping session gives up.
	629	+	if s.cfg.Clock.Now().Sub(attemptStart) >= resumeStableAttachment {
	630	+	resumes = 0
	631	+	}
	632	+	case attachmentRetry:
	633	+	// A resume attempt hit a transport-class failure before
	634	+	// attaching (the route or destination is still down): try the
	635	+	// same target again.
	636	+	outcome.target = target
	637	+	}
	638	+	resume, lastErr := outcome.target, outcome.cause
	639	+	resumes++
	640	+	if resumes > maxAttachmentResumes {
	641	+	s.logger.Warn("client_attachment_resume_exhausted", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempts", resumes-1, "error", lastErr)
	642	+	s.transition(supervisorEvent{kind: supervisorAttachEnded, err: lastErr})
	643	+	s.notifyLifecycle(LifecycleNoticeDestinationFailed)
	644	+	s.notifyAttachment(lastErr)
	645	+	return false, nil
	646	+	}
	647	+	resuming = true
	648	+	if terminated, termErr, brokerLost := s.waitResume(ctx, input, service, resumes); terminated \|\| brokerLost {
	649	+	return terminated, termErr
614	650		}
615		-	resumes = 0
616		-	s.resuming = false
617		-	// A commit from the picker overlay to another target: the previous
618		-	// attachment detached cleanly, and the swap goes through Connecting
619		-	// exactly like any committed selection. The stream identity is
620		-	// allocated now, not at commit: streams opened while the source was
621		-	// live (route publications) may have moved the connection's
622		-	// anti-replay window past an identity reserved earlier.
623	651		stream, err := service.NextStreamID()
624	652		if err != nil {
625	653		s.reportAttachmentFailure(err)
626	654		return false, nil
627	655		}
628		-	swap.request.Stream = stream
629		-	target, localProvenance = *swap, SessionEnvironmentLocalPicker
630		-	}
631		-	}
632		-
633		-	// attachOnce runs one resolved attachment end to end. A non-nil swap names the
634		-	// next target the picker overlay committed while this attachment was live.
635		-	func (s Supervisor) attachOnce(ctx context.Context, input terminalInputLifetime, service ports.BrokerService, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance) (bool, error, *pickerAttachmentTarget) {
636		-	terminated, termErr := s.attachResolved(ctx, input, service, target, localProvenance)
637		-	swap := s.pendingSwap
638		-	s.pendingSwap = nil
639		-	if terminated {
640		-	return terminated, termErr, nil
	656	+	resume.request.Stream = stream
	657	+	s.logger.Info("client_attachment_resume", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempt", resumes)
	658	+	target = resume
641	659		}
642		-	return false, nil, swap
643	660		}
644	661
645		-	func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLifetime, service ports.BrokerService, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance) (bool, error) {
	662	+	// attachResolved runs one resolved attachment attempt end to end: open the
	663	+	// exact stream under one absolute deadline, admit the worker through the
	664	+	// foreground grant, settle it, and classify the result. resuming marks an
	665	+	// attempt of the resume loop, whose transport-class failures are returned as
	666	+	// attachmentRetry instead of presented.
	667	+	func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLifetime, service ports.BrokerService, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance, resuming bool) attachmentOutcome {
646	668		request := target.request
647	669		s.transition(supervisorEvent{kind: supervisorAttachBegin})
648	670		deadline := startAttachmentDeadline(ctx, s.cfg.Clock)

@@ -657,8 +679,7 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
657	679		sessionEnv.Cwd = ""
658	680		}
659	681		if err := sessionEnv.Validate(); err != nil {
660		-	s.reportAttachmentFailure(err)
661		-	return false, nil
	682	+	return s.attemptFailed(err, resuming)
662	683		}
663	684		request.Env = append([]string(nil), sessionEnv.Env...)
664	685

@@ -666,19 +687,16 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
666	687		stream, err := service.OpenStream(deadline.Context(), request)
667	688		if err != nil {
668	689		s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
669		-	s.reportAttachmentFailure(deadline.cause(err))
670		-	return false, nil
	690	+	return s.attemptFailed(deadline.cause(err), resuming)
671	691		}
672	692		if supervisorNil(stream) {
673		-	s.reportAttachmentFailure(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"})
674		-	return false, nil
	693	+	return s.attemptFailed(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"}, resuming)
675	694		}
676	695		if err := deadline.adopt(stream); err != nil {
677	696		// The accepted absolute deadline already elapsed before this client
678	697		// saw the stream. Fail the attachment instead of restarting a budget.
679	698		_ = stream.Close()
680		-	s.reportAttachmentFailure(attachmentTimeoutError(err))
681		-	return false, nil
	699	+	return s.attemptFailed(attachmentTimeoutError(err), resuming)
682	700		}
683	701
684	702		worker, err := s.newAttachmentWorker(request, target.tab, func() error {

@@ -690,8 +708,7 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
690	708		}, sessionEnv)
691	709		if err != nil {
692	710		_ = stream.Close()
693		-	s.reportAttachmentFailure(err)
694		-	return false, nil
	711	+	return s.attemptFailed(err, resuming)
695	712		}
696	713
697	714		token := AttachmentToken{Generation: s.State().Generation, Attempt: s.nextAttachment.Add(1)}

@@ -701,24 +718,23 @@ func (s Supervisor) attachResolved(ctx context.Context, input terminalInputLif
701	718		// Another foreground owns the terminal, or the shared reader is
702	719		// claimed. Begin never closes the stream, so the supervisor does.
703	720		_ = stream.Close()
704		-	if s.resuming {
	721	+	if resuming {
705	722		// A resume must say why it stopped instead of dropping silently
706		-	// to the picker.
707		-	s.resuming = false
	723	+	// to the picker, so the busy error is presented, never retried.
708	724		s.reportAttachmentFailure(errResumeForegroundBusy)
709		-	return false, nil
	725	+	return attachmentEndedOutcome()
710	726		}
711	727		s.transition(supervisorEvent{kind: supervisorAttachEnded})
712		-	return false, nil
	728	+	return attachmentEndedOutcome()
713	729		}
714		-	return s.settleAttachment(ctx, input, service, deadline, run, request)
	730	+	return s.settleAttachment(ctx, input, service, deadline, run, request, resuming)
715	731		}
716	732
717	733		// settleAttachment joins one admitted attachment while watching the parent
718	734		// context, terminal EOF, and broker-connection loss. The deadline is the only
719	735		// other stop: when it fires before the worker settled, the attachment ends as
720	736		// a typed timeout without claiming attachment.
721		-	func (s Supervisor) settleAttachment(ctx context.Context, input terminalInputLifetime, service ports.BrokerService, deadline attachmentDeadline, run attachmentRun, request ports.BrokerOpenStreamRequest) (bool, error) {
	737	+	func (s Supervisor) settleAttachment(ctx context.Context, input terminalInputLifetime, service ports.BrokerService, deadline attachmentDeadline, run attachmentRun, request ports.BrokerOpenStreamRequest, resuming bool) attachmentOutcome {
722	738		settled := make(chan attachmentSettlement, 1)
723	739		go func() {
724	740		event, adopted := run.Wait(deadline.Context())

@@ -794,39 +810,39 @@ settlement:
794	810		if !terminated && !brokerLost {
795	811		s.takeSettledNavigation(service, run.token, request)
796	812		}
	813	+	// The overlay or a daemon navigation may have committed another target
	814	+	// while this attachment was live; this attempt consumes it either way.
	815	+	swap := s.pendingSwap
	816	+	s.pendingSwap = nil
797	817		if terminated {
798		-	s.pendingSwap = nil
799		-	return true, termErr
800		-	}
801		-	if s.pendingSwap != nil {
802		-	if !brokerLost {
803		-	// The overlay committed another target and this attachment
804		-	// detached for it: go straight to Connecting for the swap, never
805		-	// through the plain picker.
806		-	return false, nil
807		-	}
808		-	s.pendingSwap = nil
	818	+	return attachmentTerminatedOutcome(termErr)
	819	+	}
	820	+	if swap != nil && !brokerLost {
	821	+	// The overlay committed another target and this attachment detached
	822	+	// for it: go straight to Connecting for the swap, never through the
	823	+	// plain picker.
	824	+	return attachmentOutcome{kind: attachmentSwap, target: *swap}
809	825		}
810	826		if deadline.TimedOut() && !result.adopted {
811	827		timeout := attachmentTimeoutError(errAttachmentDeadline)
812		-	if s.deferResumeFailure(timeout) {
813		-	return false, nil
	828	+	if outcome, ok := s.resumeRetry(timeout, resuming); ok {
	829	+	return outcome
814	830		}
815	831		s.transition(supervisorEvent{kind: supervisorAttachEnded, err: timeout})
816	832		s.notifyAttachment(timeout)
817		-	return false, nil
	833	+	return attachmentEndedOutcome()
818	834		}
819	835		if errors.Is(result.event.Err, errDetachAndExit) {
820	836		s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "explicit_detach")
821	837		s.transition(supervisorEvent{kind: supervisorAttachEnded})
822	838		s.notifyLifecycle(LifecycleNoticeDetachAndExit)
823		-	return true, nil
	839	+	return attachmentTerminatedOutcome(nil)
824	840		}
825	841		if errors.Is(result.event.Err, errDetachToPicker) {
826	842		s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "detach_to_picker")
827	843		s.transition(supervisorEvent{kind: supervisorAttachEnded})
828	844		s.notifyLifecycle(LifecycleNoticeDetachToPicker)
829		-	return false, nil
	845	+	return attachmentEndedOutcome()
830	846		}
831	847		if result.adopted && result.event.Kind == AttachmentEventLost && run.fg.Attached() {
832	848		// The stream carrying a live attachment was lost while the broker stayed

@@ -834,23 +850,20 @@ settlement:
834	850		// session and tab that were showing instead of dropping to the picker.
835	851		if resume, ok := resumeTarget(request, run.fg); ok {
836	852		s.logger.Warn("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String(), "error", result.event.Err, "cause", errors.Unwrap(result.event.Err), "resume", true)
837		-	s.resume = &resume
838		-	s.resumeErr = result.event.Err
839		-	return false, nil
	853	+	return attachmentOutcome{kind: attachmentResume, target: resume, cause: result.event.Err}
840	854		}
841	855		}
842	856		if result.adopted && (result.event.Kind == AttachmentEventLost \|\| result.event.Kind == AttachmentEventFailed) {
843	857		s.logger.Warn("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String(), "error", result.event.Err, "cause", errors.Unwrap(result.event.Err))
844	858		// A resume attempt whose stream was lost before attaching is still a
845	859		// transport failure: retry instead of presenting it.
846		-	if result.event.Kind == AttachmentEventLost && s.resuming {
847		-	s.resumeErr = result.event.Err
848		-	return false, nil
	860	+	if result.event.Kind == AttachmentEventLost && resuming {
	861	+	return attachmentOutcome{kind: attachmentRetry, cause: result.event.Err}
849	862		}
850	863		s.transition(supervisorEvent{kind: supervisorAttachEnded, err: result.event.Err})
851	864		s.notifyLifecycle(LifecycleNoticeDestinationFailed)
852	865		s.notifyAttachment(result.event.Err)
853		-	return false, nil
	866	+	return attachmentEndedOutcome()
854	867		}
855	868		// An orderly end, or a run the supervisor retired for a broker loss,
856	869		// returns to the picker with no attachment failure reported. A stream that

@@ -861,7 +874,7 @@ settlement:
861	874		s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String())
862	875		s.notifyLifecycle(LifecycleNoticeSessionEnded)
863	876		}
864		-	return false, nil
	877	+	return attachmentEndedOutcome()
865	878		}
866	879
867	880		// maxAttachmentResumes bounds how many times one lost attachment reconnects

@@ -916,16 +929,25 @@ func (s Supervisor) waitResume(ctx context.Context, input terminalInputLifetim
916	929		}
917	930		}
918	931
919		-	// deferResumeFailure records a transport-class failure of a resume attempt so
920		-	// runResolvedAttachment retries it while Connecting stays up. Any other
921		-	// failure, or a failure outside a resume, is presented by the caller.
922		-	func (s *Supervisor) deferResumeFailure(err error) bool {
923		-	if !s.resuming \|\| !resumableFailure(err) {
924		-	return false
	932	+	// resumeRetry turns a transport-class failure of a resume attempt into an
	933	+	// attachmentRetry outcome, so runResolvedAttachment retries while Connecting
	934	+	// stays up. Any other failure, or a failure outside a resume, is not retried.
	935	+	func (s *Supervisor) resumeRetry(err error, resuming bool) (attachmentOutcome, bool) {
	936	+	if !resuming \|\| !resumableFailure(err) {
	937	+	return attachmentOutcome{}, false
925	938		}
926	939		s.logger.Info("client_attachment_resume_failed", "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
927		-	s.resumeErr = err
928		-	return true
	940	+	return attachmentOutcome{kind: attachmentRetry, cause: err}, true
	941	+	}
	942	+
	943	+	// attemptFailed ends one attachment attempt that failed before settlement:
	944	+	// a resume retries a transport-class failure, anything else is presented.
	945	+	func (s *Supervisor) attemptFailed(err error, resuming bool) attachmentOutcome {
	946	+	if outcome, ok := s.resumeRetry(err, resuming); ok {
	947	+	return outcome
	948	+	}
	949	+	s.reportAttachmentFailure(err)
	950	+	return attachmentEndedOutcome()
929	951		}
930	952
931	953		// resumableFailure reports whether a resume attempt failed for a transport

@@ -948,12 +970,8 @@ func resumableFailure(err error) bool {
948	970		}
949	971		}
950	972
951		-	// reportAttachmentFailure returns to the picker with a typed, visible failure,
952		-	// unless a resume attempt defers it for a retry.
	973	+	// reportAttachmentFailure returns to the picker with a typed, visible failure.
953	974		func (s *Supervisor) reportAttachmentFailure(err error) {
954		-	if s.deferResumeFailure(err) {
955		-	return
956		-	}
957	975		s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err})
958	976		// Classification order matters: a local refusal never dialed a destination,
959	977		// so it must not be reported as one, and only a genuinely unknown outcome

 	adopted bool
 }
+// attachmentOutcomeKind is what one attachment attempt tells the resume loop
+// to do next.
+type attachmentOutcomeKind uint8
+const (
+	// attachmentEnded returns to the picker; any failure is already presented.
+	attachmentEnded attachmentOutcomeKind = iota
+	// attachmentTerminated ends the whole run (cancellation, terminal EOF, or
+	// detach-and-exit); cause is the run's termination error, possibly nil.
+	attachmentTerminated
+	// attachmentSwap attaches to target, which the picker overlay or a daemon
+	// navigation committed while this attachment was live.
+	attachmentSwap
+	// attachmentResume reconnects to target after a live attachment was lost
+	// with cause.
+	attachmentResume
+	// attachmentRetry repeats a resume attempt that hit the transport-class
+	// failure cause before attaching.
+	attachmentRetry
+)
+// attachmentOutcome is the complete result of one attachment attempt.
+type attachmentOutcome struct {
+	kind   attachmentOutcomeKind
+	target pickerAttachmentTarget
+	cause  error
+}
+func attachmentEndedOutcome() attachmentOutcome { return attachmentOutcome{kind: attachmentEnded} }
+func attachmentTerminatedOutcome(err error) attachmentOutcome {
+	return attachmentOutcome{kind: attachmentTerminated, cause: err}
+}
 // resolveCommittedStreamRequest turns exactly one committed catalogue key into
 // the exact broker stream request the user committed. The stream ID is
 // allocated by the broker service — the single allocator on its own connection
 		picker.SetOwnsInput(true)
 		input.acquirePicker()
 	}()
-	defer func() { s.resuming, s.resumeErr, s.resume = false, nil, nil }()
+	// resuming is true while the loop reconnects a lost attachment: attempts
+	// then return transport-class failures as attachmentRetry instead of
+	// presenting them, so a retry stays on Connecting.
+	resuming := false
 	resumes := 0
 	for {
-		s.resumeErr = nil
 		attemptStart := s.cfg.Clock.Now()
-		terminated, termErr, swap := s.attachOnce(ctx, input, service, target, localProvenance)
+		outcome := s.attachResolved(ctx, input, service, target, localProvenance, resuming)
-		if terminated {
+		switch outcome.kind {
-			return terminated, termErr
+		case attachmentTerminated:
-		}
+			return true, outcome.cause
-		if swap == nil {
+		case attachmentEnded:
-			resume := s.resume
+			// The attempt already presented a final outcome (session gone,
-			s.resume = nil
+			// ended, refused).
-			switch {
+			return false, nil
-			case resume != nil:
+		case attachmentSwap:
-				// A live attachment was lost. One that stayed up long enough
+			resumes = 0
-				// starts a fresh resume budget; one that drops right after
+			resuming = false
-				// attaching keeps counting, so a flapping session gives up.
+			// A commit from the picker overlay to another target: the previous
-				if s.cfg.Clock.Now().Sub(attemptStart) >= resumeStableAttachment {
+			// attachment detached cleanly, and the swap goes through Connecting
-					resumes = 0
+			// exactly like any committed selection. The stream identity is
-				}
+			// allocated now, not at commit: streams opened while the source was
-			case s.resuming && s.resumeErr != nil:
+			// live (route publications) may have moved the connection's
-				// A resume attempt hit a transport-class failure before
+			// anti-replay window past an identity reserved earlier.
-				// attaching (the route or destination is still down): try the
-				// same target again.
-				retry := target
-				resume = &retry
-			default:
-				// No resume, or the attempt already presented a final outcome
-				// (session gone, ended, refused).
-				return false, nil
-			}
-			lastErr := s.resumeErr
-			resumes++
-			if resumes > maxAttachmentResumes {
-				s.logger.Warn("client_attachment_resume_exhausted", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempts", resumes-1, "error", lastErr)
-				s.resuming = false
-				s.transition(supervisorEvent{kind: supervisorAttachEnded, err: lastErr})
-				s.notifyLifecycle(LifecycleNoticeDestinationFailed)
-				s.notifyAttachment(lastErr)
-				return false, nil
-			}
-			s.resuming = true
-			if terminated, termErr, brokerLost := s.waitResume(ctx, input, service, resumes); terminated || brokerLost {
-				return terminated, termErr
-			}
 			stream, err := service.NextStreamID()
 			if err != nil {
-				s.resuming = false
 				s.reportAttachmentFailure(err)
 				return false, nil
 			}
-			resume.request.Stream = stream
+			target, localProvenance = outcome.target, SessionEnvironmentLocalPicker
-			s.logger.Info("client_attachment_resume", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempt", resumes)
+			target.request.Stream = stream
-			target = *resume
 			continue
+		case attachmentResume:
+			// A live attachment was lost. One that stayed up long enough
+			// starts a fresh resume budget; one that drops right after
+			// attaching keeps counting, so a flapping session gives up.
+			if s.cfg.Clock.Now().Sub(attemptStart) >= resumeStableAttachment {
+				resumes = 0
+			}
+		case attachmentRetry:
+			// A resume attempt hit a transport-class failure before
+			// attaching (the route or destination is still down): try the
+			// same target again.
+			outcome.target = target
+		}
+		resume, lastErr := outcome.target, outcome.cause
+		resumes++
+		if resumes > maxAttachmentResumes {
+			s.logger.Warn("client_attachment_resume_exhausted", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempts", resumes-1, "error", lastErr)
+			s.transition(supervisorEvent{kind: supervisorAttachEnded, err: lastErr})
+			s.notifyLifecycle(LifecycleNoticeDestinationFailed)
+			s.notifyAttachment(lastErr)
+			return false, nil
+		}
+		resuming = true
+		if terminated, termErr, brokerLost := s.waitResume(ctx, input, service, resumes); terminated || brokerLost {
+			return terminated, termErr
 		}
-		resumes = 0
-		s.resuming = false
-		// A commit from the picker overlay to another target: the previous
-		// attachment detached cleanly, and the swap goes through Connecting
-		// exactly like any committed selection. The stream identity is
-		// allocated now, not at commit: streams opened while the source was
-		// live (route publications) may have moved the connection's
-		// anti-replay window past an identity reserved earlier.
 		stream, err := service.NextStreamID()
 		if err != nil {
 			s.reportAttachmentFailure(err)
 			return false, nil
 		}
-		swap.request.Stream = stream
+		resume.request.Stream = stream
-		target, localProvenance = *swap, SessionEnvironmentLocalPicker
+		s.logger.Info("client_attachment_resume", "endpoint", resume.request.Endpoint, "session", resume.request.Target.SessionName, "attempt", resumes)
-	}
+		target = resume
-}
-// attachOnce runs one resolved attachment end to end. A non-nil swap names the
-// next target the picker overlay committed while this attachment was live.
-func (s *Supervisor) attachOnce(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance) (bool, error, *pickerAttachmentTarget) {
-	terminated, termErr := s.attachResolved(ctx, input, service, target, localProvenance)
-	swap := s.pendingSwap
-	s.pendingSwap = nil
-	if terminated {
-		return terminated, termErr, nil
 	}
-	return false, nil, swap
 }
-func (s *Supervisor) attachResolved(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance) (bool, error) {
+// attachResolved runs one resolved attachment attempt end to end: open the
+// exact stream under one absolute deadline, admit the worker through the
+// foreground grant, settle it, and classify the result. resuming marks an
+// attempt of the resume loop, whose transport-class failures are returned as
+// attachmentRetry instead of presented.
+func (s *Supervisor) attachResolved(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance, resuming bool) attachmentOutcome {
 	request := target.request
 	s.transition(supervisorEvent{kind: supervisorAttachBegin})
 	deadline := startAttachmentDeadline(ctx, s.cfg.Clock)
 		sessionEnv.Cwd = ""
 	}
 	if err := sessionEnv.Validate(); err != nil {
-		s.reportAttachmentFailure(err)
+		return s.attemptFailed(err, resuming)
-		return false, nil
 	}
 	request.Env = append([]string(nil), sessionEnv.Env...)
 	stream, err := service.OpenStream(deadline.Context(), request)
 	if err != nil {
 		s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
-		s.reportAttachmentFailure(deadline.cause(err))
+		return s.attemptFailed(deadline.cause(err), resuming)
-		return false, nil
 	}
 	if supervisorNil(stream) {
-		s.reportAttachmentFailure(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"})
+		return s.attemptFailed(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"}, resuming)
-		return false, nil
 	}
 	if err := deadline.adopt(stream); err != nil {
 		// The accepted absolute deadline already elapsed before this client
 		// saw the stream. Fail the attachment instead of restarting a budget.
 		_ = stream.Close()
-		s.reportAttachmentFailure(attachmentTimeoutError(err))
+		return s.attemptFailed(attachmentTimeoutError(err), resuming)
-		return false, nil
 	}
 	worker, err := s.newAttachmentWorker(request, target.tab, func() error {
 	}, sessionEnv)
 	if err != nil {
 		_ = stream.Close()
-		s.reportAttachmentFailure(err)
+		return s.attemptFailed(err, resuming)
-		return false, nil
 	}
 	token := AttachmentToken{Generation: s.State().Generation, Attempt: s.nextAttachment.Add(1)}
 		// Another foreground owns the terminal, or the shared reader is
 		// claimed. Begin never closes the stream, so the supervisor does.
 		_ = stream.Close()
-		if s.resuming {
+		if resuming {
 			// A resume must say why it stopped instead of dropping silently
-			// to the picker.
+			// to the picker, so the busy error is presented, never retried.
-			s.resuming = false
 			s.reportAttachmentFailure(errResumeForegroundBusy)
-			return false, nil
+			return attachmentEndedOutcome()
 		}
 		s.transition(supervisorEvent{kind: supervisorAttachEnded})
-		return false, nil
+		return attachmentEndedOutcome()
 	}
-	return s.settleAttachment(ctx, input, service, deadline, run, request)
+	return s.settleAttachment(ctx, input, service, deadline, run, request, resuming)
 }
 // settleAttachment joins one admitted attachment while watching the parent
 // context, terminal EOF, and broker-connection loss. The deadline is the only
 // other stop: when it fires before the worker settled, the attachment ends as
 // a typed timeout without claiming attachment.
-func (s *Supervisor) settleAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService, deadline *attachmentDeadline, run *attachmentRun, request ports.BrokerOpenStreamRequest) (bool, error) {
+func (s *Supervisor) settleAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService, deadline *attachmentDeadline, run *attachmentRun, request ports.BrokerOpenStreamRequest, resuming bool) attachmentOutcome {
 	settled := make(chan attachmentSettlement, 1)
 	go func() {
 		event, adopted := run.Wait(deadline.Context())
 	if !terminated && !brokerLost {
 		s.takeSettledNavigation(service, run.token, request)
 	}
+	// The overlay or a daemon navigation may have committed another target
+	// while this attachment was live; this attempt consumes it either way.
+	swap := s.pendingSwap
+	s.pendingSwap = nil
 	if terminated {
-		s.pendingSwap = nil
+		return attachmentTerminatedOutcome(termErr)
-		return true, termErr
+	}
-	}
+	if swap != nil && !brokerLost {
-	if s.pendingSwap != nil {
+		// The overlay committed another target and this attachment detached
-		if !brokerLost {
+		// for it: go straight to Connecting for the swap, never through the
-			// The overlay committed another target and this attachment
+		// plain picker.
-			// detached for it: go straight to Connecting for the swap, never
+		return attachmentOutcome{kind: attachmentSwap, target: *swap}
-			// through the plain picker.
-			return false, nil
-		}
-		s.pendingSwap = nil
 	}
 	if deadline.TimedOut() && !result.adopted {
 		timeout := attachmentTimeoutError(errAttachmentDeadline)
-		if s.deferResumeFailure(timeout) {
+		if outcome, ok := s.resumeRetry(timeout, resuming); ok {
-			return false, nil
+			return outcome
 		}
 		s.transition(supervisorEvent{kind: supervisorAttachEnded, err: timeout})
 		s.notifyAttachment(timeout)
-		return false, nil
+		return attachmentEndedOutcome()
 	}
 	if errors.Is(result.event.Err, errDetachAndExit) {
 		s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "explicit_detach")
 		s.transition(supervisorEvent{kind: supervisorAttachEnded})
 		s.notifyLifecycle(LifecycleNoticeDetachAndExit)
-		return true, nil
+		return attachmentTerminatedOutcome(nil)
 	}
 	if errors.Is(result.event.Err, errDetachToPicker) {
 		s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "detach_to_picker")
 		s.transition(supervisorEvent{kind: supervisorAttachEnded})
 		s.notifyLifecycle(LifecycleNoticeDetachToPicker)
-		return false, nil
+		return attachmentEndedOutcome()
 	}
 	if result.adopted && result.event.Kind == AttachmentEventLost && run.fg.Attached() {
 		// The stream carrying a live attachment was lost while the broker stayed
 		// session and tab that were showing instead of dropping to the picker.
 		if resume, ok := resumeTarget(request, run.fg); ok {
 			s.logger.Warn("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String(), "error", result.event.Err, "cause", errors.Unwrap(result.event.Err), "resume", true)
-			s.resume = &resume
+			return attachmentOutcome{kind: attachmentResume, target: resume, cause: result.event.Err}
-			s.resumeErr = result.event.Err
-			return false, nil
 		}
 	}
 	if result.adopted && (result.event.Kind == AttachmentEventLost || result.event.Kind == AttachmentEventFailed) {
 		s.logger.Warn("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String(), "error", result.event.Err, "cause", errors.Unwrap(result.event.Err))
 		// A resume attempt whose stream was lost before attaching is still a
 		// transport failure: retry instead of presenting it.
-		if result.event.Kind == AttachmentEventLost && s.resuming {
+		if result.event.Kind == AttachmentEventLost && resuming {
-			s.resumeErr = result.event.Err
+			return attachmentOutcome{kind: attachmentRetry, cause: result.event.Err}
-			return false, nil
 		}
 		s.transition(supervisorEvent{kind: supervisorAttachEnded, err: result.event.Err})
 		s.notifyLifecycle(LifecycleNoticeDestinationFailed)
 		s.notifyAttachment(result.event.Err)
-		return false, nil
+		return attachmentEndedOutcome()
 	}
 	// An orderly end, or a run the supervisor retired for a broker loss,
 	// returns to the picker with no attachment failure reported. A stream that
 		s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String())
 		s.notifyLifecycle(LifecycleNoticeSessionEnded)
 	}
-	return false, nil
+	return attachmentEndedOutcome()
 }
 // maxAttachmentResumes bounds how many times one lost attachment reconnects
 	}
 }
-// deferResumeFailure records a transport-class failure of a resume attempt so
+// resumeRetry turns a transport-class failure of a resume attempt into an
-// runResolvedAttachment retries it while Connecting stays up. Any other
+// attachmentRetry outcome, so runResolvedAttachment retries while Connecting
-// failure, or a failure outside a resume, is presented by the caller.
+// stays up. Any other failure, or a failure outside a resume, is not retried.
-func (s *Supervisor) deferResumeFailure(err error) bool {
+func (s *Supervisor) resumeRetry(err error, resuming bool) (attachmentOutcome, bool) {
-	if !s.resuming || !resumableFailure(err) {
+	if !resuming || !resumableFailure(err) {
-		return false
+		return attachmentOutcome{}, false
 	}
 	s.logger.Info("client_attachment_resume_failed", "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
-	s.resumeErr = err
+	return attachmentOutcome{kind: attachmentRetry, cause: err}, true
-	return true
+}
+// attemptFailed ends one attachment attempt that failed before settlement:
+// a resume retries a transport-class failure, anything else is presented.
+func (s *Supervisor) attemptFailed(err error, resuming bool) attachmentOutcome {
+	if outcome, ok := s.resumeRetry(err, resuming); ok {
+		return outcome
+	}
+	s.reportAttachmentFailure(err)
+	return attachmentEndedOutcome()
 }
 // resumableFailure reports whether a resume attempt failed for a transport
 	}
 }
-// reportAttachmentFailure returns to the picker with a typed, visible failure,
+// reportAttachmentFailure returns to the picker with a typed, visible failure.
-// unless a resume attempt defers it for a retry.
 func (s *Supervisor) reportAttachmentFailure(err error) {
-	if s.deferResumeFailure(err) {
-		return
-	}
 	s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err})
 	// Classification order matters: a local refusal never dialed a destination,
 	// so it must not be reported as one, and only a genuinely unknown outcome

M internal/usecase/client/supervisor_attach_test.go +17 -1


@@ -1334,6 +1334,8 @@ func TestSupervisorAttachmentLossResumesSameSession(t *testing.T) {
1334	1334		loseBeforeAttach
1335	1335		// refuse fails the resumed open with a non-transport refusal.
1336	1336		refuse
	1337	+	// openUnavailable fails the resumed open with a transport-class error.
	1338	+	openUnavailable
1337	1339		// cancelDuringBackoff ends the client while the backoff runs.
1338	1340		cancelDuringBackoff
1339	1341		)

@@ -1357,6 +1359,8 @@ func TestSupervisorAttachmentLossResumesSameSession(t *testing.T) {
1357	1359		{name: "transient failures then attach", steps: append(repeat(loseBeforeAttach, maxAttachmentResumes-1), attachAndStay)},
1358	1360		{name: "exhausted returns to picker", steps: repeat(loseBeforeAttach, maxAttachmentResumes), wantPicker: true},
1359	1361		{name: "refusal is final", steps: []step{refuse}, wantPicker: true},
	1362	+	{name: "unavailable opens then attach", steps: append(repeat(openUnavailable, maxAttachmentResumes-1), attachAndStay)},
	1363	+	{name: "unavailable opens exhaust", steps: repeat(openUnavailable, maxAttachmentResumes), wantPicker: true},
1360	1364		{name: "cancel during backoff never resumes", steps: []step{cancelDuringBackoff}},
1361	1365		}
1362	1366		for _, tc := range cases {

@@ -1368,11 +1372,14 @@ func TestSupervisorAttachmentLossResumesSameSession(t *testing.T) {
1368	1372		cfg.Jitter = resumeTestJitter
1369	1373		})
1370	1374		streams := make(chan *sessionTestStream, len(tc.steps)+2)
1371		-	var refusing atomic.Bool
	1375	+	var refusing, unavailable atomic.Bool
1372	1376		harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1373	1377		if refusing.Load() {
1374	1378		return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "session gone"}
1375	1379		}
	1380	+	if unavailable.Load() {
	1381	+	return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "route down"}
	1382	+	}
1376	1383		stream := newSessionTestStream()
1377	1384		streams <- stream
1378	1385		return stream, nil

@@ -1395,10 +1402,19 @@ func TestSupervisorAttachmentLossResumesSameSession(t *testing.T) {
1395	1402		return
1396	1403		}
1397	1404		refusing.Store(s == refuse)
	1405	+	unavailable.Store(s == openUnavailable)
	1406	+	opened := len(harness.service.openedRequests())
1398	1407		fireResumeTimer(t, harness.clock, attempt)
1399	1408		if s == refuse {
1400	1409		break
1401	1410		}
	1411	+	if s == openUnavailable {
	1412	+	// The failed open is retried on Connecting, without a notice.
	1413	+	// Wait for it before the next step changes the open result.
	1414	+	require.Eventually(t, func() bool { return len(harness.service.openedRequests()) > opened }, 5*time.Second, time.Millisecond)
	1415	+	attempt++
	1416	+	continue
	1417	+	}
1402	1418		resumed := <-streams
1403	1419		if s == loseBeforeAttach {
1404	1420		resumed.fail(testStreamLoss())

 		loseBeforeAttach
 		// refuse fails the resumed open with a non-transport refusal.
 		refuse
+		// openUnavailable fails the resumed open with a transport-class error.
+		openUnavailable
 		// cancelDuringBackoff ends the client while the backoff runs.
 		cancelDuringBackoff
 	)
 		{name: "transient failures then attach", steps: append(repeat(loseBeforeAttach, maxAttachmentResumes-1), attachAndStay)},
 		{name: "exhausted returns to picker", steps: repeat(loseBeforeAttach, maxAttachmentResumes), wantPicker: true},
 		{name: "refusal is final", steps: []step{refuse}, wantPicker: true},
+		{name: "unavailable opens then attach", steps: append(repeat(openUnavailable, maxAttachmentResumes-1), attachAndStay)},
+		{name: "unavailable opens exhaust", steps: repeat(openUnavailable, maxAttachmentResumes), wantPicker: true},
 		{name: "cancel during backoff never resumes", steps: []step{cancelDuringBackoff}},
 	}
 	for _, tc := range cases {
 				cfg.Jitter = resumeTestJitter
 			})
 			streams := make(chan *sessionTestStream, len(tc.steps)+2)
-			var refusing atomic.Bool
+			var refusing, unavailable atomic.Bool
 			harness.service.setOpenStream(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
 				if refusing.Load() {
 					return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "session gone"}
 				}
+				if unavailable.Load() {
+					return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "route down"}
+				}
 				stream := newSessionTestStream()
 				streams <- stream
 				return stream, nil
 					return
 				}
 				refusing.Store(s == refuse)
+				unavailable.Store(s == openUnavailable)
+				opened := len(harness.service.openedRequests())
 				fireResumeTimer(t, harness.clock, attempt)
 				if s == refuse {
 					break
 				}
+				if s == openUnavailable {
+					// The failed open is retried on Connecting, without a notice.
+					// Wait for it before the next step changes the open result.
+					require.Eventually(t, func() bool { return len(harness.service.openedRequests()) > opened }, 5*time.Second, time.Millisecond)
+					attempt++
+					continue
+				}
 				resumed := <-streams
 				if s == loseBeforeAttach {
 					resumed.fail(testStreamLoss())

M internal/usecase/daemon/copy_scroll_ownership_test.go +2 -2


@@ -35,7 +35,7 @@ func TestCopyCacheFailedPublicationAndRetry(t *testing.T) {
35	35		ac.replaceTransport(cacheFailTransport{})
36	36		}
37	37		ac.sendMu.Lock()
38		-	require.True(t, d.emitFrame(sess, ac, &state, pending))
	38	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
39	39		require.Equal(t, before, cloneComposeCache(ac.pipelineCache))
40	40		require.Zero(t, ac.output.next)
41	41		if failure == "send" {

@@ -48,7 +48,7 @@ func TestCopyCacheFailedPublicationAndRetry(t *testing.T) {
48	48		pending = composeFrame(state, ac.pipelineCache, ac.pipelineScratch)
49	49		state.view.revision = ac.viewSnapshot().revision
50	50		ac.sendMu.Lock()
51		-	require.True(t, d.emitFrame(sess, ac, &state, pending))
	51	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
52	52		require.Equal(t, pending.cache, ac.pipelineCache)
53	53		require.NotContains(t, frameText(ac.pipelineCache.frame), "COPY")
54	54		output := unmarshalTestOutput(t, (<-sends).Payload)


@@ -35,7 +35,7 @@ func TestCopyCacheFailedPublicationAndRetry(t *testing.T) {
35		ac.replaceTransport(cacheFailTransport{})	35		ac.replaceTransport(cacheFailTransport{})
36		}	36		}
37		ac.sendMu.Lock()	37		ac.sendMu.Lock()
38	-	require.True(t, d.emitFrame(sess, ac, &state, pending))	38	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
39		require.Equal(t, before, cloneComposeCache(ac.pipelineCache))	39		require.Equal(t, before, cloneComposeCache(ac.pipelineCache))
40		require.Zero(t, ac.output.next)	40		require.Zero(t, ac.output.next)
41		if failure == "send" {	41		if failure == "send" {

@@ -48,7 +48,7 @@ func TestCopyCacheFailedPublicationAndRetry(t *testing.T) {
48		pending = composeFrame(state, ac.pipelineCache, ac.pipelineScratch)	48		pending = composeFrame(state, ac.pipelineCache, ac.pipelineScratch)
49		state.view.revision = ac.viewSnapshot().revision	49		state.view.revision = ac.viewSnapshot().revision
50		ac.sendMu.Lock()	50		ac.sendMu.Lock()
51	-	require.True(t, d.emitFrame(sess, ac, &state, pending))	51	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
52		require.Equal(t, pending.cache, ac.pipelineCache)	52		require.Equal(t, pending.cache, ac.pipelineCache)
53		require.NotContains(t, frameText(ac.pipelineCache.frame), "COPY")	53		require.NotContains(t, frameText(ac.pipelineCache.frame), "COPY")
54		output := unmarshalTestOutput(t, (<-sends).Payload)	54		output := unmarshalTestOutput(t, (<-sends).Payload)

M internal/usecase/daemon/render.go +31 -15


@@ -347,10 +347,24 @@ func (d Daemon) paintWithActivationEffect(entry session, ac *attachedClient, r
347	347		}
348	348
349	349		ac.sendMu.Lock()
	350	+	result, afterUnlock := d.paintLocked(entry, ac, tb, reset, lease, paintEffect, &marks)
	351	+	ac.sendMu.Unlock()
	352	+	afterUnlock()
	353	+	return result
	354	+	}
	355	+
	356	+	func noAfterUnlock() {}
	357	+
	358	+	// paintLocked is the complete capture/compose/publish transaction. It runs with
	359	+	// ac.sendMu held and never releases it; paint is the sole owner of that lock.
	360	+	// The returned function runs after sendMu is released, with no attachment,
	361	+	// session, tab, or pane lock held.
	362	+	func (d Daemon) paintLocked(entry session, ac attachedClient, tb tab, reset bool, lease attachmentLease, paintEffect attachmentEffect, marks *runtimeMarkBatch) (paintResult, func()) {
	363	+	sess := entry
	364	+	local := sess != nil
350	365		if paintEffect != nil {
351	366		if !paintEffect.current() {
352		-	ac.sendMu.Unlock()
353		-	return paintRejected
	367	+	return paintRejected, noAfterUnlock
354	368		}
355	369		} else {
356	370		// Direct/headless rendering has no Attachment effect and retains its

@@ -359,8 +373,7 @@ func (d Daemon) paintWithActivationEffect(entry session, ac *attachedClient, r
359	373		_, owned := entry.core().attachments[ac]
360	374		entry.core().mu.Unlock()
361	375		if !owned \|\| ac.currentAttachmentSession() != entry {
362		-	ac.sendMu.Unlock()
363		-	return paintRejected
	376	+	return paintRejected, noAfterUnlock
364	377		}
365	378		}
366	379		// Capacity is checked before any destructive capture. Refresh the atomic

@@ -376,8 +389,7 @@ func (d Daemon) paintWithActivationEffect(entry session, ac *attachedClient, r
376	389		if ac.output != nil && ac.output.atCapacity() {
377	390		// The coordinator owns the blocked interval; this guard only protects
378	391		// direct test/resize paint calls from destructive capture.
379		-	ac.sendMu.Unlock()
380		-	return paintBlockedCapacity
	392	+	return paintBlockedCapacity, noAfterUnlock
381	393		}
382	394		var uiFence uint64
383	395		if lease != nil {

@@ -392,17 +404,22 @@ func (d Daemon) paintWithActivationEffect(entry session, ac *attachedClient, r
392	404		frameCapture := ac.afterFrame.beginCapture()
393	405		overlays := ac.overlays.SnapshotForRender()
394	406		repaintAttachedClients := false
395		-	defer func() {
	407	+	var published bool
	408	+	var followUp frameFollowUp
	409	+	afterUnlock := func() {
	410	+	// Frame follow-up (notices, repaints, send-error cleanup) runs first,
	411	+	// then the captured sequence is emitted before any follow-up repaint can
	412	+	// introduce another boundary.
	413	+	if published {
	414	+	d.settleFrame(entry, ac, marks, followUp)
	415	+	}
396	416		overlays.Unlock()
397		-	// Every return below has released sendMu before reaching this defer.
398		-	// Emit the captured sequence before any follow-up repaint can introduce
399		-	// another boundary, while no attachment, session, tab, or pane lock is held.
400	417		marks.flush()
401	418		ac.afterFrame.runReady()
402	419		if repaintAttachedClients {
403	420		d.repaintAllAttachedClients()
404	421		}
405		-	}()
	422	+	}
406	423		if local {
407	424		d.refreshSessionFocusedTitles(sess)
408	425		}

@@ -460,8 +477,7 @@ func (d Daemon) paintWithActivationEffect(entry session, ac *attachedClient, r
460	477		})
461	478		endCapture(0, ok)
462	479		if !ok {
463		-	ac.sendMu.Unlock()
464		-	return paintRejected
	480	+	return paintRejected, afterUnlock
465	481		}
466	482		if ac.renderStages.capture != nil {
467	483		ac.renderStages.capture()

@@ -474,8 +490,8 @@ func (d Daemon) paintWithActivationEffect(entry session, ac *attachedClient, r
474	490		if ac.renderStages.compose != nil {
475	491		ac.renderStages.compose()
476	492		}
477		-	d.emitFrame(entry, ac, state, composed, &marks)
478		-	return paintEmitted
	493	+	followUp, published = d.publishFrameLocked(entry, ac, state, composed, marks)
	494	+	return paintEmitted, afterUnlock
479	495		}
480	496
481	497		var fallbackChromeStyles = themeui.Resolve(themeui.Theme{}, domain.ThemeAccent{Mode: domain.ThemeAccentAuto}).Styles

 	}
 	ac.sendMu.Lock()
+	result, afterUnlock := d.paintLocked(entry, ac, tb, reset, lease, paintEffect, &marks)
+	ac.sendMu.Unlock()
+	afterUnlock()
+	return result
+}
+func noAfterUnlock() {}
+// paintLocked is the complete capture/compose/publish transaction. It runs with
+// ac.sendMu held and never releases it; paint is the sole owner of that lock.
+// The returned function runs after sendMu is released, with no attachment,
+// session, tab, or pane lock held.
+func (d *Daemon) paintLocked(entry *session, ac *attachedClient, tb *tab, reset bool, lease *attachmentLease, paintEffect *attachmentEffect, marks *runtimeMarkBatch) (paintResult, func()) {
+	sess := entry
+	local := sess != nil
 	if paintEffect != nil {
 		if !paintEffect.current() {
-			ac.sendMu.Unlock()
+			return paintRejected, noAfterUnlock
-			return paintRejected
 		}
 	} else {
 		// Direct/headless rendering has no Attachment effect and retains its
 		_, owned := entry.core().attachments[ac]
 		entry.core().mu.Unlock()
 		if !owned || ac.currentAttachmentSession() != entry {
-			ac.sendMu.Unlock()
+			return paintRejected, noAfterUnlock
-			return paintRejected
 		}
 	}
 	// Capacity is checked before any destructive capture. Refresh the atomic
 	if ac.output != nil && ac.output.atCapacity() {
 		// The coordinator owns the blocked interval; this guard only protects
 		// direct test/resize paint calls from destructive capture.
-		ac.sendMu.Unlock()
+		return paintBlockedCapacity, noAfterUnlock
-		return paintBlockedCapacity
 	}
 	var uiFence uint64
 	if lease != nil {
 	frameCapture := ac.afterFrame.beginCapture()
 	overlays := ac.overlays.SnapshotForRender()
 	repaintAttachedClients := false
-	defer func() {
+	var published bool
+	var followUp frameFollowUp
+	afterUnlock := func() {
+		// Frame follow-up (notices, repaints, send-error cleanup) runs first,
+		// then the captured sequence is emitted before any follow-up repaint can
+		// introduce another boundary.
+		if published {
+			d.settleFrame(entry, ac, marks, followUp)
+		}
 		overlays.Unlock()
-		// Every return below has released sendMu before reaching this defer.
-		// Emit the captured sequence before any follow-up repaint can introduce
-		// another boundary, while no attachment, session, tab, or pane lock is held.
 		marks.flush()
 		ac.afterFrame.runReady()
 		if repaintAttachedClients {
 			d.repaintAllAttachedClients()
 		}
-	}()
+	}
 	if local {
 		d.refreshSessionFocusedTitles(sess)
 	}
 	})
 	endCapture(0, ok)
 	if !ok {
-		ac.sendMu.Unlock()
+		return paintRejected, afterUnlock
-		return paintRejected
 	}
 	if ac.renderStages.capture != nil {
 		ac.renderStages.capture()
 	if ac.renderStages.compose != nil {
 		ac.renderStages.compose()
 	}
-	d.emitFrame(entry, ac, state, composed, &marks)
+	followUp, published = d.publishFrameLocked(entry, ac, state, composed, marks)
-	return paintEmitted
+	return paintEmitted, afterUnlock
 }
 var fallbackChromeStyles = themeui.Resolve(themeui.Theme{}, domain.ThemeAccent{Mode: domain.ThemeAccentAuto}).Styles

M internal/usecase/daemon/render_capture.go +1 -1


@@ -238,7 +238,7 @@ func capturePaneRenderStateLockedInto(p *pane, visible domain.Rect, out captured
238	238		} else {
239	239		out.damage = append(out.damage[:0], damage.Damage...)
240	240		}
241		-	// Consumption is transactional: capture only snapshots damage. emitFrame
	241	+	// Consumption is transactional: capture only snapshots damage. publishFrameLocked
242	242		// acknowledges this generation after preparation and transport success.
243	243		return out
244	244		}


@@ -238,7 +238,7 @@ func capturePaneRenderStateLockedInto(p *pane, visible domain.Rect, out captured
238		} else {	238		} else {
239		out.damage = append(out.damage[:0], damage.Damage...)	239		out.damage = append(out.damage[:0], damage.Damage...)
240		}	240		}
241	-	// Consumption is transactional: capture only snapshots damage. emitFrame	241	+	// Consumption is transactional: capture only snapshots damage. publishFrameLocked
242		// acknowledges this generation after preparation and transport success.	242		// acknowledges this generation after preparation and transport success.
243		return out	243		return out
244		}	244		}

M internal/usecase/daemon/render_damage_receipt_test.go +11 -11


@@ -90,7 +90,7 @@ func TestCaptureComposeEmitAcknowledgesCollapsedPaneDamageOnlyAfterEmission(t *t
90	90		require.True(t, collapsedReceipt, "capture must retain a receipt for the collapsed pane")
91	91		require.NotEmpty(t, collapsed.screen.Damage(), "capture alone must not acknowledge collapsed-pane damage")
92	92		composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch)
93		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	93	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
94	94
95	95		<-sends
96	96		require.Empty(t, collapsed.screen.Damage(), "successful emission must acknowledge the collapsed-pane receipt")

@@ -140,14 +140,14 @@ func TestRenderDamageReceiptRetainsRealVTDamageAcrossFailedEmission(t *testing.T
140	140
141	141		state, composed := captureComposeForReceiptTest(t, sess, ac)
142	142		tt.fail(&composed, ac)
143		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	143	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
144	144		p.mu.Lock()
145	145		require.NotEmpty(t, p.screen.Damage(), "failed emission must retain real VT damage")
146	146		p.mu.Unlock()
147	147
148	148		tt.restore(sess, ac, healthy)
149	149		state, composed = captureComposeForReceiptTest(t, sess, ac)
150		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	150	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
151	151		frame := <-sends
152	152		out := unmarshalTestOutput(t, frame.Payload)
153	153		require.Contains(t, string(out.Data), "changed", "retry must emit the retained terminal bytes")

@@ -166,17 +166,17 @@ func TestPrepareFailureNotifiesAndSchedulesRecovery(t *testing.T) {
166	166		var producers []string
167	167		rc.opts.onInvalidate = func(inv renderInvalidation) {
168	168		// Both the notice repaint and the explicit recovery must re-enter only
169		-	// after emitFrame releases its attachment locks.
	169	+	// after the publication releases its attachment locks.
170	170		require.True(t, ac.sendMu.TryLock(), "invalidation must not run under sendMu")
171	171		ac.sendMu.Unlock()
172	172		producers = append(producers, inv.producer)
173	173		}
174	174		state, composed := captureComposeForReceiptTest(t, sess, ac)
175		-	// An invalid frame makes output.prepare fail after capture. emitFrame owns
176		-	// sendMu on entry and releases it before reporting and rescheduling.
	175	+	// An invalid frame makes output.prepare fail after capture. The publication
	176	+	// releases sendMu before reporting and rescheduling.
177	177		composed.frame = renderer.Frame{Width: 1}
178	178
179		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	179	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
180	180
181	181		history := d.notices.history()
182	182		require.Len(t, history, 1)

@@ -203,7 +203,7 @@ func TestPrepareFailureFallbackOnlySuppressesRecursiveNoticePaint(t *testing.T)
203	203		ac.prepareFailureFallback.Store(true)
204	204		state, composed := captureComposeForReceiptTest(t, sess, ac)
205	205		composed.frame = renderer.Frame{Width: 1}
206		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	206	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
207	207		require.Empty(t, d.notices.history())
208	208		ac.prepareFailureFallback.Store(false)
209	209

@@ -212,7 +212,7 @@ func TestPrepareFailureFallbackOnlySuppressesRecursiveNoticePaint(t *testing.T)
212	212		for want := 1; want <= 2; want++ {
213	213		state, composed = captureComposeForReceiptTest(t, sess, ac)
214	214		composed.frame = renderer.Frame{Width: 1}
215		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	215	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
216	216		require.Len(t, d.notices.history(), want)
217	217		frame := <-sends
218	218		require.Equal(t, "Output", envelopeMessageName(t, frame.Payload), "outer failure %d must repaint its notice", want)

@@ -232,7 +232,7 @@ func TestRenderDamageReceiptStaleGenerationForcesFullRedraw(t *testing.T) {
232	232		p.mu.Lock()
233	233		p.screen.Write([]byte("\x1b[2;1Hsecond"))
234	234		p.mu.Unlock()
235		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	235	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
236	236		<-sends
237	237		p.mu.Lock()
238	238		defer p.mu.Unlock()

@@ -241,7 +241,7 @@ func TestRenderDamageReceiptStaleGenerationForcesFullRedraw(t *testing.T) {
241	241
242	242		func captureComposeForReceiptTest(t testing.T, sess session, ac attachedClient) (capturedRenderState, composedRenderFrame) {
243	243		t.Helper()
244		-	ac.sendMu.Lock() // emitFrame releases the transaction lock.
	244	+	ac.sendMu.Lock() // the publication helper releases the transaction lock.
245	245		state, ok := captureRenderState(sess, ac, renderCaptureRequest{
246	246		bars: barState{},
247	247		overlays: capturedOverlayRenderState{},


@@ -90,7 +90,7 @@ func TestCaptureComposeEmitAcknowledgesCollapsedPaneDamageOnlyAfterEmission(t *t
90		require.True(t, collapsedReceipt, "capture must retain a receipt for the collapsed pane")	90		require.True(t, collapsedReceipt, "capture must retain a receipt for the collapsed pane")
91		require.NotEmpty(t, collapsed.screen.Damage(), "capture alone must not acknowledge collapsed-pane damage")	91		require.NotEmpty(t, collapsed.screen.Damage(), "capture alone must not acknowledge collapsed-pane damage")
92		composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch)	92		composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch)
93	-	require.True(t, d.emitFrame(sess, ac, state, composed))	93	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
94			94
95		<-sends	95		<-sends
96		require.Empty(t, collapsed.screen.Damage(), "successful emission must acknowledge the collapsed-pane receipt")	96		require.Empty(t, collapsed.screen.Damage(), "successful emission must acknowledge the collapsed-pane receipt")

@@ -140,14 +140,14 @@ func TestRenderDamageReceiptRetainsRealVTDamageAcrossFailedEmission(t *testing.T
140			140
141		state, composed := captureComposeForReceiptTest(t, sess, ac)	141		state, composed := captureComposeForReceiptTest(t, sess, ac)
142		tt.fail(&composed, ac)	142		tt.fail(&composed, ac)
143	-	require.True(t, d.emitFrame(sess, ac, state, composed))	143	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
144		p.mu.Lock()	144		p.mu.Lock()
145		require.NotEmpty(t, p.screen.Damage(), "failed emission must retain real VT damage")	145		require.NotEmpty(t, p.screen.Damage(), "failed emission must retain real VT damage")
146		p.mu.Unlock()	146		p.mu.Unlock()
147			147
148		tt.restore(sess, ac, healthy)	148		tt.restore(sess, ac, healthy)
149		state, composed = captureComposeForReceiptTest(t, sess, ac)	149		state, composed = captureComposeForReceiptTest(t, sess, ac)
150	-	require.True(t, d.emitFrame(sess, ac, state, composed))	150	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
151		frame := <-sends	151		frame := <-sends
152		out := unmarshalTestOutput(t, frame.Payload)	152		out := unmarshalTestOutput(t, frame.Payload)
153		require.Contains(t, string(out.Data), "changed", "retry must emit the retained terminal bytes")	153		require.Contains(t, string(out.Data), "changed", "retry must emit the retained terminal bytes")

@@ -166,17 +166,17 @@ func TestPrepareFailureNotifiesAndSchedulesRecovery(t *testing.T) {
166		var producers []string	166		var producers []string
167		rc.opts.onInvalidate = func(inv renderInvalidation) {	167		rc.opts.onInvalidate = func(inv renderInvalidation) {
168		// Both the notice repaint and the explicit recovery must re-enter only	168		// Both the notice repaint and the explicit recovery must re-enter only
169	-	// after emitFrame releases its attachment locks.	169	+	// after the publication releases its attachment locks.
170		require.True(t, ac.sendMu.TryLock(), "invalidation must not run under sendMu")	170		require.True(t, ac.sendMu.TryLock(), "invalidation must not run under sendMu")
171		ac.sendMu.Unlock()	171		ac.sendMu.Unlock()
172		producers = append(producers, inv.producer)	172		producers = append(producers, inv.producer)
173		}	173		}
174		state, composed := captureComposeForReceiptTest(t, sess, ac)	174		state, composed := captureComposeForReceiptTest(t, sess, ac)
175	-	// An invalid frame makes output.prepare fail after capture. emitFrame owns	175	+	// An invalid frame makes output.prepare fail after capture. The publication
176	-	// sendMu on entry and releases it before reporting and rescheduling.	176	+	// releases sendMu before reporting and rescheduling.
177		composed.frame = renderer.Frame{Width: 1}	177		composed.frame = renderer.Frame{Width: 1}
178			178
179	-	require.True(t, d.emitFrame(sess, ac, state, composed))	179	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
180			180
181		history := d.notices.history()	181		history := d.notices.history()
182		require.Len(t, history, 1)	182		require.Len(t, history, 1)

@@ -203,7 +203,7 @@ func TestPrepareFailureFallbackOnlySuppressesRecursiveNoticePaint(t *testing.T)
203		ac.prepareFailureFallback.Store(true)	203		ac.prepareFailureFallback.Store(true)
204		state, composed := captureComposeForReceiptTest(t, sess, ac)	204		state, composed := captureComposeForReceiptTest(t, sess, ac)
205		composed.frame = renderer.Frame{Width: 1}	205		composed.frame = renderer.Frame{Width: 1}
206	-	require.True(t, d.emitFrame(sess, ac, state, composed))	206	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
207		require.Empty(t, d.notices.history())	207		require.Empty(t, d.notices.history())
208		ac.prepareFailureFallback.Store(false)	208		ac.prepareFailureFallback.Store(false)
209			209

@@ -212,7 +212,7 @@ func TestPrepareFailureFallbackOnlySuppressesRecursiveNoticePaint(t *testing.T)
212		for want := 1; want <= 2; want++ {	212		for want := 1; want <= 2; want++ {
213		state, composed = captureComposeForReceiptTest(t, sess, ac)	213		state, composed = captureComposeForReceiptTest(t, sess, ac)
214		composed.frame = renderer.Frame{Width: 1}	214		composed.frame = renderer.Frame{Width: 1}
215	-	require.True(t, d.emitFrame(sess, ac, state, composed))	215	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
216		require.Len(t, d.notices.history(), want)	216		require.Len(t, d.notices.history(), want)
217		frame := <-sends	217		frame := <-sends
218		require.Equal(t, "Output", envelopeMessageName(t, frame.Payload), "outer failure %d must repaint its notice", want)	218		require.Equal(t, "Output", envelopeMessageName(t, frame.Payload), "outer failure %d must repaint its notice", want)

@@ -232,7 +232,7 @@ func TestRenderDamageReceiptStaleGenerationForcesFullRedraw(t *testing.T) {
232		p.mu.Lock()	232		p.mu.Lock()
233		p.screen.Write([]byte("\x1b[2;1Hsecond"))	233		p.screen.Write([]byte("\x1b[2;1Hsecond"))
234		p.mu.Unlock()	234		p.mu.Unlock()
235	-	require.True(t, d.emitFrame(sess, ac, state, composed))	235	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
236		<-sends	236		<-sends
237		p.mu.Lock()	237		p.mu.Lock()
238		defer p.mu.Unlock()	238		defer p.mu.Unlock()

@@ -241,7 +241,7 @@ func TestRenderDamageReceiptStaleGenerationForcesFullRedraw(t *testing.T) {
241			241
242		func captureComposeForReceiptTest(t testing.T, sess session, ac attachedClient) (capturedRenderState, composedRenderFrame) {	242		func captureComposeForReceiptTest(t testing.T, sess session, ac attachedClient) (capturedRenderState, composedRenderFrame) {
243		t.Helper()	243		t.Helper()
244	-	ac.sendMu.Lock() // emitFrame releases the transaction lock.	244	+	ac.sendMu.Lock() // the publication helper releases the transaction lock.
245		state, ok := captureRenderState(sess, ac, renderCaptureRequest{	245		state, ok := captureRenderState(sess, ac, renderCaptureRequest{
246		bars: barState{},	246		bars: barState{},
247		overlays: capturedOverlayRenderState{},	247		overlays: capturedOverlayRenderState{},

M internal/usecase/daemon/render_pipeline.go +57 -56


@@ -456,25 +456,25 @@ func commitDamageReceipts(receipts []damageReceipt) {
456	456		}
457	457		}
458	458
459		-	// emitFrame is the sole side-effecting half of the pipeline. The caller holds
460		-	// sendMu for the complete capture/compose/emit transaction.
461		-	func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool {
462		-	var ownedMarks runtimeMarkBatch
463		-	var marks *runtimeMarkBatch
464		-	if len(batches) != 0 {
465		-	marks = batches[0]
466		-	} else {
467		-	ownedMarks = d.newRuntimeMarkBatch()
468		-	marks = &ownedMarks
469		-	// Direct test callers retain the same guarantee as paint: observer I/O
470		-	// follows emitFrame's release of attachment ownership.
471		-	defer marks.flush()
472		-	}
	459	+	// frameFollowUp is the work a frame publication leaves for after the
	460	+	// attachment's sendMu is released: notices, repaints, and send-error cleanup
	461	+	// may re-enter rendering or take session locks, so they never run under it.
	462	+	type frameFollowUp struct {
	463	+	prepareErr error
	464	+	sendErr error
	465	+	sendTr ports.ServerConnection
	466	+	warnGraphics bool
	467	+	}
	468	+
	469	+	// publishFrameLocked is the sole side-effecting half of the pipeline: it
	470	+	// revalidates ownership, prepares, sends, and commits or aborts one captured
	471	+	// frame. The caller holds ac.sendMu for the complete capture/compose/publish
	472	+	// transaction; publishFrameLocked never releases it. It reports false when the
	473	+	// frame was rejected before preparation. The returned follow-up must be passed
	474	+	// to settleFrame after sendMu is released.
	475	+	func (d Daemon) publishFrameLocked(entry session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, marks *runtimeMarkBatch) (frameFollowUp, bool) {
473	476		if entry == nil \|\| entry.core() == nil \|\| ac == nil \|\| state == nil {
474		-	if ac != nil {
475		-	ac.sendMu.Unlock()
476		-	}
477		-	return false
	477	+	return frameFollowUp{}, false
478	478		}
479	479		entry.core().mu.Lock()
480	480		_, owned := entry.core().attachments[ac]

@@ -485,19 +485,16 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
485	485		entry.core().mu.Unlock()
486	486		if marks.attachmentEffect != nil {
487	487		if !marks.attachmentEffect.current() \|\| state.attachment != ac {
488		-	ac.sendMu.Unlock()
489		-	return false
	488	+	return frameFollowUp{}, false
490	489		}
491	490		} else {
492	491		if !owned \|\| state.attachment != ac \|\| ac.currentAttachmentSession() != entry {
493		-	ac.sendMu.Unlock()
494		-	return false
	492	+	return frameFollowUp{}, false
495	493		}
496	494		if state.lease != nil {
497	495		rc := attachmentRenderCoordinator(entry)
498	496		if rc == nil \|\| state.lease.attachment != ac \|\| !rc.leaseCurrent(state.lease, true) {
499		-	ac.sendMu.Unlock()
500		-	return false
	497	+	return frameFollowUp{}, false
501	498		}
502	499		}
503	500		}

@@ -509,28 +506,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
509	506		}
510	507		endDiff(0, err == nil)
511	508		if err != nil {
512		-	ac.sendMu.Unlock()
513		-	core := entry.core()
514		-	core.mu.Lock()
515		-	sessionName := core.name
516		-	core.mu.Unlock()
517		-	d.log.Error("render draw failed", "err", err, "session", sessionName)
518		-	// Without a coordinator reportError repaints synchronously. Suppress only
519		-	// that nested notice repaint; leave the guard before returning so a later,
520		-	// independent failed transaction can still notify the user.
521		-	if entry.renderCoordinator() == nil {
522		-	if !ac.prepareFailureFallback.CompareAndSwap(false, true) {
523		-	return true
524		-	}
525		-	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", err))
526		-	ac.prepareFailureFallback.Store(false)
527		-	return true
528		-	}
529		-	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", err))
530		-	// Notices are session-scoped, while the attachment repaint still needs
531		-	// to redraw its chrome after a failed transaction.
532		-	d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
533		-	return true
	509	+	return frameFollowUp{prepareErr: err}, true
534	510		}
535	511		var sendTr ports.ServerConnection
536	512		var sendErr error

@@ -577,8 +553,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
577	553		emitted := sendErr == nil && prepared.sent()
578	554		endEmit(uint64(len(data)), emitted)
579	555		if sendErr == nil && !emitted {
580		-	ac.sendMu.Unlock()
581		-	return true
	556	+	return frameFollowUp{}, true
582	557		}
583	558		if sendErr != nil {
584	559		prepared.abort()

@@ -588,8 +563,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
588	563		if len(data) == 0 {
589	564		prepared.commitNoSend()
590	565		if !prepared.sent() {
591		-	ac.sendMu.Unlock()
592		-	return true
	566	+	return frameFollowUp{}, true
593	567		}
594	568		}
595	569		ac.afterFrame.emitted(state.frameCapture)

@@ -657,25 +631,52 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
657	631		}
658	632		}
659	633		}
660		-	ac.sendMu.Unlock()
661		-	if suppressedGraphics && d.warnUnsupportedGraphics(ac) {
	634	+	return frameFollowUp{sendErr: sendErr, sendTr: sendTr, warnGraphics: suppressedGraphics}, true
	635	+	}
	636	+
	637	+	// settleFrame runs a publication's follow-up with sendMu released and no
	638	+	// attachment, session, tab, or pane lock held.
	639	+	func (d Daemon) settleFrame(entry session, ac attachedClient, marks runtimeMarkBatch, f frameFollowUp) {
	640	+	if f.prepareErr != nil {
	641	+	core := entry.core()
	642	+	core.mu.Lock()
	643	+	sessionName := core.name
	644	+	core.mu.Unlock()
	645	+	d.log.Error("render draw failed", "err", f.prepareErr, "session", sessionName)
	646	+	// Without a coordinator reportError repaints synchronously. Suppress only
	647	+	// that nested notice repaint; leave the guard before returning so a later,
	648	+	// independent failed transaction can still notify the user.
	649	+	if entry.renderCoordinator() == nil {
	650	+	if !ac.prepareFailureFallback.CompareAndSwap(false, true) {
	651	+	return
	652	+	}
	653	+	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", f.prepareErr))
	654	+	ac.prepareFailureFallback.Store(false)
	655	+	return
	656	+	}
	657	+	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", f.prepareErr))
	658	+	// Notices are session-scoped, while the attachment repaint still needs
	659	+	// to redraw its chrome after a failed transaction.
	660	+	d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
	661	+	return
	662	+	}
	663	+	if f.warnGraphics && d.warnUnsupportedGraphics(ac) {
662	664		d.repaintForNotice(ac)
663	665		}
664		-	if sendErr != nil {
	666	+	if f.sendErr != nil {
665	667		// A transport failure may invalidate the role gate. Release this render's
666	668		// admission first. Detachment freezes the gate and therefore cannot mutate
667	669		// ownership until any enclosing admitted operation has also ended.
668	670		if marks.attachmentEffect == nil {
669		-	d.detachOnSendError(entry, ac, sendTr)
	671	+	d.detachOnSendError(entry, ac, f.sendTr)
670	672		} else {
671	673		// Capture the exact admitted capability, including its coordinator
672	674		// lease, before End permits a new attachment publication. Reserve
673	675		// cleanup accounting before End so terminal Wait cannot race a later Add.
674	676		capability := marks.attachmentEffect.capability()
675		-	launchCleanup := d.reserveAttachmentSendErrorCleanup(capability, sendTr)
	677	+	launchCleanup := d.reserveAttachmentSendErrorCleanup(capability, f.sendTr)
676	678		marks.attachmentEffect.End()
677	679		launchCleanup()
678	680		}
679	681		}
680		-	return true
681	682		}


@@ -456,25 +456,25 @@ func commitDamageReceipts(receipts []damageReceipt) {
456		}	456		}
457		}	457		}
458			458
459	-	// emitFrame is the sole side-effecting half of the pipeline. The caller holds	459	+	// frameFollowUp is the work a frame publication leaves for after the
460	-	// sendMu for the complete capture/compose/emit transaction.	460	+	// attachment's sendMu is released: notices, repaints, and send-error cleanup
461	-	func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool {	461	+	// may re-enter rendering or take session locks, so they never run under it.
462	-	var ownedMarks runtimeMarkBatch	462	+	type frameFollowUp struct {
463	-	var marks *runtimeMarkBatch	463	+	prepareErr error
464	-	if len(batches) != 0 {	464	+	sendErr error
465	-	marks = batches[0]	465	+	sendTr ports.ServerConnection
466	-	} else {	466	+	warnGraphics bool
467	-	ownedMarks = d.newRuntimeMarkBatch()	467	+	}
468	-	marks = &ownedMarks	468	+
469	-	// Direct test callers retain the same guarantee as paint: observer I/O	469	+	// publishFrameLocked is the sole side-effecting half of the pipeline: it
470	-	// follows emitFrame's release of attachment ownership.	470	+	// revalidates ownership, prepares, sends, and commits or aborts one captured
471	-	defer marks.flush()	471	+	// frame. The caller holds ac.sendMu for the complete capture/compose/publish
472	-	}	472	+	// transaction; publishFrameLocked never releases it. It reports false when the
			473	+	// frame was rejected before preparation. The returned follow-up must be passed
			474	+	// to settleFrame after sendMu is released.
			475	+	func (d Daemon) publishFrameLocked(entry session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, marks *runtimeMarkBatch) (frameFollowUp, bool) {
473		if entry == nil \|\| entry.core() == nil \|\| ac == nil \|\| state == nil {	476		if entry == nil \|\| entry.core() == nil \|\| ac == nil \|\| state == nil {
474	-	if ac != nil {	477	+	return frameFollowUp{}, false
475	-	ac.sendMu.Unlock()
476	-	}
477	-	return false
478		}	478		}
479		entry.core().mu.Lock()	479		entry.core().mu.Lock()
480		_, owned := entry.core().attachments[ac]	480		_, owned := entry.core().attachments[ac]

@@ -485,19 +485,16 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
485		entry.core().mu.Unlock()	485		entry.core().mu.Unlock()
486		if marks.attachmentEffect != nil {	486		if marks.attachmentEffect != nil {
487		if !marks.attachmentEffect.current() \|\| state.attachment != ac {	487		if !marks.attachmentEffect.current() \|\| state.attachment != ac {
488	-	ac.sendMu.Unlock()	488	+	return frameFollowUp{}, false
489	-	return false
490		}	489		}
491		} else {	490		} else {
492		if !owned \|\| state.attachment != ac \|\| ac.currentAttachmentSession() != entry {	491		if !owned \|\| state.attachment != ac \|\| ac.currentAttachmentSession() != entry {
493	-	ac.sendMu.Unlock()	492	+	return frameFollowUp{}, false
494	-	return false
495		}	493		}
496		if state.lease != nil {	494		if state.lease != nil {
497		rc := attachmentRenderCoordinator(entry)	495		rc := attachmentRenderCoordinator(entry)
498		if rc == nil \|\| state.lease.attachment != ac \|\| !rc.leaseCurrent(state.lease, true) {	496		if rc == nil \|\| state.lease.attachment != ac \|\| !rc.leaseCurrent(state.lease, true) {
499	-	ac.sendMu.Unlock()	497	+	return frameFollowUp{}, false
500	-	return false
501		}	498		}
502		}	499		}
503		}	500		}

@@ -509,28 +506,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
509		}	506		}
510		endDiff(0, err == nil)	507		endDiff(0, err == nil)
511		if err != nil {	508		if err != nil {
512	-	ac.sendMu.Unlock()	509	+	return frameFollowUp{prepareErr: err}, true
513	-	core := entry.core()
514	-	core.mu.Lock()
515	-	sessionName := core.name
516	-	core.mu.Unlock()
517	-	d.log.Error("render draw failed", "err", err, "session", sessionName)
518	-	// Without a coordinator reportError repaints synchronously. Suppress only
519	-	// that nested notice repaint; leave the guard before returning so a later,
520	-	// independent failed transaction can still notify the user.
521	-	if entry.renderCoordinator() == nil {
522	-	if !ac.prepareFailureFallback.CompareAndSwap(false, true) {
523	-	return true
524	-	}
525	-	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", err))
526	-	ac.prepareFailureFallback.Store(false)
527	-	return true
528	-	}
529	-	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", err))
530	-	// Notices are session-scoped, while the attachment repaint still needs
531	-	// to redraw its chrome after a failed transaction.
532	-	d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
533	-	return true
534		}	510		}
535		var sendTr ports.ServerConnection	511		var sendTr ports.ServerConnection
536		var sendErr error	512		var sendErr error

@@ -577,8 +553,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
577		emitted := sendErr == nil && prepared.sent()	553		emitted := sendErr == nil && prepared.sent()
578		endEmit(uint64(len(data)), emitted)	554		endEmit(uint64(len(data)), emitted)
579		if sendErr == nil && !emitted {	555		if sendErr == nil && !emitted {
580	-	ac.sendMu.Unlock()	556	+	return frameFollowUp{}, true
581	-	return true
582		}	557		}
583		if sendErr != nil {	558		if sendErr != nil {
584		prepared.abort()	559		prepared.abort()

@@ -588,8 +563,7 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
588		if len(data) == 0 {	563		if len(data) == 0 {
589		prepared.commitNoSend()	564		prepared.commitNoSend()
590		if !prepared.sent() {	565		if !prepared.sent() {
591	-	ac.sendMu.Unlock()	566	+	return frameFollowUp{}, true
592	-	return true
593		}	567		}
594		}	568		}
595		ac.afterFrame.emitted(state.frameCapture)	569		ac.afterFrame.emitted(state.frameCapture)

@@ -657,25 +631,52 @@ func (d Daemon) emitFrame(entry session, ac attachedClient, state capturedRe
657		}	631		}
658		}	632		}
659		}	633		}
660	-	ac.sendMu.Unlock()	634	+	return frameFollowUp{sendErr: sendErr, sendTr: sendTr, warnGraphics: suppressedGraphics}, true
661	-	if suppressedGraphics && d.warnUnsupportedGraphics(ac) {	635	+	}
			636	+
			637	+	// settleFrame runs a publication's follow-up with sendMu released and no
			638	+	// attachment, session, tab, or pane lock held.
			639	+	func (d Daemon) settleFrame(entry session, ac attachedClient, marks runtimeMarkBatch, f frameFollowUp) {
			640	+	if f.prepareErr != nil {
			641	+	core := entry.core()
			642	+	core.mu.Lock()
			643	+	sessionName := core.name
			644	+	core.mu.Unlock()
			645	+	d.log.Error("render draw failed", "err", f.prepareErr, "session", sessionName)
			646	+	// Without a coordinator reportError repaints synchronously. Suppress only
			647	+	// that nested notice repaint; leave the guard before returning so a later,
			648	+	// independent failed transaction can still notify the user.
			649	+	if entry.renderCoordinator() == nil {
			650	+	if !ac.prepareFailureFallback.CompareAndSwap(false, true) {
			651	+	return
			652	+	}
			653	+	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", f.prepareErr))
			654	+	ac.prepareFailureFallback.Store(false)
			655	+	return
			656	+	}
			657	+	d.reportError(entry, domain.UserErr(domain.NoticeInternal, "display update failed", f.prepareErr))
			658	+	// Notices are session-scoped, while the attachment repaint still needs
			659	+	// to redraw its chrome after a failed transaction.
			660	+	d.invalidateRender(entry, ac, true, "render_pipeline.go:prepare-failed")
			661	+	return
			662	+	}
			663	+	if f.warnGraphics && d.warnUnsupportedGraphics(ac) {
662		d.repaintForNotice(ac)	664		d.repaintForNotice(ac)
663		}	665		}
664	-	if sendErr != nil {	666	+	if f.sendErr != nil {
665		// A transport failure may invalidate the role gate. Release this render's	667		// A transport failure may invalidate the role gate. Release this render's
666		// admission first. Detachment freezes the gate and therefore cannot mutate	668		// admission first. Detachment freezes the gate and therefore cannot mutate
667		// ownership until any enclosing admitted operation has also ended.	669		// ownership until any enclosing admitted operation has also ended.
668		if marks.attachmentEffect == nil {	670		if marks.attachmentEffect == nil {
669	-	d.detachOnSendError(entry, ac, sendTr)	671	+	d.detachOnSendError(entry, ac, f.sendTr)
670		} else {	672		} else {
671		// Capture the exact admitted capability, including its coordinator	673		// Capture the exact admitted capability, including its coordinator
672		// lease, before End permits a new attachment publication. Reserve	674		// lease, before End permits a new attachment publication. Reserve
673		// cleanup accounting before End so terminal Wait cannot race a later Add.	675		// cleanup accounting before End so terminal Wait cannot race a later Add.
674		capability := marks.attachmentEffect.capability()	676		capability := marks.attachmentEffect.capability()
675	-	launchCleanup := d.reserveAttachmentSendErrorCleanup(capability, sendTr)	677	+	launchCleanup := d.reserveAttachmentSendErrorCleanup(capability, f.sendTr)
676		marks.attachmentEffect.End()	678		marks.attachmentEffect.End()
677		launchCleanup()	679		launchCleanup()
678		}	680		}
679		}	681		}
680	-	return true
681		}	682		}

M internal/usecase/daemon/render_pipeline_cache_test.go +2 -2


@@ -53,7 +53,7 @@ func TestPipelineCachePublishesOnlyAfterEmission(t *testing.T) {
53	53		state.route.Target = protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}
54	54		state.view.revision = ac.viewSnapshot().revision
55	55		ac.sendMu.Lock()
56		-	require.True(t, d.emitFrame(sess, ac, &state, pending))
	56	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
57	57		require.Equal(t, before, cloneComposeCache(ac.pipelineCache), "failed emission must not publish any composed cache backing storage")
58	58
59	59		// A send error detaches the failed link. Re-own this test attachment with

@@ -71,7 +71,7 @@ func TestPipelineCachePublishesOnlyAfterEmission(t *testing.T) {
71	71		state.route.Target = protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}
72	72		state.view.revision = ac.viewSnapshot().revision
73	73		ac.sendMu.Lock()
74		-	require.True(t, d.emitFrame(sess, ac, &state, pending))
	74	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
75	75		require.Equal(t, pending.cache, ac.pipelineCache)
76	76		frame := <-sends
77	77		output := unmarshalTestOutput(t, frame.Payload)


@@ -53,7 +53,7 @@ func TestPipelineCachePublishesOnlyAfterEmission(t *testing.T) {
53		state.route.Target = protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}	53		state.route.Target = protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}
54		state.view.revision = ac.viewSnapshot().revision	54		state.view.revision = ac.viewSnapshot().revision
55		ac.sendMu.Lock()	55		ac.sendMu.Lock()
56	-	require.True(t, d.emitFrame(sess, ac, &state, pending))	56	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
57		require.Equal(t, before, cloneComposeCache(ac.pipelineCache), "failed emission must not publish any composed cache backing storage")	57		require.Equal(t, before, cloneComposeCache(ac.pipelineCache), "failed emission must not publish any composed cache backing storage")
58			58
59		// A send error detaches the failed link. Re-own this test attachment with	59		// A send error detaches the failed link. Re-own this test attachment with

@@ -71,7 +71,7 @@ func TestPipelineCachePublishesOnlyAfterEmission(t *testing.T) {
71		state.route.Target = protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}	71		state.route.Target = protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}
72		state.view.revision = ac.viewSnapshot().revision	72		state.view.revision = ac.viewSnapshot().revision
73		ac.sendMu.Lock()	73		ac.sendMu.Lock()
74	-	require.True(t, d.emitFrame(sess, ac, &state, pending))	74	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, pending))
75		require.Equal(t, pending.cache, ac.pipelineCache)	75		require.Equal(t, pending.cache, ac.pipelineCache)
76		frame := <-sends	76		frame := <-sends
77		output := unmarshalTestOutput(t, frame.Payload)	77		output := unmarshalTestOutput(t, frame.Payload)

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


@@ -221,7 +221,7 @@ func TestCursorCandidateDoesNotPublishDuringPreparation(t *testing.T) {
221	221		require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, candidate.next)
222	222		}
223	223
224		-	func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
	224	+	func TestPublishFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
225	225		d, sess, ac, sends := newManualSessionWithPTYs(t)
226	226		healthy := ac.transport()
227	227		beforeCursor := cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}

@@ -233,7 +233,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
233	233		composed.cursor = cursorOut{row: 3, col: 4, style: 2, hasStyle: true}
234	234
235	235		ac.sendMu.Lock()
236		-	require.True(t, d.emitFrame(sess, ac, &state, composed))
	236	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, composed))
237	237
238	238		require.Equal(t, beforeCursor, ac.output.lastCursor)
239	239		require.Zero(t, ac.output.next)

@@ -248,7 +248,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
248	248		ac.replaceTransport(healthy)
249	249		state.attachment = ac
250	250		ac.sendMu.Lock()
251		-	require.True(t, d.emitFrame(sess, ac, &state, composed))
	251	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, composed))
252	252		require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, ac.output.lastCursor)
253	253		require.Equal(t, uint64(1), ac.output.next)
254	254		out := unmarshalTestOutput(t, (<-sends).Payload)

@@ -256,7 +256,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
256	256		require.Equal(t, uint64(1), out.New)
257	257		}
258	258
259		-	func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T) {
	259	+	func TestPublishFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T) {
260	260		d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
261	261		state := cacheState("steady", 1)
262	262		state.attachment = ac

@@ -265,7 +265,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
265	265		state.view.revision = ac.viewSnapshot().revision
266	266		initial := composeFrame(state, ac.pipelineCache, ac.pipelineScratch)
267	267		ac.sendMu.Lock()
268		-	require.True(t, d.emitFrame(sess, ac, &state, initial))
	268	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, initial))
269	269		<-sends
270	270
271	271		p := newPane("collapsed", nil, domain.Size{Cols: 1, Rows: 1})

@@ -285,7 +285,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
285	285		}
286	286		noByte.cache.layoutFingerprint = "no-byte-committed"
287	287		ac.sendMu.Lock()
288		-	require.True(t, d.emitFrame(sess, ac, &state, noByte))
	288	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, noByte))
289	289
290	290		require.Equal(t, beforeNext, ac.output.next)
291	291		require.Equal(t, beforeCursor, ac.output.lastCursor)

@@ -302,7 +302,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
302	302		state.focusedPaneID = "pane-2"
303	303		state.panes[0].stableID = "pane-2"
304	304		ac.sendMu.Lock()
305		-	require.True(t, d.emitFrame(sess, ac, &state, noByte))
	305	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, noByte))
306	306		var frame wire.Envelope
307	307		select {
308	308		case frame = <-sends:

@@ -699,7 +699,7 @@ func TestNoticeStylesFromMapsWarnToDedicatedRoleDistinctFromInfo(t *testing.T) {
699	699		require.NotEqual(t, got.BoxInfo, got.BoxWarn)
700	700		}
701	701
702		-	func TestEmitFrameSkipsTransportSendWhenAttachmentEffectFenceRejects(t *testing.T) {
	702	+	func TestPublishFrameSkipsTransportSendWhenAttachmentEffectFenceRejects(t *testing.T) {
703	703		d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
704	704		token := sess.captureAttachmentCapability(ac, ac.transport())
705	705		effect, admitted := ac.beginAttachmentEffect(token)

@@ -712,7 +712,7 @@ func TestEmitFrameSkipsTransportSendWhenAttachmentEffectFenceRejects(t *testing.
712	712		state.attachment = ac
713	713		composed := composeFrame(state, composeCacheInput{})
714	714		ac.sendMu.Lock()
715		-	require.True(t, d.emitFrame(sess, ac, &state, composed, &runtimeMarkBatch{attachmentEffect: effect}))
	715	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, composed, &runtimeMarkBatch{attachmentEffect: effect}))
716	716		d.attachmentCleanupWg.Wait()
717	717
718	718		require.Zero(t, ac.output.next, "rejected transport effect must not commit output state")


@@ -221,7 +221,7 @@ func TestCursorCandidateDoesNotPublishDuringPreparation(t *testing.T) {
221		require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, candidate.next)	221		require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, candidate.next)
222		}	222		}
223			223
224	-	func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {	224	+	func TestPublishFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
225		d, sess, ac, sends := newManualSessionWithPTYs(t)	225		d, sess, ac, sends := newManualSessionWithPTYs(t)
226		healthy := ac.transport()	226		healthy := ac.transport()
227		beforeCursor := cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}	227		beforeCursor := cursorOut{valid: true, row: 1, col: 2, style: 1, hasStyle: true}

@@ -233,7 +233,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
233		composed.cursor = cursorOut{row: 3, col: 4, style: 2, hasStyle: true}	233		composed.cursor = cursorOut{row: 3, col: 4, style: 2, hasStyle: true}
234			234
235		ac.sendMu.Lock()	235		ac.sendMu.Lock()
236	-	require.True(t, d.emitFrame(sess, ac, &state, composed))	236	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, composed))
237			237
238		require.Equal(t, beforeCursor, ac.output.lastCursor)	238		require.Equal(t, beforeCursor, ac.output.lastCursor)
239		require.Zero(t, ac.output.next)	239		require.Zero(t, ac.output.next)

@@ -248,7 +248,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
248		ac.replaceTransport(healthy)	248		ac.replaceTransport(healthy)
249		state.attachment = ac	249		state.attachment = ac
250		ac.sendMu.Lock()	250		ac.sendMu.Lock()
251	-	require.True(t, d.emitFrame(sess, ac, &state, composed))	251	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, composed))
252		require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, ac.output.lastCursor)	252		require.Equal(t, cursorOut{valid: true, row: 3, col: 4, style: 2, hasStyle: true}, ac.output.lastCursor)
253		require.Equal(t, uint64(1), ac.output.next)	253		require.Equal(t, uint64(1), ac.output.next)
254		out := unmarshalTestOutput(t, (<-sends).Payload)	254		out := unmarshalTestOutput(t, (<-sends).Payload)

@@ -256,7 +256,7 @@ func TestEmitFrameFailedSendDoesNotPublishCursorOrOutputState(t *testing.T) {
256		require.Equal(t, uint64(1), out.New)	256		require.Equal(t, uint64(1), out.New)
257		}	257		}
258			258
259	-	func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T) {	259	+	func TestPublishFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T) {
260		d, sess, ac, sends := newManualSessionWithPTYs(t, nil)	260		d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
261		state := cacheState("steady", 1)	261		state := cacheState("steady", 1)
262		state.attachment = ac	262		state.attachment = ac

@@ -265,7 +265,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
265		state.view.revision = ac.viewSnapshot().revision	265		state.view.revision = ac.viewSnapshot().revision
266		initial := composeFrame(state, ac.pipelineCache, ac.pipelineScratch)	266		initial := composeFrame(state, ac.pipelineCache, ac.pipelineScratch)
267		ac.sendMu.Lock()	267		ac.sendMu.Lock()
268	-	require.True(t, d.emitFrame(sess, ac, &state, initial))	268	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, initial))
269		<-sends	269		<-sends
270			270
271		p := newPane("collapsed", nil, domain.Size{Cols: 1, Rows: 1})	271		p := newPane("collapsed", nil, domain.Size{Cols: 1, Rows: 1})

@@ -285,7 +285,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
285		}	285		}
286		noByte.cache.layoutFingerprint = "no-byte-committed"	286		noByte.cache.layoutFingerprint = "no-byte-committed"
287		ac.sendMu.Lock()	287		ac.sendMu.Lock()
288	-	require.True(t, d.emitFrame(sess, ac, &state, noByte))	288	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, noByte))
289			289
290		require.Equal(t, beforeNext, ac.output.next)	290		require.Equal(t, beforeNext, ac.output.next)
291		require.Equal(t, beforeCursor, ac.output.lastCursor)	291		require.Equal(t, beforeCursor, ac.output.lastCursor)

@@ -302,7 +302,7 @@ func TestEmitFrameNoByteSuccessCommitsTransactionWithoutStateFrame(t *testing.T)
302		state.focusedPaneID = "pane-2"	302		state.focusedPaneID = "pane-2"
303		state.panes[0].stableID = "pane-2"	303		state.panes[0].stableID = "pane-2"
304		ac.sendMu.Lock()	304		ac.sendMu.Lock()
305	-	require.True(t, d.emitFrame(sess, ac, &state, noByte))	305	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, noByte))
306		var frame wire.Envelope	306		var frame wire.Envelope
307		select {	307		select {
308		case frame = <-sends:	308		case frame = <-sends:

@@ -699,7 +699,7 @@ func TestNoticeStylesFromMapsWarnToDedicatedRoleDistinctFromInfo(t *testing.T) {
699		require.NotEqual(t, got.BoxInfo, got.BoxWarn)	699		require.NotEqual(t, got.BoxInfo, got.BoxWarn)
700		}	700		}
701			701
702	-	func TestEmitFrameSkipsTransportSendWhenAttachmentEffectFenceRejects(t *testing.T) {	702	+	func TestPublishFrameSkipsTransportSendWhenAttachmentEffectFenceRejects(t *testing.T) {
703		d, sess, ac, sends := newManualSessionWithPTYs(t, nil)	703		d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
704		token := sess.captureAttachmentCapability(ac, ac.transport())	704		token := sess.captureAttachmentCapability(ac, ac.transport())
705		effect, admitted := ac.beginAttachmentEffect(token)	705		effect, admitted := ac.beginAttachmentEffect(token)

@@ -712,7 +712,7 @@ func TestEmitFrameSkipsTransportSendWhenAttachmentEffectFenceRejects(t *testing.
712		state.attachment = ac	712		state.attachment = ac
713		composed := composeFrame(state, composeCacheInput{})	713		composed := composeFrame(state, composeCacheInput{})
714		ac.sendMu.Lock()	714		ac.sendMu.Lock()
715	-	require.True(t, d.emitFrame(sess, ac, &state, composed, &runtimeMarkBatch{attachmentEffect: effect}))	715	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, &state, composed, &runtimeMarkBatch{attachmentEffect: effect}))
716		d.attachmentCleanupWg.Wait()	716		d.attachmentCleanupWg.Wait()
717			717
718		require.Zero(t, ac.output.next, "rejected transport effect must not commit output state")	718		require.Zero(t, ac.output.next, "rejected transport effect must not commit output state")

A internal/usecase/daemon/render_publish_testhelper_test.go +26 -0

@@ -0,0 +1,26 @@
	1	+	package daemon
	2	+
	3	+	// publishCapturedFrameForTest publishes a frame captured by a test that already
	4	+	// holds ac.sendMu. Like paint, it releases sendMu before settling the
	5	+	// publication and flushing runtime marks. It reports false only when the frame
	6	+	// was rejected before preparation.
	7	+	func (d Daemon) publishCapturedFrameForTest(entry session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool {
	8	+	marks := &runtimeMarkBatch{}
	9	+	if len(batches) != 0 {
	10	+	marks = batches[0]
	11	+	} else {
	12	+	owned := d.newRuntimeMarkBatch()
	13	+	marks = &owned
	14	+	}
	15	+	followUp, published := d.publishFrameLocked(entry, ac, state, composed, marks)
	16	+	if ac != nil {
	17	+	ac.sendMu.Unlock()
	18	+	}
	19	+	if published {
	20	+	d.settleFrame(entry, ac, marks, followUp)
	21	+	}
	22	+	if len(batches) == 0 {
	23	+	marks.flush()
	24	+	}
	25	+	return published
	26	+	}

@@ -0,0 +1,26 @@
	1	+	package daemon
	2	+
	3	+	// publishCapturedFrameForTest publishes a frame captured by a test that already
	4	+	// holds ac.sendMu. Like paint, it releases sendMu before settling the
	5	+	// publication and flushing runtime marks. It reports false only when the frame
	6	+	// was rejected before preparation.
	7	+	func (d Daemon) publishCapturedFrameForTest(entry session, ac attachedClient, state capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool {
	8	+	marks := &runtimeMarkBatch{}
	9	+	if len(batches) != 0 {
	10	+	marks = batches[0]
	11	+	} else {
	12	+	owned := d.newRuntimeMarkBatch()
	13	+	marks = &owned
	14	+	}
	15	+	followUp, published := d.publishFrameLocked(entry, ac, state, composed, marks)
	16	+	if ac != nil {
	17	+	ac.sendMu.Unlock()
	18	+	}
	19	+	if published {
	20	+	d.settleFrame(entry, ac, marks, followUp)
	21	+	}
	22	+	if len(batches) == 0 {
	23	+	marks.flush()
	24	+	}
	25	+	return published
	26	+	}

M internal/usecase/daemon/ui_fence_render_test.go +1 -1


@@ -85,7 +85,7 @@ func TestUIFencePreRegistrationCaptureCannotConfirmAction(t *testing.T) {
85	85		emit := func(state *capturedRenderState) {
86	86		t.Helper()
87	87		composed := composeFrame(*state, ac.pipelineCache)
88		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	88	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
89	89		}
90	90		state := capture(true)
91	91		dispatched := make(chan bool, 1)


@@ -85,7 +85,7 @@ func TestUIFencePreRegistrationCaptureCannotConfirmAction(t *testing.T) {
85		emit := func(state *capturedRenderState) {	85		emit := func(state *capturedRenderState) {
86		t.Helper()	86		t.Helper()
87		composed := composeFrame(*state, ac.pipelineCache)	87		composed := composeFrame(*state, ac.pipelineCache)
88	-	require.True(t, d.emitFrame(sess, ac, state, composed))	88	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
89		}	89		}
90		state := capture(true)	90		state := capture(true)
91		dispatched := make(chan bool, 1)	91		dispatched := make(chan bool, 1)

M internal/usecase/daemon/ui_render_publication_test.go +2 -2


@@ -32,7 +32,7 @@ func TestRenderPublicationUsesCapturedIdentityAndRevision(t *testing.T) {
32	32		view.revision++
33	33		ac.publishView(view)
34	34		}
35		-	require.True(t, d.emitFrame(sess, ac, state, composed)) // releases sendMu
	35	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed)) // releases sendMu
36	36		if scenario == "view replaced after capture" {
37	37		require.Empty(t, drainAllFrames(sends))
38	38		require.Zero(t, ac.output.viewPublication)

@@ -59,7 +59,7 @@ func TestRenderPublishesContextWithoutTerminalBytes(t *testing.T) {
59	59		t.Fatal("capture rejected")
60	60		}
61	61		composed := composeFrame(*state, ac.pipelineCache)
62		-	require.True(t, d.emitFrame(sess, ac, state, composed))
	62	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
63	63		}
64	64		paint(true)
65	65		initial := unmarshalTestOutput(t, awaitFrame(t, sends, "Output").Payload)


@@ -32,7 +32,7 @@ func TestRenderPublicationUsesCapturedIdentityAndRevision(t *testing.T) {
32		view.revision++	32		view.revision++
33		ac.publishView(view)	33		ac.publishView(view)
34		}	34		}
35	-	require.True(t, d.emitFrame(sess, ac, state, composed)) // releases sendMu	35	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed)) // releases sendMu
36		if scenario == "view replaced after capture" {	36		if scenario == "view replaced after capture" {
37		require.Empty(t, drainAllFrames(sends))	37		require.Empty(t, drainAllFrames(sends))
38		require.Zero(t, ac.output.viewPublication)	38		require.Zero(t, ac.output.viewPublication)

@@ -59,7 +59,7 @@ func TestRenderPublishesContextWithoutTerminalBytes(t *testing.T) {
59		t.Fatal("capture rejected")	59		t.Fatal("capture rejected")
60		}	60		}
61		composed := composeFrame(*state, ac.pipelineCache)	61		composed := composeFrame(*state, ac.pipelineCache)
62	-	require.True(t, d.emitFrame(sess, ac, state, composed))	62	+	require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
63		}	63		}
64		paint(true)	64		paint(true)
65		initial := unmarshalTestOutput(t, awaitFrame(t, sends, "Output").Payload)	65		initial := unmarshalTestOutput(t, awaitFrame(t, sends, "Output").Payload)