fix(daemon): close phase 3 output races

commit: fdbb41bebeabb32e02c90f0878969705d525a71e
author: brice · 2mo ago · 2026-08-03 10:30 UTC
committer: brice · 2mo ago · 2026-08-04 03:19 UTC
parent: 8086da1
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

8 files · +314 -54

M internal/usecase/daemon/attachment_render_test.go +3 -0


@@ -103,6 +103,9 @@ func TestAttachmentPaintFanoutDoesNotWaitForBlockedPeer(t *testing.T) {
103	103		select {
104	104		case frame := <-secondTransport.sends:
105	105		require.Equal(t, ports.MsgOutput, frame.Type)
	106	+	out, err := ports.UnmarshalOutput(frame.Payload)
	107	+	require.NoError(t, err)
	108	+	require.True(t, out.Full, "fan-out peers must send a fresh first frame even after another attachment acknowledges shared damage")
106	109		case <-time.After(time.Second):
107	110		t.Fatal("second attachment waited for first attachment transport")
108	111		}


@@ -103,6 +103,9 @@ func TestAttachmentPaintFanoutDoesNotWaitForBlockedPeer(t *testing.T) {
103	select {	103		select {
104	case frame := <-secondTransport.sends:	104		case frame := <-secondTransport.sends:
105	require.Equal(t, ports.MsgOutput, frame.Type)	105		require.Equal(t, ports.MsgOutput, frame.Type)
		106	+	out, err := ports.UnmarshalOutput(frame.Payload)
		107	+	require.NoError(t, err)
		108	+	require.True(t, out.Full, "fan-out peers must send a fresh first frame even after another attachment acknowledges shared damage")
106	case <-time.After(time.Second):	109		case <-time.After(time.Second):
107	t.Fatal("second attachment waited for first attachment transport")	110		t.Fatal("second attachment waited for first attachment transport")
108	}	111		}

M internal/usecase/daemon/client.go +34 -10


@@ -344,6 +344,27 @@ func (ac *attachedClient) sendExpectedTransportForAttachment(expected transportS
344	344		return err
345	345		}
346	346
	347	+	// sendExpectedSideEffectForAttachment captures the output epoch, view revision,
	348	+	// and window while sendMu is held. A resize or handoff therefore cannot put a
	349	+	// side-effect frame from the old attachment state onto the new transport state.
	350	+	func (ac attachedClient) sendExpectedSideEffectForAttachment(expected transportSnapshot, data []byte, echoAck uint64, ticket attachmentEffectTicket) error {
	351	+	ac.sendMu.Lock()
	352	+	defer ac.sendMu.Unlock()
	353	+	if ticket == nil {
	354	+	return errAttachmentTransition
	355	+	}
	356	+	if err := ac.beginExpectedTransportSendLocked(expected, ticket); err != nil {
	357	+	return errAttachmentTransition
	358	+	}
	359	+	frame := ac.output.sideEffectLocked(data, echoAck)
	360	+	err := expected.transport.Send(frame)
	361	+	if err != nil {
	362	+	ticket.reportTransportFailure(expected)
	363	+	}
	364	+	ticket.endTransportSend()
	365	+	return err
	366	+	}
	367	+
347	368		// boundedSend sends f to ac with a deadline watchdog: if the send (including
348	369		// waiting on sendMu behind a wedged paint) does not complete within
349	370		// detachNotifyTimeout, the transport is force-closed, failing the in-flight

@@ -373,24 +394,26 @@ func (d Daemon) boundedSendOutputErr(ac attachedClient, b []byte) error {
373	394		}
374	395
375	396		func (d Daemon) boundedSendOutputErrTransport(ac attachedClient, b []byte) (ports.Transport, error) {
376		-	frame := ac.output.sideEffect(b, ac.echoAck.Load())
377	397		expected := ac.transportSnapshot()
378	398		if expected.transport == nil {
379	399		return nil, errors.New("client transport is nil")
380	400		}
381		-	ac.sendMu.Lock()
382		-	if !ac.transportSnapshotCurrent(expected) {
383		-	ac.sendMu.Unlock()
384		-	return expected.transport, errTransportReplaced
385		-	}
386	401		if owned, ok := expected.transport.(ports.OwnedSynchronousTransport); ok {
387		-	err := owned.SendSynchronous(frame)
388		-	ac.sendMu.Unlock()
	402	+	ac.sendMu.Lock()
	403	+	defer ac.sendMu.Unlock()
	404	+	if !ac.transportSnapshotCurrent(expected) {
	405	+	return expected.transport, errTransportReplaced
	406	+	}
	407	+	err := owned.SendSynchronous(ac.output.sideEffectLocked(b, ac.echoAck.Load()))
389	408		return expected.transport, err
390	409		}
391		-	ac.sendMu.Unlock()
392	410		return d.boundedSendWith(expected.transport, func() error {
393		-	return ac.sendExpectedTransport(expected, frame)
	411	+	ac.sendMu.Lock()
	412	+	defer ac.sendMu.Unlock()
	413	+	if !ac.transportSnapshotCurrent(expected) {
	414	+	return errTransportReplaced
	415	+	}
	416	+	return expected.transport.Send(ac.output.sideEffectLocked(b, ac.echoAck.Load()))
394	417		})
395	418		}
396	419

@@ -605,6 +628,7 @@ func (d *Daemon) ensureAttachmentRenderCoordinatorPrelocked(entry attachmentSess
605	628		d.paint(entry, attachment, fanoutReset, lease)
606	629		} else if len(attachments) > 1 {
607	630		go d.paint(entry, attachment, fanoutReset, lease)
	631	+
608	632		} else {
609	633		d.paint(entry, attachment, fanoutReset, lease)
610	634		}

 	return err
 }
+// sendExpectedSideEffectForAttachment captures the output epoch, view revision,
+// and window while sendMu is held. A resize or handoff therefore cannot put a
+// side-effect frame from the old attachment state onto the new transport state.
+func (ac *attachedClient) sendExpectedSideEffectForAttachment(expected transportSnapshot, data []byte, echoAck uint64, ticket *attachmentEffectTicket) error {
+	ac.sendMu.Lock()
+	defer ac.sendMu.Unlock()
+	if ticket == nil {
+		return errAttachmentTransition
+	}
+	if err := ac.beginExpectedTransportSendLocked(expected, ticket); err != nil {
+		return errAttachmentTransition
+	}
+	frame := ac.output.sideEffectLocked(data, echoAck)
+	err := expected.transport.Send(frame)
+	if err != nil {
+		ticket.reportTransportFailure(expected)
+	}
+	ticket.endTransportSend()
+	return err
+}
 // boundedSend sends f to ac with a deadline watchdog: if the send (including
 // waiting on sendMu behind a wedged paint) does not complete within
 // detachNotifyTimeout, the transport is force-closed, failing the in-flight
 }
 func (d *Daemon) boundedSendOutputErrTransport(ac *attachedClient, b []byte) (ports.Transport, error) {
-	frame := ac.output.sideEffect(b, ac.echoAck.Load())
 	expected := ac.transportSnapshot()
 	if expected.transport == nil {
 		return nil, errors.New("client transport is nil")
 	}
-	ac.sendMu.Lock()
-	if !ac.transportSnapshotCurrent(expected) {
-		ac.sendMu.Unlock()
-		return expected.transport, errTransportReplaced
-	}
 	if owned, ok := expected.transport.(ports.OwnedSynchronousTransport); ok {
-		err := owned.SendSynchronous(frame)
+		ac.sendMu.Lock()
-		ac.sendMu.Unlock()
+		defer ac.sendMu.Unlock()
+		if !ac.transportSnapshotCurrent(expected) {
+			return expected.transport, errTransportReplaced
+		}
+		err := owned.SendSynchronous(ac.output.sideEffectLocked(b, ac.echoAck.Load()))
 		return expected.transport, err
 	}
-	ac.sendMu.Unlock()
 	return d.boundedSendWith(expected.transport, func() error {
-		return ac.sendExpectedTransport(expected, frame)
+		ac.sendMu.Lock()
+		defer ac.sendMu.Unlock()
+		if !ac.transportSnapshotCurrent(expected) {
+			return errTransportReplaced
+		}
+		return expected.transport.Send(ac.output.sideEffectLocked(b, ac.echoAck.Load()))
 	})
 }
 					d.paint(entry, attachment, fanoutReset, lease)
 				} else if len(attachments) > 1 {
 					go d.paint(entry, attachment, fanoutReset, lease)
 				} else {
 					d.paint(entry, attachment, fanoutReset, lease)
 				}

