# 任务编排系统 实现计划

> **面向 AI 代理的工作者：** 必需子技能：使用 superpowers:subagent-driven-development 逐任务实现此计划。步骤使用复选框（`- [ ]`）语法来跟踪进度。

**目标：** 在现有 `tasks` 表之上新增任务编排模块（含环 DAG：多前置 AND/OR、超时/失败兜底、计数/条件循环终止），后端提供 Flow/Execution 管理与 Quartz 轮询，cfc-web 管理端提供 jsPlumb 可视化编排画布。

**架构：** 新增 4 张表（flows / edges / executions / node_instances），核心编排引擎 `OrchestrationEngine` 负责节点触发、循环终止、兜底分支，通过 `OrchestrationPollingJob`（Quartz）轮询评估，节点执行复用 `TaskService` 创建 `tasks` 实例。`TaskService` 完成/失败点回调编排引擎（仅编排节点任务生效）。

**技术栈：** Spring Boot 2.7.18 + MyBatis-Plus 3.5.3.1 + Java 8 + Quartz（新增依赖）；cfc-web Vue 2 + Element UI + jsPlumb。

**设计规格：** `docs/superpowers/specs/2026-09-19-task-orchestration-design.md`

---

## 文件清单

### 后端新增
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/entity/TaskOrchestrationFlow.java` — flows 表实体
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/entity/TaskOrchestrationEdge.java` — edges 表实体
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/entity/TaskOrchestrationExecution.java` — executions 表实体
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/entity/TaskOrchestrationNodeInstance.java` — node_instances 表实体
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/mapper/TaskOrchestrationFlowMapper.java`
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/mapper/TaskOrchestrationEdgeMapper.java`
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/mapper/TaskOrchestrationExecutionMapper.java`
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/mapper/TaskOrchestrationNodeInstanceMapper.java`
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationEngine.java` — 编排引擎（核心）
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationFlowService.java` — Flow CRUD
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationExecutionService.java` — Execution 生命周期
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/controller/OrchestrationController.java` — REST 接口
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/task/OrchestrationPollingJob.java` — Quartz 轮询 Job
- **创建** `cfc-backend/src/main/java/com/etotem/cfc/config/QuartzConfig.java` — Quartz 调度配置

### 后端修改
- **修改** `cfc-backend/pom.xml` — 新增 `spring-boot-starter-quartz` 依赖
- **修改** `cfc-backend/src/main/java/com/etotem/cfc/config/DatabaseInitializer.java` — 新增 4 张表迁移
- **修改** `cfc-backend/src/main/resources/schema.sql` — 同步 4 张表建表语句
- **修改** `cfc-backend/src/main/java/com/etotem/cfc/service/TaskService.java` — 完成/失败点回调编排引擎

### 前端新增（cfc-web）
- **创建** `cfc-web/src/views/orchestration/OrchestrationFlow.vue` — 流列表 + 入口
- **创建** `cfc-web/src/views/orchestration/FlowEditor.vue` — jsPlumb 编排画布
- **创建** `cfc-web/src/views/orchestration/ExecutionDetail.vue` — 执行详情只读页

### 前端修改（cfc-web）
- **修改** `cfc-web/src/router/index.js` — 注册 3 个路由
- **修改** `cfc-web/src/api/orchestration.js`（新建）— 接口封装
- **修改** `cfc-web/package.json` — 新增 jsplumb 依赖

### 测试新增
- **创建** `cfc-backend/src/test/java/com/etotem/cfc/orchestration/OrchestrationEngineTest.java` — 引擎单元测试

---

## 任务分解

### 任务 1：后端 — 实体 + Mapper + 数据迁移

**文件：**
- 创建：4 个 entity（`TaskOrchestrationFlow.java` / `TaskOrchestrationEdge.java` / `TaskOrchestrationExecution.java` / `TaskOrchestrationNodeInstance.java`）
- 创建：4 个 Mapper
- 修改：`DatabaseInitializer.java`
- 修改：`schema.sql`

- [ ] **步骤 1：创建 4 个实体类**

按 `Task.java` 现有模式（`@Data` + `@TableName` + `@TableId(type = IdType.AUTO)` + `implements Serializable`）。字段严格对应设计文档 §3：

`TaskOrchestrationFlow.java`（`@TableName("task_orchestration_flows")`）：
```java
package com.etotem.cfc.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.io.Serializable;
import java.util.Date;

@Data
@TableName("task_orchestration_flows")
public class TaskOrchestrationFlow implements Serializable {

    @TableId(type = IdType.AUTO)
    private Long id;

    private String name;

    private String description;

    private Long creatorId;

    private Long familyId;

    private Integer version;

    private String status;

    private String scheduleCron;

    private String configJson;

    private Date createdAt;

    private Date updatedAt;
}
```

`TaskOrchestrationEdge.java`（`@TableName("task_orchestration_edges")`）：
```java
package com.etotem.cfc.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.io.Serializable;
import java.util.Date;

@Data
@TableName("task_orchestration_edges")
public class TaskOrchestrationEdge implements Serializable {

    @TableId(type = IdType.AUTO)
    private Long id;

    private Long flowId;

    private String fromNodeId;

    private String toNodeId;

    private String edgeType;

    private String operator;

    private Integer sortOrder;

    private Date createdAt;
}
```

`TaskOrchestrationExecution.java`（`@TableName("task_orchestration_executions")`）：
```java
package com.etotem.cfc.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.io.Serializable;
import java.util.Date;

@Data
@TableName("task_orchestration_executions")
public class TaskOrchestrationExecution implements Serializable {

    @TableId(type = IdType.AUTO)
    private Long id;

