feat(flow): 删除旧的流程通知表和合并表的迁移脚本

- 删除了 `20261014_flow_first_batch.sql` 和 `20261014_flow_merge_tables.sql` 文件
- 更新 `FlowNoticeStatusEnum`，新增 REMOVING 状态
- 修改 `FlowNotice` 实体，确保在插入时保留 userId 值
- 在 `FlowDurableConfiguration` 中添加事务提交后撤销失效通知的逻辑
- 更新 `IFlowNoticeService` 接口，调整通知事件的处理逻辑
- 实现 `FlowNoticeServiceImpl` 中的通知撤销逻辑，确保不误删新通知
- 新增 `FlowMessageRemovalDTO` 数据传输对象，处理通知撤销请求
- 更新 `SysMessageService` 和 `SysMessageController`，支持内部调用撤销流程通知
- 添加 `SysMessageDeletionTests` 测试用例，验证撤销逻辑的正确性

db/0pigxx_boot.sql

@@ -1568,7 +1568,8 @@ CREATE TABLE `sys_message_relation` (
   `update_by` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '更新人',
   `del_flag` char(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '删除时间',
   `tenant_id` bigint DEFAULT NULL COMMENT '租户',
-  PRIMARY KEY (`id`)
+  PRIMARY KEY (`id`),
+  KEY `idx_relation_msg_user` (`msg_id`,`user_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='系统消息推送记录';
 
 
@@ -1852,7 +1853,7 @@ CREATE TABLE `flow_notice` (
   `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID',
   `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID',
   `created_at` timestamp(3) NOT NULL COMMENT '创建时间',
-  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，CANCELLED 已取消，DISABLED 已停用',
+  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，REMOVING 待撤销，CANCELLED 已取消，DISABLED 已停用',
   `scheduled_at` timestamp(3) NULL DEFAULT NULL COMMENT '计划投递或到期时间',
   `sent_at` timestamp(3) NULL DEFAULT NULL COMMENT '投递到统一站内信的时间',
   `reminder_key` varchar(64) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '超时提醒任务ID，每个任务最多一条',
@@ -1860,7 +1861,8 @@ CREATE TABLE `flow_notice` (
   UNIQUE KEY `uk_notice_event` (`tenant_id`,`user_id`,`event_key`),
   UNIQUE KEY `uk_notice_reminder` (`tenant_id`,`reminder_key`),
   KEY `idx_notice_owner` (`tenant_id`,`user_id`,`created_at`),
-  KEY `idx_notice_due` (`status`,`scheduled_at`)
+  KEY `idx_notice_due` (`status`,`scheduled_at`),
+  KEY `idx_notice_task` (`tenant_id`,`task_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件';
 
 -- ----------------------------

db/2pigxx.sql

@@ -1519,7 +1519,8 @@ CREATE TABLE `sys_message_relation` (
   `update_by` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '更新人',
   `del_flag` char(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '删除时间',
   `tenant_id` bigint DEFAULT NULL COMMENT '租户',
-  PRIMARY KEY (`id`)
+  PRIMARY KEY (`id`),
+  KEY `idx_relation_msg_user` (`msg_id`,`user_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='系统消息推送记录';
 
 DROP TABLE IF EXISTS `sys_system_config`;

db/3pigxx_flow.sql

@@ -319,7 +319,7 @@ CREATE TABLE `flow_notice` (
   `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID',
   `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID',
   `created_at` timestamp(3) NOT NULL COMMENT '创建时间',
-  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，CANCELLED 已取消，DISABLED 已停用',
+  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，REMOVING 待撤销，CANCELLED 已取消，DISABLED 已停用',
   `scheduled_at` timestamp(3) NULL DEFAULT NULL COMMENT '计划投递或到期时间',
   `sent_at` timestamp(3) NULL DEFAULT NULL COMMENT '投递到统一站内信的时间',
   `reminder_key` varchar(64) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '超时提醒任务ID，每个任务最多一条',
@@ -327,7 +327,8 @@ CREATE TABLE `flow_notice` (
   UNIQUE KEY `uk_notice_event` (`tenant_id`,`user_id`,`event_key`),
   UNIQUE KEY `uk_notice_reminder` (`tenant_id`,`reminder_key`),
   KEY `idx_notice_owner` (`tenant_id`,`user_id`,`created_at`),
-  KEY `idx_notice_due` (`status`,`scheduled_at`)
+  KEY `idx_notice_due` (`status`,`scheduled_at`),
+  KEY `idx_notice_task` (`tenant_id`,`task_id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件';
 
 SET FOREIGN_KEY_CHECKS = 1;

db/migration/20261013_flow_node_version.md

@@ -1,69 +0,0 @@
-# 节点配置版本隔离升级
-
-本次只隔离节点配置。业务 flowId 不变；流程图、表单定义的完整历史快照不在本次范围。任务、抄送和实例详情使用对应版本的字段权限，当前表单中未被该快照声明的字段默认隐藏；这不替代服务端的提交字段鉴权。已结束实例的详情依赖 Flowable 历史数据来确定版本，历史已清理时会明确报错，不能回退最新版。
-
-## 上线顺序
-
-1. 暂停流程发布、发起、任务处理和异步执行器，备份流程业务表及 Flowable 数据。不得新旧版本混跑：旧代码会按 flowId + nodeId 删除节点记录。
-2. 对已有库执行同目录 20261013_flow_node_version.sql；新库直接使用初始化脚本。升级 SQL 只执行一次，不删除旧配置。
-3. 核对存量定义与节点配置，建立明确的版本映射。至少覆盖所有运行中定义；如需要查询历史任务、抄送或使用旧定义，也必须覆盖对应历史定义。
-4. 将已核验配置写入对应 process_definition_id，包含 root 和所有业务节点。每个租户、定义、节点只能有一条快照。按该版本的已核验表单字段集合补齐各节点 formPerms：原来未配置/空值的字段按原有前端默认语义补 E，已有 H/R/E 保留；包含 children 中的嵌套字段及带 field 的容器。不能用当前表单的新增字段补旧版权限。无法恢复原字段集合时，此版本迁移尚未完成。
-5. 复查覆盖情况，部署全部新代码后恢复业务。无法确认映射时不要恢复受影响流程；运行期会明确报“缺少节点版本快照”，不会回退读取最新版。
-
-## 只读盘点
-
-在工作流所在数据库执行以下查询；若自定义了 Flowable 表名前缀，请先调整。
-
-    SELECT DISTINCT d.TENANT_ID_, d.KEY_, d.ID_, d.VERSION_, d.DEPLOYMENT_ID_
-    FROM ACT_RU_EXECUTION e
-    JOIN ACT_RE_PROCDEF d ON d.ID_ = e.PROC_DEF_ID_
-    ORDER BY d.TENANT_ID_, d.KEY_, d.VERSION_;
-
-    SELECT p.tenant_id, p.flow_id, p.unique_id, d.ID_, d.VERSION_, p.process
-    FROM process p
-    JOIN ACT_RE_PROCDEF d
-      ON d.DEPLOYMENT_ID_ = p.unique_id
-     AND d.KEY_ = p.flow_id
-     AND d.TENANT_ID_ = CAST(p.tenant_id AS CHAR)
-    WHERE p.del_flag = '0';
-
-    SELECT id, tenant_id, flow_id, node_id, del_flag, process_definition_id, data
-    FROM process_node_data
-    ORDER BY tenant_id, flow_id, node_id, id;
-
-process.unique_id 存储部署 ID，可以定位当前模板的引擎定义。它只能帮助定位当前发布版本，不代表所有旧节点记录都属于该版本。旧节点表保留了逻辑删除数据，但没有发布版本标记；不能按行 ID 或时间顺序自动配对。
-
-核对节点 JSON 与相应发布资料/备份，包括审批人、多实例模式、拒绝策略、字段权限、触发器和通知地址。配置中的敏感业务内容不要写入审查日志。
-
-## 存量写入模板
-
-下面只是单行映射示例，所有占位值都必须替换为已核验结果。先 SELECT 确认目标行和定义的租户、业务 key、版本一致，再在受控事务中执行。不要按 flowId 批量绑定最新版。
-
-    UPDATE process_node_data n
-    JOIN ACT_RE_PROCDEF d
-      ON d.ID_ = '<已核验的流程定义ID>'
-     AND d.KEY_ = n.flow_id
-     AND d.TENANT_ID_ = CAST(n.tenant_id AS CHAR)
-    SET n.process_definition_id = d.ID_, n.del_flag = '0'
-    WHERE n.id = <已核验的节点记录ID>
-      AND n.tenant_id = <目标租户ID>
-      AND n.process_definition_id IS NULL;
-
-一条旧记录如果需属于多个已验证版本，应分别插入不同 ID 的快照副本；不要反复改写同一行的版本 ID。若需从原始设计 JSON 重建节点，注意前端的 nodeName/childNode 与数据库 JSON 的 name/children 映射，并保存所有节点。无法恢复的历史版本应明确列为阻塞，不能用最新版替代。
-
-迁移后检查所有运行实例是否有 root 快照：
-
-    SELECT DISTINCT d.ID_, d.KEY_, d.TENANT_ID_
-    FROM ACT_RU_EXECUTION e
-    JOIN ACT_RE_PROCDEF d ON d.ID_ = e.PROC_DEF_ID_
-    LEFT JOIN process_node_data n
-      ON n.process_definition_id = d.ID_
-     AND CAST(n.tenant_id AS CHAR) = d.TENANT_ID_
-     AND n.node_id = 'root' AND n.del_flag = '0'
-    WHERE n.id IS NULL;
-
-该查询必须无遗漏；还需对照设计检查全部业务节点，不能仅用 root 存在代替完整性核验。新发布时程序先按当时的表单字段集合补齐默认 E 权限，再在部署事务中写快照，后续不得人工修改已发布数据。
-
-## 回退
-
-不能在产生新快照后直接恢复旧代码继续写流程。需要先冻结流程写入，再根据备份和期间新增实例单独制定回退方案。不要直接删除快照或批量清空版本列。

db/migration/20261013_flow_node_version.sql

@@ -1,6 +0,0 @@
--- 升级已有数据库；全新数据库使用 0pigxx_boot.sql 或 3pigxx_flow.sql。
--- 执行前停止流程写入并备份；本脚本不猜测存量节点与流程版本的对应关系。
--- 必须完成同目录说明中的存量核对后再上线应用。
-ALTER TABLE process_node_data
-    ADD COLUMN process_definition_id varchar(64) DEFAULT NULL COMMENT '引擎流程定义ID，发布版本快照',
-    ADD UNIQUE KEY uk_node_definition (tenant_id, process_definition_id, node_id);

db/migration/20261014_flow_first_batch.md

@@ -1,61 +0,0 @@
-# 流程第一批能力升级
-
-## 范围与入口
-
-- 流程设计器：保存设计草稿、设计草稿箱；保存基本信息、表单和节点树，不部署流程。
-- 流程列表：申请草稿箱；动态表单发起抽屉可保存表单值和自选审批人。按用户和租户隔离，版本号防止多窗口覆盖。
-- 待办页：流程通知，包含任务分配、发起人催办和超时提醒；点击通知沿用现有任务详情权限检查。历史通知可能对应已完成或已转交任务。
-- 我发起的：运行中流程可催办，同一实例十分钟内最多一次；仅发起人可以操作。回调记录可查看状态和手动重试失败事件。
-- 审批节点 / 事件通知：配置自然时长分钟数，0 或留空关闭；最多 525600 分钟。计时基于任务创建时间，不因转交重新计时，到期只提醒当时的当前审批人一次。
-
-本批申请草稿适用于流程引擎的动态表单。自定义业务组件自行提交、拥有独立的数据模型，继续由业务页面保存草稿；未擅自修改业务表单合同。申请草稿记录发布 deployment ID，恢复时版本变化会拒绝自动恢复并保留原草稿。普通草稿只在明确保存或设计器点击下一步时持久化，未保存的输入不会自动落库。
-
-## 升级顺序
-
-1. 备份数据库并停止本次流程写入，确认上一次节点版本隔离迁移已经完成。
-2. 在流程与引擎使用的同一 MySQL 数据源执行 `20261014_flow_first_batch.sql`。新增 `flow_draft`、`flow_notice`、`flow_callback` 三张表，不改动存量业务表；新装脚本 `0pigxx_boot.sql`、`3pigxx_flow.sql` 已包含相同建表语句。
-3. 部署后端与管理端。不要混用新旧后端：旧版本不会创建持久回调与通知记录。
-4. 使用测试租户检查保存和恢复两类草稿、任务分配与转交、催办限频、到期提醒、回调 503 后重试。
-
-新部署的监听器负责此后发生的任务事件；不自动重放过去的任务通知或已完成回调，不自动为已存在且没有再次分配的任务补建超时记录。
-
-## 已安装五表版本的兼容升级
-
-仅已经执行过旧五表脚本的环境需要此步骤；新装或尚未安装本批能力的用户只执行三表建表脚本。
-
-1. 停止旧后端及其他流程写入，完整备份原五张表。恢复备份前也必须停止新后端，不能混跑两版。
-2. 执行三表建表脚本 `20261014_flow_first_batch.sql`，再执行 `20261014_flow_merge_tables.sql`（使用支持 `DELIMITER` 的 MySQL 客户端）。合并脚本只补字段和索引；数据部分使用事务。发生异常时数据部分回滚，已完成的 DDL 保留，排查后可重新执行。
-3. 合并脚本保留所有草稿和现有通知的 ID、已读状态；把超时监听关联到已有超时通知，或转换为待触发记录。已提醒的监听不会重新排队；未启用截止时间的旧记录为 `DISABLED`。旧回调的 ID、状态、次数、租约、地址和载荷保持原值，尝试历史按原 ID 合入 JSON 数组。孤立审计记录会报错并回滚，不会静默丢弃。
-4. 核对下面两个查询结果均为 0，并核对备份中的草稿、通知和回调业务字段。合并脚本可重复执行，不会重复追加历史或重置已处理的提醒。
-5. 旧 `flow_task_watch` 和 `flow_callback_attempt` 默认保留为只读核对来源，新代码不再访问。确认备份可用、迁移完整后可清理这两张旧表。旧监听中的 `last_urge_at` 仅写不读，原值保存在旧表和备份；催办限频仍以通知记录为准。
-
-```sql
--- 未迁移的监听数
-SELECT COUNT(*) FROM flow_task_watch w
-WHERE NOT EXISTS (
- SELECT 1 FROM flow_notice n WHERE n.tenant_id=w.tenant_id AND n.reminder_key=w.task_id
-);
--- 未迁移的回调尝试数
-SELECT COUNT(*) FROM flow_callback_attempt a
-LEFT JOIN flow_callback c ON c.id=a.callback_id AND c.tenant_id=a.tenant_id
-WHERE c.id IS NULL OR JSON_SEARCH(COALESCE(c.history,'[]'),'one',a.id,NULL,'$[*].id') IS NULL;
-```
-
-## 事务和投递约定
-
-- 通知与回调写入使用 `JdbcTemplate`，与 Flowable 共用 Spring 事务及数据源；独立监听器 `isFailOnException=true`，写入失败会回滚任务操作，避免任务完成但回调丢失。
-- 新表 SQL 显式指定租户、用户或引擎事件身份；后台跨租户扫描仅取内部记录，HTTP API 不允许指定他人的用户或租户。
-- `flow_notice` 同时保存已生成通知和待触发提醒。`status` 区分 `PENDING`、`SENT`、`CANCELLED`、`DISABLED`；通知列表和已读操作只接受 `SENT`。`reminder_key` 按租户和任务唯一，到期才确定当前接收人；任务结束取消尚未发送的提醒，已发送记录保留。任务暂停或暂无处理人时保留 `PENDING`，将 `scheduled_at` 推迟十秒后再检查，避免占满最早一批扫描记录；恢复后继续发送，不重新计算完整超时时长。
-- 后台默认每轮完成后间隔 10 秒扫描，可配置 `flow.delivery.scan-delay`（毫秒）。回调与提醒使用本模块独立的两线程调度器；每批上限为 20 个回调、100 个提醒，不承诺精确到秒的触发。
-- HTTP 回调使用任务完成时固定的目标地址、JSON 载荷和事件 ID。只接受 2xx；重定向、其他状态码、网络错误均进入失败处理。请求超时 10 秒，不携带用户 Bearer token。
-- 首次共尝试 5 次，失败后按 30、60、120、240 秒退避。60 秒租约与条件更新允许多实例竞争；进程崩溃后可回收过期租约继续投递，达到尝试上限的未知结果标为失败。
-- 请求头 `X-Flow-Event-Id` 和 `Idempotency-Key` 使用同一个稳定事件 ID。语义是**至少一次投递**：接收方必须把事件 ID 去重与自身业务写入放在同一事务中，才能避免“已处理但响应丢失”造成重复副作用。
-- 手动重试仅允许实例发起人对失败记录操作，保留事件 ID、目标和载荷，再给予 5 次尝试。累计次数不清零。
-- `flow_callback.history` 以 JSON 数组保存每次尝试和手动重试的审计记录，与主记录状态在同一事务中更新，不截断历史。审计归属由父记录的租户和事件 ID 确定。STARTED 表示尝试已开始但没有确认结果；可能是进程退出或租约失效。页面只展示状态、次数和最近结果，不返回回调载荷和目标 URL。
-- 本批可靠投递针对节点“事件通知”HTTP 回调；同步触发器业务调用、Redis 流程状态广播维持原有行为。
-
-## 验证证据
-
-`FlowWorkspaceTests` 使用真实 Spring Flowable、共享事务的 H2、生产建表语句（H2 MySQL 兼容模式）和本地 HTTP 服务，覆盖草稿持久化、所有者隔离、乐观锁、分配/转交通知、超时去重、催办限频、503 重试、固定事件 ID，以及回调写入失败后引擎任务回滚。H2 验证不等同于真实 MySQL 或完整应用运行验证。
-
-相关既有节点版本、会签和任务服务测试一并执行。管理端进行指定文件 ESLint、Prettier、Vue SFC 编译，以及已有版本隔离浏览器回归。三表合并额外验证了历史追加、手动重试保留旧历史、已完成任务不再提醒。迁移在本地 OceanBase seekdb 的临时数据库使用带数据样本验证，并检查重复执行和孤立审计回滚；临时库测试后清理。真实账号完整联调尚未执行，不把接口隔离回归当成真实后端验收。

db/migration/20261014_flow_first_batch.sql

@@ -1,27 +0,0 @@
--- 第一批持久化能力：三张业务表。已安装旧五表版本时，建表后另执行 20261014_flow_merge_tables.sql。
--- 升级前备份并停写，先执行 DDL 再部署应用；不自动迁移运行中的任务。
-CREATE TABLE IF NOT EXISTS flow_draft (
- id varchar(36) NOT NULL, tenant_id bigint NOT NULL, user_id bigint NOT NULL,
- kind varchar(16) NOT NULL, flow_id varchar(64) NOT NULL DEFAULT '', title varchar(200) NOT NULL,
- payload longtext NOT NULL, revision bigint NOT NULL, updated_at timestamp(3) NOT NULL,
- PRIMARY KEY(id), KEY idx_draft_owner(tenant_id,user_id,kind,updated_at)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-CREATE TABLE IF NOT EXISTS flow_notice (
- id varchar(36) NOT NULL, tenant_id bigint NOT NULL, user_id varchar(64) NOT NULL,
- event_key varchar(180) NOT NULL, kind varchar(16) NOT NULL, title varchar(200) NOT NULL,
- task_id varchar(64) NOT NULL, instance_id varchar(64) NOT NULL,
- created_at timestamp(3) NOT NULL, read_at timestamp(3) NULL,
- status varchar(16) NOT NULL DEFAULT 'SENT', scheduled_at timestamp(3) NULL, sent_at timestamp(3) NULL,
- reminder_key varchar(64) NULL,
- PRIMARY KEY(id), UNIQUE KEY uk_notice_event(tenant_id,user_id,event_key), KEY idx_notice_owner(tenant_id,user_id,created_at),
- UNIQUE KEY uk_notice_reminder(tenant_id,reminder_key), KEY idx_notice_due(status,scheduled_at)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
-CREATE TABLE IF NOT EXISTS flow_callback (
- id varchar(36) NOT NULL, tenant_id bigint NOT NULL, task_id varchar(64) NOT NULL,
- instance_id varchar(64) NOT NULL, target_url varchar(2000) NOT NULL, payload longtext NOT NULL,
- status varchar(16) NOT NULL, attempts int NOT NULL DEFAULT 0, max_attempts int NOT NULL DEFAULT 5, next_at timestamp(3) NOT NULL,
- history longtext NULL,
- lease_token varchar(36) NULL, lease_until timestamp(3) NULL, last_error varchar(500) NULL,
- created_at timestamp(3) NOT NULL, updated_at timestamp(3) NOT NULL,
- PRIMARY KEY(id), UNIQUE KEY uk_callback_task(tenant_id,task_id), KEY idx_callback_due(status,next_at)
-) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

db/migration/20261014_flow_merge_tables.sql

@@ -1,100 +0,0 @@
--- 仅用于已经安装旧五表版本的库。先备份、停止流程写入及后台扫描，并执行三表建表脚本。
--- 可重复执行；旧表暂保留作为核对来源，新代码不再访问。核验并备份后再清理旧表。
-DELIMITER $$
-DROP PROCEDURE IF EXISTS migrate_flow_three_tables$$
-CREATE PROCEDURE migrate_flow_three_tables()
-BEGIN
- DECLARE finished INT DEFAULT 0;
- DECLARE v_task VARCHAR(64);
- DECLARE v_tenant BIGINT;
- DECLARE v_instance VARCHAR(64);
- DECLARE v_due TIMESTAMP(3);
- DECLARE v_reminded TIMESTAMP(3);
- DECLARE v_id VARCHAR(36);
- DECLARE v_callback VARCHAR(36);
- DECLARE v_attempt INT;
- DECLARE v_result VARCHAR(16);
- DECLARE v_detail VARCHAR(500);
- DECLARE v_created TIMESTAMP(3);
- DECLARE v_history LONGTEXT;
- DECLARE watches CURSOR FOR SELECT task_id,tenant_id,instance_id,due_at,reminded_at FROM flow_task_watch;
- DECLARE audits CURSOR FOR SELECT id,callback_id,tenant_id,attempt_no,result,detail,created_at FROM flow_callback_attempt ORDER BY created_at,id;
- DECLARE CONTINUE HANDLER FOR NOT FOUND SET finished=1;
-
- IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='status') THEN
-  ALTER TABLE flow_notice ADD COLUMN status VARCHAR(16) NOT NULL DEFAULT 'SENT';
- END IF;
- IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='scheduled_at') THEN
-  ALTER TABLE flow_notice ADD COLUMN scheduled_at TIMESTAMP(3) NULL;
- END IF;
- IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='sent_at') THEN
-  ALTER TABLE flow_notice ADD COLUMN sent_at TIMESTAMP(3) NULL;
- END IF;
- IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='reminder_key') THEN
-  ALTER TABLE flow_notice ADD COLUMN reminder_key VARCHAR(64) NULL;
- END IF;
- IF NOT EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='flow_notice' AND index_name='uk_notice_reminder') THEN
-  ALTER TABLE flow_notice ADD UNIQUE KEY uk_notice_reminder(tenant_id,reminder_key);
- END IF;
- IF NOT EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='flow_notice' AND index_name='idx_notice_due') THEN
-  ALTER TABLE flow_notice ADD KEY idx_notice_due(status,scheduled_at);
- END IF;
- IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_callback' AND column_name='history') THEN
-  ALTER TABLE flow_callback ADD COLUMN history LONGTEXT NULL;
- END IF;
-
- -- DDL 自动提交；以下数据搬迁独立事务，任何异常回滚数据部分，修正后可重跑。
- BEGIN
-  DECLARE EXIT HANDLER FOR SQLEXCEPTION
-  BEGIN
-   ROLLBACK;
-   RESIGNAL;
-  END;
-  START TRANSACTION;
-  UPDATE flow_notice SET sent_at=created_at WHERE status='SENT' AND sent_at IS NULL;
-  IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='flow_task_watch') THEN
-   SET finished=0;
-   OPEN watches;
-   watch_loop: LOOP
-    FETCH watches INTO v_task,v_tenant,v_instance,v_due,v_reminded;
-    IF finished=1 THEN LEAVE watch_loop; END IF;
-    IF NOT EXISTS (SELECT 1 FROM flow_notice WHERE tenant_id=v_tenant AND reminder_key=v_task) THEN
-     IF (SELECT COUNT(*) FROM flow_notice WHERE tenant_id=v_tenant AND event_key=CONCAT('OVERDUE:',v_task) AND kind='OVERDUE') > 1 THEN
-      SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='Multiple overdue notices for one watch; inspect before migration';
-     END IF;
-     IF EXISTS (SELECT 1 FROM flow_notice WHERE tenant_id=v_tenant AND event_key=CONCAT('OVERDUE:',v_task) AND kind='OVERDUE') THEN
-      UPDATE flow_notice SET reminder_key=v_task,scheduled_at=v_due,sent_at=COALESCE(v_reminded,sent_at)
-       WHERE tenant_id=v_tenant AND event_key=CONCAT('OVERDUE:',v_task) AND kind='OVERDUE';
-     ELSE
-      INSERT INTO flow_notice(id,tenant_id,user_id,event_key,kind,title,task_id,instance_id,created_at,status,scheduled_at,sent_at,reminder_key)
-       VALUES(UUID(),v_tenant,'',CONCAT('OVERDUE:',v_task),'OVERDUE','流程任务',v_task,v_instance,CURRENT_TIMESTAMP(3),
-        CASE WHEN v_reminded IS NOT NULL THEN 'SENT' WHEN v_due IS NULL THEN 'DISABLED' ELSE 'PENDING' END,v_due,v_reminded,v_task);
-     END IF;
-    END IF;
-   END LOOP;
-   CLOSE watches;
-  END IF;
-  IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='flow_callback_attempt') THEN
-   SET finished=0;
-   OPEN audits;
-   audit_loop: LOOP
-    FETCH audits INTO v_id,v_callback,v_tenant,v_attempt,v_result,v_detail,v_created;
-    IF finished=1 THEN LEAVE audit_loop; END IF;
-    IF NOT EXISTS (SELECT 1 FROM flow_callback WHERE id=v_callback AND tenant_id=v_tenant) THEN
-     SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='Orphan callback attempt; retain old tables and inspect';
-    END IF;
-    SELECT COALESCE(history,'[]') INTO v_history FROM flow_callback WHERE id=v_callback AND tenant_id=v_tenant FOR UPDATE;
-    IF JSON_SEARCH(v_history,'one',v_id,NULL,'$[*].id') IS NULL THEN
-     UPDATE flow_callback SET history=JSON_ARRAY_APPEND(v_history,'$',JSON_OBJECT('id',v_id,'attempt_no',v_attempt,
-       'result',v_result,'detail',v_detail,'created_at',DATE_FORMAT(v_created,'%Y-%m-%d %H:%i:%s.%f')))
-      WHERE id=v_callback AND tenant_id=v_tenant;
-    END IF;
-   END LOOP;
-   CLOSE audits;
-  END IF;
-  COMMIT;
- END;
-END$$
-CALL migrate_flow_three_tables()$$
-DROP PROCEDURE migrate_flow_three_tables$$
-DELIMITER ;

pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeStatusEnum.java

@@ -19,7 +19,10 @@ public enum FlowNoticeStatusEnum {
 	/** 已投递到统一站内信，已读状态由 sys_message_relation 维护。 */
 	SENT("SENT"),
 
-	/** 任务已经结束，取消尚未发送的提醒。 */
+	/** 任务结束或接收人变更，等待删除统一站内信的接收记录。 */
+	REMOVING("REMOVING"),
+
+	/** 任务已经结束，提醒已取消或接收记录已删除。 */
 	CANCELLED("CANCELLED"),
 
 	/** 旧监听没有截止时间，迁移后保留但不参与扫描。 */

pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java

@@ -1,6 +1,8 @@
 package com.pig4cloud.pigx.flow.entity;
 
+import com.baomidou.mybatisplus.annotation.FieldStrategy;
 import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.pig4cloud.pigx.common.core.util.TenantTable;
@@ -21,7 +23,8 @@ public class FlowNotice {
 	/** 所属租户 ID。 */
 	private Long tenantId;
 
-	/** 接收人 ID；超时提醒到期前为空串。 */
+	/** 接收人 ID；超时提醒到期前为空串，插入时保留该值。 */
+	@TableField(insertStrategy = FieldStrategy.NOT_NULL)
 	private String userId;
 
 	/** 事件去重键，与租户、接收人组成唯一键。 */

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java

@@ -2,6 +2,8 @@ package com.pig4cloud.pigx.flow.config;
 
 import com.pig4cloud.pigx.flow.service.IFlowNoticeService;
 import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener;
+import com.pig4cloud.pigx.flow.support.listeners.FlowNoticeRemovalEvent;
+import org.springframework.transaction.event.TransactionalEventListener;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.spring.SpringProcessEngineConfiguration;
@@ -95,4 +97,22 @@ public class FlowDurableConfiguration {
 		}
 	}
 
+	/**
+	 * 引擎事务提交后立即撤销失效通知；事务回滚时不执行，避免丢失提醒。失败只记录日志，由扫描器重试。
+	 * @param event 已置为 REMOVING 的通知事件
+	 */
+	@TransactionalEventListener(fallbackExecution = true)
+	public void removeNoticesAfterCommit(FlowNoticeRemovalEvent event) {
+		IFlowNoticeService service = noticeService.getObject();
+		for (String id : event.noticeIds()) {
+			try {
+				service.deliverNotice(id);
+			}
+			catch (Exception exception) {
+				// REMOVING 已随引擎事务提交，扫描器会再次尝试；不能把已成功的审批报告为失败。
+				log.warn("流程站内信暂未删除，事件 {}", id, exception);
+			}
+		}
+	}
+
 }

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java