M internal/usecase/daemon/output_state_stream.go +7 -0


@@ -185,7 +185,14 @@ func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision ui
185	185		})}
186	186		}
187	187
	188	+	// sideEffect builds an output frame without advancing the state stream. When
	189	+	// an attachment is present, callers must hold its sendMu; attachment-bound
	190	+	// sends use sideEffectLocked after validating their transport incarnation.
188	191		func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) ports.Frame {
	192	+	return s.sideEffectLocked(data, echoAck)
	193	+	}
	194	+
	195	+	func (s *outputStateStream) sideEffectLocked(data []byte, echoAck uint64) ports.Frame {
189	196		_, epoch, viewRevision := s.fence()
190	197		size := domain.Size{Cols: 1, Rows: 1}
191	198		if s != nil && s.attachment != nil && s.attachment.size.Valid() {


@@ -185,7 +185,14 @@ func marshalOutputState(data []byte, epoch, base, next, echoAck, viewRevision ui
185	})}	185		})}
186	}	186		}
187		187
		188	+	// sideEffect builds an output frame without advancing the state stream. When
		189	+	// an attachment is present, callers must hold its sendMu; attachment-bound
		190	+	// sends use sideEffectLocked after validating their transport incarnation.
188	func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) ports.Frame {	191		func (s *outputStateStream) sideEffect(data []byte, echoAck uint64) ports.Frame {
		192	+	return s.sideEffectLocked(data, echoAck)
		193	+	}
		194	+
		195	+	func (s *outputStateStream) sideEffectLocked(data []byte, echoAck uint64) ports.Frame {
189	_, epoch, viewRevision := s.fence()	196		_, epoch, viewRevision := s.fence()
190	size := domain.Size{Cols: 1, Rows: 1}	197		size := domain.Size{Cols: 1, Rows: 1}
191	if s != nil && s.attachment != nil && s.attachment.size.Valid() {	198		if s != nil && s.attachment != nil && s.attachment.size.Valid() {

M internal/usecase/daemon/proxy_link.go +32 -38


@@ -602,54 +602,48 @@ func (d Daemon) runProxyTransport(ctx context.Context, p proxySession, generat
602	602		}
603	603		}
604	604
605		-	// handleProxySideEffect fans out a stateless remote effect to every attachment
606		-	// currently owning this proxy. Each attachment gets an independent capability,
607		-	// effect ticket, output snapshot, and exact transport-incarnation fence.
	605	+	// handleProxySideEffect forwards a stateless remote effect to every live local
	606	+	// attachment. Connection admission and transport incarnation checks make a stale
	607	+	// handoff a harmless drop rather than a send to the client's next session.
608	608		func (d Daemon) handleProxySideEffect(p proxySession, generation uint64, out ports.Output) error {
609	609		if p == nil \|\| !p.currentLinkGeneration(generation) {
610	610		return nil
611	611		}
	612	+	attachments := snapshotAttachmentSession(p)
612	613		var sendErr error
613		-	for _, ac := range snapshotAttachmentSession(p) {
	614	+	for _, ac := range attachments {
	615	+	if ac == nil \|\| ac.output == nil {
	616	+	continue
	617	+	}
	618	+	expected := ac.transportSnapshot()
	619	+	if expected.transport == nil {
	620	+	continue
	621	+	}
	622	+	token := attachmentToken(p, ac, expected.transport)
	623	+	if token.sess != p {
	624	+	continue
	625	+	}
	626	+	ticket, admitted := ac.beginAttachmentEffect(token)
	627	+	if !admitted {
	628	+	continue
	629	+	}
614	630		if !p.currentLinkGeneration(generation) {
615		-	break
	631	+	ticket.End()
	632	+	return nil
	633	+	}
	634	+	err := ac.sendExpectedSideEffectForAttachment(expected, out.Data, ac.echoAck.Load(), ticket)
	635	+	ticket.End()
	636	+	if err == nil \|\| errors.Is(err, errAttachmentTransition) {
	637	+	continue
616	638		}
617		-	if err := d.sendProxySideEffect(p, generation, ac, out); err != nil && sendErr == nil {
	639	+	// Try every live attachment before retiring the remote link for a failed
	640	+	// client send; one blocked or dead peer must not starve the others.
	641	+	if sendErr == nil {
618	642		sendErr = err
619	643		}
620	644		}
621		-	return sendErr
622		-	}
623		-
624		-	func (d Daemon) sendProxySideEffect(p proxySession, generation uint64, ac *attachedClient, out ports.Output) error {
625		-	if p == nil \|\| ac == nil \|\| ac.output == nil \|\| !p.currentLinkGeneration(generation) {
626		-	return nil
627		-	}
628		-	expected := ac.transportSnapshot()
629		-	if expected.transport == nil {
630		-	return nil
631		-	}
632		-	token := attachmentToken(p, ac, expected.transport)
633		-	if token.sess != p \|\| !token.current() {
634		-	return nil
635		-	}
636		-	ticket, admitted := ac.beginAttachmentEffect(token)
637		-	if !admitted {
638		-	return nil
639		-	}
640		-	defer ticket.End()
641		-	if !p.currentLinkGeneration(generation) \|\| !token.current() {
642		-	return nil
643		-	}
644		-	frame := ac.output.sideEffect(out.Data, ac.echoAck.Load())
645		-	if !p.currentLinkGeneration(generation) \|\| !token.current() {
646		-	return nil
647		-	}
648		-	if err := ac.sendExpectedTransportForAttachment(expected, frame, ticket); err != nil {
649		-	if errors.Is(err, errAttachmentTransition) {
650		-	return nil
651		-	}
652		-	return fmt.Errorf("proxy session: side-effect send: %w: %w", errProxyLinkSend, err)
	645	+	if sendErr != nil {
	646	+	return fmt.Errorf("proxy session: side-effect send: %w: %w", errProxyLinkSend, sendErr)
653	647		}
654	648		return nil
655	649		}

 	}
 }
-// handleProxySideEffect fans out a stateless remote effect to every attachment
+// handleProxySideEffect forwards a stateless remote effect to every live local
-// currently owning this proxy. Each attachment gets an independent capability,
+// attachment. Connection admission and transport incarnation checks make a stale
-// effect ticket, output snapshot, and exact transport-incarnation fence.
+// handoff a harmless drop rather than a send to the client's next session.
 func (d *Daemon) handleProxySideEffect(p *proxySession, generation uint64, out ports.Output) error {
 	if p == nil || !p.currentLinkGeneration(generation) {
 		return nil
 	}
+	attachments := snapshotAttachmentSession(p)
 	var sendErr error
-	for _, ac := range snapshotAttachmentSession(p) {
+	for _, ac := range attachments {
+		if ac == nil || ac.output == nil {
+			continue
+		}
+		expected := ac.transportSnapshot()
+		if expected.transport == nil {
+			continue
+		}
+		token := attachmentToken(p, ac, expected.transport)
+		if token.sess != p {
+			continue
+		}
+		ticket, admitted := ac.beginAttachmentEffect(token)
+		if !admitted {
+			continue
+		}
 		if !p.currentLinkGeneration(generation) {
-			break
+			ticket.End()
+			return nil
+		}
+		err := ac.sendExpectedSideEffectForAttachment(expected, out.Data, ac.echoAck.Load(), ticket)
+		ticket.End()
+		if err == nil || errors.Is(err, errAttachmentTransition) {
+			continue
 		}
-		if err := d.sendProxySideEffect(p, generation, ac, out); err != nil && sendErr == nil {
+		// Try every live attachment before retiring the remote link for a failed
+		// client send; one blocked or dead peer must not starve the others.
+		if sendErr == nil {
 			sendErr = err
 		}
 	}
-	return sendErr
+	if sendErr != nil {
-}
+		return fmt.Errorf("proxy session: side-effect send: %w: %w", errProxyLinkSend, sendErr)
-func (d *Daemon) sendProxySideEffect(p *proxySession, generation uint64, ac *attachedClient, out ports.Output) error {
-	if p == nil || ac == nil || ac.output == nil || !p.currentLinkGeneration(generation) {
-		return nil
-	}
-	expected := ac.transportSnapshot()
-	if expected.transport == nil {
-		return nil
-	}
-	token := attachmentToken(p, ac, expected.transport)
-	if token.sess != p || !token.current() {
-		return nil
-	}
-	ticket, admitted := ac.beginAttachmentEffect(token)
-	if !admitted {
-		return nil
-	}
-	defer ticket.End()
-	if !p.currentLinkGeneration(generation) || !token.current() {
-		return nil
-	}
-	frame := ac.output.sideEffect(out.Data, ac.echoAck.Load())
-	if !p.currentLinkGeneration(generation) || !token.current() {
-		return nil
-	}
-	if err := ac.sendExpectedTransportForAttachment(expected, frame, ticket); err != nil {
-		if errors.Is(err, errAttachmentTransition) {
-			return nil
-		}
-		return fmt.Errorf("proxy session: side-effect send: %w: %w", errProxyLinkSend, err)
 	}
 	return nil
 }

M internal/usecase/daemon/proxy_output.go +33 -6


@@ -15,24 +15,51 @@ func (p *proxySession) applyOutputForGeneration(generation uint64, out ports.Out
15	15		if p.linkGeneration != generation {
16	16		return 0, false, false
17	17		}
	18	+	invalid := func() (uint64, bool, bool) {
	19	+	if p.outputResetRequested {
	20	+	return 0, false, false
	21	+	}
	22	+	p.outputResetRequested = true
	23	+	return 0, true, false
	24	+	}
	25	+	if out.Epoch == 0 {
	26	+	return invalid()
	27	+	}
18	28		if out.New == 0 {
	29	+	// New == 0 is an ordinary byte side effect, not a state transition.
	30	+	// Match the thin client: it is accepted before the first state-bearing
	31	+	// frame, and thereafter only within the current epoch.
	32	+	if out.Base != 0 \|\| out.Full \|\| (p.outputReady && out.Epoch != p.outputEpoch) {
	33	+	return invalid()
	34	+	}
19	35		return 0, false, len(out.Data) != 0
20	36		}
21		-	if out.Base == 0 {
	37	+	if !p.outputReady {
	38	+	if out.Base != 0 \|\| !out.Full {
	39	+	return invalid()
	40	+	}
22	41		p.outputEpoch = out.Epoch
23	42		p.outputState = out.New
24	43		p.outputReady = true
25	44		p.outputResetRequested = false
26	45		return out.New, false, len(out.Data) != 0
27	46		}
28		-	if !p.outputReady \|\| p.outputResetRequested \|\| out.Epoch != p.outputEpoch \|\| out.Base != p.outputState \|\| out.New != out.Base+1 {
29		-	if !p.outputResetRequested {
30		-	p.outputResetRequested = true
31		-	requestReset = true
	47	+	if out.Epoch < p.outputEpoch {
	48	+	return invalid()
	49	+	}
	50	+	if out.Epoch == p.outputEpoch {
	51	+	if p.outputResetRequested \|\| out.Full \|\| out.Base != p.outputState \|\| out.New != out.Base+1 {
	52	+	return invalid()
32	53		}
33		-	return 0, requestReset, false
	54	+	p.outputState = out.New
	55	+	return out.New, false, len(out.Data) != 0
34	56		}
	57	+	if out.Base != 0 \|\| !out.Full {
	58	+	return invalid()
	59	+	}
	60	+	p.outputEpoch = out.Epoch
35	61		p.outputState = out.New
	62	+	p.outputResetRequested = false
36	63		return out.New, false, len(out.Data) != 0
37	64		}
38	65


@@ -15,24 +15,51 @@ func (p *proxySession) applyOutputForGeneration(generation uint64, out ports.Out
15		if p.linkGeneration != generation {	15		if p.linkGeneration != generation {
16		return 0, false, false	16		return 0, false, false
17		}	17		}
			18	+	invalid := func() (uint64, bool, bool) {
			19	+	if p.outputResetRequested {
			20	+	return 0, false, false
			21	+	}
			22	+	p.outputResetRequested = true
			23	+	return 0, true, false
			24	+	}
			25	+	if out.Epoch == 0 {
			26	+	return invalid()
			27	+	}
18		if out.New == 0 {	28		if out.New == 0 {
			29	+	// New == 0 is an ordinary byte side effect, not a state transition.
			30	+	// Match the thin client: it is accepted before the first state-bearing
			31	+	// frame, and thereafter only within the current epoch.
			32	+	if out.Base != 0 \|\| out.Full \|\| (p.outputReady && out.Epoch != p.outputEpoch) {
			33	+	return invalid()
			34	+	}
19		return 0, false, len(out.Data) != 0	35		return 0, false, len(out.Data) != 0
20		}	36		}
21	-	if out.Base == 0 {	37	+	if !p.outputReady {
			38	+	if out.Base != 0 \|\| !out.Full {
			39	+	return invalid()
			40	+	}
22		p.outputEpoch = out.Epoch	41		p.outputEpoch = out.Epoch
23		p.outputState = out.New	42		p.outputState = out.New
24		p.outputReady = true	43		p.outputReady = true
25		p.outputResetRequested = false	44		p.outputResetRequested = false
26		return out.New, false, len(out.Data) != 0	45		return out.New, false, len(out.Data) != 0
27		}	46		}
28	-	if !p.outputReady \|\| p.outputResetRequested \|\| out.Epoch != p.outputEpoch \|\| out.Base != p.outputState \|\| out.New != out.Base+1 {	47	+	if out.Epoch < p.outputEpoch {
29	-	if !p.outputResetRequested {	48	+	return invalid()
30	-	p.outputResetRequested = true	49	+	}
31	-	requestReset = true	50	+	if out.Epoch == p.outputEpoch {
			51	+	if p.outputResetRequested \|\| out.Full \|\| out.Base != p.outputState \|\| out.New != out.Base+1 {
			52	+	return invalid()
32		}	53		}
33	-	return 0, requestReset, false	54	+	p.outputState = out.New
			55	+	return out.New, false, len(out.Data) != 0
34		}	56		}
			57	+	if out.Base != 0 \|\| !out.Full {
			58	+	return invalid()
			59	+	}
			60	+	p.outputEpoch = out.Epoch
35		p.outputState = out.New	61		p.outputState = out.New
			62	+	p.outputResetRequested = false
36		return out.New, false, len(out.Data) != 0	63		return out.New, false, len(out.Data) != 0
37		}	64		}
38			65

M internal/usecase/daemon/proxy_output_test.go +171 -0


@@ -594,3 +594,174 @@ func (d Daemon) handleLinkFrame(p proxySession, generation uint64, frame ports
594	594		return proxyLinkStop, fmt.Errorf("unexpected proxy server frame type %d", frame.Type)
595	595		}
596	596		}
	597	+
	598	+	func TestProxyOutputRejectsStaleStateAndRequestsOneReset(t *testing.T) {
	599	+	tests := []struct {
	600	+	name string
	601	+	state ports.Output
	602	+	stale ports.Output
	603	+	}{
	604	+	{
	605	+	name: "duplicate full",
	606	+	state: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true},
	607	+	stale: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true, Data: []byte("duplicate")},
	608	+	},
	609	+	{
	610	+	name: "older epoch",
	611	+	state: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true},
	612	+	stale: ports.Output{Epoch: 2, Base: 0, New: 5, Full: true, Data: []byte("older")},
	613	+	},
	614	+	{
	615	+	name: "base-zero non-full",
	616	+	state: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true},
	617	+	stale: ports.Output{Epoch: 3, Base: 0, New: 5, Data: []byte("not full")},
	618	+	},
	619	+	}
	620	+	for _, tt := range tests {
	621	+	t.Run(tt.name, func(t *testing.T) {
	622	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
	623	+	require.NoError(t, err)
	624	+	proxy.linkGeneration = 1
	625	+	ack, reset, changed := proxy.applyOutputForGeneration(1, tt.state)
	626	+	require.Equal(t, uint64(4), ack)
	627	+	require.False(t, reset)
	628	+	require.Equal(t, len(tt.state.Data) != 0, changed)
	629	+
	630	+	ack, reset, changed = proxy.applyOutputForGeneration(1, tt.stale)
	631	+	require.Zero(t, ack)
	632	+	require.True(t, reset)
	633	+	require.False(t, changed)
	634	+	ack, reset, changed = proxy.applyOutputForGeneration(1, tt.stale)
	635	+	require.Zero(t, ack)
	636	+	require.False(t, reset)
	637	+	require.False(t, changed)
	638	+	})
	639	+	}
	640	+	}
	641	+
	642	+	func TestProxyOutputFansOutToEveryLiveAttachment(t *testing.T) {
	643	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
	644	+	require.NoError(t, err)
	645	+	remote := newProxyTestTransport()
	646	+	proxy.transport = remote
	647	+	proxy.linkGeneration = 1
	648	+	d := &Daemon{}
	649	+
	650	+	attachments := make([]*proxyTestTransport, 0, 2)
	651	+	for id := byte(1); id <= 2; id++ {
	652	+	clientTransport := newProxyTestTransport()
	653	+	ac := &attachedClient{tr: clientTransport, output: newOutputStateStream(), size: proxy.contentSize}
	654	+	ac.output.attachment = ac
	655	+	ac.clientID[0] = id
	656	+	ac.setSession(proxy)
	657	+	proxy.sessionCore.mu.Lock()
	658	+	require.True(t, proxy.registerAttachmentLocked(ac))
	659	+	proxy.sessionCore.mu.Unlock()
	660	+	attachments = append(attachments, clientTransport)
	661	+	}
	662	+
	663	+	result, err := d.handleLinkFrame(proxy, 1, ports.Frame{
	664	+	Type: ports.MsgOutput,
	665	+	Payload: ports.MarshalOutput(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("full")}),
	666	+	})
	667	+	require.NoError(t, err)
	668	+	require.Equal(t, proxyLinkResume, result)
	669	+	for _, transport := range attachments {
	670	+	frame := <-transport.sent
	671	+	require.Equal(t, ports.MsgOutput, frame.Type)
	672	+	out, err := ports.UnmarshalOutput(frame.Payload)
	673	+	require.NoError(t, err)
	674	+	require.Zero(t, out.New, "local proxy forwarding must not create a second ACK path")
	675	+	require.Equal(t, []byte("full"), out.Data)
	676	+	}
	677	+	ackFrame := <-remote.sent
	678	+	require.Equal(t, ports.MsgAck, ackFrame.Type)
	679	+	}
	680	+
	681	+	func TestProxyOutputAttachAfterStreamReceivesCurrentFullOutput(t *testing.T) {
	682	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
	683	+	require.NoError(t, err)
	684	+	remote := newProxyTestTransport()
	685	+	proxy.transport = remote
	686	+	proxy.linkGeneration = 1
	687	+	d := &Daemon{}
	688	+
	689	+	firstTransport := newProxyTestTransport()
	690	+	first := &attachedClient{tr: firstTransport, output: newOutputStateStream(), size: proxy.contentSize}
	691	+	first.output.attachment = first
	692	+	first.clientID[0] = 1
	693	+	first.setSession(proxy)
	694	+	proxy.sessionCore.mu.Lock()
	695	+	require.True(t, proxy.registerAttachmentLocked(first))
	696	+	proxy.sessionCore.mu.Unlock()
	697	+
	698	+	handle := func(output ports.Output) {
	699	+	t.Helper()
	700	+	result, err := d.handleLinkFrame(proxy, 1, ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(output)})
	701	+	require.NoError(t, err)
	702	+	require.Equal(t, proxyLinkResume, result)
	703	+	}
	704	+	handle(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("before attach")})
	705	+	<-firstTransport.sent
	706	+	<-remote.sent
	707	+
	708	+	secondTransport := newProxyTestTransport()
	709	+	second := &attachedClient{tr: secondTransport, output: newOutputStateStream(), size: proxy.contentSize}
	710	+	second.output.attachment = second
	711	+	second.clientID[0] = 2
	712	+	second.setSession(proxy)
	713	+	proxy.sessionCore.mu.Lock()
	714	+	require.True(t, proxy.registerAttachmentLocked(second))
	715	+	proxy.sessionCore.mu.Unlock()
	716	+
	717	+	handle(ports.Output{Epoch: 4, Base: 0, New: 1, Full: true, Data: []byte("current full")})
	718	+	for _, transport := range []*proxyTestTransport{firstTransport, secondTransport} {
	719	+	frame := <-transport.sent
	720	+	require.Equal(t, ports.MsgOutput, frame.Type)
	721	+	out, err := ports.UnmarshalOutput(frame.Payload)
	722	+	require.NoError(t, err)
	723	+	require.Zero(t, out.New)
	724	+	require.False(t, out.Full, "local proxy side effects must remain stateless")
	725	+	require.Equal(t, []byte("current full"), out.Data)
	726	+	}
	727	+	ackFrame := <-remote.sent
	728	+	require.Equal(t, ports.MsgAck, ackFrame.Type)
	729	+	}
	730	+
	731	+	func TestProxyOutputInvalidStateNeverForwardsBytes(t *testing.T) {
	732	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
	733	+	require.NoError(t, err)
	734	+	remote := newProxyTestTransport()
	735	+	proxy.transport = remote
	736	+	proxy.linkGeneration = 1
	737	+	clientTransport := newProxyTestTransport()
	738	+	ac := &attachedClient{tr: clientTransport, output: newOutputStateStream(), size: proxy.contentSize}
	739	+	ac.output.attachment = ac
	740	+	ac.setSession(proxy)
	741	+	proxy.sessionCore.mu.Lock()
	742	+	require.True(t, proxy.registerAttachmentLocked(ac))
	743	+	proxy.sessionCore.mu.Unlock()
	744	+	d := &Daemon{}
	745	+
	746	+	_, err = d.handleLinkFrame(proxy, 1, ports.Frame{
	747	+	Type: ports.MsgOutput,
	748	+	Payload: ports.MarshalOutput(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("full")}),
	749	+	})
	750	+	require.NoError(t, err)
	751	+	<-clientTransport.sent
	752	+	<-remote.sent
	753	+
	754	+	result, err := d.handleLinkFrame(proxy, 1, ports.Frame{
	755	+	Type: ports.MsgOutput,
	756	+	Payload: ports.MarshalOutput(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("stale")}),
	757	+	})
	758	+	require.NoError(t, err)
	759	+	require.Equal(t, proxyLinkResume, result)
	760	+	select {
	761	+	case frame := <-clientTransport.sent:
	762	+	t.Fatalf("stale proxy output forwarded as %v", frame.Type)
	763	+	default:
	764	+	}
	765	+	reset := <-remote.sent
	766	+	require.Equal(t, ports.MsgOutputResetRequest, reset.Type)
	767	+	}