    private Long flowId;

    private Integer flowVersion;

    private Long familyId;

    private Long familyMemberId;

    private String triggerSource;

    private String status;

    private Date startedAt;

    private Date finishedAt;

    private String errorReason;

    private Date createdAt;

    private Date updatedAt;
}
```

`TaskOrchestrationNodeInstance.java`（`@TableName("task_orchestration_node_instances")`）：
```java
package com.etotem.cfc.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import lombok.Data;

import java.io.Serializable;
import java.util.Date;

@Data
@TableName("task_orchestration_node_instances")
public class TaskOrchestrationNodeInstance implements Serializable {

    @TableId(type = IdType.AUTO)
    private Long id;

    private Long executionId;

    private String nodeId;

    private Long taskId;

    private String status;

    private Integer generation;

    private Date conditionMetAt;

    private String loopTerminationReason;

    private Date startedAt;

    private Date completedAt;

    private String errorReason;

    private Date createdAt;

    private Date updatedAt;
}
```

- [ ] **步骤 2：创建 4 个 Mapper 接口**

每个 Mapper 继承 `BaseMapper<T>`，模式参考 `TaskMapper`：

```java
package com.etotem.cfc.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.etotem.cfc.entity.TaskOrchestrationFlow;

public interface TaskOrchestrationFlowMapper extends BaseMapper<TaskOrchestrationFlow> {
}
```

（其余 3 个 Mapper 同理：`TaskOrchestrationEdgeMapper` / `TaskOrchestrationExecutionMapper` / `TaskOrchestrationNodeInstanceMapper`）

- [ ] **步骤 3：在 DatabaseInitializer 新增迁移**

在 `runMigrations()` 末尾（搜索 `// 迁移` 找到最新编号，递增），加入 4 张表的 `CREATE TABLE IF NOT EXISTS` 语句，内容与设计文档 §11 DDL 完全一致（含 `condition_met_at` 列和唯一索引 `uk_node_instance`）。

- [ ] **步骤 4：同步 schema.sql**

在 `schema.sql` 末尾追加相同的 4 张表 `CREATE TABLE IF NOT EXISTS` 语句（与迁移脚本保持一致）。

- [ ] **步骤 5：编译验证**

运行：`cd cfc-backend && mvn clean compile`
预期：BUILD SUCCESS，无编译错误。

- [ ] **步骤 6：Commit**

```bash
git add cfc-backend/src/main/java/com/etotem/cfc/entity/TaskOrchestration*.java \
        cfc-backend/src/main/java/com/etotem/cfc/mapper/TaskOrchestration*Mapper.java \
        cfc-backend/src/main/java/com/etotem/cfc/config/DatabaseInitializer.java \
        cfc-backend/src/main/resources/schema.sql
git commit -m "feat: 任务编排 4 张新表实体/Mapper/迁移/schema"
```

---

### 任务 2：后端 — 编排引擎 OrchestrationEngine（核心）

**文件：**
- 创建：`cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationEngine.java`
- 创建：`cfc-backend/src/test/java/com/etotem/cfc/orchestration/OrchestrationEngineTest.java`

- [ ] **步骤 1：编写失败的单元测试（AND/OR 语义 + 循环终止 + 幂等）**

`OrchestrationEngineTest.java`：
```java
package com.etotem.cfc.orchestration;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.etotem.cfc.entity.TaskOrchestrationEdge;
import com.etotem.cfc.entity.TaskOrchestrationNodeInstance;
import com.etotem.cfc.service.OrchestrationEngine;
import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.List;

import static org.junit.jupiter.api.Assertions.*;

public class OrchestrationEngineTest {

    private OrchestrationEngine engine = new OrchestrationEngine();

    // 前置 AND：两个源节点都 completed 才触发
    @Test
    public void testAndGateRequiresAllCompleted() {
        List<TaskOrchestrationEdge> inEdges = Arrays.asList(
            edge("n1", "n3", "AND"),
            edge("n2", "n3", "AND")
        );
        List<TaskOrchestrationNodeInstance> nodes = Arrays.asList(
            node("n1", "completed"),
            node("n2", "pending")
        );
        assertFalse(engine.isUnlocked("n3", inEdges, nodes));
    }

    // 前置 OR：任一源节点 completed 即触发
    @Test
    public void testOrGateUnlocksWhenAnyCompleted() {
        List<TaskOrchestrationEdge> inEdges = Arrays.asList(
            edge("n1", "n3", "OR"),
            edge("n2", "n3", "OR")
        );
        List<TaskOrchestrationNodeInstance> nodes = Arrays.asList(
            node("n1", "completed"),
            node("n2", "pending")
        );
        assertTrue(engine.isUnlocked("n3", inEdges, nodes));
    }

    // 循环终止：generation 达到 max_loops 时置 terminated
    @Test
    public void testLoopTerminationByCount() {
        JSONObject cfg = JSON.parseObject("{\"max_loops\":3}");
        TaskOrchestrationNodeInstance ni = node("n1", "in_progress");
        ni.setGeneration(3); // 已达上限
        engine.applyLoopTermination(ni, cfg, 3);
        assertEquals("terminated", ni.getStatus());
        assertEquals("max_loops_reached", ni.getLoopTerminationReason());
    }

    private TaskOrchestrationEdge edge(String from, String to, String op) {
        TaskOrchestrationEdge e = new TaskOrchestrationEdge();
        e.setFromNodeId(from);
        e.setToNodeId(to);
        e.setOperator(op);
        return e;
    }

    private TaskOrchestrationNodeInstance node(String id, String status) {
        TaskOrchestrationNodeInstance n = new TaskOrchestrationNodeInstance();
        n.setNodeId(id);
        n.setStatus(status);
        return n;
    }
}
```

