diff --git a/chasm/lib/channel/channel.go b/chasm/lib/channel/channel.go
index f94b0268e52..da39689d1f3 100644
--- a/chasm/lib/channel/channel.go
+++ b/chasm/lib/channel/channel.go
@@ -14,6 +14,7 @@ import (
 	channelpb "go.temporal.io/server/chasm/lib/channel/gen/channelpb/v1"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/backoff"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -32,12 +33,12 @@ type Channel struct {
 	// one of them rather than rewriting the whole ring.
 	Retained chasm.Map[int64, *channelpb.Notification]
 
-	// Keyed by workflow id. One workflow id has one open run, so a later run
-	// that subscribes replaces the entry of the run before it.
-	WorkflowListeners chasm.Map[string, *channelpb.WorkflowListener]
-
-	// Keyed by listener id.
-	CallbackListeners chasm.Map[string, *channelpb.CallbackListener]
+	// Keyed by listener id. A workflow listens under its workflow id: one
+	// workflow id has one open run, so a later run that subscribes takes the
+	// entry of the run before it. A callback listens under its request id.
+	// One key space, so a callback whose request id is the id of a workflow
+	// listening here is refused.
+	Listeners chasm.Map[string, *channelpb.Listener]
 }
 
 // readOnly hides the mutable half of a context. Reaching a data node through a
