feat: show proposed changes with clickable entity links in approval tasks

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/elements/nodes/userTask/CreateTask.java

@@ -69,6 +69,7 @@
 import org.openmetadata.service.governance.workflows.WorkflowVariableHandler;
 import org.openmetadata.service.governance.workflows.elements.TriggerFactory;
 import org.openmetadata.service.governance.workflows.elements.nodes.userTask.helper.WorkflowVariableResolver;
+import org.openmetadata.service.governance.workflows.util.ChangePreviewUtils;
 import org.openmetadata.service.jdbi3.CollectionDAO;
 import org.openmetadata.service.jdbi3.TaskRepository;
 import org.openmetadata.service.resources.feeds.MessageParser;
@@ -446,6 +447,15 @@ private Task createOrUpdateTask(
       terminateDeletedWorkflowManagedDraftTask(delegateTask, requestedTaskId);
       return null;
     }
+    Object priorApprovalPayload =
+        existingTask == null
+            ? findPriorOpenApprovalPayload(
+                taskRepository,
+                entity,
+                taskCategory,
+                resolvedWorkflowDefinitionId,
+                workflowInstanceId)
+            : null;
     if (existingTask != null) {
       LOG.info(
           "[CreateTask] Updating existing task '{}' stage='{}' workflowAssignees={} requestedAssignees={}",
@@ -488,7 +498,10 @@ private Task createOrUpdateTask(
       updatedTask.setUpdatedAt(System.currentTimeMillis());
       updatedTask.setUpdatedBy(updatedBy);
       updatedTask.setPayload(
-          requestedPayload != null ? requestedPayload : updatedTask.getPayload());
+          requestedPayload != null
+              ? ChangePreviewUtils.preserveProposedChanges(
+                  requestedPayload, updatedTask.getPayload())
+              : updatedTask.getPayload());
       if (effectiveWorkflowDefinitionId != null) {
         updatedTask.setWorkflowDefinitionId(effectiveWorkflowDefinitionId);
       }
@@ -516,6 +529,8 @@ private Task createOrUpdateTask(
         updatedTask.setDueDate(effectiveDueDate);
       }
       updatedTask.setPayload(withGrantExpirationDate(stageStatus, updatedTask.getPayload()));
+      updatedTask.setPayload(
+          applyProposedChangesIfApproval(taskType, entity, updatedTask.getPayload()));
       if (requestedExternalReference != null) {
         updatedTask.setExternalReference(
             JsonUtils.convertValue(requestedExternalReference, TaskExternalReference.class));
@@ -552,7 +567,12 @@ private Task createOrUpdateTask(
             .withAvailableTransitions(availableTransitions)
             .withDescription(
                 taskDescription != null ? taskDescription : buildTaskDescription(entity, taskType))
-            .withPayload(requestedPayload != null ? requestedPayload : payload)
+            .withPayload(
+                ChangePreviewUtils.preserveProposedChanges(
+                    requestedPayload != null
+                        ? requestedPayload
+                        : (payload != null ? payload : priorApprovalPayload),
+                    priorApprovalPayload))
             .withCreatedAt(System.currentTimeMillis())
             .withUpdatedAt(System.currentTimeMillis())
             .withUpdatedBy(updatedBy);
@@ -578,6 +598,7 @@ private Task createOrUpdateTask(
       task.setDueDate(effectiveDueDate);
     }
     task.setPayload(withGrantExpirationDate(stageStatus, task.getPayload()));
+    task.setPayload(applyProposedChangesIfApproval(taskType, entity, task.getPayload()));
     if (requestedExternalReference != null) {
       task.setExternalReference(
           JsonUtils.convertValue(requestedExternalReference, TaskExternalReference.class));
@@ -658,6 +679,45 @@ private void supersedePriorApprovalTask(
     }
   }
 
+  /**
+   * Find an open prior approval task on the same entity bound to the same workflow definition but
+   * a different workflow instance, and return its payload. Used to carry forward the {@code
+   * proposedChanges} blob across re-edits, since each entity edit spawns a fresh workflow run that
+   * creates a new Task (the prior task is closed asynchronously by {@link
+   * #supersedePriorApprovalTask}). Returns {@code null} when no eligible prior task exists.
+   */
+  static Object findPriorOpenApprovalPayload(
+      TaskRepository taskRepository,
+      EntityInterface entity,
+      TaskCategory taskCategory,
+      UUID currentWorkflowDefinitionId,
+      UUID currentWorkflowInstanceId) {
+    Object priorPayload = null;
+    if (taskCategory == TaskCategory.Approval && entity != null) {
+      try {
+        priorPayload =
+            taskRepository
+                .listNonTerminalTasksByEntityAndCategory(
+                    entity.getFullyQualifiedName(), taskCategory)
+                .stream()
+                .filter(
+                    prior ->
+                        isSupersedablePriorApprovalTask(
+                            prior, currentWorkflowDefinitionId, currentWorkflowInstanceId))
+                .map(Task::getPayload)
+                .filter(java.util.Objects::nonNull)
+                .findFirst()
+                .orElse(null);
+      } catch (Exception e) {
+        LOG.warn(
+            "[CreateTask] Failed to lookup prior approval task payload for entity '{}': {}",
+            entity.getFullyQualifiedName(),
+            e.getMessage());
+      }
+    }
+    return priorPayload;
+  }
+
   static boolean isSupersedablePriorApprovalTask(
       Task prior, UUID currentWorkflowDefinitionId, UUID currentWorkflowInstanceId) {
     return prior != null
@@ -894,6 +954,21 @@ static Object withGrantExpirationDate(TaskEntityStatus stageStatus, Object paylo
     return merged;
   }
 
+  /**
+   * For approval task types (GlossaryApproval, RequestApproval), augment the task payload with a
+   * {@code proposedChanges} map computed from the target entity's change description, merged
+   * against any existing {@code proposedChanges} on the prior payload (set-cancellation). Returns
+   * the payload unchanged for non-approval task types or when the entity carries no change
+   * description.
+   */
+  static Object applyProposedChangesIfApproval(
+      TaskEntityType taskType, EntityInterface entity, Object payload) {
+    if (taskType != TaskEntityType.GlossaryApproval && taskType != TaskEntityType.RequestApproval) {
+      return payload;
+    }
+    return ChangePreviewUtils.buildProposedChangesPayload(entity, payload);
+  }
+
   static Long parseMillisFromIso8601Duration(String duration, Long fallback) {
     try {
       return System.currentTimeMillis() + Duration.parse(duration).toMillis();

openmetadata-service/src/main/java/org/openmetadata/service/governance/workflows/util/ChangePreviewUtils.java

@@ -26,19 +26,21 @@
 import java.util.stream.Stream;
 import lombok.extern.slf4j.Slf4j;
 import org.openmetadata.schema.EntityInterface;
-import org.openmetadata.schema.entity.feed.Thread;
 import org.openmetadata.schema.type.ChangeDescription;
 import org.openmetadata.schema.type.FieldChange;
 import org.openmetadata.schema.utils.JsonUtils;
 
 /**
- * Builds and merges structured change-preview JSON stored in {@code thread.message} for approval
- * tasks. The message format is a JSON object mapping field names to {@code {added, removed}} arrays
- * of human-readable identifiers (tagFQN, fullyQualifiedName, displayName, or name).
+ * Builds and merges structured change-preview data stored under the {@code proposedChanges} key in
+ * a task payload. The change map is a {@code Map<String, FieldDiff>} of field name to {@code
+ * {added, removed}} arrays of human-readable identifiers (tagFQN, fullyQualifiedName, displayName,
+ * or name).
  */
 @Slf4j
 public final class ChangePreviewUtils {
 
+  public static final String PROPOSED_CHANGES_KEY = "proposedChanges";
+
   private static final List<String> ID_KEYS =
       List.of("tagFQN", "fullyQualifiedName", "displayName", "name");
 
@@ -112,25 +114,30 @@ private static List<String> minus(List<String> source, Collection<String> exclus
   public static Map<String, FieldDiff> buildChangeMap(ChangeDescription changeDescription) {
     Map<String, FieldDiff> result = new LinkedHashMap<>();
     for (FieldChange fieldChange : listOrEmpty(changeDescription.getFieldsAdded())) {
-      result.put(
-          fieldChange.getName(),
-          new FieldDiff(extractIdentifiers(fieldChange.getNewValue()), List.of()));
+      accumulate(
+          result, fieldChange.getName(), extractIdentifiers(fieldChange.getNewValue()), List.of());
     }
     for (FieldChange fieldChange : listOrEmpty(changeDescription.getFieldsDeleted())) {
-      result.put(
-          fieldChange.getName(),
-          new FieldDiff(List.of(), extractIdentifiers(fieldChange.getOldValue())));
+      accumulate(
+          result, fieldChange.getName(), List.of(), extractIdentifiers(fieldChange.getOldValue()));
     }
     for (FieldChange fieldChange : listOrEmpty(changeDescription.getFieldsUpdated())) {
-      result.put(
+      accumulate(
+          result,
           fieldChange.getName(),
-          new FieldDiff(
-              extractIdentifiers(fieldChange.getNewValue()),
-              extractIdentifiers(fieldChange.getOldValue())));
+          extractIdentifiers(fieldChange.getNewValue()),
+          extractIdentifiers(fieldChange.getOldValue()));
     }
     return result;
   }
 
+  private static void accumulate(
+      Map<String, FieldDiff> result, String field, List<String> added, List<String> removed) {
+    FieldDiff incoming = new FieldDiff(added, removed);
+    FieldDiff prior = result.get(field);
+    result.put(field, prior == null ? incoming : prior.merge(incoming));
+  }
+
   public static Map<String, FieldDiff> mergeChangeMaps(
       Map<String, FieldDiff> oldMap, Map<String, FieldDiff> newMap) {
     Map<String, FieldDiff> merged = new LinkedHashMap<>(oldMap);
@@ -160,24 +167,132 @@ public static Map<String, FieldDiff> parseChangeMap(String message) {
     }
   }
 
-  public static void applyChangePreview(
-      Thread taskThread, EntityInterface entity, String oldMessage) {
-    taskThread.withCardStyle(null).withFieldOperation(null).withFeedInfo(null);
-    final ChangeDescription changeDescription = entity.getChangeDescription();
+  public static Map<String, FieldDiff> extractProposedChanges(Object payload) {
+    if (!(payload instanceof Map<?, ?> payloadMap)) return new LinkedHashMap<>();
+    Object existing = payloadMap.get(PROPOSED_CHANGES_KEY);
+    if (!(existing instanceof Map<?, ?> existingMap)) return new LinkedHashMap<>();
+    Map<String, FieldDiff> result = new LinkedHashMap<>();
+    for (Map.Entry<?, ?> entry : existingMap.entrySet()) {
+      String field = String.valueOf(entry.getKey());
+      FieldDiff diff = coerceFieldDiff(entry.getValue());
+      if (diff != null) {
+        result.put(field, diff);
+      }
+    }
+    return result;
+  }
+
+  /**
+   * Accepts either an in-memory {@link FieldDiff} record (freshly built and not yet round-tripped
+   * through JSON) or a raw {@code Map<String, List<String>>} read back from the persisted task
+   * payload, and returns a {@link FieldDiff}. Returns {@code null} when the value is neither
+   * shape.
+   */
+  private static FieldDiff coerceFieldDiff(Object value) {
+    if (value instanceof FieldDiff fd) return fd;
+    if (value instanceof Map<?, ?> diffMap) {
+      return new FieldDiff(
+          coerceStringList(diffMap.get("added")), coerceStringList(diffMap.get("removed")));
+    }
+    return null;
+  }
+
+  private static List<String> coerceStringList(Object value) {
+    if (!(value instanceof Collection<?> collection)) return List.of();
+    return collection.stream().filter(java.util.Objects::nonNull).map(String::valueOf).toList();
+  }
+
+  /**
+   * Build a new task payload that carries the merged proposed-changes map under {@link
+   * #PROPOSED_CHANGES_KEY}. Returns {@code existingPayload} unchanged when the entity has no
+   * change description; returns a payload with the {@code proposedChanges} key removed when the
+   * merged map is empty (e.g. all changes cancelled out across re-edits).
+   */
+  public static Object buildProposedChangesPayload(EntityInterface entity, Object existingPayload) {
+    if (entity == null) return existingPayload;
+    ChangeDescription changeDescription = pickIncrementalOrFull(entity);
     if (hasNoChanges(changeDescription)) {
-      taskThread.withMessage(oldMessage != null ? oldMessage : "{}");
-      return;
+      if (LOG.isDebugEnabled()) {
+        LOG.debug(
+            "[ChangePreview] entity='{}' v={} changeDescription is empty/null; carrying existing payload",
+            entity.getFullyQualifiedName(),
+            entity.getVersion());
+      }
+      return existingPayload;
     }
     try {
-      Map<String, FieldDiff> merged =
-          mergeChangeMaps(parseChangeMap(oldMessage), buildChangeMap(changeDescription));
-      taskThread.withMessage(merged.isEmpty() ? "{}" : JsonUtils.pojoToJson(merged));
+      Map<String, FieldDiff> priorMap = extractProposedChanges(existingPayload);
+      Map<String, FieldDiff> newMap = buildChangeMap(changeDescription);
+      Map<String, FieldDiff> merged = mergeChangeMaps(priorMap, newMap);
+      if (LOG.isDebugEnabled()) {
+        LOG.debug(
+            "[ChangePreview] entity='{}' v={} cdAdded={} cdDeleted={} cdUpdated={} priorPayload={} newDiff={} merged={}",
+            entity.getFullyQualifiedName(),
+            entity.getVersion(),
+            listOrEmpty(changeDescription.getFieldsAdded()),
+            listOrEmpty(changeDescription.getFieldsDeleted()),
+            listOrEmpty(changeDescription.getFieldsUpdated()),
+            priorMap,
+            newMap,
+            merged);
+      }
+      Map<String, Object> updated = cloneAsMutableMap(existingPayload);
+      if (merged.isEmpty()) {
+        updated.remove(PROPOSED_CHANGES_KEY);
+      } else {
+        updated.put(PROPOSED_CHANGES_KEY, merged);
+      }
+      return updated;
     } catch (Exception e) {
       LOG.warn(
-          "Failed to build change preview for approval task on {}",
+          "Failed to build proposed-changes payload for approval task on {}",
           entity.getFullyQualifiedName(),
           e);
-      taskThread.withMessage(oldMessage != null ? oldMessage : "{}");
+      return existingPayload;
+    }
+  }
+
+  /**
+   * Prefer {@code incrementalChangeDescription} (per-edit hop diff) over {@code changeDescription}
+   * (cumulative-between-versions diff). The cumulative form double-counts when prior task payload
+   * has already merged an intermediate state: e.g. a tag added in v0.3 then removed in v0.4 is
+   * already cancelled out in the prior task payload, but v0.4's cumulative changeDescription
+   * still reports the removal, which would re-introduce it on the {@code removed} side.
+   * Incremental change description always reflects just the latest patch, which is the right
+   * unit of work to fold into the running merge.
+   */
+  private static ChangeDescription pickIncrementalOrFull(EntityInterface entity) {
+    ChangeDescription incremental = entity.getIncrementalChangeDescription();
+    if (!hasNoChanges(incremental)) {
+      return incremental;
+    }
+    return entity.getChangeDescription();
+  }
+
+  /**
+   * When a workflow-driven update replaces the entire task payload via a {@code requestedPayload}
+   * variable, any accumulated {@code proposedChanges} on the prior payload would be lost. If the
+   * replacement payload does not carry its own {@code proposedChanges}, copy the key over from
+   * the prior payload so the running merge base is preserved. Returns {@code requestedPayload}
+   * unchanged when there is nothing to preserve.
+   */
+  public static Object preserveProposedChanges(Object requestedPayload, Object priorPayload) {
+    if (!(requestedPayload instanceof Map<?, ?> requestedMap)) return requestedPayload;
+    if (requestedMap.get(PROPOSED_CHANGES_KEY) != null) return requestedPayload;
+    if (!(priorPayload instanceof Map<?, ?> priorMap)) return requestedPayload;
+    Object priorProposed = priorMap.get(PROPOSED_CHANGES_KEY);
+    if (priorProposed == null) return requestedPayload;
+    Map<String, Object> merged = cloneAsMutableMap(requestedPayload);
+    merged.put(PROPOSED_CHANGES_KEY, priorProposed);
+    return merged;
+  }
+
+  private static Map<String, Object> cloneAsMutableMap(Object payload) {
+    if (!(payload instanceof Map<?, ?> source)) return new LinkedHashMap<>();
+    Map<String, Object> copy = new LinkedHashMap<>();
+    for (Map.Entry<?, ?> entry : source.entrySet()) {
+      copy.put(String.valueOf(entry.getKey()), entry.getValue());
     }
+    return copy;
   }
 }