@@ -594,3 +594,174 @@ func (d Daemon) handleLinkFrame(p proxySession, generation uint64, frame ports
594	return proxyLinkStop, fmt.Errorf("unexpected proxy server frame type %d", frame.Type)	594		return proxyLinkStop, fmt.Errorf("unexpected proxy server frame type %d", frame.Type)
595	}	595		}
596	}	596		}
		597	+
		598	+	func TestProxyOutputRejectsStaleStateAndRequestsOneReset(t *testing.T) {
		599	+	tests := []struct {
		600	+	name string
		601	+	state ports.Output
		602	+	stale ports.Output
		603	+	}{
		604	+	{
		605	+	name: "duplicate full",
		606	+	state: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true},
		607	+	stale: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true, Data: []byte("duplicate")},
		608	+	},
		609	+	{
		610	+	name: "older epoch",
		611	+	state: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true},
		612	+	stale: ports.Output{Epoch: 2, Base: 0, New: 5, Full: true, Data: []byte("older")},
		613	+	},
		614	+	{
		615	+	name: "base-zero non-full",
		616	+	state: ports.Output{Epoch: 3, Base: 0, New: 4, Full: true},
		617	+	stale: ports.Output{Epoch: 3, Base: 0, New: 5, Data: []byte("not full")},
		618	+	},
		619	+	}
		620	+	for _, tt := range tests {
		621	+	t.Run(tt.name, func(t *testing.T) {
		622	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
		623	+	require.NoError(t, err)
		624	+	proxy.linkGeneration = 1
		625	+	ack, reset, changed := proxy.applyOutputForGeneration(1, tt.state)
		626	+	require.Equal(t, uint64(4), ack)
		627	+	require.False(t, reset)
		628	+	require.Equal(t, len(tt.state.Data) != 0, changed)
		629	+
		630	+	ack, reset, changed = proxy.applyOutputForGeneration(1, tt.stale)
		631	+	require.Zero(t, ack)
		632	+	require.True(t, reset)
		633	+	require.False(t, changed)
		634	+	ack, reset, changed = proxy.applyOutputForGeneration(1, tt.stale)
		635	+	require.Zero(t, ack)
		636	+	require.False(t, reset)
		637	+	require.False(t, changed)
		638	+	})
		639	+	}
		640	+	}
		641	+
		642	+	func TestProxyOutputFansOutToEveryLiveAttachment(t *testing.T) {
		643	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
		644	+	require.NoError(t, err)
		645	+	remote := newProxyTestTransport()
		646	+	proxy.transport = remote
		647	+	proxy.linkGeneration = 1
		648	+	d := &Daemon{}
		649	+
		650	+	attachments := make([]*proxyTestTransport, 0, 2)
		651	+	for id := byte(1); id <= 2; id++ {
		652	+	clientTransport := newProxyTestTransport()
		653	+	ac := &attachedClient{tr: clientTransport, output: newOutputStateStream(), size: proxy.contentSize}
		654	+	ac.output.attachment = ac
		655	+	ac.clientID[0] = id
		656	+	ac.setSession(proxy)
		657	+	proxy.sessionCore.mu.Lock()
		658	+	require.True(t, proxy.registerAttachmentLocked(ac))
		659	+	proxy.sessionCore.mu.Unlock()
		660	+	attachments = append(attachments, clientTransport)
		661	+	}
		662	+
		663	+	result, err := d.handleLinkFrame(proxy, 1, ports.Frame{
		664	+	Type: ports.MsgOutput,
		665	+	Payload: ports.MarshalOutput(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("full")}),
		666	+	})
		667	+	require.NoError(t, err)
		668	+	require.Equal(t, proxyLinkResume, result)
		669	+	for _, transport := range attachments {
		670	+	frame := <-transport.sent
		671	+	require.Equal(t, ports.MsgOutput, frame.Type)
		672	+	out, err := ports.UnmarshalOutput(frame.Payload)
		673	+	require.NoError(t, err)
		674	+	require.Zero(t, out.New, "local proxy forwarding must not create a second ACK path")
		675	+	require.Equal(t, []byte("full"), out.Data)
		676	+	}
		677	+	ackFrame := <-remote.sent
		678	+	require.Equal(t, ports.MsgAck, ackFrame.Type)
		679	+	}
		680	+
		681	+	func TestProxyOutputAttachAfterStreamReceivesCurrentFullOutput(t *testing.T) {
		682	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
		683	+	require.NoError(t, err)
		684	+	remote := newProxyTestTransport()
		685	+	proxy.transport = remote
		686	+	proxy.linkGeneration = 1
		687	+	d := &Daemon{}
		688	+
		689	+	firstTransport := newProxyTestTransport()
		690	+	first := &attachedClient{tr: firstTransport, output: newOutputStateStream(), size: proxy.contentSize}
		691	+	first.output.attachment = first
		692	+	first.clientID[0] = 1
		693	+	first.setSession(proxy)
		694	+	proxy.sessionCore.mu.Lock()
		695	+	require.True(t, proxy.registerAttachmentLocked(first))
		696	+	proxy.sessionCore.mu.Unlock()
		697	+
		698	+	handle := func(output ports.Output) {
		699	+	t.Helper()
		700	+	result, err := d.handleLinkFrame(proxy, 1, ports.Frame{Type: ports.MsgOutput, Payload: ports.MarshalOutput(output)})
		701	+	require.NoError(t, err)
		702	+	require.Equal(t, proxyLinkResume, result)
		703	+	}
		704	+	handle(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("before attach")})
		705	+	<-firstTransport.sent
		706	+	<-remote.sent
		707	+
		708	+	secondTransport := newProxyTestTransport()
		709	+	second := &attachedClient{tr: secondTransport, output: newOutputStateStream(), size: proxy.contentSize}
		710	+	second.output.attachment = second
		711	+	second.clientID[0] = 2
		712	+	second.setSession(proxy)
		713	+	proxy.sessionCore.mu.Lock()
		714	+	require.True(t, proxy.registerAttachmentLocked(second))
		715	+	proxy.sessionCore.mu.Unlock()
		716	+
		717	+	handle(ports.Output{Epoch: 4, Base: 0, New: 1, Full: true, Data: []byte("current full")})
		718	+	for _, transport := range []*proxyTestTransport{firstTransport, secondTransport} {
		719	+	frame := <-transport.sent
		720	+	require.Equal(t, ports.MsgOutput, frame.Type)
		721	+	out, err := ports.UnmarshalOutput(frame.Payload)
		722	+	require.NoError(t, err)
		723	+	require.Zero(t, out.New)
		724	+	require.False(t, out.Full, "local proxy side effects must remain stateless")
		725	+	require.Equal(t, []byte("current full"), out.Data)
		726	+	}
		727	+	ackFrame := <-remote.sent
		728	+	require.Equal(t, ports.MsgAck, ackFrame.Type)
		729	+	}
		730	+
		731	+	func TestProxyOutputInvalidStateNeverForwardsBytes(t *testing.T) {
		732	+	proxy, err := newProxySession(domain.RemoteSessionKey{Host: "host", Name: "session"}, domain.Size{Cols: 80, Rows: 24})
		733	+	require.NoError(t, err)
		734	+	remote := newProxyTestTransport()
		735	+	proxy.transport = remote
		736	+	proxy.linkGeneration = 1
		737	+	clientTransport := newProxyTestTransport()
		738	+	ac := &attachedClient{tr: clientTransport, output: newOutputStateStream(), size: proxy.contentSize}
		739	+	ac.output.attachment = ac
		740	+	ac.setSession(proxy)
		741	+	proxy.sessionCore.mu.Lock()
		742	+	require.True(t, proxy.registerAttachmentLocked(ac))
		743	+	proxy.sessionCore.mu.Unlock()
		744	+	d := &Daemon{}
		745	+
		746	+	_, err = d.handleLinkFrame(proxy, 1, ports.Frame{
		747	+	Type: ports.MsgOutput,
		748	+	Payload: ports.MarshalOutput(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("full")}),
		749	+	})
		750	+	require.NoError(t, err)
		751	+	<-clientTransport.sent
		752	+	<-remote.sent
		753	+
		754	+	result, err := d.handleLinkFrame(proxy, 1, ports.Frame{
		755	+	Type: ports.MsgOutput,
		756	+	Payload: ports.MarshalOutput(ports.Output{Epoch: 3, Base: 0, New: 1, Full: true, Data: []byte("stale")}),
		757	+	})
		758	+	require.NoError(t, err)
		759	+	require.Equal(t, proxyLinkResume, result)
		760	+	select {
		761	+	case frame := <-clientTransport.sent:
		762	+	t.Fatalf("stale proxy output forwarded as %v", frame.Type)
		763	+	default:
		764	+	}
		765	+	reset := <-remote.sent
		766	+	require.Equal(t, ports.MsgOutputResetRequest, reset.Type)
		767	+	}

