fix: send actual EndTime in UpdateActionStatus to fix inflated durations (#7089)

Signed-off-by: Kevin Su <pingsutw@apache.org>

Author: pingsutw

actions/k8s/client.go

@@ -458,7 +458,8 @@ func (c *ActionsClient) notifyRunService(ctx context.Context, taskAction *execut
 	// On ADDED: create the action record in the DB (deduplicated via bloom filter).
 	if eventType == watch.Added {
 		actionKey := []byte(buildTaskActionName(update.ActionID))
-		if c.recordedFilter != nil && c.recordedFilter.Contains(ctx, actionKey) {
+		isDuplicate := c.recordedFilter != nil && c.recordedFilter.Contains(ctx, actionKey)
+		if isDuplicate {
 			logger.Debugf(ctx, "Skipping duplicate RecordAction for %s", update.ActionID.Name)
 		} else {
 			recordReq := &workflow.RecordActionRequest{
@@ -498,6 +499,25 @@ func (c *ActionsClient) notifyRunService(ctx context.Context, taskAction *execut
 				c.recordedFilter.Add(ctx, actionKey)
 			}
 		}
+
+		// When a child action appears, the parent must already be running (it
+		// created the child). Promote the parent to RUNNING so the UI doesn't
+		// stay stuck on INITIALIZING while children are executing.
+		if !isDuplicate && update.ParentActionName != "" {
+			parentID := &common.ActionIdentifier{
+				Run:  update.ActionID.Run,
+				Name: update.ParentActionName,
+			}
+			parentStatusReq := &workflow.UpdateActionStatusRequest{
+				ActionId: parentID,
+				Status: &workflow.ActionStatus{
+					Phase: common.ActionPhase_ACTION_PHASE_RUNNING,
+				},
+			}
+			if _, err := c.runClient.UpdateActionStatus(ctx, connect.NewRequest(parentStatusReq)); err != nil {
+				logger.Warnf(ctx, "Failed to promote parent action %s to RUNNING: %v", update.ParentActionName, err)
+			}
+		}
 	}
 
 	if update.Phase != common.ActionPhase_ACTION_PHASE_UNSPECIFIED {
@@ -507,7 +527,7 @@ func (c *ActionsClient) notifyRunService(ctx context.Context, taskAction *execut
 				Phase:       update.Phase,
 				Attempts:    taskAction.Status.Attempts,
 				CacheStatus: taskAction.Status.CacheStatus,
-			},
+				},
 		}
 		if _, err := c.runClient.UpdateActionStatus(ctx, connect.NewRequest(statusReq)); err != nil {
 			logger.Warnf(ctx, "Failed to update action status in run service for %s: %v", update.ActionID.Name, err)

actions/k8s/client_test.go

@@ -262,3 +262,240 @@ func TestApplyRunSpecToTaskAction_ProjectsRuntimeSettings(t *testing.T) {
 	assert.Equal(t, "platform", taskAction.Labels["team"])
 	assert.Equal(t, "sdk", taskAction.Annotations["owner"])
 }
+
+func TestNotifyRunService_ChildAddedPromotesParentToRunning(t *testing.T) {
+	ctx := context.Background()
+
+	mockClient := runmocks.NewInternalRunServiceClient(t)
+	c := &ActionsClient{
+		runClient:   mockClient,
+		subscribers: make(map[string]map[chan *ActionUpdate]struct{}),
+	}
+
+	runID := &common.RunIdentifier{
+		Org:     "org",
+		Project: "proj",
+		Domain:  "dev",
+		Name:    "run1",
+	}
+
+	ta := &executorv1.TaskAction{
+		Spec: executorv1.TaskActionSpec{
+			Org:        runID.Org,
+			Project:    runID.Project,
+			Domain:     runID.Domain,
+			RunName:    runID.Name,
+			ActionName: "child-1",
+		},
+	}
+	update := &ActionUpdate{
+		ActionID: &common.ActionIdentifier{
+			Run:  runID,
+			Name: "child-1",
+		},
+		ParentActionName: "run1",
+	}
+
+	// Expect RecordAction for the child
+	mockClient.On("RecordAction", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once()
+
+	// Expect UpdateActionStatus for the PARENT with RUNNING phase
+	mockClient.On("UpdateActionStatus", mock.Anything, mock.MatchedBy(func(req *connect.Request[workflow.UpdateActionStatusRequest]) bool {
+		return req.Msg.GetActionId().GetName() == "run1" &&
+			req.Msg.GetStatus().GetPhase() == common.ActionPhase_ACTION_PHASE_RUNNING
+	})).Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once()
+
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	mockClient.AssertNumberOfCalls(t, "RecordAction", 1)
+	mockClient.AssertNumberOfCalls(t, "UpdateActionStatus", 1)
+}
+
+func TestNotifyRunService_SkipsTerminalAddedEventsOnlyWhenInBloomFilter(t *testing.T) {
+	ctx := context.Background()
+
+	mockClient := runmocks.NewInternalRunServiceClient(t)
+	filter, err := fastcheck.NewOppoBloomFilter(128, promutils.NewTestScope())
+	require.NoError(t, err)
+
+	c := &ActionsClient{
+		runClient:      mockClient,
+		recordedFilter: filter,
+		subscribers:    make(map[string]map[chan *ActionUpdate]struct{}),
+	}
+
+	// Create a terminal TaskAction CRD (SUCCEEDED)
+	ta := &executorv1.TaskAction{
+		Spec: executorv1.TaskActionSpec{
+			Org:        "org",
+			Project:    "proj",
+			Domain:     "dev",
+			RunName:    "run1",
+			ActionName: "completed-action",
+		},
+		Status: executorv1.TaskActionStatus{
+			Conditions: []metav1.Condition{
+				{Type: string(executorv1.ConditionTypeSucceeded), Status: "True"},
+			},
+		},
+	}
+	update := &ActionUpdate{
+		ActionID: &common.ActionIdentifier{
+			Run:  &common.RunIdentifier{Org: "org", Project: "proj", Domain: "dev", Name: "run1"},
+			Name: "completed-action",
+		},
+		Phase: common.ActionPhase_ACTION_PHASE_SUCCEEDED,
+	}
+
+	// First ADDED event (cold start, not in bloom filter): should process normally
+	mockClient.On("RecordAction", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once()
+	mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil)
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	mockClient.AssertNumberOfCalls(t, "RecordAction", 1)
+	mockClient.AssertNumberOfCalls(t, "UpdateActionStatus", 1)
+
+	// Action should now be in the bloom filter
+	actionKey := []byte(buildTaskActionName(update.ActionID))
+	assert.True(t, filter.Contains(ctx, actionKey))
+
+	// Second ADDED event (reconnect, in bloom filter): should skip RecordAction
+	// but still call UpdateActionStatus.
+	c.notifyRunService(ctx, ta, update, watch.Added)
+	mockClient.AssertNumberOfCalls(t, "RecordAction", 1)       // no new RecordAction
+	mockClient.AssertNumberOfCalls(t, "UpdateActionStatus", 2) // one more UpdateActionStatus
+}
+
+func TestNotifyRunService_ProcessesNonTerminalAddedEvents(t *testing.T) {
+	ctx := context.Background()
+
+	mockClient := runmocks.NewInternalRunServiceClient(t)
+	filter, err := fastcheck.NewOppoBloomFilter(128, promutils.NewTestScope())
+	require.NoError(t, err)
+
+	c := &ActionsClient{
+		runClient:      mockClient,
+		recordedFilter: filter,
+		subscribers:    make(map[string]map[chan *ActionUpdate]struct{}),
+	}
+
+	// Create a non-terminal TaskAction CRD (QUEUED)
+	ta := &executorv1.TaskAction{
+		Spec: executorv1.TaskActionSpec{
+			Org:        "org",
+			Project:    "proj",
+			Domain:     "dev",
+			RunName:    "run1",
+			ActionName: "queued-action",
+		},
+		Status: executorv1.TaskActionStatus{
+			Conditions: []metav1.Condition{
+				{Type: string(executorv1.ConditionTypeProgressing), Status: "True", Reason: string(executorv1.ConditionReasonQueued)},
+			},
+		},
+	}
+	update := &ActionUpdate{
+		ActionID: &common.ActionIdentifier{
+			Run:  &common.RunIdentifier{Org: "org", Project: "proj", Domain: "dev", Name: "run1"},
+			Name: "queued-action",
+		},
+		Phase: common.ActionPhase_ACTION_PHASE_QUEUED,
+	}
+
+	// Non-terminal ADDED events should be processed normally
+	mockClient.On("RecordAction", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once()
+	mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Once()
+
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	mockClient.AssertNumberOfCalls(t, "RecordAction", 1)
+	mockClient.AssertNumberOfCalls(t, "UpdateActionStatus", 1)
+}
+
+func TestNotifyRunService_DuplicateAddedSkipsRecordAction(t *testing.T) {
+	ctx := context.Background()
+
+	mockClient := runmocks.NewInternalRunServiceClient(t)
+	filter, err := fastcheck.NewOppoBloomFilter(128, promutils.NewTestScope())
+	require.NoError(t, err)
+
+	c := &ActionsClient{
+		runClient:      mockClient,
+		recordedFilter: filter,
+		subscribers:    make(map[string]map[chan *ActionUpdate]struct{}),
+	}
+
+	ta, update := newTestActionUpdate("action-dup")
+	update.Phase = common.ActionPhase_ACTION_PHASE_RUNNING
+
+	// First call — should process normally
+	mockClient.On("RecordAction", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once()
+	mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil)
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	// Second call (duplicate ADDED) — should skip RecordAction but still call UpdateActionStatus
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	mockClient.AssertNumberOfCalls(t, "RecordAction", 1)
+	mockClient.AssertNumberOfCalls(t, "UpdateActionStatus", 2)
+}
+
+func TestNotifyRunService_TerminalDuplicateRepairsTimestamps(t *testing.T) {
+	ctx := context.Background()
+
+	mockClient := runmocks.NewInternalRunServiceClient(t)
+	filter, err := fastcheck.NewOppoBloomFilter(128, promutils.NewTestScope())
+	require.NoError(t, err)
+
+	c := &ActionsClient{
+		runClient:      mockClient,
+		recordedFilter: filter,
+		subscribers:    make(map[string]map[chan *ActionUpdate]struct{}),
+	}
+
+	ta, update := newTestActionUpdate("action-terminal-dup")
+	update.Phase = common.ActionPhase_ACTION_PHASE_SUCCEEDED
+
+	// First call — should process normally (RecordAction + UpdateActionStatus)
+	mockClient.On("RecordAction", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once()
+	mockClient.On("UpdateActionStatus", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.UpdateActionStatusResponse]{}, nil).Times(2)
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	// Second call (terminal duplicate ADDED) — should skip RecordAction but
+	// still call UpdateActionStatus to repair missing timestamps.
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	mockClient.AssertNumberOfCalls(t, "RecordAction", 1)       // no new RecordAction
+	mockClient.AssertNumberOfCalls(t, "UpdateActionStatus", 2) // one more UpdateActionStatus
+}
+
+func TestNotifyRunService_RootActionAddedDoesNotPromoteParent(t *testing.T) {
+	ctx := context.Background()
+
+	mockClient := runmocks.NewInternalRunServiceClient(t)
+	c := &ActionsClient{
+		runClient:   mockClient,
+		subscribers: make(map[string]map[chan *ActionUpdate]struct{}),
+	}
+
+	// Root action has no parent
+	ta, update := newTestActionUpdate("action-root")
+
+	mockClient.On("RecordAction", mock.Anything, mock.Anything).
+		Return(&connect.Response[workflow.RecordActionResponse]{}, nil).Once()
+
+	c.notifyRunService(ctx, ta, update, watch.Added)
+
+	mockClient.AssertNumberOfCalls(t, "RecordAction", 1)
+	// No UpdateActionStatus should be called for root (no parent to promote)
+	mockClient.AssertNumberOfCalls(t, "UpdateActionStatus", 0)
+}

executor/pkg/controller/taskaction_controller.go

@@ -384,7 +384,6 @@ func (r *TaskActionReconciler) buildActionEvent(
 
 	info := phaseInfo.Info()
 	updatedTime := updatedTimestamp(taskAction.Status.PhaseHistory)
-	reportedTime := reportedTimestamp(info)
 
 	event := &workflow.ActionEvent{
 		Id:            actionID,
@@ -396,7 +395,7 @@ func (r *TaskActionReconciler) buildActionEvent(
 		Cluster:       r.cluster,
 		Outputs:       outputRefs(taskAction.Spec.RunOutputBase, taskAction.Spec.ActionName),
 		ClusterEvents: toClusterEvents(info, updatedTime),
-		ReportedTime:  reportedTime,
+		ReportedTime:  timestamppb.New(time.Now()),
 	}
 
 	if info != nil {
@@ -430,13 +429,6 @@ func updatedTimestamp(history []flyteorgv1.PhaseTransition) *timestamppb.Timesta
 	return timestamppb.Now()
 }
 
-func reportedTimestamp(info *pluginsCore.TaskInfo) *timestamppb.Timestamp {
-	if info != nil && info.ReportedAt != nil {
-		return timestamppb.New(*info.ReportedAt)
-	}
-	return timestamppb.Now()
-}
-
 func outputRefs(runOutputBase, actionName string) *task.OutputReferences {
 	if runOutputBase == "" {
 		return nil
@@ -517,7 +509,7 @@ func cacheStatusFromExternalResources(resources []*pluginsCore.ExternalResource)
 }
 
 // taskActionStatusChanged reports whether any status field has changed between old and new,
-// covering plugin phase, state, state version, observability JSON, and conditions.
+// covering plugin phase, state, state version, observability JSON, conditions, and phase history.
 func taskActionStatusChanged(oldStatus, newStatus flyteorgv1.TaskActionStatus) bool {
 	if oldStatus.StateJSON != newStatus.StateJSON ||
 		oldStatus.PluginStateVersion != newStatus.PluginStateVersion ||
@@ -532,6 +524,10 @@ func taskActionStatusChanged(oldStatus, newStatus flyteorgv1.TaskActionStatus) b
 		return true
 	}
 
+	if len(oldStatus.PhaseHistory) != len(newStatus.PhaseHistory) {
+		return true
+	}
+
 	return !reflect.DeepEqual(oldStatus.Conditions, newStatus.Conditions)
 }
 
@@ -596,7 +592,7 @@ func mapPhaseToConditions(ta *flyteorgv1.TaskAction, info pluginsCore.PhaseInfo)
 			flyteorgv1.ConditionReasonAborted, msg)
 	}
 
-	// Append to PhaseHistory if this is a new phase (dedup by checking last entry)
+	// Append to PhaseHistory if this is a new phase (dedup by checking last entry).
 	if phaseName != "" {
 		n := len(ta.Status.PhaseHistory)
 		if n == 0 || ta.Status.PhaseHistory[n-1].Phase != phaseName {