diff --git a/chasm/lib/activity/streams.go b/chasm/lib/activity/streams.go
index f1cf80c3e5b..de1b0c53b8f 100644
--- a/chasm/lib/activity/streams.go
+++ b/chasm/lib/activity/streams.go
@@ -40,5 +40,14 @@ func (a *Activity) AppendToOwnedStream(
 			"activity execution closed with status %v, so its streams take no more records",
 			InternalStatusToAPIStatus(a.GetStatus()))
 	}
-	return a.ownedStreams().Append(mctx, key, req)
+	result, err := a.ownedStreams().Append(mctx, key, req)
+	if err != nil || result.Deduplicated {
+		return result, err
+	}
+	// A standalone activity has no linked channels, so its streams notify an
+	// independent channel named by the activity and the stream.
+	owner := mctx.ExecutionKey()
+	a.ownedStreams().Get(mctx, key).ScheduleChannelNotify(mctx,
+		stream.ActivityChannelName(owner.BusinessID, key), owner.RunID, req.Limits)
+	return result, nil
 }
diff --git a/chasm/lib/channel/linked.go b/chasm/lib/channel/linked.go
index 5f354d5d9dc..99d042738f8 100644
--- a/chasm/lib/channel/linked.go
+++ b/chasm/lib/channel/linked.go
@@ -54,6 +54,27 @@ func (c *Channel) NotifyLinked(
 	mctx chasm.MutableContext,
 	n *channelpb.Notification,
 	limits Limits,
+) (LinkedNotifyResult, error) {
+	return c.notifyLinked(mctx, n, limits, true)
+}
+
+// NotifyLinkedListeners accepts a notification for the ring and the callback
+// listeners without handing it to the owner. The channel a stream drives
+// uses it: the owning run learns of its own stream through its cursor, and a
+// Workflow Task per append would be a cost the stream did not have.
+func (c *Channel) NotifyLinkedListeners(
+	mctx chasm.MutableContext,
+	n *channelpb.Notification,
+	limits Limits,
+) (LinkedNotifyResult, error) {
+	return c.notifyLinked(mctx, n, limits, false)
+}
+
+func (c *Channel) notifyLinked(
+	mctx chasm.MutableContext,
+	n *channelpb.Notification,
+	limits Limits,
+	wakeOwner bool,
 ) (LinkedNotifyResult, error) {
 	limits = limits.withDefaults()
 	if err := CheckNotification(n, limits.MaxMetadataBytes); err != nil {
@@ -69,9 +90,12 @@ func (c *Channel) NotifyLinked(
 		out.Advanced = true
 	}
 	c.touch(mctx)
-	if c.OwnerHolds(n.GetCounter()) {
+	switch {
+	case !wakeOwner:
+		out.OwnerHeld = true
+	case c.OwnerHolds(n.GetCounter()):
 		out.OwnerHeld = true
-	} else {
+	default:
 		out.OwnerFolded = c.State.GetOwnerPending() != nil
 		c.State.OwnerPending = n
 	}
diff --git a/chasm/lib/channel/linked_test.go b/chasm/lib/channel/linked_test.go
index 0a6422e1486..d65f1c8992c 100644
--- a/chasm/lib/channel/linked_test.go
+++ b/chasm/lib/channel/linked_test.go
@@ -46,6 +46,28 @@ func TestLinkedNotifyReachesTheOwnerInOneWrite(t *testing.T) {
 	require.Equal(t, "owner", polled[0].GetLinkedTo().GetWorkflowId())
 }
 
+// A notify for the listeners alone joins the ring and reaches the callbacks,
+// and leaves the owner with nothing pending: the channel a stream drives
+// must not wake the run for its own stream.
+func TestLinkedNotifyListenersLeavesTheOwnerAlone(t *testing.T) {
+	c, mctx := newTestLinkedChannel(t)
+
+	result, err := c.NotifyLinkedListeners(mctx, note(1), Limits{})
+	require.NoError(t, err)
+	require.True(t, result.Advanced)
+	require.True(t, result.OwnerHeld, "nothing for the owner to take")
+	require.False(t, c.HasOwnerPending())
+	require.Equal(t, int64(1), c.LatestCounter())
+	polled, err := c.Poll(mctx, PollRequest{})
+	require.NoError(t, err)
+	require.Equal(t, []int64{1}, counters(polled))
+	require.Equal(t, "owner", polled[0].GetLinkedTo().GetWorkflowId())
+
+	_, err = c.NotifyLinked(mctx, note(2), Limits{})
+	require.NoError(t, err)
+	require.Equal(t, int64(2), c.State.GetOwnerPending().GetCounter(), "a plain notify still wakes")
+}
+
 // The owner folds like a subscribed listener: a repeat at or below what it
 // holds pending, or what its unstarted task carries, changes nothing; a
 // higher counter replaces the pending entry; and once the task has started
diff --git a/chasm/lib/stream/channel.go b/chasm/lib/stream/channel.go
new file mode 100644
index 00000000000..fa1ed9da176
--- /dev/null
+++ b/chasm/lib/stream/channel.go
@@ -0,0 +1,105 @@
+package stream
+
+import (
+	"strconv"
+
+	commonpb "go.temporal.io/api/common/v1"
+	"go.temporal.io/server/chasm"
+	channelpb "go.temporal.io/server/chasm/lib/channel/gen/channelpb/v1"
+	streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
+	"go.temporal.io/server/common/payload"
+)
+
+// A change to a stream notifies the notification channel named by the stream,
+// so callback and polling listeners learn of a native stream the way they
+// learn of an external one. The name is derived from the stream's identity
+// alone, so a client derives the same name without asking the server. The
+// workflow consumers of a stream keep their cursors and are not told twice.
+
+// ChannelNamePrefix starts the name of every channel a stream drives.
+const ChannelNamePrefix = "stream/"
+
+// ChannelName is the independent channel a standalone stream notifies.
+func ChannelName(streamID string) string {
+	return ChannelNamePrefix + streamID
+}
+
+// OwnedChannelName is the channel, linked to the owning run, that a stream a
+// workflow owns notifies.
+func OwnedChannelName(name string) string {
+	return ChannelNamePrefix + name
+}
+
+// ActivityChannelName is the channel a stream an activity owns notifies: linked
+// to the workflow run for an activity a workflow scheduled, independent for a
+// standalone activity, which has no linked channels of its own.
+func ActivityChannelName(activityID, name string) string {
+	return ChannelNamePrefix + activityID + "/" + name
+}
+
+// ClosedMetadataKey is set, to true, on the notification for a close.
+const ClosedMetadataKey = "closed"
+
+// ClosedMetadata is the metadata of a close notification.
+func ClosedMetadata() map[string]*commonpb.Payload {
+	// A bool encodes without error.
+	closed, _ := payload.Encode(true)
+	return map[string]*commonpb.Payload{ClosedMetadataKey: closed}
+}
+
+// ChangeNotification describes the stream's latest change for its channel:
+// the change sequence is the counter, the head after the change is the
+// position in the form a native cursor names one, and a close says so.
+//
+// The run is what a native cursor pairs the offset with: the owning run of an
+// attached stream, and the stream's own id for a standalone one, which has no
+// run a client would know.
+func (s *Stream) ChangeNotification(channel, run string) *channelpb.Notification {
+	n := &channelpb.Notification{
+		Channel:  channel,
+		Counter:  s.State.GetChangeSequence(),
+		Position: EncodePosition(run, s.State.GetHeadOffset()),
+	}
+	if s.State.GetClosed() {
+		n.Metadata = ClosedMetadata()
+	}
+	return n
+}
+
+// EncodePosition is a native position as a client's cursor names it.
+func EncodePosition(run string, offset int64) []byte {
+	return []byte(run + ":" + strconv.FormatInt(offset, 10))
+}
+
+// ScheduleChannelNotify arms a task to hand the latest change to the stream's
+// channel, for a stream whose channel is an execution of its own: a
+// standalone stream's, or that of a stream a standalone activity owns. One
+// task is outstanding at a time and reads the latest change when it runs.
+// Tasks of one execution may run in any order, and a change arriving at the
+// channel below one it already holds is folded away, so a task per change
+// would lose the earlier ones from the ring; one task at a time keeps the
+// counters the channel sees in order, with a burst arriving as its newest.
+func (s *Stream) ScheduleChannelNotify(
+	mctx chasm.MutableContext,
+	channel, run string,
+	limits Limits,
+) {
+	if mctx == nil || limits.ChannelNotifyOff || s.State.GetChannelNotifyPending() {
+		return
+	}
+	s.State.ChannelNotifyPending = true
+	mctx.AddTask(s, chasm.TaskAttributes{ScheduledTime: mctx.Now(s)},
+		&streamlib.StreamNotifyChannelTask{Channel: channel, Run: run})
+}
+
+// TakeChannelChange is the channel task's read of the stream. Lowering the
+// flag in the transition that reads the change is what makes the coalescing
+// safe: a change that commits after this one sees the flag down and arms the
+// next task.
+func (s *Stream) TakeChannelChange(
+	_ chasm.MutableContext,
+	task *streamlib.StreamNotifyChannelTask,
+) (*channelpb.Notification, error) {
+	s.State.ChannelNotifyPending = false
+	return s.ChangeNotification(task.GetChannel(), task.GetRun()), nil
+}
diff --git a/chasm/lib/stream/channel_test.go b/chasm/lib/stream/channel_test.go
new file mode 100644
index 00000000000..4d6bfeedadd
--- /dev/null
+++ b/chasm/lib/stream/channel_test.go
@@ -0,0 +1,84 @@
+package stream
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/server/chasm"
+	streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
+	"go.temporal.io/server/common/payload"
+)
+
+func newChannelTestContext() *chasm.MockMutableContext {
+	now := time.Unix(1_700_000_000, 0).UTC()
+	return &chasm.MockMutableContext{MockContext: chasm.MockContext{
+		HandleNow: func(chasm.Component) time.Time { return now },
+	}}
+}
+
+// Every append and the close are one change each, a producer's retry none,
+// and the notification for a change carries the sequence, the head paired
+// with the run, and closed only on the close.
+func TestStreamChangeNotification(t *testing.T) {
+	ctx := newChannelTestContext()
+	s := newTestStream(t)
+
+	_, err := s.AddMessages(ctx, AddMessagesRequest{Records: msgs("a", "b")})
+	require.NoError(t, err)
+	n := s.ChangeNotification("stream/s", "run-1")
+	require.Equal(t, "stream/s", n.GetChannel())
+	require.Equal(t, int64(1), n.GetCounter())
+	require.Equal(t, "run-1:2", string(n.GetPosition()))
+	require.Empty(t, n.GetMetadata())
+
+	retry := AddMessagesRequest{Records: msgs("c"), ProducerID: "p", Sequence: 1}
+	_, err = s.AddMessages(ctx, retry)
+	require.NoError(t, err)
+	replay, err := s.AddMessages(ctx, retry)
+	require.NoError(t, err)
+	require.True(t, replay.Deduplicated)
+	require.Equal(t, int64(2), s.ChangeNotification("stream/s", "run-1").GetCounter(),
+		"a replay is not a change")
+
+	require.NoError(t, s.CloseAndSchedule(ctx, nil))
+	n = s.ChangeNotification("stream/s", "run-1")
+	require.Equal(t, int64(3), n.GetCounter())
+	require.Equal(t, "run-1:3", string(n.GetPosition()))
+	var closed bool
+	require.NoError(t, payload.Decode(n.GetMetadata()[ClosedMetadataKey], &closed))
+	require.True(t, closed)
+	s.Close(ctx.Now(s), nil)
+	require.Equal(t, int64(3), s.State.GetChangeSequence(), "a second close is not a change")
+}
+
+// One channel task is outstanding at a time, it reads the latest change when
+// it runs and lowers the flag, and the switch leaves none.
+func TestStreamScheduleChannelNotify(t *testing.T) {
+	ctx := newChannelTestContext()
+	s := newTestStream(t)
+	_, err := s.AddMessages(ctx, AddMessagesRequest{Records: msgs("a")})
+	require.NoError(t, err)
+
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{ChannelNotifyOff: true})
+	require.Empty(t, ctx.Tasks)
+
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{})
+	require.Len(t, ctx.Tasks, 1)
+	task, ok := ctx.Tasks[0].Payload.(*streamlib.StreamNotifyChannelTask)
+	require.True(t, ok, "%T", ctx.Tasks[0].Payload)
+	require.Equal(t, "stream/s", task.GetChannel())
+	require.Equal(t, "s", task.GetRun())
+
+	_, err = s.AddMessages(ctx, AddMessagesRequest{Records: msgs("b")})
+	require.NoError(t, err)
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{})
+	require.Len(t, ctx.Tasks, 1, "one task at a time")
+
+	n, err := s.TakeChannelChange(ctx, task)
+	require.NoError(t, err)
+	require.Equal(t, int64(2), n.GetCounter(), "the task reads the latest change")
+	require.Equal(t, "s:2", string(n.GetPosition()))
+	s.ScheduleChannelNotify(ctx, "stream/s", "s", Limits{})
+	require.Len(t, ctx.Tasks, 2, "the flag is down, so the next change arms a task")
+}
diff --git a/chasm/lib/stream/config.go b/chasm/lib/stream/config.go
index 2459453623c..23f586d7fe9 100644
--- a/chasm/lib/stream/config.go
+++ b/chasm/lib/stream/config.go
@@ -128,6 +128,11 @@ const (
 	// every single stream stays inside its own budget. Half the error limit,
 	// which leaves the rest of mutable state its own room.
 	OwnedStreamsMaxBytesPerWorkflow = 4 << 20
+
+	// ChannelRetainedNotifications bounds the ring of a channel a stream drives,
+	// well below a linked channel's default, since the stream holds the data a
+	// poller would otherwise need the ring for.
+	ChannelRetainedNotifications = 16
 )
 
 var (
@@ -141,6 +146,17 @@ var (
 		`Whether the stream service and the workflow stream commands are available to a
 namespace. Off by default.`,
 	)
+	NotifyChannelSetting = dynamicconfig.NewNamespaceBoolSetting(
+		"stream.notifyChannel",
+		true,
+		`Whether a change to a native stream notifies the notification channel named by the
+stream. Off creates no channel state from streams.`,
+	)
+	ChannelRetainedNotificationsSetting = dynamicconfig.NewNamespaceIntSetting(
+		"stream.channelRetainedNotifications",
+		ChannelRetainedNotifications,
+		`How many notifications the channel named by a native stream retains for pollers.`,
+	)
 	MaxConsumeItemsPerTaskSetting = dynamicconfig.NewNamespaceIntSetting(
 		"stream.maxConsumeItemsPerTask",
 		MaxConsumeItemsPerTask,
@@ -290,10 +306,16 @@ type Config struct {
 	AppendRecordsPerSecond dynamicconfig.IntPropertyFnWithNamespaceFilter
 	AppendBytesPerSecond   dynamicconfig.IntPropertyFnWithNamespaceFilter
 	PollsPerSecond         dynamicconfig.IntPropertyFnWithNamespaceFilter
+
+	NotifyChannel                dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	ChannelRetainedNotifications dynamicconfig.IntPropertyFnWithNamespaceFilter
 }
 
 func NewConfig(dc *dynamicconfig.Collection) *Config {
 	return &Config{
+		NotifyChannel:                NotifyChannelSetting.Get(dc),
+		ChannelRetainedNotifications: ChannelRetainedNotificationsSetting.Get(dc),
+
 		Enabled:                    EnabledSetting.Get(dc),
 		MaxIDLength:                dynamicconfig.MaxIDLengthLimit.Get(dc),
 		RetentionRecheckInterval:   RetentionRecheckIntervalSetting.Get(dc),
@@ -338,6 +360,12 @@ type Limits struct {
 	ReplayMaxRecords                int
 	ReplayMaxBytes                  int
 	ReplayMaxPages                  int
+
+	// ChannelNotifyOff stops a stream's changes from reaching the channel
+	// named by the stream. The zero value is on, so a Limits built by hand
+	// behaves as a namespace with the default setting.
+	ChannelNotifyOff             bool
+	ChannelRetainedNotifications int
 }
 
 // LimitsFor resolves the limits for a namespace. A nil Config, which is what
@@ -347,6 +375,9 @@ func (c *Config) LimitsFor(namespaceName string) Limits {
 		return DefaultLimits()
 	}
 	return Limits{
+		ChannelNotifyOff:             !c.NotifyChannel(namespaceName),
+		ChannelRetainedNotifications: c.ChannelRetainedNotifications(namespaceName),
+
 		MaxConsumeItemsPerTask:     c.MaxConsumeItemsPerTask(namespaceName),
 		MaxConsumeBytesPerTask:     c.MaxConsumeBytesPerTask(namespaceName),
 		MaxProducersPerStream:      c.MaxProducersPerStream(namespaceName),
@@ -410,5 +441,6 @@ func (l Limits) withDefaults() Limits {
 	fill(&l.ReplayMaxRecords, ReplayMaxRecords)
 	fill(&l.ReplayMaxBytes, ReplayMaxBytes)
 	fill(&l.ReplayMaxPages, ReplayMaxPages)
+	fill(&l.ChannelRetainedNotifications, ChannelRetainedNotifications)
 	return l
 }
diff --git a/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go b/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go
index 8e4f7c4b2be..7514b6408ec 100644
--- a/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go
+++ b/chasm/lib/stream/gen/streampb/v1/stream_state.pb.go
@@ -67,8 +67,14 @@ type StreamState struct {
 	// schedule none of their own; the check re-arms itself while the stream
 	// holds records and lowers the flag when it holds none.
 	AgeTaskPending bool `protobuf:"varint,18,opt,name=age_task_pending,json=ageTaskPending,proto3" json:"age_task_pending,omitempty"`
-	unknownFields  protoimpl.UnknownFields
-	sizeCache      protoimpl.SizeCache
+	// One per append and one per close, the counter of the notification each
+	// sends to the channel named by the stream. Grows for the life of the stream.
+	ChangeSequence int64 `protobuf:"varint,19,opt,name=change_sequence,json=changeSequence,proto3" json:"change_sequence,omitempty"`
+	// A task to notify the stream's channel is outstanding. Changes while it is
+	// set schedule none of their own; the task reads the latest when it runs.
+	ChannelNotifyPending bool `protobuf:"varint,20,opt,name=channel_notify_pending,json=channelNotifyPending,proto3" json:"channel_notify_pending,omitempty"`
+	unknownFields        protoimpl.UnknownFields
+	sizeCache            protoimpl.SizeCache
 }
 
 func (x *StreamState) Reset() {
@@ -199,6 +205,20 @@ func (x *StreamState) GetAgeTaskPending() bool {
 	return false
 }
 
+func (x *StreamState) GetChangeSequence() int64 {
+	if x != nil {
+		return x.ChangeSequence
+	}
+	return 0
+}
+
+func (x *StreamState) GetChannelNotifyPending() bool {
+	if x != nil {
+		return x.ChannelNotifyPending
+	}
+	return false
+}
+
 // Hard bounds on what a stream may hold. Distinct from StreamLifecycle.max_items,
 // which reclaims the oldest messages: a budget refuses the newest.
 type StreamBudget struct {
@@ -627,7 +647,7 @@ var File_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto protorefle
 
 const file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDesc = "" +
 	"\n" +
-	"<temporal/server/chasm/lib/stream/proto/v1/stream_state.proto\x12)temporal.server.chasm.lib.stream.proto.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\"\xa4\b\n" +
+	"<temporal/server/chasm/lib/stream/proto/v1/stream_state.proto\x12)temporal.server.chasm.lib.stream.proto.v1\x1a\x1egoogle/protobuf/duration.proto\x1a\x1fgoogle/protobuf/timestamp.proto\x1a$temporal/api/common/v1/message.proto\"\x83\t\n" +
 	"\vStreamState\x12\x1f\n" +
 	"\vhead_offset\x18\x01 \x01(\x03R\n" +
 	"headOffset\x12\x1f\n" +
@@ -647,7 +667,9 @@ const file_temporal_server_chasm_lib_stream_proto_v1_stream_state_proto_rawDesc
 	"\x0enotify_pending\x18\x10 \x01(\bR\rnotifyPending\x12\x1d\n" +
 	"\n" +
 	"held_bytes\x18\x11 \x01(\x03R\theldBytes\x12(\n" +
-	"\x10age_task_pending\x18\x12 \x01(\bR\x0eageTaskPending\x1aw\n" +
+	"\x10age_task_pending\x18\x12 \x01(\bR\x0eageTaskPending\x12'\n" +
+	"\x0fchange_sequence\x18\x13 \x01(\x03R\x0echangeSequence\x124\n" +
+	"\x16channel_notify_pending\x18\x14 \x01(\bR\x14channelNotifyPending\x1aw\n" +
 	"\x0eProducersEntry\x12\x10\n" +
 	"\x03key\x18\x01 \x01(\tR\x03key\x12O\n" +
 	"\x05value\x18\x02 \x01(\v29.temporal.server.chasm.lib.stream.proto.v1.ProducerCursorR\x05value:\x028\x01\x1aw\n" +
diff --git a/chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go b/chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go
index 074c5801a6d..fed8032d299 100644
--- a/chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go
+++ b/chasm/lib/stream/gen/streampb/v1/tasks.go-helpers.pb.go
@@ -115,3 +115,40 @@ func (this *StreamAgeTask) Equal(that interface{}) bool {
 
 	return proto.Equal(this, that1)
 }
+
+// Marshal an object of type StreamNotifyChannelTask to the protobuf v3 wire format
+func (val *StreamNotifyChannelTask) Marshal() ([]byte, error) {
+	return proto.Marshal(val)
+}
+
+// Unmarshal an object of type StreamNotifyChannelTask from the protobuf v3 wire format
+func (val *StreamNotifyChannelTask) Unmarshal(buf []byte) error {
+	return proto.Unmarshal(buf, val)
+}
+
+// Size returns the size of the object, in bytes, once serialized
+func (val *StreamNotifyChannelTask) Size() int {
+	return proto.Size(val)
+}
+
+// Equal returns whether two StreamNotifyChannelTask 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 *StreamNotifyChannelTask) Equal(that interface{}) bool {
+	if that == nil {
+		return this == nil
+	}
+
+	var that1 *StreamNotifyChannelTask
+	switch t := that.(type) {
+	case *StreamNotifyChannelTask:
+		that1 = t
+	case StreamNotifyChannelTask:
+		that1 = &t
+	default:
+		return false
+	}
+
+	return proto.Equal(this, that1)
+}
diff --git a/chasm/lib/stream/gen/streampb/v1/tasks.pb.go b/chasm/lib/stream/gen/streampb/v1/tasks.pb.go
index c0d613ebb72..8d205233c27 100644
--- a/chasm/lib/stream/gen/streampb/v1/tasks.pb.go
+++ b/chasm/lib/stream/gen/streampb/v1/tasks.pb.go
@@ -141,6 +141,64 @@ func (*StreamAgeTask) Descriptor() ([]byte, []int) {
 	return file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDescGZIP(), []int{2}
 }
 
+// Fires after a change to a stream whose channel lives in another execution:
+// a standalone stream, or one a standalone activity owns. One is outstanding
+// at a time and reads the latest change when it runs, so changes reach the
+// channel in order and a burst reaches it as its newest change.
+type StreamNotifyChannelTask struct {
+	state   protoimpl.MessageState `protogen:"open.v1"`
+	Channel string                 `protobuf:"bytes,1,opt,name=channel,proto3" json:"channel,omitempty"`
+	// What a native position pairs the offset with: the owning run, or the
+	// stream's own id for a standalone stream.
+	Run           string `protobuf:"bytes,2,opt,name=run,proto3" json:"run,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *StreamNotifyChannelTask) Reset() {
+	*x = StreamNotifyChannelTask{}
+	mi := &file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes[3]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *StreamNotifyChannelTask) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*StreamNotifyChannelTask) ProtoMessage() {}
+
+func (x *StreamNotifyChannelTask) ProtoReflect() protoreflect.Message {
+	mi := &file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes[3]
+	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 StreamNotifyChannelTask.ProtoReflect.Descriptor instead.
+func (*StreamNotifyChannelTask) Descriptor() ([]byte, []int) {
+	return file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDescGZIP(), []int{3}
+}
+
+func (x *StreamNotifyChannelTask) GetChannel() string {
+	if x != nil {
+		return x.Channel
+	}
+	return ""
+}
+
+func (x *StreamNotifyChannelTask) GetRun() string {
+	if x != nil {
+		return x.Run
+	}
+	return ""
+}
+
 var File_temporal_server_chasm_lib_stream_proto_v1_tasks_proto protoreflect.FileDescriptor
 
 const file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDesc = "" +
@@ -148,7 +206,10 @@ const file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDesc = "" +
 	"5temporal/server/chasm/lib/stream/proto/v1/tasks.proto\x12)temporal.server.chasm.lib.stream.proto.v1\"\x15\n" +
 	"\x13StreamRetentionTask\"\x1b\n" +
 	"\x19StreamNotifyConsumersTask\"\x0f\n" +
-	"\rStreamAgeTaskB>Z<go.temporal.io/server/chasm/lib/stream/gen/streampb;streampbb\x06proto3"
+	"\rStreamAgeTask\"E\n" +
+	"\x17StreamNotifyChannelTask\x12\x18\n" +
+	"\achannel\x18\x01 \x01(\tR\achannel\x12\x10\n" +
+	"\x03run\x18\x02 \x01(\tR\x03runB>Z<go.temporal.io/server/chasm/lib/stream/gen/streampb;streampbb\x06proto3"
 
 var (
 	file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDescOnce sync.Once
@@ -162,11 +223,12 @@ func file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDescGZIP() []
 	return file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDescData
 }
 
-var file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_msgTypes = make([]protoimpl.MessageInfo, 4)
 var file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_goTypes = []any{
 	(*StreamRetentionTask)(nil),       // 0: temporal.server.chasm.lib.stream.proto.v1.StreamRetentionTask
 	(*StreamNotifyConsumersTask)(nil), // 1: temporal.server.chasm.lib.stream.proto.v1.StreamNotifyConsumersTask
 	(*StreamAgeTask)(nil),             // 2: temporal.server.chasm.lib.stream.proto.v1.StreamAgeTask
+	(*StreamNotifyChannelTask)(nil),   // 3: temporal.server.chasm.lib.stream.proto.v1.StreamNotifyChannelTask
 }
 var file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_depIdxs = []int32{
 	0, // [0:0] is the sub-list for method output_type
@@ -187,7 +249,7 @@ func file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDesc), len(file_temporal_server_chasm_lib_stream_proto_v1_tasks_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   3,
+			NumMessages:   4,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/chasm/lib/stream/proto/v1/stream_state.proto b/chasm/lib/stream/proto/v1/stream_state.proto
index b514a576179..6ced22ad2fe 100644
--- a/chasm/lib/stream/proto/v1/stream_state.proto
+++ b/chasm/lib/stream/proto/v1/stream_state.proto
@@ -63,6 +63,12 @@ message StreamState {
   // schedule none of their own; the check re-arms itself while the stream
   // holds records and lowers the flag when it holds none.
   bool age_task_pending = 18;
+  // One per append and one per close, the counter of the notification each
+  // sends to the channel named by the stream. Grows for the life of the stream.
+  int64 change_sequence = 19;
+  // A task to notify the stream's channel is outstanding. Changes while it is
+  // set schedule none of their own; the task reads the latest when it runs.
+  bool channel_notify_pending = 20;
 }
 
 // Hard bounds on what a stream may hold. Distinct from StreamLifecycle.max_items,
diff --git a/chasm/lib/stream/proto/v1/tasks.proto b/chasm/lib/stream/proto/v1/tasks.proto
index fa35057e3cf..30759f5d593 100644
--- a/chasm/lib/stream/proto/v1/tasks.proto
+++ b/chasm/lib/stream/proto/v1/tasks.proto
@@ -20,3 +20,14 @@ message StreamNotifyConsumersTask {}
 // lifecycle's retention. It truncates what aged, stopping at an active
 // consumer's floor, and re-arms itself while the stream holds records.
 message StreamAgeTask {}
+
+// Fires after a change to a stream whose channel lives in another execution:
+// a standalone stream, or one a standalone activity owns. One is outstanding
+// at a time and reads the latest change when it runs, so changes reach the
+// channel in order and a burst reaches it as its newest change.
+message StreamNotifyChannelTask {
+  string channel = 1;
+  // What a native position pairs the offset with: the owning run, or the
+  // stream's own id for a standalone stream.
+  string run = 2;
+}
diff --git a/chasm/lib/stream/service/fx.go b/chasm/lib/stream/service/fx.go
index 756b4820fc4..f7a4dfde92a 100644
--- a/chasm/lib/stream/service/fx.go
+++ b/chasm/lib/stream/service/fx.go
@@ -17,6 +17,9 @@ var HistoryModule = fx.Module(
 		newHandler,
 		newRetentionTaskHandler,
 		newNotifyConsumersTaskHandler,
+		// Takes the channel service's routed client, which the channel module
+		// provides to the same History process.
+		newNotifyChannelTaskHandler,
 		newAgeTaskHandler,
 		newLibrary,
 	),
diff --git a/chasm/lib/stream/service/handler.go b/chasm/lib/stream/service/handler.go
index d09e67534f2..978e3b94d6c 100644
--- a/chasm/lib/stream/service/handler.go
+++ b/chasm/lib/stream/service/handler.go
@@ -370,7 +370,16 @@ func (h *handler) AddMessages(
 
 	result, _, err := chasm.UpdateComponent(ctx,
 		refForRun(req.GetNamespaceId(), in.GetStreamId(), in.GetRunId()),
-		(*stream.Stream).AddMessages, addReq)
+		func(
+			s *stream.Stream, mctx chasm.MutableContext, r stream.AddMessagesRequest,
+		) (stream.AddMessagesResult, error) {
+			out, err := s.AddMessages(mctx, r)
+			if err == nil && !out.Deduplicated {
+				s.ScheduleChannelNotify(mctx,
+					stream.ChannelName(in.GetStreamId()), in.GetStreamId(), limits)
+			}
+			return out, err
+		}, addReq)
 	if err != nil {
 		return nil, err
 	}
@@ -1170,11 +1179,19 @@ func (h *handler) CloseStream(
 ) (*streamlib.CloseStreamResponse, error) {
 	in := req.GetFrontendRequest()
 	ctx = h.withCallerInfo(ctx, req.GetNamespaceId())
+	limits := h.limitsFor(req.GetNamespaceId())
 	_, _, err := chasm.UpdateComponent(
 		ctx,
 		refFor(req.GetNamespaceId(), in.GetStreamId()),
 		func(s *stream.Stream, mctx chasm.MutableContext, reason *commonpb.Payload) (struct{}, error) {
-			return struct{}{}, s.CloseAndSchedule(mctx, reason)
+			if s.State.GetClosed() {
+				return struct{}{}, nil
+			}
+			if err := s.CloseAndSchedule(mctx, reason); err != nil {
+				return struct{}{}, err
+			}
+			s.ScheduleChannelNotify(mctx, stream.ChannelName(in.GetStreamId()), in.GetStreamId(), limits)
+			return struct{}{}, nil
 		},
 		in.GetReason(),
 	)
diff --git a/chasm/lib/stream/service/library.go b/chasm/lib/stream/service/library.go
index 825c295e126..3458c3c13e7 100644
--- a/chasm/lib/stream/service/library.go
+++ b/chasm/lib/stream/service/library.go
@@ -22,6 +22,7 @@ type library struct {
 	handler         *handler
 	retention       *retentionTaskHandler
 	notifyConsumers *notifyConsumersTaskHandler
+	notifyChannel   *notifyChannelTaskHandler
 	age             *ageTaskHandler
 }
 
@@ -29,9 +30,16 @@ func newLibrary(
 	h *handler,
 	retention *retentionTaskHandler,
 	notifyConsumers *notifyConsumersTaskHandler,
+	notifyChannel *notifyChannelTaskHandler,
 	age *ageTaskHandler,
 ) *library {
-	return &library{handler: h, retention: retention, notifyConsumers: notifyConsumers, age: age}
+	return &library{
+		handler:         h,
+		retention:       retention,
+		notifyConsumers: notifyConsumers,
+		notifyChannel:   notifyChannel,
+		age:             age,
+	}
 }
 
 // componentOnlyLibrary registers the component without the service, which is
@@ -83,6 +91,10 @@ func (l *library) Tasks() []*chasm.RegistrableTask {
 			"streamNotifyConsumers",
 			l.notifyConsumers,
 		),
+		chasm.NewRegistrableSideEffectTask(
+			"streamNotifyChannel",
+			l.notifyChannel,
+		),
 		chasm.NewRegistrablePureTask(
 			"streamAge",
 			l.age,
diff --git a/chasm/lib/stream/service/tasks.go b/chasm/lib/stream/service/tasks.go
index ecc2beb415d..771f10ac86e 100644
--- a/chasm/lib/stream/service/tasks.go
+++ b/chasm/lib/stream/service/tasks.go
@@ -3,9 +3,11 @@ package service
 import (
 	"context"
 	"errors"
+	"fmt"
 
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/chasm"
+	channelpb "go.temporal.io/server/chasm/lib/channel/gen/channelpb/v1"
 	"go.temporal.io/server/chasm/lib/stream"
 	streampb "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
 	"go.temporal.io/server/common/headers"
@@ -295,6 +297,78 @@ func newNotifyConsumersTaskHandler(
 	}
 }
 
+// notifyChannelTaskHandler tells the channel named for a stream about one
+// change. The channel is an execution of its own, so the notify goes out
+// through the channel service to be routed, as any writer's would.
+type notifyChannelTaskHandler struct {
+	chasm.SideEffectTaskHandlerBase[*streampb.StreamNotifyChannelTask]
+
+	namespaceRegistry namespace.Registry
+	logger            log.Logger
+	channels          channelpb.ChannelServiceClient
+}
+
+func newNotifyChannelTaskHandler(
+	namespaceRegistry namespace.Registry,
+	logger log.Logger,
+	channels channelpb.ChannelServiceClient,
+) *notifyChannelTaskHandler {
+	return &notifyChannelTaskHandler{
+		namespaceRegistry: namespaceRegistry,
+		logger:            logger,
+		channels:          channels,
+	}
+}
+
+// Validate accepts the task unconditionally, for the reason the consumer
+// notify does: the change that scheduled it raised the flag that keeps later
+// changes from scheduling their own, and Execute is what lowers it.
+func (h *notifyChannelTaskHandler) Validate(
+	_ chasm.Context,
+	_ *stream.Stream,
+	_ chasm.TaskInvocation,
+	_ *streampb.StreamNotifyChannelTask,
+) (bool, error) {
+	return true, nil
+}
+
+// Execute reads the latest change and notifies the channel. The request id
+// names the change, so a retry of the task after an ambiguous failure is
+// answered rather than accepted twice.
+func (h *notifyChannelTaskHandler) Execute(
+	ctx context.Context,
+	ref chasm.ComponentRef,
+	_ chasm.TaskAttributes,
+	task *streampb.StreamNotifyChannelTask,
+) error {
+	ctx, namespaceName := backgroundCallerContext(ctx, h.namespaceRegistry, ref.NamespaceID)
+	n, _, err := chasm.UpdateComponent(ctx, ref, (*stream.Stream).TakeChannelChange, task)
+	if err != nil {
+		return err
+	}
+	if n.GetCounter() == 0 {
+		return nil
+	}
+	callCtx, cancel := context.WithTimeout(ctx, stream.RoutedCallTimeout)
+	defer cancel()
+	_, err = h.channels.NotifyChannel(callCtx, &channelpb.NotifyChannelRequest{
+		NamespaceId: ref.NamespaceID,
+		FrontendRequest: &channelpb.NotifyChannelInput{
+			Namespace:    namespaceName,
+			Notification: n,
+			Identity:     "stream",
+			RequestId:    fmt.Sprintf("stream-change:%s:%d", n.GetChannel(), n.GetCounter()),
+		},
+	})
+	if err != nil {
+		h.logger.Error("failed to notify the channel named by a stream",
+			tag.NewStringTag("stream-id", ref.BusinessID),
+			tag.NewStringTag("channel", task.GetChannel()),
+			tag.Error(err))
+	}
+	return err
+}
+
 // Validate accepts the task unconditionally.
 //
 // The scheduling append already decided there was someone to tell, and it
diff --git a/chasm/lib/stream/stream.go b/chasm/lib/stream/stream.go
index 625ad11e855..fc9222a9821 100644
--- a/chasm/lib/stream/stream.go
+++ b/chasm/lib/stream/stream.go
@@ -245,6 +245,7 @@ func (s *Stream) AddMessages(
 	s.State.HeadOffset = first + count
 	s.State.AppendedBytes += int64(len(blob.Data))
 	s.State.HeldBytes += int64(len(blob.Data))
+	s.State.ChangeSequence++
 	if req.ProducerID != "" {
 		if s.State.Producers == nil {
 			s.State.Producers = make(map[string]*streamlib.ProducerCursor)
@@ -560,6 +561,7 @@ func (s *Stream) Close(now time.Time, reason *commonpb.Payload) time.Time {
 	s.State.Closed = true
 	s.State.CloseReason = reason
 	s.State.CloseTime = timestamppb.New(now)
+	s.State.ChangeSequence++
 
 	retention := s.State.GetLifecycle().GetRetention().AsDuration()
 	if retention <= 0 {
diff --git a/chasm/lib/workflow/stream_admission_test.go b/chasm/lib/workflow/stream_admission_test.go
index d70131489cb..19e5847d34b 100644
--- a/chasm/lib/workflow/stream_admission_test.go
+++ b/chasm/lib/workflow/stream_admission_test.go
@@ -260,7 +260,7 @@ func TestCloseActivityStreamsEndsOnlyThatActivity(t *testing.T) {
 
 	require.True(t, w.HasOpenActivityStreams(ctx, "act"))
 	require.False(t, w.HasOpenActivityStreams(ctx, "never-wrote"))
-	require.NoError(t, w.CloseActivityStreams(ctx, "act"))
+	require.NoError(t, w.CloseActivityStreams(ctx, "act", stream.Limits{}))
 	require.False(t, w.HasOpenActivityStreams(ctx, "act"))
 
 	closed := func(key string) bool { return w.OwnedStream(ctx, key).State.GetClosed() }
diff --git a/chasm/lib/workflow/stream_channel.go b/chasm/lib/workflow/stream_channel.go
new file mode 100644
index 00000000000..89f56fa0472
--- /dev/null
+++ b/chasm/lib/workflow/stream_channel.go
@@ -0,0 +1,53 @@
+package workflow
+
+import (
+	"net/url"
+	"strings"
+
+	"go.temporal.io/server/chasm"
+	"go.temporal.io/server/chasm/lib/channel"
+	"go.temporal.io/server/chasm/lib/stream"
+)
+
+// notifyStreamChannel tells the channel linked to this run under the stream's
+// name about the stream's latest change, creating the channel on the first.
+// The run itself is not woken: it learns of a stream it owns through its
+// cursor, and a Workflow Task per append would be a cost the stream did not
+// have before the channel. The callback listeners are handed the change and
+// the ring keeps it for pollers.
+func (w *Workflow) notifyStreamChannel(
+	mctx chasm.MutableContext,
+	key string,
+	s *stream.Stream,
+	limits stream.Limits,
+) error {
+	if limits.ChannelNotifyOff {
+		return nil
+	}
+	name := streamChannelName(key)
+	// One channel per stream, over and above what the run links by hand, so
+	// a stream that fits is never refused its channel for room.
+	c, err := w.LinkedChannelOrNew(mctx, name,
+		limits.MaxOwnedStreamsPerWorkflow+channel.DefaultMaxLinkedChannelsPerWorkflow)
+	if err != nil {
+		return err
+	}
+	_, err = c.NotifyLinkedListeners(mctx, s.ChangeNotification(name, mctx.ExecutionKey().RunID),
+		channel.Limits{LinkedRetainedNotifications: limits.ChannelRetainedNotifications})
+	return err
+}
+
+// streamChannelName is the channel a stream held under the key notifies: the
+// stream's own name for one the workflow owns, the activity and the name for
+// one of its activities owns.
+func streamChannelName(key string) string {
+	if !IsActivityStreamKey(key) {
+		return stream.OwnedChannelName(key)
+	}
+	escaped, name, _ := strings.Cut(strings.TrimPrefix(key, activityStreamPrefix), "/")
+	activityID, err := url.PathUnescape(escaped)
+	if err != nil {
+		activityID = escaped
+	}
+	return stream.ActivityChannelName(activityID, name)
+}
diff --git a/chasm/lib/workflow/stream_channel_test.go b/chasm/lib/workflow/stream_channel_test.go
new file mode 100644
index 00000000000..f27f5de47e5
--- /dev/null
+++ b/chasm/lib/workflow/stream_channel_test.go
@@ -0,0 +1,58 @@
+package workflow
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.temporal.io/server/chasm/lib/stream"
+	"go.temporal.io/server/common/payload"
+)
+
+// The channel a held stream notifies is named by the stream alone for one
+// the workflow owns, and by the activity and the stream for one of its
+// activities owns, with the activity id as the caller wrote it.
+func TestStreamChannelName(t *testing.T) {
+	require.Equal(t, "stream/output", streamChannelName("output"))
+	require.Equal(t, "stream/model/call/out",
+		streamChannelName(ActivityStreamKey("model/call", "out")))
+	require.Equal(t, "stream/act/output", streamChannelName(ActivityStreamKey("act", "output")))
+}
+
+// An append to a stream the run holds notifies the linked channel of the
+// stream's name without waking the run, the activity's end closes its streams
+// with a last change, and the switch leaves no channel behind.
+func TestStreamAppendNotifiesLinkedChannel(t *testing.T) {
+	ctx := newOwnerContext()
+	w := &Workflow{}
+	_, err := w.AppendToOwnedStream(ctx, "output", stream.AddMessagesRequest{
+		Records: budgetTestRecords(1),
+	})
+	require.NoError(t, err)
+	c, ok := w.LinkedChannel(ctx, "stream/output")
+	require.True(t, ok, "the first change creates the channel")
+	require.Equal(t, int64(1), c.LatestCounter())
+	require.Equal(t, int64(1), c.RetainedCount())
+	require.Equal(t, "run-1:1", string(c.State.GetLatest().GetPosition()))
+	require.False(t, c.HasOwnerPending(), "the run is not woken by its own stream")
+	require.False(t, w.HasPendingChannelNotifications(ctx))
+
+	key := ActivityStreamKey("act", "output")
+	_, err = w.AppendToOwnedStream(ctx, key, stream.AddMessagesRequest{Records: budgetTestRecords(1)})
+	require.NoError(t, err)
+	require.NoError(t, w.CloseActivityStreams(ctx, "act", stream.Limits{}))
+	c, ok = w.LinkedChannel(ctx, "stream/act/output")
+	require.True(t, ok)
+	require.Equal(t, int64(2), c.LatestCounter())
+	var closed bool
+	require.NoError(t, payload.Decode(
+		c.State.GetLatest().GetMetadata()[stream.ClosedMetadataKey], &closed))
+	require.True(t, closed)
+	require.False(t, w.HasPendingChannelNotifications(ctx))
+
+	off := &Workflow{}
+	_, err = off.AppendToOwnedStream(ctx, "output", stream.AddMessagesRequest{
+		Records: budgetTestRecords(1), Limits: stream.Limits{ChannelNotifyOff: true},
+	})
+	require.NoError(t, err)
+	require.Zero(t, off.LinkedChannelCount())
+}
diff --git a/chasm/lib/workflow/stream_commands.go b/chasm/lib/workflow/stream_commands.go
index 9f8f7e57a90..c348d78ec84 100644
--- a/chasm/lib/workflow/stream_commands.go
+++ b/chasm/lib/workflow/stream_commands.go
@@ -71,9 +71,20 @@ func (w *Workflow) HasOpenActivityStreams(ctx chasm.Context, activityID string)
 // be released: the workflow is still running, so the rule that ends a stream
 // with its execution does not reach these. A retry is not terminal, so the
 // next attempt keeps writing to the same streams.
-func (w *Workflow) CloseActivityStreams(mctx chasm.MutableContext, activityID string) error {
+func (w *Workflow) CloseActivityStreams(
+	mctx chasm.MutableContext,
+	activityID string,
+	limits stream.Limits,
+) error {
 	for _, key := range w.activityStreamKeys(activityID) {
-		if err := w.Streams[key].Get(mctx).CloseAndSchedule(mctx, nil); err != nil {
+		s := w.Streams[key].Get(mctx)
+		if s.State.GetClosed() {
+			continue
+		}
+		if err := s.CloseAndSchedule(mctx, nil); err != nil {
+			return err
+		}
+		if err := w.notifyStreamChannel(mctx, key, s, limits); err != nil {
 			return err
 		}
 	}
@@ -184,7 +195,7 @@ func handleAppendStreamRecordsCommand(
 	}
 	wf.RecordStreamRecordsAppended(
 		name, result.FirstOffset, result.NextOffset, opts.WorkflowTaskCompletedEventID)
-	return nil
+	return wf.notifyStreamChannel(chasmCtx, name, s, limits)
 }
 
 // handleSubscribeStreamCommand registers this workflow as a consumer.
diff --git a/chasm/lib/workflow/workflow.go b/chasm/lib/workflow/workflow.go
index 2bbcaca38b1..c5b1446ed09 100644
--- a/chasm/lib/workflow/workflow.go
+++ b/chasm/lib/workflow/workflow.go
@@ -915,5 +915,9 @@ func (w *Workflow) AppendToOwnedStream(
 	key string,
 	req stream.AddMessagesRequest,
 ) (stream.AddMessagesResult, error) {
-	return w.ownedStreams().Append(mctx, key, req)
+	result, err := w.ownedStreams().Append(mctx, key, req)
+	if err != nil || result.Deduplicated {
+		return result, err
+	}
+	return result, w.notifyStreamChannel(mctx, key, w.ownedStreams().Get(mctx, key), req.Limits)
 }
diff --git a/service/history/workflow/mutable_state_impl.go b/service/history/workflow/mutable_state_impl.go
index 5a09d11d03d..b24cc4f5d77 100644
--- a/service/history/workflow/mutable_state_impl.go
+++ b/service/history/workflow/mutable_state_impl.go
@@ -898,7 +898,8 @@ func (ms *MutableStateImpl) closeActivityStreams(activityID string) error {
 	if err != nil {
 		return err
 	}
-	return wf.CloseActivityStreams(mutableCtx, activityID)
+	return wf.CloseActivityStreams(mutableCtx, activityID,
+		ms.config.Stream.LimitsFor(ms.GetNamespaceEntry().Name().String()))
 }
 
 // deleteTerminalActivity deletes an activity that reached a terminal status,
diff --git a/tests/stream_channel_test.go b/tests/stream_channel_test.go
new file mode 100644
index 00000000000..e180d7f0835
--- /dev/null
+++ b/tests/stream_channel_test.go
@@ -0,0 +1,317 @@
+package tests
+
+import (
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	commonpb "go.temporal.io/api/common/v1"
+	notificationpb "go.temporal.io/api/notification/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/chasm/lib/activity"
+	"go.temporal.io/server/chasm/lib/stream"
+	streamlib "go.temporal.io/server/chasm/lib/stream/gen/streampb/v1"
+	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/payload"
+	"go.temporal.io/server/common/testing/await"
+	"go.temporal.io/server/tests/testcore"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+// A change to a native stream notifies the channel named by the stream, so a
+// client learns of the stream the way it learns of an external one. These
+// cases write through the stream service and read the channel.
+
+// newStreamChannelEnv serves the channel calls and the stream calls from one
+// cluster, with activities on for the activity-owned cases.
+func newStreamChannelEnv(
+	t *testing.T,
+	opts ...testcore.TestOption,
+) (*channelTestEnv, *streamTestEnv) {
+	c := newChannelTestEnv(t, opts...)
+	on := []dynamicconfig.ConstrainedValue{{
+		Constraints: dynamicconfig.Constraints{Namespace: c.ns},
+		Value:       true,
+	}}
+	cluster := c.env.GetTestCluster()
+	cluster.OverrideDynamicConfig(t, dynamicconfig.EnableChasm, on)
+	cluster.OverrideDynamicConfig(t, activity.Enabled, on)
+	return c, newStreamTestEnvFrom(t, c.env)
+}
+
+// requireChange checks one stream-driven notification: the change sequence
+// as the counter, the head paired with the run as the position, and closed
+// only on the close.
+func requireChange(
+	t *testing.T,
+	n *notificationpb.Notification,
+	channel, run string,
+	counter, head int64,
+	closed bool,
+) {
+	t.Helper()
+	require.Equal(t, channel, n.GetChannel())
+	require.Equal(t, counter, n.GetCounter(), "counter of %q", channel)
+	require.Equal(t, string(stream.EncodePosition(run, head)), string(n.GetPosition()))
+	if !closed {
+		require.Empty(t, n.GetMetadata(), "metadata of change %d", counter)
+		return
+	}
+	var isClosed bool
+	require.NoError(t, payload.Decode(n.GetMetadata()[stream.ClosedMetadataKey], &isClosed))
+	require.True(t, isClosed)
+}
+
+func (c *channelTestEnv) pollIndependent(name string, after int64) []*notificationpb.Notification {
+	c.t.Helper()
+	resp, err := c.env.FrontendClient().PollChannel(c.ctx(), &workflowservice.PollChannelRequest{
+		Namespace: c.ns, Channel: name, AfterCounter: after,
+	})
+	require.NoError(c.t, err)
+	return resp.GetNotifications()
+}
+
+// awaitIndependent waits for the channel's latest notification to reach the
+// counter, since a standalone stream reaches its channel through a task, and
+// returns what the channel retains above the given counter. The task reads
+// the latest change when it runs, so a burst may arrive as fewer
+// notifications than changes, in order.
+func (c *channelTestEnv) awaitIndependent(
+	name string,
+	after, latest int64,
+) []*notificationpb.Notification {
+	c.t.Helper()
+	var out []*notificationpb.Notification
+	await.Require(c.ctx(), c.t, func(t *await.T) {
+		resp, err := c.env.FrontendClient().PollChannel(c.ctx(), &workflowservice.PollChannelRequest{
+			Namespace: c.ns, Channel: name, AfterCounter: after,
+		})
+		require.NoError(t, err)
+		out = resp.GetNotifications()
+		require.NotEmpty(t, out)
+		require.Equal(t, latest, out[len(out)-1].GetCounter())
+	}, 20*time.Second, 50*time.Millisecond)
+	for i := 1; i < len(out); i++ {
+		require.Less(c.t, out[i-1].GetCounter(), out[i].GetCounter(), "in order")
+	}
+	return out
+}
+
+func (c *channelTestEnv) registerCallback(
+	name, url string,
+	owner *commonpb.Execution,
+) string {
+	c.t.Helper()
+	resp, 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",
+			Execution: owner,
+		})
+	require.NoError(c.t, err)
+	return resp.GetListenerId()
+}
+
+func awaitCounters(t *testing.T, recorder *callbackRecorder, want ...string) {
+	t.Helper()
+	await.Require(t.Context(), t, func(t *await.T) {
+		require.Equal(t, want, recorder.counters())
+	}, 20*time.Second, 50*time.Millisecond)
+}
+
+// A stream a workflow owns notifies the channel of the stream's name linked
+// to the run: one notification per append, whatever the batch holds, with
+// the head as its position. The owner is not woken by its own stream, a
+// callback registered on the channel is handed the latest and then each
+// change, and the stream's own readers see nothing different.
+func TestStreamChannelWorkflowOwnedStream(t *testing.T) {
+	c, s := newStreamChannelEnv(t)
+	id := "stream-channel-wf-" + uuid.NewString()
+	runID := c.startIdle(id)
+	owner := &streamlib.StreamOwner{Kind: streamlib.STREAM_OWNER_KIND_WORKFLOW, Id: id}
+	name := stream.OwnedChannelName("output")
+	linked := linkedOwner(id, "")
+
+	_, err := s.addOwned(t, owner, "output", &streamlib.AddWorkflowMessagesInput{
+		Records: streamMsgs("", "a", "b"),
+	})
+	require.NoError(t, err)
+	resp, err := c.pollLinked(linked, name, 0, 0, 0)
+	require.NoError(t, err)
+	require.Len(t, resp.GetNotifications(), 1, "one notification per append call")
+	requireChange(t, resp.GetNotifications()[0], name, runID, 1, 2, false)
+	require.Equal(t, id, resp.GetNotifications()[0].GetLinkedTo().GetBusinessId())
+	require.False(t, c.hasPendingTask(id), "the owner is not woken by its own stream")
+
+	recorder, url := newCallbackRecorder(t)
+	c.registerCallback(name, url, linked)
+	awaitCounters(t, recorder, "1")
+
+	_, err = s.addOwned(t, owner, "output", &streamlib.AddWorkflowMessagesInput{
+		Records: streamMsgs("", "c"),
+	})
+	require.NoError(t, err)
+	resp, err = c.pollLinked(linked, name, 1, 0, 0)
+	require.NoError(t, err)
+	require.Len(t, resp.GetNotifications(), 1)
+	requireChange(t, resp.GetNotifications()[0], name, runID, 2, 3, false)
+	awaitCounters(t, recorder, "1", "2")
+	require.False(t, c.hasPendingTask(id))
+
+	// A producer's retry appends nothing and notifies nothing.
+	retry := &streamlib.AddWorkflowMessagesInput{
+		Records: streamMsgs("", "d"), ProducerId: "p", Sequence: 1,
+	}
+	_, err = s.addOwned(t, owner, "output", retry)
+	require.NoError(t, err)
+	_, err = s.addOwned(t, owner, "output", retry)
+	require.NoError(t, err)
+	desc := c.mustDescribeLinked(id, name)
+	require.Equal(t, notificationpb.CHANNEL_KIND_LINKED, desc.GetKind())
+	require.Equal(t, int64(3), desc.GetLatest().GetCounter())
+	require.Equal(t, int32(3), desc.GetRetainedCount())
+	awaitCounters(t, recorder, "1", "2", "3")
+
+	require.Equal(t, []string{"a", "b", "c", "d"},
+		bodies(s.pollOwned(t, owner, "output", 0, false).GetRecords()))
+}
+
+// A standalone stream notifies the independent channel of its id through a
+// task that carries the latest change. A callback that registers after
+// three appends is handed the third once, and the close is the last change,
+// marked closed.
+func TestStreamChannelStandaloneStream(t *testing.T) {
+	c, s := newStreamChannelEnv(t)
+	ctx := s.ctx()
+	streamID := "stream-channel-sa-" + uuid.NewString()
+	name := stream.ChannelName(streamID)
+	s.create(ctx, t, streamID)
+	_, err := s.add(ctx, t, streamID, &streamlib.AddMessagesInput{Records: streamMsgs("", "a")})
+	require.NoError(t, err)
+	first := c.awaitIndependent(name, 0, 1)
+	requireChange(t, first[0], name, streamID, 1, 1, false)
+	require.Nil(t, first[0].GetLinkedTo())
+
+	for _, body := range []string{"b", "c"} {
+		_, err := s.add(ctx, t, streamID, &streamlib.AddMessagesInput{Records: streamMsgs("", body)})
+		require.NoError(t, err)
+	}
+	changes := c.awaitIndependent(name, 1, 3)
+	last := changes[len(changes)-1]
+	requireChange(t, last, name, streamID, 3, 3, false)
+
+	recorder, url := newCallbackRecorder(t)
+	c.registerCallback(name, url, nil)
+	awaitCounters(t, recorder, "3")
+
+	_, err = s.client.CloseStream(ctx, &streamlib.CloseStreamRequest{
+		FrontendRequest: &streamlib.CloseStreamInput{Namespace: s.ns, StreamId: streamID},
+	})
+	require.NoError(t, err)
+	closed := c.awaitIndependent(name, 3, 4)
+	require.Len(t, closed, 1)
+	requireChange(t, closed[0], name, streamID, 4, 3, true)
+	awaitCounters(t, recorder, "3", "4")
+
+	// A second close changes nothing.
+	_, err = s.client.CloseStream(ctx, &streamlib.CloseStreamRequest{
+		FrontendRequest: &streamlib.CloseStreamInput{Namespace: s.ns, StreamId: streamID},
+	})
+	require.NoError(t, err)
+	desc := c.describe(name)
+	require.Equal(t, notificationpb.CHANNEL_KIND_INDEPENDENT, desc.GetKind())
+	require.Equal(t, int64(4), desc.GetLatest().GetCounter())
+	require.Len(t, desc.GetListeners(), 1)
+}
+
+// A stream an activity of a workflow owns notifies a channel named by the
+// activity and the stream, linked to the workflow run, and the activity's
+// completion closes the stream, which is the last change.
+func TestStreamChannelWorkflowActivityStream(t *testing.T) {
+	c, s := newStreamChannelEnv(t)
+	id := "stream-channel-wfa-" + uuid.NewString()
+	runID := scheduleStreamingActivity(t, c.env, s, id, "model-call", nil)
+	task := pollActivityTask(t, c.env, s, id+"-tq")
+	owner := &streamlib.StreamOwner{
+		Kind: streamlib.STREAM_OWNER_KIND_WORKFLOW_ACTIVITY, Id: id, ActivityId: "model-call",
+	}
+	name := stream.ActivityChannelName("model-call", "output")
+
+	_, err := s.addOwned(t, owner, "output", &streamlib.AddWorkflowMessagesInput{
+		Records: attemptRecords(1, "token"),
+	})
+	require.NoError(t, err)
+	resp, err := c.pollLinked(linkedOwner(id, ""), name, 0, 0, 0)
+	require.NoError(t, err)
+	require.Len(t, resp.GetNotifications(), 1)
+	requireChange(t, resp.GetNotifications()[0], name, runID, 1, 1, false)
+
+	completeActivityAttempt(t, c.env, s, task)
+	await.Require(c.ctx(), t, func(at *await.T) {
+		require.True(at, s.describeOwned(t, owner, "output").GetClosed())
+	}, 20*time.Second, 50*time.Millisecond)
+	resp, err = c.pollLinked(linkedOwner(id, ""), name, 1, 0, 0)
+	require.NoError(t, err)
+	require.Len(t, resp.GetNotifications(), 1)
+	requireChange(t, resp.GetNotifications()[0], name, runID, 2, 1, true)
+}
+
+// A stream a standalone activity owns has no linked channel to live in, so
+// it notifies the independent channel named by the activity and the stream.
+func TestStreamChannelStandaloneActivityStream(t *testing.T) {
+	c, s := newStreamChannelEnv(t)
+	activityID := "stream-channel-saa-" + uuid.NewString()
+	started, err := c.env.FrontendClient().StartActivityExecution(s.ctx(),
+		&workflowservice.StartActivityExecutionRequest{
+			Namespace:           s.ns,
+			ActivityId:          activityID,
+			ActivityType:        &commonpb.ActivityType{Name: "streaming-activity"},
+			TaskQueue:           c.taskQueue(activityID),
+			StartToCloseTimeout: durationpb.New(time.Minute),
+			RequestId:           uuid.NewString(),
+		})
+	require.NoError(t, err)
+	pollActivityTask(t, c.env, s, activityID+"-tq")
+	owner := &streamlib.StreamOwner{Kind: streamlib.STREAM_OWNER_KIND_ACTIVITY, Id: activityID}
+	name := stream.ActivityChannelName(activityID, "output")
+
+	_, err = s.addOwned(t, owner, "output", &streamlib.AddWorkflowMessagesInput{
+		Records: attemptRecords(1, "token one", "token two"),
+	})
+	require.NoError(t, err)
+	changes := c.awaitIndependent(name, 0, 1)
+	require.Len(t, changes, 1)
+	requireChange(t, changes[0], name, started.GetRunId(), 1, 2, false)
+}
+
+// With the switch off, streams create no channel state at all.
+func TestStreamChannelSwitchOff(t *testing.T) {
+	c, s := newStreamChannelEnv(t, testcore.WithDynamicConfig(stream.NotifyChannelSetting, false))
+	id := "stream-channel-off-" + uuid.NewString()
+	c.startIdle(id)
+	owner := &streamlib.StreamOwner{Kind: streamlib.STREAM_OWNER_KIND_WORKFLOW, Id: id}
+	_, err := s.addOwned(t, owner, "output", &streamlib.AddWorkflowMessagesInput{
+		Records: streamMsgs("", "a"),
+	})
+	require.NoError(t, err)
+	desc := c.mustDescribeLinked(id, stream.OwnedChannelName("output"))
+	require.Nil(t, desc.GetLatest(), "the linked name exists by construction and holds nothing")
+	require.Zero(t, desc.GetRetainedCount())
+
+	ctx := s.ctx()
+	streamID := "stream-channel-off-sa-" + uuid.NewString()
+	s.create(ctx, t, streamID)
+	_, err = s.add(ctx, t, streamID, &streamlib.AddMessagesInput{Records: streamMsgs("", "a")})
+	require.NoError(t, err)
+	require.Equal(t, []string{"a"}, bodies(s.poll(ctx, t, streamID, 0).GetRecords()))
+	_, err = c.env.FrontendClient().DescribeChannel(c.ctx(), &workflowservice.DescribeChannelRequest{
+		Namespace: c.ns, Channel: stream.ChannelName(streamID),
+	})
+	requireNotFound(t, err)
+}