> 说明：`isUnlocked` / `applyLoopTermination` 是引擎将要暴露的纯函数方法，便于单元测试（不依赖数据库）。

- [ ] **步骤 2：运行测试确认失败**

运行：`cd cfc-backend && mvn test -Dtest=OrchestrationEngineTest`
预期：编译失败，`OrchestrationEngine` 类不存在。

- [ ] **步骤 3：实现 OrchestrationEngine**

`OrchestrationEngine.java` 核心方法（注入 4 个 Mapper + `TaskService` + `TaskTemplateMapper`，`@Resource` DI）：

```java
package com.etotem.cfc.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.etotem.cfc.entity.*;
import com.etotem.cfc.mapper.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
@Service
public class OrchestrationEngine {

    @Resource
    private TaskOrchestrationNodeInstanceMapper nodeInstanceMapper;
    @Resource
    private TaskOrchestrationEdgeMapper edgeMapper;
    @Resource
    private TaskOrchestrationExecutionMapper executionMapper;
    @Resource
    private TaskOrchestrationFlowMapper flowMapper;
    @Resource
    private TaskTemplateMapper taskTemplateMapper;
    @Resource
    private TaskService taskService;

    /**
     * 纯函数：判断目标节点是否满足入边触发条件。
     * @return true = 可触发
     */
    public boolean isUnlocked(String targetNodeId,
                              List<TaskOrchestrationEdge> inEdges,
                              List<TaskOrchestrationNodeInstance> nodes) {
        if (inEdges == null || inEdges.isEmpty()) {
            return true; // 无前置，start_node 直接触发
        }
        Map<String, String> statusById = nodes.stream()
            .collect(Collectors.toMap(TaskOrchestrationNodeInstance::getNodeId,
                                      TaskOrchestrationNodeInstance::getStatus));
        // 按 operator 分组：AND 组内全 completed，OR 组内任一 completed
        boolean hasAnd = false;
        boolean hasOr = false;
        boolean orSatisfied = false;
        for (TaskOrchestrationEdge e : inEdges) {
            String src = e.getFromNodeId();
            String srcStatus = statusById.getOrDefault(src, "pending");
            if ("AND".equalsIgnoreCase(e.getOperator())) {
                hasAnd = true;
                if (!"completed".equals(srcStatus)) {
                    return false; // 任一 AND 源未完成即不触发
                }
            } else { // OR
                hasOr = true;
                if ("completed".equals(srcStatus)) {
                    orSatisfied = true;
                }
            }
        }
        if (hasAnd && !hasOr) {
            return true; // 纯 AND，全部完成（前面已通过 return false 检查）
        }
        if (!hasAnd && hasOr) {
            return orSatisfied; // 纯 OR
        }
        // AND + OR 混合：AND 全完成（已通过检查）且 OR 至少一个完成
        return true; // 到达此处说明 AND 已全完成，OR 是否有满足不影响（按 AND 优先语义）
    }

    /**
     * 纯函数：应用循环终止规则。
     */
    public void applyLoopTermination(TaskOrchestrationNodeInstance ni,
                                     JSONObject nodeCfg,
                                     int maxLoops) {
        if (ni.getGeneration() != null && ni.getGeneration() >= maxLoops) {
            ni.setStatus("terminated");
            ni.setLoopTerminationReason("max_loops_reached");
        }
    }

    /**
     * 事件驱动评估：节点完成/失败/超时后调用，扫描 pending 节点并触发满足条件的。
     */
    public void evaluateFlow(Long executionId) {
        // 1. 加载执行实例
        TaskOrchestrationExecution exec = executionMapper.selectById(executionId);
        if (exec == null || !"running".equals(exec.getStatus())) {
            return;
        }
        // 2. 加载该 execution 的所有节点实例
        List<TaskOrchestrationNodeInstance> nodes = nodeInstanceMapper.selectList(
            new LambdaQueryWrapper<TaskOrchestrationNodeInstance>()
                .eq(TaskOrchestrationNodeInstance::getExecutionId, executionId));
        // 3. 加载 flow 的边（此处需从 execution.flowId 关联 edges，但版本快照简化：直接读当前边表）
        List<TaskOrchestrationEdge> edges = edgeMapper.selectList(
            new LambdaQueryWrapper<TaskOrchestrationEdge>()
                .eq(TaskOrchestrationEdge::getFlowId, exec.getFlowId()));
        // 4. 对每个 pending 节点，检查入边是否满足
        Set<String> nodeIds = nodes.stream()
            .map(TaskOrchestrationNodeInstance::getNodeId).collect(Collectors.toSet());
        for (String nodeId : nodeIds) {
            List<TaskOrchestrationNodeInstance> pendingNodes = nodes.stream()
                .filter(n -> "pending".equals(n.getStatus()) && nodeId.equals(n.getNodeId()))
                .collect(Collectors.toList());
            if (pendingNodes.isEmpty()) continue;
            List<TaskOrchestrationEdge> inEdges = edges.stream()
                .filter(e -> nodeId.equals(e.getToNodeId()))
                .collect(Collectors.toList());
            if (isUnlocked(nodeId, inEdges, nodes)) {
                triggerNode(exec, pendingNodes.get(0), inEdges);
            }
        }
        // 5. 判断 flow 是否完成（所有节点均非 pending/running/in_progress）
        boolean allTerminal = nodes.stream().allMatch(n ->
            !"pending".equals(n.getStatus())
            && !"in_progress".equals(n.getStatus())
            && !"started".equals(n.getStatus()));
        if (allTerminal) {
            exec.setStatus("completed");
            exec.setFinishedAt(new Date());
            executionMapper.updateById(exec);
        }
    }

    private void triggerNode(TaskOrchestrationExecution exec,
                             TaskOrchestrationNodeInstance ni,
                             List<TaskOrchestrationEdge> inEdges) {
        // 幂等：唯一索引兜底，这里先检查
        Long existingCount = nodeInstanceMapper.selectCount(
            new LambdaQueryWrapper<TaskOrchestrationNodeInstance>()
                .eq(TaskOrchestrationNodeInstance::getExecutionId, ni.getExecutionId())
                .eq(TaskOrchestrationNodeInstance::getNodeId, ni.getNodeId())
                .eq(TaskOrchestrationNodeInstance::getGeneration, ni.getGeneration())
                .ne(TaskOrchestrationNodeInstance::getStatus, "pending"));
        if (existingCount != null && existingCount > 0) return;
        // triggerNode 完整实现见任务 3（步骤 1），此处暂不展开，
        // 任务 2 仅保留 evaluateFlow 骨架及 isUnlocked / applyLoopTermination 纯函数
    }

    // triggerNode、createTaskFromTemplate、triggerFailedEdges 等完整实现见任务 3（步骤 1）
}
```

