diff --git a/internal/server/event_deliver.go b/internal/server/event_deliver.go
deleted file mode 100644
index 1e5dad6..0000000
--- a/internal/server/event_deliver.go
+++ /dev/null
@@ -1,185 +0,0 @@
-package server
-
-import (
-	"cmp"
-	"encoding/json"
-	"log/slog"
-	"time"
-
-	"github.com/mamonth/oasmock/internal/extensions"
-	"github.com/mamonth/oasmock/internal/loader"
-	"github.com/mamonth/oasmock/internal/runtime"
-)
-
-func (b *eventBus) deliver(sub channelSubscription, payload map[string]any) {
-	b.deliverTo(sub, payload, nil)
-}
-
-// deliverTargeted delivers a built-in event to a single candidate connection.
-// The connection bucket (if any) is evaluated against that one recipient only;
-// with no connection conditions the message is pushed to the recipient alone
-// (RS.EVT.24, RS.EXT.26).
-func (b *eventBus) deliverTargeted(sub channelSubscription, payload map[string]any, recipient ConsumerInfo) {
-	b.deliverTo(sub, payload, &recipient)
-}
-
-// deliverTo runs the shared delayed-emission + delivery pipeline for a
-// subscription. When target is non-nil, delivery is restricted to that single
-// candidate (built-in connect recipient).
-func (b *eventBus) deliverTo(sub channelSubscription, payload map[string]any, target *ConsumerInfo) {
-	if len(sub.messages) == 0 {
-		return
-	}
-	if sub.delay > 0 {
-		ms := sub.delay
-		sub.delay = 0
-		go func() {
-			select {
-			case <-b.done:
-				return
-			case <-time.After(time.Duration(ms) * time.Millisecond):
-			}
-			b.deliverTo(sub, payload, target)
-		}()
-		return
-	}
-	deliverable := sub.messages[0]
-	addr := sub.address
-	prefix := deliverable.prefix
-	eventName := sub.event
-	opID := "event:" + cmp.Or(eventName, anyEventIdentity) + ":" + addr
-
-	b.deliverExample(sub, deliverable.spec.Examples, addr, prefix, eventName, payload, opID, target)
-}
-
-// stateEnvEvaluator wires the fixed state and environment sources shared by
-// every emission path (periodic deliveries have no event/connection context).
-func (b *eventBus) stateEnvEvaluator(prefix string) runtime.Evaluator {
-	eval := runtime.NewEvaluator()
-	eval.AddSource(runtime.SourceState, b.renderer.NewStateSource(prefix))
-	eval.AddSource(runtime.SourceEnv, b.renderer.NewEnvSource())
-	return eval
-}
-
-// eventEvaluator wires the fixed emission sources (state, env, event) plus an
-// optional per-connection source into a fresh evaluator.
-func (b *eventBus) eventEvaluator(state, env runtime.DataSource, eventName string, payload map[string]any, connection *runtime.ConnectionSource) runtime.Evaluator {
-	eval := runtime.NewEvaluator()
-	eval.AddSource(runtime.SourceState, state)
-	eval.AddSource(runtime.SourceEnv, env)
-	eval.AddSource(runtime.SourceEvent, &runtime.EventSource{Name: eventName, Data: payload})
-	if connection != nil {
-		eval.AddSource(runtime.SourceConnection, connection)
-	}
-	return eval
-}
-
-// renderExample renders a single example's payload against an evaluator,
-// honoring x-mock-skip and x-mock-set-state. It returns the rendered body, or
-// nil when the example is skipped or rendering fails (verbose-logged).
-func (b *eventBus) renderExample(view *MessageExampleView, eval runtime.Evaluator, prefix, opID string) []byte {
-	if extensions.ValueSkip(view) {
-		return nil
-	}
-	if stateMap, ok := extensions.ValueSetState(view); ok {
-		b.renderer.ApplySetState(stateMap, eval, prefix)
-	}
-	body, err := b.renderer.RenderAsyncPayload(view, eval)
-	if err != nil {
-		if b.verbose {
-			slog.Debug("Example delivery render failed", "opID", opID, "err", err)
-		}
-		return nil
-	}
-	return body
-}
-
-// evaluateConnectionBucket evaluates an example's connection conditions
-// against one candidate recipient. An empty bucket matches every candidate.
-func (b *eventBus) evaluateConnectionBucket(bucket extensions.ParamsMatch, state, env runtime.DataSource, eventName string, payload map[string]any, candidate ConsumerInfo) (bool, error) {
-	if len(bucket) == 0 {
-		return true, nil
-	}
-	eval := b.eventEvaluator(state, env, eventName, payload, connectionSourceFromInfo(candidate))
-	return extensions.EvaluateParamsMatch(bucket, eval)
-}
-
-// deliverExample runs the shared selection + render + recipient-partition
-// pipeline for one subscription's examples. When target is non-nil, delivery
-// is restricted to that single candidate (built-in connect recipient).
-func (b *eventBus) deliverExample(sub channelSubscription, examples []*loader.MessageExampleSpec, addr, prefix, eventName string, payload map[string]any, opID string, target *ConsumerInfo) {
-	// Fixed (non-connection) sources evaluated once per emission.
-	state := b.renderer.NewStateSource(prefix)
-	env := b.renderer.NewEnvSource()
-
-	for _, example := range examples {
-		view := &MessageExampleView{spec: example}
-		common, connection := b.partitionedMatch(view)
-		var connSource *runtime.ConnectionSource
-		if target != nil {
-			connSource = connectionSourceFromInfo(*target)
-		}
-		evaluator := b.eventEvaluator(state, env, eventName, payload, connSource)
-		if len(common) > 0 {
-			ok, cErr := extensions.EvaluateParamsMatch(common, evaluator)
-			if cErr != nil || !ok {
-				continue
-			}
-		}
-		body := b.renderExample(view, evaluator, prefix, opID)
-		if body == nil {
-			continue
-		}
-		if target != nil {
-			// Built-in recipient: evaluate the connection bucket against the
-			// single candidate and deliver on match (or immediately when there
-			// is no connection bucket).
-			ok, okErr := b.evaluateConnectionBucket(connection, state, env, eventName, payload, *target)
-			if okErr != nil || !ok {
-				continue
-			}
-			b.notifyPush(addr, target.ConnectionID, body)
-			b.bus.PushTo(*target, addr, body)
-			continue
-		}
-		if len(connection) == 0 {
-			// Broadcast fast path (RS.EXT.25).
-			b.notifyPush(addr, "", body)
-			b.bus.SignalRPush(addr, body)
-			b.bus.WSBroadcast(addr, body)
-			continue
-		}
-		// Per-connection partition: evaluate the connection bucket against each
-		// candidate with its connection context (design D6).
-		for _, candidate := range b.bus.Candidates(addr) {
-			ok, okErr := b.evaluateConnectionBucket(connection, state, env, eventName, payload, candidate)
-			if okErr != nil || !ok {
-				continue
-			}
-			b.notifyPush(addr, candidate.ConnectionID, body)
-			b.bus.PushTo(candidate, addr, body)
-		}
-	}
-}
-
-// notifyPush emits a push envelope to the management observer (RS.AMG.25).
-func (b *eventBus) notifyPush(channel, connectionID string, body []byte) {
-	if b.observer == nil {
-		return
-	}
-	var payload map[string]any
-	if err := json.Unmarshal(body, &payload); err != nil {
-		payload = map[string]any{"raw": string(body)}
-	}
-	env := manageEnvelope{Type: "push"}
-	env.Push = &managePushEnvelope{Channel: channel, ConnectionID: connectionID, Payload: payload}
-	b.observer(env)
-}
-
-// partitionedMatch splits an example's x-mock-match into common conditions
-// (evaluated once per emission) and connection conditions (evaluated per
-// candidate recipient). A nil/absent match yields empty buckets.
-func (b *eventBus) partitionedMatch(view *MessageExampleView) (extensions.ParamsMatch, extensions.ParamsMatch) {
-	match, _ := extensions.ValueMatch(view)
-	return extensions.PartitionConnectionConditions(extensions.ParamsMatch(match))
-}
diff --git a/internal/server/event_server.go b/internal/server/event_server.go
index dd67a75..b7f3faf 100644
--- a/internal/server/event_server.go
+++ b/internal/server/event_server.go
@@ -4,7 +4,6 @@ import (
 	"cmp"
 	"fmt"
 	"log/slog"
-	"sync"
 	"time"
 
 	"github.com/mamonth/oasmock/internal/asyncapi"
@@ -12,58 +11,39 @@ import (
 	"github.com/mamonth/oasmock/internal/loader"
 )
 
-// eventBus is the pure-fabrication coordinator behind the event driver
-// (design D8). It owns the event broker and renders + delivers subscribed
-// messages through the MessageRenderer and ConsumerBus contracts, so it never
-// reaches into Server.
+// eventBus orchestrates the event driver (design D8). It owns the subscription
+// broker and the interval scheduler, and delegates message rendering/delivery
+// to the messageDelivery engine. It never reaches into Server.
 type eventBus struct {
 	broker    *eventBroker
-	renderer  MessageRenderer
-	bus       ConsumerBus
 	scheduler *jobScheduler
+	delivery  *messageDelivery
 	verbose   bool
-	// wait sleeps for a delayed emission (design D4/D5); injectable so tests
-	// stay hermetic.
-	wait func(time.Duration)
-	// observer, when set, is invoked with every emitted envelope so the
-	// management stream can mirror fired events and deliveries (RS.AMG.24-25).
+	// observer, when set, is invoked with event and schedule envelopes so the
+	// management stream can mirror fired events and job lifecycle. Push
+	// envelopes are emitted by the delivery engine (RS.AMG.24-27).
 	observer func(env manageEnvelope)
-	// done is closed on shutdown so pending delayed emissions no longer run.
-	done    chan struct{}
-	stopOne sync.Once
 }
 
-// doneChannel returns the bus shutdown signal so callers that schedule work on
-// their own goroutines (e.g. delayed management pushes) can cancel it when the
-// bus shuts down.
-func (b *eventBus) doneChannel() <-chan struct{} {
-	if b == nil {
-		ch := make(chan struct{})
-		close(ch)
-		return ch
-	}
-	return b.done
-}
-
-// setObserver installs the management-stream observer.
+// setObserver installs the management-stream observer on the bus and the
+// delivery engine (pushes).
 func (b *eventBus) setObserver(observer func(env manageEnvelope)) {
 	b.observer = observer
+	b.delivery.setObserver(observer)
 }
 
-// newEventBus wires a broker whose delivery goes through the renderer and
-// consumer bus.
+// newEventBus wires a broker whose delivery goes through the messageDelivery
+// engine, plus an interval scheduler.
 func newEventBus(renderer MessageRenderer, bus ConsumerBus, verbose bool) *eventBus {
+	delivery := newMessageDelivery(renderer, bus, verbose)
 	b := &eventBus{
-		renderer:  renderer,
-		bus:       bus,
 		scheduler: newJobScheduler(),
+		delivery:  delivery,
 		verbose:   verbose,
-		wait:      time.Sleep,
-		done:      make(chan struct{}),
 	}
 	b.broker = &eventBroker{
 		byEvent: make(map[string][]channelSubscription),
-		deliver: b.deliver,
+		deliver: delivery.deliver,
 		done:    make(chan struct{}),
 	}
 	return b
@@ -100,8 +80,20 @@ func (b *eventBus) fireTargeted(name string, payload map[string]any, firingSchem
 		return
 	}
 	for _, sub := range subs {
-		b.deliverTargeted(sub, payload, recipient)
+		b.delivery.deliverTargeted(sub, payload, recipient)
+	}
+}
+
+// doneChannel returns the bus shutdown signal so callers that schedule work on
+// their own goroutines (e.g. delayed management pushes) can cancel it when the
+// bus shuts down.
+func (b *eventBus) doneChannel() <-chan struct{} {
+	if b == nil {
+		ch := make(chan struct{})
+		close(ch)
+		return ch
 	}
+	return b.delivery.doneChannel()
 }
 
 // hasSubscribers reports whether any event-driven example could match an
@@ -111,13 +103,15 @@ func (b *eventBus) hasSubscribers(name, schema string) bool {
 }
 
 // shutdown cancels all periodic interval jobs and cancels pending delayed
-// emissions so no delivery happens after shutdown.
+// emissions (broker and delivery) so no delivery happens after shutdown.
 func (b *eventBus) shutdown() {
-	if b == nil || b.scheduler == nil {
+	if b == nil {
 		return
 	}
-	b.scheduler.shutdown()
-	b.stopOne.Do(func() { close(b.done) })
+	if b.scheduler != nil {
+		b.scheduler.shutdown()
+	}
+	b.delivery.shutdown()
 	b.broker.stop()
 }
 
@@ -289,7 +283,7 @@ func (b *eventBus) registerPeriodic(address, prefix, exampleID string, spec *loa
 		exampleID: exampleID,
 		channel:   address,
 		deliver: func() {
-			b.deliverPeriodic(address, prefix, spec, opID)
+			b.delivery.deliverPeriodic(address, prefix, spec, opID)
 		},
 	})
 	go b.scheduler.run(job)
@@ -320,19 +314,6 @@ func (b *eventBus) removeIntervalJob(jobID string) {
 	}
 }
 
