feat(events): structured event framework + namespace & replication li… (#10890)

…fecycle events

Add common/events: a pluggable events.Handler for emitting structured
("wide") events, with a typed Encoder (incl. Any for whole-object
values), a global event registry (NewEventDef) with startup
duplicate-name validation, a default handler that logs each event as one
line keyed by the event type, and a noop handler. Wired through the
server via WithCustomEventHandler, catalog validation at bootstrap, a
per-service fx provider, and a GetEventHandler() accessor on
ShardContext.

Define two events on the framework:
- NamespaceLifecycle: a generic, phase-discriminated namespace event
(stable identity fields + a nested "details" object). Emitters supply
the phases.
- ReplicationLifecycle: traces a replication task sent -> executing ->
applied across sync_workflow_state / sync_versioned_transition /
verify_versioned_transition, emitted at the stream sender, the passive
executables, and the ndc workflow-state replicator (post-apply
mutable-state summary, no extra read).

## What changed?
^
## Why?
Improve lifecycle observability

## How did you test it?
- [x] built
- [x] run locally and tested manually
- [ ] covered by existing tests
- [x] added new unit test(s)
- [ ] added new functional test(s)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Stephan Behnke <stephanos@users.noreply.github.com>

common/dynamicconfig/constants.go

@@ -1672,6 +1672,11 @@ leaves the membership ring, giving in-flight long-polls time to drain before the
 		true,
 		`EnableReplicationStream turn on replication stream`,
 	)
+	EmitReplicationLifecycleEvents = NewGlobalBoolSetting(
+		"history.emitReplicationLifecycleEvents",
+		false,
+		`EmitReplicationLifecycleEvents controls whether the history service emits ReplicationLifecycle wide events (sent/executing/applied phases). Cluster-level; default off.`,
+	)
 	EnableCloseInboundReplicationStreamOnShutdown = NewGlobalBoolSetting(
 		"history.enableCloseInboundReplicationStreamOnShutdown",
 		true,

common/wideevents/attrs.go

@@ -0,0 +1,18 @@
+package wideevents
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"go.opentelemetry.io/otel/log"
+)
+
+// jsonAttr records v as a compact JSON string under key rather than a nested structure, so the
+// logging / ingestion layer keeps one low-cardinality field (e.g. details) instead of flattening
+// into a dynamic key per leaf path. Used for composite fields (maps/slices/whole objects).
+func jsonAttr(key string, v any) log.KeyValue {
+	if b, err := json.Marshal(v); err == nil {
+		return log.String(key, string(b))
+	}
+	return log.String(key, fmt.Sprintf("%v", v))
+}

common/wideevents/events.go

@@ -0,0 +1,53 @@
+package wideevents
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/log"
+	"go.opentelemetry.io/otel/log/noop"
+)
+
+// instrumentationName is the OTEL instrumentation scope for events emitted by this package.
+const instrumentationName = "go.temporal.io/server/common/wideevents"
+
+// Payload is the data of one wide event. Each event type implements it by supplying its stable
+// event name and contributing its fields as OTEL log attributes.
+type Payload interface {
+	// EventName is the stable name of the event type. It becomes the log record's event name so
+	// events are easy to trace/grep by type.
+	EventName() string
+	// Attributes returns the event's fields as OTEL log key-values.
+	Attributes() []log.KeyValue
+}
+
+// NewLogger returns the OTEL logger used to emit events for serviceName. serviceName is attached
+// as an instrumentation-scope attribute (OTEL semconv service.name) so every emitted event carries
+// it, replacing per-event common-tag plumbing.
+func NewLogger(lp log.LoggerProvider, serviceName string) log.Logger {
+	return lp.Logger(
+		instrumentationName,
+		log.WithInstrumentationAttributes(attribute.String("service.name", serviceName)),
+	)
+}
+
+// NoopLogger returns a logger that discards all events. Safe default for tests and for
+// deployments that have not opted in to a real LoggerProvider.
+func NoopLogger() log.Logger {
+	return noop.NewLoggerProvider().Logger(instrumentationName)
+}
+
+// Emit writes p as a single OTEL log record via logger. A nil logger is a safe no-op so call sites
+// never need to guard. Batching, export, and serialization are handled by the LoggerProvider that
+// produced logger.
+func Emit(logger log.Logger, p Payload) {
+	if logger == nil {
+		return
+	}
+	var rec log.Record
+	rec.SetSeverity(log.SeverityInfo)
+	// The event type is the record's event name so events are easy to trace/grep by type.
+	rec.SetEventName(p.EventName())
+	rec.AddAttributes(p.Attributes()...)
+	logger.Emit(context.Background(), rec)
+}

common/wideevents/events_test.go

@@ -0,0 +1,96 @@
+package wideevents
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/log"
+	"go.opentelemetry.io/otel/log/embedded"
+)
+
+// captureLogger records emitted records for assertions.
+type captureLogger struct {
+	embedded.Logger
+	records []log.Record
+}
+
+func (c *captureLogger) Emit(_ context.Context, r log.Record) { c.records = append(c.records, r) }
+func (c *captureLogger) Enabled(context.Context, log.EnabledParameters) bool {
+	return true
+}
+
+// captureProvider hands out a single captureLogger and records the scope name and options it was
+// created with.
+type captureProvider struct {
+	embedded.LoggerProvider
+	logger *captureLogger
+	name   string
+	cfg    log.LoggerConfig
+}
+
+func (p *captureProvider) Logger(name string, opts ...log.LoggerOption) log.Logger {
+	p.name = name
+	p.cfg = log.NewLoggerConfig(opts...)
+	return p.logger
+}
+
+type sampleEvent struct {
+	A string
+	B int64
+}
+
+func (e sampleEvent) EventName() string { return "sample" }
+func (e sampleEvent) Attributes() []log.KeyValue {
+	return []log.KeyValue{log.String("a", e.A), log.Int64("b", e.B)}
+}
+
+func TestEmitWritesEventWithAttributes(t *testing.T) {
+	lg := &captureLogger{}
+
+	Emit(lg, sampleEvent{A: "x", B: 7})
+
+	require.Len(t, lg.records, 1)
+	rec := lg.records[0]
+	// The event type is the record's event name, for easy tracing/grepping.
+	require.Equal(t, "sample", rec.EventName())
+
+	got := map[string]log.Value{}
+	rec.WalkAttributes(func(kv log.KeyValue) bool {
+		got[kv.Key] = kv.Value
+		return true
+	})
+	require.Equal(t, "x", got["a"].AsString())
+	require.Equal(t, int64(7), got["b"].AsInt64())
+}
+
+func TestEmitNilLoggerIsNoop(t *testing.T) {
+	require.NotPanics(t, func() {
+		Emit(nil, sampleEvent{A: "x", B: 1})
+	})
+}
+
+func TestNewLoggerAttachesServiceNameAsScopeAttribute(t *testing.T) {
+	p := &captureProvider{logger: &captureLogger{}}
+	_ = NewLogger(p, "history")
+
+	require.Equal(t, instrumentationName, p.name)
+
+	var found bool
+	set := p.cfg.InstrumentationAttributes()
+	for iter := set.Iter(); iter.Next(); {
+		kv := iter.Attribute()
+		if kv.Key == attribute.Key("service.name") {
+			require.Equal(t, "history", kv.Value.AsString())
+			found = true
+		}
+	}
+	require.True(t, found, "service.name scope attribute present")
+}
+
+func TestNoopLoggerDiscards(t *testing.T) {
+	require.NotPanics(t, func() {
+		Emit(NoopLogger(), sampleEvent{A: "x", B: 1})
+	})
+}

common/wideevents/namespace_events.go

@@ -0,0 +1,32 @@
+package wideevents
+
+import "go.opentelemetry.io/otel/log"
+
+// NamespaceLifecycleEventName is the stable event name for the generic, phase-discriminated wide
+// event describing namespace-level activity (replication, failover, configuration, admission,
+// handover, etc.). This package owns only the stable envelope: the set of phase values and the
+// contents of Details are supplied by the emitter.
+const NamespaceLifecycleEventName = "namespace_lifecycle"
+
+// NamespaceLifecyclePayload is the NamespaceLifecycle payload. The identity fields are stable;
+// all phase-specific data goes in Details and is emitted as a single nested "details" object.
+type NamespaceLifecyclePayload struct {
+	Phase       string
+	Namespace   string
+	NamespaceID string
+	Details     map[string]any
+}
+
+func (p NamespaceLifecyclePayload) EventName() string { return NamespaceLifecycleEventName }
+
+func (p NamespaceLifecyclePayload) Attributes() []log.KeyValue {
+	attrs := []log.KeyValue{
+		log.String("phase", p.Phase),
+		log.String("namespace", p.Namespace),
+		log.String("namespace_id", p.NamespaceID),
+	}
+	if len(p.Details) > 0 {
+		attrs = append(attrs, jsonAttr("details", p.Details))
+	}
+	return attrs
+}

common/wideevents/namespace_events_test.go