> **注意**：任务 2 仅定义引擎骨架（`evaluateFlow`、`isUnlocked`、`applyLoopTermination`、`onTaskCompleted`、`onTaskFailed`、`onTaskTimeout`、`findByTaskId`）。`triggerNode` 的完整实现（含 `createTaskFromTemplate` 模板字段映射）见任务 3 步骤 1。

- [ ] **步骤 4：运行测试确认通过**

运行：`cd cfc-backend && mvn test -Dtest=OrchestrationEngineTest`
预期：PASS（4 个测试全部通过）。

- [ ] **步骤 5：编译验证**

运行：`cd cfc-backend && mvn clean compile`
预期：BUILD SUCCESS。

- [ ] **步骤 6：Commit**

```bash
git add cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationEngine.java \
        cfc-backend/src/test/java/com/etotem/cfc/orchestration/OrchestrationEngineTest.java
git commit -m "feat: 编排引擎核心（AND/OR 触发 + 循环终止 + evaluateFlow）"
```

---

### 任务 3：后端 — 节点任务创建 + TaskService 回调

**文件：**
- 修改：`cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationEngine.java` — 补全 `triggerNode`
- 修改：`cfc-backend/src/main/java/com/etotem/cfc/service/TaskService.java` — 完成/失败点回调

- [ ] **步骤 1：补全 triggerNode（加载模板 → 创建 tasks → 回写）**

在 `OrchestrationEngine` 中补全 `triggerNode` 方法体，替换步骤 2 中的 TODO 部分。核心逻辑：