M internal/usecase/daemon/render_capture.go +7 -0


@@ -224,6 +224,13 @@ func captureLocalRenderState(
224	224		return nil, false
225	225		}
226	226
	227	+	// A reset is an attachment-local capture boundary. The shared pane damage
	228	+	// receipt may have been acknowledged by another fan-out attachment before
	229	+	// this capture runs, so retaining this attachment's prior pane snapshots
	230	+	// would compose an old frame with a fresh reset epoch.
	231	+	if reset {
	232	+	ac.captureFrames = nil
	233	+	}
227	234		scratch := &ac.renderScratch
228	235		scratch.statusTabs = append(scratch.statusTabs[:0], bars.status.tabs...)
229	236		bars.status.tabs = scratch.statusTabs


@@ -224,6 +224,13 @@ func captureLocalRenderState(
224	return nil, false	224		return nil, false
225	}	225		}
226		226
		227	+	// A reset is an attachment-local capture boundary. The shared pane damage
		228	+	// receipt may have been acknowledged by another fan-out attachment before
		229	+	// this capture runs, so retaining this attachment's prior pane snapshots
		230	+	// would compose an old frame with a fresh reset epoch.
		231	+	if reset {
		232	+	ac.captureFrames = nil
		233	+	}
227	scratch := &ac.renderScratch	234		scratch := &ac.renderScratch
228	scratch.statusTabs = append(scratch.statusTabs[:0], bars.status.tabs...)	235		scratch.statusTabs = append(scratch.statusTabs[:0], bars.status.tabs...)
229	bars.status.tabs = scratch.statusTabs	236		bars.status.tabs = scratch.statusTabs

