暂停任务开发

fs-ai-call-task/src/main/java/com/fs/app/service/CallTaskService.java

@@ -28,6 +28,7 @@ public class CallTaskService {
 
     private final RedisCache redisCache2;
     private final CompanyWorkflowEngine companyWorkflowEngine;
+    private final ICompanyVoiceRoboticService companyVoiceRoboticService;
     @Autowired
     @Qualifier("cidWorkFlowExecutor")
     private Executor cidWorkFlowExecutor;
@@ -41,6 +42,8 @@ public class CallTaskService {
         String delayCallKeyPrefix = AiCallTaskNode.getDelayCallKeyPrefix(cidGroupNo,null) + "*";
         Collection<String> keys = redisCache2.keys(delayCallKeyPrefix);
         log.info("共扫描到 {} 个待处理键", keys.size());
+        // 本地缓存已查询的任务暂停状态，避免同一批次重复查询
+        Map<Long, Boolean> pausedCache = new ConcurrentHashMap<>();
         keys.parallelStream().forEach(key -> {
             try {
                 //doExec
@@ -49,17 +52,29 @@ public class CallTaskService {
                         ExecutionContext context = redisCache2.getCacheObject(key);
                         if (context == null) {
                             log.warn("工作流延时任务context为空，跳过 - key: {}", key);
+                            redisCache2.deleteObject(key);
+                            return;
+                        }
+                        // 任务暂停守卫检查（roboticId即CompanyVoiceRobotic.id，是实际暂停操作的目标）
+                        Long taskId = context.getVariable("roboticId", Long.class);
+                        if (taskId != null && pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id))) {
+                            // 延时key是时间分片前缀，下一分钟就不会再扫到，直接删除
+                            // 同步context信息到DB exec，供恢复时resumePausedInstances使用
+                            context.setVariable("callSource", "callTaskTimer");
+                            context.setVariable("_delayTargetNodeKey", context.getCurrentNodeKey());
+                            companyWorkflowEngine.updateExecVariables(context.getWorkflowInstanceId(), context.getVariables());
+                            log.info("任务已暂停，删除延时key并同步exec，等待恢复时从DB重建 - taskId: {}, key: {}", taskId, key);
+                            redisCache2.deleteObject(key);
                             return;
                         }
                         context.setVariable("callRedisKey", key);
                         context.setVariable("callSource", "callTaskTimer");
                         companyWorkflowEngine.timeDoExecute(context.getWorkflowInstanceId(), context.getCurrentNodeKey(), context.getVariables());
+                        redisCache2.deleteObject(key);
                     } catch (Exception e) {
                         log.error("处理工作流延时任务异常 - key: {}", key, e);
                     }
-                }, cidWorkFlowExecutor).thenRun(()->{
-                    redisCache2.deleteObject(key);
-                });
+                }, cidWorkFlowExecutor);
             } catch (Exception ex) {
                 log.error("处理工作流延时任务异常 - key: {}", key, ex);
             }

fs-cid-workflow/src/main/java/com/fs/app/service/CidWorkflowTaskService.java

@@ -1,5 +1,7 @@
 package com.fs.app.service;
 import com.fs.common.core.redis.RedisCache;
+import com.alibaba.fastjson.JSONObject;
+import com.fs.common.utils.StringUtils;
 import com.fs.company.domain.*;
 import com.fs.company.mapper.*;
 import com.fs.company.param.ExecutionContext;