```java
private void triggerNode(TaskOrchestrationExecution exec,
                         TaskOrchestrationNodeInstance ni,
                         List<TaskOrchestrationEdge> inEdges) {
    // 幂等检查（唯一索引兜底）
    Long existingCount = nodeInstanceMapper.selectCount(
        new LambdaQueryWrapper<TaskOrchestrationNodeInstance>()
            .eq(TaskOrchestrationNodeInstance::getExecutionId, ni.getExecutionId())
            .eq(TaskOrchestrationNodeInstance::getNodeId, ni.getNodeId())
            .eq(TaskOrchestrationNodeInstance::getGeneration, ni.getGeneration())
            .ne(TaskOrchestrationNodeInstance::getStatus, "pending"));
    if (existingCount != null && existingCount > 0) return;

    // 1. 读 flow 的 config_json，拿到节点配置
    TaskOrchestrationFlow flow = flowMapper.selectById(exec.getFlowId());
    JSONObject flowCfg = JSON.parseObject(flow.getConfigJson());
    JSONObject nodeCfg = null;
    for (Object o : flowCfg.getJSONArray("nodes")) {
        JSONObject n = (JSONObject) o;
        if (ni.getNodeId().equals(n.getString("id"))) {
            nodeCfg = n;
            break;
        }
    }
    if (nodeCfg == null) {
        ni.setStatus("failed");
        ni.setErrorReason("节点配置不存在");
        nodeInstanceMapper.updateById(ni);
        return;
    }

    // 2. 加载任务模板
    String templateRef = nodeCfg.getString("task_template_ref");
    TaskTemplate template = taskTemplateMapper.selectById(Long.valueOf(templateRef));
    if (template == null) {
        ni.setStatus("failed");
        ni.setErrorReason("任务模板不存在: " + templateRef);
        nodeInstanceMapper.updateById(ni);
        triggerFailedEdges(exec, ni); // 触发 failed 边
        return;
    }

    // 3. 创建 tasks 实例（调用 TaskService.createTask(Long, CreateTaskDTO)）
    // 根据模板配置填充 DTO，关键字段映射：
    //   title/description：从 template 读取
    //   points：从 template.pts 读取
    //   deadline：nodeCfg.getInteger("timeout_minutes") 对应 task.deadline
    //   is_daily_progress：nodeCfg.boolean("is_daily_progress")
    //   source_type/source_id：flow config_json 中的对应字段
    //   memberOnly：template.integer("memberOnly")
    CreateTaskDTO dto = new CreateTaskDTO();
    dto.setTitle(template.getTitle());
    dto.setDescription(template.getDescription());
    dto.setPoints(template.getPoints());
    dto.setTaskType(template.getTaskType());
    dto.setFrequency(template.getFrequency());
    dto.setMaxFrequency(template.getMaxFrequency());
    dto.setNeedReview(template.getNeedReview());
    dto.setReviewType(template.getReviewType());
    dto.setReviewByCategory(template.getReviewByCategory());
    dto.setCompleteTypes(template.getCompleteTypes());
    dto.setEarliestStart(template.getEarliestStart());
    dto.setLatestEnd(template.getLatestEnd());
    dto.setExecutorType(nodeCfg.getString("executorType") != null ? nodeCfg.getString("executorType") : "child");
    dto.setExecutorId(ni.getFamilyMemberId());
    dto.setMemberId(ni.getFamilyMemberId());
    dto.setMinigameCode(nodeCfg.getString("minigameCode"));
    dto.setRepeatType(nodeCfg.getString("repeatType"));
    dto.setDuration(nodeCfg.getInteger("duration"));
    dto.setDimensionCode(nodeCfg.getString("dimensionCode"));
    dto.setDimensionWeights(nodeCfg.getString("dimensionWeights"));
    dto.setMemberOnly(nodeCfg.getInteger("memberOnly") != null ? nodeCfg.getInteger("memberOnly") : 0);
    dto.setPrerequisiteTaskId(nodeCfg.getLong("prerequisiteTaskId"));
    dto.setActionType(nodeCfg.getString("actionType"));
    dto.setActionConfig(nodeCfg.getString("actionConfig"));
    dto.setRequireInput(nodeCfg.getInteger("requireInput"));
    dto.setStartRequired(nodeCfg.getInteger("startRequired"));
    dto.setMinDurationSeconds(nodeCfg.getInteger("minDurationSeconds"));
    dto.setSourceType(nodeCfg.getString("sourceType"));
    dto.setSourceId(nodeCfg.getLong("sourceId"));
    dto.setIsDailyProgress(nodeCfg.getInteger("is_daily_progress"));
    dto.setTargetValue(nodeCfg.getInteger("targetValue"));
    // 超时时间：deadline = now + timeout_minutes（若配置）
    Long taskId = taskService.createTask(ni.getFamilyMemberId(), dto);

    // 4. 回写 task_id + 状态
    ni.setTaskId(taskId);
    ni.setStatus("in_progress");
    ni.setStartedAt(new Date());
    nodeInstanceMapper.updateById(ni);
}

private void triggerFailedEdges(TaskOrchestrationExecution exec,
                                TaskOrchestrationNodeInstance ni) {
    // 触发所有 edge_type=failed 且 from_node_id=ni.nodeId 的出边目标节点（创建 pending 实例）
    List<TaskOrchestrationEdge> failedEdges = edgeMapper.selectList(
        new LambdaQueryWrapper<TaskOrchestrationEdge>()
            .eq(TaskOrchestrationEdge::getFlowId, exec.getFlowId())
            .eq(TaskOrchestrationEdge::getFromNodeId, ni.getNodeId())
            .eq(TaskOrchestrationEdge::getEdgeType, "failed"));
    for (TaskOrchestrationEdge e : failedEdges) {
        // 幂等：检查同execution_id + to_node_id + generation=0 的 pending 实例是否存在
        Long exists = nodeInstanceMapper.selectCount(
            new LambdaQueryWrapper<TaskOrchestrationNodeInstance>()
                .eq(TaskOrchestrationNodeInstance::getExecutionId, exec.getId())
                .eq(TaskOrchestrationNodeInstance::getNodeId, e.getToNodeId())
                .eq(TaskOrchestrationNodeInstance::getGeneration, 0)
                .eq(TaskOrchestrationNodeInstance::getStatus, "pending"));
        if (exists != null && exists > 0) continue; // 已存在则跳过
        TaskOrchestrationNodeInstance newNi = new TaskOrchestrationNodeInstance();
        newNi.setExecutionId(exec.getId());
        newNi.setNodeId(e.getToNodeId());
        newNi.setGeneration(0);
        newNi.setStatus("pending");
        newNi.setCreatedAt(new Date());
        nodeInstanceMapper.insert(newNi);
    }
}
```

> **关键**：`triggerFailedEdges` 通过查询唯一索引 `(execution_id, node_id, generation)` 实现幂等，避免 Quartz 轮询与回调同时触发导致重复创建节点任务。`createTaskFromTemplate` 内部需调用 `TaskService.createTask(CreateTaskDTO)`，字段映射（title/description/points/needReview/deadline）参考 `TaskService` 现有模板实例化逻辑。**实施时先读取 `TaskService.createTask` 方法签名确认参数**，不可臆测。

- [ ] **步骤 2：在 TaskService 完成点回调编排引擎**

定位 `TaskService` 中任务完成（状态置 completed）的代码点，末尾添加（仅编排节点任务生效）：

```java
@Resource
@Lazy
private OrchestrationEngine orchestrationEngine;

// 在任务完成逻辑后：
orchestrationEngine.onTaskCompleted(taskId);
```

在 `OrchestrationEngine` 新增回调方法：

```java
public void onTaskCompleted(Long taskId) {
    TaskOrchestrationNodeInstance ni = findByTaskId(taskId);
    if (ni == null) return; // 非编排节点任务，零开销
    ni.setStatus("completed");
    ni.setCompletedAt(new Date());
    nodeInstanceMapper.updateById(ni);
    evaluateFlow(ni.getExecutionId());
}

private TaskOrchestrationNodeInstance findByTaskId(Long taskId) {
    return nodeInstanceMapper.selectOne(
        new LambdaQueryWrapper<TaskOrchestrationNodeInstance>()
            .eq(TaskOrchestrationNodeInstance::getTaskId, taskId));
}
```

- [ ] **步骤 3：在 TaskService 失败点回调编排引擎**

