Add wide events for parent-child workflow lifecycle debugging (#11707)

## What changed? - Added `parent_child_lifecycle` wide events at critical parent-child failure and recovery points. - Recorded parent/child execution identities, initiated event ID/version, local task metadata, errors, and locally observed workflow state such as `Zombie`. - Added visibility into standby verification and parent resend scheduling, execution, deduplication, and completion. - Reused the existing `history.emitReplicationLifecycleEvents` flag. - Added and extended unit tests for event payloads and emission paths. ## Why? Parent-child failures across shards and clusters are difficult to diagnose from existing metrics alone. These events help distinguish missing workflows, missing first workflow tasks, completion verification failures, workflow ID conflicts, orphaned children, and parent resend progress. They can also be correlated with replication lifecycle and task-discard events. ## How did you test it? - [x] built - [ ] run locally and tested manually - [x] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) ### output from local e2e test <details> <summary>Wide Events for Parent resend scenario </summary> ``` [ { "attempt": 1, "details": { "child_workflow_state": "Completed", "event_type": "parent_child_lifecycle", "local_task_id": 1048587, "message": "Standby child completion verification started with parent resend requested", "operation": "standby_verification", "outcome": "started", "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "phase": "verify_child_completion", "version": 1 }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "parent_initiated_id": 5, "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "phase": "executing", "run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "shard": 2, "task_type": "TransferCloseExecution", "workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827" }, { "attempt": 0, "details": { "child_run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "child_workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827", "event_type": "parent_child_lifecycle", "initial_error_type": "serviceerror.NotFound", "local_cluster": "standby_annmr", "message": "Parent workflow resend checkpoint", "operation": "standby_verification_sync_state", "outcome": "started", "parent_initiated_id": 5, "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "parent_workflow_state": "", "phase": "parent_resend", "stage": "sync_workflow_state" }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "phase": "executing", "run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "shard": 1, "source_cluster": "active_annmr", "task_type": "sync_workflow_state", "workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee" }, { "details": { "child_run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "child_workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827", "event_type": "parent_child_lifecycle", "initial_error_type": "serviceerror.NotFound", "local_cluster": "standby_annmr", "message": "Parent workflow resend checkpoint", "operation": "standby_verification_sync_state", "outcome": "succeeded", "parent_initiated_id": 5, "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "parent_workflow_state": "Running", "phase": "parent_resend" }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "outcome": "verified", "phase": "applied", "run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "shard": 1, "source_cluster": "active_annmr", "task_type": "sync_workflow_state", "workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee" }, { "details": { "attempt": 1, "child_workflow_state": "Completed", "event_type": "parent_child_lifecycle", "local_task_id": 1048587, "message": "Standby child completion verification completed", "operation": "standby_verification", "outcome": "verified", "parent_initiated_version": 1, "parent_namespace_id": "691a916b-7f7a-4284-93af-782495421958", "phase": "verify_child_completion", "version": 1 }, "event_name": "replication_lifecycle", "namespace": "test-namespace-974d10a5-e0ef-451a-bac1-0834ec45f658", "namespace_id": "691a916b-7f7a-4284-93af-782495421958", "outcome": "verified", "parent_initiated_id": 5, "parent_run_id": "01a034ee-4749-795e-875c-07c66ddaf4a7", "parent_workflow_id": "parent-d30f61cf-70b9-4cfd-94af-85bc0af99eee", "phase": "applied", "run_id": "01a034ee-475b-7e96-9e3f-4239bd23f79a", "shard": 2, "task_type": "TransferCloseExecution", "workflow_id": "child-211e426f-e194-4492-8534-32d30fefb827" } ] ``` </details> ## Potential risks Enabling replication lifecycle events will produce additional wide-event volume.
2026-08-30 18:41:49 -07:00 · 2026-08-25 16:00:13 -07:00
parent a02e33e571
commit 534045279e
9 changed files with 748 additions and 34 deletions
--- a/common/dynamicconfig/constants.go
+++ b/common/dynamicconfig/constants.go
@@ -1739,7 +1739,7 @@ branch token is not the execution's current one, but still serves the read.`,
 	EmitReplicationLifecycleEvents = NewGlobalBoolSetting(
 		"history.emitReplicationLifecycleEvents",
 		false,
-		`EmitReplicationLifecycleEvents controls whether the history service emits ReplicationLifecycle wide events (sent/executing/applied/skipped/error phases). Cluster-level; default off.`,
+		`EmitReplicationLifecycleEvents controls whether the history service emits ReplicationLifecycle wide events, including parent-child diagnostics. Cluster-level; default off.`,
 	)
 	EnableCloseInboundReplicationStreamOnShutdown = NewGlobalBoolSetting(
 		"history.enableCloseInboundReplicationStreamOnShutdown",
--- a/common/wideevents/replication_events.go
+++ b/common/wideevents/replication_events.go
@@ -104,6 +104,27 @@ const (
 	ReplOperationDLQWrite                         = "dlq_write"
 )

+// Parent-child diagnostics reuse the ReplicationLifecycle event and store this vocabulary in
+// details. Keeping the values here gives producers a shared contract without adding an event table.
+const (
+	ParentChildLifecycleEventType = "parent_child_lifecycle"
+
+	ParentChildPhaseVerifyChildCompletion = "verify_child_completion"
+	ParentChildPhaseParentResend          = "parent_resend"
+
+	ParentChildOutcomeNotFound          = "not_found"
+	ParentChildOutcomeCompletionMissing = "completion_missing"
+	ParentChildOutcomeScheduled         = "scheduled"
+	ParentChildOutcomeStarted           = "started"
+	ParentChildOutcomeVerified          = "verified"
+	ParentChildOutcomeIgnored           = "ignored"
+	ParentChildOutcomeSucceeded         = "succeeded"
+	ParentChildOutcomeSourceNotFound    = "source_not_found"
+	ParentChildOutcomeDeduplicated      = "deduplicated"
+	ParentChildOutcomeLimited           = "limited"
+	ParentChildOutcomeFailed            = "failed"
+)
+
 // EmitReplicationError normalizes the common error envelope and emits it into the existing
 // replication_lifecycle event, not a separate event table.
 func EmitReplicationError(
--- a/service/history/api/verifychildworkflowcompletionrecorded/api.go
+++ b/service/history/api/verifychildworkflowcompletionrecorded/api.go
@@ -3,6 +3,7 @@ package verifychildworkflowcompletionrecorded
 import (
 	"context"
 	"errors"
+	"maps"
 	"time"

 	commonpb "go.temporal.io/api/common/v1"
@@ -23,6 +24,8 @@ import (
 	"go.temporal.io/server/common/persistence/transitionhistory"
 	"go.temporal.io/server/common/persistence/versionhistory"
 	"go.temporal.io/server/common/rpc"
+	"go.temporal.io/server/common/util"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/history/api"
 	"go.temporal.io/server/service/history/consts"
 	historyi "go.temporal.io/server/service/history/interfaces"
@@ -33,7 +36,10 @@ func verifyChildExecution(
 	workflowConsistencyChecker api.WorkflowConsistencyChecker,
 	request *historyservice.VerifyChildExecutionCompletionRecordedRequest,
 ) (versionedTransition *persistencespb.VersionedTransition,
-	versionHistories *historyspb.VersionHistories, retError error) {
+	versionHistories *historyspb.VersionHistories,
+	parentWorkflowState string,
+	retError error,
+) {
 	workflowLease, err := workflowConsistencyChecker.GetWorkflowLease(
 		ctx,
 		request.Clock,
@@ -49,28 +55,29 @@ func verifyChildExecution(
 		locks.PriorityLow,
 	)
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, "", err
 	}
 	defer func() { workflowLease.GetReleaseFn()(retError) }()

 	mutableState := workflowLease.GetMutableState()
+	parentWorkflowState = mutableState.GetExecutionState().GetState().String()
 	if !mutableState.IsWorkflowExecutionRunning() &&
 		mutableState.GetExecutionState().State != enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE {
 		// parent has already completed and can't be blocked after failover.
-		return nil, nil, nil
+		return nil, nil, parentWorkflowState, nil
 	}

 	onCurrentBranch, err := api.IsHistoryEventOnCurrentBranch(mutableState, request.ParentInitiatedId, request.ParentInitiatedVersion)
 	if err != nil {
 		// initiated event not found on any branch
-		return nil, nil, consts.ErrWorkflowNotReady
+		return nil, nil, parentWorkflowState, consts.ErrWorkflowNotReady
 	}

 	if !onCurrentBranch {
 		// due to conflict resolution, the initiated event may on a different branch of the workflow.
 		// we don't have to do anything and can simply return not found error. Standby logic
 		// after seeing this error will give up verification.
-		return nil, nil, consts.ErrChildExecutionNotFound
+		return nil, nil, parentWorkflowState, consts.ErrChildExecutionNotFound
 	}

 	ci, isRunning := mutableState.GetChildExecutionInfo(request.ParentInitiatedId)
@@ -78,15 +85,15 @@ func verifyChildExecution(
 		if ci.StartedEventId != common.EmptyEventID &&
 			ci.GetStartedWorkflowId() != request.ChildExecution.GetWorkflowId() {
 			// this can happen since we may not have the initiated version
-			return nil, nil, consts.ErrChildExecutionNotFound
+			return nil, nil, parentWorkflowState, consts.ErrChildExecutionNotFound
 		}

-		return nil, nil, consts.ErrWorkflowNotReady
+		return nil, nil, parentWorkflowState, consts.ErrWorkflowNotReady
 	}

 	versionedTransition = transitionhistory.CopyVersionedTransition(transitionhistory.LastVersionedTransition(mutableState.GetExecutionInfo().TransitionHistory))
 	versionHistories = versionhistory.CopyVersionHistories(mutableState.GetExecutionInfo().VersionHistories)
-	return versionedTransition, versionHistories, nil
+	return versionedTransition, versionHistories, parentWorkflowState, nil
 }

 func Invoke(
@@ -102,7 +109,7 @@ func Invoke(
 	}

 	resendParent := false
-	versionedTransition, versionHistories, errVerify := verifyChildExecution(ctx, workflowConsistencyChecker, request)
+	versionedTransition, versionHistories, parentWorkflowState, errVerify := verifyChildExecution(ctx, workflowConsistencyChecker, request)
 	switch errVerify.(type) {
 	case nil:
 		return &historyservice.VerifyChildExecutionCompletionRecordedResponse{}, nil
@@ -114,12 +121,14 @@ func Invoke(
 	}

 	metricsHandler := shardContext.GetMetricsHandler()
+	emitLifecycle := shardContext.GetConfig().EmitReplicationLifecycleEvents()
+	asyncResend := shardContext.GetConfig().EnableAsyncParentWorkflowResend()

 	// The measured resend, run either inline or in the background.
 	resend := func(ctx context.Context) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) {
 		metrics.ParentWorkflowResendAttempts.With(metricsHandler).Record(1)
 		startTime := time.Now().UTC()
-		resp, err := resendParentAndVerify(ctx, request, workflowConsistencyChecker, shardContext, namespaceID, versionedTransition, versionHistories, errVerify)
+		resp, err := resendParentAndVerify(ctx, request, workflowConsistencyChecker, shardContext, namespaceID, versionedTransition, versionHistories, errVerify, parentWorkflowState, emitLifecycle)
 		metrics.ParentWorkflowResendLatency.With(metricsHandler).Record(time.Since(startTime))
 		if err != nil {
 			recordResendFailure(shardContext, metricsHandler, request, err)
@@ -127,7 +136,7 @@ func Invoke(
 		return resp, err
 	}

-	if !shardContext.GetConfig().EnableAsyncParentWorkflowResend() {
+	if !asyncResend {
 		return resend(ctx)
 	}

@@ -138,9 +147,15 @@ func Invoke(
 	// retry while an earlier resend runs, so without these a stale parent, or a namespace with many
 	// of them, would spawn goroutines without bound.
 	parentKey := definition.NewWorkflowKey(request.NamespaceId, request.ParentExecution.WorkflowId, request.ParentExecution.RunId)
-	claimed, atCapacity := inFlightResends.tryClaim(parentKey, shardContext.GetConfig().ParentWorkflowResendMaxInFlight())
+	maxInFlight := shardContext.GetConfig().ParentWorkflowResendMaxInFlight()
+	claimed, atCapacity := inFlightResends.tryClaim(parentKey, maxInFlight)
 	if atCapacity {
 		metrics.ParentWorkflowResendLimited.With(metricsHandler).Record(1)
+		if emitLifecycle {
+			details := parentResendEventDetails(errVerify)
+			details["max_in_flight"] = maxInFlight
+			emitParentResendLifecycleEvent(shardContext, request, parentWorkflowState, wideevents.ParentChildOutcomeLimited, nil, details)
+		}
 		shardContext.GetLogger().Warn("Dropped parent workflow resend, shard is at its in-flight limit",
 			tag.WorkflowNamespaceID(request.GetNamespaceId()),
 			tag.NewStringTag("parent-workflow-id", request.ParentExecution.GetWorkflowId()),
@@ -153,8 +168,28 @@ func Invoke(
 	}
 	if !claimed {
 		metrics.ParentWorkflowResendSkipped.With(metricsHandler).Record(1)
+		if emitLifecycle {
+			emitParentResendLifecycleEvent(
+				shardContext,
+				request,
+				parentWorkflowState,
+				wideevents.ParentChildOutcomeDeduplicated,
+				nil,
+				parentResendEventDetails(errVerify),
+			)
+		}
 		return nil, errVerify
 	}
+	if emitLifecycle {
+		emitParentResendLifecycleEvent(
+			shardContext,
+			request,
+			parentWorkflowState,
+			wideevents.ParentChildOutcomeScheduled,
+			nil,
+			parentResendEventDetails(errVerify),
+		)
+	}

 	// The context is detached from the request, which gRPC cancels when this handler returns, and
 	// rooted at the shard lifecycle so the work stops with the shard.
@@ -205,26 +240,47 @@ func resendParentAndVerify(
 	versionedTransition *persistencespb.VersionedTransition,
 	versionHistories *historyspb.VersionHistories,
 	errVerify error,
+	parentWorkflowState string,
+	emitLifecycle bool,
 ) (*historyservice.VerifyChildExecutionCompletionRecordedResponse, error) {
 	// Resend parent workflow from source cluster

 	clusterMetadata := shardContext.GetClusterMetadata()
 	targetClusterInfo := clusterMetadata.GetAllClusterInfo()[clusterMetadata.GetCurrentClusterName()]
+	activeClusterName := ""
+	emitResult := func(outcome string, eventErr error, stage string) {
+		if !emitLifecycle {
+			return
+		}
+		details := parentResendEventDetails(errVerify)
+		if activeClusterName != "" {
+			details["source_cluster"] = activeClusterName
+		}
+		if stage != "" {
+			details["stage"] = stage
+		}
+		emitParentResendLifecycleEvent(shardContext, request, parentWorkflowState, outcome, eventErr, details)
+	}

 	namespaceEntry, err := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespaceID)
 	if err != nil {
+		emitResult(wideevents.ParentChildOutcomeFailed, err, "resolve_namespace")
 		return nil, err
 	}

-	activeClusterName := namespaceEntry.ActiveClusterName(namespace.RoutingKey{ID: request.ParentExecution.WorkflowId})
+	activeClusterName = namespaceEntry.ActiveClusterName(namespace.RoutingKey{ID: request.ParentExecution.WorkflowId})
 	if activeClusterName == clusterMetadata.GetCurrentClusterName() {
-		return nil, errors.New("namespace becomes active when processing task as standby")
+		err = errors.New("namespace becomes active when processing task as standby")
+		emitResult(wideevents.ParentChildOutcomeFailed, err, "resolve_source_cluster")
+		return nil, err
 	}

 	remoteAdminClient, err := shardContext.GetRemoteAdminClient(activeClusterName)
 	if err != nil {
+		emitResult(wideevents.ParentChildOutcomeFailed, err, "resolve_remote_client")
 		return nil, err
 	}
+	emitResult(wideevents.ParentChildOutcomeStarted, nil, "sync_workflow_state")

 	resp, err := remoteAdminClient.SyncWorkflowState(ctx, &adminservice.SyncWorkflowStateRequest{
 		NamespaceId: request.NamespaceId,
@@ -243,30 +299,119 @@ func resendParentAndVerify(
 			// parent workflow is not found on source cluster,
 			// we can return empty response to indicate that verification is done
 			// TODO: add parent workflow to workflowNotFoundCache
+			emitResult(wideevents.ParentChildOutcomeSourceNotFound, err, "sync_workflow_state")
 			return &historyservice.VerifyChildExecutionCompletionRecordedResponse{}, nil
 		}
 		if _, ok := errors.AsType[*serviceerror.FailedPrecondition](err); ok {
 			// Unable to perform sync state. Transition history maybe disabled in source cluster.
+			emitResult(wideevents.ParentChildOutcomeFailed, err, "sync_workflow_state")
 			return nil, errVerify
 		}
+		emitResult(wideevents.ParentChildOutcomeFailed, err, "sync_workflow_state")
 		return nil, err
 	}

 	engine, err := shardContext.GetEngine(ctx)
 	if err != nil {
+		emitResult(wideevents.ParentChildOutcomeFailed, err, "get_engine")
 		return nil, err
 	}
 	err = engine.ReplicateVersionedTransition(ctx, chasm.WorkflowArchetypeID, resp.VersionedTransitionArtifact, activeClusterName)
 	if err != nil {
 		if !errors.Is(err, consts.ErrDuplicate) {
+			emitResult(wideevents.ParentChildOutcomeFailed, err, "replicate_versioned_transition")
 			return nil, err
 		}
 	}

 	// Verify child execution again after resending parent workflow
-	_, _, err = verifyChildExecution(ctx, workflowConsistencyChecker, request)
+	_, _, observedParentWorkflowState, err := verifyChildExecution(ctx, workflowConsistencyChecker, request)
+	if observedParentWorkflowState != "" {
+		parentWorkflowState = observedParentWorkflowState
+	}
 	if err != nil {
+		emitResult(wideevents.ParentChildOutcomeFailed, err, "verify_after_resend")
 		return nil, err
 	}
+	emitResult(wideevents.ParentChildOutcomeSucceeded, nil, "")
 	return &historyservice.VerifyChildExecutionCompletionRecordedResponse{}, nil
 }
+
+func parentResendEventDetails(initialError error) map[string]any {
+	return map[string]any{
+		"initial_error_type": util.ErrorType(initialError),
+	}
+}
+
+func emitParentResendLifecycleEvent(
+	shardContext historyi.ShardContext,
+	request *historyservice.VerifyChildExecutionCompletionRecordedRequest,
+	parentWorkflowState string,
+	outcome string,
+	err error,
+	details map[string]any,
+) {
+	eventDetails := make(map[string]any, len(details)+11)
+	maps.Copy(eventDetails, details)
+	eventDetails["event_type"] = wideevents.ParentChildLifecycleEventType
+	eventDetails["phase"] = wideevents.ParentChildPhaseParentResend
+	eventDetails["outcome"] = outcome
+	eventDetails["local_cluster"] = shardContext.GetClusterMetadata().GetCurrentClusterName()
+	eventDetails["parent_namespace_id"] = request.GetNamespaceId()
+	eventDetails["parent_workflow_id"] = request.GetParentExecution().GetWorkflowId()
+	eventDetails["parent_run_id"] = request.GetParentExecution().GetRunId()
+	eventDetails["parent_workflow_state"] = parentWorkflowState
+	eventDetails["child_workflow_id"] = request.GetChildExecution().GetWorkflowId()
+	eventDetails["child_run_id"] = request.GetChildExecution().GetRunId()
+	eventDetails["parent_initiated_id"] = request.GetParentInitiatedId()
+	eventDetails["parent_initiated_version"] = request.GetParentInitiatedVersion()
+
+	namespaceName := ""
+	if entry, namespaceErr := shardContext.GetNamespaceRegistry().GetNamespaceByID(namespace.ID(request.GetNamespaceId())); namespaceErr == nil {
+		namespaceName = entry.Name().String()
+	}
+	sourceCluster, ok := eventDetails["source_cluster"].(string)
+	if !ok {
+		sourceCluster = ""
+	}
+	delete(eventDetails, "source_cluster")
+	if err != nil {
+		eventDetails["error"] = err.Error()
+		eventDetails["error_type"] = util.ErrorType(err)
+	}
+	payload := wideevents.ReplicationLifecyclePayload{
+		TaskType:      wideevents.ReplTaskSyncWorkflowState,
+		Shard:         shardContext.GetShardID(),
+		Namespace:     namespaceName,
+		NamespaceID:   request.GetNamespaceId(),
+		WorkflowID:    request.GetParentExecution().GetWorkflowId(),
+		RunID:         request.GetParentExecution().GetRunId(),
+		SourceCluster: sourceCluster,
+	}
+	switch outcome {
+	case wideevents.ParentChildOutcomeScheduled,
+		wideevents.ParentChildOutcomeStarted,
+		wideevents.ParentChildOutcomeDeduplicated:
+		eventDetails["operation"] = wideevents.ReplOperationStandbyVerificationSyncState
+		eventDetails["message"] = "Parent workflow resend checkpoint"
+		payload.Phase = wideevents.ReplicationExecuting
+		payload.Details = eventDetails
+		wideevents.Emit(shardContext.GetEventLogger(), payload)
+	case wideevents.ParentChildOutcomeSucceeded, wideevents.ParentChildOutcomeSourceNotFound:
+		eventDetails["operation"] = wideevents.ReplOperationStandbyVerificationSyncState
+		eventDetails["message"] = "Parent workflow resend checkpoint"
+		payload.Phase = wideevents.ReplicationApplied
+		payload.Outcome = wideevents.ParentChildOutcomeVerified
+		payload.Details = eventDetails
+		wideevents.Emit(shardContext.GetEventLogger(), payload)
+	default:
+		wideevents.EmitReplicationError(
+			shardContext.GetEventLogger(),
+			payload,
+			wideevents.ReplOperationStandbyVerificationSyncState,
+			"Parent workflow resend checkpoint",
+			err,
+			eventDetails,
+		)
+	}
+}
--- a/service/history/history_engine2_test.go
+++ b/service/history/history_engine2_test.go
@@ -48,8 +48,11 @@ import (
 	"go.temporal.io/server/common/searchattribute/sadefs"
 	serviceerrors "go.temporal.io/server/common/serviceerror"
 	"go.temporal.io/server/common/tasktoken"
+	"go.temporal.io/server/common/testing/await"
 	"go.temporal.io/server/common/testing/protorequire"
 	"go.temporal.io/server/common/testing/testvars"
+	"go.temporal.io/server/common/util"
+	"go.temporal.io/server/common/wideevents"
 	"go.temporal.io/server/service/history/api"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/consts"
@@ -89,6 +92,7 @@ type (
 		mockClusterMetadata         *cluster.MockMetadata
 		mockVisibilityManager       *manager.MockVisibilityManager
 		mockWorkflowStateReplicator *ndc.MockWorkflowStateReplicator
+		parentChildEventCapture     *parentChildEventCapture

 		workflowCache    wcache.Cache
 		historyEngine    *historyEngineImpl
@@ -142,6 +146,7 @@ func (s *engine2Suite) SetupTest() {
 	s.mockMemoryScheduledQueue.EXPECT().NotifyNewTasks(gomock.Any()).AnyTimes()

 	s.config = tests.NewDynamicConfig()
+	s.parentChildEventCapture = &parentChildEventCapture{}
 	mockShard := shard.NewTestContext(
 		s.controller,
 		&persistencespb.ShardInfo{
@@ -149,6 +154,7 @@ func (s *engine2Suite) SetupTest() {
 			RangeId: 1,
 		},
 		s.config,
+		s.parentChildEventCapture,
 	)
 	reg := hsm.NewRegistry()
 	err := workflow.RegisterStateMachine(reg)
@@ -2684,6 +2690,8 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_WorkflowNotExi
 func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent() {
 	// Inline resend: the RPC pulls and re-verifies before returning.
 	s.config.EnableAsyncParentWorkflowResend = func() bool { return false }
+	s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true)
+	capture := s.parentChildEventCapture

 	request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{
 		NamespaceId: tests.ParentNamespaceID.String(),
@@ -2705,7 +2713,7 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent()
 	mockClusterMetadata := cluster.NewMockMetadata(s.controller)
 	mockClusterMetadata.EXPECT().GetClusterID().Return(tests.Version).AnyTimes()
 	mockClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestAlternativeClusterName).AnyTimes()
-	mockClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo)
+	mockClusterMetadata.EXPECT().GetAllClusterInfo().Return(cluster.TestAllClusterInfo).AnyTimes()
 	mockClusterMetadata.EXPECT().ClusterNameForFailoverVersion(true, tests.Version).Return(cluster.TestCurrentClusterName).AnyTimes()
 	s.mockShard.SetClusterMetadata(mockClusterMetadata)

@@ -2740,17 +2748,13 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent()
 	).Return(resp, nil)
 	s.mockWorkflowStateReplicator.EXPECT().ReplicateVersionedTransition(gomock.Any(), chasm.WorkflowArchetypeID, resp.VersionedTransitionArtifact, cluster.TestCurrentClusterName).Return(nil)

-	// prepare closed workflow
+	// The parent can land as a zombie after conflict resolution; preserve that state in the event.
 	ms := workflow.TestGlobalMutableState(s.historyEngine.shardContext, s.mockEventsCache, log.NewTestLogger(), tests.Version, tests.WorkflowID, tests.RunID)
 	addWorkflowExecutionStartedEvent(ms, &commonpb.WorkflowExecution{
 		WorkflowId: tests.WorkflowID,
 		RunId:      tests.RunID,
 	}, "wType", "testTaskQueue", payloads.EncodeString("input"), 25*time.Second, 20*time.Second, 200*time.Second, "identity")
-	_, err := ms.AddTimeoutWorkflowEvent(
-		enumspb.RETRY_STATE_RETRY_POLICY_NOT_SET,
-		uuid.NewString(),
-	)
-	s.NoError(err)
+	ms.GetExecutionState().State = enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE
 	ms.GetExecutionInfo().VersionHistories = &historyspb.VersionHistories{
 		CurrentVersionHistoryIndex: 0,
 		Histories: []*historyspb.VersionHistory{
@@ -2767,14 +2771,79 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParent()
 	gwmsResponse := &persistence.GetWorkflowExecutionResponse{State: wfMs}
 	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(gwmsResponse, nil)

-	_, err = s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), request)
-	s.NoError(err)
+	_, err := s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), request)
+	s.Require().NoError(err)
+
+	s.Require().Equal([]string{
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeSucceeded),
+	}, parentChildOutcomes(capture))
+	records := parentChildRecords(capture)
+	s.Require().Len(records, 2)
+	attributes := wideEventAttributes(records[0])
+	details := wideEventDetails(records[0])
+	s.Equal(string(wideevents.ReplicationExecuting), attributes["phase"].AsString())
+	s.Equal(wideevents.ParentChildPhaseParentResend, details["phase"])
+	s.Equal(cluster.TestAlternativeClusterName, details["local_cluster"])
+	s.Equal(int64(1), attributes["shard"].AsInt64())
+	s.Equal(request.GetParentExecution().GetWorkflowId(), attributes["workflow_id"].AsString())
+	s.Equal(request.GetParentExecution().GetWorkflowId(), details["parent_workflow_id"])
+	s.Equal(request.GetChildExecution().GetWorkflowId(), details["child_workflow_id"])
+	s.InDelta(float64(request.GetParentInitiatedId()), details["parent_initiated_id"], 0)
+	s.InDelta(float64(request.GetParentInitiatedVersion()), details["parent_initiated_version"], 0)
+	s.Equal(cluster.TestCurrentClusterName, attributes["source_cluster"].AsString())
+	s.Equal(util.ErrorType(&serviceerror.NotFound{}), details["initial_error_type"])
+	s.Equal("sync_workflow_state", details["stage"])
+	s.Equal(string(wideevents.ReplicationApplied), wideEventAttributes(records[1])["phase"].AsString())
+	s.Equal(wideevents.ParentChildOutcomeVerified, wideEventAttributes(records[1])["outcome"].AsString())
+	s.Equal(enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE.String(), wideEventDetails(records[1])["parent_workflow_state"])
+
+	// Source NotFound is a terminal, successful outcome: there is no parent state left to pull.
+	sourceMissingRequest := &historyservice.VerifyChildExecutionCompletionRecordedRequest{
+		NamespaceId: request.GetNamespaceId(),
+		ParentExecution: &commonpb.WorkflowExecution{
+			WorkflowId: "parent missing on source",
+			RunId:      uuid.NewString(),
+		},
+		ChildExecution:         request.GetChildExecution(),
+		ParentInitiatedId:      request.GetParentInitiatedId(),
+		ParentInitiatedVersion: request.GetParentInitiatedVersion(),
+		ResendParent:           true,
+	}
+	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, &serviceerror.NotFound{})
+	s.mockShard.Resource.RemoteAdminClient.EXPECT().SyncWorkflowState(
+		gomock.Any(),
+		&adminservice.SyncWorkflowStateRequest{
+			NamespaceId: sourceMissingRequest.NamespaceId,
+			Execution:   sourceMissingRequest.ParentExecution,
+			ArchetypeId: chasm.WorkflowArchetypeID,
+			TargetClusterId: int32(
+				cluster.TestAlternativeClusterInitialFailoverVersion,
+			),
+		},
+	).Return(nil, serviceerror.NewNotFound("parent missing on source"))
+
+	_, err = s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), sourceMissingRequest)
+	s.Require().NoError(err)
+	s.Require().Equal([]string{
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeSucceeded),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeSourceNotFound),
+	}, parentChildOutcomes(capture))
+	sourceNotFoundRecord := parentChildRecords(capture)[3]
+	s.Equal(string(wideevents.ReplicationApplied), wideEventAttributes(sourceNotFoundRecord)["phase"].AsString())
+	s.Equal(wideevents.ParentChildOutcomeVerified, wideEventAttributes(sourceNotFoundRecord)["outcome"].AsString())
+	sourceNotFoundDetails := wideEventDetails(sourceNotFoundRecord)
+	s.Equal(util.ErrorType(serviceerror.NewNotFound("")), sourceNotFoundDetails["error_type"])
 }

 // Async resend: the RPC returns the verification error immediately and the pull runs in the
 // background.
 func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentAsync() {
 	s.config.EnableAsyncParentWorkflowResend = func() bool { return true }
+	s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true)
+	capture := s.parentChildEventCapture

 	request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{
 		NamespaceId: tests.ParentNamespaceID.String(),
@@ -2819,12 +2888,60 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentAs
 	case <-time.After(10 * time.Second):
 		s.Fail("background resend did not call SyncWorkflowState")
 	}
+	await.RequireTrue(s.T(), func() bool {
+		return len(parentChildOutcomes(capture)) >= 3
+	}, 10*time.Second, 10*time.Millisecond)
+	s.Require().Equal([]string{
+		string(wideevents.ParentChildOutcomeScheduled),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeFailed),
+	}, parentChildOutcomes(capture))
+	records := parentChildRecords(capture)
+	attributes := wideEventAttributes(records[2])
+	details := wideEventDetails(records[2])
+	s.Equal(util.ErrorType(serviceerror.NewUnavailable("source cluster unavailable")), details["error_type"])
+	s.Equal(util.ErrorType(&serviceerror.NotFound{}), details["initial_error_type"])
+	s.Equal("sync_workflow_state", details["stage"])
+	s.Equal(cluster.TestCurrentClusterName, attributes["source_cluster"].AsString())
+}
+
+func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentLimited() {
+	s.config.EnableAsyncParentWorkflowResend = func() bool { return true }
+	s.config.ParentWorkflowResendMaxInFlight = func() int { return 0 }
+	s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true)
+	capture := s.parentChildEventCapture
+
+	request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{
+		NamespaceId: tests.ParentNamespaceID.String(),
+		ParentExecution: &commonpb.WorkflowExecution{
+			WorkflowId: tests.WorkflowID,
+			RunId:      tests.RunID,
+		},
+		ChildExecution: &commonpb.WorkflowExecution{
+			WorkflowId: "child workflowId",
+			RunId:      "child runId",
+		},
+		ParentInitiatedId:      123,
+		ParentInitiatedVersion: 100,
+		ResendParent:           true,
+	}
+	s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).Return(nil, &serviceerror.NotFound{})
+
+	_, err := s.historyEngine.VerifyChildExecutionCompletionRecorded(metrics.AddMetricsContext(context.Background()), request)
+	var notFound *serviceerror.NotFound
+	s.Require().ErrorAs(err, &notFound)
+	s.Require().Equal([]string{string(wideevents.ParentChildOutcomeLimited)}, parentChildOutcomes(capture))
+	details := wideEventDetails(parentChildRecords(capture)[0])
+	s.Equal(util.ErrorType(&serviceerror.NotFound{}), details["initial_error_type"])
+	s.InDelta(0, details["max_in_flight"], 0)
 }

 // TestVerifyChildExecutionCompletionRecorded_ResendParentDeduped asserts that a second attempt for
 // the same parent does not start a concurrent resend while the first is still running.
 func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentDeduped() {
 	s.config.EnableAsyncParentWorkflowResend = func() bool { return true }
+	s.config.EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true)
+	capture := s.parentChildEventCapture

 	request := &historyservice.VerifyChildExecutionCompletionRecordedRequest{
 		NamespaceId: tests.ParentNamespaceID.String(),
@@ -2853,6 +2970,12 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentDe
 	// Times(1): only one resend may reach the source. It blocks so the second attempt overlaps.
 	entered := make(chan struct{})
 	release := make(chan struct{})
+	released := false
+	defer func() {
+		if !released {
+			close(release)
+		}
+	}()
 	s.mockShard.Resource.RemoteAdminClient.EXPECT().SyncWorkflowState(gomock.Any(), gomock.Any()).
 		DoAndReturn(func(context.Context, *adminservice.SyncWorkflowStateRequest, ...grpc.CallOption) (*adminservice.SyncWorkflowStateResponse, error) {
 			close(entered)
@@ -2873,8 +2996,25 @@ func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_ResendParentDe
 	// Second attempt while the first is in flight: must not start another resend.
 	_, err = s.historyEngine.VerifyChildExecutionCompletionRecorded(ctx, request)
 	s.Error(err)
+	s.Require().Equal([]string{
+		string(wideevents.ParentChildOutcomeScheduled),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeDeduplicated),
+	}, parentChildOutcomes(capture))
+	deduplicatedRecord := parentChildRecords(capture)[2]
+	s.Equal(string(wideevents.ReplicationExecuting), wideEventAttributes(deduplicatedRecord)["phase"].AsString())

 	close(release)
+	released = true
+	await.RequireTrue(s.T(), func() bool {
+		return len(parentChildOutcomes(capture)) >= 4
+	}, 10*time.Second, 10*time.Millisecond)
+	s.Require().Equal([]string{
+		string(wideevents.ParentChildOutcomeScheduled),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeDeduplicated),
+		string(wideevents.ParentChildOutcomeFailed),
+	}, parentChildOutcomes(capture))
 }

 func (s *engine2Suite) TestVerifyChildExecutionCompletionRecorded_WorkflowClosed() {
--- a/service/history/parent_child_events.go
+++ b/service/history/parent_child_events.go
@@ -0,0 +1,205 @@
+package history
+
+import (
+	"go.temporal.io/api/serviceerror"
+	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/wideevents"
+	historyi "go.temporal.io/server/service/history/interfaces"
+	"go.temporal.io/server/service/history/tasks"
+)
+
+func emitChildCompletionVerificationStarted(
+	shardContext historyi.ShardContext,
+	task *tasks.CloseExecutionTask,
+	parentNamespaceID string,
+	parentWorkflowID string,
+	parentRunID string,
+	parentInitiatedID int64,
+	parentInitiatedVersion int64,
+	childWorkflowState string,
+	resendParent bool,
+	attempt int,
+) {
+	if !resendParent || !parentChildLifecycleEnabled(shardContext) {
+		return
+	}
+	payload, details := parentChildEventForCloseTask(
+		shardContext,
+		task,
+		parentNamespaceID,
+		parentWorkflowID,
+		parentRunID,
+		parentInitiatedID,
+		parentInitiatedVersion,
+		childWorkflowState,
+		attempt,
+	)
+	details["phase"] = wideevents.ParentChildPhaseVerifyChildCompletion
+	details["outcome"] = wideevents.ParentChildOutcomeStarted
+	emitParentChildReplicationEvent(
+		shardContext,
+		payload,
+		wideevents.ReplicationExecuting,
+		wideevents.ReplOperationStandbyVerification,
+		"Standby child completion verification started with parent resend requested",
+		nil,
+		details,
+	)
+}
+
+func emitChildCompletionVerificationResult(
+	shardContext historyi.ShardContext,
+	task *tasks.CloseExecutionTask,
+	parentNamespaceID string,
+	parentWorkflowID string,
+	parentRunID string,
+	parentInitiatedID int64,
+	parentInitiatedVersion int64,
+	childWorkflowState string,
+	resendParent bool,
+	attempt int,
+	err error,
+) {
+	if !resendParent || !parentChildLifecycleEnabled(shardContext) {
+		return
+	}
+	outcome, emit := parentChildVerificationOutcome(
+		err,
+		wideevents.ParentChildOutcomeNotFound,
+		wideevents.ParentChildOutcomeCompletionMissing,
+	)
+	switch err.(type) {
+	case nil:
+		outcome, emit = wideevents.ParentChildOutcomeVerified, true
+	case *serviceerror.NamespaceNotFound, *serviceerror.Unimplemented:
+		outcome, emit = wideevents.ParentChildOutcomeIgnored, true
+	default:
+	}
+	if !emit {
+		return
+	}
+	payload, details := parentChildEventForCloseTask(
+		shardContext,
+		task,
+		parentNamespaceID,
+		parentWorkflowID,
+		parentRunID,
+		parentInitiatedID,
+		parentInitiatedVersion,
+		childWorkflowState,
+		attempt,
+	)
+	details["phase"] = wideevents.ParentChildPhaseVerifyChildCompletion
+	details["outcome"] = outcome
+	if attempt >= 0 {
+		details["attempt"] = attempt
+	}
+	replicationPhase := wideevents.ReplicationError
+	if err == nil || outcome == wideevents.ParentChildOutcomeIgnored {
+		replicationPhase = wideevents.ReplicationApplied
+	}
+	emitParentChildReplicationEvent(
+		shardContext,
+		payload,
+		replicationPhase,
+		wideevents.ReplOperationStandbyVerification,
+		"Standby child completion verification completed",
+		err,
+		details,
+	)
+}
+
+func parentChildEventForCloseTask(
+	shardContext historyi.ShardContext,
+	task *tasks.CloseExecutionTask,
+	parentNamespaceID string,
+	parentWorkflowID string,
+	parentRunID string,
+	parentInitiatedID int64,
+	parentInitiatedVersion int64,
+	childWorkflowState string,
+	attempt int,
+) (wideevents.ReplicationLifecyclePayload, map[string]any) {
+	payload, details := parentChildEventForTask(shardContext, task, attempt)
+	payload.ParentWorkflowID = parentWorkflowID
+	payload.ParentRunID = parentRunID
+	payload.ParentInitiatedID = parentInitiatedID
+	details["parent_namespace_id"] = parentNamespaceID
+	details["child_workflow_state"] = childWorkflowState
+	details["parent_initiated_version"] = parentInitiatedVersion
+	return payload, details
+}
+
+func parentChildEventForTask(
+	shardContext historyi.ShardContext,
+	task tasks.Task,
+	attempt int,
+) (wideevents.ReplicationLifecyclePayload, map[string]any) {
+	namespaceName := ""
+	if name, err := shardContext.GetNamespaceRegistry().GetNamespaceName(namespace.ID(task.GetNamespaceID())); err == nil {
+		namespaceName = name.String()
+	}
+	details := map[string]any{
+		"event_type":    wideevents.ParentChildLifecycleEventType,
+		"local_task_id": task.GetTaskID(),
+	}
+	if versionedTask, ok := task.(tasks.HasVersion); ok {
+		details["version"] = versionedTask.GetVersion()
+	}
+	return wideevents.ReplicationLifecyclePayload{
+		TaskType:    task.GetType().String(),
+		Shard:       shardContext.GetShardID(),
+		Namespace:   namespaceName,
+		NamespaceID: task.GetNamespaceID(),
+		WorkflowID:  task.GetWorkflowID(),
+		RunID:       task.GetRunID(),
+		Attempt:     int32(attempt),
+	}, details
+}
+
+func emitParentChildReplicationEvent(
+	shardContext historyi.ShardContext,
+	payload wideevents.ReplicationLifecyclePayload,
+	phase wideevents.ReplicationPhase,
+	operation string,
+	message string,
+	err error,
+	details map[string]any,
+) {
+	if phase == wideevents.ReplicationError {
+		wideevents.EmitReplicationError(shardContext.GetEventLogger(), payload, operation, message, err, details)
+		return
+	}
+	details["operation"] = operation
+	details["message"] = message
+	payload.Phase = phase
+	payload.Details = details
+	if phase == wideevents.ReplicationApplied {
+		payload.Outcome = wideevents.ParentChildOutcomeVerified
+	}
+	wideevents.Emit(shardContext.GetEventLogger(), payload)
+}
+
+func parentChildLifecycleEnabled(shardContext historyi.ShardContext) bool {
+	return shardContext.GetConfig().EmitReplicationLifecycleEvents()
+}
+
+func parentChildVerificationOutcome(
+	err error,
+	notFoundOutcome string,
+	notReadyOutcome string,
+) (string, bool) {
+	if err == nil {
+		return "", false
+	}
+	switch err.(type) {
+	case *serviceerror.NotFound:
+		return notFoundOutcome, true
+	case *serviceerror.WorkflowNotReady:
+		return notReadyOutcome, true
+	case *serviceerror.NamespaceNotFound, *serviceerror.Unimplemented:
+		return "", false
+	default:
+		return wideevents.ParentChildOutcomeFailed, true
+	}
+}
--- a/service/history/parent_child_events_test.go
+++ b/service/history/parent_child_events_test.go
@@ -0,0 +1,80 @@
+package history
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+
+	otellog "go.opentelemetry.io/otel/log"
+	"go.opentelemetry.io/otel/log/embedded"
+	"go.temporal.io/server/common/wideevents"
+)
+
+type parentChildEventCapture struct {
+	embedded.Logger
+
+	mu      sync.RWMutex
+	records []otellog.Record
+}
+
+func (c *parentChildEventCapture) Emit(_ context.Context, record otellog.Record) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.records = append(c.records, record.Clone())
+}
+
+func (*parentChildEventCapture) Enabled(context.Context, otellog.EnabledParameters) bool {
+	return true
+}
+
+func (c *parentChildEventCapture) snapshot() []otellog.Record {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+	records := make([]otellog.Record, len(c.records))
+	for i, record := range c.records {
+		records[i] = record.Clone()
+	}
+	return records
+}
+
+func wideEventAttributes(record otellog.Record) map[string]otellog.Value {
+	attributes := make(map[string]otellog.Value, record.AttributesLen())
+	record.WalkAttributes(func(attribute otellog.KeyValue) bool {
+		attributes[attribute.Key] = attribute.Value
+		return true
+	})
+	return attributes
+}
+
+func parentChildOutcomes(capture *parentChildEventCapture) []string {
+	var outcomes []string
+	for _, record := range parentChildRecords(capture) {
+		if outcome, ok := wideEventDetails(record)["outcome"].(string); ok {
+			outcomes = append(outcomes, outcome)
+		}
+	}
+	return outcomes
+}
+
+func parentChildRecords(capture *parentChildEventCapture) []otellog.Record {
+	var records []otellog.Record
+	for _, record := range capture.snapshot() {
+		if record.EventName() == wideevents.ReplicationLifecycleEventName &&
+			wideEventDetails(record)["event_type"] == wideevents.ParentChildLifecycleEventType {
+			records = append(records, record)
+		}
+	}
+	return records
+}
+
+func wideEventDetails(record otellog.Record) map[string]any {
+	attribute, ok := wideEventAttributes(record)["details"]
+	if !ok {
+		return nil
+	}
+	var details map[string]any
+	if err := json.Unmarshal([]byte(attribute.AsString()), &details); err != nil {
+		return nil
+	}
+	return details
+}
--- a/service/history/shard/context_testutil.go
+++ b/service/history/shard/context_testutil.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"sync"

