From 2d721450f6978e65d524c228a58224f164df3983 Mon Sep 17 00:00:00 2001
From: Moe Dashti <mohammad.dashti@temporal.io>
Date: Fri, 2 Oct 2026 17:40:04 -0700
Subject: [PATCH 1/2] Held workflow and callback listeners in one table of
 callbacks.

A workflow listener is an internal callback whose data names the run, so the
table has one record shape and the delivery dispatches on the variant.
Describe and the metrics derive the kind from it.
---
 chasm/lib/channel/channel.go                  | 265 +++++++++++-------
 chasm/lib/channel/channel_test.go             |  80 +++++-
 .../gen/channelpb/v1/message.go-helpers.pb.go |  40 +--
 .../channel/gen/channelpb/v1/message.pb.go    | 249 ++++++++--------
 chasm/lib/channel/linked.go                   |   8 +-
 chasm/lib/channel/linked_test.go              |   4 +-
 chasm/lib/channel/proto/v1/message.proto      |  33 ++-
 chasm/lib/channel/service/deliver.go          |  30 +-
 chasm/lib/channel/service/handler.go          |  15 +-
 chasm/lib/channel/service/tasks.go            |   4 +-
 tests/notification_channel_test.go            |   4 +-
 11 files changed, 442 insertions(+), 290 deletions(-)

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..49e141bc18c 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)

From 6ddbc643902d24f8bc98ebdf7491debca0f61889 Mon Sep 17 00:00:00 2001
From: Moe Dashti <mohammad.dashti@temporal.io>
Date: Fri, 2 Oct 2026 17:40:04 -0700
Subject: [PATCH 2/2] Covered ordered callback posts and re-creation after
 retention.

The channel kept one post in flight per callback listener and folded what
arrived meanwhile, and a channel deleted by the idle task started fresh on
the next notify. Neither had a test.
---
 tests/notification_channel_test.go | 90 +++++++++++++++++++++++++++++-
 1 file changed, 88 insertions(+), 2 deletions(-)

diff --git a/tests/notification_channel_test.go b/tests/notification_channel_test.go
index 49e141bc18c..c9b415e823e 100644
--- a/tests/notification_channel_test.go
+++ b/tests/notification_channel_test.go
@@ -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)