同理，在任务失败/放弃逻辑点调用 `orchestrationEngine.onTaskFailed(taskId)`，引擎实现：

```java
public void onTaskFailed(Long taskId) {
    TaskOrchestrationNodeInstance ni = findByTaskId(taskId);
    if (ni == null) return;
    ni.setStatus("failed");
    ni.setCompletedAt(new Date());
    nodeInstanceMapper.updateById(ni);
    // 触发 failed 边
    TaskOrchestrationExecution exec = executionMapper.selectById(ni.getExecutionId());
    triggerFailedEdges(exec, ni);
    evaluateFlow(ni.getExecutionId());
}
```

- [ ] **步骤 4：编译验证**

运行：`cd cfc-backend && mvn clean compile`
预期：BUILD SUCCESS（确认 `TaskService.createTask` 签名匹配，`flowMapper` 已注入）。

- [ ] **步骤 5：Commit**

```bash
git add cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationEngine.java \
        cfc-backend/src/main/java/com/etotem/cfc/service/TaskService.java
git commit -m "feat: 节点任务创建 + TaskService 完成/失败回调编排引擎"
```

---

### 任务 4：后端 — Quartz 轮询 + timeout 检查

**文件：**
- 修改：`cfc-backend/pom.xml` — 新增 Quartz 依赖
- 创建：`cfc-backend/src/main/java/com/etotem/cfc/config/QuartzConfig.java`
- 创建：`cfc-backend/src/main/java/com/etotem/cfc/task/OrchestrationPollingJob.java`
- 修改：`cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationEngine.java` — 新增 `onTaskTimeout`

- [ ] **步骤 1：pom.xml 新增 Quartz 依赖**

在 `<dependencies>` 中加入：
```xml
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>
```

- [ ] **步骤 2：创建 QuartzConfig**

```java
package com.etotem.cfc.config;

import org.quartz.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class QuartzConfig {

    @Bean
    public JobDetail orchestrationPollingJobDetail() {
        return JobBuilder.newJob(OrchestrationPollingJob.class)
            .withIdentity("orchestrationPollingJob")
            .storeDurably()
            .build();
    }

    @Bean
    public Trigger orchestrationPollingTrigger() {
        return TriggerBuilder.newTrigger()
            .forJob(orchestrationPollingJobDetail())
            .withIdentity("orchestrationPollingTrigger")
            .withSchedule(SimpleScheduleBuilder.simpleSchedule()
                .withIntervalInSeconds(30)
                .repeatForever())
            .build();
    }
}
```

> **注意**：`OrchestrationPollingJob` 在 `com.etotem.cfc.task` 包，`QuartzConfig` 在 `com.etotem.cfc.config` 包，需 import 对应类。

- [ ] **步骤 3：创建 OrchestrationPollingJob**

```java
package com.etotem.cfc.task;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.etotem.cfc.entity.Task;
import com.etotem.cfc.entity.TaskOrchestrationExecution;
import com.etotem.cfc.entity.TaskOrchestrationNodeInstance;
import com.etotem.cfc.mapper.TaskMapper;
import com.etotem.cfc.mapper.TaskOrchestrationExecutionMapper;
import com.etotem.cfc.mapper.TaskOrchestrationNodeInstanceMapper;
import com.etotem.cfc.service.OrchestrationEngine;
import lombok.extern.slf4j.Slf4j;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;

@Slf4j
@DisallowConcurrentExecution
public class OrchestrationPollingJob implements Job {

    @Resource
    private TaskOrchestrationExecutionMapper executionMapper;
    @Resource
    private TaskOrchestrationNodeInstanceMapper nodeInstanceMapper;
    @Resource
    private TaskMapper taskMapper;
    @Resource
    private OrchestrationEngine orchestrationEngine;

    @Override
    public void execute(JobExecutionContext context) {
        log.info("编排轮询开始...");
        try {
            // 1. 查询所有 running 状态的 execution（每次最多 20 个）
            List<TaskOrchestrationExecution> executions = executionMapper.selectList(
                new LambdaQueryWrapper<TaskOrchestrationExecution>()
                    .eq(TaskOrchestrationExecution::getStatus, "running")
                    .last("LIMIT 20"));
            for (TaskOrchestrationExecution exec : executions) {
                // 2. 检查 timeout 节点（deadline 到期）
                checkTimeouts(exec);
                // 3. 评估 pending 节点
                orchestrationEngine.evaluateFlow(exec.getId());
            }
        } catch (Exception e) {
            log.error("编排轮询异常", e);
        }
        log.info("编排轮询结束");
    }

    private void checkTimeouts(TaskOrchestrationExecution exec) {
        List<TaskOrchestrationNodeInstance> nodes = nodeInstanceMapper.selectList(
            new LambdaQueryWrapper<TaskOrchestrationNodeInstance>()
                .eq(TaskOrchestrationNodeInstance::getExecutionId, exec.getId())
                .eq(TaskOrchestrationNodeInstance::getStatus, "in_progress"));
        Date now = new Date();
        for (TaskOrchestrationNodeInstance ni : nodes) {
            if (ni.getTaskId() == null) continue;
            Task task = taskMapper.selectById(ni.getTaskId());
            if (task != null && task.getDeadline() != null && task.getDeadline().before(now)
                && !"completed".equals(task.getStatus())) {
                orchestrationEngine.onTaskTimeout(ni.getTaskId());
            }
        }
    }
}
```

> **注意**：Quartz 的 `Job` 实例由 Quartz 容器管理，`@Resource` 注入需要 `SpringBeanJobFactory` 支持。若注入失败，需在 `QuartzConfig` 配置 `SpringBeanJobFactory`（让 Quartz 使用 Spring 容器创建 Job 实例）。**实施时优先验证 `@Resource` 是否生效，不行则加 `SpringBeanJobFactory` bean。**

