diff --git a/adapter/autoselect_history.go b/adapter/autoselect_history.go
index 9daef11..e60fef3 100644
--- a/adapter/autoselect_history.go
+++ b/adapter/autoselect_history.go
@@ -1,10 +1,63 @@
 package adapter
 
 import (
+	"encoding/json"
 	"sync"
 	"time"
 )
 
+// UserFailureKind distinguishes how a member's user traffic failed.
+type UserFailureKind string
+
+const (
+	// UserFailureDial is a failure to establish the connection to the member.
+	UserFailureDial UserFailureKind = "dial"
+	// UserFailureStall is a data-plane stall on an already-dialed connection:
+	// the tunnel carried traffic, then went silent for the idle window.
+	UserFailureStall UserFailureKind = "stall"
+	// UserFailureReset is a non-benign transport error (e.g. RST) on an
+	// already-established connection — the tunnel broke mid-stream rather
+	// than going quiet.
+	UserFailureReset   UserFailureKind = "reset"
+	UserFailureUnknown UserFailureKind = "unknown"
+)
+
+// UserFailure stores when a user-traffic failure happened and what kind it was.
+type UserFailure struct {
+	At   time.Time       `json:"at"`
+	Kind UserFailureKind `json:"kind,omitempty"`
+}
+
+func normalizeUserFailureKind(kind UserFailureKind) UserFailureKind {
+	switch kind {
+	case UserFailureDial, UserFailureStall, UserFailureReset:
+		return kind
+	default:
+		return UserFailureUnknown
+	}
+}
+
+// UnmarshalJSON accepts both the current object form and the legacy bare
+// timestamp form. Legacy entries are tagged as unknown.
+func (u *UserFailure) UnmarshalJSON(data []byte) error {
+	var at time.Time
+	if err := json.Unmarshal(data, &at); err == nil {
+		u.At = at
+		u.Kind = UserFailureUnknown
+		return nil
+	}
+
+	type rawUserFailure UserFailure
+	var raw rawUserFailure
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return err
+	}
+
+	*u = UserFailure(raw)
+	u.Kind = normalizeUserFailureKind(u.Kind)
+	return nil
+}
+
 // TagHistory is a point-in-time snapshot of one member's selection
 // state, mirroring the in-memory history kept by the MutableAutoSelect
 // group. Probe outcomes feed the scalar fields; real user-traffic