@@ -0,0 +1,37 @@
+package wideevents
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNamespaceLifecycleEventName(t *testing.T) {
+	require.Equal(t, "namespace_lifecycle", NamespaceLifecyclePayload{}.EventName())
+}
+
+// TestNamespaceLifecycleFieldSetLocked pins the complete set of field names NamespaceLifecycle can
+// emit. This set is the event's published wire contract that downstream consumers depend on. If
+// this test fails you have added, removed, or renamed an emitted field: do so deliberately, get
+// the change reviewed, and then update `want` to match.
+func TestNamespaceLifecycleFieldSetLocked(t *testing.T) {
+	// want pins both the field set and the emitted values (the published wire contract).
+	// Composite fields (details) are emitted as a compact JSON string.
+	want := map[string]any{
+		"phase":        "route_computed",
+		"namespace":    "ns",
+		"namespace_id": "ns-id",
+		"details":      `{"k":"v"}`,
+	}
+
+	got := valueMap(NamespaceLifecyclePayload{
+		Phase:       "route_computed",
+		Namespace:   "ns",
+		NamespaceID: "ns-id",
+		Details:     map[string]any{"k": "v"},
+	}.Attributes())
+
+	require.Equal(t, want, got,
+		"NamespaceLifecycle emitted field set or values changed; this alters the event's published "+
+			"wire contract. Make the change deliberately, get it reviewed, then update `want`.")
+}

common/wideevents/replication_events.go

@@ -0,0 +1,200 @@
+package wideevents
+
+import (
+	"go.opentelemetry.io/otel/log"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+)
+
+// ReplicationLifecycleEventName is the stable event name for the ReplicationLifecycle wide event,
+// which traces a replication task sent -> executing -> applied.
+const ReplicationLifecycleEventName = "replication_lifecycle"
+
+type ReplicationPhase string
+
+const (
+	ReplicationSent      ReplicationPhase = "sent"
+	ReplicationExecuting ReplicationPhase = "executing"
+	ReplicationApplied   ReplicationPhase = "applied"
+)
+
+const (
+	ReplTaskSyncWorkflowState         = "sync_workflow_state"
+	ReplTaskSyncVersionedTransition   = "sync_versioned_transition"
+	ReplTaskVerifyVersionedTransition = "verify_versioned_transition"
+)
+
+type ReplicationLifecyclePayload struct {
+	Phase    ReplicationPhase
+	TaskType string
+	Shard    int32
+	// task identity + cross-phase join key: (namespace_id, workflow_id, run_id, transition_count, task_type)
+	Namespace       string
+	NamespaceID     string
+	WorkflowID      string
+	RunID           string
+	FailoverVersion int64
+	TransitionCount int64
+	// parent info: populated at every phase where mutable state is in hand (sent + applied) when
+	// this workflow is itself a child. Only the child->parent direction is emitted (a child points
+	// at its parent); the parent->children direction is intentionally not recorded here.
+	ParentWorkflowID  string
+	ParentRunID       string
+	ParentInitiatedID int64
+	Details           map[string]any
+	// sent-only
+	NewRunID     string
+	IsFirstSync  bool
+	FirstEventID int64
+	NextEventID  int64
+	// received-only
+	Attempt int32
+	// event_version_history is the (event_id, version) branch. It is emitted on executing (from the
+	// task) and applied (from the resulting mutable state); the version disambiguates which history
+	// branch the events are on.
+	EventVersionHistory []VersionHistoryEntry
+	// applied-only: post-apply mutable-state SUMMARY (no blob)
+	State               string
+	Status              string
+	AppliedNextEventID  int64
+	TransitionHistory   []VersionedTransitionEntry
+	LastEventID         int64
+	LastEventVersion    int64
+	Outcome             string
+	Error               string
+	NewExecutionRunID   string
+	ResetRunID          string
+	SignalCount         int64
+	ActivityCount       int64
+	UserTimerCount      int64
+	ChildExecutionCount int64
+	UpdateCount         int64
+}
+
+// VersionedTransitionEntry is one entry of a workflow's transition history.
+type VersionedTransitionEntry struct {
+	FailoverVersion int64 `json:"failover_version"`
+	TransitionCount int64 `json:"transition_count"`
+}
+
+// VersionHistoryEntry is one (event_id, version) point of an event version history branch.
+type VersionHistoryEntry struct {
+	EventID int64 `json:"event_id"`
+	Version int64 `json:"version"`
+}
+
+func (p ReplicationLifecyclePayload) EventName() string { return ReplicationLifecycleEventName }
+
+func (p ReplicationLifecyclePayload) Attributes() []log.KeyValue {
+	attrs := []log.KeyValue{
+		log.String("phase", string(p.Phase)),
+		log.String("task_type", p.TaskType),
+		log.Int64("shard", int64(p.Shard)),
+		log.String("namespace", p.Namespace),
+		log.String("namespace_id", p.NamespaceID),
+		log.String("workflow_id", p.WorkflowID),
+		log.String("run_id", p.RunID),
+	}
+	if p.FailoverVersion != 0 || p.TransitionCount != 0 {
+		attrs = append(attrs,
+			log.Int64("failover_version", p.FailoverVersion),
+			log.Int64("transition_count", p.TransitionCount),
+		)
+	}
+	// parent fields are phase-independent: emitted on any phase that populated them (sent +
+	// applied). Guards keep them absent when not applicable (e.g. executing, or a workflow that is
+	// not a child).
+	if p.ParentWorkflowID != "" {
+		attrs = append(attrs,
+			log.String("parent_workflow_id", p.ParentWorkflowID),
+			log.String("parent_run_id", p.ParentRunID),
+		)
+		if p.ParentInitiatedID != 0 {
+			attrs = append(attrs, log.Int64("parent_initiated_id", p.ParentInitiatedID))
+		}
+	}
+	if len(p.Details) > 0 {
+		attrs = append(attrs, jsonAttr("details", p.Details))
+	}
+	if len(p.EventVersionHistory) > 0 {
+		attrs = append(attrs, jsonAttr("event_version_history", p.EventVersionHistory))
+	}
+	switch p.Phase {
+	case ReplicationSent:
+		attrs = p.appendSent(attrs)
+	case ReplicationExecuting:
+		attrs = append(attrs, log.Int64("attempt", int64(p.Attempt)))
+	case ReplicationApplied:
+		attrs = p.appendApplied(attrs)
+	default:
+	}
+	return attrs
+}
+
+func (p ReplicationLifecyclePayload) appendSent(attrs []log.KeyValue) []log.KeyValue {
+	if p.NewRunID != "" {
+		attrs = append(attrs, log.String("new_run_id", p.NewRunID))
+	}
+	attrs = append(attrs, log.Bool("is_first_sync", p.IsFirstSync))
+	if p.FirstEventID != 0 {
+		attrs = append(attrs, log.Int64("first_event_id", p.FirstEventID))
+	}
+	if p.NextEventID != 0 {
+		attrs = append(attrs, log.Int64("next_event_id", p.NextEventID))
+	}
+	return attrs
+}
+
+func (p ReplicationLifecyclePayload) appendApplied(attrs []log.KeyValue) []log.KeyValue {
+	attrs = append(attrs, log.String("outcome", p.Outcome))
+	if p.Error != "" {
+		attrs = append(attrs, log.String("error", p.Error))
+	}
+	if p.State == "" {
+		return attrs
+	}
+	attrs = append(attrs,
+		log.String("state", p.State),
+		log.String("status", p.Status),
+		log.Int64("applied_next_event_id", p.AppliedNextEventID),
+	)
+	if len(p.TransitionHistory) > 0 {
+		attrs = append(attrs, jsonAttr("transition_history", p.TransitionHistory))
+	}
+	attrs = append(attrs, log.Int64("last_event_id", p.LastEventID))
+	if p.LastEventVersion != 0 {
+		attrs = append(attrs, log.Int64("last_event_version", p.LastEventVersion))
+	}
+	if p.NewExecutionRunID != "" {
+		attrs = append(attrs, log.String("new_execution_run_id", p.NewExecutionRunID))
+	}
+	if p.ResetRunID != "" {
+		attrs = append(attrs, log.String("reset_run_id", p.ResetRunID))
+	}
+	if p.SignalCount != 0 {
+		attrs = append(attrs, log.Int64("signal_count", p.SignalCount))
+	}
+	if p.ActivityCount != 0 {
+		attrs = append(attrs, log.Int64("activity_count", p.ActivityCount))
+	}
+	if p.UserTimerCount != 0 {
+		attrs = append(attrs, log.Int64("user_timer_count", p.UserTimerCount))
+	}
+	if p.ChildExecutionCount != 0 {
+		attrs = append(attrs, log.Int64("child_execution_count", p.ChildExecutionCount))
+	}
+	if p.UpdateCount != 0 {
+		attrs = append(attrs, log.Int64("update_count", p.UpdateCount))
+	}
+	return attrs
+}
+
+// PopulateParentInfo records the parent workflow identity when info describes a child workflow.
+// A nil info or a workflow with no parent is a no-op, so call sites never need to guard.
+func (p *ReplicationLifecyclePayload) PopulateParentInfo(info *persistencespb.WorkflowExecutionInfo) {
+	if info.GetParentWorkflowId() == "" {
+		return
+	}
+	p.ParentWorkflowID = info.GetParentWorkflowId()
+	p.ParentRunID = info.GetParentRunId()
+	p.ParentInitiatedID = info.GetParentInitiatedId()
+}