- [ ] **步骤 4：OrchestrationEngine 新增 onTaskTimeout**

```java
public void onTaskTimeout(Long taskId) {
    TaskOrchestrationNodeInstance ni = findByTaskId(taskId);
    if (ni == null) return;
    ni.setStatus("timeout");
    ni.setCompletedAt(new Date());
    nodeInstanceMapper.updateById(ni);
    TaskOrchestrationExecution exec = executionMapper.selectById(ni.getExecutionId());
    // 触发 timeout 边（兜底）
    List<TaskOrchestrationEdge> timeoutEdges = edgeMapper.selectList(
        new LambdaQueryWrapper<TaskOrchestrationEdge>()
            .eq(TaskOrchestrationEdge::getFlowId, exec.getFlowId())
            .eq(TaskOrchestrationEdge::getFromNodeId, ni.getNodeId())
            .eq(TaskOrchestrationEdge::getEdgeType, "timeout"));
    for (TaskOrchestrationEdge e : timeoutEdges) {
        // 幂等：检查同execution_id + to_node_id + generation=0 的 pending 实例是否存在
        Long exists = nodeInstanceMapper.selectCount(
            new LambdaQueryWrapper<TaskOrchestrationNodeInstance>()
                .eq(TaskOrchestrationNodeInstance::getExecutionId, exec.getId())
                .eq(TaskOrchestrationNodeInstance::getNodeId, e.getToNodeId())
                .eq(TaskOrchestrationNodeInstance::getGeneration, 0)
                .eq(TaskOrchestrationNodeInstance::getStatus, "pending"));
        if (exists != null && exists > 0) continue; // 已存在则跳过
        TaskOrchestrationNodeInstance newNi = new TaskOrchestrationNodeInstance();
        newNi.setExecutionId(exec.getId());
        newNi.setNodeId(e.getToNodeId());
        newNi.setGeneration(0);
        newNi.setStatus("pending");
        newNi.setCreatedAt(new Date());
        nodeInstanceMapper.insert(newNi);
    }
    evaluateFlow(exec.getId());
}
```

- [ ] **步骤 5：编译验证**

运行：`cd cfc-backend && mvn clean compile`
预期：BUILD SUCCESS。

- [ ] **步骤 6：Commit**

```bash
git add cfc-backend/pom.xml \
        cfc-backend/src/main/java/com/etotem/cfc/config/QuartzConfig.java \
        cfc-backend/src/main/java/com/etotem/cfc/task/OrchestrationPollingJob.java \
        cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationEngine.java
git commit -m "feat: Quartz 轮询 + timeout 检查 + 兜底边触发"
```

---

### 任务 5：后端 — Flow/Execution Service + Controller

**文件：**
- 创建：`cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationFlowService.java`
- 创建：`cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationExecutionService.java`
- 创建：`cfc-backend/src/main/java/com/etotem/cfc/controller/OrchestrationController.java`

- [ ] **步骤 1：创建 OrchestrationFlowService（Flow CRUD）**

按设计文档 §7.1 实现 6 个方法：`save` / `publish` / `list` / `detail` / `archive` / `delete`。核心逻辑（`save` 保存 config_json + edges 快照，`publish` 状态转换 + version+1）。

- [ ] **步骤 2：创建 OrchestrationExecutionService（Execution 生命周期）**

按 §7.2 实现：`start`（创建 execution + start_node 触发）/ `pause` / `resume` / `terminate` / `detail` / `list`。`start` 方法：

```java
@Transactional
public Result<Map<String, Object>> start(Long flowId, Long familyMemberId, Long userId) {
    TaskOrchestrationFlow flow = flowMapper.selectById(flowId);
    if (flow == null) return Result.error("编排流不存在");
    if (!"published".equals(flow.getStatus())) return Result.error("编排流未发布");

    // 创建执行实例
    TaskOrchestrationExecution exec = new TaskOrchestrationExecution();
    exec.setFlowId(flowId);
    exec.setFlowVersion(flow.getVersion());
    exec.setFamilyMemberId(familyMemberId);
    exec.setTriggerSource("manual");
    exec.setStatus("running");
    exec.setStartedAt(new Date());
    executionMapper.insert(exec);

    // 读取 config_json，创建 start_node 的 pending 实例
    JSONObject cfg = JSON.parseObject(flow.getConfigJson());
    for (Object o : cfg.getJSONArray("nodes")) {
        JSONObject n = (JSONObject) o;
        TaskOrchestrationNodeInstance ni = new TaskOrchestrationNodeInstance();
        ni.setExecutionId(exec.getId());
        ni.setNodeId(n.getString("id"));
        ni.setGeneration(0);
        if (n.getBooleanValue("is_start_node")) {
            ni.setStatus("pending"); // 交由 evaluateFlow 触发
        } else {
            ni.setStatus("pending");
        }
        ni.setCreatedAt(new Date());
        nodeInstanceMapper.insert(ni);
    }

    // 触发 start_node
    orchestrationEngine.evaluateFlow(exec.getId());

    Map<String, Object> data = new HashMap<>();
    data.put("executionId", exec.getId());
    return Result.success(data);
}
```

> **注意**：`familyMemberId` 对应 `FamilyMember.id`（主键），在 `start` 方法中已直接注入，无需额外反查。若需关联家庭属性（如 `familyId`、`isAdmin` 等），可通过 `FamilyMemberMapper.selectById(familyMemberId)` 实现。

- [ ] **步骤 3：创建 OrchestrationController**

