fix(dgram): distinguish UDP contact from stream progress

commit: 246a824448cf8f5e0b7026d759604eb35e8aa707
author: brice · 3mo ago · 2026-07-10 06:20 UTC
parent: 993f038
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

6 files · +336 -127

A internal/adapters/dgram/health.go +50 -0

@@ -0,0 +1,50 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"time"
	5	+
	6	+	"github.com/bnema/vev/internal/ports"
	7	+	)
	8	+
	9	+	// healthTracker separates authenticated path contact from useful stream
	10	+	// progress. Transport.mu protects every field.
	11	+	type healthTracker struct {
	12	+	lastPacket time.Time
	13	+	lastRecord time.Time
	14	+	lastProgress time.Time
	15	+	}
	16	+
	17	+	func newHealthTracker(now time.Time) healthTracker {
	18	+	return healthTracker{lastPacket: now, lastRecord: now, lastProgress: now}
	19	+	}
	20	+
	21	+	func (h *healthTracker) authenticatedPacket(now time.Time) { h.lastPacket = now }
	22	+
	23	+	func (h *healthTracker) completeRecord(now time.Time) { h.lastRecord = now }
	24	+
	25	+	func (h *healthTracker) ackProgress(now time.Time) { h.lastProgress = now }
	26	+
	27	+	func (h healthTracker) decide(now time.Time, hasPending bool, degradedAfter, probeAfter, offlineAfter, deadAfter time.Duration) (ports.LinkState, bool, bool) {
	28	+	packetAge := now.Sub(h.lastPacket)
	29	+	switch {
	30	+	case deadAfter > 0 && packetAge >= deadAfter:
	31	+	return ports.LinkStateDead, false, true
	32	+	case offlineAfter > 0 && packetAge >= offlineAfter:
	33	+	return ports.LinkStateOffline, false, false
	34	+	case probeAfter > 0 && packetAge >= probeAfter:
	35	+	return ports.LinkStateProbing, true, false
	36	+	}
	37	+
	38	+	qualityAt := h.lastRecord
	39	+	if h.lastProgress.After(qualityAt) {
	40	+	qualityAt = h.lastProgress
	41	+	}
	42	+	qualityAge := now.Sub(qualityAt)
	43	+	if hasPending && probeAfter > 0 && qualityAge >= probeAfter {
	44	+	return ports.LinkStateProbing, true, false
	45	+	}
	46	+	if degradedAfter > 0 && qualityAge >= degradedAfter {
	47	+	return ports.LinkStateDegraded, false, false
	48	+	}
	49	+	return ports.LinkStateConnected, false, false
	50	+	}

@@ -0,0 +1,50 @@
	1	+	package dgram
	2	+
	3	+	import (
	4	+	"time"
	5	+
	6	+	"github.com/bnema/vev/internal/ports"
	7	+	)
	8	+
	9	+	// healthTracker separates authenticated path contact from useful stream
	10	+	// progress. Transport.mu protects every field.
	11	+	type healthTracker struct {
	12	+	lastPacket time.Time
	13	+	lastRecord time.Time
	14	+	lastProgress time.Time
	15	+	}
	16	+
	17	+	func newHealthTracker(now time.Time) healthTracker {
	18	+	return healthTracker{lastPacket: now, lastRecord: now, lastProgress: now}
	19	+	}
	20	+
	21	+	func (h *healthTracker) authenticatedPacket(now time.Time) { h.lastPacket = now }
	22	+
	23	+	func (h *healthTracker) completeRecord(now time.Time) { h.lastRecord = now }
	24	+
	25	+	func (h *healthTracker) ackProgress(now time.Time) { h.lastProgress = now }
	26	+
	27	+	func (h healthTracker) decide(now time.Time, hasPending bool, degradedAfter, probeAfter, offlineAfter, deadAfter time.Duration) (ports.LinkState, bool, bool) {
	28	+	packetAge := now.Sub(h.lastPacket)
	29	+	switch {
	30	+	case deadAfter > 0 && packetAge >= deadAfter:
	31	+	return ports.LinkStateDead, false, true
	32	+	case offlineAfter > 0 && packetAge >= offlineAfter:
	33	+	return ports.LinkStateOffline, false, false
	34	+	case probeAfter > 0 && packetAge >= probeAfter:
	35	+	return ports.LinkStateProbing, true, false
	36	+	}
	37	+
	38	+	qualityAt := h.lastRecord
	39	+	if h.lastProgress.After(qualityAt) {
	40	+	qualityAt = h.lastProgress
	41	+	}
	42	+	qualityAge := now.Sub(qualityAt)
	43	+	if hasPending && probeAfter > 0 && qualityAge >= probeAfter {
	44	+	return ports.LinkStateProbing, true, false
	45	+	}
	46	+	if degradedAfter > 0 && qualityAge >= degradedAfter {
	47	+	return ports.LinkStateDegraded, false, false
	48	+	}
	49	+	return ports.LinkStateConnected, false, false
	50	+	}

M internal/adapters/dgram/transport.go +37 -74