@@ -29,6 +31,7 @@ public class CidWorkflowTaskService {
     Integer cidGroupNo;
     private final CompanyAiWorkflowExecMapper companyAiWorkflowExecMapper;
     private final CompanyWorkflowEngine companyWorkflowEngine;
+    private final ICompanyVoiceRoboticService companyVoiceRoboticService;
 
 
     private final RedisCache redisCache;
@@ -48,9 +51,20 @@ public class CidWorkflowTaskService {
         System.out.println(companyAiWorkflowExecs);
         log.info("runCidWorkflow得到租户id：{}",TenantHelper.getTenantId());
         if (null != companyAiWorkflowExecs && companyAiWorkflowExecs.size() > 0) {
+            // 本地缓存已查询的任务暂停状态，避免同一批次重复查询
+            Map<Long, Boolean> pausedCache = new HashMap<>();
             companyAiWorkflowExecs.forEach(exec -> {
 //                cidWorkFlowExecutor.execute(() -> {
                     try {
+                        // 任务暂停守卫检查（从 variables JSON 中提取 roboticId，即 CompanyVoiceRobotic.id）
+                        Long taskId = extractRoboticIdFromExec(exec);
+                        if (taskId != null) {
+                            boolean paused = pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id));
+                            if (paused) {
+                                log.debug("任务已暂停，跳过执行 - taskId: {}, execId: {}", taskId, exec.getId());
+                                return;
+                            }
+                        }
                         companyWorkflowEngine.executeNode(exec.getWorkflowInstanceId(), exec.getCurrentNodeKey());
                     } catch (Exception e) {
                         log.error("处理就绪任务异常 - exec: {}", exec, e);
@@ -67,9 +81,20 @@ public class CidWorkflowTaskService {
         List<CompanyAiWorkflowExec> companyAiWorkflowExecs = companyAiWorkflowExecMapper.selectExecListWithTimeAvailableByStatusAndGroupNo(ExecutionStatusEnum.PENDING.getValue(), cidGroupNo);
         log.info("activateTimeAvailableTask得到租户id：{}",TenantHelper.getTenantId());
         if (null != companyAiWorkflowExecs && companyAiWorkflowExecs.size() > 0) {
+            // 本地缓存已查询的任务暂停状态，避免同一批次重复查询
+            Map<Long, Boolean> pausedCache = new HashMap<>();
             companyAiWorkflowExecs.forEach(exec -> {
 //                cidWorkFlowExecutor.execute(() -> {
                     try {
+                        // 任务暂停守卫检查（从 variables JSON 中提取 roboticId，即 CompanyVoiceRobotic.id）
+                        Long taskId = extractRoboticIdFromExec(exec);
+                        if (taskId != null) {
+                            boolean paused = pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id));
+                            if (paused) {
+                                log.debug("任务已暂停，跳过执行 - taskId: {}, execId: {}", taskId, exec.getId());
+                                return;
+                            }
+                        }
                         companyWorkflowEngine.executeNode(exec.getWorkflowInstanceId(), exec.getCurrentNodeKey());
                     } catch (Exception e) {
                         log.error("处理就绪任务异常 - exec: {}", exec, e);
@@ -87,11 +112,23 @@ public class CidWorkflowTaskService {
         String s = String.format(AbstractWorkflowNode.CONTINUE_TIMER_EXECUTE_KEY_PREFIX, cidGroupNo);
         Collection<String> keys = redisCache.keys(s);
         log.info("定时服务执行共扫描到 {} 个待处理键", keys.size());
+        // 本地缓存已查询的任务暂停状态，避免同一批次重复查询
+        Map<Long, Boolean> pausedCache = new HashMap<>();
         // 使用 cidWorkFlowExecutor 替代 parallelStream，确保租户ID正确传递
         keys.forEach(key -> {
             cidWorkFlowExecutor.execute(() -> {
                 try {
                     ExecutionContext context = redisCache.getCacheObject(key);
+                    if (context == null) {
+                        redisCache.deleteObject(key);
+                        return;
+                    }
+                    // 任务暂停守卫检查（CONTINUE:TIMER key 不含时间分片，下次扫描还能找到，暂停时保留key）
+                    Long taskId = context.getVariable("roboticId", Long.class);
+                    if (taskId != null && pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id))) {
+                        log.info("任务已暂停，保留CONTINUE:TIMER key等待下次扫描 - taskId: {}, key: {}", taskId, key);
+                        return;
+                    }
                     CompanyAiWorkflowExec exec = companyAiWorkflowExecMapper.selectExecWithTimeAvailableByInstanceId(context.getWorkflowInstanceId());
                     if(null == exec){
                         return;
@@ -109,4 +146,22 @@ public class CidWorkflowTaskService {
         });
     }
 
+    /**
+     * 从 exec 记录的 variables JSON 中提取 roboticId（CompanyVoiceRobotic.id）
+     * 用于暂停守卫检查，fallback 到 exec.getBusinessKey()
+     */
+    private Long extractRoboticIdFromExec(CompanyAiWorkflowExec exec) {
+        if (StringUtils.isNotBlank(exec.getVariables())) {
+            try {
+                JSONObject vars = JSONObject.parseObject(exec.getVariables());
+                if (vars != null && vars.containsKey("roboticId")) {
+                    return vars.getLong("roboticId");
+                }
+            } catch (Exception e) {
+                log.warn("extractRoboticIdFromExec解析variables失败 - execId: {}", exec.getId(), e);
+            }
+        }
+        return exec.getBusinessKey();
+    }
+
 }

fs-company/src/main/java/com/fs/company/controller/company/CompanyVoiceRoboticController.java

@@ -16,6 +16,7 @@ import com.fs.common.core.domain.AjaxResult;
 import com.fs.common.core.domain.R;
 import com.fs.common.core.page.TableDataInfo;
 import com.fs.common.enums.BusinessType;
+import com.fs.common.utils.SecurityUtils;
 import com.fs.common.utils.ServletUtils;
 import com.fs.common.utils.StringUtils;
 import com.fs.common.utils.poi.ExcelUtil;
@@ -35,6 +36,7 @@ import com.fs.company.vo.WorkflowExecRecordVo;
 import com.fs.crm.service.ICrmCustomerPropertyService;
 import com.fs.framework.security.LoginUser;
 import com.fs.framework.service.TokenService;
+import com.fs.wxcid.utils.TenantHelper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -354,8 +356,10 @@ public class CompanyVoiceRoboticController extends BaseController
         });
     }
 
+    @Log(title = "外呼任务暂停操作", businessType = BusinessType.UPDATE)
     @PostMapping("/pauseRoboticActive")
     public R pauseRoboticActive(@RequestBody PauseRoboticActiveParam param){
+        TenantHelper.setTenantId(SecurityUtils.getTenantId());
         return companyVoiceRoboticService.pauseRoboticActive(param);
     }
 }

fs-service/src/main/java/com/fs/company/param/ExecutionContext.java

@@ -62,14 +62,34 @@ public class ExecutionContext {
 
 
     /**
-     * 获取本地变量
+     * 获取本地变量（支持数值类型兼容转换，如 Integer → Long）
      */
     public <T> T getVariable(String key, Class<T> type) {
         Object value = this.variables.get(key);
         if (value == null) {
             return null;
         }
-        return type.isInstance(value) ? type.cast(value) : null;
+        if (type.isInstance(value)) {
+            return type.cast(value);
+        }
+        // 数值类型兼容转换：JSON反序列化可能将数字存为Integer，但调用方请求Long
+        if (value instanceof Number) {
+            Number num = (Number) value;
+            if (type == Long.class || type == long.class) {
+                return type.cast(num.longValue());
+            } else if (type == Integer.class || type == int.class) {
+                return type.cast(num.intValue());
+            } else if (type == Double.class || type == double.class) {
+                return type.cast(num.doubleValue());
+            } else if (type == Float.class || type == float.class) {
+                return type.cast(num.floatValue());
+            }
+        }
+        // String类型兼容
+        if (type == String.class) {
+            return type.cast(value.toString());
+        }
+        return null;
     }
 
     /**
@@ -80,14 +100,32 @@ public class ExecutionContext {
     }
 
     /**
-     * 获取全局变量
+     * 获取全局变量（支持数值类型兼容转换）
      */
     public <T> T getGlobalVariable(String key, Class<T> type) {
         Object value = this.globalVariables.get(key);
         if (value == null) {
             return null;
         }
-        return type.isInstance(value) ? type.cast(value) : null;
+        if (type.isInstance(value)) {
+            return type.cast(value);
+        }
+        if (value instanceof Number) {
+            Number num = (Number) value;
+            if (type == Long.class || type == long.class) {
+                return type.cast(num.longValue());
+            } else if (type == Integer.class || type == int.class) {
+                return type.cast(num.intValue());
+            } else if (type == Double.class || type == double.class) {
+                return type.cast(num.doubleValue());
+            } else if (type == Float.class || type == float.class) {
+                return type.cast(num.floatValue());
+            }
+        }
+        if (type == String.class) {
+            return type.cast(value.toString());
+        }
+        return null;
     }
     public ExecutionContext clone() {
         ExecutionContext cloned = new ExecutionContext();

fs-service/src/main/java/com/fs/company/service/CompanyWorkflowEngine.java

@@ -62,6 +62,13 @@ public interface CompanyWorkflowEngine {
      * @param inputData
      */
     void timeDoExecute(String workflowInstanceId, String nodeKey, Map<String, Object> inputData);
+
+    /**
+     * 更新exec记录的variables字段（用于暂停时同步延时上下文到DB）
+     * @param workflowInstanceId
+     * @param variables 最新的variables map
+     */
+    void updateExecVariables(String workflowInstanceId, Map<String, Object> variables);
     /**
      * 创建sip任务
      * @param roboticId

fs-service/src/main/java/com/fs/company/service/ICompanyVoiceRoboticService.java

@@ -116,4 +116,13 @@ public interface ICompanyVoiceRoboticService extends IService<CompanyVoiceRoboti
     void addNewExec4Task(Long taskId,Long crmCustomerId,String traceId);
 
     R pauseRoboticActive(PauseRoboticActiveParam param);
+
+    /**
+     * 判断任务是否处于暂停状态
+     * 优先从Redis读取，Redis无数据则查DB并回填
+     *
+     * @param taskId 任务ID
+     * @return true=暂停中 false=非暂停
+     */
+    boolean isTaskPaused(Long taskId);
 }

fs-service/src/main/java/com/fs/company/service/impl/CompanyVoiceRoboticCallLogCallphoneServiceImpl.java

@@ -18,12 +18,14 @@ import com.fs.common.utils.DateUtils;
 import com.fs.common.utils.StringUtils;
 import com.fs.common.utils.spring.SpringUtils;
 import com.fs.company.domain.*;
+import com.fs.company.mapper.CompanyAiWorkflowExecMapper;
 import com.fs.company.mapper.CompanyVoiceRoboticBusinessMapper;
 import com.fs.company.mapper.CompanyVoiceRoboticCallLogCallphoneMapper;
 import com.fs.company.mapper.CompanyVoiceRoboticCalleesMapper;
 import com.fs.company.mapper.CompanyWxAccountMapper;
 import com.fs.company.service.CompanyWorkflowEngine;
 import com.fs.company.service.ICompanyVoiceRoboticCallLogCallphoneService;
+import com.fs.company.service.ICompanyVoiceRoboticService;
 import com.fs.company.vo.CidConfigVO;
 import com.fs.company.vo.CompanyVoiceRoboticCallLogCallPhoneVO;
 import com.fs.company.vo.CompanyVoiceRoboticCallLogCount;
@@ -38,6 +40,7 @@ import com.fs.wxcid.utils.TenantHelper;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
 import org.springframework.security.core.context.SecurityContextHolder;
@@ -88,6 +91,11 @@ public class CompanyVoiceRoboticCallLogCallphoneServiceImpl extends ServiceImpl<
 
     @Autowired
     SysDictTypeServiceImpl sysDictTypeService;
+    @Autowired
+    @Lazy
+    ICompanyVoiceRoboticService companyVoiceRoboticService;
+    @Autowired
+    CompanyAiWorkflowExecMapper companyAiWorkflowExecMapper;
     /**
      * 查询调用日志_ai打电话
      *
@@ -372,14 +380,21 @@ public class CompanyVoiceRoboticCallLogCallphoneServiceImpl extends ServiceImpl<
                         JSONObject bizJson = JSONObject.parseObject(result.getBizJson());
                         JSONObject userData = JSONObject.parseObject(redisCache2.getCacheObject(EASYCALL_WORKFLOW_REDIS_KEY +  bizJson.getString("callBackUuid")), JSONObject.class);
                         if (null != userData && userData.containsKey("callBackUuid") && userData.containsKey("workflowInstanceId") && userData.containsKey("nodeKey")) {
-                            Map<String, Object> param = new HashMap<>();
-                            param.put("callBackUuid", userData.getString("callBackUuid"));
-                            param.put("callSource", "callBack");
-                            CompletableFuture.runAsync(() -> {
-                                companyWorkflowEngine.resumeFromBlockingNode(userData.getString("workflowInstanceId"), userData.getString("nodeKey"), param);
-                            }, cidWorkFlowExecutor).thenRun(() -> {
-                                redisCache2.deleteObject(EASYCALL_WORKFLOW_REDIS_KEY +  bizJson.getString("callBackUuid"));
-                            });
+                            // 暂停检查：任务暂停中则标记回调已到达但不触发工作流继续执行
+                            if (companyVoiceRoboticService.isTaskPaused(callees.getRoboticId())) {
+                                log.info("任务暂停中，EasyCall外呼回调数据已保存但不触发工作流继续执行 - taskId: {}, workflowInstanceId: {}",
+                                        callees.getRoboticId(), userData.getString("workflowInstanceId"));
+                                markCallbackReceivedInVariables(userData.getString("workflowInstanceId"));
+                            } else {
+                                Map<String, Object> param = new HashMap<>();
+                                param.put("callBackUuid", userData.getString("callBackUuid"));
+                                param.put("callSource", "callBack");
+                                CompletableFuture.runAsync(() -> {
+                                    companyWorkflowEngine.resumeFromBlockingNode(userData.getString("workflowInstanceId"), userData.getString("nodeKey"), param);
+                                }, cidWorkFlowExecutor).thenRun(() -> {
+                                    redisCache2.deleteObject(EASYCALL_WORKFLOW_REDIS_KEY +  bizJson.getString("callBackUuid"));
+                                });
+                            }
                         }
                         redisCache2.deleteObject(bizJson.getString("callBackUuid"));
                     }
@@ -409,6 +424,34 @@ public class CompanyVoiceRoboticCallLogCallphoneServiceImpl extends ServiceImpl<
 
     }
 
+    /**
+     * 标记工作流实例的variables中回调已到达（任务暂停时使用，恢复时会检查此标记）
+     */
+    private void markCallbackReceivedInVariables(String workflowInstanceId) {
+        try {
+            CompanyAiWorkflowExec exec = companyAiWorkflowExecMapper.selectByWorkflowInstanceId(workflowInstanceId);
+            if (exec == null) {
+                log.warn("markCallbackReceived: 未找到工作流实例 - workflowInstanceId: {}", workflowInstanceId);
+                return;
+            }
+            String variablesJson = exec.getVariables();
+            JSONObject variables;
+            if (StringUtils.isNotBlank(variablesJson)) {
+                variables = JSONObject.parseObject(variablesJson);
+            } else {
+                variables = new JSONObject();
+            }
+            variables.put("pause_callback_received", true);
+            CompanyAiWorkflowExec update = new CompanyAiWorkflowExec();
+            update.setWorkflowInstanceId(workflowInstanceId);
+            update.setVariables(variables.toJSONString());
+            companyAiWorkflowExecMapper.updateByWorkflowInstanceId(update);
+            log.info("markCallbackReceived: 已标记回调到达 - workflowInstanceId: {}", workflowInstanceId);
+        } catch (Exception e) {
+            log.error("markCallbackReceived: 标记回调到达失败 - workflowInstanceId: {}", workflowInstanceId, e);
+        }
+    }
+
     @Async("callLogExcutor")
     public void asyncInsertCompanyVoiceRoboticCallLogBatch(List<CompanyVoiceRoboticCallLogCallphone> list) {
         try {

fs-service/src/main/java/com/fs/company/service/impl/CompanyVoiceRoboticServiceImpl.java

@@ -23,6 +23,7 @@ import com.fs.common.core.redis.RedisCacheT;
 import com.fs.common.exception.CustomException;
 import com.fs.common.exception.base.BaseException;
 import com.fs.common.service.impl.SmsServiceImpl;
+import com.fs.company.service.impl.call.EasyCallTaskControlService;
 import com.fs.common.utils.*;
 import com.fs.common.utils.spring.SpringUtils;
 import com.fs.company.domain.*;
@@ -140,6 +141,9 @@ public class CompanyVoiceRoboticServiceImpl extends ServiceImpl<CompanyVoiceRobo
 
     private final ICrmCustomerAnalyzeService crmCustomerAnalyzeService;
 
+    private final EasyCallTaskControlService easyCallTaskControlService;
+    private final CompanySiptaskInfoMapper companySiptaskInfoMapper;
+
     /** EasyCall intent 意向度重试队列 Redis key 前缀，value 为已重试次数 */
     private static final String EASYCALL_INTENT_RETRY_KEY = "easycall:intent:retry:";
     /** intent 意向度等待重试最大次数（每次间隔约30秒，最多等待 5*30=150秒） */
@@ -1960,17 +1964,267 @@ public class CompanyVoiceRoboticServiceImpl extends ServiceImpl<CompanyVoiceRobo
     public R pauseRoboticActive(PauseRoboticActiveParam param) {
         //暂停任务
         if (ACTIVE_TYPE_PAUSE.equals(param.getActiveType())) {
+            CompanyVoiceRobotic robotic = selectCompanyVoiceRoboticById(param.getTaskId());
+            if (robotic == null || robotic.getTaskStatus() != 1) {
+                return R.error("任务不在执行中状态，无法暂停");
+            }
+            robotic.setTaskStatus(2);
+            updateCompanyVoiceRobotic(robotic);
+            // 更新Redis缓存
+            redisCache2.setCacheObject("task:status:" + param.getTaskId(), 2);
+            // 调用EasyCall暂停
+            pauseEasyCallTask(robotic);
+            return R.ok("暂停成功");
+        }
+        //恢复任务继续进入可运行
+        else if (ACTIVE_TYPE_CONTINUE.equals(param.getActiveType())) {
+            CompanyVoiceRobotic robotic = selectCompanyVoiceRoboticById(param.getTaskId());
+            if (robotic == null || robotic.getTaskStatus() != 2) {
+                return R.error("任务不在中断状态，无法继续");
+            }
+            robotic.setTaskStatus(1);
+            updateCompanyVoiceRobotic(robotic);
+            // 更新Redis缓存
+            redisCache2.setCacheObject("task:status:" + param.getTaskId(), 1);
+            // 调用EasyCall恢复
+            resumeEasyCallTask(robotic);
+            // 异步执行恢复扫描（分批处理，避免阻塞接口）
+            // 通过SpringUtils获取代理对象调用，确保@Async生效
+            SpringUtils.getBean(CompanyVoiceRoboticServiceImpl.class).resumePausedInstances(param.getTaskId());
+            return R.ok("继续成功");
+        }
 
-            // 暂停任务更新
+        return R.error("操作类型无效");
+    }
 
-            // 暂停任务创建的三方外呼任务
+    /**
+     * 暂停该任务关联的所有EasyCall外呼任务
+     */
+    private void pauseEasyCallTask(CompanyVoiceRobotic robotic) {
+        try {
+            CompanySiptaskInfo query = new CompanySiptaskInfo();
+            query.setTaskId(robotic.getId());
+            List<CompanySiptaskInfo> sipTasks = companySiptaskInfoMapper.selectCompanySiptaskInfoList(query);
+            if (sipTasks != null && !sipTasks.isEmpty()) {
+                for (CompanySiptaskInfo sipTask : sipTasks) {
+                    if (sipTask.getBatchId() != null) {
+                        easyCallTaskControlService.pauseTask(sipTask.getBatchId());
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error("暂停EasyCall任务异常, roboticId={}", robotic.getId(), e);
+        }
+    }
 
+    /**
+     * 恢复该任务关联的所有EasyCall外呼任务
+     */
+    private void resumeEasyCallTask(CompanyVoiceRobotic robotic) {
+        try {
+            CompanySiptaskInfo query = new CompanySiptaskInfo();
+            query.setTaskId(robotic.getId());
+            List<CompanySiptaskInfo> sipTasks = companySiptaskInfoMapper.selectCompanySiptaskInfoList(query);
+            if (sipTasks != null && !sipTasks.isEmpty()) {
+                for (CompanySiptaskInfo sipTask : sipTasks) {
+                    if (sipTask.getBatchId() != null) {
+                        easyCallTaskControlService.resumeTask(sipTask.getBatchId());
+                    }
+                }
+            }
+        } catch (Exception e) {
+            log.error("恢复EasyCall任务异常, roboticId={}", robotic.getId(), e);
         }
-        //恢复任务继续进入可运行
-        else if (ACTIVE_TYPE_CONTINUE.equals(param.getActiveType())) {
+    }
+
+    /**
+     * 异步恢复暂停期间被阻塞的工作流实例
+     * 查询该任务下所有PAUSED(4)/WAITING(5)状态的工作流实例，分批处理
+     *
+     * @param taskId 任务ID
+     */
+    @Async("cidWorkFlowExecutor")
+    public void resumePausedInstances(Long taskId) {
+        try {
+            log.info("开始恢复暂停实例, taskId={}", taskId);
+
+            // 先查询该任务下所有业务记录的ID（businessKey存的是CompanyVoiceRoboticBusiness.id，而非taskId）
+            LambdaQueryWrapper<CompanyVoiceRoboticBusiness> bizWrapper = new LambdaQueryWrapper<>();
+            bizWrapper.eq(CompanyVoiceRoboticBusiness::getRoboticId, taskId)
+                    .select(CompanyVoiceRoboticBusiness::getId);
+            List<CompanyVoiceRoboticBusiness> bizList = companyVoiceRoboticBusinessMapper.selectList(bizWrapper);
+            if (bizList == null || bizList.isEmpty()) {
+                log.info("未找到业务记录, taskId={}", taskId);
+                return;
+            }
+            List<Long> businessIds = bizList.stream().map(CompanyVoiceRoboticBusiness::getId).collect(java.util.stream.Collectors.toList());
+
+            // 查询该任务下所有PAUSED和WAITING状态的工作流实例
+            LambdaQueryWrapper<CompanyAiWorkflowExec> queryWrapper = new LambdaQueryWrapper<>();
+            queryWrapper.in(CompanyAiWorkflowExec::getBusinessKey, businessIds)
+                    .in(CompanyAiWorkflowExec::getStatus,
+                            ExecutionStatusEnum.PAUSED.getValue(),
+                            ExecutionStatusEnum.WAITING.getValue(),
+                            ExecutionStatusEnum.PENDING.getValue());
+            List<CompanyAiWorkflowExec> execList = companyAiWorkflowExecMapper.selectList(queryWrapper);
+
+            if (execList == null || execList.isEmpty()) {
+                log.info("无需恢复的暂停实例, taskId={}", taskId);
+                return;
+            }
+
+            log.info("找到待恢复实例 {} 个, taskId={}", execList.size(), taskId);
 
+            int batchSize = 50;
+            for (int i = 0; i < execList.size(); i++) {
+                CompanyAiWorkflowExec exec = execList.get(i);
+                try {
+                    processResumeInstance(exec);
+                } catch (Exception e) {
+                    log.error("恢复实例异常, instanceId={}", exec.getWorkflowInstanceId(), e);
+                }
+
+                // 每处理50个暂停2秒，避免压力过大
+                if ((i + 1) % batchSize == 0 && i + 1 < execList.size()) {
+                    try {
+                        Thread.sleep(2000);
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                        log.warn("恢复暂停实例被中断, taskId={}", taskId);
+                        return;
+                    }
+                }
+            }
+
+            log.info("恢复暂停实例完成, taskId={}, count={}", taskId, execList.size());
+        } catch (Exception e) {
+            log.error("恢复暂停实例整体异常, taskId={}", taskId, e);
+        }
+    }
+
+    /**
+     * 处理单个待恢复的工作流实例
+     */
+    private void processResumeInstance(CompanyAiWorkflowExec exec) {
+        Integer status = exec.getStatus();
+        Integer nodeType = exec.getCurrentNodeType();
+        String instanceId = exec.getWorkflowInstanceId();
+        String nodeKey = exec.getCurrentNodeKey();
+
+        // exec.getVariables() 存储的是 variables Map 的 JSON（不是完整 ExecutionContext）
+        Map<String, Object> inputData = new HashMap<>();
+        if (StringUtils.isNotEmpty(exec.getVariables())) {
+            try {
+                Map<String, Object> parsed = JSON.parseObject(exec.getVariables(), Map.class);
+                if (parsed != null) {
+                    inputData = parsed;
+                }
+            } catch (Exception e) {
+                log.error("反序列化variables失败, instanceId={}", instanceId, e);
+                return;
+            }
         }
 
-        return R.ok("操作成功");
+        // PAUSED状态处理
+        if (ExecutionStatusEnum.PAUSED.getValue() == status) {
+            // AI_CALL_TASK + PAUSED：检查是否有回调
+            if (NodeTypeEnum.AI_CALL_TASK.getValue().equals(nodeType)) {
+                if (hasCallbackReceived(inputData)) {
+                    log.info("恢复AI外呼PAUSED实例, instanceId={}", instanceId);
+                    companyWorkflowEngine.resumeFromBlockingNode(instanceId, nodeKey, inputData);
+                }
+            }
+            // AI_ADD_WX_TASK / AI_QW_ADD_WX_TASK + PAUSED：检查是否有回调
+            else if (NodeTypeEnum.AI_ADD_WX_TASK.getValue().equals(nodeType)
+                    || NodeTypeEnum.AI_QW_ADD_WX_TASK.getValue().equals(nodeType)
+                    || NodeTypeEnum.AI_ADD_WX_TASK_NEW.getValue().equals(nodeType)) {
+                if (hasCallbackReceived(inputData)) {
+                    log.info("恢复加微PAUSED实例, instanceId={}, nodeType={}", instanceId, nodeType);
+                    companyWorkflowEngine.resumeFromBlockingNode(instanceId, nodeKey, inputData);
+                }
+            }
+        }
+        // WAITING状态处理
+        else if (ExecutionStatusEnum.WAITING.getValue() == status) {
+            // 优先从 variables 中获取延时目标节点key（暂停时同步存入的）
+            String targetNodeKey = inputData.containsKey("_delayTargetNodeKey")
+                    ? (String) inputData.get("_delayTargetNodeKey") : nodeKey;
+            // AI_CALL_TASK + WAITING：延时已过期，触发timeDoExecute
+            if (NodeTypeEnum.AI_CALL_TASK.getValue().equals(nodeType)) {
+                log.info("恢复AI外呼WAITING实例, instanceId={}, targetNodeKey={}", instanceId, targetNodeKey);
+                companyWorkflowEngine.timeDoExecute(instanceId, targetNodeKey, inputData);
+            }
+            // AI_ADD_WX_TASK / AI_QW_ADD_WX_TASK / AI_ADD_WX_TASK_NEW + WAITING
+            else if (NodeTypeEnum.AI_ADD_WX_TASK.getValue().equals(nodeType)
+                    || NodeTypeEnum.AI_QW_ADD_WX_TASK.getValue().equals(nodeType)
+                    || NodeTypeEnum.AI_ADD_WX_TASK_NEW.getValue().equals(nodeType)) {
+                if (hasCallbackReceived(inputData)) {
+                    log.info("恢复加微WAITING实例, instanceId={}, nodeType={}", instanceId, nodeType);
+                    companyWorkflowEngine.resumeFromBlockingNode(instanceId, nodeKey, inputData);
+                } else {
+                    // 无回调但有延时目标节点（加微延时场景）
+                    if (inputData.containsKey("_delayTargetNodeKey")) {
+                        log.info("恢复加微WAITING延时实例, instanceId={}, targetNodeKey={}", instanceId, targetNodeKey);
+                        companyWorkflowEngine.timeDoExecute(instanceId, targetNodeKey, inputData);
+                    }
+                }
+            }
+        }
+        // PENDING状态处理：重建Redis CONTINUE:TIMER:EXECUTE key
+        else if (ExecutionStatusEnum.PENDING.getValue() == status) {
+            if (!inputData.isEmpty()) {
+                ExecutionContext resumeContext = new ExecutionContext();
+                resumeContext.setWorkflowInstanceId(instanceId);
+                resumeContext.setCurrentNodeKey(nodeKey);
+                resumeContext.setVariables(inputData);
+                resumeContext.setBusinessId(exec.getBusinessKey());
+                log.info("重建PENDING实例Redis key, instanceId={}", instanceId);
+                rebuildContinueTimerKey(exec, resumeContext);
+            }
+        }
+    }
+
+    /**
+     * 检查variables中是否有回调标记
+     */
+    private boolean hasCallbackReceived(Map<String, Object> variables) {
+        if (variables == null) return false;
+        Object flag = variables.get("pause_callback_received");
+        return Boolean.TRUE.equals(flag) || "true".equals(String.valueOf(flag));
+    }
+
+    /**
+     * 重建CONTINUE:TIMER:EXECUTE Redis key
+     */
+    private void rebuildContinueTimerKey(CompanyAiWorkflowExec exec, ExecutionContext context) {
+        try {
+            Integer groupNo = exec.getCidGroupNo();
+            Date now = new Date();
+            int hour = now.getHours();
+            int minute = now.getMinutes();
+            String redisKey = "CONTINUE:TIMER:EXECUTE:" + groupNo + ":" + hour + ":" + minute;
+            String contextJson = JSON.toJSONString(context);
+            redisCache2.setCacheObject(redisKey + ":" + exec.getWorkflowInstanceId(), contextJson);
+        } catch (Exception e) {
+            log.error("重建CONTINUE:TIMER:EXECUTE Redis key异常, instanceId={}", exec.getWorkflowInstanceId(), e);
+        }
+    }
+
+    @Override
+    public boolean isTaskPaused(Long taskId) {
+        if (taskId == null) return false;
+        // 优先从Redis读取
+        Integer status = redisCache2.getCacheObject("task:status:" + taskId);
+        if (status != null) {
+            return status == 2;
+        }
+        // Redis无数据则查DB
+        CompanyVoiceRobotic robotic = companyVoiceRoboticMapper.selectCompanyVoiceRoboticById(taskId);
+        if (robotic != null) {
+            // 回填Redis
+            redisCache2.setCacheObject("task:status:" + taskId, robotic.getTaskStatus());
+            return robotic.getTaskStatus() != null && robotic.getTaskStatus() == 2;
+        }
+        return false;
     }
 }

fs-service/src/main/java/com/fs/company/service/impl/CompanyWorkflowEngineImpl.java

@@ -623,4 +623,18 @@ public class CompanyWorkflowEngineImpl implements CompanyWorkflowEngine {
         return null;
     }
 
+    @Override
+    public void updateExecVariables(String workflowInstanceId, Map<String, Object> variables) {
+        try {
+            CompanyAiWorkflowExec update = new CompanyAiWorkflowExec();
+            update.setWorkflowInstanceId(workflowInstanceId);
+            update.setVariables(objectMapper.writeValueAsString(variables));
+            update.setLastUpdateTime(LocalDateTime.now());
+            currentExecutionMapper.updateByWorkflowInstanceId(update);
+            log.info("更新exec variables成功 - workflowInstanceId: {}", workflowInstanceId);
+        } catch (Exception e) {
+            log.error("更新exec variables失败 - workflowInstanceId: {}", workflowInstanceId, e);
+        }
+    }
+
 }

fs-service/src/main/java/com/fs/company/service/impl/call/EasyCallTaskControlService.java

@@ -0,0 +1,39 @@
+package com.fs.company.service.impl.call;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+/**
+ * EasyCallCenter365 任务暂停/恢复控制服务
+ * <p>
+ * 通过HTTP接口调用EasyCall平台，实现对第三方外呼任务的暂停和恢复操作。
+ * 当前为Mock实现，待提供真实接口地址后替换。
+ */
+@Service
+@Slf4j
+public class EasyCallTaskControlService {
+
+    @Value("${easycall.api.base-url:http://localhost:8870}")
+    private String easyCallBaseUrl;
+
+    /**
+     * 暂停EasyCall外呼任务
+     *
+     * @param batchId EasyCall任务批次ID
+     */
+    public void pauseTask(Long batchId) {
+        // TODO: 待提供真实接口地址后替换
+        log.info("[Mock] 调用EasyCall暂停任务接口: batchId={}", batchId);
+    }
+
+    /**
+     * 恢复EasyCall外呼任务
+     *
+     * @param batchId EasyCall任务批次ID
+     */
+    public void resumeTask(Long batchId) {
+        // TODO: 待提供真实接口地址后替换
+        log.info("[Mock] 调用EasyCall恢复任务接口: batchId={}", batchId);
+    }
+}

fs-service/src/main/java/com/fs/company/service/impl/call/node/AbstractWorkflowNode.java

@@ -9,6 +9,7 @@ import com.fs.common.utils.spring.SpringUtils;
 import com.fs.company.domain.*;
 import com.fs.company.mapper.*;
 import com.fs.company.param.ExecutionContext;
+import com.fs.company.service.ICompanyVoiceRoboticService;
 import com.fs.company.service.IWorkflowNode;
 import com.fs.company.vo.ExecutionResult;
 import com.fs.enums.ExecutionStatusEnum;
@@ -45,6 +46,7 @@ public abstract class AbstractWorkflowNode implements IWorkflowNode {
     public static final WorkflowNodeFactory workflowNodeFactory = SpringUtils.getBean(WorkflowNodeFactory.class);
     public static final ObjectMapper objectMapper = new ObjectMapper();
     public static final RedissonClient redissonClient = SpringUtils.getBean(RedissonClient.class);
+    public static final ICompanyVoiceRoboticService companyVoiceRoboticService = SpringUtils.getBean(ICompanyVoiceRoboticService.class);
     protected static final String NODE_EXEC_LOCK_PREFIX = "node_exec_lock_";
     protected static final String CONTINUE_TIMER_EXECUTE_KEY = "CONTINUE:TIMER:EXECUTE:%s:%s:%s:";
     public static final String CONTINUE_TIMER_EXECUTE_KEY_PREFIX = "CONTINUE:TIMER:EXECUTE:%s:*";
@@ -108,6 +110,13 @@ public abstract class AbstractWorkflowNode implements IWorkflowNode {
 
     @Override
     public ExecutionResult continueExecute(ExecutionContext context) {
+        // 暂停守卫：任务暂停时不处理继续请求
+        Long continueTaskId = getTaskIdFromContext(context);
+        if (isTaskPaused(continueTaskId)) {
+            log.info("任务已暂停，跳过continueExecute - taskId: {}, workflowInstanceId: {}, nodeKey: {}",
+                    continueTaskId, context.getWorkflowInstanceId(), nodeKey);
+            return null;
+        }
         if(!runnable(context) && getType() != NodeTypeEnum.END){
             log.info("当前流程已到达结束节点，节点继续执行失败：- {},- {} -,{}" , nodeName, nodeKey, context.getWorkflowInstanceId());
             return null;
@@ -477,6 +486,23 @@ public abstract class AbstractWorkflowNode implements IWorkflowNode {
         return context;
     }
 
+    /**
+     * 创建带businessId(roboticId)的执行上下文，用于延时任务中暂停守卫检查
+     */
+    protected ExecutionContext createExecutionContextWithBusinessId(String workflowInstanceId, String nodeKey) {
+        ExecutionContext context = createExecutionContext(workflowInstanceId, nodeKey);
+        try {
+            CompanyVoiceRoboticBusiness business = companyVoiceRoboticBusinessMapper
+                    .selectCompanyVoiceRoboticBusinessByWorkflowInstanceId(workflowInstanceId);
+            if (business != null) {
+                context.setBusinessId(business.getRoboticId());
+            }
+        } catch (Exception e) {
+            log.warn("createExecutionContextWithBusinessId获取roboticId失败 - workflowInstanceId: {}", workflowInstanceId, e);
+        }
+        return context;
+    }
+
     protected void execPointNextNode(ExecutionContext context) {
         try {
             CompanyAiWorkflowExec update = new CompanyAiWorkflowExec();
@@ -499,6 +525,14 @@ public abstract class AbstractWorkflowNode implements IWorkflowNode {
         if (StringUtils.isBlank(edge.getTargetNodeKey())) {
             return;
         }
+        // 暂停守卫：任务暂停时不触发下一节点执行，将实例状态设为READY并指向下一节点
+        Long nextTaskId = getTaskIdFromContext(context);
+        if (isTaskPaused(nextTaskId)) {
+            log.info("任务已暂停，runNextNode中止执行 - taskId: {}, workflowInstanceId: {}, nextNodeKey: {}",
+                    nextTaskId, context.getWorkflowInstanceId(), edge.getTargetNodeKey());
+            updateExecToReady(context, edge.getTargetNodeKey());
+            return;
+        }
         ExecutionContext nextContext = context.clone();
         CompanyWorkflowNode nextNode = getNodeByKey(edge.getTargetNodeKey());
         nextContext.setCurrentNodeKey(nextNode.getNodeKey());
@@ -550,4 +584,55 @@ public abstract class AbstractWorkflowNode implements IWorkflowNode {
     public static String getContinueTimerExecuteKey(Integer groupNo, LocalTime time) {
         return String.format(CONTINUE_TIMER_EXECUTE_KEY, groupNo,time.getHour(), time.getMinute());
     }
+
+    /**
+     * 从执行上下文中获取任务ID（roboticId）
+     * 通过workflowInstanceId查询业务关联表获取
+     */
+    protected Long getTaskIdFromContext(ExecutionContext context) {
+        try {
+            CompanyVoiceRoboticBusiness business = companyVoiceRoboticBusinessMapper
+                    .selectCompanyVoiceRoboticBusinessByWorkflowInstanceId(context.getWorkflowInstanceId());
+            if (business != null) {
+                return business.getRoboticId();
+            }
+        } catch (Exception e) {
+            log.warn("获取taskId失败 - workflowInstanceId: {}", context.getWorkflowInstanceId(), e);
+        }
+        return null;
+    }
+
+    /**
+     * 判断任务是否处于暂停状态
+     */
+    protected boolean isTaskPaused(Long taskId) {
+        if (taskId == null) {
+            return false;
+        }
+        return companyVoiceRoboticService.isTaskPaused(taskId);
+    }
+
+    /**
+     * 将工作流实例状态设为READY并指向下一节点（暂停守卫使用）
+     * 恢复后CidTask定时扫描会自然拾取READY状态的实例继续执行
+     */
+    protected void updateExecToReady(ExecutionContext context, String nextNodeKey) {
+        try {
+            CompanyWorkflowNode nextNode = getNodeByKey(nextNodeKey);
+            CompanyAiWorkflowExec update = new CompanyAiWorkflowExec();
+            update.setWorkflowInstanceId(context.getWorkflowInstanceId());
+            update.setStatus(ExecutionStatusEnum.READY.getValue());
+            update.setCurrentNodeKey(nextNodeKey);
+            update.setCurrentNodeName(nextNode.getNodeName());
+            update.setCurrentNodeType(NodeTypeEnum.fromCode(nextNode.getNodeType()).getValue());
+            update.setLastUpdateTime(LocalDateTime.now());
+            update.setVariables(objectMapper.writeValueAsString(context.getVariables()));
+            companyAiWorkflowExecMapper.updateByWorkflowInstanceId(update);
+            log.info("任务暂停，工作流实例已设为READY状态 - workflowInstanceId: {}, nextNodeKey: {}",
+                    context.getWorkflowInstanceId(), nextNodeKey);
+        } catch (Exception e) {
+            log.error("更新工作流为READY状态失败 - workflowInstanceId: {}, nextNodeKey: {}",
+                    context.getWorkflowInstanceId(), nextNodeKey, e);
+        }
+    }
 }

fs-service/src/main/java/com/fs/company/service/impl/call/node/AiAddWxTaskNode.java

@@ -13,6 +13,8 @@ import com.fs.company.param.ExecutionContext;
 import com.fs.company.vo.AiAddWxConfigVO;
 import com.fs.company.vo.AiCallWorkflowConditionVo;
 import com.fs.company.vo.ExecutionResult;
+
+import java.util.concurrent.TimeUnit;
 import com.fs.enums.ExecutionStatusEnum;
 import com.fs.enums.NodeTypeEnum;
 import lombok.extern.slf4j.Slf4j;
@@ -256,7 +258,7 @@ public class AiAddWxTaskNode extends AbstractWorkflowNode {
      * @param workflowInstanceId
      */
     public void doneAddwx(String workflowInstanceId) {
-        ExecutionContext context = createExecutionContext(workflowInstanceId, nodeKey);
+        ExecutionContext context = createExecutionContextWithBusinessId(workflowInstanceId, nodeKey);
         context.setVariable("lastNodeKey", nodeKey);
         //启动定时节点倒计时
         CompanyAiWorkflowExec exec = companyAiWorkflowExecMapper.selectByWorkflowInstanceId(context.getWorkflowInstanceId());
@@ -289,7 +291,7 @@ public class AiAddWxTaskNode extends AbstractWorkflowNode {
                     String redisKey = getDelayAddWxKeyPrefix(exec.getCidGroupNo(),l) + workflowInstanceId;
                     ExecutionContext nextContext = context.clone();
                     nextContext.setCurrentNodeKey(edge.getTargetNodeKey());
-                    super.redisCache.setCacheObject(redisKey, nextContext);
+                    super.redisCache.setCacheObject(redisKey, nextContext, 1, TimeUnit.DAYS);
                 }
             }
         });

fs-service/src/main/java/com/fs/company/service/impl/call/node/AiCallTaskNode.java

@@ -28,7 +28,9 @@ import com.fs.enums.ExecutionStatusEnum;
 import com.fs.enums.NodeTypeEnum;
 import com.fs.enums.TaskTypeEnum;
 import com.fs.his.config.CidPhoneConfig;
+import com.fs.his.utils.PhoneUtil;
 import com.fs.system.service.ISysConfigService;
+import com.fs.wxcid.utils.TenantHelper;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.*;
@@ -360,9 +362,11 @@ public class AiCallTaskNode extends AbstractWorkflowNode {
         addListParam.setBatchId(batchId);
         // 构建号码条目，bizJson 传入默认客户信息占位符
         EasyCallPhoneItemVO phoneItem = new EasyCallPhoneItemVO();
+        // todo
+//        phoneItem.setPhoneNum(PhoneUtil.decryptPhone(callees.getPhone()));
         phoneItem.setPhoneNum(callees.getPhone());
         // bizJson 默认传入客户姓名占位，运行时可根据实际业务填充
-        phoneItem.setBizJson(new JSONObject().fluentPut("custName", callees.getUserName()).fluentPut("callBackUuid",callBackUuid).fluentPut("callBackUrl",callBackUrl));
+        phoneItem.setBizJson(new JSONObject().fluentPut("custName", callees.getUserName()).fluentPut("tenantId", TenantHelper.getTenantId()).fluentPut("callBackUuid",callBackUuid).fluentPut("callBackUrl",callBackUrl));
         addListParam.setPhoneList(Collections.singletonList(phoneItem));
         easyCallService.addCommonCallList(addListParam, null);
         log.info("workflowCallPhoneOne4EasyCall: 名单追加成功 - batchId: {}, phone: {}",

fs-service/src/main/java/com/fs/company/service/impl/call/node/AiQwAddWxTaskNode.java

@@ -208,7 +208,7 @@ public class AiQwAddWxTaskNode extends AbstractWorkflowNode {
      *
      */
     public void doneQwAddWx(String workflowInstanceId) {
-        ExecutionContext context = createExecutionContext(workflowInstanceId, nodeKey);
+        ExecutionContext context = createExecutionContextWithBusinessId(workflowInstanceId, nodeKey);
         context.setVariable("lastNodeKey", nodeKey);
         //启动定时节点倒计时
         CompanyAiWorkflowExec exec = companyAiWorkflowExecMapper.selectByWorkflowInstanceId(context.getWorkflowInstanceId());

fs-service/src/main/java/com/fs/crm/utils/CrmCustomerAiTagUtil.java

@@ -27,6 +27,7 @@ import com.fs.crm.vo.CrmCustomerAiTagVo;
 import com.fs.fastgptApi.param.ChatParam;
 import com.fs.fastgptApi.service.ChatService;
 import com.fs.hisapi.util.MapUtil;
+import com.fs.system.mapper.SysConfigMapper;
 import com.fs.system.mapper.SysDictDataMapper;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -56,15 +57,19 @@ public class CrmCustomerAiTagUtil {
 
     // 1. 声明静态变量
     private static RedisTemplate redisTemplate;
+    private static SysConfigMapper sysConfigMapper;
 
     // 2. 注入实例到非静态变量
     @Autowired
     @Qualifier("redisTemplate")
     private RedisTemplate redisTemplateInstance;
+    @Autowired
+    private SysConfigMapper sysConfigM;
     @PostConstruct
     public void initStatic() {
         APP_KEY = this.appKey;
         redisTemplate = this.redisTemplateInstance;
+        sysConfigMapper = this.sysConfigM;
     }
 
     public static List<CrmCustomerAiTagVo> getCrmCustomerAiTag(CrmCustomerAiTagParam content) throws JsonProcessingException {
@@ -144,8 +149,13 @@ public class CrmCustomerAiTagUtil {
             param.setMessages(messageList);
             ChatService chatService = SpringUtils.getBean(ChatService.class);
             AiHostProper aiHost = SpringUtils.getBean(AiHostProper.class);
-
-            return chatService.initiatingTakeChat(param, aiHost.getAiApi(), appKey);
+            com.fs.config.saas.ProjectConfig projectConfig = com.fs.config.saas.ProjectConfig.getFromDB(sysConfigMapper);
+            String aiApiUrl ="";
+            if(null != projectConfig && null != projectConfig.getIpad()){
+                aiApiUrl = projectConfig.getIpad().getAiApi();
+            }
+//            return chatService.initiatingTakeChat(param, aiHost.getAiApi(), appKey);
+            return chatService.initiatingTakeChat(param, aiApiUrl, appKey);
         } catch (Exception e) {
             throw new RuntimeException("AI服务调用失败", e);
         }

fs-service/src/main/resources/mapper/company/CompanyVoiceRoboticMapper.xml

@@ -183,6 +183,7 @@
             <if test="createTime != null">create_time = #{createTime},</if>
             <if test="createUser != null">create_user = #{createUser},</if>
             <if test="companyAiWorkflowId != null">company_ai_workflow_id = #{companyAiWorkflowId},</if>
+            <if test="taskStatus != null">task_status = #{taskStatus},</if>
         </trim>
         where id = #{id}
     </update>
@@ -231,7 +232,7 @@
     <select id="selectSceneTaskByCompanyIdAndType" resultType="CompanyVoiceRobotic">
         select * from company_voice_robotic where company_id = #{companyId}
                                               and scene_type = #{sceneType}
-                                              and task_status = 1
+                                              and task_status IN (1, 2)
                                               and del_flag = 0
         order by create_time desc
     </select>

fs-wx-api/src/main/java/com/fs/app/websocket/service/WebSocketServer.java

@@ -17,6 +17,7 @@ import com.fs.company.mapper.CompanyWxAccountMapper;
 import com.fs.company.mapper.CompanyWxClientMapper;
 import com.fs.company.service.CompanyWorkflowEngine;
 import com.fs.company.service.impl.CompanyWxServiceImpl;
+import com.fs.company.service.ICompanyVoiceRoboticService;
 import com.fs.core.config.TenantConfigContext;
 import com.fs.tenant.domain.TenantInfo;
 import com.fs.tenant.mapper.TenantInfoMapper;
@@ -40,10 +41,13 @@ import javax.websocket.server.ServerEndpoint;
 import java.io.IOException;
 import java.lang.reflect.Method;
 import java.time.LocalDateTime;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.Date;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
 
 
 @Slf4j
@@ -62,7 +66,8 @@ public class WebSocketServer {
     CidIpadServerMapper cidIpadServerMapper = SpringUtils.getBean(CidIpadServerMapper.class);
     CompanyWorkflowEngine companyWorkflowEngine = SpringUtils.getBean(CompanyWorkflowEngine.class);
     TenantInfoMapper tenantInfoMapper = SpringUtils.getBean(TenantInfoMapper.class);
-
+    ICompanyVoiceRoboticService companyVoiceRoboticService = SpringUtils.getBean(ICompanyVoiceRoboticService.class);
+    Executor cidWorkFlowExecutor = SpringUtils.getBean("cidWorkFlowExecutor");
     //发送消息
     public <T> void sendMessage(Session session, ResultMsgVo<T> data) {
         if (session != null) {
@@ -205,14 +210,30 @@ public class WebSocketServer {
                         List<CompanyWxClient> clients = companyWxClientMapper.selectWxV2(companyWxAccount.getId(), wxContact.getPhone());
                         log.info("更新联系人2：{}", clients);
                         if (clients != null) {
-                            clients.parallelStream().forEach(e -> {
-                                e.setIsAdd(1);
-                                e.setRemark(addResultWxVo.getRemark());
-                                e.setWxName(addResultWxVo.getUserName());
-                                e.setSuccessAddTime(LocalDateTime.now());
-                                companyWxClientMapper.updateById(e);
-                                companyWxService.triggerWorkflowOnAddWxSuccess(e.getId());
+                            List<CompletableFuture<Void>> futures = new ArrayList<>();
+                            clients.forEach(e -> {
+                                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
+                                    try {
+                                        e.setIsAdd(1);
+                                        e.setRemark(addResultWxVo.getRemark());
+                                        e.setWxName(addResultWxVo.getUserName());
+                                        e.setSuccessAddTime(LocalDateTime.now());
+                                        companyWxClientMapper.updateById(e);
+                                        // 暂停检查：任务暂停中则跳过工作流触发，加微结果已保存，恢复时resumePausedInstances会自动检测并补触发
+                                        if (e.getRoboticId() != null && companyVoiceRoboticService.isTaskPaused(e.getRoboticId())) {
+                                            log.info("任务暂停中，加微结果已保存但不触发工作流继续执行 - taskId: {}, wxClientId: {}", e.getRoboticId(), e.getId());
+                                            // 标记回调已到达，便于恢复时补触发
+                                            markAddWxCallbackReceived(e.getId());
+                                            return;
+                                        }
+                                        companyWxService.triggerWorkflowOnAddWxSuccess(e.getId());
+                                    } catch (Exception ex) {
+                                        log.error("处理加微回调异常 - wxClientId: {}", e.getId(), ex);
+                                    }
+                                }, cidWorkFlowExecutor);
+                                futures.add(future);
                             });
+                            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
                         }
 //                    if(null != addResultWxVo && StringUtils.isNotBlank(addResultWxVo.getBizJson())){
 //                        JSONObject jsonObject = JSONObject.parseObject(addResultWxVo.getBizJson());
@@ -256,6 +277,7 @@ public class WebSocketServer {
         }
         try {
             TenantInfo tenantInfo = tenantInfoMapper.getTenByCode(tenantCode);
+            TenantHelper.setTenantId(tenantInfo.getId());
             Object manager = SpringUtils.getBean("tenantDataSourceManager");
             Method method = manager.getClass().getMethod("ensureSwitchByTenantId", Long.class);
             method.invoke(manager, tenantInfo.getId());
@@ -277,6 +299,43 @@ public class WebSocketServer {
     }
 
 
+    /**
+     * 标记加微回调已到达（暂停期间），便于恢复时补触发
+     */
+    private void markAddWxCallbackReceived(Long wxClientId) {
+        try {
+            com.fs.company.mapper.CompanyAiWorkflowExecMapper execMapper =
+                    SpringUtils.getBean(com.fs.company.mapper.CompanyAiWorkflowExecMapper.class);
+            // 查找WAITING状态的加微工作流实例（AI_ADD_WX_TASK_NEW=10）
+            com.fs.company.domain.CompanyAiWorkflowExec exec = execMapper.selectWaitingAddWxWorkflowByWxClientId(
+                    wxClientId, 5, 10);
+            // 未找到新类型，尝试老类型（AI_ADD_WX_TASK=8）
+            if (exec == null) {
+                exec = execMapper.selectWaitingAddWxWorkflowByWxClientId(wxClientId, 5, 8);
+            }
+            if (exec == null) {
+                log.debug("markAddWxCallbackReceived: 未找到对应工作流实例 - wxClientId: {}", wxClientId);
+                return;
+            }
+            String variablesJson = exec.getVariables();
+            com.alibaba.fastjson.JSONObject variables;
+            if (com.fs.common.utils.StringUtils.isNotBlank(variablesJson)) {
+                variables = com.alibaba.fastjson.JSONObject.parseObject(variablesJson);
+            } else {
+                variables = new com.alibaba.fastjson.JSONObject();
+            }
+            variables.put("pause_callback_received", true);
+            com.fs.company.domain.CompanyAiWorkflowExec update = new com.fs.company.domain.CompanyAiWorkflowExec();
+            update.setWorkflowInstanceId(exec.getWorkflowInstanceId());
+            update.setVariables(variables.toJSONString());
+            execMapper.updateByWorkflowInstanceId(update);
+            log.info("markAddWxCallbackReceived: 已标记加微回调到达 - wxClientId: {}, workflowInstanceId: {}",
+                    wxClientId, exec.getWorkflowInstanceId());
+        } catch (Exception e) {
+            log.error("markAddWxCallbackReceived: 标记失败 - wxClientId: {}", wxClientId, e);
+        }
+    }
+
     public void finalHandle() {
         try {
             TenantConfigContext.clear();

fs-wx-task/src/main/java/com/fs/app/service/WxTaskService.java

@@ -855,21 +855,39 @@ public class WxTaskService {
 //        Set<String> keys = redisKeyScanner.scanMatchKey(delayAddWxKeyPrefix);
         Collection<String> keys = redisCache2.keys(delayAddWxKeyPrefix);
         log.info("cidWorkflowAddWxRun共扫描到 {} 个待处理键", keys.size());
+        // 本地缓存已查询的任务暂停状态，避免同一批次重复查询
+        Map<Long, Boolean> pausedCache = new ConcurrentHashMap<>();
         keys.parallelStream().forEach(key -> {
             try {
                 //doExec
                 CompletableFuture.runAsync(()->{
                     try {
                         ExecutionContext context = redisCache2.getCacheObject(key);
+                        if (context == null) {
+                            log.warn("工作流延时任务context为空，跳过 - key: {}", key);
+                            redisCache2.deleteObject(key);
+                            return;
+                        }
+                        // 任务暂停守卫检查（roboticId即CompanyVoiceRobotic.id，是实际暂停操作的目标）
+                        Long taskId = context.getVariable("roboticId", Long.class);
+                        if (taskId != null && pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id))) {
+                            // 延时key是时间分片前缀，下一分钟就不会再扫到，直接删除
+                            // 同步context信息到DB exec，供恢复时resumePausedInstances使用
+                            context.setVariable("callSource", "addWxTimer");
+                            context.setVariable("_delayTargetNodeKey", context.getCurrentNodeKey());
+                            companyWorkflowEngine.updateExecVariables(context.getWorkflowInstanceId(), context.getVariables());
+                            log.info("任务已暂停，删除延时key并同步exec，等待恢复时从DB重建 - taskId: {}, key: {}", taskId, key);
+                            redisCache2.deleteObject(key);
+                            return;
+                        }
                         context.setVariable("callRedisKey",key);
                         context.setVariable("callSource","addWxTimer");
                         companyWorkflowEngine.timeDoExecute(context.getWorkflowInstanceId(),context.getCurrentNodeKey(),context.getVariables());
+                        redisCache2.deleteObject(key);
                     } catch (Exception e) {
                         log.error("处理工作流延时任务异常 - key: {}", key, e);
                     }
-                }, cidWorkFlowExecutor).thenRun(()->{
-                    redisCache2.deleteObject(key);
-                });
+                }, cidWorkFlowExecutor);
 
             } catch (Exception ex) {
                 log.error("处理工作流延时任务异常 - key: {}", key, ex);
@@ -909,6 +927,22 @@ public class WxTaskService {
                 return;
             }
 
+            // 任务暂停守卫检查：过滤掉已暂停任务的数据
+            Map<Long, Boolean> pausedCache = new HashMap<>();
+            list = list.stream().filter(item -> {
+                Long taskId = item.getRoboticId();
+                if (taskId == null) return true;
+                boolean paused = pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id));
+                if (paused) {
+                    log.debug("任务已暂停，跳过加微 - taskId: {}, clientId: {}", taskId, item.getRoboticWxId());
+                }
+                return !paused;
+            }).collect(Collectors.toList());
+            if (list.isEmpty()) {
+                log.info("过滤暂停任务后无需要加微的数据");
+                return;
+            }
+
             // 构建客户映射
             Map<Long, CompanyWxClient4WorkFlowVO> clientMap = PubFun.listToMapByGroupObject(
                     list, CompanyWxClient4WorkFlowVO::getAccountId);
@@ -1587,6 +1621,23 @@ public class WxTaskService {
             if (clients.isEmpty()) {
                 return;
             }
+
+            // 任务暂停守卫检查：过滤掉已暂停任务的数据
+            Map<Long, Boolean> pausedCache = new HashMap<>();
+            clients = clients.stream().filter(client -> {
+                Long taskId = client.getRoboticId();
+                if (taskId == null) return true;
+                boolean paused = pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id));
+                if (paused) {
+                    log.debug("任务已暂停，跳过加微结果查询 - taskId: {}, clientId: {}", taskId, client.getId());
+                }
+                return !paused;
+            }).collect(Collectors.toList());
+            if (clients.isEmpty()) {
+                log.info("过滤暂停任务后无需要查询加微结果的数据");
+                return;
+            }
+
             // 处理每个客户的加微结果
             List<CompanyWxClient> upClientList = new CopyOnWriteArrayList<>();
             List<CompletableFuture<Void>> futures = new ArrayList<>();
@@ -1626,21 +1677,39 @@ public class WxTaskService {
         String delayAddWxKeyPrefix = AiQwAddWxTaskNode.getDelayAddWxKeyPrefix(cidGroupNo,null) + "*";
         Collection<String> keys = redisCache2.keys(delayAddWxKeyPrefix);
         log.info("企微加微共扫描到 {} 个待处理键", keys.size());
+        // 本地缓存已查询的任务暂停状态，避免同一批次重复查询
+        Map<Long, Boolean> pausedCache = new ConcurrentHashMap<>();
         keys.parallelStream().forEach(key -> {
             try {
                 //doExec
                 CompletableFuture.runAsync(()->{
                     try {
                         ExecutionContext context = redisCache2.getCacheObject(key);
+                        if (context == null) {
+                            log.warn("企微加微工作流延时任务context为空，跳过 - key: {}", key);
+                            redisCache2.deleteObject(key);
+                            return;
+                        }
+                        // 任务暂停守卫检查（roboticId即CompanyVoiceRobotic.id，是实际暂停操作的目标）
+                        Long taskId = context.getVariable("roboticId", Long.class);
+                        if (taskId != null && pausedCache.computeIfAbsent(taskId, id -> companyVoiceRoboticService.isTaskPaused(id))) {
+                            // 延时key是时间分片前缀，下一分钟就不会再扫到，直接删除
+                            // 同步context信息到DB exec，供恢复时resumePausedInstances使用
+                            context.setVariable("callSource", "qwAddWxTimer");
+                            context.setVariable("_delayTargetNodeKey", context.getCurrentNodeKey());
+                            companyWorkflowEngine.updateExecVariables(context.getWorkflowInstanceId(), context.getVariables());
+                            log.info("任务已暂停，删除延时key并同步exec，等待恢复时从DB重建 - taskId: {}, key: {}", taskId, key);
+                            redisCache2.deleteObject(key);
+                            return;
+                        }
                         context.setVariable("callRedisKey",key);
                         context.setVariable("callSource","qwAddWxTimer");
                         companyWorkflowEngine.timeDoExecute(context.getWorkflowInstanceId(),context.getCurrentNodeKey(),context.getVariables());
+                        redisCache2.deleteObject(key);
                     } catch (Exception e) {
                         log.error("处理工作流延时任务异常 - key: {}", key, e);
                     }
-                }, cidWorkFlowExecutor).thenRun(()->{
-                    redisCache2.deleteObject(key);
-                });
+                }, cidWorkFlowExecutor);
 
             } catch (Exception ex) {
                 log.error("处理工作流延时任务异常 - key: {}", key, ex);