Add hedge-trigger, final-outcome, attempted-payload and excluded-leader metrics

AGENTS.md

@@ -15,7 +15,9 @@ read it when planning changes to the areas it covers, and keep it current:
 
 - [`docs/internal/metrics.md`](docs/internal/metrics.md) — how metrics are split
   (transport vs producer-state vs warpstream-specific) and the franz-go/`kprom`
-  drop-in-compatibility contract. Update it when changing metrics.
+  drop-in-compatibility contract. Update it when that categorization or parity
+  contract changes. Do not catalog individual custom metrics here; their
+  Prometheus help strings are the source of truth.
 - [`docs/internal/tracing.md`](docs/internal/tracing.md) — how the client drives
   franz-go's produce-record hooks on its own produce path to support tracing (e.g.
   `kotel`) as a drop-in. Update it when changing hook or tracing behaviour.

pkg/wgo/agentpool.go

@@ -3,6 +3,7 @@ package wgo
 import (
 	"context"
 	"fmt"
+	"slices"
 	"sort"
 	"sync/atomic"
 	"time"
@@ -92,6 +93,8 @@ func (p *AgentPool) refresh(ctx context.Context) (removed []int32, dropped leade
 		newAgents = append(newAgents, b.NodeID)
 	}
 	sort.Slice(newAgents, func(i, j int) bool { return newAgents[i] < newAgents[j] })
+	// A NodeID listed twice is still one agent.
+	newAgents = slices.Compact(newAgents)
 
 	agentSet := make(map[int32]struct{}, len(newAgents))
 	for _, id := range newAgents {
@@ -224,3 +227,25 @@ func diffRemovedAgents(old []int32, newSet map[int32]struct{}) []int32 {
 	}
 	return removed
 }
+
+// diffAgentMembership counts NodeIDs that appeared or disappeared between two
+// sorted, unique agent lists.
+func diffAgentMembership(old, new []int32) (added, removed int) {
+	i, j := 0, 0
+	for i < len(old) && j < len(new) {
+		switch {
+		case old[i] == new[j]:
+			i++
+			j++
+		case old[i] < new[j]:
+			removed++
+			i++
+		default:
+			added++
+			j++
+		}
+	}
+	removed += len(old) - i
+	added += len(new) - j
+	return added, removed
+}

pkg/wgo/agentpool_test.go

@@ -366,6 +366,41 @@ func TestBuildLeadersAndTopicIDs(t *testing.T) {
 
 func stringPtr(s string) *string { return &s }
 
+func TestAgentPool_DiffAgentMembership(t *testing.T) {
+	tests := map[string]struct {
+		old         []int32
+		new         []int32
+		wantAdded   int
+		wantRemoved int
+	}{
+		"unchanged": {
+			old: []int32{1, 2, 3}, new: []int32{1, 2, 3},
+		},
+		"one added": {
+			old: []int32{1, 2}, new: []int32{1, 2, 3}, wantAdded: 1,
+		},
+		"one removed": {
+			old: []int32{1, 2, 3}, new: []int32{1, 3}, wantRemoved: 1,
+		},
+		"replace one": {
+			old: []int32{1, 2, 3}, new: []int32{1, 4}, wantAdded: 1, wantRemoved: 2,
+		},
+		"empty to some": {
+			old: nil, new: []int32{1, 2}, wantAdded: 2,
+		},
+		"all removed": {
+			old: []int32{1, 2}, new: nil, wantRemoved: 2,
+		},
+	}
+	for name, tc := range tests {
+		t.Run(name, func(t *testing.T) {
+			added, removed := diffAgentMembership(tc.old, tc.new)
+			assert.Equal(t, tc.wantAdded, added)
+			assert.Equal(t, tc.wantRemoved, removed)
+		})
+	}
+}
+
 func TestDiffRemovedAgents(t *testing.T) {
 	tests := map[string]struct {
 		old      []int32

pkg/wgo/client.go

@@ -151,7 +151,7 @@ func NewWarpstreamClient(logger kgo.Logger, reg prometheus.Registerer, opts ...O
 		refreshCancel:  refreshCancel,
 		refreshNowCh:   make(chan struct{}, 1),
 	}
-	// Count a dropped leader and do not nudge: the refresh goroutine starts below.
+	// Publish the excluded-leader count and do not nudge: the refresh goroutine starts below.
 	// The periodic tick fetches the next snapshot.
 	c.noteLeaderDrops(dropped, false)
 	// Demoter sits on top of the lazy pool strategy so refresh-driven
@@ -503,12 +503,14 @@ func (c *WarpstreamClient) refreshPool(trigger metadataRefreshTrigger) {
 	c.noteLeaderDrops(dropped, true)
 }
 
-// noteLeaderDrops counts and logs excluded leaders. nudge asks for another fetch.
+// noteLeaderDrops publishes the excluded-leader count and logs it when it is
+// above zero. nudge asks for another fetch. The gauge is set even at zero so it
+// clears when the exclusion does.
 func (c *WarpstreamClient) noteLeaderDrops(dropped leaderDrops, nudge bool) {
+	c.metrics.agentPoolExcludedLeaders.Set(float64(dropped.Count))
 	if dropped.Count == 0 {
 		return
 	}
-	c.metrics.agentPoolLeaderDroppedTotal.Add(float64(dropped.Count))
 	log(c.logger, kgo.LogLevelWarn, "warpstream agentpool: leaders excluded from map",
 		"count", dropped.Count,
 		"first_topic", dropped.Topic,