-// deliverPeriodic renders a periodically driven example against current state
-// and environment and broadcasts it to the channel's consumers.
-func (b *eventBus) deliverPeriodic(address, prefix string, spec *loader.MessageExampleSpec, opID string) {
-	view := &MessageExampleView{spec: spec}
-	body := b.renderExample(view, b.stateEnvEvaluator(prefix), prefix, opID)
-	if body == nil {
-		return
-	}
-	b.notifyPush(address, "", body)
-	b.bus.SignalRPush(address, body)
-	b.bus.WSBroadcast(address, body)
-}
-
 // derivedExamples maps one spec example into the example specs to register.
 // An example without x-send-events maps to itself. A legacy x-send-events
 // example maps through the deprecation shim: each entry becomes the unified
diff --git a/internal/server/hubmanager.go b/internal/server/hubmanager.go
index c7e4ff6..9610ef5 100644
--- a/internal/server/hubmanager.go
+++ b/internal/server/hubmanager.go
@@ -64,10 +64,7 @@ func (m *hubManager) hubChannelForAddress(address string) (*signalRHub, string)
 // hasConnection reports whether a connection id is active on any hub.
 func (m *hubManager) hasConnection(id string) bool {
 	for _, hub := range m.hubs {
-		hub.mu.Lock()
-		_, ok := hub.conns[id]
-		hub.mu.Unlock()
-		if ok {
+		if hub.conns.hasConnection(id) {
 			return true
 		}
 	}
@@ -79,7 +76,7 @@ func (m *hubManager) hasConnection(id string) bool {
 func (m *hubManager) SignalRPush(address string, payload []byte) {
 	hub, channelID := m.hubChannelForAddress(address)
 	if hub != nil {
-		hub.pushToStreams(channelID, payload, channelID)
+		hub.conns.pushToStreams(channelID, payload, channelID)
 	}
 }
 
@@ -112,7 +109,7 @@ func (m *hubManager) Candidates(address string) []ConsumerInfo {
 	}
 	if hub, channelID := m.hubChannelForAddress(address); hub != nil {
 		seen := make(map[string]int) // connectionID -> index in out
-		for _, st := range hub.openStreamsForChannel(channelID) {
+		for _, st := range hub.conns.openStreamsForChannel(channelID) {
 			connID := st["connectionId"]
 			if idx, ok := seen[connID]; ok {
 				out[idx].Streams = append(out[idx].Streams, st)
@@ -122,8 +119,8 @@ func (m *hubManager) Candidates(address string) []ConsumerInfo {
 			out = append(out, ConsumerInfo{
 				ConnectionID: connID,
 				Channel:      address,
-				Query:        hub.connectionMetadata(connID),
-				Headers:      hub.connectionHeaders(connID),
+				Query:        hub.conns.connectionMetadata(connID),
+				Headers:      hub.conns.connectionHeaders(connID),
 				Streams:      []map[string]string{st},
 			})
 		}
@@ -131,28 +128,6 @@ func (m *hubManager) Candidates(address string) []ConsumerInfo {
 	return out
 }
 
-// connectionMetadata returns the upgrade-time query metadata of a connection
-// (for {$connection.query.*} evaluation); nil when unknown.
-func (h *signalRHub) connectionMetadata(connID string) map[string][]string {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	if sc, ok := h.conns[connID]; ok {
-		return sc.query
-	}
-	return nil
-}
-
-// connectionHeaders returns the upgrade-time header metadata of a connection;
-// nil when unknown. Header keys are lower-cased at capture time.
-func (h *signalRHub) connectionHeaders(connID string) map[string][]string {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	if sc, ok := h.conns[connID]; ok {
-		return sc.headers
-	}
-	return nil
-}
-
 // PushTo delivers a payload to one candidate consumer: its raw ws connection,
 // or its SignalR stream network (falling back to a server invocation).
 func (m *hubManager) PushTo(consumer ConsumerInfo, address string, payload []byte) {
@@ -166,5 +141,5 @@ func (m *hubManager) PushTo(consumer ConsumerInfo, address string, payload []byt
 	if hub == nil {
 		return
 	}
-	hub.pushToConnection(consumer.ConnectionID, channelID, payload, channelID)
+	hub.conns.pushToConnection(consumer.ConnectionID, channelID, payload, channelID)
 }
diff --git a/internal/server/management_async.go b/internal/server/management_async.go
index 691c532..28a5143 100644
--- a/internal/server/management_async.go
+++ b/internal/server/management_async.go
@@ -178,7 +178,7 @@ func (s *Server) handleAsyncConsumers(w http.ResponseWriter, r *http.Request) {
 		for _, hub := range s.hubMgr.hubs {
 			for channelID := range hub.channels {
 				address := asyncAddressWithPrefix(hub.prefix, hub.channels[channelID].Address)
-				for _, st := range hub.openStreamsForChannel(channelID) {
+				for _, st := range hub.conns.openStreamsForChannel(channelID) {
 					consumers = append(consumers, consumerInfo{
 						ConnectionID: st["connectionId"],
 						Channel:      address,
@@ -189,7 +189,7 @@ func (s *Server) handleAsyncConsumers(w http.ResponseWriter, r *http.Request) {
 		}
 	} else if hub := s.hubForAddress(channel); hub != nil {
 		if id := matchingHubChannel(hub, channel); id != "" {
-			for _, st := range hub.openStreamsForChannel(id) {
+			for _, st := range hub.conns.openStreamsForChannel(id) {
 				consumers = append(consumers, consumerInfo{
 					ConnectionID: st["connectionId"],
 					Channel:      channel,
@@ -238,13 +238,11 @@ func (s *Server) handleAsyncDisconnect(w http.ResponseWriter, r *http.Request) {
 
 	// SignalR hub connection (RS.AMG.14-15).
 	for _, hub := range s.hubMgr.hubs {
-		hub.mu.Lock()
-		if sc, ok := hub.conns[req.ConnectionID]; ok {
-			hub.mu.Unlock()
+		if sc, ok := hub.conns.connection(req.ConnectionID); ok {
 			s.disconnectWS(sc.writer, req)
+			hub.conns.unregister(req.ConnectionID)
 			return
 		}
-		hub.mu.Unlock()
 	}
 
 	// Raw ws connection.
diff --git a/internal/server/management_async_lifecycle_test.go b/internal/server/management_async_lifecycle_test.go
index 01dc01f..8e9d08b 100644
--- a/internal/server/management_async_lifecycle_test.go
+++ b/internal/server/management_async_lifecycle_test.go
@@ -303,15 +303,13 @@ func TestPushEndpoint_TargetedSignalR(t *testing.T) {
 	require.NoError(t, err) // handshake reply
 
 	hub := srv.hubMgr.hubs[0]
-	hub.mu.Lock()
 	var streamConnID string
-	for id, sc := range hub.conns {
+	for id, sc := range hub.conns.connections() {
 		if len(sc.streams) > 0 {
 			streamConnID = id
 			break
 		}
 	}
-	hub.mu.Unlock()
 	require.NotEmpty(t, streamConnID)
 
 	body := `{"channel":"/priceFeed","connectionId":"` + streamConnID + `","payload":{"seq":1}}`
diff --git a/internal/server/message_delivery.go b/internal/server/message_delivery.go
new file mode 100644
index 0000000..0e204ed
--- /dev/null
+++ b/internal/server/message_delivery.go
@@ -0,0 +1,254 @@
+package server
+
+import (
+	"cmp"
+	"encoding/json"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/mamonth/oasmock/internal/extensions"
+	"github.com/mamonth/oasmock/internal/loader"
+	"github.com/mamonth/oasmock/internal/runtime"
+)
+
+// messageDelivery renders and delivers subscribed AsyncAPI message examples
+// through the MessageRenderer and ConsumerBus contracts. It is the cohesive
+// delivery engine behind the event driver (design D8): it owns the sources of
+// per-emission rendering (state, env, event, connection), the recipient
+// partition, the delayed-emission cancellation and the push/observer side
+// effects. It never reaches into Server or the broker/scheduler registries.
+type messageDelivery struct {
+	renderer MessageRenderer
+	bus      ConsumerBus
+	verbose  bool
+
+	// observer, when set, is invoked with push envelopes so the management
+	// stream can mirror deliveries (RS.AMG.25).
+	observer func(env manageEnvelope)
+	// done is closed on shutdown so pending delayed emissions no longer run.
+	done    chan struct{}
+	stopOne sync.Once
+}
+
+// newMessageDelivery wires a delivery engine over a renderer and consumer bus.
+func newMessageDelivery(renderer MessageRenderer, bus ConsumerBus, verbose bool) *messageDelivery {
+	return &messageDelivery{
+		renderer: renderer,
+		bus:      bus,
+		verbose:  verbose,
+		done:     make(chan struct{}),
+	}
+}
+
+// setObserver installs the management-stream push observer.
+func (d *messageDelivery) setObserver(observer func(env manageEnvelope)) {
+	d.observer = observer
+}
+
+// doneChannel returns the delivery shutdown signal so callers that schedule
+// work on their own goroutines can cancel it on shutdown.
+func (d *messageDelivery) doneChannel() <-chan struct{} {
+	if d == nil {
+		ch := make(chan struct{})
+		close(ch)
+		return ch
+	}
+	return d.done
+}
+
+// shutdown cancels pending delayed emissions so no delivery happens after
+// shutdown.
+func (d *messageDelivery) shutdown() {
+	if d == nil {
+		return
+	}
+	d.stopOne.Do(func() { close(d.done) })
+}
+
+// deliver delivers an event payload to a subscription's consumers (broadcast).
+func (d *messageDelivery) deliver(sub channelSubscription, payload map[string]any) {
+	d.deliverTo(sub, payload, nil)
+}
+
+// deliverTargeted delivers a built-in event to a single candidate connection.
+// The connection bucket (if any) is evaluated against that one recipient only;
+// with no connection conditions the message is pushed to the recipient alone
+// (RS.EVT.24, RS.EXT.26).
+func (d *messageDelivery) deliverTargeted(sub channelSubscription, payload map[string]any, recipient ConsumerInfo) {
+	d.deliverTo(sub, payload, &recipient)
+}
+
+// deliverTo runs the shared delayed-emission + delivery pipeline for a
+// subscription. When target is non-nil, delivery is restricted to that single
+// candidate (built-in connect recipient).
+func (d *messageDelivery) deliverTo(sub channelSubscription, payload map[string]any, target *ConsumerInfo) {
+	if len(sub.messages) == 0 {
+		return
+	}
+	if sub.delay > 0 {
+		ms := sub.delay
+		sub.delay = 0
+		go func() {
+			select {
+			case <-d.done:
+				return
+			case <-time.After(time.Duration(ms) * time.Millisecond):
+			}
+			d.deliverTo(sub, payload, target)
+		}()
+		return
+	}
+	deliverable := sub.messages[0]
+	addr := sub.address
+	prefix := deliverable.prefix
+	eventName := sub.event
+	opID := "event:" + cmp.Or(eventName, anyEventIdentity) + ":" + addr
+
+	d.deliverExample(sub, deliverable.spec.Examples, addr, prefix, eventName, payload, opID, target)
+}
+
+// stateEnvEvaluator wires the fixed state and environment sources shared by
+// every emission path (periodic deliveries have no event/connection context).
+func (d *messageDelivery) stateEnvEvaluator(prefix string) runtime.Evaluator {
+	eval := runtime.NewEvaluator()
+	eval.AddSource(runtime.SourceState, d.renderer.NewStateSource(prefix))
+	eval.AddSource(runtime.SourceEnv, d.renderer.NewEnvSource())
+	return eval
+}
+
+// eventEvaluator wires the fixed emission sources (state, env, event) plus an
+// optional per-connection source into a fresh evaluator.
+func (d *messageDelivery) eventEvaluator(state, env runtime.DataSource, eventName string, payload map[string]any, connection *runtime.ConnectionSource) runtime.Evaluator {
+	eval := runtime.NewEvaluator()
+	eval.AddSource(runtime.SourceState, state)
+	eval.AddSource(runtime.SourceEnv, env)
+	eval.AddSource(runtime.SourceEvent, &runtime.EventSource{Name: eventName, Data: payload})
+	if connection != nil {
+		eval.AddSource(runtime.SourceConnection, connection)
+	}
+	return eval
+}
+
+// renderExample renders a single example's payload against an evaluator,
+// honoring x-mock-skip and x-mock-set-state. It returns the rendered body, or
+// nil when the example is skipped or rendering fails (verbose-logged).
+func (d *messageDelivery) renderExample(view *MessageExampleView, eval runtime.Evaluator, prefix, opID string) []byte {
+	if extensions.ValueSkip(view) {
+		return nil
+	}
+	if stateMap, ok := extensions.ValueSetState(view); ok {
+		d.renderer.ApplySetState(stateMap, eval, prefix)
+	}
+	body, err := d.renderer.RenderAsyncPayload(view, eval)
+	if err != nil {
+		if d.verbose {
+			slog.Debug("Example delivery render failed", "opID", opID, "err", err)
+		}
+		return nil
+	}
+	return body
+}
+
+// evaluateConnectionBucket evaluates an example's connection conditions
+// against one candidate recipient. An empty bucket matches every candidate.
+func (d *messageDelivery) evaluateConnectionBucket(bucket extensions.ParamsMatch, state, env runtime.DataSource, eventName string, payload map[string]any, candidate ConsumerInfo) (bool, error) {
+	if len(bucket) == 0 {
+		return true, nil
+	}
+	eval := d.eventEvaluator(state, env, eventName, payload, connectionSourceFromInfo(candidate))
+	return extensions.EvaluateParamsMatch(bucket, eval)
+}
+
+// deliverExample runs the shared selection + render + recipient-partition
+// pipeline for one subscription's examples. When target is non-nil, delivery
+// is restricted to that single candidate (built-in connect recipient).
+func (d *messageDelivery) deliverExample(sub channelSubscription, examples []*loader.MessageExampleSpec, addr, prefix, eventName string, payload map[string]any, opID string, target *ConsumerInfo) {
+	// Fixed (non-connection) sources evaluated once per emission.
+	state := d.renderer.NewStateSource(prefix)
+	env := d.renderer.NewEnvSource()
+
+	for _, example := range examples {
+		view := &MessageExampleView{spec: example}
+		common, connection := d.partitionedMatch(view)
+		var connSource *runtime.ConnectionSource
+		if target != nil {
+			connSource = connectionSourceFromInfo(*target)
+		}
+		evaluator := d.eventEvaluator(state, env, eventName, payload, connSource)
+		if len(common) > 0 {
+			ok, cErr := extensions.EvaluateParamsMatch(common, evaluator)
+			if cErr != nil || !ok {
+				continue
+			}
+		}
+		body := d.renderExample(view, evaluator, prefix, opID)
+		if body == nil {
+			continue
+		}
+		if target != nil {
+			// Built-in recipient: evaluate the connection bucket against the
+			// single candidate and deliver on match (or immediately when there
+			// is no connection bucket).
+			ok, okErr := d.evaluateConnectionBucket(connection, state, env, eventName, payload, *target)
+			if okErr != nil || !ok {
+				continue
+			}
+			d.notifyPush(addr, target.ConnectionID, body)
+			d.bus.PushTo(*target, addr, body)
+			continue
+		}
+		if len(connection) == 0 {
+			// Broadcast fast path (RS.EXT.25).
+			d.notifyPush(addr, "", body)
+			d.bus.SignalRPush(addr, body)
+			d.bus.WSBroadcast(addr, body)
+			continue
+		}
+		// Per-connection partition: evaluate the connection bucket against each
+		// candidate with its connection context (design D6).
+		for _, candidate := range d.bus.Candidates(addr) {
+			ok, okErr := d.evaluateConnectionBucket(connection, state, env, eventName, payload, candidate)
+			if okErr != nil || !ok {
+				continue
+			}
+			d.notifyPush(addr, candidate.ConnectionID, body)
+			d.bus.PushTo(candidate, addr, body)
+		}
+	}
+}
+
+// deliverPeriodic renders a periodically driven example against current state
+// and environment and broadcasts it to the channel's consumers.
+func (d *messageDelivery) deliverPeriodic(address, prefix string, spec *loader.MessageExampleSpec, opID string) {
+	view := &MessageExampleView{spec: spec}
+	body := d.renderExample(view, d.stateEnvEvaluator(prefix), prefix, opID)
+	if body == nil {
+		return
+	}
+	d.notifyPush(address, "", body)
+	d.bus.SignalRPush(address, body)
+	d.bus.WSBroadcast(address, body)
+}
+
+// notifyPush emits a push envelope to the management observer (RS.AMG.25).
+func (d *messageDelivery) notifyPush(channel, connectionID string, body []byte) {
+	if d.observer == nil {
+		return
+	}
+	var payload map[string]any
+	if err := json.Unmarshal(body, &payload); err != nil {
+		payload = map[string]any{"raw": string(body)}
+	}
+	env := manageEnvelope{Type: "push"}
+	env.Push = &managePushEnvelope{Channel: channel, ConnectionID: connectionID, Payload: payload}
+	d.observer(env)
+}
+
+// partitionedMatch splits an example's x-mock-match into common conditions
+// (evaluated once per emission) and connection conditions (evaluated per
+// candidate recipient). A nil/absent match yields empty buckets.
+func (d *messageDelivery) partitionedMatch(view *MessageExampleView) (extensions.ParamsMatch, extensions.ParamsMatch) {
+	match, _ := extensions.ValueMatch(view)
+	return extensions.PartitionConnectionConditions(extensions.ParamsMatch(match))
+}
diff --git a/internal/server/signalr_conn.go b/internal/server/signalr_conn.go
index a1d56af..4761f21 100644
--- a/internal/server/signalr_conn.go
+++ b/internal/server/signalr_conn.go
@@ -145,13 +145,13 @@ func (h *signalRHub) handleStreamInvocation(sc *signalRConnection, env signalREn
 	}))
 
 	// Hold the stream open and register it (RS.SHR.4, RS.SHR.21).
-	h.mu.Lock()
+	h.conns.mu.Lock()
 	sc.streams[env.InvocationID] = &signalRStream{
 		invocationID: env.InvocationID,
 		channelID:    channelID,
 		connID:       sc.id,
 	}
-	h.mu.Unlock()
+	h.conns.mu.Unlock()
 }
 
 // handleInvocation answers a one-shot Invocation by operation ID with a
@@ -181,22 +181,14 @@ func (h *signalRHub) handleInvocation(sc *signalRConnection, env signalREnvelope
 
 // handleCancelInvocation closes an open stream (RS.SHR.17).
 func (h *signalRHub) handleCancelInvocation(sc *signalRConnection, env signalREnvelope) {
-	h.mu.Lock()
+	h.conns.mu.Lock()
 	if st, ok := sc.streams[env.InvocationID]; ok {
-		h.unregisterStream(st)
+		h.conns.unregisterStream(st)
 	}
-	h.mu.Unlock()
+	h.conns.mu.Unlock()
 	h.writeCompletion(sc, env.InvocationID, "")
 }
 
-// unregisterStream removes a stream from its connection registry.
-// The caller must hold h.mu.
-func (h *signalRHub) unregisterStream(st *signalRStream) {
-	if sc, ok := h.conns[st.connID]; ok {
-		delete(sc.streams, st.invocationID)
-	}
-}
-
 // writeCompletion sends a completion envelope.
 func (h *signalRHub) writeCompletion(sc *signalRConnection, invocationID, errMsg string) {
 	env := signalREnvelope{Type: signalRTypeCompletion, InvocationID: invocationID}
@@ -227,23 +219,3 @@ func (h *signalRHub) renderOperation(opID string) (int, []byte, error) {
 	opKey := "signalr:operation:" + opID
 	return h.renderer.RenderMessageSpecs(specs, h.prefix, opKey, InboundMessage{})
 }
-
-// streamDelivery is one snapshotted write to perform after the hub lock is
-// released, so network I/O never blocks other hub operations (negotiate,
-func (h *signalRHub) openStreamsForChannel(channelID string) []map[string]string {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	var out []map[string]string
-	for _, sc := range h.conns {
-		for invocationID, st := range sc.streams {
-			if st.channelID == channelID {
-				out = append(out, map[string]string{
-					"connectionId": sc.id,
-					"invocationId": invocationID,
-					"streamId":     st.channelID,
-				})
-			}
-		}
-	}
-	return out
-}
diff --git a/internal/server/signalr_hub.go b/internal/server/signalr_hub.go
index ec5d363..d332095 100644
--- a/internal/server/signalr_hub.go
+++ b/internal/server/signalr_hub.go
@@ -3,9 +3,7 @@ package server
 import (
 	"encoding/json"
 	"net/http"
-	"strconv"
 	"strings"
-	"sync"
 
 	"github.com/gorilla/websocket"
 	"github.com/mamonth/oasmock/internal/asyncapi"
@@ -26,8 +24,9 @@ type signalRAvailableTransport struct {
 }
 
 // signalRHub is the SignalR overlay serving a single AsyncAPI document
-// declared with root-level x-signalr (design D7). It depends only on the
-// MessageRenderer surface, never on the whole Server.
+// declared with root-level x-signalr (design D7). It owns the HTTP transport
+// (negotiate/upgrade), the protocol framing and the document model; connection
+// and open-stream state lives in the signalRConnRegistry it delegates to.
 type signalRHub struct {
 	renderer MessageRenderer
 	path     string // hub path, e.g. "/hub"
@@ -40,10 +39,8 @@ type signalRHub struct {
 	// lifecycle notifications (D5).
 	hooks builtInHooks
 
-	mu     sync.Mutex
-	tokens map[string]string // connection token -> connection id
-	conns  map[string]*signalRConnection
-	idSeq  int
+	// conns owns tokens, connections and open streams.
+	conns *signalRConnRegistry
 }
 
 // setHooks wires built-in trigger and lifecycle callbacks into the hub.
@@ -58,7 +55,6 @@ type signalRConnection struct {
 	conn    *websocket.Conn
 	writer  *wsWriter
 	streams map[string]*signalRStream // invocationId -> open stream
-	server  *signalRHub
 	// query/headers capture the upgrade-time metadata for {$connection.*}
 	// evaluation (RS.EXT.27).
 	query   map[string][]string
@@ -80,8 +76,7 @@ func newSignalRHub(renderer MessageRenderer, doc *asyncapi.Document, prefix stri
 		prefix:   prefix,
 		channels: make(map[string]*asyncapi.Channel),
 		ops:      make(map[string]*asyncapi.Operation),
-		tokens:   make(map[string]string),
-		conns:    make(map[string]*signalRConnection),
+		conns:    newSignalRConnRegistry(),
 	}
 	if doc != nil {
 		hub.path = signalRPath(doc)
@@ -149,7 +144,7 @@ func (h *signalRHub) negotiate(w http.ResponseWriter, r *http.Request) {
 		writeJSONError(w, http.StatusBadRequest, "unsupported transport "+transport)
 		return
 	}
-	token, connID := h.issueToken()
+	token, connID := h.conns.issueToken()
 	resp := signalRNegotiate{
 		ConnectionToken:  token,
 		ConnectionID:     connID,
@@ -171,37 +166,6 @@ func isSignalRWebSockets(transport string) bool {
 	return strings.EqualFold(transport, "webSockets") || strings.EqualFold(transport, "websockets")
 }
 
-// issueToken creates and records a connection token.
-func (h *signalRHub) issueToken() (token, connID string) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	h.idSeq++
-	connID = "signalr-" + strconv.Itoa(h.idSeq)
-	token = connID + "-t"
-	h.tokens[token] = connID
-	return token, connID
-}
-
-// consumeToken validates and consumes a token, binding the connection.
-func (h *signalRHub) consumeToken(token string) (string, bool) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	connID, ok := h.tokens[token]
-	if ok {
-		delete(h.tokens, token)
-	}
-	return connID, ok
-}
-
-// freshToken returns a token for a connection id (no pre-correlation).
-func (h *signalRHub) freshToken() (token, connID string) {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	h.idSeq++
-	connID = "signalr-fresh-" + strconv.Itoa(h.idSeq)
-	return connID + "-t", connID
-}
-
 // serveUpgrade handles a WebSocket upgrade to the hub path (RS.SHR.11-13).
 func (h *signalRHub) serveUpgrade(w http.ResponseWriter, r *http.Request) {
 	// Only the WebSockets transport can upgrade (RS.SHR.10).
@@ -214,14 +178,14 @@ func (h *signalRHub) serveUpgrade(w http.ResponseWriter, r *http.Request) {
 	token := idParam
 	if idParam != "" {
 		var ok bool
-		connID, ok = h.consumeToken(idParam)
+		connID, ok = h.conns.consumeToken(idParam)
 		if !ok {
 			writeJSONError(w, http.StatusNotFound, "unknown connection token")
 			return
 		}
 	} else {
 		// Fresh internally generated connection id (RS.SHR.13).
-		_, connID = h.freshToken()
+		_, connID = h.conns.freshToken()
 		token = connID + "-t"
 	}
 
@@ -238,13 +202,10 @@ func (h *signalRHub) serveUpgrade(w http.ResponseWriter, r *http.Request) {
 		conn:    conn,
 		writer:  wr,
 		streams: make(map[string]*signalRStream),
-		server:  h,
 		query:   r.URL.Query(),
 		headers: lowerHeaderKeys(r.Header),
 	}
-	h.mu.Lock()
-	h.conns[connID] = sc
-	h.mu.Unlock()
+	h.conns.register(sc)
 
 	channel := hubDefaultChannel(h)
 	info := ConsumerInfo{
@@ -261,12 +222,10 @@ func (h *signalRHub) serveUpgrade(w http.ResponseWriter, r *http.Request) {
 	}
 
 	defer func() {
-		h.mu.Lock()
 		// The connection's open streams are discarded with the connection
-		// object: removing connID makes them undiscoverable (openStreamsForChannel
-		// iterates h.conns), so no separate stream cleanup is needed.
-		delete(h.conns, connID)
-		h.mu.Unlock()
+		// object: unregistering makes them undiscoverable (openStreamsForChannel
+		// iterates the registry), so no separate stream cleanup is needed.
+		h.conns.unregister(connID)
 		wr.close()
 		if h.hooks.OnDisconnect != nil {
 			h.hooks.OnDisconnect(channel, connID)
@@ -303,5 +262,3 @@ func (s *Server) registerSignalRHubs(r interface {
 		r.Get(hub.upgradePath(), hub.serveUpgrade)
 	}
 }
-
-// openStreamsForChannel returns open-stream descriptions for a channel.
diff --git a/internal/server/signalr_hub_test.go b/internal/server/signalr_hub_test.go
index 027ac21..1bf4275 100644
--- a/internal/server/signalr_hub_test.go
+++ b/internal/server/signalr_hub_test.go
@@ -79,15 +79,15 @@ func TestSignalRHub_TokenCorrelation(t *testing.T) {
 	srv, _, _, _ := newMockedServerWithGeneratedMocks(t, Config{HistorySize: DefaultHistorySize})
 	hub := newSignalRHubAtPath(srv, "/hub", "", nil)
 
-	token, connID := hub.issueToken()
+	token, connID := hub.conns.issueToken()
 	require.NotEmpty(t, token)
 	require.NotEmpty(t, connID)
 
-	gotConnID, ok := hub.consumeToken(token)
+	gotConnID, ok := hub.conns.consumeToken(token)
 	assert.True(t, ok, "issued token must correlate with its connection id")
 	assert.Equal(t, connID, gotConnID)
 	// The token is consumed on correlation, so it cannot be reused.
-	_, ok = hub.consumeToken(token)
+	_, ok = hub.conns.consumeToken(token)
 	assert.False(t, ok, "consumed token must not correlate again")
 }
 
@@ -105,11 +105,11 @@ func TestSignalRHub_FreshToken(t *testing.T) {
 	srv, _, _, _ := newMockedServerWithGeneratedMocks(t, Config{HistorySize: DefaultHistorySize})
 	hub := newSignalRHubAtPath(srv, "/hub", "", nil)
 
-	token, connID := hub.freshToken()
+	token, connID := hub.conns.freshToken()
 	require.NotEmpty(t, token)
 	require.NotEmpty(t, connID)
 	// A fresh token is not correlated until an upgrade presents it.
-	_, ok := hub.consumeToken(token)
+	_, ok := hub.conns.consumeToken(token)
 	assert.False(t, ok, "fresh token is not yet correlated")
 }
 
@@ -239,7 +239,6 @@ func TestHubManager_CandidatesDeduplicatesStreams(t *testing.T) {
 	srv, _, _, _ := newMockedServerWithGeneratedMocks(t, Config{HistorySize: DefaultHistorySize})
 	hub := newSignalRHubAtPath(srv, "/hub", "", nil)
 	hub.channels["priceFeed"] = &asyncapi.Channel{ID: "priceFeed", Address: "/price"}
-	hub.mu.Lock()
 	sc := &signalRConnection{
 		id:      "signalr-1",
 		writer:  newWSWriter(nil),
@@ -247,8 +246,7 @@ func TestHubManager_CandidatesDeduplicatesStreams(t *testing.T) {
 	}
 	sc.streams["inv-1"] = &signalRStream{invocationID: "inv-1", channelID: "priceFeed", connID: "signalr-1"}
 	sc.streams["inv-2"] = &signalRStream{invocationID: "inv-2", channelID: "priceFeed", connID: "signalr-1"}
-	hub.conns["signalr-1"] = sc
-	hub.mu.Unlock()
+	hub.conns.register(sc)
 
 	mgr := &hubManager{hubs: []*signalRHub{hub}}
 	candidates := mgr.Candidates("/price")
@@ -278,15 +276,13 @@ func TestSignalRHub_OpenStreamsForChannel(t *testing.T) {
 		streams: make(map[string]*signalRStream),
 	}
 	sc.streams["inv-1"] = &signalRStream{invocationID: "inv-1", channelID: "priceFeed", connID: "signalr-1"}
-	hub.mu.Lock()
-	hub.conns["signalr-1"] = sc
-	hub.mu.Unlock()
+	hub.conns.register(sc)
 
-	streams := hub.openStreamsForChannel("priceFeed")
+	streams := hub.conns.openStreamsForChannel("priceFeed")
 	require.Len(t, streams, 1)
 	assert.Equal(t, "signalr-1", streams[0]["connectionId"])
 	assert.Equal(t, "inv-1", streams[0]["invocationId"])
 	assert.Equal(t, "priceFeed", streams[0]["streamId"])
 
-	assert.Empty(t, hub.openStreamsForChannel("otherChannel"))
+	assert.Empty(t, hub.conns.openStreamsForChannel("otherChannel"))
 }
diff --git a/internal/server/signalr_integration_test.go b/internal/server/signalr_integration_test.go
index a259b53..03aa072 100644
--- a/internal/server/signalr_integration_test.go
+++ b/internal/server/signalr_integration_test.go
@@ -320,7 +320,7 @@ func TestSignalR_PushToOpenStream(t *testing.T) {
 	require.NoError(t, err) // snapshot
 
 	hub := srv.hubMgr.hubs[0]
-	hub.pushToStreams("priceFeed", []byte(`{"symbol":"BTC","price":60000}`), "priceFeed")
+	hub.conns.pushToStreams("priceFeed", []byte(`{"symbol":"BTC","price":60000}`), "priceFeed")
 
 	_, msg, err := conn.ReadMessage()
 	require.NoError(t, err)
@@ -350,7 +350,7 @@ func TestSignalR_PushWithoutOpenStream(t *testing.T) {
 	// No stream opened: place a marker by sending a ping after which we expect
 	// only the server Invocation for the push.
 	hub := srv.hubMgr.hubs[0]
-	hub.pushToStreams("priceFeed", []byte(`{"symbol":"BTC","price":60000}`), "priceFeed")
+	hub.conns.pushToStreams("priceFeed", []byte(`{"symbol":"BTC","price":60000}`), "priceFeed")
 
 	_ = conn.SetReadDeadline(time.Now().Add(3 * time.Second))
 	_, msg, err := conn.ReadMessage()
diff --git a/internal/server/signalr_push.go b/internal/server/signalr_push.go
deleted file mode 100644
index 8b14b94..0000000
--- a/internal/server/signalr_push.go
+++ /dev/null
@@ -1,86 +0,0 @@
-package server
-
-import (
-	"encoding/json"
-	"strconv"
-)
-
-type streamDelivery struct {
-	writer *wsWriter
-	env    signalREnvelope
-}
-
-// pushToStreams emits a templated payload into all open streams of a channel;
-// when no stream is open it sends a server Invocation (RS.SHR.18-19).
-func (h *signalRHub) pushToStreams(channelID string, payload []byte, target string) {
-	h.mu.Lock()
-	deliveries := h.buildStreamDeliveries(channelID, payload, target, nil)
-	h.mu.Unlock()
-	for _, d := range deliveries {
-		d.writer.write(encodeSignalRMessage(d.env))
-	}
-}
-
-// pushToConnection pushes a payload to one connection's open streams for the
-// channel, falling back to a server Invocation on that connection when no
-// stream is open (RS.AMG.5, RS.SHR.18-19).
-func (h *signalRHub) pushToConnection(connectionID, channelID string, payload []byte, target string) {
-	h.mu.Lock()
-	sc, ok := h.conns[connectionID]
-	var deliveries []streamDelivery
-	if ok {
-		deliveries = h.buildStreamDeliveries(channelID, payload, target, sc)
-	}
-	h.mu.Unlock()
-	for _, d := range deliveries {
-		d.writer.write(encodeSignalRMessage(d.env))
-	}
-}
-
-// buildStreamDeliveries snapshots the writes needed to deliver a payload to a
-// channel's open streams, falling back to per-connection server Invocations
-// (each with a distinct server-assigned id) when no stream matches. The caller
-// must hold h.mu; when conn is non-nil delivery is restricted to that single
-// connection.
-func (h *signalRHub) buildStreamDeliveries(channelID string, payload []byte, target string, conn *signalRConnection) []streamDelivery {
-	var out []streamDelivery
-	emitInvocation := func(sc *signalRConnection) {
-		h.idSeq++
-		out = append(out, streamDelivery{writer: sc.writer, env: signalREnvelope{
-			Type:         signalRTypeInvocation,
-			InvocationID: "srv-" + target + "-" + strconv.Itoa(h.idSeq),
-			Target:       target,
-			Arguments:    []any{json.RawMessage(payload)},
-		}})
-	}
-	matched := false
-	writeStreamItems := func(sc *signalRConnection) {
-		for invocationID, st := range sc.streams {
-			if st.channelID != channelID {
-				continue
-			}
-			matched = true
-			out = append(out, streamDelivery{writer: sc.writer, env: signalREnvelope{
-				Type:         signalRTypeStreamItem,
-				InvocationID: invocationID,
-				Item:         json.RawMessage(payload),
-			}})
-		}
-	}
-	if conn != nil {
-		writeStreamItems(conn)
-		if !matched {
-			emitInvocation(conn)
-		}
-		return out
-	}
-	for _, sc := range h.conns {
-		writeStreamItems(sc)
-	}
-	if !matched {
-		for _, sc := range h.conns {
-			emitInvocation(sc)
-		}
-	}
-	return out
-}
diff --git a/internal/server/signalr_registry.go b/internal/server/signalr_registry.go
new file mode 100644
index 0000000..ac912c8
--- /dev/null
+++ b/internal/server/signalr_registry.go
@@ -0,0 +1,230 @@
+package server
+
+import (
+	"encoding/json"
+	"strconv"
+	"sync"
+)
+
+// streamDelivery is one snapshotted write to perform after the registry lock
+// is released, so network I/O never blocks other hub operations.
+type streamDelivery struct {
+	writer *wsWriter
+	env    signalREnvelope
+}
+
+// signalRConnRegistry owns the connection, token and open-stream state of a
+// SignalR hub, plus the payload-delivery helpers that address those
+// connections. It is the single owner of the hub's mu so transport handling
+// (negotiate/upgrade/read loop) never reaches into connection state directly.
+type signalRConnRegistry struct {
+	mu     sync.Mutex
+	tokens map[string]string // connection token -> connection id
+	conns  map[string]*signalRConnection
+	idSeq  int
+}
+
+func newSignalRConnRegistry() *signalRConnRegistry {
+	return &signalRConnRegistry{
+		tokens: make(map[string]string),
+		conns:  make(map[string]*signalRConnection),
+	}
+}
+
+// issueToken creates and records a connection token.
+func (r *signalRConnRegistry) issueToken() (token, connID string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.idSeq++
+	connID = "signalr-" + strconv.Itoa(r.idSeq)
+	token = connID + "-t"
+	r.tokens[token] = connID
+	return token, connID
+}
+
+// consumeToken validates and consumes a token, binding the connection.
+func (r *signalRConnRegistry) consumeToken(token string) (string, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	connID, ok := r.tokens[token]
+	if ok {
+		delete(r.tokens, token)
+	}
+	return connID, ok
+}
+
+// freshToken returns a token for a connection id (no pre-correlation).
+func (r *signalRConnRegistry) freshToken() (token, connID string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.idSeq++
+	connID = "signalr-fresh-" + strconv.Itoa(r.idSeq)
+	return connID + "-t", connID
+}
+
+// register adds a connection to the registry.
+func (r *signalRConnRegistry) register(sc *signalRConnection) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.conns[sc.id] = sc
+}
+
+// unregister removes a connection; its open streams are discarded with the
+// connection object, so no separate stream cleanup is needed (they become
+// undiscoverable through openStreamsForChannel which iterates r.conns).
+func (r *signalRConnRegistry) unregister(id string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	delete(r.conns, id)
+}
+
+// hasConnection reports whether a connection id is active.
+func (r *signalRConnRegistry) hasConnection(id string) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	_, ok := r.conns[id]
+	return ok
+}
+
+// connection returns a registered connection, or nil.
+func (r *signalRConnRegistry) connection(id string) (*signalRConnection, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	sc, ok := r.conns[id]
+	return sc, ok
+}
+
+// connections returns a snapshot of all registered connections keyed by id.
+func (r *signalRConnRegistry) connections() map[string]*signalRConnection {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make(map[string]*signalRConnection, len(r.conns))
+	for id, sc := range r.conns {
+		out[id] = sc
+	}
+	return out
+}
+
+// connectionMetadata returns the upgrade-time query metadata of a connection
+// (for {$connection.query.*} evaluation); nil when unknown.
+func (r *signalRConnRegistry) connectionMetadata(connID string) map[string][]string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if sc, ok := r.conns[connID]; ok {
+		return sc.query
+	}
+	return nil
+}
+
+// connectionHeaders returns the upgrade-time header metadata of a connection;
+// nil when unknown. Header keys are lower-cased at capture time.
+func (r *signalRConnRegistry) connectionHeaders(connID string) map[string][]string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if sc, ok := r.conns[connID]; ok {
+		return sc.headers
+	}
+	return nil
+}
+
+// openStreamsForChannel returns open-stream descriptions for a channel.
+func (r *signalRConnRegistry) openStreamsForChannel(channelID string) []map[string]string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var out []map[string]string
+	for _, sc := range r.conns {
+		for invocationID, st := range sc.streams {
+			if st.channelID == channelID {
+				out = append(out, map[string]string{
+					"connectionId": sc.id,
+					"invocationId": invocationID,
+					"streamId":     st.channelID,
+				})
+			}
+		}
+	}
+	return out
+}
+
+// unregisterStream removes a stream from its connection registry. The caller
+// must hold r.mu.
+func (r *signalRConnRegistry) unregisterStream(st *signalRStream) {
+	if sc, ok := r.conns[st.connID]; ok {
+		delete(sc.streams, st.invocationID)
+	}
+}
+
+// pushToStreams emits a templated payload into all open streams of a channel;
+// when no stream is open it sends a server Invocation (RS.SHR.18-19).
+func (r *signalRConnRegistry) pushToStreams(channelID string, payload []byte, target string) {
+	r.mu.Lock()
+	deliveries := r.buildStreamDeliveries(channelID, payload, target, nil)
+	r.mu.Unlock()
+	for _, d := range deliveries {
+		d.writer.write(encodeSignalRMessage(d.env))
+	}
+}
+
+// pushToConnection pushes a payload to one connection's open streams for the
+// channel, falling back to a server Invocation on that connection when no
+// stream is open (RS.AMG.5, RS.SHR.18-19).
+func (r *signalRConnRegistry) pushToConnection(connectionID, channelID string, payload []byte, target string) {
+	r.mu.Lock()
+	sc, ok := r.conns[connectionID]
+	var deliveries []streamDelivery
+	if ok {
+		deliveries = r.buildStreamDeliveries(channelID, payload, target, sc)
+	}
+	r.mu.Unlock()
+	for _, d := range deliveries {
+		d.writer.write(encodeSignalRMessage(d.env))
+	}
+}
+
+// buildStreamDeliveries snapshots the writes needed to deliver a payload to a
+// channel's open streams, falling back to per-connection server Invocations
+// (each with a distinct server-assigned id) when no stream matches. The caller
+// must hold r.mu; when conn is non-nil delivery is restricted to that single
+// connection.
+func (r *signalRConnRegistry) buildStreamDeliveries(channelID string, payload []byte, target string, conn *signalRConnection) []streamDelivery {
+	var out []streamDelivery
+	emitInvocation := func(sc *signalRConnection) {
+		r.idSeq++
+		out = append(out, streamDelivery{writer: sc.writer, env: signalREnvelope{
+			Type:         signalRTypeInvocation,
+			InvocationID: "srv-" + target + "-" + strconv.Itoa(r.idSeq),
+			Target:       target,
+			Arguments:    []any{json.RawMessage(payload)},
+		}})
+	}
+	matched := false
+	writeStreamItems := func(sc *signalRConnection) {
+		for invocationID, st := range sc.streams {
+			if st.channelID != channelID {
+				continue
+			}
+			matched = true
+			out = append(out, streamDelivery{writer: sc.writer, env: signalREnvelope{
+				Type:         signalRTypeStreamItem,
+				InvocationID: invocationID,
+				Item:         json.RawMessage(payload),
+			}})
+		}
+	}
+	if conn != nil {
+		writeStreamItems(conn)
+		if !matched {
+			emitInvocation(conn)
+		}
+		return out
+	}
+	for _, sc := range r.conns {
+		writeStreamItems(sc)
+	}
+	if !matched {
+		for _, sc := range r.conns {
+			emitInvocation(sc)
+		}
+	}
+	return out
+}