@@ -43,20 +43,22 @@ public interface IFlowNoticeService extends IService<FlowNotice> {
 	void remind(long tenant, String taskId);
 
 	/**
-	 * 查询最多 100 条待投递的通知事件。
+	 * 查询最多 100 条到期的待投递或待撤销通知事件。
 	 * @return 只包含 ID 的通知事件
 	 */
 	List<FlowNotice> listDueNotices();
 
 	/**
-	 * 投递通知事件到统一站内信；行锁防止扫描器并发发送同一事件。
+	 * 在独立事务中处理一条通知事件，行锁防止扫描器与提交后回调并发处理同一事件。
+	 * <p>
+	 * READY 且任务仍归原接收人时投递，否则撤销；REMOVING 删除原接收人的接收记录后置为 CANCELLED。
 	 * @param id 通知事件 ID
-	 * @throws IllegalStateException 站内信服务返回失败，由扫描器延迟重试
+	 * @throws IllegalStateException 站内信发送或删除失败，状态保持不变，由扫描器延迟重试
 	 */
 	void deliverNotice(String id);
 
 	/**
-	 * 投递失败后延迟一分钟再试。
+	 * 投递或撤销失败后延迟一分钟再试。
 	 * @param id 通知事件 ID
 	 */
 	void delayNotice(String id);

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java

@@ -5,8 +5,10 @@ import cn.hutool.core.util.EnumUtil;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO;
 import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService;
+import com.pig4cloud.pigx.common.core.constant.CommonConstants;
 import com.pig4cloud.pigx.common.core.constant.enums.YesNoEnum;
 import com.pig4cloud.pigx.common.data.tenant.TenantBroker;
 import com.pig4cloud.pigx.flow.constant.FlowNoticeKindEnum;
@@ -19,6 +21,7 @@ import com.pig4cloud.pigx.flow.entity.FlowNotice;
 import com.pig4cloud.pigx.flow.entity.ProcessInstanceRecord;
 import com.pig4cloud.pigx.flow.mapper.FlowNoticeMapper;
 import com.pig4cloud.pigx.flow.support.listeners.NodeTaskCompleteNotify;
+import com.pig4cloud.pigx.flow.support.listeners.FlowNoticeRemovalEvent;
 import com.pig4cloud.pigx.flow.service.IFlowNoticeService;
 import com.pig4cloud.pigx.flow.service.IProcessInstanceRecordService;
 import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
@@ -28,6 +31,8 @@ import org.flowable.engine.TaskService;
 import org.flowable.task.service.impl.persistence.entity.TaskEntity;
 import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
 import java.time.LocalDateTime;
 import java.util.List;
@@ -49,6 +54,8 @@ public class FlowNoticeServiceImpl extends ServiceImpl<FlowNoticeMapper, FlowNot
 
 	private final RemoteMessageService messageService;
 
+	private final ApplicationEventPublisher eventPublisher;
+
 	/**
 	 * {@inheritDoc}
 	 */
@@ -57,17 +64,23 @@ public class FlowNoticeServiceImpl extends ServiceImpl<FlowNoticeMapper, FlowNot
 		if (StrUtil.isBlank(task.getTenantId()) || StrUtil.isBlank(task.getProcessDefinitionId()))
 			return;
 		long tenant = Long.parseLong(task.getTenantId());
-		if (FlowableEngineEventType.ENTITY_DELETED.name().equals(event)) {
+		boolean ended = FlowableEngineEventType.ENTITY_DELETED.name().equals(event)
+				|| FlowableEngineEventType.TASK_COMPLETED.name().equals(event);
+		if (ended) {
 			cancelReminder(tenant, task.getId());
-			return;
+			removeTaskNotices(tenant, task.getId(), null);
 		}
+		else if (FlowableEngineEventType.TASK_ASSIGNED.name().equals(event)) {
+			removeTaskNotices(tenant, task.getId(), task.getAssignee());
+		}
+		if (FlowableEngineEventType.ENTITY_DELETED.name().equals(event))
+			return;
 		Node node = TenantBroker.applyAs(tenant,
 				t -> nodeService.getPublishedNode(task.getProcessDefinitionId(), task.getTaskDefinitionKey())
 					.getData());
 		if (Objects.isNull(node))
 			return;
 		if (FlowableEngineEventType.TASK_COMPLETED.name().equals(event)) {
-			cancelReminder(tenant, task.getId());
 			if (StrUtil.isNotBlank(node.getEventConfig()))
 				nodeTaskCompleteNotify.notifyTaskCompleted(task, node.getEventConfig());
 			return;
@@ -222,7 +235,7 @@ public class FlowNoticeServiceImpl extends ServiceImpl<FlowNoticeMapper, FlowNot
 	@Override
 	public List<FlowNotice> listDueNotices() {
 		return lambdaQuery().select(FlowNotice::getId)
-			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode())
+			.in(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode(), FlowNoticeStatusEnum.REMOVING.getCode())
 			.le(FlowNotice::getScheduledAt, LocalDateTime.now())
 			.orderByAsc(FlowNotice::getScheduledAt)
 			.page(new Page<>(1, 100, false))
@@ -233,14 +246,25 @@ public class FlowNoticeServiceImpl extends ServiceImpl<FlowNoticeMapper, FlowNot
 	 * {@inheritDoc}
 	 */
 	@Override
-	@Transactional(rollbackFor = Exception.class)
+	@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
 	public void deliverNotice(String id) {
 		FlowNotice row = lambdaQuery().eq(FlowNotice::getId, id)
-			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode())
+			.in(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode(), FlowNoticeStatusEnum.REMOVING.getCode())
 			.last("FOR UPDATE")
 			.one();
 		if (Objects.isNull(row))
 			return;
+		if (FlowNoticeStatusEnum.REMOVING.getCode().equals(row.getStatus())) {
+			removeNotice(row);
+			return;
+		}
+		// 出队前核对任务与接收人，避免完成或转办后的旧通知再次投递。
+		var task = taskService.createTaskQuery().taskId(row.getTaskId())
+			.taskTenantId(String.valueOf(row.getTenantId())).singleResult();
+		if (Objects.isNull(task) || !Objects.equals(task.getAssignee(), row.getUserId())) {
+			removeNotice(row);
+			return;
+		}
 		MessageNoticeDTO notice = new MessageNoticeDTO();
 		// 站内信分类，upms 尚无对应常量。
 		notice.setCategory("1");
@@ -268,10 +292,41 @@ public class FlowNoticeServiceImpl extends ServiceImpl<FlowNoticeMapper, FlowNot
 	public void delayNotice(String id) {
 		lambdaUpdate().set(FlowNotice::getScheduledAt, LocalDateTime.now().plusSeconds(60))
 			.eq(FlowNotice::getId, id)
-			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode())
+			.in(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode(), FlowNoticeStatusEnum.REMOVING.getCode())
 			.update();
 	}
 
+	/** 引擎事务内持久化撤销意图；提交后立即执行，远程失败由原投递扫描器重试。 */
+	private void removeTaskNotices(long tenant, String taskId, String currentAssignee) {
+		var rows = lambdaQuery().eq(FlowNotice::getTenantId, tenant)
+			.eq(FlowNotice::getTaskId, taskId)
+			.ne(StrUtil.isNotBlank(currentAssignee), FlowNotice::getUserId, currentAssignee)
+			.in(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode(), FlowNoticeStatusEnum.SENT.getCode())
+			.list();
+		if (rows.isEmpty())
+			return;
+		List<String> ids = rows.stream().map(FlowNotice::getId).toList();
+		lambdaUpdate().in(FlowNotice::getId, ids)
+			.eq(FlowNotice::getTenantId, tenant)
+			.in(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode(), FlowNoticeStatusEnum.SENT.getCode())
+			.set(FlowNotice::getStatus, FlowNoticeStatusEnum.REMOVING.getCode())
+			.set(FlowNotice::getScheduledAt, LocalDateTime.now()).update();
+		eventPublisher.publishEvent(new FlowNoticeRemovalEvent(ids));
+	}
+
+	private void removeNotice(FlowNotice row) {
+		FlowMessageRemovalDTO removal = new FlowMessageRemovalDTO();
+		removal.setTaskId(row.getTaskId());
+		removal.setRequestKey("flow:" + row.getId());
+		removal.setUserId(Long.parseLong(row.getUserId()));
+		var result = TenantBroker.applyAs(row.getTenantId(), t -> messageService.removeFlowNotice(removal));
+		if (Objects.isNull(result) || result.getCode() != CommonConstants.SUCCESS
+				|| !Boolean.TRUE.equals(result.getData()))
+			throw new IllegalStateException("流程站内信删除失败");
+		lambdaUpdate().eq(FlowNotice::getId, row.getId())
+			.set(FlowNotice::getStatus, FlowNoticeStatusEnum.CANCELLED.getCode()).update();
+	}
+
 	private void cancelReminder(long tenant, String task) {
 		lambdaUpdate().set(FlowNotice::getStatus, FlowNoticeStatusEnum.CANCELLED.getCode())
 			.eq(FlowNotice::getTenantId, tenant)

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowNoticeRemovalEvent.java

@@ -0,0 +1,11 @@
+package com.pig4cloud.pigx.flow.support.listeners;
+
+import java.util.List;
+
+/**
+ * 引擎事务提交后立即撤销通知；失败仍由持久化状态驱动重试。
+ *
+ * @param noticeIds 已置为 REMOVING 的通知事件 ID
+ */
+public record FlowNoticeRemovalEvent(List<String> noticeIds) {
+}

pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeLifecycleTests.java

@@ -0,0 +1,153 @@
+package com.pig4cloud.pigx.flow.service.impl;
+
+import com.baomidou.mybatisplus.annotation.FieldStrategy;
+import com.baomidou.mybatisplus.core.config.GlobalConfig;
+import com.baomidou.mybatisplus.core.toolkit.GlobalConfigUtils;
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder;
+import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO;
+import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService;
+import com.pig4cloud.pigx.common.core.util.R;
+import com.pig4cloud.pigx.flow.entity.FlowNotice;
+import com.pig4cloud.pigx.flow.mapper.FlowNoticeMapper;
+import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
+import com.pig4cloud.pigx.flow.support.listeners.FlowNoticeRemovalEvent;
+import org.apache.ibatis.mapping.Environment;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mybatis.spring.SqlSessionTemplate;
+import org.mybatis.spring.transaction.SpringManagedTransactionFactory;
+import org.flowable.engine.TaskService;
+import org.flowable.task.api.TaskQuery;
+import org.flowable.task.service.impl.persistence.entity.TaskEntity;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.test.util.ReflectionTestUtils;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+class FlowNoticeLifecycleTests {
+
+	private JdbcTemplate jdbc;
+	private FlowNoticeServiceImpl service;
+	private RemoteMessageService messages;
+	private ApplicationEventPublisher events;
+	private TransactionTemplate transaction;
+
+	@BeforeEach
+	void setUp() {
+		var dataSource = new JdbcDataSource();
+		dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1");
+		jdbc = new JdbcTemplate(dataSource);
+		jdbc.execute("create table flow_notice(id varchar(64) primary key, tenant_id bigint, user_id varchar(64) not null, "
+				+ "event_key varchar(255), kind varchar(32), title varchar(255), task_id varchar(64), "
+				+ "instance_id varchar(64), created_at timestamp, status varchar(32), scheduled_at timestamp, "
+				+ "sent_at timestamp, reminder_key varchar(64))");
+		var configuration = new MybatisConfiguration();
+		configuration.setMapUnderscoreToCamelCase(true);
+		configuration.setEnvironment(new Environment("test", new SpringManagedTransactionFactory(), dataSource));
+		GlobalConfigUtils.setGlobalConfig(configuration, new GlobalConfig()
+				.setDbConfig(new GlobalConfig.DbConfig().setInsertStrategy(FieldStrategy.NOT_EMPTY)));
+		configuration.addMapper(FlowNoticeMapper.class);
+		var factory = new MybatisSqlSessionFactoryBuilder().build(configuration);
+		messages = mock(RemoteMessageService.class);
+		events = mock(ApplicationEventPublisher.class);
+		var tasks = mock(TaskService.class);
+		var query = mock(TaskQuery.class, RETURNS_SELF);
+		when(tasks.createTaskQuery()).thenReturn(query);
+		var nodes = mock(IProcessNodeDataService.class);
+		when(nodes.getPublishedNode(any(), any())).thenReturn(R.ok());
+		service = new FlowNoticeServiceImpl(null, null, nodes, tasks, messages, events);
+		ReflectionTestUtils.setField(service, "baseMapper", new SqlSessionTemplate(factory).getMapper(FlowNoticeMapper.class));
+		transaction = new TransactionTemplate(new DataSourceTransactionManager(dataSource));
+		when(messages.removeFlowNotice(any())).thenReturn(R.ok(true));
+	}
+
+	@Test
+	void pendingReminderPersistsEmptyRecipientWithNotEmptyGlobalStrategy() {
+		var reminder = new FlowNotice();
+		reminder.setId("pending-reminder");
+		reminder.setUserId("");
+		reminder.setStatus("PENDING");
+		reminder.setKind("OVERDUE");
+		assertThat(service.save(reminder)).isTrue();
+		assertThat(jdbc.queryForObject("select user_id from flow_notice where id=?",
+				String.class, reminder.getId())).isEmpty();
+	}
+
+	@ParameterizedTest
+	@ValueSource(strings = { "ENTITY_DELETED", "TASK_COMPLETED", "TASK_ASSIGNED" })
+	void endedOrTransferredTaskRemovesOnlyObsoleteNotices(String event) {
+		insert("todo", 1, "task-A", "SENT", null);
+		insert("urge", 1, "task-A", "SENT", null);
+		insert("queued", 1, "task-A", "READY", null);
+		insert("overdue", 1, "task-A", "PENDING", "task-A");
+		insert("next-task", 1, "task-B", "SENT", null);
+		insert("other-tenant", 2, "task-A", "SENT", null);
+		insert("current-owner", 1, "task-A", "SENT", null);
+		jdbc.update("update flow_notice set user_id='2002' where id='current-owner'");
+		var task = mock(TaskEntity.class);
+		when(task.getId()).thenReturn("task-A");
+		when(task.getTenantId()).thenReturn("1");
+		when(task.getProcessDefinitionId()).thenReturn("process-A");
+		when(task.getTaskDefinitionKey()).thenReturn("node-A");
+		when(task.getAssignee()).thenReturn("2002");
+		transaction.executeWithoutResult(status -> service.record(task, event));
+		verifyNoInteractions(messages);
+		verify(events).publishEvent(any(FlowNoticeRemovalEvent.class));
+		assertThat(state("overdue")).isEqualTo(event.equals("TASK_ASSIGNED") ? "PENDING" : "CANCELLED");
+		for (String id : new String[] { "todo", "urge", "queued" }) {
+			assertThat(state(id)).isEqualTo("REMOVING");
+			transaction.executeWithoutResult(status -> service.deliverNotice(id));
+			assertThat(state(id)).isEqualTo("CANCELLED");
+		}
+		verify(messages, times(3)).removeFlowNotice(argThat(removal ->
+				removal.getTaskId().equals("task-A") && removal.getUserId().equals(1001L)
+						&& removal.getRequestKey().startsWith("flow:")));
+		verify(messages, never()).sendNotice(any());
+		if (event.equals("TASK_ASSIGNED")) {
+			assertThat(state("current-owner")).isEqualTo("SENT");
+		}
+		else {
+			assertThat(state("current-owner")).isEqualTo("REMOVING");
+			transaction.executeWithoutResult(status -> service.deliverNotice("current-owner"));
+			assertThat(state("current-owner")).isEqualTo("CANCELLED");
+		}
+		assertThat(state("next-task")).isEqualTo("SENT");
+		assertThat(state("other-tenant")).isEqualTo("SENT");
+	}
+
+	@Test
+	void failedRemovalCanRetryAndAnEndedTaskIsNeverDeliveredAgain() {
+		// READY 可能是发送成功但响应丢失的重试；即使任务已结束也要尝试清理原消息。
+		insert("queued", 1, "task-A", "READY", null);
+		when(messages.removeFlowNotice(any())).thenReturn(R.failed("暂时不可用"));
+		assertThatThrownBy(() -> transaction.executeWithoutResult(status -> service.deliverNotice("queued")))
+				.isInstanceOf(IllegalStateException.class);
+		assertThat(state("queued")).isEqualTo("READY");
+		when(messages.removeFlowNotice(any())).thenReturn(R.ok(true));
+		transaction.executeWithoutResult(status -> service.deliverNotice("queued"));
+		assertThat(state("queued")).isEqualTo("CANCELLED");
+		verify(messages, never()).sendNotice(any());
+		verify(messages, times(2)).removeFlowNotice(any(FlowMessageRemovalDTO.class));
+	}
+
+	private void insert(String id, long tenant, String task, String status, String reminder) {
+		jdbc.update("insert into flow_notice(id,tenant_id,user_id,task_id,status,reminder_key,scheduled_at) values(?,?,?,?,?,?,CURRENT_TIMESTAMP)",
+				id, tenant, "1001", task, status, reminder);
+	}
+
+	private String state(String id) {
+		return jdbc.queryForObject("select status from flow_notice where id=?", String.class, id);
+	}
+}

pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/FlowMessageRemovalDTO.java

@@ -0,0 +1,29 @@
+package com.pig4cloud.pigx.admin.api.dto;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import lombok.Data;
+
+/** 精确撤销某次流程通知的接收记录，避免误删转办后新产生的通知。 */
+@Data
+public class FlowMessageRemovalDTO {
+
+	/**
+	 * 流程任务 ID，对应站内信 business_id
+	 */
+	@NotBlank
+	private String taskId;
+
+	/**
+	 * 通知事件标识，格式为 flow: 加通知事件 ID，对应站内信 request_key
+	 */
+	@NotBlank
+	private String requestKey;
+
+	/**
+	 * 原接收人用户 ID
+	 */
+	@NotNull
+	private Long userId;
+
+}

pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/feign/RemoteMessageService.java

@@ -19,6 +19,7 @@
 
 package com.pig4cloud.pigx.admin.api.feign;
 
+import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageEmailDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageHookDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO;
@@ -76,4 +77,14 @@ public interface RemoteMessageService {
     @PostMapping("/sysMessage/internal/notice")
     R sendNotice(@RequestBody MessageNoticeDTO noticeDTO);
 
+    /**
+     * 逻辑删除指定流程通知在原接收人下的接收记录；仅供流程服务内部调用。
+     *
+     * @param removal 任务 ID、通知事件标识和原接收人
+     * @return 处理成功为 true，记录不存在也视为成功
+     */
+    @NoToken
+    @PostMapping("/sysMessage/internal/flow-notice/remove")
+    R<Boolean> removeFlowNotice(@RequestBody FlowMessageRemovalDTO removal);
+
 }

pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java

@@ -6,6 +6,7 @@ import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageEmailDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageHookDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO;
@@ -285,4 +286,16 @@ public class SysMessageController {
         return sysMessageService.sendNotice(noticeDTO);
     }
 
+    /**
+     * 流程完成、取消或转办后，逻辑删除对应通知在原接收人下的接收记录；仅限内部服务调用。
+     * @param removal 任务 ID、通知事件标识和原接收人
+     * @return 处理成功为 true，记录不存在也视为成功
+     */
+    @Inner
+    @PostMapping("/internal/flow-notice/remove")
+    @Operation(summary = "内部撤销流程站内信")
+    public R<Boolean> removeFlowNotice(@Valid @RequestBody FlowMessageRemovalDTO removal) {
+        return R.ok(sysMessageService.removeFlowNotice(removal));
+    }
+
 }

pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java

@@ -3,6 +3,7 @@ package com.pig4cloud.pigx.admin.service;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.extension.service.IService;
+import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageEmailDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageHookDTO;
 import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO;
@@ -121,4 +122,13 @@ public interface SysMessageService extends IService<SysMessageEntity> {
      */
     R sendNotice(MessageNoticeDTO noticeDTO);
 
+    /**
+     * 逻辑删除指定流程通知在原接收人下的接收记录，不删除消息正文和其他接收人的记录。
+     *
+     * @param removal 任务 ID、通知事件标识（须以 flow: 开头）和原接收人
+     * @return 处理成功为 true，记录不存在也视为成功
+     * @throws IllegalArgumentException 参数缺失或通知事件标识不属于流程通知
+     */
+    Boolean removeFlowNotice(FlowMessageRemovalDTO removal);
+
 }

pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java

@@ -654,4 +654,24 @@ public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMess
         return R.ok(messageVO.getId());
     }
 
+    @Override
+    public Boolean removeFlowNotice(FlowMessageRemovalDTO removal) {
+        if (Objects.isNull(removal) || Objects.isNull(removal.getUserId()) || StrUtil.isBlank(removal.getTaskId())
+                || !StrUtil.startWith(removal.getRequestKey(), "flow:")) {
+            throw new IllegalArgumentException("流程通知删除参数不完整");
+        }
+        // 租户过滤由现有 MyBatis 拦截器执行；事件键避免误删同一任务重新分配后的通知。
+        var messages = baseMapper.selectList(Wrappers.<SysMessageEntity>lambdaQuery()
+                .select(SysMessageEntity::getId)
+                .eq(SysMessageEntity::getBusinessType, "FLOW_TASK")
+                .eq(SysMessageEntity::getBusinessId, removal.getTaskId())
+                .eq(SysMessageEntity::getRequestKey, removal.getRequestKey()));
+        for (var message : messages) {
+            relationMapper.delete(Wrappers.<SysMessageRelationEntity>lambdaQuery()
+                    .eq(SysMessageRelationEntity::getMsgId, message.getId())
+                    .eq(SysMessageRelationEntity::getUserId, removal.getUserId()));
+        }
+        return Boolean.TRUE;
+    }
+
 }

