test(events,metrics): assert exactly-once, cover the new counter, drop an overclaim

Codex round 7.

The confirmation bound's comment said it bounds establishment. It does not: the
dial and go-redis's HELLO/AUTH handshake run inside client.Subscribe before the
timer starts and are bounded by the CLIENT's DialTimeout instead, so the worst
case composes to roughly DialTimeout plus this. Anyone reasoning about connect
latency needs both numbers.

The concurrency test asserted topology — subscriber count and pendingSubs —
while reading one event per channel, so a duplicate from a second establishment
could sit in the channel undetected. It now asserts exactly-once, which is the
behaviour the topology was standing in for.

The new Observer method, counter and deployment contract had no adapter test.
Added, including the half that matters: the count must NOT also land on the
reset series, since an adapter that merged them would pass a total-only
assertion while destroying the distinction an operator acts on.

And the abandonment test now says out loud that it fabricates its state, because
the establishing caller is blocked inside Subscribe and nobody can unsubscribe
it — the same reason the retry it exercises is defence in depth rather than a
reachable path.

Claude-Session: https://claude.ai/code/session_01JVDBKbgn3Xt7ndW1YoYd8X

internal/events/redis_bus.go

@@ -1836,6 +1836,15 @@ const stragglerWindow = 30 * time.Second
 // defaultSubscribeConfirmTimeout bounds the wait for Redis to acknowledge a new
 // subscription (BUG-2747).
 //
+// IT BOUNDS THE ACKNOWLEDGEMENT, NOT THE WHOLE OF ESTABLISHMENT (codex round
+// 7). The dial and go-redis's HELLO/AUTH handshake happen inside
+// client.Subscribe, before this timer starts, and they are bounded by the
+// CLIENT's DialTimeout — 5s by default, and not by any context we pass. So a
+// stalled dial followed by this wait composes to roughly DialTimeout + this,
+// and anyone reasoning about worst-case connect latency needs both numbers.
+// BUG-2749 carries the same arithmetic for the admission slot that is held
+// across it.
+//
 // SHORT BECAUSE THE DISTRIBUTION HAS NO MIDDLE, not as a guess at how fast
 // Redis is. Establishment either completes in single-digit milliseconds or does
 // not complete at all, so past the top of the fast mode, waiting longer buys

internal/events/redis_subscribe_confirm_test.go

@@ -195,6 +195,16 @@ func TestConcurrentFirstSubscribersShareOneRedisSubscription(t *testing.T) {
 		case <-time.After(3 * time.Second):
 			t.Fatalf("caller %d did not receive the event", i)
 		}
+		// EXACTLY ONCE, not merely at least once (codex round 7). Two
+		// establishments mean two receive loops fanning the same event out
+		// twice; reading one and moving on would leave the duplicate sitting
+		// in the channel undetected, and the subscriber-count assertion above
+		// is topology rather than the behaviour anyone cares about.
+		select {
+		case dup := <-ch:
+			t.Fatalf("caller %d received a second copy of the event (%d): the workspace has more than one subscription", i, dup.ID)
+		case <-time.After(250 * time.Millisecond):
+		}
 		b.Unsubscribe(ch)
 	}
 }
@@ -642,6 +652,16 @@ func TestAConfirmedSubscriptionIsNeverLeftMarkedUnconfirmed(t *testing.T) {
 // nobody will keep, and returns holding a channel wired to nothing — and
 // PERMANENTLY, because its own registration makes wsCounts non-zero so no
 // later caller establishes either. A dead stream that looks alive.
+//
+// IT FABRICATES THE STATE ON PURPOSE, and that is worth stating rather than
+// hiding (codex round 7). It empties the maps directly instead of calling
+// Unsubscribe because the establishing caller is blocked inside Subscribe and
+// its channel does not exist yet — nobody can unsubscribe it. That is the same
+// reason the retry it exercises is DEFENCE IN DEPTH: with the record retired in
+// the same critical section as the decision, a joiner cannot reach this state
+// through the public surface at all. So this pins the recovery path, not a
+// reachable defect, and it is the honest way to have any coverage of a path
+// whose unreachability is an argument rather than a measurement.
 func TestAJoinerIsNotStrandedByAnAbandonedEstablishment(t *testing.T) {
 	mr := miniredis.RunT(t)
 	client := redis.NewClient(&redis.Options{Addr: mr.Addr()})

internal/metrics/events_observer_test.go

@@ -46,6 +46,14 @@ func TestEventsObserverMapsEachEventToItsOwnCounter(t *testing.T) {
 	obs.ReceiveLoopExited()
 	obs.ReceiveLoopExited()
 
+	// Counted on its OWN counter, not folded into the reset series. The two
+	// answer different questions — a reset says coverage ended, this says a
+	// stream was admitted whose coverage is undescribable — and an adapter that
+	// merged them would satisfy a total-only assertion while destroying the
+	// distinction an operator acts on (BUG-2747).
+	obs.SubscriptionUnconfirmed()
+	obs.SubscriptionUnconfirmed()
+
 	assertCounter(t, m, "pad_event_resume_gaps_total", nil, 2)
 	// The reason must land on a LABELLED series, not on the bare counter: an
 	// adapter that dropped the label would satisfy a total-only assertion and
@@ -58,6 +66,11 @@ func TestEventsObserverMapsEachEventToItsOwnCounter(t *testing.T) {
 		map[string]string{"reason": events.ResetReasonCounterBackward}, 2)
 	assertCounter(t, m, "pad_event_sequence_resets_total",
 		map[string]string{"reason": events.ResetReasonEpochRegressed}, 1)
+	assertCounter(t, m, "pad_event_subscription_unconfirmed_total", nil, 2)
+	// ...and it did NOT leak into the reset series, which is the half a
+	// merged-counter adapter would still pass without.
+	assertCounter(t, m, "pad_event_sequence_resets_total",
+		map[string]string{"reason": events.ResetReasonSubscriptionUnconfirmed}, 0)
 	assertCounter(t, m, "pad_event_sequence_resets_total",
 		map[string]string{"reason": events.ResetReasonUndecodableMessage}, 5)
 	assertCounter(t, m, "pad_event_receive_loop_exits_total", nil, 5)