diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java
index 058afcb3fc71..6cc850d6f141 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/enums/AlertType.java
@@ -29,7 +29,7 @@ public enum AlertType {
 
     /**
      * 0 workflow instance failure, 1 workflow instance success, 2 workflow instance blocked, 3 workflow instance timeout, 4 fault tolerance warning,
-     * 5 task failure, 6 task success, 7 task timeout
+     * 5 task failure, 6 task success, 7 task timeout, 8 task result
       */
     WORKFLOW_INSTANCE_FAILURE(0, "workflow instance failure"),
     WORKFLOW_INSTANCE_SUCCESS(1, "workflow instance success"),
@@ -39,6 +39,7 @@ public enum AlertType {
     TASK_FAILURE(5, "task failure"),
     TASK_SUCCESS(6, "task success"),
     TASK_TIMEOUT(7, "task timeout"),
+    TASK_RESULT(8, "task result"),
     ;
 
     AlertType(int code, String descp) {
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
index 19ed2d70f2df..4cc2f2c891a7 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/AlertDao.java
@@ -108,6 +108,33 @@ public int addAlert(Alert alert) {
         return count;
     }
 
+    /**
+     * Insert a task-result alert idempotently. If an alert with the same sign,
+     * workflow instance id and alert type already exists, the insert is skipped.
+     * <p>This guards against duplicate inserts caused by at-least-once delivery
+     * of task success lifecycle events.
+     *
+     * @param alert alert, must have sign, workflowInstanceId and alertType set
+     * @return insert count (1 if inserted, 0 if skipped)
+     */
+    public int addTaskResultAlert(Alert alert) {
+        if (null == alert.getAlertGroupId() || NumberUtils.INTEGER_ZERO.equals(alert.getAlertGroupId())) {
+            log.warn("the value of alertGroupId is null or 0 ");
+            return 0;
+        }
+
+        String sign = generateSign(alert);
+        alert.setSign(sign);
+        int count = alertMapper.insertTaskResultAlertIfAbsent(alert);
+        if (count > 0) {
+            log.info("add task result alert to db , alert: {}", alert);
+        } else {
+            log.info("skip duplicate task result alert, sign: {}, workflowInstanceId: {}", sign,
+                    alert.getWorkflowInstanceId());
+        }
+        return count;
+    }
+
     /**
      * update alert sending(execution) status
      *
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java
index 0d60891e55f5..dcfdf1ec0e86 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/AlertMapper.java
@@ -43,6 +43,12 @@ List<Alert> listingAlertByStatus(@Param("minAlertId") int minAlertId, @Param("al
     void insertAlertWhenServerCrash(@Param("alert") Alert alert,
                                     @Param("crashAlarmSuppressionStartTime") Date crashAlarmSuppressionStartTime);
 
+    /**
+     * Insert a task-result alert only if no alert with the same sign, workflow instance id and alert type
+     * already exists. This makes the insert idempotent against at-least-once event delivery.
+     */
+    int insertTaskResultAlertIfAbsent(@Param("alert") Alert alert);
+
     void deleteByWorkflowInstanceId(@Param("workflowInstanceId") Integer processInstanceId);
 
     List<Alert> selectByWorkflowInstanceId(@Param("workflowInstanceId") Integer processInstanceId);
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml
index 7891dd91376d..df6a3a783556 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/AlertMapper.xml
@@ -51,6 +51,34 @@
         having count(*) = 0
     </insert>
 
+    <!--
+        Idempotent insert for task-result alerts.  Only inserts a row when no alert
+        with the same sign, workflow_instance_id and alert_type already exists.
+        This guards against duplicate inserts caused by at-least-once event delivery.
+    -->
+    <insert id="insertTaskResultAlertIfAbsent">
+        insert into t_ds_alert(sign, title, content, alert_status, warning_type, log, alertgroup_id, create_time,
+                               update_time, project_code, workflow_definition_code, workflow_instance_id, alert_type)
+        SELECT #{alert.sign},
+               #{alert.title},
+               #{alert.content},
+               #{alert.alertStatus.code},
+               #{alert.warningType.code},
+               #{alert.log},
+               #{alert.alertGroupId},
+               #{alert.createTime},
+               #{alert.updateTime},
+               #{alert.projectCode},
+               #{alert.workflowDefinitionCode},
+               #{alert.workflowInstanceId},
+               #{alert.alertType.code}
+        from t_ds_alert
+        where sign = #{alert.sign}
+          and workflow_instance_id = #{alert.workflowInstanceId}
+          and alert_type = #{alert.alertType.code}
+        having count(*) = 0
+    </insert>
+
     <select id="listingAlertByStatus" resultType="org.apache.dolphinscheduler.dao.entity.Alert">
         select
         <include refid="baseSql"/>
diff --git a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/AlertDaoTest.java b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/AlertDaoTest.java
index b284a4cd0967..9cf3bb2bb1fc 100644
--- a/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/AlertDaoTest.java
+++ b/dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/repository/impl/AlertDaoTest.java
@@ -18,6 +18,8 @@
 package org.apache.dolphinscheduler.dao.repository.impl;
 
 import org.apache.dolphinscheduler.common.enums.AlertStatus;
+import org.apache.dolphinscheduler.common.enums.AlertType;
+import org.apache.dolphinscheduler.common.enums.WarningType;
 import org.apache.dolphinscheduler.dao.AlertDao;
 import org.apache.dolphinscheduler.dao.BaseDaoTest;
 import org.apache.dolphinscheduler.dao.entity.Alert;
@@ -66,4 +68,45 @@ void testSendServerStoppedAlert() {
                 .count();
         Assertions.assertEquals(1L, count);
     }
+
+    @Test
+    void testAddTaskResultAlertIdempotent() {
+        String content = "[{\"taskName\":\"sql-task-1\",\"result\":\"ok\"}]";
+        int workflowInstanceId = 999999;
+
+        Alert alert = new Alert();
+        alert.setTitle("SQL Task Result");
+        alert.setContent(content);
+        alert.setWarningType(WarningType.SUCCESS);
+        alert.setAlertGroupId(1);
+        alert.setAlertStatus(AlertStatus.WAIT_EXECUTION);
+        alert.setWorkflowInstanceId(workflowInstanceId);
+        alert.setAlertType(AlertType.TASK_RESULT);
+        alert.setCreateTime(new java.util.Date());
+
+        // First insert should succeed
+        int firstCount = alertDao.addTaskResultAlert(alert);
+        Assertions.assertEquals(1, firstCount);
+
+        // Second insert with the same content + workflowInstanceId + alertType should be skipped
+        Alert duplicateAlert = new Alert();
+        duplicateAlert.setTitle("SQL Task Result");
+        duplicateAlert.setContent(content);
+        duplicateAlert.setWarningType(WarningType.SUCCESS);
+        duplicateAlert.setAlertGroupId(1);
+        duplicateAlert.setAlertStatus(AlertStatus.WAIT_EXECUTION);
+        duplicateAlert.setWorkflowInstanceId(workflowInstanceId);
+        duplicateAlert.setAlertType(AlertType.TASK_RESULT);
+        duplicateAlert.setCreateTime(new java.util.Date());
+
+        int secondCount = alertDao.addTaskResultAlert(duplicateAlert);
+        Assertions.assertEquals(0, secondCount);
+
+        // Verify only one alert row exists for this workflow instance
+        long count = alertDao.listAlerts(workflowInstanceId)
+                .stream()
+                .filter(a -> a.getAlertType() == AlertType.TASK_RESULT)
+                .count();
+        Assertions.assertEquals(1L, count);
+    }
 }
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
index 9c7f41e1b527..4568768bcaa0 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/event/TaskSuccessLifecycleEvent.java
@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event;
 
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.AbstractTaskLifecycleEvent;
@@ -41,6 +42,10 @@ public class TaskSuccessLifecycleEvent extends AbstractTaskLifecycleEvent {
 
     private final List<Property> varPool;
 
+    private final boolean needAlert;
+
+    private final TaskAlertInfo taskAlertInfo;
+
     @Override
     public ILifecycleEventType getEventType() {
         return TaskLifecycleEventType.SUCCEEDED;
@@ -52,6 +57,7 @@ public String toString() {
                 "task=" + taskExecution.getName() +
                 ", endTime=" + endTime +
                 ", varPool='" + varPool + '\'' +
+                ", needAlert=" + needAlert +
                 '}';
     }
 }
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java
index 86a6ec179927..b75aec89b333 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/lifecycle/handler/TaskSuccessLifecycleEventHandler.java
@@ -17,6 +17,7 @@
 
 package org.apache.dolphinscheduler.server.master.engine.task.lifecycle.handler;
 
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.server.master.engine.ILifecycleEventType;
 import org.apache.dolphinscheduler.server.master.engine.task.client.TaskExecutorClient;
 import org.apache.dolphinscheduler.server.master.engine.task.execution.ITaskExecution;
@@ -24,18 +25,26 @@
 import org.apache.dolphinscheduler.server.master.engine.task.lifecycle.event.TaskSuccessLifecycleEvent;
 import org.apache.dolphinscheduler.server.master.engine.task.statemachine.ITaskStateAction;
 import org.apache.dolphinscheduler.server.master.engine.workflow.execution.IWorkflowExecution;
+import org.apache.dolphinscheduler.service.alert.WorkflowAlertManager;
 import org.apache.dolphinscheduler.task.executor.eventbus.ITaskExecutorLifecycleEventReporter;
 import org.apache.dolphinscheduler.task.executor.events.TaskExecutorLifecycleEventType;
 
+import lombok.extern.slf4j.Slf4j;
+
 import org.springframework.stereotype.Component;
 
+@Slf4j
 @Component
 public class TaskSuccessLifecycleEventHandler extends AbstractTaskLifecycleEventHandler<TaskSuccessLifecycleEvent> {
 
     private final TaskExecutorClient taskExecutorClient;
 
-    public TaskSuccessLifecycleEventHandler(final TaskExecutorClient taskExecutorClient) {
+    private final WorkflowAlertManager workflowAlertManager;
+
+    public TaskSuccessLifecycleEventHandler(final TaskExecutorClient taskExecutorClient,
+                                            final WorkflowAlertManager workflowAlertManager) {
         this.taskExecutorClient = taskExecutorClient;
+        this.workflowAlertManager = workflowAlertManager;
     }
 
     @Override
@@ -43,7 +52,25 @@ public void handle(final ITaskStateAction taskStateAction,
                        final IWorkflowExecution workflowExecution,
                        final ITaskExecution taskExecution,
                        final TaskSuccessLifecycleEvent taskSuccessEvent) {
+        // 1. State transition + DB persistence (may throw if state mismatch)
         taskStateAction.onSucceedEvent(workflowExecution, taskExecution, taskSuccessEvent);
+
+        // 2. Persist task-result alert only after the success state transition is confirmed
+        if (taskSuccessEvent.isNeedAlert()) {
+            final TaskAlertInfo taskAlertInfo = taskSuccessEvent.getTaskAlertInfo();
+            if (taskAlertInfo != null && taskAlertInfo.getAlertGroupId() != null
+                    && taskAlertInfo.getAlertGroupId() > 0) {
+                workflowAlertManager.sendTaskResultAlert(
+                        taskExecution.getWorkflowInstance(),
+                        taskExecution.getTaskInstance(),
+                        taskAlertInfo);
+            } else {
+                log.warn("Task: {} need alert but alertGroupId is invalid, skip sending alert",
+                        taskExecution.getName());
+            }
+        }
+
+        // 3. ACK the worker — only after state transition and alert persistence are done
         taskExecutorClient.ackTaskExecutorLifecycleEvent(
                 taskExecution,
                 new ITaskExecutorLifecycleEventReporter.TaskExecutorLifecycleEventAck(
diff --git a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java
index 278f5160dc80..0566c839858b 100644
--- a/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java
+++ b/dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/rpc/TaskExecutorEventListenerImpl.java
@@ -116,6 +116,8 @@ public void onTaskExecutorSuccess(final TaskExecutorSuccessLifecycleEvent taskEx
                     .taskExecution(taskExecution)
                     .endTime(new Date(taskExecutorSuccessLifecycleEvent.getEndTime()))
                     .varPool(taskExecutorSuccessLifecycleEvent.getVarPool())
+                    .needAlert(taskExecutorSuccessLifecycleEvent.isNeedAlert())
+                    .taskAlertInfo(taskExecutorSuccessLifecycleEvent.getTaskAlertInfo())
                     .build();
             taskExecution.getWorkflowEventBus().publish(taskSuccessEvent);
         } finally {
diff --git a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java
index 9335e9551eb7..906955859fcc 100644
--- a/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java
+++ b/dolphinscheduler-service/src/main/java/org/apache/dolphinscheduler/service/alert/WorkflowAlertManager.java
@@ -34,6 +34,7 @@
 import org.apache.dolphinscheduler.dao.repository.ProjectDao;
 import org.apache.dolphinscheduler.dao.repository.UserDao;
 import org.apache.dolphinscheduler.dao.repository.WorkflowDefinitionLogDao;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 
 import java.util.ArrayList;
 import java.util.Date;
@@ -208,4 +209,34 @@ public void sendWorkflowTimeoutAlert(WorkflowInstance workflowInstance) {
         alertDao.sendWorkflowTimeoutAlert(workflowInstance, projectUser);
     }
 
+    /**
+     * send task result alert
+     *
+     * @param workflowInstance workflow instance
+     * @param taskInstance     task instance
+     * @param taskAlertInfo    task alert info
+     */
+    public void sendTaskResultAlert(WorkflowInstance workflowInstance,
+                                    TaskInstance taskInstance,
+                                    TaskAlertInfo taskAlertInfo) {
+        if (taskAlertInfo == null || taskAlertInfo.getAlertGroupId() == null) {
+            return;
+        }
+
+        Alert alert = new Alert();
+        alert.setTitle(taskAlertInfo.getTitle());
+        alert.setContent(taskAlertInfo.getContent());
+        alert.setWarningType(WarningType.SUCCESS);
+        alert.setCreateTime(new Date());
+        alert.setAlertGroupId(taskAlertInfo.getAlertGroupId());
+        alert.setProjectCode(workflowInstance.getProjectCode());
+        alert.setWorkflowDefinitionCode(workflowInstance.getWorkflowDefinitionCode());
+        alert.setWorkflowInstanceId(workflowInstance.getId());
+        alert.setAlertType(taskAlertInfo.getAlertType() != null
+                ? taskAlertInfo.getAlertType()
+                : AlertType.TASK_RESULT);
+        alertDao.addTaskResultAlert(alert);
+        log.info("Send task result alert for task: {} in workflow: {}",
+                taskInstance.getName(), workflowInstance.getName());
+    }
 }
diff --git a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
index 947d1edf3633..2c55c80c365c 100644
--- a/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
+++ b/dolphinscheduler-task-executor/src/main/java/org/apache/dolphinscheduler/task/executor/events/TaskExecutorSuccessLifecycleEvent.java
@@ -19,6 +19,7 @@
 
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
 
 import java.util.List;
@@ -48,6 +49,10 @@ public class TaskExecutorSuccessLifecycleEvent extends AbstractTaskExecutorLifec
 
     private Long latestReportTime;
 
+    private boolean needAlert;
+
+    private TaskAlertInfo taskAlertInfo;
+
     public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecutor) {
         final TaskExecutionContext taskExecutionContext = taskExecutor.getTaskExecutionContext();
         return TaskExecutorSuccessLifecycleEvent.builder()
@@ -57,6 +62,8 @@ public static TaskExecutorSuccessLifecycleEvent of(final ITaskExecutor taskExecu
                 .varPool(taskExecutionContext.getVarPool())
                 .type(TaskExecutorLifecycleEventType.SUCCESS)
                 .endTime(taskExecutionContext.getEndTime())
+                .needAlert(taskExecutionContext.isNeedAlert())
+                .taskAlertInfo(taskExecutionContext.getTaskAlertInfo())
                 .build();
     }
 
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
index 0c8f208fcbae..530205c4556d 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/AbstractTask.java
@@ -19,7 +19,6 @@
 
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
-import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters;
 
 import java.util.Map;
@@ -56,10 +55,6 @@ public abstract class AbstractTask {
      */
     protected volatile int exitStatusCode = -1;
 
-    protected boolean needAlert = false;
-
-    protected TaskAlertInfo taskAlertInfo;
-
     /**
      * constructor
      *
@@ -109,22 +104,6 @@ public void setAppIds(String appIds) {
         this.appIds = appIds;
     }
 
-    public boolean getNeedAlert() {
-        return needAlert;
-    }
-
-    public void setNeedAlert(boolean needAlert) {
-        this.needAlert = needAlert;
-    }
-
-    public TaskAlertInfo getTaskAlertInfo() {
-        return taskAlertInfo;
-    }
-
-    public void setTaskAlertInfo(TaskAlertInfo taskAlertInfo) {
-        this.taskAlertInfo = taskAlertInfo;
-    }
-
     /**
      * get task parameters
      *
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
index 85d6f682196b..a072258011df 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java
@@ -19,6 +19,7 @@
 
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
 import org.apache.dolphinscheduler.plugin.task.api.resource.ResourceContext;
 
@@ -130,6 +131,10 @@ public class TaskExecutionContext implements Serializable {
 
     private final long firstDispatchTime = System.currentTimeMillis();
 
+    private boolean needAlert;
+
+    private TaskAlertInfo taskAlertInfo;
+
     public int increaseDispatchFailTimes() {
         return ++dispatchFailTimes;
     }
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java
index e72242e93473..d2c22e565a79 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/model/TaskAlertInfo.java
@@ -17,6 +17,8 @@
 
 package org.apache.dolphinscheduler.plugin.task.api.model;
 
+import org.apache.dolphinscheduler.common.enums.AlertType;
+
 import lombok.Data;
 
 @Data
@@ -27,4 +29,7 @@ public class TaskAlertInfo {
     private String content;
 
     private Integer alertGroupId;
+
+    private AlertType alertType;
+
 }
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
index 113dfbcd9eb8..a3516d1e9ade 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/main/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTask.java
@@ -17,6 +17,7 @@
 
 package org.apache.dolphinscheduler.plugin.task.sql;
 
+import org.apache.dolphinscheduler.common.enums.AlertType;
 import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.plugin.datasource.api.plugin.DataSourceClientProvider;
@@ -287,13 +288,21 @@ private String resultProcess(ResultSet resultSet) throws Exception {
             }
         }
 
-        String result = resultJSONArray.isEmpty() ? JSONUtils.toJsonString(generateEmptyRow(resultSet))
-                : JSONUtils.toJsonString(resultJSONArray);
+        String result;
+        ArrayNode alertArray = resultJSONArray;
+        if (resultJSONArray.isEmpty()) {
+            ArrayNode emptyRow = generateEmptyRow(resultSet);
+            result = JSONUtils.toJsonString(emptyRow);
+            alertArray = emptyRow;
+        } else {
+            result = JSONUtils.toJsonString(resultJSONArray);
+        }
 
         if (Boolean.TRUE.equals(sqlParameters.getSendEmail())) {
-            sendAttachment(sqlParameters.getGroupId(), StringUtils.isNotEmpty(sqlParameters.getTitle())
-                    ? sqlParameters.getTitle()
-                    : taskExecutionContext.getTaskName() + " query result sets", result);
+            log.info("SendAlert is enabled, preparing task result alert");
+            prepareTaskResultAlert(alertArray);
+        } else {
+            log.info("SendAlert is not enabled, skip task result alert");
         }
         log.debug("execute sql result : {}", result);
         return result;
@@ -320,18 +329,33 @@ private ArrayNode generateEmptyRow(ResultSet resultSet) throws SQLException {
     }
 
     /**
-     * send alert as an attachment
+     * Prepare task result alert info.
+     * Truncate the alert content to displayRows to avoid oversized RPC payload.
      *
-     * @param title   title
-     * @param content content
+     * @param resultJSONArray the full query result JSON array
      */
-    private void sendAttachment(int groupId, String title, String content) {
-        setNeedAlert(Boolean.TRUE);
+    private void prepareTaskResultAlert(ArrayNode resultJSONArray) {
         TaskAlertInfo taskAlertInfo = new TaskAlertInfo();
-        taskAlertInfo.setAlertGroupId(groupId);
-        taskAlertInfo.setContent(content);
-        taskAlertInfo.setTitle(title);
-        setTaskAlertInfo(taskAlertInfo);
+        taskAlertInfo.setAlertGroupId(sqlParameters.getGroupId());
+        taskAlertInfo.setTitle(StringUtils.isNotEmpty(sqlParameters.getTitle())
+                ? sqlParameters.getTitle()
+                : taskExecutionContext.getTaskName() + " query result sets");
+        // Truncate content to displayRows to avoid oversized RPC payload
+        int alertRows = sqlParameters.getDisplayRows() > 0 ? sqlParameters.getDisplayRows()
+                : TaskConstants.DEFAULT_DISPLAY_ROWS;
+        alertRows = Math.min(alertRows, resultJSONArray.size());
+        ArrayNode alertContent = JSONUtils.createArrayNode();
+        for (int i = 0; i < alertRows; i++) {
+            alertContent.add(resultJSONArray.get(i));
+        }
+        taskAlertInfo.setContent(JSONUtils.toJsonString(alertContent));
+        taskAlertInfo.setAlertType(AlertType.TASK_RESULT);
+
+        taskExecutionContext.setNeedAlert(true);
+        taskExecutionContext.setTaskAlertInfo(taskAlertInfo);
+        log.info("Prepare task result alert: title={}, alertGroupId={}, alertType={}, totalRows={}, alertRows={}",
+                taskAlertInfo.getTitle(), taskAlertInfo.getAlertGroupId(), taskAlertInfo.getAlertType(),
+                resultJSONArray.size(), alertRows);
     }
 
     private String executeQuery(Connection connection, SqlBinds sqlBinds, String handlerType) throws Exception {
diff --git a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java
index 92f0fa0702d2..ce2b7945f7fa 100644
--- a/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java
+++ b/dolphinscheduler-task-plugin/dolphinscheduler-task-sql/src/test/java/org/apache/dolphinscheduler/plugin/task/sql/SqlTaskTest.java
@@ -20,6 +20,7 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import org.apache.dolphinscheduler.common.enums.AlertType;
 import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
@@ -29,6 +30,7 @@
 import org.apache.dolphinscheduler.plugin.task.api.enums.Direct;
 import org.apache.dolphinscheduler.plugin.task.api.enums.ResourceType;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.plugin.task.api.model.TaskAlertInfo;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.SqlParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.DataSourceParameters;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
@@ -489,4 +491,240 @@ private ResourceParametersHelper getResourceParametersHelperWithDatasourceType(D
         return resourceParametersHelper;
     }
 
+    /**
+     * Helper: create a SqlTask with sendEmail enabled, invoke prepareTaskResultAlert via reflection,
+     * and return the TaskExecutionContext for assertions.
+     */
+    private TaskExecutionContext createSqlTaskWithAlert(ArrayNode resultArray, int displayRows, String title,
+                                                        int groupId) {
+        String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\""
+                + ",\"sendEmail\":true"
+                + ",\"displayRows\":" + displayRows
+                + ",\"groupId\":" + groupId
+                + (title != null ? ",\"title\":\"" + title + "\"" : "")
+                + "}";
+
+        TaskExecutionContext ctx = new TaskExecutionContext();
+        ctx.setTaskParams(taskParams);
+        ctx.setTaskName("test_sql_task");
+        ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE));
+
+        SqlTask task = new SqlTask(ctx);
+        // Call prepareTaskResultAlert via reflection
+        try {
+            Method method = SqlTask.class.getDeclaredMethod("prepareTaskResultAlert", ArrayNode.class);
+            method.setAccessible(true);
+            method.invoke(task, resultArray);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+        return ctx;
+    }
+
+    /**
+     * When sendEmail is true, prepareTaskResultAlert should set needAlert on TaskExecutionContext
+     * and populate TaskAlertInfo with correct alertGroupId, title, alertType and truncated content.
+     */
+    @Test
+    void testPrepareTaskResultAlert_setsNeedAlertAndTaskAlertInfo() {
+        ArrayNode resultArray = JSONUtils.createArrayNode();
+        resultArray.add(JSONUtils.parseObject("{\"id\":\"1\",\"name\":\"alice\"}"));
+        resultArray.add(JSONUtils.parseObject("{\"id\":\"2\",\"name\":\"bob\"}"));
+
+        TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "My Alert Title", 5);
+
+        Assertions.assertTrue(ctx.isNeedAlert(), "needAlert should be true when sendEmail is enabled");
+
+        TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+        Assertions.assertNotNull(alertInfo, "taskAlertInfo should not be null");
+        Assertions.assertEquals(5, alertInfo.getAlertGroupId());
+        Assertions.assertEquals("My Alert Title", alertInfo.getTitle());
+        Assertions.assertEquals(AlertType.TASK_RESULT, alertInfo.getAlertType());
+        Assertions.assertNotNull(alertInfo.getContent());
+
+        // Content should contain all rows (within displayRows limit)
+        ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+        Assertions.assertEquals(2, contentArray.size());
+    }
+
+    /**
+     * Alert content should be truncated to displayRows to avoid oversized RPC payload.
+     */
+    @Test
+    void testPrepareTaskResultAlert_truncatesToDisplayRows() {
+        ArrayNode resultArray = JSONUtils.createArrayNode();
+        for (int i = 0; i < 50; i++) {
+            resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}"));
+        }
+
+        // displayRows = 3, but result has 50 rows -> content should be truncated to 3
+        TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 3, null, 1);
+
+        TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+        Assertions.assertNotNull(alertInfo);
+
+        ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+        Assertions.assertEquals(3, contentArray.size(), "Alert content should be truncated to displayRows");
+        Assertions.assertEquals("0", contentArray.get(0).get("id").asText());
+        Assertions.assertEquals("1", contentArray.get(1).get("id").asText());
+        Assertions.assertEquals("2", contentArray.get(2).get("id").asText());
+    }
+
+    /**
+     * When displayRows is 0 (unset), should default to TaskConstants.DEFAULT_DISPLAY_ROWS (10).
+     */
+    @Test
+    void testPrepareTaskResultAlert_usesDefaultDisplayRowsWhenUnset() {
+        ArrayNode resultArray = JSONUtils.createArrayNode();
+        for (int i = 0; i < 20; i++) {
+            resultArray.add(JSONUtils.parseObject("{\"id\":\"" + i + "\"}"));
+        }
+
+        // displayRows = 0 means unset -> should default to DEFAULT_DISPLAY_ROWS (10)
+        TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 0, "default rows test", 2);
+
+        TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+        Assertions.assertNotNull(alertInfo);
+
+        ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+        Assertions.assertEquals(TaskConstants.DEFAULT_DISPLAY_ROWS, contentArray.size(),
+                "Should default to DEFAULT_DISPLAY_ROWS when displayRows is 0");
+    }
+
+    /**
+     * When title is provided, it should be used as the alert title.
+     */
+    @Test
+    void testPrepareTaskResultAlert_usesCustomTitle() {
+        ArrayNode resultArray = JSONUtils.createArrayNode();
+        resultArray.add(JSONUtils.parseObject("{\"id\":\"1\"}"));
+
+        TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, "Custom Title", 1);
+
+        TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+        Assertions.assertNotNull(alertInfo);
+        Assertions.assertEquals("Custom Title", alertInfo.getTitle());
+    }
+
+    /**
+     * When title is empty, should use taskName + " query result sets" as default.
+     */
+    @Test
+    void testPrepareTaskResultAlert_usesDefaultTitleWhenEmpty() {
+        ArrayNode resultArray = JSONUtils.createArrayNode();
+        resultArray.add(JSONUtils.parseObject("{\"id\":\"1\"}"));
+
+        TaskExecutionContext ctx = createSqlTaskWithAlert(resultArray, 10, null, 1);
+
+        TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+        Assertions.assertNotNull(alertInfo);
+        Assertions.assertEquals("test_sql_task query result sets", alertInfo.getTitle());
+    }
+
+    /**
+     * When sendEmail is false, resultProcess should NOT set needAlert or taskAlertInfo.
+     */
+    @Test
+    void testResultProcess_sendEmailDisabled_doesNotSetNeedAlert() throws Exception {
+        // Build a SqlTask with sendEmail = false
+        String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\",\"sendEmail\":false}";
+
+        TaskExecutionContext ctx = new TaskExecutionContext();
+        ctx.setTaskParams(taskParams);
+        ctx.setTaskName("no_alert_task");
+        ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE));
+
+        SqlTask task = new SqlTask(ctx);
+
+        ResultSet mockResultSet = mock(ResultSet.class);
+        ResultSetMetaData mockMetaData = mock(ResultSetMetaData.class);
+        when(mockResultSet.getMetaData()).thenReturn(mockMetaData);
+        when(mockMetaData.getColumnCount()).thenReturn(1);
+        when(mockMetaData.getColumnLabel(1)).thenReturn("id");
+        when(mockResultSet.next()).thenReturn(true, false);
+        when(mockResultSet.getObject(1)).thenReturn("1");
+
+        Method resultProcessMethod = SqlTask.class.getDeclaredMethod("resultProcess", ResultSet.class);
+        resultProcessMethod.setAccessible(true);
+        resultProcessMethod.invoke(task, mockResultSet);
+
+        Assertions.assertFalse(ctx.isNeedAlert(), "needAlert should remain false when sendEmail is disabled");
+        Assertions.assertNull(ctx.getTaskAlertInfo(), "taskAlertInfo should remain null when sendEmail is disabled");
+    }
+
+    /**
+     * When the query result is empty, resultProcess should still prepare the alert
+     * using the generated empty row so the user is notified that the query returned no data.
+     */
+    @Test
+    void testResultProcess_emptyResultSet_prepareAlertWithEmptyRow() throws Exception {
+        String taskParams = "{\"type\":\"HIVE\",\"datasource\":1,\"sql\":\"select 1\""
+                + ",\"sendEmail\":true,\"displayRows\":10,\"groupId\":3,\"title\":\"empty result\"}";
+
+        TaskExecutionContext ctx = new TaskExecutionContext();
+        ctx.setTaskParams(taskParams);
+        ctx.setTaskName("empty_result_task");
+        ctx.setResourceParametersHelper(getResourceParametersHelperWithDatasourceType(DbType.HIVE));
+
+        SqlTask task = new SqlTask(ctx);
+
+        ResultSet mockResultSet = mock(ResultSet.class);
+        ResultSetMetaData mockMetaData = mock(ResultSetMetaData.class);
+        when(mockResultSet.getMetaData()).thenReturn(mockMetaData);
+        when(mockMetaData.getColumnCount()).thenReturn(2);
+        when(mockMetaData.getColumnLabel(1)).thenReturn("id");
+        when(mockMetaData.getColumnLabel(2)).thenReturn("name");
+        when(mockResultSet.next()).thenReturn(false); // empty result set
+
+        Method resultProcessMethod = SqlTask.class.getDeclaredMethod("resultProcess", ResultSet.class);
+        resultProcessMethod.setAccessible(true);
+        resultProcessMethod.invoke(task, mockResultSet);
+
+        Assertions.assertTrue(ctx.isNeedAlert(), "needAlert should be true even for empty result set");
+
+        TaskAlertInfo alertInfo = ctx.getTaskAlertInfo();
+        Assertions.assertNotNull(alertInfo);
+        Assertions.assertEquals(3, alertInfo.getAlertGroupId());
+        Assertions.assertEquals("empty result", alertInfo.getTitle());
+        Assertions.assertEquals(AlertType.TASK_RESULT, alertInfo.getAlertType());
+
+        // Content should be the empty row
+        ArrayNode contentArray = JSONUtils.parseArray(alertInfo.getContent());
+        Assertions.assertEquals(1, contentArray.size(), "Empty result alert should contain one empty row");
+        Assertions.assertEquals("", contentArray.get(0).get("id").asText());
+        Assertions.assertEquals("", contentArray.get(0).get("name").asText());
+    }
+
+    /**
+     * Verify that TaskExecutorSuccessLifecycleEvent.of() carries needAlert and taskAlertInfo
+     * from TaskExecutionContext to the event, ensuring the alert info survives the Worker→Master RPC.
+     * Since task-executor module has no test infrastructure, we validate indirectly by confirming
+     * that TaskExecutionContext properly stores and exposes these fields for the event builder to read.
+     */
+    @Test
+    void testTaskExecutionContext_carriesAlertInfoForEventPropagation() {
+        // Simulate what SqlTask.prepareTaskResultAlert does
+        TaskExecutionContext ctx = new TaskExecutionContext();
+        ctx.setNeedAlert(true);
+
+        TaskAlertInfo alertInfo = new TaskAlertInfo();
+        alertInfo.setTitle("test title");
+        alertInfo.setContent("[{\"id\":\"1\"}]");
+        alertInfo.setAlertGroupId(7);
+        alertInfo.setAlertType(AlertType.TASK_RESULT);
+        ctx.setTaskAlertInfo(alertInfo);
+
+        // Verify the fields are retrievable — this is what TaskExecutorSuccessLifecycleEvent.of() reads
+        Assertions.assertTrue(ctx.isNeedAlert());
+        Assertions.assertNotNull(ctx.getTaskAlertInfo());
+        Assertions.assertEquals("test title", ctx.getTaskAlertInfo().getTitle());
+        Assertions.assertEquals(7, ctx.getTaskAlertInfo().getAlertGroupId());
+        Assertions.assertEquals(AlertType.TASK_RESULT, ctx.getTaskAlertInfo().getAlertType());
+
+        // Verify default state when alert is not set
+        TaskExecutionContext ctx2 = new TaskExecutionContext();
+        Assertions.assertFalse(ctx2.isNeedAlert());
+        Assertions.assertNull(ctx2.getTaskAlertInfo());
+    }
+
 }