pigx-upms/pigx-upms-biz/src/test/java/com/pig4cloud/pigx/admin/service/impl/SysMessageDeletionTests.java

@@ -4,6 +4,11 @@ import com.baomidou.mybatisplus.core.MybatisConfiguration;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
 import com.pig4cloud.pigx.admin.api.entity.SysMessageRelationEntity;
+import com.pig4cloud.pigx.admin.api.entity.SysMessageEntity;
+import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO;
+import com.pig4cloud.pigx.admin.mapper.SysMessageMapper;
+import org.springframework.test.util.ReflectionTestUtils;
+import java.util.List;
 import com.pig4cloud.pigx.admin.mapper.SysMessageRelationMapper;
 import com.pig4cloud.pigx.common.security.service.PigxUser;
 import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
@@ -35,5 +40,26 @@ class SysMessageDeletionTests {
         assertThat(condition.getValue().getSqlSegment()).contains("msg_id =", "user_id =");
         assertThat(condition.getValue().getParamNameValuePairs().values()).containsExactlyInAnyOrder(123L, 1001L);
         verifyNoMoreInteractions(relations);
+
+        // 流程撤销必须同时匹配任务、事件和接收人，不能误删新分配的通知。
+        TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "message"), SysMessageEntity.class);
+        var messages = mock(SysMessageMapper.class);
+        ReflectionTestUtils.setField(service, "baseMapper", messages);
+        var message = new SysMessageEntity();
+        message.setId(456L);
+        when(messages.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of(message));
+        var removal = new FlowMessageRemovalDTO();
+        removal.setTaskId("task-A");
+        removal.setRequestKey("flow:event-A");
+        removal.setUserId(1002L);
+        service.removeFlowNotice(removal);
+        var messageCondition = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        verify(messages).selectList(messageCondition.capture());
+        assertThat(messageCondition.getValue().getSqlSegment()).contains("business_type =", "business_id =", "request_key =");
+        assertThat(messageCondition.getValue().getParamNameValuePairs().values())
+                .containsExactlyInAnyOrder("FLOW_TASK", "task-A", "flow:event-A");
+        verify(relations, times(2)).delete(condition.capture());
+        assertThat(condition.getValue().getSqlSegment()).contains("msg_id =", "user_id =");
+        assertThat(condition.getValue().getParamNameValuePairs().values()).containsExactlyInAnyOrder(456L, 1002L);
     }
 }