@@ -13,22 +66,19 @@ import (
 // lets the probe URL through while dropping user traffic shows up as
 // elevated UserFailures with healthy probe scalars simultaneously.
 type TagHistory struct {
-	// LastSuccessDelayMs is the most recent successful probe RTT
-	// (clamped to ≥1 ms). 0 means "no successful probe yet" — rank
-	// uses it as a sentinel to deprioritize against measured peers.
+	// LastSuccessDelayMs is the most recent successful probe RTT.
+	// 0 means "no successful probe yet".
 	LastSuccessDelayMs uint32 `json:"last_success_delay_ms,omitempty"`
-	// LastOutcomeAt is the timestamp of the most recent probe outcome
-	// (success or failure). Used by the probe-cycle freshSince gate to
-	// distinguish "probed this cycle" from "stale outcome."
+	// LastOutcomeAt is the timestamp of the most recent probe outcome.
 	LastOutcomeAt time.Time `json:"last_outcome_at,omitempty"`
 	// ConsecutiveFailures counts probe failures since the last probe
 	// success. Resets to zero on probe success.
 	ConsecutiveFailures uint32 `json:"consecutive_failures,omitempty"`
-	// UserFailures is the sliding window of user-traffic failure
-	// timestamps (dial errors and data-plane stalls). Probe successes
-	// never enter this window; entries age out naturally on the next
-	// mutation older than the group's userFailureWindow.
-	UserFailures []time.Time `json:"user_failures,omitempty"`
+	// UserFailures is the sliding window of user-traffic failures, each
+	// tagged with its kind. Probe successes never enter this window; entries
+	// age out naturally on the next mutation older than the group's
+	// userFailureWindow.
+	UserFailures []UserFailure `json:"user_failures,omitempty"`
 	// HardDemoted reports whether the member has reached the hard-demote
 	// tier from its own recorded probe and user-traffic failures.
 	HardDemoted bool      `json:"hard_demoted,omitempty"`
@@ -164,7 +214,7 @@ func cloneTagHistory(h *TagHistory) *TagHistory {
 	}
 	out := *h
 	if len(h.UserFailures) > 0 {
-		out.UserFailures = make([]time.Time, len(h.UserFailures))
+		out.UserFailures = make([]UserFailure, len(h.UserFailures))
 		copy(out.UserFailures, h.UserFailures)
 	}
 	return &out
diff --git a/adapter/autoselect_history_test.go b/adapter/autoselect_history_test.go
new file mode 100644
index 0000000..17a9164
--- /dev/null
+++ b/adapter/autoselect_history_test.go
@@ -0,0 +1,47 @@
+package adapter
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUserFailure_UnmarshalJSON_ObjectForm(t *testing.T) {
+	at := time.Date(2026, 7, 6, 12, 0, 0, 0, time.UTC)
+	data, err := json.Marshal(UserFailure{At: at, Kind: UserFailureStall})
+	require.NoError(t, err)
+
+	var got UserFailure
+	require.NoError(t, json.Unmarshal(data, &got))
+	assert.True(t, got.At.Equal(at))
+	assert.Equal(t, UserFailureStall, got.Kind)
+}
+
+func TestUserFailure_UnmarshalJSON_LegacyTimestamp(t *testing.T) {
+	// Snapshots persisted before the kind was tracked stored each failure
+	// as a bare RFC3339 timestamp string. Those must still load, tagged as
+	// unknown, so an in-place upgrade doesn't fail to read history.
+	at := time.Date(2026, 7, 6, 12, 0, 0, 0, time.UTC)
+	legacy, err := json.Marshal(at)
+	require.NoError(t, err)
+
+	var got UserFailure
+	require.NoError(t, json.Unmarshal(legacy, &got))
+	assert.True(t, got.At.Equal(at))
+	assert.Equal(t, UserFailureUnknown, got.Kind)
+}
+
+func TestTagHistory_UnmarshalJSON_LegacyUserFailures(t *testing.T) {
+	// A whole TagHistory persisted with the legacy []time.Time window must
+	// decode without error.
+	legacy := `{"user_failures":["2026-07-06T12:00:00Z","2026-07-06T12:01:00Z"]}`
+	var got TagHistory
+	require.NoError(t, json.Unmarshal([]byte(legacy), &got))
+	require.Len(t, got.UserFailures, 2)
+	for _, f := range got.UserFailures {
+		assert.Equal(t, UserFailureUnknown, f.Kind)
+	}
+}
diff --git a/protocol/group/mutableautoselect.go b/protocol/group/mutableautoselect.go
index 7a9bc2c..79cca7a 100644
--- a/protocol/group/mutableautoselect.go
+++ b/protocol/group/mutableautoselect.go
@@ -469,7 +469,7 @@ func (s *MutableAutoSelect) DialContext(ctx context.Context, network string, des
 	// runLadder's gating both see it. For nested groups, the inner tag
 	// isn't a member of this group, so recordUserFailure would drop on
 	// the member gate.
-	s.recordUserFailure(outerTag)
+	s.recordUserFailure(outerTag, adapter.UserFailureDial)
 
 	// Fast failover: try one alternate from the current rank before
 	// kicking the ladder. The active 3-min cadence keeps the probe
@@ -483,7 +483,7 @@ func (s *MutableAutoSelect) DialContext(ctx context.Context, network string, des
 			return s.wrapStream(conn, alt), nil
 		}
 		s.logger.ErrorContext(ctx, err)
-		s.recordUserFailure(altTag)
+		s.recordUserFailure(altTag, adapter.UserFailureDial)
 		outerTag = altTag
 	}
 
@@ -512,7 +512,7 @@ func (s *MutableAutoSelect) ListenPacket(ctx context.Context, destination M.Sock
 		return s.wrapPacket(conn, o), nil
 	}
 	s.logger.ErrorContext(ctx, err)
-	s.recordUserFailure(outerTag)
+	s.recordUserFailure(outerTag, adapter.UserFailureDial)
 
 	alt, altErr := s.selectForExcluding("udp", outerTag)
 	if altErr == nil {
@@ -523,7 +523,7 @@ func (s *MutableAutoSelect) ListenPacket(ctx context.Context, destination M.Sock
 			return s.wrapPacket(conn, alt), nil
 		}
 		s.logger.ErrorContext(ctx, err)
-		s.recordUserFailure(altTag)
+		s.recordUserFailure(altTag, adapter.UserFailureDial)
 		outerTag = altTag
 	}
 
@@ -957,17 +957,17 @@ func (s *MutableAutoSelect) mutateHistory(tag string, fn func(*localHistory, tim
 	return true
 }
 
-// recordUserFailure appends a single user-traffic failure timestamp to
-// the member's sliding window and persists the snapshot. Dial errors
-// and confirmed data-plane stalls both pass through here; each counts
+// recordUserFailure appends a single user-traffic failure of the given
+// kind to the member's sliding window and persists the snapshot. Dial
+// errors and confirmed data-plane stalls both pass through here; each counts
 // as one failure, with softFailLimit tripping the soft tier and
 // consecutiveFailLimit tripping the hard tier. Returns true if the
 // failure was persisted; false on dedup or non-member tag, so callers
 // can suppress downstream side effects (e.g. ladder kicks) on collapsed
 // events.
-func (s *MutableAutoSelect) recordUserFailure(tag string) bool {
+func (s *MutableAutoSelect) recordUserFailure(tag string, kind adapter.UserFailureKind) bool {
 	return s.mutateHistory(tag, func(h *localHistory, now time.Time) bool {
-		return h.addUserFailure(now, s.hist.userFailureWindow, s.hist.userFailureDedupeWindow)
+		return h.addUserFailure(adapter.UserFailure{At: now, Kind: kind}, s.hist.userFailureWindow, s.hist.userFailureDedupeWindow)
 	})
 }
 
diff --git a/protocol/group/mutableautoselect_dataplane.go b/protocol/group/mutableautoselect_dataplane.go
index cb18b81..048fe01 100644
--- a/protocol/group/mutableautoselect_dataplane.go
+++ b/protocol/group/mutableautoselect_dataplane.go
@@ -8,23 +8,25 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/getlantern/lantern-box/adapter"
 )
 
-// makeHooks returns the (onStall, onActivity) callback pair wired into
+// makeHooks returns the (onFailure, onActivity) callback pair wired into
 // the data-plane wrappers. Callbacks re-look-up the member's history
 // at fire time so a Remove+Add cycle can't strand writes on a stale
-// entry. The stall callback short-circuits when recordUserFailure
+// entry. The failure callback short-circuits when recordUserFailure
 // reports a dedup/non-member, so a single broken outbound with N
 // orphan idle conns doesn't trigger N redundant full-fleet probe
 // sweeps.
-func (s *MutableAutoSelect) makeHooks(outerTag string) (onStall, onActivity func()) {
-	onStall = func() {
-		if !s.recordUserFailure(outerTag) {
+func (s *MutableAutoSelect) makeHooks(outerTag string) (onFailure func(adapter.UserFailureKind), onActivity func()) {
+	onFailure = func(kind adapter.UserFailureKind) {
+		if !s.recordUserFailure(outerTag, kind) {
 			return
 		}
 		go s.runLadder(outerTag)
 	}
-	return onStall, s.bumpActive
+	return onFailure, s.bumpActive
 }
 
 // dataPlaneWatchdog is the no-traffic stall timer shared by the stream
@@ -45,7 +47,7 @@ func (s *MutableAutoSelect) makeHooks(outerTag string) (onStall, onActivity func
 // identical to a broken tunnel without this gate.
 type dataPlaneWatchdog struct {
 	idle            time.Duration
-	onStall         func()
+	onFailure       func(adapter.UserFailureKind)
 	onActivity      func()
 	provedReadBytes uint64
 	readBytes       atomic.Uint64
@@ -57,10 +59,10 @@ type dataPlaneWatchdog struct {
 	closeOnce       sync.Once
 }
 
-func (w *dataPlaneWatchdog) init(idle time.Duration, provedReadBytes uint64, onStall, onActivity func()) {
+func (w *dataPlaneWatchdog) init(idle time.Duration, provedReadBytes uint64, onFailure func(adapter.UserFailureKind), onActivity func()) {
 	w.idle = idle
 	w.provedReadBytes = provedReadBytes
-	w.onStall = onStall
+	w.onFailure = onFailure
 	w.onActivity = onActivity
 	w.timer = time.AfterFunc(idle, w.fireStall)
 }
@@ -87,7 +89,7 @@ func (w *dataPlaneWatchdog) noteIO(n int, err error, isRead bool) {
 	}
 	// Attribute mid-stream transport failures immediately.
 	if isDataPlaneFailure(err) {
-		w.fireFailure()
+		w.fireResetFailure()
 		return
 	}
 	// Ignore empty I/O and benign terminal conditions.
@@ -122,7 +124,7 @@ func (w *dataPlaneWatchdog) noteIO(n int, err error, isRead bool) {
 
 // closeWatchdog sets stalled=true before Stop so any concurrent noteIO
 // short-circuits and any concurrent fireStall CAS-fails — late I/O can't
-// deliver a phantom onStall after Close.
+// deliver a phantom onFailure after Close.
 func (w *dataPlaneWatchdog) closeWatchdog() (firstClose bool) {
 	w.closeOnce.Do(func() {
 		w.stalled.Store(true)
@@ -148,25 +150,25 @@ func (w *dataPlaneWatchdog) fireStall() {
 	if !w.fired.CompareAndSwap(false, true) {
 		return
 	}
-	if w.onStall != nil {
-		w.onStall()
+	if w.onFailure != nil {
+		w.onFailure(adapter.UserFailureStall)
 	}
 }
 
-// fireFailure demotes on a mid-stream transport error. Unlike fireStall it skips
+// fireResetFailure demotes on a mid-stream transport error. Unlike fireStall it skips
 // the proven/lastWasWrite gates — an explicit reset is unambiguous even before a
 // conn is proven. Fire and attribution happens at most once.
-func (w *dataPlaneWatchdog) fireFailure() {
+func (w *dataPlaneWatchdog) fireResetFailure() {
 	if !w.stalled.CompareAndSwap(false, true) {
 		return
 	}
 	if !w.fired.CompareAndSwap(false, true) {
 		return
 	}
-	if w.onStall != nil {
-		// run onStall in a goroutine to avoid deadlocks: the callback may call back
+	if w.onFailure != nil {
+		// run onFailure in a goroutine to avoid deadlocks: the callback may call back
 		// into the selector, which may hold locks that the data-plane IO path also needs.
-		go w.onStall()
+		go w.onFailure(adapter.UserFailureReset)
 	}
 }
 
@@ -177,9 +179,9 @@ type dataPlaneStream struct {
 	dataPlaneWatchdog
 }
 
-func newDataPlaneStream(c net.Conn, idle time.Duration, provedReadBytes uint64, onStall, onActivity func()) *dataPlaneStream {
+func newDataPlaneStream(c net.Conn, idle time.Duration, provedReadBytes uint64, onFailure func(adapter.UserFailureKind), onActivity func()) *dataPlaneStream {
 	d := &dataPlaneStream{Conn: c}
-	d.init(idle, provedReadBytes, onStall, onActivity)
+	d.init(idle, provedReadBytes, onFailure, onActivity)
 	return d
 }
 
@@ -215,9 +217,15 @@ type dataPlanePacket struct {
 	dataPlaneWatchdog
 }
 
-func newDataPlanePacket(c net.PacketConn, idle time.Duration, provedReadBytes uint64, onStall, onActivity func()) *dataPlanePacket {
+func newDataPlanePacket(
+	c net.PacketConn,
+	idle time.Duration,
+	provedReadBytes uint64,
+	onFailure func(adapter.UserFailureKind),
+	onActivity func(),
+) *dataPlanePacket {
 	d := &dataPlanePacket{PacketConn: c}
-	d.init(idle, provedReadBytes, onStall, onActivity)
+	d.init(idle, provedReadBytes, onFailure, onActivity)
 	return d
 }
 
diff --git a/protocol/group/mutableautoselect_history.go b/protocol/group/mutableautoselect_history.go
index b89a9cb..bb4a27b 100644
--- a/protocol/group/mutableautoselect_history.go
+++ b/protocol/group/mutableautoselect_history.go
@@ -1,6 +1,7 @@
 package group
 
 import (
+	"slices"
 	"sync"
 	"time"
 
@@ -58,7 +59,7 @@ type localHistory struct {
 	lastSuccessDelayMs  uint32
 	lastOutcomeAt       time.Time
 	consecutiveFailures uint32
-	userFailures        []time.Time
+	userFailures        []adapter.UserFailure
 }
 
 func newLocalHistory() *localHistory { return &localHistory{} }
@@ -84,26 +85,27 @@ func (h *localHistory) recordProbeFailure(now time.Time) {
 	h.lastOutcomeAt = now
 }
 
-// addUserFailure appends a user-traffic failure timestamp and prunes
-// entries older than window in place. A data-plane stall and a dial
-// error each count as one failure; hard demotion requires three in
-// the window.
+// addUserFailure appends the failure and prunes entries older than window
+// in place. A data-plane stall and a dial error each count as one failure;
+// hard demotion requires three in the window.
 //
 // dedupe collapses bursts to a single failure: if the most recent
 // entry is newer than dedupe, the append is dropped. A single broken
 // outbound with many idle conns hitting their stall timer in sequence
 // would otherwise inflate the count out of proportion to the event.
+// The dedupe window spans kinds so a dial error immediately followed by
+// a stall on the retry counts once, matching the one-event intent.
 // Returns true if the failure was recorded.
-func (h *localHistory) addUserFailure(now time.Time, window, dedupe time.Duration) bool {
+func (h *localHistory) addUserFailure(failure adapter.UserFailure, window, dedupe time.Duration) bool {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	if dedupe > 0 && len(h.userFailures) > 0 {
 		last := h.userFailures[len(h.userFailures)-1]
-		if now.Sub(last) < dedupe {
+		if failure.At.Sub(last.At) < dedupe {
 			return false
 		}
 	}
-	h.userFailures = pruneUserFailures(append(h.userFailures, now), now, window)
+	h.userFailures = pruneUserFailures(append(h.userFailures, failure), failure.At, window)
 	return true
 }
 
@@ -117,7 +119,7 @@ func (h *localHistory) userFailureCount(now time.Time, window time.Duration) uin
 	return uint32(len(h.userFailures))
 }
 
-func (h *localHistory) snapshot(now time.Time, window time.Duration) (lastDelay uint32, lastAt time.Time, consec uint32, userFails []time.Time) {
+func (h *localHistory) snapshot(now time.Time, window time.Duration) (lastDelay uint32, lastAt time.Time, consec uint32, userFails []adapter.UserFailure) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 	h.userFailures = pruneUserFailures(h.userFailures, now, window)
@@ -125,7 +127,7 @@ func (h *localHistory) snapshot(now time.Time, window time.Duration) (lastDelay
 	lastAt = h.lastOutcomeAt
 	consec = h.consecutiveFailures
 	if len(h.userFailures) > 0 {
-		userFails = make([]time.Time, len(h.userFailures))
+		userFails = make([]adapter.UserFailure, len(h.userFailures))
 		copy(userFails, h.userFailures)
 	}
 	return
@@ -150,7 +152,7 @@ func (h *localHistory) toTagHistory(updatedAt time.Time, p historyParams) *adapt
 		UpdatedAt:           updatedAt,
 	}
 	if len(h.userFailures) > 0 {
-		out.UserFailures = make([]time.Time, len(h.userFailures))
+		out.UserFailures = make([]adapter.UserFailure, len(h.userFailures))
 		copy(out.UserFailures, h.userFailures)
 	}
 	return out
@@ -158,8 +160,8 @@ func (h *localHistory) toTagHistory(updatedAt time.Time, p historyParams) *adapt
 
 // pruneUserFailures returns the subset of failures whose age is within
 // window. Ages are clamped to zero to tolerate clock skew between
-// cached entries and now.
-func pruneUserFailures(failures []time.Time, now time.Time, window time.Duration) []time.Time {
+// cached entries and now. failures must be sorted by At ascending.
+func pruneUserFailures(failures []adapter.UserFailure, now time.Time, window time.Duration) []adapter.UserFailure {
 	if len(failures) == 0 {
 		return nil
 	}
@@ -167,28 +169,30 @@ func pruneUserFailures(failures []time.Time, now time.Time, window time.Duration
 		// window<=0 disables aging — keep what we have.
 		return failures
 	}
-	out := failures[:0]
-	for _, t := range failures {
-		if ageWithin(now, t, window) {
-			out = append(out, t)
-		}
-	}
-	if len(out) == 0 {
+	firstLive := slices.IndexFunc(failures, func(f adapter.UserFailure) bool {
+		return ageWithin(now, f.At, window)
+	})
+	switch firstLive {
+	case 0:
+		return failures
+	case -1:
 		return nil
+	default:
+		copy(failures, failures[firstLive:])
+		return failures[:len(failures)-firstLive]
 	}
-	return out
 }
 
 // countUserFailuresInWindow returns the number of entries whose age is
 // within window without allocating. Use on the read-only rank hot path
 // where pruneUserFailures' in-place mutation isn't needed.
-func countUserFailuresInWindow(failures []time.Time, now time.Time, window time.Duration) uint32 {
+func countUserFailuresInWindow(failures []adapter.UserFailure, now time.Time, window time.Duration) uint32 {
 	if window <= 0 {
 		return uint32(len(failures))
 	}
 	var n uint32
-	for _, t := range failures {
-		if ageWithin(now, t, window) {
+	for _, f := range failures {
+		if ageWithin(now, f.At, window) {
 			n++
 		}
 	}
@@ -199,11 +203,7 @@ func countUserFailuresInWindow(failures []time.Time, now time.Time, window time.
 // (cached entries with future timestamps from clock skew) clamp to 0
 // so they still count as "in window."
 func ageWithin(now, t time.Time, window time.Duration) bool {
-	age := now.Sub(t)
-	if age < 0 {
-		age = 0
-	}
-	return age < window
+	return max(0, now.Sub(t)) < window
 }
 
 // hydrateLocalHistory rebuilds a localHistory from a persisted snapshot.
@@ -219,7 +219,11 @@ func hydrateLocalHistory(t *adapter.TagHistory, now time.Time, window time.Durat
 	h.lastOutcomeAt = t.LastOutcomeAt
 	h.consecutiveFailures = t.ConsecutiveFailures
 	if len(t.UserFailures) > 0 {
-		copies := make([]time.Time, len(t.UserFailures))
+		// ensure the slice is sorted by At ascending
+		slices.SortFunc(t.UserFailures, func(a, b adapter.UserFailure) int {
+			return a.At.Compare(b.At)
+		})
+		copies := make([]adapter.UserFailure, len(t.UserFailures))
 		copy(copies, t.UserFailures)
 		h.userFailures = pruneUserFailures(copies, now, window)
 	}
diff --git a/protocol/group/mutableautoselect_test.go b/protocol/group/mutableautoselect_test.go
index 9391c7d..980cece 100644
--- a/protocol/group/mutableautoselect_test.go
+++ b/protocol/group/mutableautoselect_test.go
@@ -84,14 +84,18 @@ func TestPruneUserFailures(t *testing.T) {
 		want   int
 	}{
 		{"empty", nil, time.Minute, 0},
-		{"all in window", []time.Time{now.Add(-1 * time.Minute), now.Add(-2 * time.Minute)}, 5 * time.Minute, 2},
-		{"some aged out", []time.Time{now.Add(-1 * time.Minute), now.Add(-10 * time.Minute)}, 5 * time.Minute, 1},
+		{"all in window", []time.Time{now.Add(-2 * time.Minute), now.Add(-1 * time.Minute)}, 5 * time.Minute, 2},
+		{"some aged out", []time.Time{now.Add(-10 * time.Minute), now.Add(-1 * time.Minute)}, 5 * time.Minute, 1},
 		{"future-stamped clamps age to 0", []time.Time{now.Add(time.Hour)}, 5 * time.Minute, 1},
 		{"all aged out returns nil", []time.Time{now.Add(-10 * time.Minute)}, 5 * time.Minute, 0},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := pruneUserFailures(append([]time.Time(nil), tt.in...), now, tt.window)
+			in := make([]adapter.UserFailure, len(tt.in))
+			for i, at := range tt.in {
+				in[i] = adapter.UserFailure{At: at, Kind: adapter.UserFailureStall}
+			}
+			got := pruneUserFailures(in, now, tt.window)
 			assert.Len(t, got, tt.want)
 		})
 	}
@@ -211,14 +215,14 @@ func TestToTagHistory_HardDemoted(t *testing.T) {
 	// are not hard demoted.
 	soft := newLocalHistory()
 	for i := uint32(0); i < p.softFailLimit; i++ {
-		soft.addUserFailure(now.Add(-time.Duration(i)*time.Second), p.userFailureWindow, 0)
+		soft.addUserFailure(adapter.UserFailure{At: now.Add(-time.Duration(i) * time.Second), Kind: adapter.UserFailureStall}, p.userFailureWindow, 0)
 	}
 	assert.False(t, soft.toTagHistory(now, p).HardDemoted,
 		"user failures below the hard limit are not hard demoted")
 
 	userHard := newLocalHistory()
 	for i := uint32(0); i < p.consecutiveFailLimit; i++ {
-		userHard.addUserFailure(now.Add(-time.Duration(i)*time.Second), p.userFailureWindow, 0)
+		userHard.addUserFailure(adapter.UserFailure{At: now.Add(-time.Duration(i) * time.Second), Kind: adapter.UserFailureStall}, p.userFailureWindow, 0)
 	}
 	assert.True(t, userHard.toTagHistory(now, p).HardDemoted,
 		"user failures at the hard limit are hard demoted")
@@ -231,8 +235,8 @@ func TestLocalHistory_UserFailuresIndependentOfProbeOutcomes(t *testing.T) {
 	// laundered to clean by a passing probe.
 	h := newLocalHistory()
 	now := time.Now()
-	h.addUserFailure(now, 5*time.Minute, 0)
-	h.addUserFailure(now.Add(time.Second), 5*time.Minute, 0)
+	h.addUserFailure(adapter.UserFailure{At: now, Kind: adapter.UserFailureStall}, 5*time.Minute, 0)
+	h.addUserFailure(adapter.UserFailure{At: now.Add(time.Second), Kind: adapter.UserFailureStall}, 5*time.Minute, 0)
 	assert.Equal(t, uint32(2), h.userFailureCount(now.Add(2*time.Second), 5*time.Minute))
 	h.recordProbeSuccess(100, now.Add(3*time.Second))
 	assert.Equal(t, uint32(2), h.userFailureCount(now.Add(4*time.Second), 5*time.Minute),
@@ -245,8 +249,8 @@ func TestLocalHistory_UserFailuresAgeOut(t *testing.T) {
 	// regardless of whether traffic gets routed through it.
 	h := newLocalHistory()
 	now := time.Now()
-	h.addUserFailure(now, 5*time.Minute, 0)
-	h.addUserFailure(now.Add(time.Minute), 5*time.Minute, 0)
+	h.addUserFailure(adapter.UserFailure{At: now, Kind: adapter.UserFailureStall}, 5*time.Minute, 0)
+	h.addUserFailure(adapter.UserFailure{At: now.Add(time.Minute), Kind: adapter.UserFailureStall}, 5*time.Minute, 0)
 	assert.Equal(t, uint32(2), h.userFailureCount(now.Add(time.Minute), 5*time.Minute))
 	// Six minutes later, both are stale.
 	assert.Equal(t, uint32(0), h.userFailureCount(now.Add(6*time.Minute), 5*time.Minute),
@@ -262,12 +266,12 @@ func TestLocalHistory_AddUserFailureDedupesBurst(t *testing.T) {
 	now := time.Now()
 	dedupe := 30 * time.Second
 
-	require.True(t, h.addUserFailure(now, 5*time.Minute, dedupe), "first failure must record")
-	require.False(t, h.addUserFailure(now.Add(time.Second), 5*time.Minute, dedupe), "burst within dedupe must drop")
-	require.False(t, h.addUserFailure(now.Add(29*time.Second), 5*time.Minute, dedupe), "still within dedupe must drop")
+	require.True(t, h.addUserFailure(adapter.UserFailure{At: now, Kind: adapter.UserFailureStall}, 5*time.Minute, dedupe), "first failure must record")
+	require.False(t, h.addUserFailure(adapter.UserFailure{At: now.Add(time.Second), Kind: adapter.UserFailureStall}, 5*time.Minute, dedupe), "burst within dedupe must drop")
+	require.False(t, h.addUserFailure(adapter.UserFailure{At: now.Add(29 * time.Second), Kind: adapter.UserFailureStall}, 5*time.Minute, dedupe), "still within dedupe must drop")
 	assert.Equal(t, uint32(1), h.userFailureCount(now.Add(30*time.Second), 5*time.Minute), "burst collapsed to one")
 
-	require.True(t, h.addUserFailure(now.Add(30*time.Second), 5*time.Minute, dedupe), "at-or-past dedupe must record")
+	require.True(t, h.addUserFailure(adapter.UserFailure{At: now.Add(30 * time.Second), Kind: adapter.UserFailureStall}, 5*time.Minute, dedupe), "at-or-past dedupe must record")
 	assert.Equal(t, uint32(2), h.userFailureCount(now.Add(30*time.Second), 5*time.Minute))
 }
 
@@ -276,9 +280,9 @@ func TestHydrateLocalHistory_DropsAgedUserFailures(t *testing.T) {
 	persisted := &adapter.TagHistory{
 		LastSuccessDelayMs: 120,
 		LastOutcomeAt:      now.Add(-time.Minute),
-		UserFailures: []time.Time{
-			now.Add(-30 * time.Minute), // outside 5-min window
-			now.Add(-1 * time.Minute),  // inside window
+		UserFailures: []adapter.UserFailure{
+			{At: now.Add(-30 * time.Minute), Kind: adapter.UserFailureDial}, // outside 5-min window
+			{At: now.Add(-1 * time.Minute), Kind: adapter.UserFailureStall}, // inside window
 		},
 		UpdatedAt: now.Add(-time.Minute),
 	}
@@ -370,7 +374,7 @@ func addUserFailureN(s *MutableAutoSelect, tag string, n int) {
 	h := s.historyForLocked(tag)
 	now := time.Now()
 	for i := 0; i < n; i++ {
-		h.addUserFailure(now.Add(time.Duration(i)*time.Millisecond), s.hist.userFailureWindow, 0)
+		h.addUserFailure(adapter.UserFailure{At: now.Add(time.Duration(i) * time.Millisecond), Kind: adapter.UserFailureStall}, s.hist.userFailureWindow, 0)
 	}
 }
 
@@ -559,7 +563,7 @@ func TestDataPlaneStream_StallSuppressedUntilProven(t *testing.T) {
 	var calls atomic.Uint32
 	const provedReadBytes = 100
 	d := newDataPlaneStream(stallConn{}, 10*time.Millisecond, provedReadBytes,
-		func() { calls.Add(1) }, nil)
+		func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 	time.Sleep(50 * time.Millisecond)
 	assert.Equal(t, uint32(0), calls.Load(),
@@ -573,7 +577,7 @@ func TestDataPlaneStream_StallFiresAfterProven(t *testing.T) {
 	var calls atomic.Uint32
 	const provedReadBytes = 50
 	d := newDataPlaneStream(echoConn{}, 30*time.Millisecond, provedReadBytes,
-		func() { calls.Add(1) }, nil)
+		func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	// Drive enough Read bytes to prove the conn.
@@ -603,7 +607,7 @@ func TestDataPlaneStream_StallSuppressedOnReadOnlyIdle(t *testing.T) {
 	var calls atomic.Uint32
 	const provedReadBytes = 50
 	d := newDataPlaneStream(echoConn{}, 30*time.Millisecond, provedReadBytes,
-		func() { calls.Add(1) }, nil)
+		func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	_, err := d.Read(make([]byte, provedReadBytes))
@@ -617,14 +621,14 @@ func TestDataPlaneStream_StallSuppressedOnReadOnlyIdle(t *testing.T) {
 }
 
 func TestDataPlaneStream_CloseIsIdempotent(t *testing.T) {
-	d := newDataPlaneStream(stallConn{}, time.Hour, 0, func() {}, nil)
+	d := newDataPlaneStream(stallConn{}, time.Hour, 0, func(adapter.UserFailureKind) {}, nil)
 	require.NoError(t, d.Close())
 	assert.NoError(t, d.Close(), "second Close should be a no-op")
 }
 
 func TestDataPlaneStream_NoStallAfterClose(t *testing.T) {
 	var calls atomic.Uint32
-	d := newDataPlaneStream(stallConn{}, time.Hour, 0, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(stallConn{}, time.Hour, 0, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	d.proven.Store(true)       // simulate a proven conn so fireStall isn't gated on the proven check
 	d.lastWasWrite.Store(true) // ...nor on the write-without-read gate
 	d.Close()
@@ -634,7 +638,7 @@ func TestDataPlaneStream_NoStallAfterClose(t *testing.T) {
 
 func TestDataPlanePacket_NoStallAfterClose(t *testing.T) {
 	var calls atomic.Uint32
-	d := newDataPlanePacket(stallPacketConn{}, time.Hour, 0, func() { calls.Add(1) }, nil)
+	d := newDataPlanePacket(stallPacketConn{}, time.Hour, 0, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	d.proven.Store(true)
 	d.lastWasWrite.Store(true)
 	d.Close()
@@ -724,8 +728,8 @@ func TestRecordProbeOutcome_DropsOutcomesForRemovedTag(t *testing.T) {
 
 func TestRecordUserFailure_AppendsAndBoundedByMembership(t *testing.T) {
 	s, _ := newTestMUR(t, "a")
-	assert.True(t, s.recordUserFailure("a"), "first call must persist")
-	assert.False(t, s.recordUserFailure("a"), "immediate burst must return false (deduped)")
+	assert.True(t, s.recordUserFailure("a", adapter.UserFailureDial), "first call must persist")
+	assert.False(t, s.recordUserFailure("a", adapter.UserFailureStall), "immediate burst must return false (deduped)")
 	s.access.Lock()
 	h, ok := s.peekHistoryLocked("a")
 	s.access.Unlock()
@@ -733,7 +737,7 @@ func TestRecordUserFailure_AppendsAndBoundedByMembership(t *testing.T) {
 	assert.Equal(t, uint32(1), h.userFailureCount(time.Now(), s.hist.userFailureWindow),
 		"successive calls within dedupe window must collapse")
 
-	assert.False(t, s.recordUserFailure("nonexistent"), "non-member must return false")
+	assert.False(t, s.recordUserFailure("nonexistent", adapter.UserFailureDial), "non-member must return false")
 	s.access.Lock()
 	_, exists := s.peekHistoryLocked("nonexistent")
 	s.access.Unlock()
@@ -746,13 +750,41 @@ func TestMakeHooks_StallAppendsSingleUserFailure(t *testing.T) {
 	// one. (Spec change from earlier "one-shot hard demote.")
 	s, _ := newTestMUR(t, "a")
 	onStall, _ := s.makeHooks("a")
-	onStall()
+	onStall(adapter.UserFailureStall)
 	s.access.Lock()
 	h, ok := s.peekHistoryLocked("a")
 	s.access.Unlock()
 	require.True(t, ok)
 	assert.Equal(t, uint32(1), h.userFailureCount(time.Now(), s.hist.userFailureWindow),
 		"a single stall must contribute exactly one user-failure timestamp")
+	require.Len(t, h.userFailures, 1)
+	assert.Equal(t, adapter.UserFailureStall, h.userFailures[0].Kind,
+		"the stall hook must record the failure as a stall")
+}
+
+func TestMakeHooks_PropagatesFailureKind(t *testing.T) {
+	s, _ := newTestMUR(t, "a")
+	onStall, _ := s.makeHooks("a")
+	onStall(adapter.UserFailureReset)
+	s.access.Lock()
+	h, ok := s.peekHistoryLocked("a")
+	s.access.Unlock()
+	require.True(t, ok)
+	require.Len(t, h.userFailures, 1)
+	assert.Equal(t, adapter.UserFailureReset, h.userFailures[0].Kind,
+		"makeHooks must record the failure under the kind the watchdog reports")
+}
+
+func TestRecordUserFailure_AttributesDialKind(t *testing.T) {
+	s, _ := newTestMUR(t, "a")
+	require.True(t, s.recordUserFailure("a", adapter.UserFailureDial))
+	s.access.Lock()
+	h, ok := s.peekHistoryLocked("a")
+	s.access.Unlock()
+	require.True(t, ok)
+	require.Len(t, h.userFailures, 1)
+	assert.Equal(t, adapter.UserFailureDial, h.userFailures[0].Kind,
+		"a dial-site failure must record the failure as a dial error")
 }
 
 func TestRecordUserFailure_DemotesTagInNextSelectFor(t *testing.T) {
@@ -949,7 +981,7 @@ func TestNextProbeInterval(t *testing.T) {
 
 func TestDataPlaneStream_OnActivityFiresOnNonEmptyIO(t *testing.T) {
 	var activity atomic.Uint32
-	d := newDataPlaneStream(echoConn{}, time.Hour, 0, func() {}, func() { activity.Add(1) })
+	d := newDataPlaneStream(echoConn{}, time.Hour, 0, func(adapter.UserFailureKind) {}, func() { activity.Add(1) })
 	defer d.Close()
 	_, err := d.Write([]byte("hi"))
 	require.NoError(t, err, "Write should succeed on echoConn")
@@ -1008,7 +1040,7 @@ func (timeoutErr) Error() string   { return "i/o timeout" }
 func (timeoutErr) Timeout() bool   { return true }
 func (timeoutErr) Temporary() bool { return false }
 
-// fireFailure dispatches onStall on its own goroutine, so failure assertions
+// fireResetFailure dispatches onFailure on its own goroutine, so failure assertions
 // poll and non-failure assertions let that goroutine settle before checking.
 const attributeSettle = 50 * time.Millisecond
 
@@ -1019,7 +1051,7 @@ func TestDataPlaneStream_MidStreamReset_ConnReset_AttributesFailure(t *testing.T
 	var calls atomic.Uint32
 	const provedReadBytes = 1 << 20 // never reached by the 16-byte read below
 	c := &failingConn{err: connResetErr(), firstN: 16}
-	d := newDataPlaneStream(c, time.Hour, provedReadBytes, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(c, time.Hour, provedReadBytes, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	buf := make([]byte, 32)
@@ -1037,12 +1069,46 @@ func TestDataPlaneStream_MidStreamReset_ConnReset_AttributesFailure(t *testing.T
 		"mid-stream reset on an unproven conn must attribute exactly one failure")
 }
 
+func TestDataPlaneStream_MidStreamReset_AttributesResetKind(t *testing.T) {
+	// fireResetFailure must report the failure as a reset, not a stall, so the two
+	// mid-stream failure modes stay separable downstream.
+	var got atomic.Value // adapter.UserFailureKind
+	c := &failingConn{err: connResetErr(), firstN: 16}
+	d := newDataPlaneStream(c, time.Hour, 1<<20, func(k adapter.UserFailureKind) { got.Store(k) }, nil)
+	defer d.Close()
+
+	buf := make([]byte, 32)
+	_, err := d.Read(buf)
+	require.NoError(t, err)
+	_, err = d.Read(buf)
+	require.Error(t, err)
+
+	require.Eventually(t, func() bool { return got.Load() != nil },
+		time.Second, 5*time.Millisecond, "a mid-stream reset must attribute a failure")
+	assert.Equal(t, adapter.UserFailureReset, got.Load().(adapter.UserFailureKind),
+		"a mid-stream reset must be recorded as a reset")
+}
+
+func TestDataPlaneStream_Stall_AttributesStallKind(t *testing.T) {
+	// The idle-timeout path must report a stall, complementing the reset path.
+	var got atomic.Value // adapter.UserFailureKind
+	d := newDataPlaneStream(stallConn{}, time.Hour, 0, func(k adapter.UserFailureKind) { got.Store(k) }, nil)
+	defer d.Close()
+	d.proven.Store(true)
+	d.lastWasWrite.Store(true)
+	d.fireStall()
+
+	require.NotNil(t, got.Load(), "a stall must attribute a failure")
+	assert.Equal(t, adapter.UserFailureStall, got.Load().(adapter.UserFailureKind),
+		"an idle stall must be recorded as a stall")
+}
+
 func TestDataPlaneStream_MidStreamReset_ErrClosed_AttributesFailure(t *testing.T) {
 	// net.ErrClosed reaching noteIO without a prior local Close means the
 	// inner outbound tore its own fd down — a broken tunnel, so demote.
 	var calls atomic.Uint32
 	c := &failingConn{err: net.ErrClosed, firstN: 8}
-	d := newDataPlaneStream(c, time.Hour, 4, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(c, time.Hour, 4, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	buf := make([]byte, 32)
@@ -1060,7 +1126,7 @@ func TestDataPlaneStream_MidStreamReset_WritePath_AttributesFailure(t *testing.T
 	// The failure path must fire on a reset Write too, not just Read.
 	var calls atomic.Uint32
 	c := &failingConn{err: connResetErr(), failWrite: true}
-	d := newDataPlaneStream(c, time.Hour, 4, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(c, time.Hour, 4, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	_, err := d.Write([]byte("req"))
@@ -1075,7 +1141,7 @@ func TestDataPlanePacket_MidStreamReset_AttributesFailure(t *testing.T) {
 	// the same way as the stream path.
 	var calls atomic.Uint32
 	c := &failingPacketConn{err: connResetErr()}
-	d := newDataPlanePacket(c, time.Hour, 4, func() { calls.Add(1) }, nil)
+	d := newDataPlanePacket(c, time.Hour, 4, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	_, _, err := d.ReadFrom(make([]byte, 32))
@@ -1089,7 +1155,7 @@ func TestDataPlaneStream_EOFDoesNotAttribute(t *testing.T) {
 	// io.EOF is an ordinary stream end, not a tunnel failure.
 	var calls atomic.Uint32
 	c := &failingConn{err: io.EOF, firstN: 8}
-	d := newDataPlaneStream(c, time.Hour, 4, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(c, time.Hour, 4, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	buf := make([]byte, 32)
@@ -1108,7 +1174,7 @@ func TestDataPlaneStream_TimeoutDoesNotAttribute(t *testing.T) {
 	// double-count and usurp that role.
 	var calls atomic.Uint32
 	c := &failingConn{err: timeoutErr{}, firstN: 8}
-	d := newDataPlaneStream(c, time.Hour, 4, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(c, time.Hour, 4, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	buf := make([]byte, 32)
@@ -1128,7 +1194,7 @@ func TestDataPlaneStream_NoAttributeAfterClose(t *testing.T) {
 	// failure.
 	var calls atomic.Uint32
 	c := &failingConn{err: net.ErrClosed}
-	d := newDataPlaneStream(c, time.Hour, 1, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(c, time.Hour, 1, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	require.True(t, d.closeWatchdog(), "first close")
 
 	_, err := d.Read(make([]byte, 8))
@@ -1142,7 +1208,7 @@ func TestDataPlaneStream_FailureFiresOnce(t *testing.T) {
 	// Repeated errored reads attribute at most once per conn (the fired CAS).
 	var calls atomic.Uint32
 	c := &failingConn{err: connResetErr()}
-	d := newDataPlaneStream(c, time.Hour, 1, func() { calls.Add(1) }, nil)
+	d := newDataPlaneStream(c, time.Hour, 1, func(adapter.UserFailureKind) { calls.Add(1) }, nil)
 	defer d.Close()
 
 	for i := range 3 {