@@ -51,10 +52,9 @@ func readOnly(ctx chasm.Context) chasm.Context {
 // a notify or a registration, which stamps the first activity.
 func NewChannel(_ chasm.MutableContext) *Channel {
 	return &Channel{
-		State:             &channelpb.ChannelState{},
-		Retained:          make(chasm.Map[int64, *channelpb.Notification]),
-		WorkflowListeners: make(chasm.Map[string, *channelpb.WorkflowListener]),
-		CallbackListeners: make(chasm.Map[string, *channelpb.CallbackListener]),
+		State:     &channelpb.ChannelState{},
+		Retained:  make(chasm.Map[int64, *channelpb.Notification]),
+		Listeners: make(chasm.Map[string, *channelpb.Listener]),
 	}
 }
 
@@ -70,11 +70,8 @@ func (c *Channel) Terminate(
 	mctx chasm.MutableContext,
 	_ chasm.TerminateComponentRequest,
 ) (chasm.TerminateComponentResponse, error) {
-	for id := range c.WorkflowListeners {
-		delete(c.WorkflowListeners, id)
-	}
-	for id := range c.CallbackListeners {
-		delete(c.CallbackListeners, id)
+	for id := range c.Listeners {
+		delete(c.Listeners, id)
 	}
 	c.scheduleIdleCheck(mctx, DefaultLimits())
 	return chasm.TerminateComponentResponse{}, nil
@@ -91,7 +88,55 @@ func (c *Channel) LifecycleState(_ chasm.Context) chasm.LifecycleState {
 
 // ListenerCount is how many listeners the channel holds, of both kinds.
 func (c *Channel) ListenerCount() int {
-	return len(c.WorkflowListeners) + len(c.CallbackListeners)
+	return len(c.Listeners)
+}
+
+// The listener_kind metrics tag, derived from the callback variant.
+const (
+	ListenerKindWorkflow = "workflow"
+	ListenerKindCallback = "callback"
+)
+
+// ListenerKind names what a callback reaches: a workflow run for an internal
+// callback, an HTTP endpoint for a Nexus one.
+func ListenerKind(cb *commonpb.Callback) string {
+	if cb.GetInternal() != nil {
+		return ListenerKindWorkflow
+	}
+	return ListenerKindCallback
+}
+
+// IsWorkflowListener reports whether the listener is a workflow run, reached
+// through the routed delivery rather than an HTTP post.
+func IsWorkflowListener(l *channelpb.Listener) bool {
+	return l.GetCallback().GetInternal() != nil
+}
+
+// WorkflowTargetOf decodes the run an internal callback reaches. It reports
+// false for an HTTP callback, and for data that does not parse, which the
+// delivery then has no way to reach.
+func WorkflowTargetOf(l *channelpb.Listener) (*channelpb.WorkflowTarget, bool) {
+	internal := l.GetCallback().GetInternal()
+	if internal == nil {
+		return nil, false
+	}
+	target := &channelpb.WorkflowTarget{}
+	if err := proto.Unmarshal(internal.GetData(), target); err != nil {
+		return nil, false
+	}
+	return target, true
+}
+
+// setWorkflowTarget stores the run under the listener's internal callback.
+func setWorkflowTarget(l *channelpb.Listener, target *channelpb.WorkflowTarget) error {
+	data, err := proto.Marshal(target)
+	if err != nil {
+		return err
+	}
+	l.Callback = &commonpb.Callback{Variant: &commonpb.Callback_Internal_{
+		Internal: &commonpb.Callback_Internal{Data: data},
+	}}
+	return nil
 }
 
 // CheckNotification refuses a notification the channel cannot accept. The
@@ -207,8 +252,9 @@ func (c *Channel) touch(mctx chasm.MutableContext) {
 
 // FanOut is what the fan-out task delivers and to whom.
 type FanOut struct {
-	Latest            *channelpb.Notification
-	WorkflowListeners []*channelpb.WorkflowListener
+	Latest *channelpb.Notification
+	// The workflow listeners, for the task to reach on their own shards.
+	Workflows []*channelpb.WorkflowTarget
 	// Callback listeners that already held the notification, in flight or
 	// pending, or whose pending one it replaced.
 	CallbackFolded int
@@ -238,19 +284,14 @@ func (c *Channel) TakeFanOut(mctx chasm.MutableContext, _ struct{}) (FanOut, err
 		return FanOut{}, err
 	}
 	out.CallbackStarted, out.CallbackFolded = started, folded
-
-	view := readOnly(mctx)
-	for _, id := range slices.Sorted(maps.Keys(c.WorkflowListeners)) {
-		out.WorkflowListeners = append(out.WorkflowListeners,
-			common.CloneProto(c.WorkflowListeners[id].Get(view)))
-	}
+	out.Workflows = c.WorkflowTargets(mctx)
 	return out, nil
 }
 
 // callbackHolds reports whether a callback listener already holds a
 // notification at this counter or above, in flight or pending. Only what it
 // holds counts: once a post is done, the same counter is news again.
-func callbackHolds(listener *channelpb.CallbackListener, counter int64) bool {
+func callbackHolds(listener *channelpb.Listener, counter int64) bool {
 	return (listener.GetInFlight() != nil && listener.GetInFlight().GetCounter() >= counter) ||
 		(listener.GetPending() != nil && listener.GetPending().GetCounter() >= counter)
 }
@@ -259,8 +300,9 @@ func callbackHolds(listener *channelpb.CallbackListener, counter int64) bool {
 // notification, which is when handing it over writes the channel.
 func (c *Channel) CallbacksNeed(ctx chasm.Context, counter int64) bool {
 	view := readOnly(ctx)
-	for _, field := range c.CallbackListeners {
-		if !callbackHolds(field.Get(view), counter) {
+	for _, field := range c.Listeners {
+		listener := field.Get(view)
+		if !IsWorkflowListener(listener) && !callbackHolds(listener, counter) {
 			return true
 		}
 	}
@@ -277,12 +319,16 @@ func (c *Channel) HandToCallbacks(
 	n *channelpb.Notification,
 ) (started int, folded int, err error) {
 	view := readOnly(mctx)
-	for _, id := range slices.Sorted(maps.Keys(c.CallbackListeners)) {
-		if callbackHolds(c.CallbackListeners[id].Get(view), n.GetCounter()) {
+	for _, id := range slices.Sorted(maps.Keys(c.Listeners)) {
+		current := c.Listeners[id].Get(view)
+		if IsWorkflowListener(current) {
+			continue
+		}
+		if callbackHolds(current, n.GetCounter()) {
 			folded++
 			continue
 		}
-		listener := c.CallbackListeners[id].Get(mctx)
+		listener := c.Listeners[id].Get(mctx)
 		listener.HandedCounter = max(listener.GetHandedCounter(), n.GetCounter())
 		if listener.GetInFlight() != nil {
 			if listener.GetPending() != nil {
@@ -301,12 +347,14 @@ func (c *Channel) HandToCallbacks(
 	return started, folded, nil
 }
 
-// WorkflowListenerList returns the workflow listeners in workflow id order.
-func (c *Channel) WorkflowListenerList(ctx chasm.Context) []*channelpb.WorkflowListener {
+// WorkflowTargets returns the workflow listeners in workflow id order.
+func (c *Channel) WorkflowTargets(ctx chasm.Context) []*channelpb.WorkflowTarget {
 	view := readOnly(ctx)
-	out := make([]*channelpb.WorkflowListener, 0, len(c.WorkflowListeners))
-	for _, id := range slices.Sorted(maps.Keys(c.WorkflowListeners)) {
-		out = append(out, common.CloneProto(c.WorkflowListeners[id].Get(view)))
+	var out []*channelpb.WorkflowTarget
+	for _, id := range slices.Sorted(maps.Keys(c.Listeners)) {
+		if target, ok := WorkflowTargetOf(c.Listeners[id].Get(view)); ok {
+			out = append(out, target)
+		}
 	}
 	return out
 }
@@ -327,7 +375,7 @@ func CallbackDestination(cb *commonpb.Callback) (string, error) {
 
 func (c *Channel) scheduleCallback(
 	mctx chasm.MutableContext,
-	listener *channelpb.CallbackListener,
+	listener *channelpb.Listener,
 ) error {
 	destination, err := CallbackDestination(listener.GetCallback())
 	if err != nil {
@@ -343,6 +391,7 @@ func (c *Channel) scheduleCallback(
 
 // WorkflowRegistration describes a workflow run that subscribed.
 type WorkflowRegistration struct {
+	NamespaceID         string
 	WorkflowID          string
 	RunID               string
 	FirstExecutionRunID string
@@ -367,28 +416,44 @@ func (c *Channel) RegisterWorkflowListener(
 	limits := reg.Limits.withDefaults()
 	c.touch(mctx)
 	latest := common.CloneProto(c.State.GetLatest())
-	if field, ok := c.WorkflowListeners[reg.WorkflowID]; ok {
-		if field.Get(readOnly(mctx)).GetRunId() == reg.RunID {
+	if field, ok := c.Listeners[reg.WorkflowID]; ok {
+		target, isWorkflow := WorkflowTargetOf(field.Get(readOnly(mctx)))
+		if !isWorkflow {
+			return nil, serviceerror.NewInvalidArgumentf(
+				"listener id %q belongs to a callback on this channel", reg.WorkflowID)
+		}
+		if target.GetRunId() == reg.RunID {
 			return nil, nil
 		}
+		target.RunId = reg.RunID
+		target.FirstExecutionRunId = reg.FirstExecutionRunID
 		listener := field.Get(mctx)
-		listener.RunId = reg.RunID
-		listener.FirstExecutionRunId = reg.FirstExecutionRunID
+		if err := setWorkflowTarget(listener, target); err != nil {
+			return nil, err
+		}
 		listener.RegisteredTime = timestamppb.New(mctx.Now(c))
 		return latest, nil
 	}
 	if err := c.checkListenerRoom(limits.MaxListeners); err != nil {
 		return nil, err
 	}
-	if c.WorkflowListeners == nil {
-		c.WorkflowListeners = make(chasm.Map[string, *channelpb.WorkflowListener])
+	listener := &channelpb.Listener{
+		ListenerId:     reg.WorkflowID,
+		RegisteredTime: timestamppb.New(mctx.Now(c)),
 	}
-	c.WorkflowListeners[reg.WorkflowID] = chasm.NewDataField(mctx, &channelpb.WorkflowListener{
+	err := setWorkflowTarget(listener, &channelpb.WorkflowTarget{
+		NamespaceId:         reg.NamespaceID,
 		WorkflowId:          reg.WorkflowID,
 		RunId:               reg.RunID,
 		FirstExecutionRunId: reg.FirstExecutionRunID,
-		RegisteredTime:      timestamppb.New(mctx.Now(c)),
 	})
+	if err != nil {
+		return nil, err
+	}
+	if c.Listeners == nil {
+		c.Listeners = make(chasm.Map[string, *channelpb.Listener])
+	}
+	c.Listeners[reg.WorkflowID] = chasm.NewDataField(mctx, listener)
 	return latest, nil
 }
 
@@ -407,12 +472,17 @@ func (c *Channel) checkListenerRoom(maxListeners int) error {
 func (c *Channel) RekeyWorkflowListener(
 	mctx chasm.MutableContext,
 	workflowID, fromRunID, toRunID string,
-) {
-	field, ok := c.WorkflowListeners[workflowID]
-	if !ok || field.Get(readOnly(mctx)).GetRunId() != fromRunID {
-		return
+) error {
+	field, ok := c.Listeners[workflowID]
+	if !ok {
+		return nil
 	}
-	field.Get(mctx).RunId = toRunID
+	target, isWorkflow := WorkflowTargetOf(field.Get(readOnly(mctx)))
+	if !isWorkflow || target.GetRunId() != fromRunID {
+		return nil
+	}
+	target.RunId = toRunID
+	return setWorkflowTarget(field.Get(mctx), target)
 }
 
 // ForgetWorkflowListener drops a listener no run of its workflow answers for
@@ -422,11 +492,15 @@ func (c *Channel) ForgetWorkflowListener(
 	workflowID, runID string,
 	limits Limits,
 ) {
-	field, ok := c.WorkflowListeners[workflowID]
-	if !ok || field.Get(readOnly(mctx)).GetRunId() != runID {
+	field, ok := c.Listeners[workflowID]
+	if !ok {
 		return
 	}
-	delete(c.WorkflowListeners, workflowID)
+	target, isWorkflow := WorkflowTargetOf(field.Get(readOnly(mctx)))
+	if !isWorkflow || target.GetRunId() != runID {
+		return
+	}
+	delete(c.Listeners, workflowID)
 	c.scheduleIdleCheck(mctx, limits.withDefaults())
 }
 
@@ -437,9 +511,9 @@ type CallbackRegistration struct {
 	Limits    Limits
 }
 
-// RegisterCallbackListener adds a callback listener and returns its id. A
-// registration whose request id the channel has seen returns the listener
-// that request made, so a retry adds nothing.
+// RegisterCallbackListener adds a callback listener and returns its id, which
+// is the request id. A registration whose request id the channel has seen
+// returns the listener that request made, so a retry adds nothing.
 //
 // A new listener is posted the latest notification the channel holds, once,
 // for the same reason a new workflow listener is handed it.
@@ -456,20 +530,21 @@ func (c *Channel) RegisterCallbackListener(
 	limits := reg.Limits.withDefaults()
 	c.touch(mctx)
 
-	view := readOnly(mctx)
-	for id, field := range c.CallbackListeners {
-		if field.Get(view).GetRequestId() == reg.RequestID {
-			return id, nil
+	id := reg.RequestID
+	if field, ok := c.Listeners[id]; ok {
+		if IsWorkflowListener(field.Get(readOnly(mctx))) {
+			return "", serviceerror.NewInvalidArgumentf(
+				"request id %q is the id of a workflow listening on this channel", id)
 		}
+		return id, nil
 	}
 	if err := c.checkListenerRoom(limits.MaxListeners); err != nil {
 		return "", err
 	}
-	if c.CallbackListeners == nil {
-		c.CallbackListeners = make(chasm.Map[string, *channelpb.CallbackListener])
+	if c.Listeners == nil {
+		c.Listeners = make(chasm.Map[string, *channelpb.Listener])
 	}
-	id := reg.RequestID
-	listener := &channelpb.CallbackListener{
+	listener := &channelpb.Listener{
 		ListenerId:     id,
 		Callback:       common.CloneProto(reg.Callback),
 		RequestId:      reg.RequestID,
@@ -482,7 +557,7 @@ func (c *Channel) RegisterCallbackListener(
 			return "", err
 		}
 	}
-	c.CallbackListeners[id] = chasm.NewDataField(mctx, listener)
+	c.Listeners[id] = chasm.NewDataField(mctx, listener)
 	return id, nil
 }
 
@@ -498,14 +573,12 @@ func (c *Channel) UnregisterListener(
 	if listenerID == "" {
 		return serviceerror.NewInvalidArgument("listener id is required")
 	}
-	if _, ok := c.WorkflowListeners[listenerID]; ok {
-		if _, isCallback := c.CallbackListeners[listenerID]; !isCallback {
-			return serviceerror.NewInvalidArgumentf(
-				"listener %q is a workflow; it stops listening when its run ends", listenerID)
-		}
+	if field, ok := c.Listeners[listenerID]; ok && IsWorkflowListener(field.Get(readOnly(mctx))) {
+		return serviceerror.NewInvalidArgumentf(
+			"listener %q is a workflow, which stops listening when its run ends", listenerID)
 	}
 	c.touch(mctx)
-	delete(c.CallbackListeners, listenerID)
+	delete(c.Listeners, listenerID)
 	c.scheduleIdleCheck(mctx, limits.withDefaults())
 	return nil
 }
@@ -527,13 +600,13 @@ func (c *Channel) CallbackDeliveryFor(
 	ctx chasm.Context,
 	task *channelpb.ChannelCallbackTask,
 ) (CallbackDelivery, bool) {
-	field, ok := c.CallbackListeners[task.GetListenerId()]
+	field, ok := c.Listeners[task.GetListenerId()]
 	if !ok {
 		return CallbackDelivery{}, false
 	}
 	listener := field.Get(readOnly(ctx))
-	if listener.GetInFlight() == nil || listener.GetTaskSequence() != task.GetSequence() ||
-		listener.GetNextAttemptTime() != nil {
+	if IsWorkflowListener(listener) || listener.GetInFlight() == nil ||
+		listener.GetTaskSequence() != task.GetSequence() || listener.GetNextAttemptTime() != nil {
 		return CallbackDelivery{}, false
 	}
 	return CallbackDelivery{
@@ -566,7 +639,7 @@ func (c *Channel) CompleteCallbackDelivery(
 	mctx chasm.MutableContext,
 	outcome CallbackOutcome,
 ) (bool, error) {
-	field, ok := c.CallbackListeners[outcome.ListenerID]
+	field, ok := c.Listeners[outcome.ListenerID]
 	if !ok {
 		return false, nil
 	}
@@ -619,7 +692,7 @@ func attemptFailure(err error, nonRetryable bool) *failurepb.Failure {
 
 // BackoffDone reports whether a backoff task still has an attempt to resume.
 func (c *Channel) BackoffDone(ctx chasm.Context, task *channelpb.ChannelCallbackBackoffTask) bool {
-	field, ok := c.CallbackListeners[task.GetListenerId()]
+	field, ok := c.Listeners[task.GetListenerId()]
 	if !ok {
 		return false
 	}
@@ -636,7 +709,7 @@ func (c *Channel) ResumeCallback(
 	if !c.BackoffDone(mctx, task) {
 		return nil
 	}
-	listener := c.CallbackListeners[task.GetListenerId()].Get(mctx)
+	listener := c.Listeners[task.GetListenerId()].Get(mctx)
 	listener.NextAttemptTime = nil
 	return c.scheduleCallback(mctx, listener)
 }
@@ -696,29 +769,29 @@ func (c *Channel) Describe(ctx chasm.Context, _ struct{}) (Snapshot, error) {
 		RetainedCount:    c.RetainedCount(),
 		LastActivityTime: c.State.GetLastActivityTime().AsTime(),
 	}
-	for _, id := range slices.Sorted(maps.Keys(c.WorkflowListeners)) {
-		listener := c.WorkflowListeners[id].Get(view)
-		out.Listeners = append(out.Listeners, &channelpb.ChannelListenerInfo{
-			ListenerId: id,
-			Variant: &channelpb.ChannelListenerInfo_Workflow_{
+	var workflows, callbacks []*channelpb.ChannelListenerInfo
+	for _, id := range slices.Sorted(maps.Keys(c.Listeners)) {
+		listener := c.Listeners[id].Get(view)
+		info := &channelpb.ChannelListenerInfo{
+			ListenerId:     id,
+			RegisteredTime: common.CloneProto(listener.GetRegisteredTime()),
+		}
+		if target, ok := WorkflowTargetOf(listener); ok {
+			info.Variant = &channelpb.ChannelListenerInfo_Workflow_{
 				Workflow: &channelpb.ChannelListenerInfo_Workflow{
-					WorkflowId: listener.GetWorkflowId(),
-					RunId:      listener.GetRunId(),
+					WorkflowId: target.GetWorkflowId(),
+					RunId:      target.GetRunId(),
 				},
-			},
-			RegisteredTime: common.CloneProto(listener.GetRegisteredTime()),
-		})
-	}
-	for _, id := range slices.Sorted(maps.Keys(c.CallbackListeners)) {
-		listener := c.CallbackListeners[id].Get(view)
-		out.Listeners = append(out.Listeners, &channelpb.ChannelListenerInfo{
-			ListenerId: id,
-			Variant: &channelpb.ChannelListenerInfo_Callback{
-				Callback: common.CloneProto(listener.GetCallback()),
-			},
-			RegisteredTime: common.CloneProto(listener.GetRegisteredTime()),
-		})
+			}
+			workflows = append(workflows, info)
+			continue
+		}
+		info.Variant = &channelpb.ChannelListenerInfo_Callback{
+			Callback: common.CloneProto(listener.GetCallback()),
+		}
+		callbacks = append(callbacks, info)
 	}
+	out.Listeners = append(workflows, callbacks...)
 	return out, nil
 }
 
diff --git a/chasm/lib/channel/channel_test.go b/chasm/lib/channel/channel_test.go
index 4dd2b0b3905..a36883e6b31 100644
--- a/chasm/lib/channel/channel_test.go
+++ b/chasm/lib/channel/channel_test.go
@@ -123,7 +123,7 @@ func TestBurstCoalescesIntoOneFanOut(t *testing.T) {
 	fanOut, err := c.TakeFanOut(mctx, struct{}{})
 	require.NoError(t, err)
 	require.Equal(t, int64(3), fanOut.Latest.GetCounter())
-	require.Len(t, fanOut.WorkflowListeners, 1)
+	require.Len(t, fanOut.Workflows, 1)
 	require.False(t, c.State.GetFanOutPending())
 
 	_, err = c.Notify(mctx, note(4), Limits{})
@@ -186,7 +186,7 @@ func TestCallbackFoldsWhileBusy(t *testing.T) {
 	})
 	require.NoError(t, err)
 	require.True(t, delivered)
-	require.Nil(t, c.CallbackListeners[id].Get(mctx).GetInFlight())
+	require.Nil(t, c.Listeners[id].Get(mctx).GetInFlight())
 }
 
 // A retryable failure backs off and tries the same notification again; a
@@ -228,7 +228,7 @@ func TestCallbackRetryBacksOff(t *testing.T) {
 	})
 	require.NoError(t, err)
 	require.False(t, delivered)
-	listener := c.CallbackListeners[id].Get(mctx)
+	listener := c.Listeners[id].Get(mctx)
 	require.Nil(t, listener.GetInFlight())
 	require.True(t, listener.GetLastAttemptFailure().GetApplicationFailureInfo().GetNonRetryable())
 }
@@ -250,7 +250,7 @@ func TestListenerTable(t *testing.T) {
 	})
 	require.NoError(t, err)
 	require.Equal(t, first, again)
-	listener := c.CallbackListeners[first].Get(mctx)
+	listener := c.Listeners[first].Get(mctx)
 	require.Equal(t, int64(5), listener.GetInFlight().GetCounter(), "a new listener is posted the latest")
 	require.Equal(t, int64(5), listener.GetHandedCounter())
 	require.Len(t, tasksOf[*channelpb.ChannelCallbackTask](mctx), 1, "once, not per retry")
@@ -266,7 +266,7 @@ func TestListenerTable(t *testing.T) {
 	require.NoError(t, registerWorkflow(c, mctx, WorkflowRegistration{
 		WorkflowID: "wf", RunID: "run-2", Limits: limits,
 	}), "a later run of a known workflow takes its entry, not a new one")
-	require.Equal(t, "run-2", c.WorkflowListeners["wf"].Get(mctx).GetRunId())
+	require.Equal(t, "run-2", workflowRun(t, c, mctx, "wf"))
 
 	_, err = c.RegisterCallbackListener(mctx, CallbackRegistration{
 		RequestID: "req-3", Callback: &commonpb.Callback{},
@@ -294,10 +294,10 @@ func TestWorkflowListenerRekeyAndForget(t *testing.T) {
 		WorkflowID: "wf", RunID: "run-1",
 	}))
 
-	c.RekeyWorkflowListener(mctx, "wf", "run-1", "run-2")
-	require.Equal(t, "run-2", c.WorkflowListeners["wf"].Get(mctx).GetRunId())
-	c.RekeyWorkflowListener(mctx, "wf", "run-1", "run-3")
-	require.Equal(t, "run-2", c.WorkflowListeners["wf"].Get(mctx).GetRunId(), "stale re-key")
+	require.NoError(t, c.RekeyWorkflowListener(mctx, "wf", "run-1", "run-2"))
+	require.Equal(t, "run-2", workflowRun(t, c, mctx, "wf"))
+	require.NoError(t, c.RekeyWorkflowListener(mctx, "wf", "run-1", "run-3"))
+	require.Equal(t, "run-2", workflowRun(t, c, mctx, "wf"), "stale re-key")
 
 	c.ForgetWorkflowListener(mctx, "wf", "run-1", Limits{})
 	require.Equal(t, 1, c.ListenerCount(), "stale forget")
@@ -361,6 +361,66 @@ func registerWorkflow(c *Channel, mctx chasm.MutableContext, reg WorkflowRegistr
 	return err
 }
 
+// workflowRun is the run the channel has on record for a workflow id.
+func workflowRun(t *testing.T, c *Channel, ctx chasm.Context, workflowID string) string {
+	t.Helper()
+	field, ok := c.Listeners[workflowID]
+	require.True(t, ok, "no listener %q", workflowID)
+	target, ok := WorkflowTargetOf(field.Get(ctx))
+	require.True(t, ok, "listener %q is not a workflow", workflowID)
+	return target.GetRunId()
+}
+
+// A workflow listener is an internal callback whose data names the run, held
+// in the one table next to the HTTP callbacks. The two kinds share the id
+// space, so neither can take an id the other holds.
+func TestWorkflowListenerIsAnInternalCallback(t *testing.T) {
+	c, mctx, _ := newTestChannel(t)
+	_, err := c.RegisterWorkflowListener(mctx, WorkflowRegistration{
+		NamespaceID: "ns", WorkflowID: "wf", RunID: "run-1", FirstExecutionRunID: "run-0",
+	})
+	require.NoError(t, err)
+	listener := c.Listeners["wf"].Get(mctx)
+	require.True(t, IsWorkflowListener(listener))
+	require.Equal(t, ListenerKindWorkflow, ListenerKind(listener.GetCallback()))
+	require.NotEmpty(t, listener.GetCallback().GetInternal().GetData())
+	require.Nil(t, listener.GetInFlight(), "a run folds on its own shard, not here")
+	target, ok := WorkflowTargetOf(listener)
+	require.True(t, ok)
+	require.Equal(t, "ns", target.GetNamespaceId())
+	require.Equal(t, "wf", target.GetWorkflowId())
+	require.Equal(t, "run-1", target.GetRunId())
+	require.Equal(t, "run-0", target.GetFirstExecutionRunId())
+
+	var invalid *serviceerror.InvalidArgument
+	_, err = c.RegisterCallbackListener(mctx, CallbackRegistration{
+		RequestID: "wf", Callback: testCallback(),
+	})
+	require.ErrorAs(t, err, &invalid, "a callback cannot take a workflow's id")
+	id, err := c.RegisterCallbackListener(mctx, CallbackRegistration{
+		RequestID: "cb", Callback: testCallback(),
+	})
+	require.NoError(t, err)
+	callback := c.Listeners[id].Get(mctx)
+	require.False(t, IsWorkflowListener(callback))
+	require.Equal(t, ListenerKindCallback, ListenerKind(callback.GetCallback()))
+	_, ok = WorkflowTargetOf(callback)
+	require.False(t, ok)
+	_, err = c.RegisterWorkflowListener(mctx, WorkflowRegistration{WorkflowID: "cb", RunID: "run-1"})
+	require.ErrorAs(t, err, &invalid, "a workflow cannot take a callback's id")
+
+	require.Equal(t, 2, c.ListenerCount())
+	targets := c.WorkflowTargets(mctx)
+	require.Len(t, targets, 1)
+	require.Equal(t, "run-1", targets[0].GetRunId())
+	snapshot, err := c.Describe(mctx, struct{}{})
+	require.NoError(t, err)
+	require.Len(t, snapshot.Listeners, 2)
+	require.Equal(t, "run-1", snapshot.Listeners[0].GetWorkflow().GetRunId(), "workflows first")
+	require.Equal(t, testCallback().GetNexus().GetUrl(),
+		snapshot.Listeners[1].GetCallback().GetNexus().GetUrl())
+}
+
 // A run new to the channel is handed the latest notification, so a write that
 // landed before it subscribed still wakes it; a repeat registration and an
 // empty channel hand nothing.
@@ -411,5 +471,5 @@ func TestRepeatHandsOnlyToCallbacksThatLackIt(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 1, started)
 	require.Len(t, tasksOf[*channelpb.ChannelCallbackTask](mctx), 2)
-	require.Empty(t, c.WorkflowListenerList(mctx))
+	require.Empty(t, c.WorkflowTargets(mctx))
 }
diff --git a/chasm/lib/channel/gen/channelpb/v1/message.go-helpers.pb.go b/chasm/lib/channel/gen/channelpb/v1/message.go-helpers.pb.go
index f5089f2e084..3332a2c00e2 100644
--- a/chasm/lib/channel/gen/channelpb/v1/message.go-helpers.pb.go
+++ b/chasm/lib/channel/gen/channelpb/v1/message.go-helpers.pb.go
@@ -42,35 +42,35 @@ func (this *Notification) Equal(that interface{}) bool {
 	return proto.Equal(this, that1)
 }
 
-// Marshal an object of type WorkflowListener to the protobuf v3 wire format
-func (val *WorkflowListener) Marshal() ([]byte, error) {
+// Marshal an object of type Listener to the protobuf v3 wire format
+func (val *Listener) Marshal() ([]byte, error) {
 	return proto.Marshal(val)
 }
 
-// Unmarshal an object of type WorkflowListener from the protobuf v3 wire format
-func (val *WorkflowListener) Unmarshal(buf []byte) error {
+// Unmarshal an object of type Listener from the protobuf v3 wire format
+func (val *Listener) Unmarshal(buf []byte) error {
 	return proto.Unmarshal(buf, val)
 }
 
 // Size returns the size of the object, in bytes, once serialized
-func (val *WorkflowListener) Size() int {
+func (val *Listener) Size() int {
 	return proto.Size(val)
 }
 
-// Equal returns whether two WorkflowListener values are equivalent by recursively
+// Equal returns whether two Listener values are equivalent by recursively
 // comparing the message's fields.
 // For more information see the documentation for
 // https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
-func (this *WorkflowListener) Equal(that interface{}) bool {
+func (this *Listener) Equal(that interface{}) bool {
 	if that == nil {
 		return this == nil
 	}
 
-	var that1 *WorkflowListener
+	var that1 *Listener
 	switch t := that.(type) {
-	case *WorkflowListener:
+	case *Listener:
 		that1 = t
-	case WorkflowListener:
+	case Listener:
 		that1 = &t
 	default:
 		return false
@@ -79,35 +79,35 @@ func (this *WorkflowListener) Equal(that interface{}) bool {
 	return proto.Equal(this, that1)
 }
 
-// Marshal an object of type CallbackListener to the protobuf v3 wire format
-func (val *CallbackListener) Marshal() ([]byte, error) {
+// Marshal an object of type WorkflowTarget to the protobuf v3 wire format
+func (val *WorkflowTarget) Marshal() ([]byte, error) {
 	return proto.Marshal(val)
 }
 
-// Unmarshal an object of type CallbackListener from the protobuf v3 wire format
-func (val *CallbackListener) Unmarshal(buf []byte) error {
+// Unmarshal an object of type WorkflowTarget from the protobuf v3 wire format
+func (val *WorkflowTarget) Unmarshal(buf []byte) error {
 	return proto.Unmarshal(buf, val)
 }
 
 // Size returns the size of the object, in bytes, once serialized
-func (val *CallbackListener) Size() int {
+func (val *WorkflowTarget) Size() int {
 	return proto.Size(val)
 }
 
-// Equal returns whether two CallbackListener values are equivalent by recursively
+// Equal returns whether two WorkflowTarget values are equivalent by recursively
 // comparing the message's fields.
 // For more information see the documentation for
 // https://pkg.go.dev/google.golang.org/protobuf/proto#Equal
-func (this *CallbackListener) Equal(that interface{}) bool {
+func (this *WorkflowTarget) Equal(that interface{}) bool {
 	if that == nil {
 		return this == nil
 	}
 
-	var that1 *CallbackListener
+	var that1 *WorkflowTarget
 	switch t := that.(type) {
-	case *CallbackListener:
+	case *WorkflowTarget:
 		that1 = t
-	case CallbackListener:
+	case WorkflowTarget:
 		that1 = &t
 	default:
 		return false
diff --git a/chasm/lib/channel/gen/channelpb/v1/message.pb.go b/chasm/lib/channel/gen/channelpb/v1/message.pb.go
index bb43ce83f22..1aab63bd334 100644
--- a/chasm/lib/channel/gen/channelpb/v1/message.pb.go
+++ b/chasm/lib/channel/gen/channelpb/v1/message.pb.go
@@ -108,83 +108,16 @@ func (x *Notification) GetLinkedTo() *v1.WorkflowExecution {
 	return nil
 }
 
-// WorkflowListener is a workflow run that subscribed to the channel.
-type WorkflowListener struct {
-	state      protoimpl.MessageState `protogen:"open.v1"`
-	WorkflowId string                 `protobuf:"bytes,1,opt,name=workflow_id,json=workflowId,proto3" json:"workflow_id,omitempty"`
-	// The run that subscribed, or the one the subscription was re-keyed to.
-	RunId               string                 `protobuf:"bytes,2,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"`
-	FirstExecutionRunId string                 `protobuf:"bytes,3,opt,name=first_execution_run_id,json=firstExecutionRunId,proto3" json:"first_execution_run_id,omitempty"`
-	RegisteredTime      *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=registered_time,json=registeredTime,proto3" json:"registered_time,omitempty"`
-	unknownFields       protoimpl.UnknownFields
-	sizeCache           protoimpl.SizeCache
-}
-
-func (x *WorkflowListener) Reset() {
-	*x = WorkflowListener{}
-	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[1]
-	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-	ms.StoreMessageInfo(mi)
-}
-
-func (x *WorkflowListener) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*WorkflowListener) ProtoMessage() {}
-
-func (x *WorkflowListener) ProtoReflect() protoreflect.Message {
-	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[1]
-	if x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
-	}
-	return mi.MessageOf(x)
-}
-
-// Deprecated: Use WorkflowListener.ProtoReflect.Descriptor instead.
-func (*WorkflowListener) Descriptor() ([]byte, []int) {
-	return file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDescGZIP(), []int{1}
-}
-
-func (x *WorkflowListener) GetWorkflowId() string {
-	if x != nil {
-		return x.WorkflowId
-	}
-	return ""
-}
-
-func (x *WorkflowListener) GetRunId() string {
-	if x != nil {
-		return x.RunId
-	}
-	return ""
-}
-
-func (x *WorkflowListener) GetFirstExecutionRunId() string {
-	if x != nil {
-		return x.FirstExecutionRunId
-	}
-	return ""
-}
-
-func (x *WorkflowListener) GetRegisteredTime() *timestamppb.Timestamp {
-	if x != nil {
-		return x.RegisteredTime
-	}
-	return nil
-}
-
-// CallbackListener is an HTTP endpoint the channel posts its notifications to.
-type CallbackListener struct {
+// Listener is one entry of the channel's listener table. The callback says
+// where a notification goes: a Nexus callback is an HTTP endpoint the channel
+// posts to, and an internal callback carries a WorkflowTarget and reaches the
+// run on its own shard through the routed delivery.
+type Listener struct {
 	state      protoimpl.MessageState `protogen:"open.v1"`
 	ListenerId string                 `protobuf:"bytes,1,opt,name=listener_id,json=listenerId,proto3" json:"listener_id,omitempty"`
 	Callback   *v1.Callback           `protobuf:"bytes,2,opt,name=callback,proto3" json:"callback,omitempty"`
-	// Request id of the registration, so a retried registration finds this
-	// listener rather than adding another.
+	// Request id of a callback registration, so a retried registration finds
+	// this listener rather than adding another. Empty for a workflow.
 	RequestId      string                 `protobuf:"bytes,3,opt,name=request_id,json=requestId,proto3" json:"request_id,omitempty"`
 	RegisteredTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=registered_time,json=registeredTime,proto3" json:"registered_time,omitempty"`
 	// The notification being delivered, if any. One delivery is in flight at a
@@ -207,21 +140,21 @@ type CallbackListener struct {
 	sizeCache     protoimpl.SizeCache
 }
 
-func (x *CallbackListener) Reset() {
-	*x = CallbackListener{}
-	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[2]
+func (x *Listener) Reset() {
+	*x = Listener{}
+	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[1]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
 
-func (x *CallbackListener) String() string {
+func (x *Listener) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*CallbackListener) ProtoMessage() {}
+func (*Listener) ProtoMessage() {}
 
-func (x *CallbackListener) ProtoReflect() protoreflect.Message {
-	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[2]
+func (x *Listener) ProtoReflect() protoreflect.Message {
+	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[1]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -232,88 +165,159 @@ func (x *CallbackListener) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use CallbackListener.ProtoReflect.Descriptor instead.
-func (*CallbackListener) Descriptor() ([]byte, []int) {
-	return file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDescGZIP(), []int{2}
+// Deprecated: Use Listener.ProtoReflect.Descriptor instead.
+func (*Listener) Descriptor() ([]byte, []int) {
+	return file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDescGZIP(), []int{1}
 }
 
-func (x *CallbackListener) GetListenerId() string {
+func (x *Listener) GetListenerId() string {
 	if x != nil {
 		return x.ListenerId
 	}
 	return ""
 }
 
-func (x *CallbackListener) GetCallback() *v1.Callback {
+func (x *Listener) GetCallback() *v1.Callback {
 	if x != nil {
 		return x.Callback
 	}
 	return nil
 }
 
-func (x *CallbackListener) GetRequestId() string {
+func (x *Listener) GetRequestId() string {
 	if x != nil {
 		return x.RequestId
 	}
 	return ""
 }
 
-func (x *CallbackListener) GetRegisteredTime() *timestamppb.Timestamp {
+func (x *Listener) GetRegisteredTime() *timestamppb.Timestamp {
 	if x != nil {
 		return x.RegisteredTime
 	}
 	return nil
 }
 
-func (x *CallbackListener) GetInFlight() *Notification {
+func (x *Listener) GetInFlight() *Notification {
 	if x != nil {
 		return x.InFlight
 	}
 	return nil
 }
 
-func (x *CallbackListener) GetPending() *Notification {
+func (x *Listener) GetPending() *Notification {
 	if x != nil {
 		return x.Pending
 	}
 	return nil
 }
 
-func (x *CallbackListener) GetHandedCounter() int64 {
+func (x *Listener) GetHandedCounter() int64 {
 	if x != nil {
 		return x.HandedCounter
 	}
 	return 0
 }
 
-func (x *CallbackListener) GetAttempt() int32 {
+func (x *Listener) GetAttempt() int32 {
 	if x != nil {
 		return x.Attempt
 	}
 	return 0
 }
 
-func (x *CallbackListener) GetNextAttemptTime() *timestamppb.Timestamp {
+func (x *Listener) GetNextAttemptTime() *timestamppb.Timestamp {
 	if x != nil {
 		return x.NextAttemptTime
 	}
 	return nil
 }
 
-func (x *CallbackListener) GetLastAttemptFailure() *v11.Failure {
+func (x *Listener) GetLastAttemptFailure() *v11.Failure {
 	if x != nil {
 		return x.LastAttemptFailure
 	}
 	return nil
 }
 
-func (x *CallbackListener) GetTaskSequence() int64 {
+func (x *Listener) GetTaskSequence() int64 {
 	if x != nil {
 		return x.TaskSequence
 	}
 	return 0
 }
 
+// WorkflowTarget is what an internal callback's data holds: the workflow run
+// a notification is routed to.
+type WorkflowTarget struct {
+	state       protoimpl.MessageState `protogen:"open.v1"`
+	NamespaceId string                 `protobuf:"bytes,1,opt,name=namespace_id,json=namespaceId,proto3" json:"namespace_id,omitempty"`
+	WorkflowId  string                 `protobuf:"bytes,2,opt,name=workflow_id,json=workflowId,proto3" json:"workflow_id,omitempty"`
+	// The run that subscribed, or the one the subscription was re-keyed to.
+	RunId               string `protobuf:"bytes,3,opt,name=run_id,json=runId,proto3" json:"run_id,omitempty"`
+	FirstExecutionRunId string `protobuf:"bytes,4,opt,name=first_execution_run_id,json=firstExecutionRunId,proto3" json:"first_execution_run_id,omitempty"`
+	unknownFields       protoimpl.UnknownFields
+	sizeCache           protoimpl.SizeCache
+}
+
+func (x *WorkflowTarget) Reset() {
+	*x = WorkflowTarget{}
+	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[2]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *WorkflowTarget) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*WorkflowTarget) ProtoMessage() {}
+
+func (x *WorkflowTarget) ProtoReflect() protoreflect.Message {
+	mi := &file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes[2]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use WorkflowTarget.ProtoReflect.Descriptor instead.
+func (*WorkflowTarget) Descriptor() ([]byte, []int) {
+	return file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDescGZIP(), []int{2}
+}
+
+func (x *WorkflowTarget) GetNamespaceId() string {
+	if x != nil {
+		return x.NamespaceId
+	}
+	return ""
+}
+
+func (x *WorkflowTarget) GetWorkflowId() string {
+	if x != nil {
+		return x.WorkflowId
+	}
+	return ""
+}
+
+func (x *WorkflowTarget) GetRunId() string {
+	if x != nil {
+		return x.RunId
+	}
+	return ""
+}
+
+func (x *WorkflowTarget) GetFirstExecutionRunId() string {
+	if x != nil {
+		return x.FirstExecutionRunId
+	}
+	return ""
+}
+
 var File_temporal_server_chasm_lib_channel_proto_v1_message_proto protoreflect.FileDescriptor
 
 const file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDesc = "" +
@@ -327,14 +331,8 @@ const file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDesc = ""
 	"\tlinked_to\x18\x05 \x01(\v2).temporal.api.common.v1.WorkflowExecutionR\blinkedTo\x1a\\\n" +
 	"\rMetadataEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x125\n" +
-	"\x05value\x18\x02 \x01(\v2\x1f.temporal.api.common.v1.PayloadR\x05value:\x028\x01\"\xc4\x01\n" +
-	"\x10WorkflowListener\x12\x1f\n" +
-	"\vworkflow_id\x18\x01 \x01(\tR\n" +
-	"workflowId\x12\x15\n" +
-	"\x06run_id\x18\x02 \x01(\tR\x05runId\x123\n" +
-	"\x16first_execution_run_id\x18\x03 \x01(\tR\x13firstExecutionRunId\x12C\n" +
-	"\x0fregistered_time\x18\x04 \x01(\v2\x1a.google.protobuf.TimestampR\x0eregisteredTime\"\x82\x05\n" +
-	"\x10CallbackListener\x12\x1f\n" +
+	"\x05value\x18\x02 \x01(\v2\x1f.temporal.api.common.v1.PayloadR\x05value:\x028\x01\"\xfa\x04\n" +
+	"\bListener\x12\x1f\n" +
 	"\vlistener_id\x18\x01 \x01(\tR\n" +
 	"listenerId\x12<\n" +
 	"\bcallback\x18\x02 \x01(\v2 .temporal.api.common.v1.CallbackR\bcallback\x12\x1d\n" +
@@ -348,7 +346,13 @@ const file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDesc = ""
 	"\x11next_attempt_time\x18\t \x01(\v2\x1a.google.protobuf.TimestampR\x0fnextAttemptTime\x12R\n" +
 	"\x14last_attempt_failure\x18\n" +
 	" \x01(\v2 .temporal.api.failure.v1.FailureR\x12lastAttemptFailure\x12#\n" +
-	"\rtask_sequence\x18\v \x01(\x03R\ftaskSequenceBAZ?go.temporal.io/server/chasm/lib/channel/gen/channelpb;channelpbb\x06proto3"
+	"\rtask_sequence\x18\v \x01(\x03R\ftaskSequence\"\xa0\x01\n" +
+	"\x0eWorkflowTarget\x12!\n" +
+	"\fnamespace_id\x18\x01 \x01(\tR\vnamespaceId\x12\x1f\n" +
+	"\vworkflow_id\x18\x02 \x01(\tR\n" +
+	"workflowId\x12\x15\n" +
+	"\x06run_id\x18\x03 \x01(\tR\x05runId\x123\n" +
+	"\x16first_execution_run_id\x18\x04 \x01(\tR\x13firstExecutionRunIdBAZ?go.temporal.io/server/chasm/lib/channel/gen/channelpb;channelpbb\x06proto3"
 
 var (
 	file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDescOnce sync.Once
@@ -365,31 +369,30 @@ func file_temporal_server_chasm_lib_channel_proto_v1_message_proto_rawDescGZIP()
 var file_temporal_server_chasm_lib_channel_proto_v1_message_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
 var file_temporal_server_chasm_lib_channel_proto_v1_message_proto_goTypes = []any{
 	(*Notification)(nil),          // 0: temporal.server.chasm.lib.channel.proto.v1.Notification
-	(*WorkflowListener)(nil),      // 1: temporal.server.chasm.lib.channel.proto.v1.WorkflowListener
-	(*CallbackListener)(nil),      // 2: temporal.server.chasm.lib.channel.proto.v1.CallbackListener
+	(*Listener)(nil),              // 1: temporal.server.chasm.lib.channel.proto.v1.Listener
+	(*WorkflowTarget)(nil),        // 2: temporal.server.chasm.lib.channel.proto.v1.WorkflowTarget
 	nil,                           // 3: temporal.server.chasm.lib.channel.proto.v1.Notification.MetadataEntry
 	(*v1.WorkflowExecution)(nil),  // 4: temporal.api.common.v1.WorkflowExecution
-	(*timestamppb.Timestamp)(nil), // 5: google.protobuf.Timestamp
-	(*v1.Callback)(nil),           // 6: temporal.api.common.v1.Callback
+	(*v1.Callback)(nil),           // 5: temporal.api.common.v1.Callback
+	(*timestamppb.Timestamp)(nil), // 6: google.protobuf.Timestamp
 	(*v11.Failure)(nil),           // 7: temporal.api.failure.v1.Failure
 	(*v1.Payload)(nil),            // 8: temporal.api.common.v1.Payload
 }
 var file_temporal_server_chasm_lib_channel_proto_v1_message_proto_depIdxs = []int32{
-	3,  // 0: temporal.server.chasm.lib.channel.proto.v1.Notification.metadata:type_name -> temporal.server.chasm.lib.channel.proto.v1.Notification.MetadataEntry
-	4,  // 1: temporal.server.chasm.lib.channel.proto.v1.Notification.linked_to:type_name -> temporal.api.common.v1.WorkflowExecution
-	5,  // 2: temporal.server.chasm.lib.channel.proto.v1.WorkflowListener.registered_time:type_name -> google.protobuf.Timestamp
-	6,  // 3: temporal.server.chasm.lib.channel.proto.v1.CallbackListener.callback:type_name -> temporal.api.common.v1.Callback
-	5,  // 4: temporal.server.chasm.lib.channel.proto.v1.CallbackListener.registered_time:type_name -> google.protobuf.Timestamp
-	0,  // 5: temporal.server.chasm.lib.channel.proto.v1.CallbackListener.in_flight:type_name -> temporal.server.chasm.lib.channel.proto.v1.Notification
-	0,  // 6: temporal.server.chasm.lib.channel.proto.v1.CallbackListener.pending:type_name -> temporal.server.chasm.lib.channel.proto.v1.Notification
-	5,  // 7: temporal.server.chasm.lib.channel.proto.v1.CallbackListener.next_attempt_time:type_name -> google.protobuf.Timestamp
-	7,  // 8: temporal.server.chasm.lib.channel.proto.v1.CallbackListener.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure
-	8,  // 9: temporal.server.chasm.lib.channel.proto.v1.Notification.MetadataEntry.value:type_name -> temporal.api.common.v1.Payload
-	10, // [10:10] is the sub-list for method output_type
-	10, // [10:10] is the sub-list for method input_type
-	10, // [10:10] is the sub-list for extension type_name
-	10, // [10:10] is the sub-list for extension extendee
-	0,  // [0:10] is the sub-list for field type_name
+	3, // 0: temporal.server.chasm.lib.channel.proto.v1.Notification.metadata:type_name -> temporal.server.chasm.lib.channel.proto.v1.Notification.MetadataEntry
+	4, // 1: temporal.server.chasm.lib.channel.proto.v1.Notification.linked_to:type_name -> temporal.api.common.v1.WorkflowExecution
+	5, // 2: temporal.server.chasm.lib.channel.proto.v1.Listener.callback:type_name -> temporal.api.common.v1.Callback
+	6, // 3: temporal.server.chasm.lib.channel.proto.v1.Listener.registered_time:type_name -> google.protobuf.Timestamp
+	0, // 4: temporal.server.chasm.lib.channel.proto.v1.Listener.in_flight:type_name -> temporal.server.chasm.lib.channel.proto.v1.Notification
+	0, // 5: temporal.server.chasm.lib.channel.proto.v1.Listener.pending:type_name -> temporal.server.chasm.lib.channel.proto.v1.Notification
+	6, // 6: temporal.server.chasm.lib.channel.proto.v1.Listener.next_attempt_time:type_name -> google.protobuf.Timestamp
+	7, // 7: temporal.server.chasm.lib.channel.proto.v1.Listener.last_attempt_failure:type_name -> temporal.api.failure.v1.Failure
+	8, // 8: temporal.server.chasm.lib.channel.proto.v1.Notification.MetadataEntry.value:type_name -> temporal.api.common.v1.Payload
+	9, // [9:9] is the sub-list for method output_type
+	9, // [9:9] is the sub-list for method input_type
+	9, // [9:9] is the sub-list for extension type_name
+	9, // [9:9] is the sub-list for extension extendee
+	0, // [0:9] is the sub-list for field type_name
 }
 
 func init() { file_temporal_server_chasm_lib_channel_proto_v1_message_proto_init() }
diff --git a/chasm/lib/channel/linked.go b/chasm/lib/channel/linked.go
index a975ebfdb06..b6db2b1f836 100644
--- a/chasm/lib/channel/linked.go
+++ b/chasm/lib/channel/linked.go
@@ -114,9 +114,11 @@ func (c *Channel) LinkedTo(ctx chasm.Context) *commonpb.WorkflowExecution {
 	return &commonpb.WorkflowExecution{WorkflowId: key.BusinessID, RunId: key.RunID}
 }
 
-// LinkedListenerCount counts the owner and the callback listeners.
+// LinkedListenerCount counts the owner and the callback listeners. The owner
+// is the one workflow listening and is not in the table, so the table holds
+// callbacks only.
 func (c *Channel) LinkedListenerCount() int {
-	return 1 + len(c.CallbackListeners)
+	return 1 + len(c.Listeners)
 }
 
 // OwnerHolds reports whether the owner already has a notification at this
@@ -185,7 +187,7 @@ func (c *Channel) OwnerStanding() OwnerStanding {
 		LastCounter:      c.LatestCounter(),
 		Pending:          common.CloneProto(c.State.GetOwnerPending()),
 		ScheduledCounter: c.State.GetOwnerScheduledCounter(),
-		ListenerCount:    len(c.CallbackListeners),
+		ListenerCount:    len(c.Listeners),
 		RetainedCount:    c.RetainedCount(),
 		AcceptedCount:    c.State.GetAcceptedCount(),
 	}
diff --git a/chasm/lib/channel/linked_test.go b/chasm/lib/channel/linked_test.go
index d65f1c8992c..360ef533ce3 100644
--- a/chasm/lib/channel/linked_test.go
+++ b/chasm/lib/channel/linked_test.go
@@ -150,11 +150,11 @@ func TestLinkedCallbackListeners(t *testing.T) {
 	result, err = c.NotifyLinked(mctx, note(2), Limits{})
 	require.NoError(t, err)
 	require.Equal(t, 0, result.CallbackStarted, "busy, so it folds into pending")
-	require.Equal(t, int64(2), c.CallbackListeners[id].Get(mctx).GetPending().GetCounter())
+	require.Equal(t, int64(2), c.Listeners[id].Get(mctx).GetPending().GetCounter())
 
 	late, err := c.RegisterCallbackListener(mctx, CallbackRegistration{RequestID: "r2", Callback: testCallback()})
 	require.NoError(t, err)
-	require.Equal(t, int64(2), c.CallbackListeners[late].Get(mctx).GetInFlight().GetCounter(),
+	require.Equal(t, int64(2), c.Listeners[late].Get(mctx).GetInFlight().GetCounter(),
 		"handed the latest on registration")
 	require.Equal(t, 3, c.LinkedListenerCount())
 
diff --git a/chasm/lib/channel/proto/v1/message.proto b/chasm/lib/channel/proto/v1/message.proto
index edfda374254..92c95e92f14 100644
--- a/chasm/lib/channel/proto/v1/message.proto
+++ b/chasm/lib/channel/proto/v1/message.proto
@@ -23,24 +23,21 @@ message Notification {
   temporal.api.common.v1.WorkflowExecution linked_to = 5;
 }
 
-// WorkflowListener is a workflow run that subscribed to the channel.
-message WorkflowListener {
-  string workflow_id = 1;
-  // The run that subscribed, or the one the subscription was re-keyed to.
-  string run_id = 2;
-  string first_execution_run_id = 3;
-  google.protobuf.Timestamp registered_time = 4;
-}
-
-// CallbackListener is an HTTP endpoint the channel posts its notifications to.
-message CallbackListener {
+// Listener is one entry of the channel's listener table. The callback says
+// where a notification goes: a Nexus callback is an HTTP endpoint the channel
+// posts to, and an internal callback carries a WorkflowTarget and reaches the
+// run on its own shard through the routed delivery.
+message Listener {
   string listener_id = 1;
   temporal.api.common.v1.Callback callback = 2;
-  // Request id of the registration, so a retried registration finds this
-  // listener rather than adding another.
+  // Request id of a callback registration, so a retried registration finds
+  // this listener rather than adding another. Empty for a workflow.
   string request_id = 3;
   google.protobuf.Timestamp registered_time = 4;
 
+  // The fields below track an HTTP post. A workflow listener keeps none of
+  // them: its run folds what it is handed on its own shard.
+
   // The notification being delivered, if any. One delivery is in flight at a
   // time per listener.
   Notification in_flight = 5;
@@ -58,3 +55,13 @@ message CallbackListener {
   // one delivery and a stale one finds nothing to do.
   int64 task_sequence = 11;
 }
+
+// WorkflowTarget is what an internal callback's data holds: the workflow run
+// a notification is routed to.
+message WorkflowTarget {
+  string namespace_id = 1;
+  string workflow_id = 2;
+  // The run that subscribed, or the one the subscription was re-keyed to.
+  string run_id = 3;
+  string first_execution_run_id = 4;
+}
diff --git a/chasm/lib/channel/service/deliver.go b/chasm/lib/channel/service/deliver.go
index bff2e3b2868..0eb134069f0 100644
--- a/chasm/lib/channel/service/deliver.go
+++ b/chasm/lib/channel/service/deliver.go
@@ -38,14 +38,14 @@ func (h *workflowDeliverer) deliverAll(
 	ctx context.Context,
 	ref chasm.ComponentRef,
 	ns string,
-	listeners []*channelpb.WorkflowListener,
+	targets []*channelpb.WorkflowTarget,
 	n *channelpb.Notification,
 ) error {
 	var group errgroup.Group
 	group.SetLimit(fanOutConcurrency)
-	for _, listener := range listeners {
+	for _, target := range targets {
 		group.Go(func() error {
-			return h.deliverOne(ctx, ref, ns, listener, n)
+			return h.deliverOne(ctx, ref, ns, target, n)
 		})
 	}
 	return group.Wait()
@@ -58,17 +58,17 @@ func (h *workflowDeliverer) deliverOne(
 	ctx context.Context,
 	ref chasm.ComponentRef,
 	ns string,
-	listener *channelpb.WorkflowListener,
+	target *channelpb.WorkflowTarget,
 	n *channelpb.Notification,
 ) error {
 	callCtx, cancel := context.WithTimeout(ctx, channel.RoutedCallTimeout)
 	defer cancel()
 	response, err := h.routed.DeliverChannelNotification(callCtx, &channelpb.DeliverChannelNotificationRequest{
-		NamespaceId: ref.NamespaceID,
+		NamespaceId: target.GetNamespaceId(),
 		FrontendRequest: &channelpb.DeliverChannelNotificationInput{
 			Namespace:    ns,
-			WorkflowId:   listener.GetWorkflowId(),
-			RunId:        listener.GetRunId(),
+			WorkflowId:   target.GetWorkflowId(),
+			RunId:        target.GetRunId(),
 			Notification: n,
 		},
 	})
@@ -78,7 +78,7 @@ func (h *workflowDeliverer) deliverOne(
 		// the transport's, and dropping a listener on that would be a guess.
 		h.logger.Warn("failed to deliver a channel notification to a workflow",
 			tag.NewStringTag("channel", ref.BusinessID),
-			tag.WorkflowID(listener.GetWorkflowId()),
+			tag.WorkflowID(target.GetWorkflowId()),
 			tag.Error(err))
 		return err
 	}
@@ -87,16 +87,16 @@ func (h *workflowDeliverer) deliverOne(
 	switch {
 	case out.GetListenerClosed():
 		_, _, err = chasm.UpdateComponent(ctx, ref,
-			func(c *channel.Channel, mctx chasm.MutableContext, l *channelpb.WorkflowListener) (struct{}, error) {
-				c.ForgetWorkflowListener(mctx, l.GetWorkflowId(), l.GetRunId(), limits)
+			func(c *channel.Channel, mctx chasm.MutableContext, _ struct{}) (struct{}, error) {
+				c.ForgetWorkflowListener(mctx, target.GetWorkflowId(), target.GetRunId(), limits)
 				return struct{}{}, nil
-			}, listener)
+			}, struct{}{})
 	case out.GetSuccessorRunId() != "":
 		_, _, err = chasm.UpdateComponent(ctx, ref,
-			func(c *channel.Channel, mctx chasm.MutableContext, l *channelpb.WorkflowListener) (struct{}, error) {
-				c.RekeyWorkflowListener(mctx, l.GetWorkflowId(), l.GetRunId(), out.GetSuccessorRunId())
-				return struct{}{}, nil
-			}, listener)
+			func(c *channel.Channel, mctx chasm.MutableContext, _ struct{}) (struct{}, error) {
+				return struct{}{}, c.RekeyWorkflowListener(
+					mctx, target.GetWorkflowId(), target.GetRunId(), out.GetSuccessorRunId())
+			}, struct{}{})
 	default:
 		// Delivered, or a duplicate. The listener stays as it is.
 	}
diff --git a/chasm/lib/channel/service/handler.go b/chasm/lib/channel/service/handler.go
index dcbafccf148..233e83e2ea6 100644
--- a/chasm/lib/channel/service/handler.go
+++ b/chasm/lib/channel/service/handler.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"time"
 
+	commonpb "go.temporal.io/api/common/v1"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/channel"
@@ -206,7 +207,7 @@ type repeatRead struct {
 	// it is.
 	repeat    bool
 	listeners int
-	workflows []*channelpb.WorkflowListener
+	workflows []*channelpb.WorkflowTarget
 	// Some callback listener does not hold the notification.
 	callbacksNeed bool
 }
@@ -227,7 +228,7 @@ func (h *handler) readRepeat(
 			return repeatRead{
 				repeat:        true,
 				listeners:     c.ListenerCount(),
-				workflows:     c.WorkflowListenerList(cctx),
+				workflows:     c.WorkflowTargets(cctx),
 				callbacksNeed: c.CallbacksNeed(cctx, counter),
 			}, nil
 		}, n.GetCounter())
@@ -516,6 +517,7 @@ func (h *handler) RegisterWorkflowListener(
 		return nil, err
 	}
 	reg := channel.WorkflowRegistration{
+		NamespaceID:         req.GetNamespaceId(),
 		WorkflowID:          in.GetWorkflowId(),
 		RunID:               in.GetRunId(),
 		FirstExecutionRunID: in.GetFirstExecutionRunId(),
@@ -670,6 +672,11 @@ func (h *handler) DeliverChannelNotification(
 }
 
 var (
-	workflowKindTag = metrics.StringTag("listener_kind", "workflow")
-	callbackKindTag = metrics.StringTag("listener_kind", "callback")
+	workflowKindTag = metrics.StringTag("listener_kind", channel.ListenerKindWorkflow)
+	callbackKindTag = metrics.StringTag("listener_kind", channel.ListenerKindCallback)
 )
+
+// listenerKindTag derives the kind tag from the listener's callback variant.
+func listenerKindTag(cb *commonpb.Callback) metrics.Tag {
+	return metrics.StringTag("listener_kind", channel.ListenerKind(cb))
+}
diff --git a/chasm/lib/channel/service/tasks.go b/chasm/lib/channel/service/tasks.go
index 7bc23eeac14..414a65c5863 100644
--- a/chasm/lib/channel/service/tasks.go
+++ b/chasm/lib/channel/service/tasks.go
@@ -105,7 +105,7 @@ func (h *fanOutTaskHandler) Execute(
 		return nil
 	}
 
-	return h.deliverer.deliverAll(ctx, ref, ns, fanOut.WorkflowListeners, fanOut.Latest)
+	return h.deliverer.deliverAll(ctx, ref, ns, fanOut.Workflows, fanOut.Latest)
 }
 
 // Discard lowers the coalescing flag the scheduling notify raised. Left up, no
@@ -217,7 +217,7 @@ func (h *callbackTaskHandler) Execute(
 			kind = linkedKindTag
 		}
 		metrics.ChannelNotificationsDelivered.With(h.metricsHandler).Record(
-			1, metrics.NamespaceTag(ns), kind, callbackKindTag)
+			1, metrics.NamespaceTag(ns), kind, listenerKindTag(delivery.Callback))
 	}
 	if postErr != nil && retryable {
 		// Reported to the queue as the destination being down, which feeds its
diff --git a/tests/notification_channel_test.go b/tests/notification_channel_test.go
index a0736b90e11..c9b415e823e 100644
--- a/tests/notification_channel_test.go
+++ b/tests/notification_channel_test.go
@@ -282,9 +282,9 @@ func (c *channelTestEnv) awaitCallbackPending(name, listenerID string, counter i
 				Archetype: channelservice.Archetype,
 			})
 		require.NoError(t, err)
-		node, ok := resp.GetDatabaseMutableState().GetChasmNodes()["CallbackListeners#"+listenerID]
+		node, ok := resp.GetDatabaseMutableState().GetChasmNodes()["Listeners#"+listenerID]
 		require.True(t, ok)
-		var listener channelpb.CallbackListener
+		var listener channelpb.Listener
 		require.NoError(t, proto.Unmarshal(node.GetData().GetData(), &listener))
 		require.Equal(t, counter, listener.GetPending().GetCounter())
 	}, 20*time.Second, 50*time.Millisecond)
@@ -490,7 +490,9 @@ type callbackRecorder struct {
 	bodies   []map[string]any
 	channels []string
 	hold     chan struct{}
-	arrived  chan struct{}
+	// How long each request takes to answer, for a slow endpoint.
+	delay   time.Duration
+	arrived chan struct{}
 }
 
 func newCallbackRecorder(t *testing.T) (*callbackRecorder, string) {
@@ -502,12 +504,18 @@ func newCallbackRecorder(t *testing.T) (*callbackRecorder, string) {
 		r.mu.Lock()
 		r.bodies = append(r.bodies, body)
 		r.channels = append(r.channels, req.Header.Get(channelservice.ChannelHeader))
-		hold := r.hold
+		hold, delay := r.hold, r.delay
 		r.mu.Unlock()
 		r.arrived <- struct{}{}
 		if hold != nil {
 			<-hold
 		}
+		if delay > 0 {
+			select {
+			case <-time.After(delay):
+			case <-req.Context().Done():
+			}
+		}
 		w.WriteHeader(http.StatusOK)
 	}))
 	t.Cleanup(server.Close)
@@ -595,6 +603,84 @@ func TestNotificationChannelCallbackListener(t *testing.T) {
 	require.Equal(t, int32(0), c.mustNotify(name, 5))
 }
 
+// Deliveries to a callback are in order: one post is in flight per listener,
+// and what arrives during it folds into the one post sent after it returns.
+// A slow endpoint under a burst of rising counters sees them rise and ends on
+// the last one.
+func TestNotificationChannelCallbackDeliveriesInOrder(t *testing.T) {
+	c := newChannelTestEnv(t)
+	name := "orders-" + uuid.NewString()
+	recorder, url := newCallbackRecorder(t)
+	recorder.mu.Lock()
+	recorder.delay = 150 * time.Millisecond
+	recorder.mu.Unlock()
+	_, err := c.env.FrontendClient().RegisterChannelListener(c.ctx(),
+		&workflowservice.RegisterChannelListenerRequest{
+			Namespace: c.ns,
+			Channel:   name,
+			Callback: &commonpb.Callback{Variant: &commonpb.Callback_Nexus_{
+				Nexus: &commonpb.Callback_Nexus{Url: url},
+			}},
+			RequestId: uuid.NewString(),
+			Identity:  "tester",
+		})
+	require.NoError(t, err)
+
+	const last = int64(12)
+	pace := time.NewTicker(40 * time.Millisecond)
+	defer pace.Stop()
+	for counter := int64(1); counter <= last; counter++ {
+		require.Equal(t, int32(1), c.mustNotify(name, counter))
+		<-pace.C
+	}
+	await.Require(c.ctx(), t, func(t *await.T) {
+		seen := recorder.counters()
+		require.NotEmpty(t, seen)
+		require.Equal(t, strconv.FormatInt(last, 10), seen[len(seen)-1])
+	}, 20*time.Second, 50*time.Millisecond)
+
+	seen := recorder.counters()
+	previous := int64(0)
+	for _, s := range seen {
+		counter, err := strconv.ParseInt(s, 10, 64)
+		require.NoError(t, err)
+		require.Greater(t, counter, previous, "deliveries out of order: %v", seen)
+		previous = counter
+	}
+	require.Equal(t, last, previous)
+}
+
+// A channel with no listeners is deleted a retention after its last activity,
+// ring included. A notify after that starts a fresh channel, and a poller
+// holding a counter from the old one sees only what the new one has.
+func TestNotificationChannelRecreatedAfterRetention(t *testing.T) {
+	c := newChannelTestEnv(t, testcore.WithDynamicConfig(channel.RetentionSetting, time.Second))
+	name := "orders-" + uuid.NewString()
+	for counter := int64(1); counter <= 3; counter++ {
+		require.Equal(t, int32(0), c.mustNotify(name, counter))
+	}
+	require.Equal(t, int32(3), c.describe(name).GetRetainedCount())
+
+	await.Require(c.ctx(), t, func(t *await.T) {
+		_, err := c.env.FrontendClient().DescribeChannel(c.ctx(), &workflowservice.DescribeChannelRequest{
+			Namespace: c.ns, Channel: name,
+		})
+		var notFound *serviceerror.NotFound
+		require.ErrorAs(t, err, &notFound, "the idle channel is still there")
+	}, 20*time.Second, 100*time.Millisecond)
+
+	require.Equal(t, int32(0), c.mustNotify(name, 7))
+	resp, err := c.env.FrontendClient().PollChannel(c.ctx(), &workflowservice.PollChannelRequest{
+		Namespace: c.ns, Channel: name, AfterCounter: 2,
+	})
+	require.NoError(t, err)
+	require.Len(t, resp.GetNotifications(), 1, "the old ring is gone with the old channel")
+	require.Equal(t, int64(7), resp.GetNotifications()[0].GetCounter())
+	desc := c.describe(name)
+	require.Equal(t, int32(1), desc.GetRetainedCount())
+	require.Equal(t, int64(7), desc.GetLatest().GetCounter())
+}
+
 // A long poll returns a notification published while it waits.
 func TestNotificationChannelPollWaits(t *testing.T) {
 	c := newChannelTestEnv(t)