common/wideevents/replication_events_test.go

@@ -0,0 +1,243 @@
+package wideevents
+
+import (
+	"maps"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/log"
+)
+
+// attrMap indexes a payload's emitted attributes by key for assertions.
+func attrMap(kvs []log.KeyValue) map[string]log.Value {
+	m := make(map[string]log.Value, len(kvs))
+	for _, kv := range kvs {
+		m[kv.Key] = kv.Value
+	}
+	return m
+}
+
+// valueToAny decodes an OTEL log.Value into a comparable Go value for value-level assertions.
+func valueToAny(v log.Value) any {
+	switch v.Kind() {
+	case log.KindString:
+		return v.AsString()
+	case log.KindInt64:
+		return v.AsInt64()
+	case log.KindFloat64:
+		return v.AsFloat64()
+	case log.KindBool:
+		return v.AsBool()
+	case log.KindBytes:
+		return v.AsBytes()
+	default:
+		return v.String()
+	}
+}
+
+// valueMap decodes a payload's emitted attributes into a key -> Go value map.
+func valueMap(kvs []log.KeyValue) map[string]any {
+	m := make(map[string]any, len(kvs))
+	for _, kv := range kvs {
+		m[kv.Key] = valueToAny(kv.Value)
+	}
+	return m
+}
+
+// mergeFields returns base with extra overlaid, without mutating either.
+func mergeFields(base, extra map[string]any) map[string]any {
+	m := make(map[string]any, len(base)+len(extra))
+	maps.Copy(m, base)
+	maps.Copy(m, extra)
+	return m
+}
+
+func TestReplicationLifecycleEventName(t *testing.T) {
+	require.Equal(t, "replication_lifecycle", ReplicationLifecyclePayload{}.EventName())
+}
+
+func TestReplicationLifecycleEncodeSent(t *testing.T) {
+	p := ReplicationLifecyclePayload{
+		Phase:           ReplicationSent,
+		TaskType:        ReplTaskSyncVersionedTransition,
+		Shard:           3,
+		NamespaceID:     "ns-id",
+		WorkflowID:      "wf-id",
+		RunID:           "run-id",
+		FailoverVersion: 5,
+		TransitionCount: 7,
+		IsFirstSync:     true,
+		FirstEventID:    1,
+		NextEventID:     8,
+	}
+	f := attrMap(p.Attributes())
+
+	require.Equal(t, "sent", f["phase"].AsString())
+	require.True(t, f["is_first_sync"].AsBool())
+	require.Equal(t, int64(1), f["first_event_id"].AsInt64())
+	require.Equal(t, int64(8), f["next_event_id"].AsInt64())
+	require.Equal(t, ReplTaskSyncVersionedTransition, f["task_type"].AsString())
+	require.Equal(t, int64(3), f["shard"].AsInt64())
+	require.Equal(t, "ns-id", f["namespace_id"].AsString())
+	require.Equal(t, "wf-id", f["workflow_id"].AsString())
+	require.Equal(t, "run-id", f["run_id"].AsString())
+	require.Equal(t, int64(5), f["failover_version"].AsInt64())
+	require.Equal(t, int64(7), f["transition_count"].AsInt64())
+	// applied-only fields must be absent for the sent phase.
+	_, ok := f["outcome"]
+	require.False(t, ok)
+}
+
+func TestReplicationLifecycleEncodeApplied(t *testing.T) {
+	p := ReplicationLifecyclePayload{
+		Phase:              ReplicationApplied,
+		TaskType:           ReplTaskSyncVersionedTransition,
+		Shard:              1,
+		NamespaceID:        "ns-id",
+		WorkflowID:         "wf-id",
+		RunID:              "run-id",
+		State:              "Running",
+		Status:             "Unspecified",
+		AppliedNextEventID: 10,
+		TransitionHistory:  []VersionedTransitionEntry{{FailoverVersion: 5, TransitionCount: 7}},
+		LastEventID:        9,
+		LastEventVersion:   5,
+		Outcome:            "applied",
+		NewExecutionRunID:  "new-run",
+		SignalCount:        6,
+		UpdateCount:        2,
+	}
+	f := attrMap(p.Attributes())
+
+	require.Equal(t, "applied", f["phase"].AsString())
+	require.Equal(t, "new-run", f["new_execution_run_id"].AsString())
+	require.Equal(t, int64(6), f["signal_count"].AsInt64())
+	require.Equal(t, int64(2), f["update_count"].AsInt64())
+	// zero-valued applied summary fields must be omitted.
+	_, ok := f["activity_count"]
+	require.False(t, ok)
+	require.Equal(t, "applied", f["outcome"].AsString())
+	require.Equal(t, "Running", f["state"].AsString())
+	require.Equal(t, "Unspecified", f["status"].AsString())
+	require.Equal(t, int64(10), f["applied_next_event_id"].AsInt64())
+	// composite fields are emitted as a compact JSON string.
+	require.JSONEq(t, `[{"failover_version":5,"transition_count":7}]`, f["transition_history"].AsString())
+	require.Equal(t, int64(9), f["last_event_id"].AsInt64())
+	require.Equal(t, int64(5), f["last_event_version"].AsInt64())
+	// sent-only fields must be absent.
+	_, ok = f["is_first_sync"]
+	require.False(t, ok)
+}
+
+func TestEmitReplicationLifecycleNilSafe(t *testing.T) {
+	require.NotPanics(t, func() {
+		Emit(nil, ReplicationLifecyclePayload{Phase: ReplicationSent})
+	})
+}
+
+// fullyPopulatedReplication returns a payload with every field set to a non-zero value so that
+// Attributes exercises every conditional branch. Phase-specific fields are only emitted for their
+// phase, so callers must union the results across all phases to see the full field set.
+func fullyPopulatedReplication(phase ReplicationPhase) ReplicationLifecyclePayload {
+	return ReplicationLifecyclePayload{
+		Phase:               phase,
+		TaskType:            ReplTaskSyncVersionedTransition,
+		Shard:               1,
+		Namespace:           "ns",
+		NamespaceID:         "ns-id",
+		WorkflowID:          "wf-id",
+		RunID:               "run-id",
+		FailoverVersion:     5,
+		TransitionCount:     7,
+		ParentWorkflowID:    "p-wf",
+		ParentRunID:         "p-run",
+		ParentInitiatedID:   3,
+		Details:             map[string]any{"k": "v"},
+		NewRunID:            "new-run",
+		IsFirstSync:         true,
+		FirstEventID:        1,
+		NextEventID:         8,
+		Attempt:             2,
+		EventVersionHistory: []VersionHistoryEntry{{EventID: 9, Version: 5}},
+		State:               "Running",
+		Status:              "Unspecified",
+		AppliedNextEventID:  10,
+		TransitionHistory:   []VersionedTransitionEntry{{FailoverVersion: 5, TransitionCount: 7}},
+		LastEventID:         9,
+		LastEventVersion:    5,
+		Outcome:             "applied",
+		Error:               "boom",
+		NewExecutionRunID:   "ne-run",
+		ResetRunID:          "reset-run",
+		SignalCount:         6,
+		ActivityCount:       4,
+		UserTimerCount:      3,
+		ChildExecutionCount: 2,
+		UpdateCount:         1,
+	}
+}
+
+// TestReplicationLifecycleFieldSetLocked pins the complete set of field names ReplicationLifecycle
+// can emit, across every phase. This set is the event's published wire contract that downstream
+// consumers depend on. If this test fails you have added, removed, or renamed an emitted field:
+// do so deliberately, get the change reviewed, and then update `want` to match.
+func TestReplicationLifecycleFieldSetLocked(t *testing.T) {
+	// base pins the phase-independent fields (and values) every phase emits. Composite fields
+	// (details, event_version_history, transition_history) are emitted as compact JSON strings.
+	base := map[string]any{
+		"task_type":             ReplTaskSyncVersionedTransition,
+		"shard":                 int64(1),
+		"namespace":             "ns",
+		"namespace_id":          "ns-id",
+		"workflow_id":           "wf-id",
+		"run_id":                "run-id",
+		"failover_version":      int64(5),
+		"transition_count":      int64(7),
+		"parent_workflow_id":    "p-wf",
+		"parent_run_id":         "p-run",
+		"parent_initiated_id":   int64(3),
+		"details":               `{"k":"v"}`,
+		"event_version_history": `[{"event_id":9,"version":5}]`,
+	}
+	// want pins each phase's complete emitted field set AND values — the published wire contract.
+	// If this fails you have added, removed, renamed, or changed the value of an emitted field:
+	// do so deliberately, get it reviewed, then update `want`.
+	want := map[ReplicationPhase]map[string]any{
+		ReplicationSent: mergeFields(base, map[string]any{
+			"phase":          "sent",
+			"new_run_id":     "new-run",
+			"is_first_sync":  true,
+			"first_event_id": int64(1),
+			"next_event_id":  int64(8),
+		}),
+		ReplicationExecuting: mergeFields(base, map[string]any{
+			"phase":   "executing",
+			"attempt": int64(2),
+		}),
+		ReplicationApplied: mergeFields(base, map[string]any{
+			"phase":                 "applied",
+			"outcome":               "applied",
+			"error":                 "boom",
+			"state":                 "Running",
+			"status":                "Unspecified",
+			"applied_next_event_id": int64(10),
+			"transition_history":    `[{"failover_version":5,"transition_count":7}]`,
+			"last_event_id":         int64(9),
+			"last_event_version":    int64(5),
+			"new_execution_run_id":  "ne-run",
+			"reset_run_id":          "reset-run",
+			"signal_count":          int64(6),
+			"activity_count":        int64(4),
+			"user_timer_count":      int64(3),
+			"child_execution_count": int64(2),
+			"update_count":          int64(1),
+		}),
+	}
+
+	for phase, wantFields := range want {
+		require.Equal(t, wantFields, valueMap(fullyPopulatedReplication(phase).Attributes()),
+			"ReplicationLifecycle %q emitted field set or values changed; this alters the event's "+
+				"published wire contract. Make the change deliberately, get it reviewed, then "+
+				"update `want`.", phase)
+	}
+}