@@ -140,15 +140,19 @@ type Transport struct {
140	140		sendDir, recvDir uint32
141	141		mtu int
142	142
143		-	mu sync.Mutex
144		-	peer net.Addr
145		-	ctr uint64
146		-	seq uint64
147		-	probeSeq uint64
148		-	pending map[uint64]*pending
149		-	closed bool
150		-	closeErr error
151		-	lastHeard time.Time
	143	+	mu sync.Mutex
	144	+	peer net.Addr
	145	+	ctr uint64
	146	+	seq uint64
	147	+	probeSeq uint64
	148	+	pending map[uint64]*pending
	149	+	closed bool
	150	+	closeErr error
	151	+	health healthTracker
	152	+	peerCounter uint64
	153	+	peerCounterSet bool
	154	+	// These mirrors preserve the established diagnostic snapshot fields while
	155	+	// health owns link-state decisions.
152	156		lastAuthenticatedPacket time.Time
153	157		lastCompleteRecord time.Time
154	158		lastACKProgress time.Time

@@ -306,7 +310,7 @@ func NewTransportWithOptions(pc net.PacketConn, peer net.Addr, key []byte, sendD
306	310		reasm: pdgram.NewReassembler(),
307	311		in: make(chan ports.Frame, 32),
308	312		done: make(chan struct{}),
309		-	lastHeard: opts.Clock.Now(),
	313	+	health: newHealthTracker(opts.Clock.Now()),
310	314		lastAuthenticatedPacket: opts.Clock.Now(),
311	315		lastCompleteRecord: opts.Clock.Now(),
312	316		lastACKProgress: opts.Clock.Now(),

@@ -782,7 +786,7 @@ func (t *Transport) readLoop(pc net.PacketConn) {
782	786		return
783	787		}
784	788		t.recvMu.Lock()
785		-	_, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
	789	+	counter, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
786	790		if err != nil {
787	791		t.recvMu.Unlock()
788	792		continue

@@ -801,7 +805,14 @@ func (t *Transport) readLoop(pc net.PacketConn) {
801	805		t.reassemblyInflight = inflight
802	806		var afterAuthenticated func()
803	807		if err == nil {
804		-	t.lastAuthenticatedPacket = t.clock.Now()
	808	+	now := t.clock.Now()
	809	+	t.health.authenticatedPacket(now)
	810	+	t.lastAuthenticatedPacket = now
	811	+	if !t.peerCounterSet \|\| counter > t.peerCounter {
	812	+	t.peer = addr
	813	+	t.peerCounter = counter
	814	+	t.peerCounterSet = true
	815	+	}
805	816		afterAuthenticated = t.afterAuthenticatedPacket
806	817		}
807	818		t.mu.Unlock()

@@ -817,7 +828,6 @@ func (t *Transport) readLoop(pc net.PacketConn) {
817	828		continue
818	829		}
819	830		t.recordCompleteRecord()
820		-	t.updatePeerFromAuthenticated(addr)
821	831		t.handleRecord(payload)
822	832		}
823	833		}

@@ -833,15 +843,9 @@ func (t *Transport) notifyMalformedFragment() {
833	843
834	844		func (t *Transport) recordCompleteRecord() {
835	845		t.mu.Lock()
836		-	t.lastCompleteRecord = t.clock.Now()
837		-	t.mu.Unlock()
838		-	}
839		-
840		-	func (t *Transport) updatePeerFromAuthenticated(addr net.Addr) {
841		-	t.mu.Lock()
842		-	t.peer = addr
843		-	t.lastHeard = t.clock.Now()
844		-	t.hoppedOffline = false
	846	+	now := t.clock.Now()
	847	+	t.health.completeRecord(now)
	848	+	t.lastCompleteRecord = now
845	849		t.mu.Unlock()
846	850		t.setLinkState(ports.LinkStateConnected, nil)
847	851		}

@@ -869,11 +873,14 @@ func (t *Transport) handleRecord(p []byte) {
869	873		acked = true
870	874		}
871	875		if acked {
872		-	t.lastACKProgress = t.clock.Now()
	876	+	now := t.clock.Now()
	877	+	t.health.ackProgress(now)
	878	+	t.lastACKProgress = now
873	879		t.notifySendWaitersLocked()
874	880		}
875	881		t.mu.Unlock()
876	882		if acked {
	883	+	t.setLinkState(ports.LinkStateConnected, nil)
877	884		t.emitDiagnostic()
878	885		}
879	886		case recProbe:

@@ -1009,7 +1016,6 @@ func (t *Transport) healthLoop() {
1009	1016		select {
1010	1017		case <-health.C():
1011	1018		t.checkSilence()
1012		-	t.checkSendStall()
1013	1019		health.Reset(t.resendAfter)
1014	1020		case <-t.done:
1015	1021		return

@@ -1205,66 +1211,22 @@ func (t *Transport) hopPacketConnOnce() {
1205	1211
1206	1212		func (t *Transport) checkSilence() {
1207	1213		t.mu.Lock()
1208		-	last := t.lastHeard
1209	1214		closed := t.closed
1210		-	degradedAfter := t.degradedAfter
1211		-	probeAfter := t.probeAfter
1212		-	offlineAfter := t.offlineAfter
1213		-	deadAfter := t.deadAfter
1214	1215		now := t.clock.Now()
	1216	+	state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
1215	1217		t.mu.Unlock()
1216	1218		if closed {
1217	1219		return
1218	1220		}
1219		-	silentFor := now.Sub(last)
1220		-	switch {
1221		-	case deadAfter > 0 && silentFor >= deadAfter:
1222		-	t.setLinkState(ports.LinkStateDead, ErrLinkDead)
	1221	+	if dead {
	1222	+	t.setLinkState(state, ErrLinkDead)
1223	1223		t.closeWithError(ErrLinkDead)
1224		-	case offlineAfter > 0 && silentFor >= offlineAfter:
1225		-	t.setLinkState(ports.LinkStateOffline, nil)
1226		-	case probeAfter > 0 && silentFor >= probeAfter:
1227		-	t.setLinkState(ports.LinkStateProbing, nil)
1228		-	t.hopPacketConnOnce()
1229		-	case degradedAfter > 0 && silentFor >= degradedAfter:
1230		-	t.setLinkState(ports.LinkStateDegraded, nil)
1231		-	}
1232		-	}
1233		-
1234		-	// checkSendStall declares the link dead when the oldest unacked reliable frame
1235		-	// has gone unacknowledged for deadAfter despite at least one retransmit. The
1236		-	// receive-silence check in checkSilence misses one-way-dead links whose receive
1237		-	// side stays fresh from peer heartbeats; this catches them at the layer that
1238		-	// owns death. enqueued is set once when a frame is accepted and never refreshed,
1239		-	// so acks (which delete pending entries) keep the oldest age bounded on a
1240		-	// progressing link — only a frame that is never acked crosses the threshold.
1241		-	func (t *Transport) checkSendStall() {
1242		-	t.mu.Lock()
1243		-	if t.closed \|\| t.deadAfter <= 0 {
1244		-	t.mu.Unlock()
1245	1224		return
1246	1225		}
1247		-	deadAfter := t.deadAfter
1248		-	now := t.clock.Now()
1249		-	var oldest time.Time
1250		-	var attempts int
1251		-	found := false
1252		-	for _, p := range t.pending {
1253		-	if p == nil \|\| p.enqueued.IsZero() {
1254		-	continue
1255		-	}
1256		-	if !found \|\| p.enqueued.Before(oldest) {
1257		-	oldest = p.enqueued
1258		-	attempts = p.attempts
1259		-	found = true
1260		-	}
1261		-	}
1262		-	t.mu.Unlock()
1263		-	if !found \|\| attempts < 1 \|\| now.Sub(oldest) < deadAfter {
1264		-	return
	1226	+	t.setLinkState(state, nil)
	1227	+	if hop {
	1228	+	t.hopPacketConnOnce()
1265	1229		}
1266		-	t.setLinkState(ports.LinkStateDead, ErrLinkDead)
1267		-	t.closeWithError(ErrLinkDead)
1268	1230		}
1269	1231
1270	1232		func (t *Transport) setLinkState(state ports.LinkState, err error) {

@@ -1276,6 +1238,7 @@ func (t *Transport) setLinkState(state ports.LinkState, err error) {
1276	1238		}
1277	1239		t.linkState = state
1278	1240		if state == ports.LinkStateConnected {
	1241	+	t.hoppedOffline = false
1279	1242		for _, p := range t.pending {
1280	1243		if p != nil {
1281	1244		p.attempts = 0

 	sendDir, recvDir uint32
 	mtu              int
-	mu                      sync.Mutex
+	mu             sync.Mutex
-	peer                    net.Addr
+	peer           net.Addr
-	ctr                     uint64
+	ctr            uint64
-	seq                     uint64
+	seq            uint64
-	probeSeq                uint64
+	probeSeq       uint64
-	pending                 map[uint64]*pending
+	pending        map[uint64]*pending
-	closed                  bool
+	closed         bool
-	closeErr                error
+	closeErr       error
-	lastHeard               time.Time
+	health         healthTracker
+	peerCounter    uint64
+	peerCounterSet bool
+	// These mirrors preserve the established diagnostic snapshot fields while
+	// health owns link-state decisions.
 	lastAuthenticatedPacket time.Time
 	lastCompleteRecord      time.Time
 	lastACKProgress         time.Time
 		reasm:                   pdgram.NewReassembler(),
 		in:                      make(chan ports.Frame, 32),
 		done:                    make(chan struct{}),
-		lastHeard:               opts.Clock.Now(),
+		health:                  newHealthTracker(opts.Clock.Now()),
 		lastAuthenticatedPacket: opts.Clock.Now(),
 		lastCompleteRecord:      opts.Clock.Now(),
 		lastACKProgress:         opts.Clock.Now(),
 			return
 		}
 		t.recvMu.Lock()
-		_, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
+		counter, pt, err := t.codec.Open(append([]byte(nil), buf[:n]...), t.recvDir, nil, t.replay)
 		if err != nil {
 			t.recvMu.Unlock()
 			continue
 		t.reassemblyInflight = inflight
 		var afterAuthenticated func()
 		if err == nil {
-			t.lastAuthenticatedPacket = t.clock.Now()
+			now := t.clock.Now()
+			t.health.authenticatedPacket(now)
+			t.lastAuthenticatedPacket = now
+			if !t.peerCounterSet || counter > t.peerCounter {
+				t.peer = addr
+				t.peerCounter = counter
+				t.peerCounterSet = true
+			}
 			afterAuthenticated = t.afterAuthenticatedPacket
 		}
 		t.mu.Unlock()
 			continue
 		}
 		t.recordCompleteRecord()
-		t.updatePeerFromAuthenticated(addr)
 		t.handleRecord(payload)
 	}
 }
 func (t *Transport) recordCompleteRecord() {
 	t.mu.Lock()
-	t.lastCompleteRecord = t.clock.Now()
+	now := t.clock.Now()
-	t.mu.Unlock()
+	t.health.completeRecord(now)
-}
+	t.lastCompleteRecord = now
-func (t *Transport) updatePeerFromAuthenticated(addr net.Addr) {
-	t.mu.Lock()
-	t.peer = addr
-	t.lastHeard = t.clock.Now()
-	t.hoppedOffline = false
 	t.mu.Unlock()
 	t.setLinkState(ports.LinkStateConnected, nil)
 }
 			acked = true
 		}
 		if acked {
-			t.lastACKProgress = t.clock.Now()
+			now := t.clock.Now()
+			t.health.ackProgress(now)
+			t.lastACKProgress = now
 			t.notifySendWaitersLocked()
 		}
 		t.mu.Unlock()
 		if acked {
+			t.setLinkState(ports.LinkStateConnected, nil)
 			t.emitDiagnostic()
 		}
 	case recProbe:
 		select {
 		case <-health.C():
 			t.checkSilence()
-			t.checkSendStall()
 			health.Reset(t.resendAfter)
 		case <-t.done:
 			return
 func (t *Transport) checkSilence() {
 	t.mu.Lock()
-	last := t.lastHeard
 	closed := t.closed
-	degradedAfter := t.degradedAfter
-	probeAfter := t.probeAfter
-	offlineAfter := t.offlineAfter
-	deadAfter := t.deadAfter
 	now := t.clock.Now()
+	state, hop, dead := t.health.decide(now, len(t.pending) > 0, t.degradedAfter, t.probeAfter, t.offlineAfter, t.deadAfter)
 	t.mu.Unlock()
 	if closed {
 		return
 	}
-	silentFor := now.Sub(last)
+	if dead {
-	switch {
+		t.setLinkState(state, ErrLinkDead)
-	case deadAfter > 0 && silentFor >= deadAfter:
-		t.setLinkState(ports.LinkStateDead, ErrLinkDead)
 		t.closeWithError(ErrLinkDead)
-	case offlineAfter > 0 && silentFor >= offlineAfter:
-		t.setLinkState(ports.LinkStateOffline, nil)
-	case probeAfter > 0 && silentFor >= probeAfter:
-		t.setLinkState(ports.LinkStateProbing, nil)
-		t.hopPacketConnOnce()
-	case degradedAfter > 0 && silentFor >= degradedAfter:
-		t.setLinkState(ports.LinkStateDegraded, nil)
-	}
-}
-// checkSendStall declares the link dead when the oldest unacked reliable frame
-// has gone unacknowledged for deadAfter despite at least one retransmit. The
-// receive-silence check in checkSilence misses one-way-dead links whose receive
-// side stays fresh from peer heartbeats; this catches them at the layer that
-// owns death. enqueued is set once when a frame is accepted and never refreshed,
-// so acks (which delete pending entries) keep the oldest age bounded on a
-// progressing link — only a frame that is never acked crosses the threshold.
-func (t *Transport) checkSendStall() {
-	t.mu.Lock()
-	if t.closed || t.deadAfter <= 0 {
-		t.mu.Unlock()
 		return
 	}
-	deadAfter := t.deadAfter
+	t.setLinkState(state, nil)
-	now := t.clock.Now()
+	if hop {
-	var oldest time.Time
+		t.hopPacketConnOnce()
-	var attempts int
-	found := false
-	for _, p := range t.pending {
-		if p == nil || p.enqueued.IsZero() {
-			continue
-		}
-		if !found || p.enqueued.Before(oldest) {
-			oldest = p.enqueued
-			attempts = p.attempts
-			found = true
-		}
-	}
-	t.mu.Unlock()
-	if !found || attempts < 1 || now.Sub(oldest) < deadAfter {
-		return
 	}
-	t.setLinkState(ports.LinkStateDead, ErrLinkDead)
-	t.closeWithError(ErrLinkDead)
 }
 func (t *Transport) setLinkState(state ports.LinkState, err error) {
 	}
 	t.linkState = state
 	if state == ports.LinkStateConnected {
+		t.hoppedOffline = false
 		for _, p := range t.pending {
 			if p != nil {
 				p.attempts = 0

M internal/adapters/dgram/transport_test.go +217 -50


@@ -877,6 +877,126 @@ func TestAuthenticatedPeerChangeUpdatesPeerAndEmitsConnected(t *testing.T) {
877	877		}
878	878		}
879	879
	880	+	func TestAuthenticatedFragmentSeparatesContactProgressAndPeerEligibility(t *testing.T) {
	881	+	aPC, bPC := newPair()
	882	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
	883	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	884	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
	885	+	})
	886	+	if err != nil {
	887	+	t.Fatal(err)
	888	+	}
	889	+	defer func() { _ = a.Close() }()
	890	+	defer func() { _ = bPC.Close() }()
	891	+
	892	+	a.setLinkState(ports.LinkStateDegraded, nil)
	893	+	<-a.LinkEvents()
	894	+	codec, err := pdgram.NewCodec(key())
	895	+	if err != nil {
	896	+	t.Fatal(err)
	897	+	}
	898	+	authenticated := make(chan struct{}, 8)
	899	+	malformed := make(chan struct{}, 8)
	900	+	a.mu.Lock()
	901	+	a.afterAuthenticatedPacket = func() { authenticated <- struct{}{} }
	902	+	a.afterMalformedFragment = func() { malformed <- struct{}{} }
	903	+	initialPacket := a.health.lastPacket
	904	+	initialRecord := a.health.lastRecord
	905	+	a.mu.Unlock()
	906	+
	907	+	sealFragment := func(counter uint64, frag pdgram.Fragment) []byte {
	908	+	t.Helper()
	909	+	raw, marshalErr := pdgram.MarshalFragment(frag)
	910	+	if marshalErr != nil {
	911	+	t.Fatal(marshalErr)
	912	+	}
	913	+	return codec.Seal(2, counter, raw, nil)
	914	+	}
	915	+	waitSignal := func(ch <-chan struct{}, what string) {
	916	+	t.Helper()
	917	+	select {
	918	+	case <-ch:
	919	+	case <-time.After(time.Second):
	920	+	t.Fatalf("timeout waiting for %s", what)
	921	+	}
	922	+	}
	923	+	assertHealth := func(wantPacket, wantRecord time.Time, wantPeer string, wantState ports.LinkState) {
	924	+	t.Helper()
	925	+	a.mu.Lock()
	926	+	gotPacket := a.health.lastPacket
	927	+	gotRecord := a.health.lastRecord
	928	+	a.mu.Unlock()
	929	+	if !gotPacket.Equal(wantPacket) \|\| !gotRecord.Equal(wantRecord) {
	930	+	t.Fatalf("health packet/record=(%v,%v), want (%v,%v)", gotPacket, gotRecord, wantPacket, wantRecord)
	931	+	}
	932	+	if got := a.Peer().String(); got != wantPeer {
	933	+	t.Fatalf("peer=%s, want %s", got, wantPeer)
	934	+	}
	935	+	if got := a.LinkState(); got != wantState {
	936	+	t.Fatalf("state=%v, want %v", got, wantState)
	937	+	}
	938	+	}
	939	+
	940	+	// Neither failed authentication nor an authenticated but invalid fragment
	941	+	// crosses the contact or roaming trust boundary.
	942	+	badAEAD := codec.Seal(2, 1, []byte("not a fragment"), nil)
	943	+	badAEAD[len(badAEAD)-1] ^= 0xff
	944	+	aPC.in <- packet{b: badAEAD, addr: testAddr("roam1")}
	945	+	aPC.in <- packet{b: codec.Seal(2, 2, []byte("not a fragment"), nil), addr: testAddr("roam1")}
	946	+	waitSignal(malformed, "malformed fragment")
	947	+	assertHealth(initialPacket, initialRecord, "b", ports.LinkStateDegraded)
	948	+
	949	+	clk.advance(time.Second)
	950	+	partial := pdgram.Fragment{Seq: 100, Index: 0, Count: 2, Data: []byte("half")}
	951	+	acceptedPacket := clk.Now()
	952	+	acceptedRaw := sealFragment(10, partial)
	953	+	aPC.in <- packet{b: acceptedRaw, addr: testAddr("roam1")}
	954	+	waitSignal(authenticated, "accepted incomplete fragment")
	955	+	assertHealth(acceptedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
	956	+
	957	+	// A replay and an Add-time fragment inconsistency cannot refresh contact or
	958	+	// migrate the peer. Reassembly diagnostics still update on the Add error.
	959	+	aPC.in <- packet{b: acceptedRaw, addr: testAddr("roam2")}
	960	+	aPC.in <- packet{b: codec.Seal(2, 11, []byte("still not a fragment"), nil), addr: testAddr("roam2")}
	961	+	waitSignal(malformed, "replay barrier")
	962	+	assertHealth(acceptedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
	963	+
	964	+	aPC.in <- packet{b: sealFragment(12, pdgram.Fragment{Seq: 100, Index: 1, Count: 3, Data: []byte("mismatch")}), addr: testAddr("roam2")}
	965	+	waitSignal(malformed, "reassembly rejection")
	966	+	assertHealth(acceptedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
	967	+
	968	+	// Fresh reordered packets refresh contact, but only the greatest accepted
	969	+	// AEAD counter is eligible to move the roaming peer. Contact alone must not
	970	+	// re-arm socket hopping while stream progress remains stalled.
	971	+	a.mu.Lock()
	972	+	a.hoppedOffline = true
	973	+	a.mu.Unlock()
	974	+	clk.advance(time.Second)
	975	+	reorderedPacket := clk.Now()
	976	+	aPC.in <- packet{b: sealFragment(9, pdgram.Fragment{Seq: 101, Index: 0, Count: 2, Data: []byte("older")}), addr: testAddr("roam2")}
	977	+	waitSignal(authenticated, "fresh reordered fragment")
	978	+	assertHealth(reorderedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
	979	+	a.mu.Lock()
	980	+	if !a.hoppedOffline {
	981	+	a.mu.Unlock()
	982	+	t.Fatal("fragment-only contact re-armed socket hopping")
	983	+	}
	984	+	a.mu.Unlock()
	985	+
	986	+	clk.advance(time.Second)
	987	+	newestPacket := clk.Now()
	988	+	aPC.in <- packet{b: sealFragment(13, pdgram.Fragment{Seq: 102, Index: 0, Count: 2, Data: []byte("newer")}), addr: testAddr("roam2")}
	989	+	waitSignal(authenticated, "new greatest fragment")
	990	+	assertHealth(newestPacket, initialRecord, "roam2", ports.LinkStateDegraded)
	991	+
	992	+	clk.advance(time.Second)
	993	+	completeAt := clk.Now()
	994	+	aPC.in <- packet{b: sealFragment(14, pdgram.Fragment{Seq: 103, Index: 0, Count: 1, Data: probeRecord(recPong, 999)}), addr: testAddr("roam2")}
	995	+	waitSignal(authenticated, "complete record")
	996	+	eventually(t, time.Second, func() bool { return a.LinkState() == ports.LinkStateConnected })
	997	+	assertHealth(completeAt, completeAt, "roam2", ports.LinkStateConnected)
	998	+	}
	999	+
880	1000		func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
881	1001		aPC, bPC := newPair()
882	1002		var hopped atomic.Bool

@@ -900,7 +1020,7 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
900	1020		t.Fatal(err)
901	1021		}
902	1022		a.mu.Lock()
903		-	a.lastHeard = time.Now().Add(-time.Second)
	1023	+	a.health.lastPacket = time.Now().Add(-time.Second)
904	1024		a.mu.Unlock()
905	1025		a.checkSilence()
906	1026		if !hopped.Load() {

@@ -940,7 +1060,7 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
940	1060		defer func() { _ = b.Close() }()
941	1061
942	1062		a.mu.Lock()
943		-	a.lastHeard = time.Now().Add(-time.Second)
	1063	+	a.health.lastPacket = time.Now().Add(-time.Second)
944	1064		a.mu.Unlock()
945	1065		a.checkSilence()
946	1066		a.mu.Lock()

@@ -966,7 +1086,7 @@ func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
966	1086		defer func() { _ = a.Close() }()
967	1087		defer func() { _ = b.Close() }()
968	1088		a.mu.Lock()
969		-	a.lastHeard = time.Now().Add(-time.Second)
	1089	+	a.health.lastPacket = time.Now().Add(-time.Second)
970	1090		a.mu.Unlock()
971	1091		a.checkSilence()
972	1092		if a.pc != aPC {

@@ -1272,7 +1392,7 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
1272	1392		defer func() { _ = a.Close() }()
1273	1393		defer func() { _ = b.Close() }()
1274	1394		a.mu.Lock()
1275		-	a.lastHeard = time.Now().Add(-time.Second)
	1395	+	a.health.lastPacket = time.Now().Add(-time.Second)
1276	1396		a.mu.Unlock()
1277	1397
1278	1398		_, err := recvErrWithin(t, a, time.Second)

@@ -1684,7 +1804,8 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) {
1684	1804		for _, tt := range tests {
1685	1805		t.Run(tt.name, func(t *testing.T) {
1686	1806		a.mu.Lock()
1687		-	a.lastHeard = time.Now().Add(-tt.age)
	1807	+	staleAt := time.Now().Add(-tt.age)
	1808	+	a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
1688	1809		a.mu.Unlock()
1689	1810		a.checkSilence()
1690	1811		if got := a.LinkState(); got != tt.want {

@@ -1702,59 +1823,103 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) {
1702	1823		}
1703	1824		}
1704	1825
1705		-	func TestSendStallDeathRequiresOldRetransmittedFrame(t *testing.T) {
	1826	+	func TestLinkHealthSeparatesPacketContactFromUsefulProgress(t *testing.T) {
	1827	+	now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
	1828	+	threshold := func(seconds int) time.Duration { return time.Duration(seconds) * time.Second }
1706	1829		tests := []struct {
1707		-	name string
1708		-	age time.Duration
1709		-	attempts int
1710		-	wantDead bool
	1830	+	name string
	1831	+	packetAge time.Duration
	1832	+	recordAge time.Duration
	1833	+	ackAge time.Duration
	1834	+	hasPending bool
	1835	+	wantState ports.LinkState
	1836	+	wantHop bool
	1837	+	wantDead bool
1711	1838		}{
1712		-	{name: "recent frame stays alive", age: time.Second, attempts: 5, wantDead: false},
1713		-	{name: "old but never retransmitted stays alive", age: 61 * time.Second, attempts: 0, wantDead: false},
1714		-	{name: "old retransmitted frame dies", age: 61 * time.Second, attempts: 3, wantDead: true},
	1839	+	{name: "fresh contact without useful progress degrades", packetAge: 0, recordAge: threshold(10), ackAge: threshold(10), wantState: ports.LinkStateDegraded},
	1840	+	{name: "pending round trip stall probes and hops", packetAge: 0, recordAge: threshold(20), ackAge: threshold(20), hasPending: true, wantState: ports.LinkStateProbing, wantHop: true},
	1841	+	{name: "fragment-only contact never becomes offline", packetAge: 0, recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateDegraded},
	1842	+	{name: "packet silence probes and hops", packetAge: threshold(20), recordAge: threshold(20), ackAge: threshold(20), wantState: ports.LinkStateProbing, wantHop: true},
	1843	+	{name: "packet silence becomes offline", packetAge: threshold(30), recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateOffline},
	1844	+	{name: "packet silence becomes dead", packetAge: threshold(60), recordAge: threshold(60), ackAge: threshold(60), wantState: ports.LinkStateDead, wantDead: true},
	1845	+	{name: "complete record restores connected", packetAge: 0, recordAge: 0, ackAge: threshold(30), hasPending: true, wantState: ports.LinkStateConnected},
	1846	+	{name: "ack progress restores connected", packetAge: 0, recordAge: threshold(30), ackAge: 0, hasPending: true, wantState: ports.LinkStateConnected},
1715	1847		}
	1848	+
1716	1849		for _, tt := range tests {
1717	1850		t.Run(tt.name, func(t *testing.T) {
1718		-	aPC, bPC := newPair()
1719		-	// Drop everything a sends so the frame never gets acked, and keep the
1720		-	// receive side fresh so only the send-stall check can declare death.
1721		-	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }
1722		-	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
1723		-	ResendAfter: time.Hour, Heartbeat: time.Hour, DeadAfter: 60 * time.Second,
1724		-	})
1725		-	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
1726		-	defer func() { _ = a.Close() }()
1727		-	defer func() { _ = b.Close() }()
1728		-
1729		-	now := time.Now()
1730		-	a.mu.Lock()
1731		-	a.lastHeard = now
1732		-	a.pending[1] = &pending{
1733		-	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stuck")},
1734		-	enqueued: now.Add(-tt.age),
1735		-	last: now,
1736		-	attempts: tt.attempts,
	1851	+	health := healthTracker{
	1852	+	lastPacket: now.Add(-tt.packetAge),
	1853	+	lastRecord: now.Add(-tt.recordAge),
	1854	+	lastProgress: now.Add(-tt.ackAge),
1737	1855		}
1738		-	a.mu.Unlock()
1739		-
1740		-	a.checkSendStall()
1741		-	if dead := a.LinkState() == ports.LinkStateDead; dead != tt.wantDead {
1742		-	t.Fatalf("dead=%v, want %v", dead, tt.wantDead)
1743		-	}
1744		-	select {
1745		-	case <-a.done:
1746		-	if !tt.wantDead {
1747		-	t.Fatal("transport closed without a send stall")
1748		-	}
1749		-	default:
1750		-	if tt.wantDead {
1751		-	t.Fatal("transport not closed after send stall")
1752		-	}
	1856	+	state, hop, dead := health.decide(now, tt.hasPending, threshold(10), threshold(20), threshold(30), threshold(60))
	1857	+	if state != tt.wantState \|\| hop != tt.wantHop \|\| dead != tt.wantDead {
	1858	+	t.Fatalf("decide()=(%v,%v,%v), want (%v,%v,%v)", state, hop, dead, tt.wantState, tt.wantHop, tt.wantDead)
1753	1859		}
1754	1860		})
1755	1861		}
1756	1862		}
1757	1863
	1864	+	func TestLinkHealthPendingStallWithFreshContactProbesOnceWithoutDying(t *testing.T) {
	1865	+	aPC, bPC := newPair()
	1866	+	var hops atomic.Int32
	1867	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
	1868	+	ResendAfter: time.Hour, Heartbeat: time.Hour, ProbeAfter: 20 * time.Second, DeadAfter: 60 * time.Second,
	1869	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
	1870	+	hops.Add(1)
	1871	+	return &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
	1872	+	},
	1873	+	})
	1874	+	defer func() { _ = a.Close() }()
	1875	+	defer func() { _ = bPC.Close() }()
	1876	+
	1877	+	now := time.Now()
	1878	+	a.mu.Lock()
	1879	+	a.health = healthTracker{lastPacket: now, lastRecord: now.Add(-time.Minute), lastProgress: now.Add(-time.Minute)}
	1880	+	a.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, enqueued: now.Add(-time.Minute), attempts: 3}
	1881	+	a.mu.Unlock()
	1882	+
	1883	+	a.checkSilence()
	1884	+	a.checkSilence()
	1885	+	if got := a.LinkState(); got != ports.LinkStateProbing {
	1886	+	t.Fatalf("state=%v, want probing", got)
	1887	+	}
	1888	+	if got := hops.Load(); got != 1 {
	1889	+	t.Fatalf("socket hops=%d, want one", got)
	1890	+	}
	1891	+	select {
	1892	+	case <-a.done:
	1893	+	t.Fatal("fresh authenticated contact must prevent pending-stall death")
	1894	+	default:
	1895	+	}
	1896	+	}
	1897	+
	1898	+	func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing.T) {
	1899	+	tr, _ := newResendTestTransport(t, 10)
	1900	+	now := tr.clock.Now()
	1901	+	tr.linkState = ports.LinkStateDegraded
	1902	+	tr.health = healthTracker{lastPacket: now, lastRecord: now.Add(-time.Minute), lastProgress: now.Add(-time.Minute)}
	1903	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: now, last: now}
	1904	+
	1905	+	tr.handleRecord(ackRecord(1))
	1906	+	if got := tr.LinkState(); got != ports.LinkStateConnected {
	1907	+	t.Fatalf("state after advancing ACK=%v, want connected", got)
	1908	+	}
	1909	+	tr.mu.Lock()
	1910	+	if tr.health.lastProgress.Before(now) {
	1911	+	tr.mu.Unlock()
	1912	+	t.Fatalf("last progress=%v, want at or after %v", tr.health.lastProgress, now)
	1913	+	}
	1914	+	tr.mu.Unlock()
	1915	+
	1916	+	tr.setLinkState(ports.LinkStateDegraded, nil)
	1917	+	tr.handleRecord(ackRecord(1))
	1918	+	if got := tr.LinkState(); got != ports.LinkStateDegraded {
	1919	+	t.Fatalf("duplicate ACK state=%v, want degraded", got)
	1920	+	}
	1921	+	}
	1922	+
1758	1923		func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1759	1924		aPC, bPC := newPair()
1760	1925		var hopped atomic.Bool

@@ -1773,7 +1938,8 @@ func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1773	1938		defer func() { _ = b.Close() }()
1774	1939
1775	1940		a.mu.Lock()
1776		-	a.lastHeard = time.Now().Add(-time.Second)
	1941	+	staleAt := time.Now().Add(-time.Second)
	1942	+	a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
1777	1943		a.mu.Unlock()
1778	1944		a.checkSilence()
1779	1945		if hopped.Load() {

@@ -1781,7 +1947,7 @@ func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1781	1947		}
1782	1948
1783	1949		a.mu.Lock()
1784		-	a.lastHeard = time.Now().Add(-2 * time.Second)
	1950	+	a.health.lastPacket = time.Now().Add(-2 * time.Second)
1785	1951		a.mu.Unlock()
1786	1952		a.checkSilence()
1787	1953		if !hopped.Load() {

@@ -1796,7 +1962,8 @@ func TestShortSilenceDoesNotKillTransport(t *testing.T) {
1796	1962		defer func() { _ = a.Close() }()
1797	1963		defer func() { _ = b.Close() }()
1798	1964		a.mu.Lock()
1799		-	a.lastHeard = time.Now().Add(-time.Second)
	1965	+	staleAt := time.Now().Add(-time.Second)
	1966	+	a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
1800	1967		a.mu.Unlock()
1801	1968		a.checkSilence()
1802	1969		if got := a.LinkState(); got != ports.LinkStateDegraded {


@@ -877,6 +877,126 @@ func TestAuthenticatedPeerChangeUpdatesPeerAndEmitsConnected(t *testing.T) {
877		}	877		}
878		}	878		}
879			879
			880	+	func TestAuthenticatedFragmentSeparatesContactProgressAndPeerEligibility(t *testing.T) {
			881	+	aPC, bPC := newPair()
			882	+	clk := newManualClock(time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC))
			883	+	a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
			884	+	Clock: clk, ResendAfter: time.Hour, Heartbeat: time.Hour,
			885	+	})
			886	+	if err != nil {
			887	+	t.Fatal(err)
			888	+	}
			889	+	defer func() { _ = a.Close() }()
			890	+	defer func() { _ = bPC.Close() }()
			891	+
			892	+	a.setLinkState(ports.LinkStateDegraded, nil)
			893	+	<-a.LinkEvents()
			894	+	codec, err := pdgram.NewCodec(key())
			895	+	if err != nil {
			896	+	t.Fatal(err)
			897	+	}
			898	+	authenticated := make(chan struct{}, 8)
			899	+	malformed := make(chan struct{}, 8)
			900	+	a.mu.Lock()
			901	+	a.afterAuthenticatedPacket = func() { authenticated <- struct{}{} }
			902	+	a.afterMalformedFragment = func() { malformed <- struct{}{} }
			903	+	initialPacket := a.health.lastPacket
			904	+	initialRecord := a.health.lastRecord
			905	+	a.mu.Unlock()
			906	+
			907	+	sealFragment := func(counter uint64, frag pdgram.Fragment) []byte {
			908	+	t.Helper()
			909	+	raw, marshalErr := pdgram.MarshalFragment(frag)
			910	+	if marshalErr != nil {
			911	+	t.Fatal(marshalErr)
			912	+	}
			913	+	return codec.Seal(2, counter, raw, nil)
			914	+	}
			915	+	waitSignal := func(ch <-chan struct{}, what string) {
			916	+	t.Helper()
			917	+	select {
			918	+	case <-ch:
			919	+	case <-time.After(time.Second):
			920	+	t.Fatalf("timeout waiting for %s", what)
			921	+	}
			922	+	}
			923	+	assertHealth := func(wantPacket, wantRecord time.Time, wantPeer string, wantState ports.LinkState) {
			924	+	t.Helper()
			925	+	a.mu.Lock()
			926	+	gotPacket := a.health.lastPacket
			927	+	gotRecord := a.health.lastRecord
			928	+	a.mu.Unlock()
			929	+	if !gotPacket.Equal(wantPacket) \|\| !gotRecord.Equal(wantRecord) {
			930	+	t.Fatalf("health packet/record=(%v,%v), want (%v,%v)", gotPacket, gotRecord, wantPacket, wantRecord)
			931	+	}
			932	+	if got := a.Peer().String(); got != wantPeer {
			933	+	t.Fatalf("peer=%s, want %s", got, wantPeer)
			934	+	}
			935	+	if got := a.LinkState(); got != wantState {
			936	+	t.Fatalf("state=%v, want %v", got, wantState)
			937	+	}
			938	+	}
			939	+
			940	+	// Neither failed authentication nor an authenticated but invalid fragment
			941	+	// crosses the contact or roaming trust boundary.
			942	+	badAEAD := codec.Seal(2, 1, []byte("not a fragment"), nil)
			943	+	badAEAD[len(badAEAD)-1] ^= 0xff
			944	+	aPC.in <- packet{b: badAEAD, addr: testAddr("roam1")}
			945	+	aPC.in <- packet{b: codec.Seal(2, 2, []byte("not a fragment"), nil), addr: testAddr("roam1")}
			946	+	waitSignal(malformed, "malformed fragment")
			947	+	assertHealth(initialPacket, initialRecord, "b", ports.LinkStateDegraded)
			948	+
			949	+	clk.advance(time.Second)
			950	+	partial := pdgram.Fragment{Seq: 100, Index: 0, Count: 2, Data: []byte("half")}
			951	+	acceptedPacket := clk.Now()
			952	+	acceptedRaw := sealFragment(10, partial)
			953	+	aPC.in <- packet{b: acceptedRaw, addr: testAddr("roam1")}
			954	+	waitSignal(authenticated, "accepted incomplete fragment")
			955	+	assertHealth(acceptedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
			956	+
			957	+	// A replay and an Add-time fragment inconsistency cannot refresh contact or
			958	+	// migrate the peer. Reassembly diagnostics still update on the Add error.
			959	+	aPC.in <- packet{b: acceptedRaw, addr: testAddr("roam2")}
			960	+	aPC.in <- packet{b: codec.Seal(2, 11, []byte("still not a fragment"), nil), addr: testAddr("roam2")}
			961	+	waitSignal(malformed, "replay barrier")
			962	+	assertHealth(acceptedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
			963	+
			964	+	aPC.in <- packet{b: sealFragment(12, pdgram.Fragment{Seq: 100, Index: 1, Count: 3, Data: []byte("mismatch")}), addr: testAddr("roam2")}
			965	+	waitSignal(malformed, "reassembly rejection")
			966	+	assertHealth(acceptedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
			967	+
			968	+	// Fresh reordered packets refresh contact, but only the greatest accepted
			969	+	// AEAD counter is eligible to move the roaming peer. Contact alone must not
			970	+	// re-arm socket hopping while stream progress remains stalled.
			971	+	a.mu.Lock()
			972	+	a.hoppedOffline = true
			973	+	a.mu.Unlock()
			974	+	clk.advance(time.Second)
			975	+	reorderedPacket := clk.Now()
			976	+	aPC.in <- packet{b: sealFragment(9, pdgram.Fragment{Seq: 101, Index: 0, Count: 2, Data: []byte("older")}), addr: testAddr("roam2")}
			977	+	waitSignal(authenticated, "fresh reordered fragment")
			978	+	assertHealth(reorderedPacket, initialRecord, "roam1", ports.LinkStateDegraded)
			979	+	a.mu.Lock()
			980	+	if !a.hoppedOffline {
			981	+	a.mu.Unlock()
			982	+	t.Fatal("fragment-only contact re-armed socket hopping")
			983	+	}
			984	+	a.mu.Unlock()
			985	+
			986	+	clk.advance(time.Second)
			987	+	newestPacket := clk.Now()
			988	+	aPC.in <- packet{b: sealFragment(13, pdgram.Fragment{Seq: 102, Index: 0, Count: 2, Data: []byte("newer")}), addr: testAddr("roam2")}
			989	+	waitSignal(authenticated, "new greatest fragment")
			990	+	assertHealth(newestPacket, initialRecord, "roam2", ports.LinkStateDegraded)
			991	+
			992	+	clk.advance(time.Second)
			993	+	completeAt := clk.Now()
			994	+	aPC.in <- packet{b: sealFragment(14, pdgram.Fragment{Seq: 103, Index: 0, Count: 1, Data: probeRecord(recPong, 999)}), addr: testAddr("roam2")}
			995	+	waitSignal(authenticated, "complete record")
			996	+	eventually(t, time.Second, func() bool { return a.LinkState() == ports.LinkStateConnected })
			997	+	assertHealth(completeAt, completeAt, "roam2", ports.LinkStateConnected)
			998	+	}
			999	+
880		func TestPortHopPreservesPendingReliableMessages(t *testing.T) {	1000		func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
881		aPC, bPC := newPair()	1001		aPC, bPC := newPair()
882		var hopped atomic.Bool	1002		var hopped atomic.Bool

@@ -900,7 +1020,7 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
900		t.Fatal(err)	1020		t.Fatal(err)
901		}	1021		}
902		a.mu.Lock()	1022		a.mu.Lock()
903	-	a.lastHeard = time.Now().Add(-time.Second)	1023	+	a.health.lastPacket = time.Now().Add(-time.Second)
904		a.mu.Unlock()	1024		a.mu.Unlock()
905		a.checkSilence()	1025		a.checkSilence()
906		if !hopped.Load() {	1026		if !hopped.Load() {

@@ -940,7 +1060,7 @@ func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
940		defer func() { _ = b.Close() }()	1060		defer func() { _ = b.Close() }()
941			1061
942		a.mu.Lock()	1062		a.mu.Lock()
943	-	a.lastHeard = time.Now().Add(-time.Second)	1063	+	a.health.lastPacket = time.Now().Add(-time.Second)
944		a.mu.Unlock()	1064		a.mu.Unlock()
945		a.checkSilence()	1065		a.checkSilence()
946		a.mu.Lock()	1066		a.mu.Lock()

@@ -966,7 +1086,7 @@ func TestServerWithoutRebindDoesNotHopPorts(t *testing.T) {
966		defer func() { _ = a.Close() }()	1086		defer func() { _ = a.Close() }()
967		defer func() { _ = b.Close() }()	1087		defer func() { _ = b.Close() }()
968		a.mu.Lock()	1088		a.mu.Lock()
969	-	a.lastHeard = time.Now().Add(-time.Second)	1089	+	a.health.lastPacket = time.Now().Add(-time.Second)
970		a.mu.Unlock()	1090		a.mu.Unlock()
971		a.checkSilence()	1091		a.checkSilence()
972		if a.pc != aPC {	1092		if a.pc != aPC {

@@ -1272,7 +1392,7 @@ func TestRecvUnblocksWhenPeerDeadAfter(t *testing.T) {
1272		defer func() { _ = a.Close() }()	1392		defer func() { _ = a.Close() }()
1273		defer func() { _ = b.Close() }()	1393		defer func() { _ = b.Close() }()
1274		a.mu.Lock()	1394		a.mu.Lock()
1275	-	a.lastHeard = time.Now().Add(-time.Second)	1395	+	a.health.lastPacket = time.Now().Add(-time.Second)
1276		a.mu.Unlock()	1396		a.mu.Unlock()
1277			1397
1278		_, err := recvErrWithin(t, a, time.Second)	1398		_, err := recvErrWithin(t, a, time.Second)

@@ -1684,7 +1804,8 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) {
1684		for _, tt := range tests {	1804		for _, tt := range tests {
1685		t.Run(tt.name, func(t *testing.T) {	1805		t.Run(tt.name, func(t *testing.T) {
1686		a.mu.Lock()	1806		a.mu.Lock()
1687	-	a.lastHeard = time.Now().Add(-tt.age)	1807	+	staleAt := time.Now().Add(-tt.age)
			1808	+	a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
1688		a.mu.Unlock()	1809		a.mu.Unlock()
1689		a.checkSilence()	1810		a.checkSilence()
1690		if got := a.LinkState(); got != tt.want {	1811		if got := a.LinkState(); got != tt.want {

@@ -1702,59 +1823,103 @@ func TestLinkStateTransitionsFromConfigurableSilence(t *testing.T) {
1702		}	1823		}
1703		}	1824		}
1704			1825
1705	-	func TestSendStallDeathRequiresOldRetransmittedFrame(t *testing.T) {	1826	+	func TestLinkHealthSeparatesPacketContactFromUsefulProgress(t *testing.T) {
			1827	+	now := time.Date(2030, 1, 2, 3, 4, 5, 0, time.UTC)
			1828	+	threshold := func(seconds int) time.Duration { return time.Duration(seconds) * time.Second }
1706		tests := []struct {	1829		tests := []struct {
1707	-	name string	1830	+	name string
1708	-	age time.Duration	1831	+	packetAge time.Duration
1709	-	attempts int	1832	+	recordAge time.Duration
1710	-	wantDead bool	1833	+	ackAge time.Duration
			1834	+	hasPending bool
			1835	+	wantState ports.LinkState
			1836	+	wantHop bool
			1837	+	wantDead bool
1711		}{	1838		}{
1712	-	{name: "recent frame stays alive", age: time.Second, attempts: 5, wantDead: false},	1839	+	{name: "fresh contact without useful progress degrades", packetAge: 0, recordAge: threshold(10), ackAge: threshold(10), wantState: ports.LinkStateDegraded},
1713	-	{name: "old but never retransmitted stays alive", age: 61 * time.Second, attempts: 0, wantDead: false},	1840	+	{name: "pending round trip stall probes and hops", packetAge: 0, recordAge: threshold(20), ackAge: threshold(20), hasPending: true, wantState: ports.LinkStateProbing, wantHop: true},
1714	-	{name: "old retransmitted frame dies", age: 61 * time.Second, attempts: 3, wantDead: true},	1841	+	{name: "fragment-only contact never becomes offline", packetAge: 0, recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateDegraded},
			1842	+	{name: "packet silence probes and hops", packetAge: threshold(20), recordAge: threshold(20), ackAge: threshold(20), wantState: ports.LinkStateProbing, wantHop: true},
			1843	+	{name: "packet silence becomes offline", packetAge: threshold(30), recordAge: threshold(30), ackAge: threshold(30), wantState: ports.LinkStateOffline},
			1844	+	{name: "packet silence becomes dead", packetAge: threshold(60), recordAge: threshold(60), ackAge: threshold(60), wantState: ports.LinkStateDead, wantDead: true},
			1845	+	{name: "complete record restores connected", packetAge: 0, recordAge: 0, ackAge: threshold(30), hasPending: true, wantState: ports.LinkStateConnected},
			1846	+	{name: "ack progress restores connected", packetAge: 0, recordAge: threshold(30), ackAge: 0, hasPending: true, wantState: ports.LinkStateConnected},
1715		}	1847		}
			1848	+
1716		for _, tt := range tests {	1849		for _, tt := range tests {
1717		t.Run(tt.name, func(t *testing.T) {	1850		t.Run(tt.name, func(t *testing.T) {
1718	-	aPC, bPC := newPair()	1851	+	health := healthTracker{
1719	-	// Drop everything a sends so the frame never gets acked, and keep the	1852	+	lastPacket: now.Add(-tt.packetAge),
1720	-	// receive side fresh so only the send-stall check can declare death.	1853	+	lastRecord: now.Add(-tt.recordAge),
1721	-	aPC.drop = func(_ []byte, addr net.Addr) bool { return addr.String() == "b" }	1854	+	lastProgress: now.Add(-tt.ackAge),
1722	-	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
1723	-	ResendAfter: time.Hour, Heartbeat: time.Hour, DeadAfter: 60 * time.Second,
1724	-	})
1725	-	b, _ := NewTransport(bPC, testAddr("a"), key(), 2, 1)
1726	-	defer func() { _ = a.Close() }()
1727	-	defer func() { _ = b.Close() }()
1728	-
1729	-	now := time.Now()
1730	-	a.mu.Lock()
1731	-	a.lastHeard = now
1732	-	a.pending[1] = &pending{
1733	-	frame: ports.Frame{Type: ports.MsgOutput, Payload: []byte("stuck")},
1734	-	enqueued: now.Add(-tt.age),
1735	-	last: now,
1736	-	attempts: tt.attempts,
1737		}	1855		}
1738	-	a.mu.Unlock()	1856	+	state, hop, dead := health.decide(now, tt.hasPending, threshold(10), threshold(20), threshold(30), threshold(60))
1739	-		1857	+	if state != tt.wantState \|\| hop != tt.wantHop \|\| dead != tt.wantDead {
1740	-	a.checkSendStall()	1858	+	t.Fatalf("decide()=(%v,%v,%v), want (%v,%v,%v)", state, hop, dead, tt.wantState, tt.wantHop, tt.wantDead)
1741	-	if dead := a.LinkState() == ports.LinkStateDead; dead != tt.wantDead {
1742	-	t.Fatalf("dead=%v, want %v", dead, tt.wantDead)
1743	-	}
1744	-	select {
1745	-	case <-a.done:
1746	-	if !tt.wantDead {
1747	-	t.Fatal("transport closed without a send stall")
1748	-	}
1749	-	default:
1750	-	if tt.wantDead {
1751	-	t.Fatal("transport not closed after send stall")
1752	-	}
1753		}	1859		}
1754		})	1860		})
1755		}	1861		}
1756		}	1862		}
1757			1863
			1864	+	func TestLinkHealthPendingStallWithFreshContactProbesOnceWithoutDying(t *testing.T) {
			1865	+	aPC, bPC := newPair()
			1866	+	var hops atomic.Int32
			1867	+	a, _ := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{
			1868	+	ResendAfter: time.Hour, Heartbeat: time.Hour, ProbeAfter: 20 * time.Second, DeadAfter: 60 * time.Second,
			1869	+	RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
			1870	+	hops.Add(1)
			1871	+	return &fakePC{addr: testAddr("a2"), in: make(chan packet, 100), peers: map[string]*fakePC{"b": bPC}}, nil
			1872	+	},
			1873	+	})
			1874	+	defer func() { _ = a.Close() }()
			1875	+	defer func() { _ = bPC.Close() }()
			1876	+
			1877	+	now := time.Now()
			1878	+	a.mu.Lock()
			1879	+	a.health = healthTracker{lastPacket: now, lastRecord: now.Add(-time.Minute), lastProgress: now.Add(-time.Minute)}
			1880	+	a.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, enqueued: now.Add(-time.Minute), attempts: 3}
			1881	+	a.mu.Unlock()
			1882	+
			1883	+	a.checkSilence()
			1884	+	a.checkSilence()
			1885	+	if got := a.LinkState(); got != ports.LinkStateProbing {
			1886	+	t.Fatalf("state=%v, want probing", got)
			1887	+	}
			1888	+	if got := hops.Load(); got != 1 {
			1889	+	t.Fatalf("socket hops=%d, want one", got)
			1890	+	}
			1891	+	select {
			1892	+	case <-a.done:
			1893	+	t.Fatal("fresh authenticated contact must prevent pending-stall death")
			1894	+	default:
			1895	+	}
			1896	+	}
			1897	+
			1898	+	func TestLinkHealthOnlyAckProgressThatRemovesPendingRestoresConnected(t *testing.T) {
			1899	+	tr, _ := newResendTestTransport(t, 10)
			1900	+	now := tr.clock.Now()
			1901	+	tr.linkState = ports.LinkStateDegraded
			1902	+	tr.health = healthTracker{lastPacket: now, lastRecord: now.Add(-time.Minute), lastProgress: now.Add(-time.Minute)}
			1903	+	tr.pending[1] = &pending{frame: ports.Frame{Type: ports.MsgOutput}, first: now, last: now}
			1904	+
			1905	+	tr.handleRecord(ackRecord(1))
			1906	+	if got := tr.LinkState(); got != ports.LinkStateConnected {
			1907	+	t.Fatalf("state after advancing ACK=%v, want connected", got)
			1908	+	}
			1909	+	tr.mu.Lock()
			1910	+	if tr.health.lastProgress.Before(now) {
			1911	+	tr.mu.Unlock()
			1912	+	t.Fatalf("last progress=%v, want at or after %v", tr.health.lastProgress, now)
			1913	+	}
			1914	+	tr.mu.Unlock()
			1915	+
			1916	+	tr.setLinkState(ports.LinkStateDegraded, nil)
			1917	+	tr.handleRecord(ackRecord(1))
			1918	+	if got := tr.LinkState(); got != ports.LinkStateDegraded {
			1919	+	t.Fatalf("duplicate ACK state=%v, want degraded", got)
			1920	+	}
			1921	+	}
			1922	+
1758		func TestHopFiresAtProbingNotDegraded(t *testing.T) {	1923		func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1759		aPC, bPC := newPair()	1924		aPC, bPC := newPair()
1760		var hopped atomic.Bool	1925		var hopped atomic.Bool

@@ -1773,7 +1938,8 @@ func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1773		defer func() { _ = b.Close() }()	1938		defer func() { _ = b.Close() }()
1774			1939
1775		a.mu.Lock()	1940		a.mu.Lock()
1776	-	a.lastHeard = time.Now().Add(-time.Second)	1941	+	staleAt := time.Now().Add(-time.Second)
			1942	+	a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
1777		a.mu.Unlock()	1943		a.mu.Unlock()
1778		a.checkSilence()	1944		a.checkSilence()
1779		if hopped.Load() {	1945		if hopped.Load() {

@@ -1781,7 +1947,7 @@ func TestHopFiresAtProbingNotDegraded(t *testing.T) {
1781		}	1947		}
1782			1948
1783		a.mu.Lock()	1949		a.mu.Lock()
1784	-	a.lastHeard = time.Now().Add(-2 * time.Second)	1950	+	a.health.lastPacket = time.Now().Add(-2 * time.Second)
1785		a.mu.Unlock()	1951		a.mu.Unlock()
1786		a.checkSilence()	1952		a.checkSilence()
1787		if !hopped.Load() {	1953		if !hopped.Load() {

@@ -1796,7 +1962,8 @@ func TestShortSilenceDoesNotKillTransport(t *testing.T) {
1796		defer func() { _ = a.Close() }()	1962		defer func() { _ = a.Close() }()
1797		defer func() { _ = b.Close() }()	1963		defer func() { _ = b.Close() }()
1798		a.mu.Lock()	1964		a.mu.Lock()
1799	-	a.lastHeard = time.Now().Add(-time.Second)	1965	+	staleAt := time.Now().Add(-time.Second)
			1966	+	a.health = healthTracker{lastPacket: staleAt, lastRecord: staleAt, lastProgress: staleAt}
1800		a.mu.Unlock()	1967		a.mu.Unlock()
1801		a.checkSilence()	1968		a.checkSilence()
1802		if got := a.LinkState(); got != ports.LinkStateDegraded {	1969		if got := a.LinkState(); got != ports.LinkStateDegraded {

M internal/ports/link.go +12 -0


@@ -3,13 +3,25 @@ package ports
3	3		import "time"
4	4
5	5		// LinkState reports best-effort transport health without changing Transport.
	6	+	// Implementations may distinguish authenticated path contact from useful
	7	+	// stream progress when classifying the link.
6	8		type LinkState int
7	9
8	10		const (
	11	+	// LinkStateConnected means a complete record or meaningful acknowledgement
	12	+	// progress was observed recently.
9	13		LinkStateConnected LinkState = iota
	14	+	// LinkStateDegraded means authenticated packets still reach the path, but
	15	+	// complete records and acknowledgement progress have stalled.
10	16		LinkStateDegraded
	17	+	// LinkStateProbing means packet contact is silent or pending data has made no
	18	+	// round-trip progress long enough for path recovery to begin.
11	19		LinkStateProbing
	20	+	// LinkStateOffline means authenticated packet contact has been absent long
	21	+	// enough for the client to resume over its fallback transport.
12	22		LinkStateOffline
	23	+	// LinkStateDead means prolonged authenticated-packet silence has made the
	24	+	// transport terminal.
13	25		LinkStateDead
14	26		)
15	27


@@ -3,13 +3,25 @@ package ports
3	import "time"	3		import "time"
4		4
5	// LinkState reports best-effort transport health without changing Transport.	5		// LinkState reports best-effort transport health without changing Transport.
		6	+	// Implementations may distinguish authenticated path contact from useful
		7	+	// stream progress when classifying the link.
6	type LinkState int	8		type LinkState int
7		9
8	const (	10		const (
		11	+	// LinkStateConnected means a complete record or meaningful acknowledgement
		12	+	// progress was observed recently.
9	LinkStateConnected LinkState = iota	13		LinkStateConnected LinkState = iota
		14	+	// LinkStateDegraded means authenticated packets still reach the path, but
		15	+	// complete records and acknowledgement progress have stalled.
10	LinkStateDegraded	16		LinkStateDegraded
		17	+	// LinkStateProbing means packet contact is silent or pending data has made no
		18	+	// round-trip progress long enough for path recovery to begin.
11	LinkStateProbing	19		LinkStateProbing
		20	+	// LinkStateOffline means authenticated packet contact has been absent long
		21	+	// enough for the client to resume over its fallback transport.
12	LinkStateOffline	22		LinkStateOffline
		23	+	// LinkStateDead means prolonged authenticated-packet silence has made the
		24	+	// transport terminal.
13	LinkStateDead	25		LinkStateDead
14	)	26		)
15		27

M internal/usecase/client/client.go +5 -0


@@ -461,6 +461,11 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
461	461		clearStatus()
462	462		continue
463	463		}
	464	+	if ev.State == ports.LinkStateDegraded {
	465	+	log.Warn("UDP link degraded")
	466	+	clearStatus()
	467	+	continue
	468	+	}
464	469		drawStatusStage(stageForLinkState(ev.State))
465	470		if ev.State == ports.LinkStateOffline {
466	471		// Stop waiting for the transport to reach Dead at 60s: exit with


@@ -461,6 +461,11 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi
461	clearStatus()	461		clearStatus()
462	continue	462		continue
463	}	463		}
		464	+	if ev.State == ports.LinkStateDegraded {
		465	+	log.Warn("UDP link degraded")
		466	+	clearStatus()
		467	+	continue
		468	+	}
464	drawStatusStage(stageForLinkState(ev.State))	469		drawStatusStage(stageForLinkState(ev.State))
465	if ev.State == ports.LinkStateOffline {	470		if ev.State == ports.LinkStateOffline {
466	// Stop waiting for the transport to reach Dead at 60s: exit with	471		// Stop waiting for the transport to reach Dead at 60s: exit with

M internal/usecase/client/reconnect_toast_test.go +15 -3


@@ -179,26 +179,38 @@ func reconnectToastHelloFromSend(t testing.T, tr reconnectToastRecordingTransp
179	179		return hello
180	180		}
181	181
182		-	func TestAttachOnceLinkStateEventsDriveReconnectUIStage(t *testing.T) {
	182	+	func TestReconnectToastDegradedClearsModalAndProbingIsVisible(t *testing.T) {
183	183		term := newReconnectToastTerminalHarness(t)
184	184		defer term.closeInput()
185	185		tr := newReconnectToastLinkTransport()
186	186		tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)}
187	187
188	188		stages := make(chan reconnectStage, 4)
	189	+	clears := make(chan struct{}, 4)
189	190		resultCh := make(chan attachResult, 1)
190	191		ms := milestones{}
191	192		go func() {
192	193		resultCh <- attachOnce(context.Background(), tr, term.term, portsmocks.NewMockClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, func() error {
193	194		_, err := term.term.EnterRaw()
194	195		return err
195		-	}, func() {}, func(stage reconnectStage) {
	196	+	}, func() { clears <- struct{}{} }, func(stage reconnectStage) {
196	197		stages <- stage
197	198		}, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler))
198	199		}()
199	200
	201	+	require.Eventually(t, func() bool { return len(clears) == 1 }, time.Second, time.Millisecond)
	202	+	<-clears // Welcome clears any stale status from an earlier attempt.
200	203		tr.events <- ports.LinkEvent{State: ports.LinkStateDegraded}
201		-	require.Equal(t, reconnectStageDegraded, <-stages)
	204	+	select {
	205	+	case <-clears:
	206	+	case <-time.After(time.Second):
	207	+	t.Fatal("degraded link did not clear stale reconnect modal")
	208	+	}
	209	+	select {
	210	+	case stage := <-stages:
	211	+	t.Fatalf("degraded link drew modal stage %v", stage)
	212	+	case <-time.After(25 * time.Millisecond):
	213	+	}
202	214		tr.events <- ports.LinkEvent{State: ports.LinkStateProbing}
203	215		require.Equal(t, reconnectStageProbingUDP, <-stages)
204	216		tr.recvCh <- reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)}

 	return hello
 }
-func TestAttachOnceLinkStateEventsDriveReconnectUIStage(t *testing.T) {
+func TestReconnectToastDegradedClearsModalAndProbingIsVisible(t *testing.T) {
 	term := newReconnectToastTerminalHarness(t)
 	defer term.closeInput()
 	tr := newReconnectToastLinkTransport()
 	tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)}
 	stages := make(chan reconnectStage, 4)
+	clears := make(chan struct{}, 4)
 	resultCh := make(chan attachResult, 1)
 	ms := milestones{}
 	go func() {
 		resultCh <- attachOnce(context.Background(), tr, term.term, portsmocks.NewMockClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, func() error {
 			_, err := term.term.EnterRaw()
 			return err
-		}, func() {}, func(stage reconnectStage) {
+		}, func() { clears <- struct{}{} }, func(stage reconnectStage) {
 			stages <- stage
 		}, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler))
 	}()
+	require.Eventually(t, func() bool { return len(clears) == 1 }, time.Second, time.Millisecond)
+	<-clears // Welcome clears any stale status from an earlier attempt.
 	tr.events <- ports.LinkEvent{State: ports.LinkStateDegraded}
-	require.Equal(t, reconnectStageDegraded, <-stages)
+	select {
+	case <-clears:
+	case <-time.After(time.Second):
+		t.Fatal("degraded link did not clear stale reconnect modal")
+	}
+	select {
+	case stage := <-stages:
+		t.Fatalf("degraded link drew modal stage %v", stage)
+	case <-time.After(25 * time.Millisecond):
+	}
 	tr.events <- ports.LinkEvent{State: ports.LinkStateProbing}
 	require.Equal(t, reconnectStageProbingUDP, <-stages)
 	tr.recvCh <- reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)}