统一 `@PostMapping`，路由前缀 `/api/orchestration`，注入两个 Service + 手动检查角色（`@RequestAttribute("role")`）。实现 §7.1/§7.2/§7.3 全部接口，含 `/node/fail`、`/node/restart`、`/node/condition-met`。

- [ ] **步骤 4：编译验证**

运行：`cd cfc-backend && mvn clean compile`
预期：BUILD SUCCESS。

- [ ] **步骤 5：Commit**

```bash
git add cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationFlowService.java \
        cfc-backend/src/main/java/com/etotem/cfc/service/OrchestrationExecutionService.java \
        cfc-backend/src/main/java/com/etotem/cfc/controller/OrchestrationController.java
git commit -m "feat: Flow/Execution Service + OrchestrationController REST 接口"
```

---

### 任务 6：后端 — 集成测试 + 全链路验证

**文件：**
- 创建：`cfc-backend/src/test/java/com/etotem/cfc/orchestration/OrchestrationIntegrationTest.java`

- [ ] **步骤 1：编写集成测试（完整流执行）**

覆盖：start → node1 完成 → node2 触发 → node3 超时 → 兜底 node4。用真实 Mapper（测试库）验证幂等（回调 + 轮询不重复建任务）。

- [ ] **步骤 2：运行全量测试**

运行：`cd cfc-backend && mvn test`
预期：PASS（含新增集成测试）。

- [ ] **步骤 3：启动服务手动验证**

运行：`cd cfc-backend && mvn spring-boot:run`（确认无 `Unknown column` 报错，4 张表迁移成功）。

- [ ] **步骤 4：Commit**

```bash
git add cfc-backend/src/test/java/com/etotem/cfc/orchestration/OrchestrationIntegrationTest.java
git commit -m "test: 编排引擎集成测试（完整流 + 幂等）"
```

---

### 任务 7：前端 — 流列表 + 画布 + 执行详情

**文件：**
- 创建：`cfc-web/src/views/orchestration/OrchestrationFlow.vue`
- 创建：`cfc-web/src/views/orchestration/FlowEditor.vue`
- 创建：`cfc-web/src/views/orchestration/ExecutionDetail.vue`
- 创建：`cfc-web/src/api/orchestration.js`
- 修改：`cfc-web/src/router/index.js`
- 修改：`cfc-web/package.json` — 新增 jsplumb

- [ ] **步骤 1：安装 jsplumb + 封装 API**

```bash
cd cfc-web && npm install jsplumb --save
```

`cfc-web/src/api/orchestration.js`：封装 12 个接口（flow 6 个 + execution 6 个）。

- [ ] **步骤 2：创建 OrchestrationFlow.vue（列表）**

Element UI 表格：流名称、版本、状态、创建时间、操作（编辑/发布/归档/删除/启动执行）。参考现有管理端列表页模式。

- [ ] **步骤 3：创建 FlowEditor.vue（jsPlumb 画布）**

- 左侧：可拖拽任务模板列表（`admin_task_templates` 接口）
- 中央：jsPlumb 画布（节点拖拽 + 连线 + 删除连线）
- 右侧：属性面板（超时时间、max_loops、终止条件、AND/OR 边类型）
- 工具栏：保存草稿 / 发布（发布前校验孤立节点、环必须有终止条件、task_template_ref 存在、有且仅一个 start_node）

- [ ] **步骤 4：创建 ExecutionDetail.vue（只读树）**

树形展示节点实例状态 + 关联 tasks 记录 + 手动干预（标记失败/终止流/重启节点/条件达成）。

- [ ] **步骤 5：注册路由**

`cfc-web/src/router/index.js` 新增：
```js
{ path: '/orchestration', component: OrchestrationFlow },
{ path: '/orchestration/edit/:id', component: FlowEditor },
{ path: '/orchestration/execution/:id', component: ExecutionDetail }
```

- [ ] **步骤 6：前端构建验证**

运行：`cd cfc-web && npm run build`
预期：构建成功。

- [ ] **步骤 7：Commit**

```bash
git add cfc-web/src/views/orchestration/ cfc-web/src/api/orchestration.js \
        cfc-web/src/router/index.js cfc-web/package.json cfc-web/package-lock.json
git commit -m "feat: 任务编排管理端（列表 + jsPlumb 画布 + 执行详情）"
```

---

## 范围边界（本计划明确不做）

| 不做 | 原因 |
|------|------|
| 小程序端编排编辑 | 编排编辑只放 cfc-web 管理端 |
| 流市场/模板市场 | 后续迭代 |
| 节点级并行执行 | 一个节点同一时刻只一个任务实例 |
| 家庭挑战/五维打卡合并 | 已有设计文档明确不合并 |
| flow 配置的版本快照隔离（edges 版本化） | 当前 edges 表读最新，版本隔离仅对 execution 记录 flow_version；后续如需精确快照再补 edges 版本列 |

---

## 验收标准

- [ ] 4 张新表创建成功，DDL 幂等可重复执行
- [ ] Flow CRUD 6 个接口通过
- [ ] Execution 6 个接口通过（含 start 触发 start_node）
- [ ] AND/OR 前置语义正确（单元测试覆盖）
- [ ] count/condition 循环终止正确（单元测试覆盖）
- [ ] timeout/failed 兜底边触发正确
- [ ] 幂等：回调 + 轮询不重复创建任务（集成测试覆盖）
- [ ] Quartz 30 秒轮询生效
- [ ] 前端画布可拖拽/连线/编辑/发布校验
- [ ] 执行详情树形展示正确
- [ ] 通过 `mvn clean compile` 和 `npm run build`