M internal/usecase/daemon/render_damage_receipt_test.go +27 -0


@@ -11,6 +11,33 @@ import (
11	11		"github.com/stretchr/testify/require"
12	12		)
13	13
	14	+	func TestResetCaptureRefreshesPaneAfterPeerAcknowledgesSharedDamage(t *testing.T) {
	15	+	_, sess, ac, _ := newManualSessionWithPTYs(t, nil)
	16	+	p := sess.tabs[0].focusedPane()
	17	+	p.mu.Lock()
	18	+	p.screen.ClearDamage()
	19	+	p.screen.Write([]byte("before"))
	20	+	p.mu.Unlock()
	21	+
	22	+	ac.sendMu.Lock()
	23	+	state, ok := captureRenderState(sess, ac, renderCaptureRequest{reset: false})
	24	+	require.True(t, ok)
	25	+	before := state.panes[0].frame.Clone()
	26	+	ac.sendMu.Unlock()
	27	+
	28	+	p.mu.Lock()
	29	+	p.screen.Write([]byte("after"))
	30	+	damage := p.screen.CaptureDamage()
	31	+	p.screen.AcknowledgeDamage(damage.Generation)
	32	+	p.mu.Unlock()
	33	+
	34	+	ac.sendMu.Lock()
	35	+	state, ok = captureRenderState(sess, ac, renderCaptureRequest{reset: true})
	36	+	ac.sendMu.Unlock()
	37	+	require.True(t, ok)
	38	+	require.NotEqual(t, before, state.panes[0].frame, "reset capture must not reuse the pane snapshot retained before shared damage was acknowledged")
	39	+	}
	40	+