+	otellog "go.opentelemetry.io/otel/log"
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
@@ -44,8 +45,9 @@ func NewTestContextWithTimeSource(
 	shardInfo *persistencespb.ShardInfo,
 	config *configs.Config,
 	timeSource clock.TimeSource,
+	eventLogger ...otellog.Logger,
 ) *ContextTest {
-	result := NewTestContext(ctrl, shardInfo, config)
+	result := NewTestContext(ctrl, shardInfo, config, eventLogger...)
 	result.timeSource = timeSource
 	result.taskKeyManager.generator.timeSource = timeSource
 	result.Resource.TimeSource = timeSource
@@ -56,15 +58,21 @@ func NewTestContext(
 	ctrl *gomock.Controller,
 	shardInfo *persistencespb.ShardInfo,
 	config *configs.Config,
+	eventLogger ...otellog.Logger,
 ) *ContextTest {
+	var logger otellog.Logger
+	if len(eventLogger) > 0 {
+		logger = eventLogger[0]
+	}
 	resourceTest := resourcetest.NewTest(ctrl, primitives.HistoryService)
 	eventsCache := events.NewMockCache(ctrl)
 	shard := newTestContext(
 		resourceTest,
 		eventsCache,
 		ContextConfigOverrides{
-			ShardInfo: shardInfo,
-			Config:    config,
+			ShardInfo:   shardInfo,
+			Config:      config,
+			EventLogger: logger,
 		},
 	)
 	return &ContextTest{
@@ -80,6 +88,7 @@ type ContextConfigOverrides struct {
 	Registry         namespace.Registry
 	ClusterMetadata  cluster.Metadata
 	ExecutionManager persistence.ExecutionManager
+	EventLogger      otellog.Logger
 }

 type StubContext struct {
@@ -141,6 +150,7 @@ func newTestContext(t *resourcetest.Test, eventsCache events.Cache, config Conte
 		lifecycleCtx:        lifecycleCtx,
 		lifecycleCancel:     lifecycleCancel,
 		queueMetricEmitter:  sync.Once{},
+		eventLogger:         config.EventLogger,

 		state:              contextStateAcquired,
 		engineFuture:       future.NewFuture[historyi.Engine](),
--- a/service/history/transfer_queue_standby_task_executor.go
+++ b/service/history/transfer_queue_standby_task_executor.go
@@ -103,7 +103,7 @@ func (t *transferQueueStandbyTaskExecutor) Execute(
 		// TODO: add error logs
 		err = nil
 	case *tasks.CloseExecutionTask:
-		err = t.processCloseExecution(ctx, task)
+		err = t.processCloseExecution(ctx, task, executable.Attempt())
 	case *tasks.DeleteExecutionTask:
 		err = t.processDeleteExecutionTask(ctx, task, false)
 	case *tasks.ChasmTask:
@@ -294,6 +294,7 @@ func (t *transferQueueStandbyTaskExecutor) processWorkflowTask(
 func (t *transferQueueStandbyTaskExecutor) processCloseExecution(
 	ctx context.Context,
 	transferTask *tasks.CloseExecutionTask,
+	attempt int,
 ) error {
 	processTaskIfClosed := true
 	actionFn := func(ctx context.Context, wfContext historyi.WorkflowContext, mutableState historyi.MutableState, release historyi.ReleaseWorkflowContextFunc) (any, error) {
@@ -339,10 +340,24 @@ func (t *transferQueueStandbyTaskExecutor) processCloseExecution(
 			parentInitiatedID := executionInfo.ParentInitiatedId
 			parentInitiatedVersion := executionInfo.ParentInitiatedVersion
 			parentClock := executionInfo.ParentClock
+			childWorkflowState := mutableState.GetExecutionState().GetState().String()

 			// no need for mutable state anymore, release workflow lock
 			release(nil)

+			emitChildCompletionVerificationStarted(
+				t.shardContext,
+				transferTask,
+				parentNamespaceID,
+				parentWorkflowID,
+				parentRunID,
+				parentInitiatedID,
+				parentInitiatedVersion,
+				childWorkflowState,
+				resendParent,
+				attempt,
+			)
+
 			_, err := t.historyRawClient.VerifyChildExecutionCompletionRecorded(ctx, &historyservice.VerifyChildExecutionCompletionRecordedRequest{
 				NamespaceId: parentNamespaceID,
 				ParentExecution: &commonpb.WorkflowExecution{
@@ -358,6 +373,19 @@ func (t *transferQueueStandbyTaskExecutor) processCloseExecution(
 				Clock:                  parentClock,
 				ResendParent:           resendParent,
 			})
+			emitChildCompletionVerificationResult(
+				t.shardContext,
+				transferTask,
+				parentNamespaceID,
+				parentWorkflowID,
+				parentRunID,
+				parentInitiatedID,
+				parentInitiatedVersion,
+				childWorkflowState,
+				resendParent,
+				attempt,
+				err,
+			)
 			switch err.(type) {
 			case nil, *serviceerror.NamespaceNotFound, *serviceerror.Unimplemented:
 				// Case 1: Target workflow is in the desired state.
@@ -491,7 +519,6 @@ func (t *transferQueueStandbyTaskExecutor) processStartChildExecution(
 		childStartedWorkflowID := childWorkflowInfo.StartedWorkflowId
 		childStartedRunID := childWorkflowInfo.StartedRunId
 		childClock := childWorkflowInfo.Clock
-
 		// no need for mutable state anymore, release workflow lock
 		release(nil)

--- a/service/history/transfer_queue_standby_task_executor_test.go
+++ b/service/history/transfer_queue_standby_task_executor_test.go
@@ -30,6 +30,7 @@ import (
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/definition"
+	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
@@ -41,6 +42,8 @@ import (
 	"go.temporal.io/server/common/telemetry"
 	"go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1"
 	"go.temporal.io/server/common/testing/protomock"
+	"go.temporal.io/server/common/util"
+	"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/hsm"
@@ -76,9 +79,10 @@ type (
 		mockRemoteAdminClient *adminservicemock.MockAdminServiceClient
 		mockChasmEngine       chasm.Engine

-		mockExecutionMgr     *persistence.MockExecutionManager
-		mockArchivalMetadata archiver.MetadataMock
-		mockArchiverProvider *provider.MockArchiverProvider
+		mockExecutionMgr        *persistence.MockExecutionManager
+		mockArchivalMetadata    archiver.MetadataMock
+		mockArchiverProvider    *provider.MockArchiverProvider
+		parentChildEventCapture *parentChildEventCapture

 		workflowCache             wcache.Cache
 		logger                    log.Logger
@@ -124,6 +128,7 @@ func (s *transferQueueStandbyTaskExecutorSuite) SetupTest() {
 	s.chasmDiscardDuration = config.ChasmStandbyTaskDiscardDelay("")

 	s.controller = gomock.NewController(s.T())
+	s.parentChildEventCapture = &parentChildEventCapture{}
 	s.mockShard = shard.NewTestContextWithTimeSource(
 		s.controller,
 		&persistencespb.ShardInfo{
@@ -131,6 +136,7 @@ func (s *transferQueueStandbyTaskExecutorSuite) SetupTest() {
 		},
 		config,
 		s.timeSource,
+		s.parentChildEventCapture,
 	)

 	reg := hsm.NewRegistry()
@@ -725,6 +731,9 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessWorkflowTask_StampMis
 }

 func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCloseExecution() {
+	capture := s.parentChildEventCapture
+	s.mockShard.GetConfig().EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true)
+
 	execution := &commonpb.WorkflowExecution{
 		WorkflowId: "some random workflow ID",
 		RunId:      uuid.NewString(),
@@ -841,8 +850,17 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCloseExecution() {
 	s.ErrorAs(resp.ExecutionErr, &verificationErr)
 	var resourceExhaustedErr *serviceerror.ResourceExhausted
 	s.ErrorAs(resp.ExecutionErr, &resourceExhaustedErr)
+	s.Require().Empty(parentChildRecords(capture))

 	s.mockShard.SetCurrentTime(s.clusterName, now.Add(s.localVerificationDuration))
+	s.mockHistoryClient.EXPECT().VerifyChildExecutionCompletionRecorded(gomock.Any(), expectedVerificationWithResendParentRequest).Return(nil, nil)
+	resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask))
+	s.NoError(resp.ExecutionErr)
+
+	s.mockHistoryClient.EXPECT().VerifyChildExecutionCompletionRecorded(gomock.Any(), expectedVerificationWithResendParentRequest).Return(nil, serviceerror.NewUnimplemented("not implemented"))
+	resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask))
+	s.NoError(resp.ExecutionErr)
+
 	s.mockHistoryClient.EXPECT().VerifyChildExecutionCompletionRecorded(gomock.Any(), expectedVerificationWithResendParentRequest).Return(nil, consts.ErrWorkflowNotReady)
 	resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask))
 	s.Equal(consts.ErrTaskRetry, resp.ExecutionErr)
@@ -862,6 +880,68 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCloseExecution() {
 	resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask))
 	s.ErrorAs(resp.ExecutionErr, &verificationErr)
 	s.Equal(randomErr, verificationErr.Unwrap())
+
+	s.Equal([]string{
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeVerified),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeIgnored),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeCompletionMissing),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeCompletionMissing),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeCompletionMissing),
+		string(wideevents.ParentChildOutcomeStarted),
+		string(wideevents.ParentChildOutcomeFailed),
+	}, parentChildOutcomes(capture))
+	records := parentChildRecords(capture)
+	s.Require().Len(records, 12)
+	attributes := wideEventAttributes(records[0])
+	details := wideEventDetails(records[0])
+	s.Equal(string(wideevents.ReplicationExecuting), attributes["phase"].AsString())
+	s.Equal(wideevents.ParentChildPhaseVerifyChildCompletion, details["phase"])
+	s.Equal(wideevents.ParentChildOutcomeStarted, details["outcome"])
+	s.Equal(s.namespaceID.String(), attributes["namespace_id"].AsString())
+	s.Equal(parentExecution.GetWorkflowId(), attributes["parent_workflow_id"].AsString())
+	s.Equal(parentExecution.GetRunId(), attributes["parent_run_id"].AsString())
+	s.Equal(execution.GetWorkflowId(), attributes["workflow_id"].AsString())
+	s.Equal(execution.GetRunId(), attributes["run_id"].AsString())
+	s.Equal(transferTask.GetType().String(), attributes["task_type"].AsString())
+	s.Equal(int64(1), attributes["attempt"].AsInt64())
+	s.Equal("Completed", details["child_workflow_state"])
+	s.Equal(parentInitiatedID, attributes["parent_initiated_id"].AsInt64())
+	s.Equal(parentNamespaceID, details["parent_namespace_id"])
+	s.InDelta(float64(parentInitiatedVersion), details["parent_initiated_version"], 0)
+	s.InDelta(float64(taskID), details["local_task_id"], 0)
+	s.InDelta(float64(transferTask.GetVersion()), details["version"], 0)
+	s.Require().NotContains(details, "attempt")
+	for _, key := range []string{
+		"child_namespace_id",
+		"child_workflow_id",
+		"child_run_id",
+		"local_cluster",
+		"local_task_type",
+		"parent_initiated_id",
+		"parent_run_id",
+		"parent_workflow_id",
+		"resend_parent_requested",
+		"verification_scope",
+	} {
+		for _, record := range records {
+			s.Require().NotContains(wideEventDetails(record), key)
+		}
+	}
+	missingDetails := wideEventDetails(records[5])
+	s.Equal(wideevents.ParentChildOutcomeCompletionMissing, missingDetails["outcome"])
+	s.Equal(util.ErrorType(consts.ErrWorkflowNotReady), missingDetails["error_type"])
+	s.InDelta(1, missingDetails["attempt"], 0)
+	verifiedAttributes := wideEventAttributes(records[1])
+	s.Require().NotContains(verifiedAttributes, "attempt")
+	s.InDelta(1, wideEventDetails(records[1])["attempt"], 0)
+	ignoredAttributes := wideEventAttributes(records[3])
+	s.Equal(string(wideevents.ReplicationApplied), ignoredAttributes["phase"].AsString())
+	s.Equal(wideevents.ParentChildOutcomeVerified, ignoredAttributes["outcome"].AsString())
 }

 func (s *transferQueueStandbyTaskExecutorSuite) TestProcessCancelExecution_Pending() {
@@ -1125,6 +1205,9 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessSignalExecution_Succe
 }

 func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_Pending() {
+	capture := s.parentChildEventCapture
+	s.mockShard.GetConfig().EmitReplicationLifecycleEvents = dynamicconfig.GetBoolPropertyFn(true)
+
 	execution := &commonpb.WorkflowExecution{
 		WorkflowId: "some random workflow ID",
 		RunId:      uuid.NewString(),
@@ -1185,7 +1268,8 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_P
 	resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask))
 	s.Equal(consts.ErrTaskRetry, resp.ExecutionErr)

-	event = addChildWorkflowExecutionStartedEvent(mutableState, event.GetEventId(), childWorkflowID, uuid.NewString(), childWorkflowType, nil)
+	childRunID := uuid.NewString()
+	event = addChildWorkflowExecutionStartedEvent(mutableState, event.GetEventId(), childWorkflowID, childRunID, childWorkflowType, nil)
 	mutableState.FlushBufferedEvents()

 	// clear the cache
@@ -1230,6 +1314,8 @@ func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_P
 	s.mockHistoryClient.EXPECT().VerifyFirstWorkflowTaskScheduled(gomock.Any(), gomock.Any()).Return(nil, nil)
 	resp = s.transferQueueStandbyTaskExecutor.Execute(context.Background(), s.newTaskExecutable(transferTask))
 	s.NoError(resp.ExecutionErr)
+
+	s.Require().Empty(parentChildRecords(capture))
 }

 func (s *transferQueueStandbyTaskExecutorSuite) TestProcessStartChildExecution_Success() {