go.mod

@@ -57,14 +57,15 @@ require (
 	github.com/urfave/cli/v2 v2.27.7
 	go.opentelemetry.io/collector/pdata v1.56.0
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0
-	go.opentelemetry.io/otel v1.43.0
+	go.opentelemetry.io/otel v1.44.0
 	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.43.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.43.0
 	go.opentelemetry.io/otel/exporters/prometheus v0.56.0
-	go.opentelemetry.io/otel/metric v1.43.0
+	go.opentelemetry.io/otel/log v0.20.0
+	go.opentelemetry.io/otel/metric v1.44.0
 	go.opentelemetry.io/otel/sdk v1.43.0
 	go.opentelemetry.io/otel/sdk/metric v1.43.0
-	go.opentelemetry.io/otel/trace v1.43.0
+	go.opentelemetry.io/otel/trace v1.44.0
 	go.temporal.io/api v1.63.3
 	go.temporal.io/auto-scaled-workers v0.0.0-20260706201056-4320b34799ee
 	go.temporal.io/sdk v1.41.1

go.sum

@@ -449,8 +449,8 @@ go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.6
 go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0/go.mod h1:Sje3i3MjSPKTSPvVWCaL8ugBzJwik3u4smCjUeuupqg=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 h1:CqXxU8VOmDefoh0+ztfGaymYbhdB/tT3zs79QaZTNGY=
 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0/go.mod h1:BuhAPThV8PBHBvg8ZzZ/Ok3idOdhWIodywz2xEcRbJo=
-go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
-go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0=
+go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU=
+go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.43.0 h1:8UQVDcZxOJLtX6gxtDt3vY2WTgvZqMQRzjsqiIHQdkc=
 go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.43.0/go.mod h1:2lmweYCiHYpEjQ/lSJBYhj9jP1zvCvQW4BqL9dnT7FQ=
 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 h1:88Y4s2C8oTui1LGM6bTWkw0ICGcOLCAI5l6zsD1j20k=
@@ -461,14 +461,16 @@ go.opentelemetry.io/otel/exporters/prometheus v0.56.0 h1:GnCIi0QyG0yy2MrJLzVrIM7
 go.opentelemetry.io/otel/exporters/prometheus v0.56.0/go.mod h1:JQcVZtbIIPM+7SWBB+T6FK+xunlyidwLp++fN0sUaOk=
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.43.0 h1:TC+BewnDpeiAmcscXbGMfxkO+mwYUwE/VySwvw88PfA=
 go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.43.0/go.mod h1:J/ZyF4vfPwsSr9xJSPyQ4LqtcTPULFR64KwTikGLe+A=
-go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM=
-go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY=
+go.opentelemetry.io/otel/log v0.20.0 h1:/5i0vuHxCLWUfChWG41K9wkM0jafruPw9NU1/RCJirs=
+go.opentelemetry.io/otel/log v0.20.0/go.mod h1:wOcMcjsZpG8x7Bak7IhSi/lg8wscV2C1VdrKCLPlt0E=
+go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc=
+go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo=
 go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg=
 go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg=
 go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw=
 go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A=
-go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A=
-go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
+go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk=
+go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE=
 go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g=
 go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk=
 go.opentelemetry.io/proto/slim/otlp v1.10.0 h1:iR97Vs/ZDR+y9TfuP9b1XBtdPWeC+OMslIBmhcLU7jM=

service/history/configs/config.go

@@ -13,6 +13,7 @@ type Config struct {
 	NumberOfShards int32
 
 	EnableReplicationStream                       dynamicconfig.BoolPropertyFn
+	EmitReplicationLifecycleEvents                dynamicconfig.BoolPropertyFn
 	EnableCloseInboundReplicationStreamOnShutdown dynamicconfig.BoolPropertyFn
 	EnableSeparateReplicationEnableFlag           dynamicconfig.BoolPropertyFn
 	HistoryReplicationDLQV2                       dynamicconfig.BoolPropertyFn
@@ -444,6 +445,7 @@ func NewConfig(
 		NumberOfShards: numberOfShards,
 
 		EnableReplicationStream:                       dynamicconfig.EnableReplicationStream.Get(dc),
+		EmitReplicationLifecycleEvents:                dynamicconfig.EmitReplicationLifecycleEvents.Get(dc),
 		EnableCloseInboundReplicationStreamOnShutdown: dynamicconfig.EnableCloseInboundReplicationStreamOnShutdown.Get(dc),
 		EnableSeparateReplicationEnableFlag:           dynamicconfig.EnableSeparateReplicationEnableFlag.Get(dc),
 		HistoryReplicationDLQV2:                       dynamicconfig.EnableHistoryReplicationDLQV2.Get(dc),

service/history/history_engine.go

@@ -280,6 +280,7 @@ func NewEngineWithShardContext(
 			serializer,
 			persistenceRateLimiter,
 			logger,
+			shard.GetEventLogger(),
 		)
 		historyEngImpl.nDCHSMStateReplicator = ndc.NewHSMStateReplicator(
 			shard,

service/history/interfaces/shard_context.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"time"
 
+	otellog "go.opentelemetry.io/otel/log"
 	commonpb "go.temporal.io/api/common/v1"
 	"go.temporal.io/server/api/adminservice/v1"
 	clockspb "go.temporal.io/server/api/clock/v1"
@@ -46,6 +47,7 @@ type (
 		GetLogger() log.Logger
 		GetThrottledLogger() log.Logger
 		GetMetricsHandler() metrics.Handler
+		GetEventLogger() otellog.Logger
 		GetTimeSource() clock.TimeSource
 
 		GetRemoteAdminClient(string) (adminservice.AdminServiceClient, error)

service/history/interfaces/shard_context_mock.go

@@ -14,6 +14,7 @@ import (
 	reflect "reflect"
 	time "time"
 
+	log "go.opentelemetry.io/otel/log"
 	common "go.temporal.io/api/common/v1"
 	adminservice "go.temporal.io/server/api/adminservice/v1"
 	clock "go.temporal.io/server/api/clock/v1"
@@ -26,7 +27,7 @@ import (
 	definition "go.temporal.io/server/common/definition"
 	finalizer "go.temporal.io/server/common/finalizer"
 	locks "go.temporal.io/server/common/locks"
-	log "go.temporal.io/server/common/log"
+	log0 "go.temporal.io/server/common/log"
 	metrics "go.temporal.io/server/common/metrics"
 	namespace "go.temporal.io/server/common/namespace"
 	persistence0 "go.temporal.io/server/common/persistence"
@@ -369,6 +370,20 @@ func (mr *MockShardContextMockRecorder) GetEngine(ctx any) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEngine", reflect.TypeOf((*MockShardContext)(nil).GetEngine), ctx)
 }
 
+// GetEventLogger mocks base method.
+func (m *MockShardContext) GetEventLogger() log.Logger {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetEventLogger")
+	ret0, _ := ret[0].(log.Logger)
+	return ret0
+}
+
+// GetEventLogger indicates an expected call of GetEventLogger.
+func (mr *MockShardContextMockRecorder) GetEventLogger() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEventLogger", reflect.TypeOf((*MockShardContext)(nil).GetEventLogger))
+}
+
 // GetEventsCache mocks base method.
 func (m *MockShardContext) GetEventsCache() events.Cache {
 	m.ctrl.T.Helper()
@@ -455,10 +470,10 @@ func (mr *MockShardContextMockRecorder) GetLifecycleContext() *gomock.Call {
 }
 
 // GetLogger mocks base method.
-func (m *MockShardContext) GetLogger() log.Logger {
+func (m *MockShardContext) GetLogger() log0.Logger {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetLogger")
-	ret0, _ := ret[0].(log.Logger)
+	ret0, _ := ret[0].(log0.Logger)
 	return ret0
 }
 
@@ -655,10 +670,10 @@ func (mr *MockShardContextMockRecorder) GetShardID() *gomock.Call {
 }
 
 // GetThrottledLogger mocks base method.
-func (m *MockShardContext) GetThrottledLogger() log.Logger {
+func (m *MockShardContext) GetThrottledLogger() log0.Logger {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetThrottledLogger")
-	ret0, _ := ret[0].(log.Logger)
+	ret0, _ := ret[0].(log0.Logger)
 	return ret0
 }
 
@@ -1200,6 +1215,20 @@ func (mr *MockControllableContextMockRecorder) GetEngine(ctx any) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEngine", reflect.TypeOf((*MockControllableContext)(nil).GetEngine), ctx)
 }
 
+// GetEventLogger mocks base method.
+func (m *MockControllableContext) GetEventLogger() log.Logger {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetEventLogger")
+	ret0, _ := ret[0].(log.Logger)
+	return ret0
+}
+
+// GetEventLogger indicates an expected call of GetEventLogger.
+func (mr *MockControllableContextMockRecorder) GetEventLogger() *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetEventLogger", reflect.TypeOf((*MockControllableContext)(nil).GetEventLogger))
+}
+
 // GetEventsCache mocks base method.
 func (m *MockControllableContext) GetEventsCache() events.Cache {
 	m.ctrl.T.Helper()
@@ -1286,10 +1315,10 @@ func (mr *MockControllableContextMockRecorder) GetLifecycleContext() *gomock.Cal
 }
 
 // GetLogger mocks base method.
-func (m *MockControllableContext) GetLogger() log.Logger {
+func (m *MockControllableContext) GetLogger() log0.Logger {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetLogger")
-	ret0, _ := ret[0].(log.Logger)
+	ret0, _ := ret[0].(log0.Logger)
 	return ret0
 }
 
@@ -1500,10 +1529,10 @@ func (mr *MockControllableContextMockRecorder) GetShardID() *gomock.Call {
 }
 
 // GetThrottledLogger mocks base method.
-func (m *MockControllableContext) GetThrottledLogger() log.Logger {
+func (m *MockControllableContext) GetThrottledLogger() log0.Logger {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "GetThrottledLogger")
-	ret0, _ := ret[0].(log.Logger)
+	ret0, _ := ret[0].(log0.Logger)
 	return ret0
 }
 

service/history/ndc/workflow_state_replicator.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	otellog "go.opentelemetry.io/otel/log"
 	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	historypb "go.temporal.io/api/history/v1"
@@ -41,6 +42,7 @@ import (
 	"go.temporal.io/server/common/quotas"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/common/softassert"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/history/consts"
 	"go.temporal.io/server/service/history/events"
 	"go.temporal.io/server/service/history/historybuilder"
@@ -74,6 +76,7 @@ type (
 		persistenceRateLimiter       quotas.RequestRateLimiter
 		enablePersistenceRateLimiter dynamicconfig.BoolPropertyFnWithNamespaceFilter
 		logger                       log.Logger
+		eventLogger                  otellog.Logger
 		taskRefresher                workflow.TaskRefresher
 	}
 )
@@ -85,6 +88,7 @@ func NewWorkflowStateReplicator(
 	eventSerializer serialization.Serializer,
 	persistenceRateLimiter quotas.RequestRateLimiter,
 	logger log.Logger,
+	eventLogger otellog.Logger,
 ) *WorkflowStateReplicatorImpl {
 
 	logger = log.With(logger, tag.ComponentWorkflowStateReplicator)
@@ -99,6 +103,7 @@ func NewWorkflowStateReplicator(
 		persistenceRateLimiter:       persistenceRateLimiter,
 		enablePersistenceRateLimiter: shardContext.GetConfig().EnableHistoryReplicationRateLimiter,
 		logger:                       logger,
+		eventLogger:                  eventLogger,
 		taskRefresher:                workflow.NewTaskRefresher(shardContext),
 	}
 }
@@ -235,6 +240,20 @@ func (r *WorkflowStateReplicatorImpl) ReplicateVersionedTransition(
 	wid := executionInfo.GetWorkflowId()
 	rid := executionState.GetRunId()
 
+	// emitApplied gates the best-effort "applied" lifecycle event; computed once here so the gate
+	// lives at the call site, like the other replication lifecycle emitters.
+	emitApplied := r.eventLogger != nil && r.shardContext.GetConfig().EmitReplicationLifecycleEvents()
+	// ms is the mutable state being applied; appliedMS is a snapshot of its post-apply state taken
+	// under the workflow lock (see the releaseFn wrapper below) for the deferred emit. Reading the
+	// live ms after the lock is released would race with the next writer.
+	var ms historyi.MutableState
+	var appliedMS *persistencespb.WorkflowMutableState
+	defer func() {
+		if emitApplied && retError == nil {
+			r.emitReplicationVersionedTransitionApplied(namespaceID, wid, rid, appliedMS)
+		}
+	}()
+
 	wfCtx, releaseFn, err := r.workflowCache.GetOrCreateChasmExecution(
 		ctx,
 		r.shardContext,
@@ -249,6 +268,18 @@ func (r *WorkflowStateReplicatorImpl) ReplicateVersionedTransition(
 	if err != nil {
 		return err
 	}
+	if emitApplied {
+		// Snapshot the post-apply mutable state at release time: the only point that is both after
+		// the apply (which mutates ms in place) and still under the workflow lock, since the apply
+		// releases via the transaction manager.
+		innerReleaseFn := releaseFn
+		releaseFn = func(err error) {
+			if err == nil && appliedMS == nil && ms != nil {
+				appliedMS = ms.CloneToProto()
+			}
+			innerReleaseFn(err)
+		}
+	}
 	defer func() {
 		if rec := recover(); rec != nil {
 			releaseFn(errPanic)
@@ -265,7 +296,7 @@ func (r *WorkflowStateReplicatorImpl) ReplicateVersionedTransition(
 		}
 	}
 
-	ms, err := wfCtx.LoadMutableState(ctx, r.shardContext)
+	ms, err = wfCtx.LoadMutableState(ctx, r.shardContext)
 	switch err.(type) {
 	case *serviceerror.NotFound:
 		return r.applySnapshot(ctx, namespaceID, wid, rid, archetypeID, wfCtx, releaseFn, nil, versionedTransitionArtifact, sourceClusterName)
@@ -342,6 +373,83 @@ func (r *WorkflowStateReplicatorImpl) ReplicateVersionedTransition(
 	}
 }
 
+// emitReplicationVersionedTransitionApplied emits a best-effort "applied" ReplicationLifecycle
+// event summarizing post-apply mutable state. ms may be nil (e.g. a fresh apply via the NotFound
+// path) in which case only identity fields are populated. It never affects control flow.
+func (r *WorkflowStateReplicatorImpl) emitReplicationVersionedTransitionApplied(
+	namespaceID namespace.ID,
+	workflowID string,
+	runID string,
+	ms *persistencespb.WorkflowMutableState,
+) {
+	var nsName string
+	if name, err := r.namespaceRegistry.GetNamespaceName(namespaceID); err == nil {
+		nsName = name.String()
+	}
+
+	payload := wideevents.ReplicationLifecyclePayload{
+		Phase:       wideevents.ReplicationApplied,
+		TaskType:    wideevents.ReplTaskSyncVersionedTransition,
+		Shard:       r.shardContext.GetShardID(),
+		Namespace:   nsName,
+		NamespaceID: namespaceID.String(),
+		WorkflowID:  workflowID,
+		RunID:       runID,
+		Outcome:     "applied",
+	}
+	if ms != nil {
+		info := ms.GetExecutionInfo()
+		state := ms.GetExecutionState()
+		payload.State = state.GetState().String()
+		payload.Status = state.GetStatus().String()
+		payload.AppliedNextEventID = ms.GetNextEventId()
+		if th := info.GetTransitionHistory(); len(th) > 0 {
+			entries := make([]wideevents.VersionedTransitionEntry, 0, len(th))
+			for _, vt := range th {
+				entries = append(entries, wideevents.VersionedTransitionEntry{
+					FailoverVersion: vt.GetNamespaceFailoverVersion(),
+					TransitionCount: vt.GetTransitionCount(),
+				})
+			}
+			payload.TransitionHistory = entries
+		}
+		if currentHistory, err := versionhistory.GetCurrentVersionHistory(info.GetVersionHistories()); err == nil {
+			if lastItem, itemErr := versionhistory.GetLastVersionHistoryItem(currentHistory); itemErr == nil {
+				payload.LastEventID = lastItem.GetEventId()
+				payload.LastEventVersion = lastItem.GetVersion()
+			}
+			items := currentHistory.GetItems()
+			history := make([]wideevents.VersionHistoryEntry, 0, len(items))
+			for _, item := range items {
+				history = append(history, wideevents.VersionHistoryEntry{
+					EventID: item.GetEventId(),
+					Version: item.GetVersion(),
+				})
+			}
+			payload.EventVersionHistory = history
+		}
+		populateAppliedExecutionSummary(&payload, info)
+		payload.PopulateParentInfo(info)
+	}
+
+	wideevents.Emit(r.eventLogger, payload)
+}
+
+// populateAppliedExecutionSummary fills the post-apply mutable-state summary fields shared across
+// applied lifecycle hooks. It is best-effort and nil-safe.
+func populateAppliedExecutionSummary(
+	payload *wideevents.ReplicationLifecyclePayload,
+	info *persistencespb.WorkflowExecutionInfo,
+) {
+	payload.NewExecutionRunID = info.GetNewExecutionRunId()
+	payload.ResetRunID = info.GetResetRunId()
+	payload.SignalCount = info.GetSignalCount()
+	payload.ActivityCount = info.GetActivityCount()
+	payload.UserTimerCount = info.GetUserTimerCount()
+	payload.ChildExecutionCount = info.GetChildExecutionCount()
+	payload.UpdateCount = info.GetUpdateCount()
+}
+
 func parseVersionedTransitionAttributes(
 	versionedTransition *replicationspb.VersionedTransitionArtifact,
 ) (

service/history/ndc/workflow_state_replicator_test.go

@@ -115,6 +115,7 @@ func (s *workflowReplicatorSuite) SetupTest() {
 		s.serializer,
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 }
 
@@ -611,6 +612,7 @@ func (s *workflowReplicatorSuite) Test_ReplicateVersionedTransition_SameBranch_S
 		s.serializer,
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)
@@ -703,6 +705,7 @@ func (s *workflowReplicatorSuite) Test_ReplicateVersionedTransition_DifferentBra
 		s.serializer,
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)
@@ -789,6 +792,7 @@ func (s *workflowReplicatorSuite) Test_ReplicateVersionedTransition_SameBranch_S
 		s.serializer,
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)
@@ -884,6 +888,7 @@ func (s *workflowReplicatorSuite) Test_ReplicateVersionedTransition_FirstTask_Sy
 		s.serializer,
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)
@@ -965,6 +970,7 @@ func (s *workflowReplicatorSuite) Test_ReplicateVersionedTransition_MutationProv
 		s.serializer,
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)
@@ -1596,6 +1602,7 @@ func (s *workflowReplicatorSuite) Test_handleFirstReplicationTask_WithSnapshot_S
 		serialization.NewSerializer(),
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)
@@ -1674,6 +1681,7 @@ func (s *workflowReplicatorSuite) Test_handleFirstReplicationTask_WithMutation_S
 		serialization.NewSerializer(),
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)
@@ -1747,6 +1755,7 @@ func (s *workflowReplicatorSuite) Test_handleFirstReplicationTask_InvalidArtifac
 		serialization.NewSerializer(),
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 
 	versionedTransitionArtifact := &replicationspb.VersionedTransitionArtifact{}
@@ -1772,6 +1781,7 @@ func (s *workflowReplicatorSuite) Test_handleFirstReplicationTask_CreateWorkflow
 		serialization.NewSerializer(),
 		quotas.NoopRequestRateLimiter,
 		s.logger,
+		nil,
 	)
 	mockTransactionManager := NewMockTransactionManager(s.controller)
 	mockTaskRefresher := workflow.NewMockTaskRefresher(s.controller)

service/history/replication/executable_sync_versioned_transition_task.go

@@ -17,6 +17,7 @@ import (
 	"go.temporal.io/server/common/persistence/versionhistory"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	ctasks "go.temporal.io/server/common/tasks"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/history/consts"
 )
 
@@ -74,6 +75,10 @@ func (e *ExecutableSyncVersionedTransitionTask) Execute() error {
 	}
 	e.MarkExecutionStart()
 
+	if e.Config.EmitReplicationLifecycleEvents() {
+		emitReplicationExecuting(e.ProcessToolBox, e.ReplicationTask(), e.WorkflowKey, wideevents.ReplTaskSyncVersionedTransition, int32(e.Attempt()))
+	}
+
 	callerInfo := getReplicaitonCallerInfo(e.GetPriority())
 	namespaceName, apply, nsError := e.GetNamespaceInfo(headers.SetCallerInfo(
 		context.Background(),

service/history/replication/executable_verify_versioned_transition_task.go

@@ -24,6 +24,7 @@ import (
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/common/softassert"
 	ctasks "go.temporal.io/server/common/tasks"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/history/consts"
 )
 
@@ -75,12 +76,26 @@ func (e *ExecutableVerifyVersionedTransitionTask) QueueID() any {
 	return e.WorkflowKey
 }
 
-func (e *ExecutableVerifyVersionedTransitionTask) Execute() error {
+func (e *ExecutableVerifyVersionedTransitionTask) Execute() (retErr error) {
 	if e.TerminalState() {
 		return nil
 	}
 	e.MarkExecutionStart()
 
+	emitLifecycle := e.Config.EmitReplicationLifecycleEvents()
+	if emitLifecycle {
+		emitReplicationExecuting(e.ProcessToolBox, e.ReplicationTask(), e.WorkflowKey, wideevents.ReplTaskVerifyVersionedTransition, int32(e.Attempt()))
+	}
+
+	// inspectedMS is the mutable-state snapshot examined during verification, captured for the
+	// best-effort "applied" lifecycle event emitted below.
+	var inspectedMS *persistencespb.WorkflowMutableState
+	defer func() {
+		if emitLifecycle {
+			e.emitReplicationVerifyApplied(inspectedMS, retErr)
+		}
+	}()
+
 	callerInfo := getReplicaitonCallerInfo(e.GetPriority())
 	namespaceName, apply, nsError := e.GetNamespaceInfo(headers.SetCallerInfo(
 		context.Background(),
@@ -107,6 +122,7 @@ func (e *ExecutableVerifyVersionedTransitionTask) Execute() error {
 	defer cancel()
 
 	ms, err := e.getMutableState(ctx, e.RunID)
+	inspectedMS = ms
 	if err != nil {
 		switch err.(type) {
 		case *serviceerror.NotFound:

service/history/replication/executable_workflow_state_task.go

@@ -17,6 +17,7 @@ import (
 	"go.temporal.io/server/common/namespace"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	ctasks "go.temporal.io/server/common/tasks"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/history/consts"
 )
 
@@ -81,6 +82,10 @@ func (e *ExecutableWorkflowStateTask) Execute() error {
 	}
 	e.MarkExecutionStart()
 
+	if e.Config.EmitReplicationLifecycleEvents() {
+		emitReplicationExecuting(e.ProcessToolBox, e.ReplicationTask(), e.WorkflowKey, wideevents.ReplTaskSyncWorkflowState, int32(e.Attempt()))
+	}
+
 	callerInfo := getReplicaitonCallerInfo(e.GetPriority())
 	namespaceName, apply, err := e.GetNamespaceInfo(headers.SetCallerInfo(
 		context.Background(),

service/history/replication/replication_events.go

@@ -0,0 +1,264 @@
+package replication
+
+import (
+	"errors"
+
+	enumsspb "go.temporal.io/server/api/enums/v1"
+	historyspb "go.temporal.io/server/api/history/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	replicationspb "go.temporal.io/server/api/replication/v1"
+	"go.temporal.io/server/common/definition"
+	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/persistence/versionhistory"
+	serviceerrors "go.temporal.io/server/common/serviceerror"
+	"go.temporal.io/server/common/wideevents"
+	"go.temporal.io/server/service/history/tasks"
+)
+
+// emitReplicationExecuting emits a best-effort "executing" ReplicationLifecycle event when an
+// executable replication task is picked up to execute on the target cluster. It never affects
+// control flow: a nil logger / unresolved shard is a no-op and namespace resolution failures
+// fall back to "".
+//
+// The event logger and shard id are taken from the target shard (resolved by namespace+workflow),
+// because the replication ProcessToolBox does not carry the event logger in all wirings.
+func emitReplicationExecuting(
+	toolBox ProcessToolBox,
+	task *replicationspb.ReplicationTask,
+	key definition.WorkflowKey,
+	taskType string,
+	attempt int32,
+) {
+	shardContext, err := toolBox.ShardController.GetShardByNamespaceWorkflow(namespace.ID(key.NamespaceID), key.WorkflowID)
+	if err != nil {
+		return
+	}
+	logger := shardContext.GetEventLogger()
+	if logger == nil {
+		return
+	}
+
+	var nsName string
+	if name, err := toolBox.NamespaceCache.GetNamespaceName(namespace.ID(key.NamespaceID)); err == nil {
+		nsName = name.String()
+	}
+
+	payload := wideevents.ReplicationLifecyclePayload{
+		Phase:       wideevents.ReplicationExecuting,
+		TaskType:    taskType,
+		Shard:       shardContext.GetShardID(),
+		Namespace:   nsName,
+		NamespaceID: key.NamespaceID,
+		WorkflowID:  key.WorkflowID,
+		RunID:       key.RunID,
+		Attempt:     attempt,
+	}
+	// Record what this attempt will try to apply, taken from the task itself: its target versioned
+	// transition and, for verify, the expected event version history. This lets a reader correlate
+	// the executing attempt with the eventual applied outcome and history branch.
+	if vt := task.GetVersionedTransition(); vt != nil {
+		payload.FailoverVersion = vt.GetNamespaceFailoverVersion()
+		payload.TransitionCount = vt.GetTransitionCount()
+	}
+	if items := task.GetVerifyVersionedTransitionTaskAttributes().GetEventVersionHistory(); len(items) > 0 {
+		payload.EventVersionHistory = versionHistoryEntries(items)
+	}
+	wideevents.Emit(logger, payload)
+}
+
+// emitReplicationSent emits a best-effort "sent" ReplicationLifecycle event for the supported
+// replication task types. It never affects control flow.
+func (s *StreamSenderImpl) emitReplicationSent(
+	task *replicationspb.ReplicationTask,
+	item tasks.Task,
+) {
+	logger := s.shardContext.GetEventLogger()
+	if logger == nil {
+		return
+	}
+
+	var taskType string
+	switch task.GetTaskType() {
+	case enumsspb.REPLICATION_TASK_TYPE_SYNC_WORKFLOW_STATE_TASK:
+		taskType = wideevents.ReplTaskSyncWorkflowState
+	case enumsspb.REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK:
+		taskType = wideevents.ReplTaskSyncVersionedTransition
+	case enumsspb.REPLICATION_TASK_TYPE_VERIFY_VERSIONED_TRANSITION_TASK:
+		taskType = wideevents.ReplTaskVerifyVersionedTransition
+	default:
+		return
+	}
+
+	nsID := item.GetNamespaceID()
+	nsName, err := s.shardContext.GetNamespaceRegistry().GetNamespaceName(namespace.ID(nsID))
+	if err != nil {
+		nsName = namespace.EmptyName
+	}
+
+	payload := wideevents.ReplicationLifecyclePayload{
+		Phase:       wideevents.ReplicationSent,
+		TaskType:    taskType,
+		Shard:       s.serverShardKey.ShardID,
+		Namespace:   nsName.String(),
+		NamespaceID: nsID,
+		WorkflowID:  item.GetWorkflowID(),
+		RunID:       item.GetRunID(),
+	}
+	if vt := task.GetVersionedTransition(); vt != nil {
+		payload.FailoverVersion = vt.GetNamespaceFailoverVersion()
+		payload.TransitionCount = vt.GetTransitionCount()
+	}
+
+	if attr := task.GetVerifyVersionedTransitionTaskAttributes(); attr != nil {
+		// verify ships no history batch, so use the task's target event id as next_event_id.
+		payload.NextEventID = attr.GetNextEventId()
+		payload.NewRunID = attr.GetNewRunId()
+	}
+
+	if rawTaskInfo := task.GetRawTaskInfo(); rawTaskInfo != nil {
+		payload.FirstEventID = rawTaskInfo.GetFirstEventId()
+		payload.NextEventID = rawTaskInfo.GetNextEventId()
+	} else if svtTask, ok := item.(*tasks.SyncVersionedTransitionTask); ok {
+		payload.FirstEventID = svtTask.FirstEventID
+		payload.NextEventID = svtTask.NextEventID
+	}
+	if attr := task.GetSyncVersionedTransitionTaskAttributes(); attr != nil {
+		payload.IsFirstSync = attr.GetVersionedTransitionArtifact().GetIsFirstSync()
+	}
+
+	// parent info, extracted from the mutable state carried in the task payload (child->parent
+	// only). The verify task ships no mutable state, so it contributes no parent info here.
+	populateSentParentInfo(&payload, task)
+
+	wideevents.Emit(logger, payload)
+}
+
+// populateSentParentInfo records the child->parent identity from the mutable state carried in the
+// task payload, when present. Task types that ship no mutable state (e.g. verify) are a no-op.
+func populateSentParentInfo(payload *wideevents.ReplicationLifecyclePayload, task *replicationspb.ReplicationTask) {
+	switch task.GetTaskType() {
+	case enumsspb.REPLICATION_TASK_TYPE_SYNC_WORKFLOW_STATE_TASK:
+		if ms := task.GetSyncWorkflowStateTaskAttributes().GetWorkflowState(); ms != nil {
+			payload.PopulateParentInfo(ms.GetExecutionInfo())
+		}
+	case enumsspb.REPLICATION_TASK_TYPE_SYNC_VERSIONED_TRANSITION_TASK:
+		art := task.GetSyncVersionedTransitionTaskAttributes().GetVersionedTransitionArtifact()
+		if art == nil {
+			return
+		}
+		if snap := art.GetSyncWorkflowStateSnapshotAttributes(); snap != nil {
+			if ms := snap.GetState(); ms != nil {
+				payload.PopulateParentInfo(ms.GetExecutionInfo())
+			}
+		} else if mut := art.GetSyncWorkflowStateMutationAttributes(); mut != nil {
+			if m := mut.GetStateMutation(); m != nil {
+				payload.PopulateParentInfo(m.GetExecutionInfo())
+			}
+		}
+	default:
+	}
+}
+
+// emitReplicationVerifyApplied emits a best-effort "applied" ReplicationLifecycle event for a
+// verify task. Outcome is "verified" when verification passed (no error) and "resend_needed" when
+// the task requested a state resend; any other error is reported with its message. ms may be nil
+// (e.g. the workflow was not found) in which case only identity fields are populated.
+func (e *ExecutableVerifyVersionedTransitionTask) emitReplicationVerifyApplied(
+	ms *persistencespb.WorkflowMutableState,
+	retErr error,
+) {
+	shardContext, err := e.ShardController.GetShardByNamespaceWorkflow(namespace.ID(e.NamespaceID), e.WorkflowID)
+	if err != nil {
+		return
+	}
+	logger := shardContext.GetEventLogger()
+	if logger == nil {
+		return
+	}
+
+	outcome := "verified"
+	errStr := ""
+	if retErr != nil {
+		var syncStateErr *serviceerrors.SyncState
+		if errors.As(retErr, &syncStateErr) {
+			outcome = "resend_needed"
+		} else {
+			outcome = "error"
+			errStr = retErr.Error()
+		}
+	}
+
+	var nsName string
+	if name, nsErr := e.NamespaceCache.GetNamespaceName(namespace.ID(e.NamespaceID)); nsErr == nil {
+		nsName = name.String()
+	}
+
+	payload := wideevents.ReplicationLifecyclePayload{
+		Phase:       wideevents.ReplicationApplied,
+		TaskType:    wideevents.ReplTaskVerifyVersionedTransition,
+		Shard:       shardContext.GetShardID(),
+		Namespace:   nsName,
+		NamespaceID: e.NamespaceID,
+		WorkflowID:  e.WorkflowID,
+		RunID:       e.RunID,
+		Outcome:     outcome,
+		Error:       errStr,
+	}
+	if vt := e.ReplicationTask().GetVersionedTransition(); vt != nil {
+		payload.FailoverVersion = vt.GetNamespaceFailoverVersion()
+		payload.TransitionCount = vt.GetTransitionCount()
+	}
+
+	// For verify, record an expected-vs-actual comparison in Details rather than a mutable-state
+	// summary. When verify triggers a resend, the state replicator emits its own applied event with
+	// the full summary, so repeating it here would only duplicate. "expected" is what the task
+	// requires the passive to have; "actual" is what the inspected mutable state currently has.
+	details := map[string]any{"expected": e.expectedState()}
+	if ms != nil {
+		details["actual"] = actualState(ms)
+	}
+	payload.Details = details
+
+	wideevents.Emit(logger, payload)
+}
+
+// expectedState describes what the verify task requires the passive to have: the target versioned
+// transition, next event id, and expected event version history.
+func (e *ExecutableVerifyVersionedTransitionTask) expectedState() map[string]any {
+	expected := map[string]any{"next_event_id": e.taskAttr.GetNextEventId()}
+	if vt := e.ReplicationTask().GetVersionedTransition(); vt != nil {
+		expected["versioned_transition"] = versionedTransitionEntry(vt)
+	}
+	if items := e.taskAttr.GetEventVersionHistory(); len(items) > 0 {
+		expected["event_version_history"] = versionHistoryEntries(items)
+	}
+	return expected
+}
+
+// actualState describes what the passive's inspected mutable state currently has.
+func actualState(ms *persistencespb.WorkflowMutableState) map[string]any {
+	info := ms.GetExecutionInfo()
+	actual := map[string]any{"next_event_id": ms.GetNextEventId()}
+	if th := info.GetTransitionHistory(); len(th) > 0 {
+		actual["versioned_transition"] = versionedTransitionEntry(th[len(th)-1])
+	}
+	if currentHistory, err := versionhistory.GetCurrentVersionHistory(info.GetVersionHistories()); err == nil {
+		actual["event_version_history"] = versionHistoryEntries(currentHistory.GetItems())
+	}
+	return actual
+}
+
+func versionedTransitionEntry(vt *persistencespb.VersionedTransition) wideevents.VersionedTransitionEntry {
+	return wideevents.VersionedTransitionEntry{
+		FailoverVersion: vt.GetNamespaceFailoverVersion(),
+		TransitionCount: vt.GetTransitionCount(),
+	}
+}
+
+func versionHistoryEntries(items []*historyspb.VersionHistoryItem) []wideevents.VersionHistoryEntry {
+	out := make([]wideevents.VersionHistoryEntry, 0, len(items))
+	for _, item := range items {
+		out = append(out, wideevents.VersionHistoryEntry{EventID: item.GetEventId(), Version: item.GetVersion()})
+	}
+	return out
+}

service/history/replication/stream_sender.go

@@ -594,6 +594,9 @@ Loop:
 				}
 				metrics.ReplicationRateLimitLatency.With(s.metrics).Record(time.Since(rlStartTime), metrics.OperationTag(TaskOperationTag(task)))
 			}
+			if s.config.EmitReplicationLifecycleEvents() {
+				s.emitReplicationSent(task, item)
+			}
 			if err := s.sendToStream(&historyservice.StreamWorkflowReplicationMessagesResponse{
 				Attributes: &historyservice.StreamWorkflowReplicationMessagesResponse_Messages{
 					Messages: &replicationspb.WorkflowReplicationMessages{

service/history/shard/context_factory.go

@@ -1,6 +1,7 @@
 package shard
 
 import (
+	otellog "go.opentelemetry.io/otel/log"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/archiver"
@@ -47,6 +48,7 @@ type (
 		HostInfoProvider            membership.HostInfoProvider
 		Logger                      log.Logger
 		MetricsHandler              metrics.Handler
+		EventLogger                 otellog.Logger
 		NamespaceRegistry           namespace.Registry
 		PayloadSerializer           serialization.Serializer
 		PersistenceExecutionManager persistence.ExecutionManager
@@ -92,6 +94,7 @@ func (c *contextFactoryImpl) CreateContext(
 		c.ClientBean,
 		c.HistoryClient,
 		c.MetricsHandler,
+		c.EventLogger,
 		c.PayloadSerializer,
 		c.TimeSource,
 		c.NamespaceRegistry,

service/history/shard/context_impl.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	otellog "go.opentelemetry.io/otel/log"
 	commonpb "go.temporal.io/api/common/v1"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/server/api/adminservice/v1"
@@ -48,6 +49,7 @@ import (
 	"go.temporal.io/server/common/rpc"
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/util"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/consts"
 	"go.temporal.io/server/service/history/events"
@@ -87,6 +89,7 @@ type (
 		stringRepr          string
 		executionManager    persistence.ExecutionManager
 		metricsHandler      metrics.Handler
+		eventLogger         otellog.Logger
 		eventsCache         events.Cache
 		closeCallback       CloseCallback
 		config              *configs.Config
@@ -2050,6 +2053,7 @@ func newContext(
 	clientBean client.Bean,
 	historyClient historyservice.HistoryServiceClient,
 	metricsHandler metrics.Handler,
+	eventLogger otellog.Logger,
 	payloadSerializer serialization.Serializer,
 	timeSource cclock.TimeSource,
 	namespaceRegistry namespace.Registry,
@@ -2087,6 +2091,7 @@ func newContext(
 		stringRepr:              fmt.Sprintf("Shard(%d)", shardID),
 		executionManager:        persistenceExecutionManager,
 		metricsHandler:          metricsHandler,
+		eventLogger:             eventLogger,
 		closeCallback:           closeCallback,
 		config:                  historyConfig,
 		finalizer:               finalizer.New(taggedLogger, metricsHandler),
@@ -2195,6 +2200,13 @@ func (s *ContextImpl) GetHistoryClient() historyservice.HistoryServiceClient {
 	return s.historyClient
 }
 
+func (s *ContextImpl) GetEventLogger() otellog.Logger {
+	if s.eventLogger == nil {
+		return wideevents.NoopLogger()
+	}
+	return s.eventLogger
+}
+
 func (s *ContextImpl) GetMetricsHandler() metrics.Handler {
 	return s.metricsHandler
 }

temporal/fx.go

@@ -12,6 +12,8 @@ import (
 	"github.com/google/uuid"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	otellog "go.opentelemetry.io/otel/log"
+	lognoop "go.opentelemetry.io/otel/log/noop"
 	"go.opentelemetry.io/otel/propagation"
 	otelresource "go.opentelemetry.io/otel/sdk/resource"
 	otelsdktrace "go.opentelemetry.io/otel/sdk/trace"
@@ -50,6 +52,7 @@ import (
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/searchattribute/sadefs"
 	"go.temporal.io/server/common/telemetry"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/frontend"
 	"go.temporal.io/server/service/history"
 	"go.temporal.io/server/service/history/replication"
@@ -125,6 +128,7 @@ type (
 		TLSConfigProvider     encryption.TLSConfigProvider
 		EsClient              esclient.Client
 		MetricsHandler        metrics.Handler
+		EventLoggerProvider   otellog.LoggerProvider
 	}
 )
 
@@ -204,6 +208,14 @@ func ServerOptionsProvider(opts []ServerOption) (serverOptionsProvider, error) {
 		}
 	}
 
+	// EventLoggerProvider backs structured ("wide") events. Select the custom OTEL LoggerProvider
+	// if injected, else a no-op provider that discards events. A deployment opts in by injecting a
+	// provider via WithCustomEventLoggerProvider.
+	eventLoggerProvider := so.eventLoggerProvider
+	if eventLoggerProvider == nil {
+		eventLoggerProvider = lognoop.NewLoggerProvider()
+	}
+
 	// DynamicConfigClient
 	dcClient := so.dynamicConfigClient
 	if dcClient == nil {
@@ -314,6 +326,7 @@ func ServerOptionsProvider(opts []ServerOption) (serverOptionsProvider, error) {
 		TLSConfigProvider:     tlsConfigProvider,
 		EsClient:              esClient,
 		MetricsHandler:        metricHandler,
+		EventLoggerProvider:   eventLoggerProvider,
 	}, nil
 }
 
@@ -359,6 +372,7 @@ type (
 		NamespaceLogger                 resource.NamespaceLogger
 		DynamicConfigClient             dynamicconfig.Client
 		MetricsHandler                  metrics.Handler
+		EventLoggerProvider             otellog.LoggerProvider
 		EsClient                        esclient.Client
 		TlsConfigProvider               encryption.TLSConfigProvider //nolint:staticcheck // should be TLSConfigProvider
 		PersistenceConfig               config.Persistence
@@ -452,6 +466,9 @@ func (params ServiceProviderParamsCommon) GetCommonServiceOptions(serviceName pr
 			func() metrics.Handler {
 				return params.MetricsHandler.WithTags(metrics.ServiceNameTag(serviceName))
 			},
+			func() otellog.Logger {
+				return wideevents.NewLogger(params.EventLoggerProvider, string(serviceName))
+			},
 			func() esclient.Client {
 				return params.EsClient
 			},

temporal/server_option.go

@@ -3,6 +3,7 @@ package temporal
 import (
 	"net/http"
 
+	otellog "go.opentelemetry.io/otel/log"
 	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/archiver/provider"
 	"go.temporal.io/server/common/authorization"
@@ -213,3 +214,14 @@ func WithCustomMetricsHandler(provider metrics.Handler) ServerOption {
 		s.metricHandler = provider
 	})
 }
+
+// WithCustomEventLoggerProvider sets a custom OTEL LoggerProvider used to emit structured
+// ("wide") events. Each service builds an events.Handler from it (see events.NewHandler). When
+// unset, events are discarded via a no-op provider.
+//
+// NOTE: this option is experimental and may be changed or removed in future release.
+func WithCustomEventLoggerProvider(loggerProvider otellog.LoggerProvider) ServerOption {
+	return applyFunc(func(s *serverOptions) {
+		s.eventLoggerProvider = loggerProvider
+	})
+}

temporal/server_options.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"slices"
 
+	otellog "go.opentelemetry.io/otel/log"
 	"go.temporal.io/server/client"
 	"go.temporal.io/server/common/archiver/provider"
 	"go.temporal.io/server/common/authorization"
@@ -59,6 +60,7 @@ type (
 		searchAttributesMapper          searchattribute.Mapper
 		customFrontendInterceptors      []grpc.UnaryServerInterceptor
 		metricHandler                   metrics.Handler
+		eventLoggerProvider             otellog.LoggerProvider
 		tokenProvider                   auth.TokenProvider
 	}
 )

tests/testcore/onebox.go

@@ -15,6 +15,7 @@ import (
 	"testing"
 	"time"
 
+	otellog "go.opentelemetry.io/otel/log"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.temporal.io/server/api/adminservice/v1"
 	"go.temporal.io/server/chasm"
@@ -47,6 +48,7 @@ import (
 	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/telemetry"
 	"go.temporal.io/server/common/testing/testhooks"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/components/nexusoperations"
 	"go.temporal.io/server/service/frontend"
 	"go.temporal.io/server/service/history"
@@ -448,6 +450,7 @@ func (c *temporalImpl) startHistory() {
 			),
 			fx.Provide(c.configProvider),
 			fx.Provide(c.GetMetricsHandler),
+			fx.Provide(func() otellog.Logger { return wideevents.NoopLogger() }),
 			fx.Provide(func() listenHostPort { return listenHostPort(host) }),
 			fx.Provide(func() httpPort { return mustPortFromAddress(c.FrontendHTTPAddress()) }),
 			fx.Provide(func() config.DCRedirectionPolicy { return config.DCRedirectionPolicy{} }),