14	41		func TestPrimaryCaptureAloneRecordsDamageReceipts(t *testing.T) {
15	42		_, sess, ac, _ := newManualSessionWithPTYs(t, nil)
16	43		p := sess.tabs[0].focusedPane()


@@ -11,6 +11,33 @@ import (
11	"github.com/stretchr/testify/require"	11		"github.com/stretchr/testify/require"
12	)	12		)
13		13
		14	+	func TestResetCaptureRefreshesPaneAfterPeerAcknowledgesSharedDamage(t *testing.T) {
		15	+	_, sess, ac, _ := newManualSessionWithPTYs(t, nil)
		16	+	p := sess.tabs[0].focusedPane()
		17	+	p.mu.Lock()
		18	+	p.screen.ClearDamage()
		19	+	p.screen.Write([]byte("before"))
		20	+	p.mu.Unlock()
		21	+
		22	+	ac.sendMu.Lock()
		23	+	state, ok := captureRenderState(sess, ac, renderCaptureRequest{reset: false})
		24	+	require.True(t, ok)
		25	+	before := state.panes[0].frame.Clone()
		26	+	ac.sendMu.Unlock()
		27	+
		28	+	p.mu.Lock()
		29	+	p.screen.Write([]byte("after"))
		30	+	damage := p.screen.CaptureDamage()
		31	+	p.screen.AcknowledgeDamage(damage.Generation)
		32	+	p.mu.Unlock()
		33	+
		34	+	ac.sendMu.Lock()
		35	+	state, ok = captureRenderState(sess, ac, renderCaptureRequest{reset: true})
		36	+	ac.sendMu.Unlock()
		37	+	require.True(t, ok)
		38	+	require.NotEqual(t, before, state.panes[0].frame, "reset capture must not reuse the pane snapshot retained before shared damage was acknowledged")
		39	+	}
		40	+
14	func TestPrimaryCaptureAloneRecordsDamageReceipts(t *testing.T) {	41		func TestPrimaryCaptureAloneRecordsDamageReceipts(t *testing.T) {
15	_, sess, ac, _ := newManualSessionWithPTYs(t, nil)	42		_, sess, ac, _ := newManualSessionWithPTYs(t, nil)
16	p := sess.tabs[0].focusedPane()	43		p := sess.tabs[0].focusedPane()