基于 OpenClaw + Claude Code 的端到端研发自动化系统
任务暂停、人工审核、流程重启逻辑完整实现
本系统实现了完整的人机协同研发自动化流程，支持在任意研发节点进行任务暂停、人工审核和流程重启。系统集成 OpenClaw 框架和 Claude Code AI 助手，覆盖从需求分析到最终验收的全流程自动化。
系统采用有限状态机 (FSM) 管理任务生命周期，确保状态转换的正确性和可追溯性。
| 当前状态 | 触发动作 | 下一状态 | 说明 |
|---|---|---|---|
| PENDING | start() | RUNNING | 任务开始执行 |
| RUNNING | pause() | PAUSED | 创建检查点后暂停 |
| RUNNING / PAUSED | request_review() | AWAITING_REVIEW | 请求人工审核 |
| AWAITING_REVIEW | submit(APPROVE) | REVIEW_APPROVED | 审核通过 |
| AWAITING_REVIEW | submit(REJECT) | REVIEW_REJECTED | 审核拒绝 |
| AWAITING_REVIEW | submit(REQUEST_CHANGES) | PAUSED | 要求修改 |
| PAUSED / REJECTED | restart() | RUNNING | 从检查点重启 |
完整的研发自动化流程包含 13 个节点，其中 7 个节点需要人工审核。
每个节点都有对应的处理器类，负责执行具体的业务逻辑：
class HumanInLoopManager: """人机协同管理器 - 核心控制类""" def pause_task(self, task_id: str, reason: str = "") -> bool: """暂停任务并创建检查点""" task = self.tasks[task_id] # 创建检查点保存上下文 checkpoint = TaskCheckpoint( node=task.current_node, status=task.status, context=task.metadata.copy(), artifacts=[] ) task.checkpoints.append(checkpoint) # 更新状态 task.status = TaskStatus.PAUSED task.metadata["pause_reason"] = reason return True def request_review(self, task_id: str, reviewer_id: str, message: str = "") -> bool: """请求人工审核""" task = self.tasks[task_id] # 自动暂停如果正在运行 if task.status == TaskStatus.RUNNING: self.pause_task(task_id, "Awaiting review") # 创建待审核记录 task.pending_review = ReviewComment( reviewer_id=reviewer_id, decision=ReviewDecision.REQUEST_CHANGES, comments=message ) task.status = TaskStatus.AWAITING_REVIEW return True def submit_review(self, task_id: str, reviewer_id: str, decision: ReviewDecision, comments: str = "") -> bool: """提交审核决策""" task = self.tasks[task_id] # 记录审核历史 review_comment = ReviewComment( reviewer_id=reviewer_id, decision=decision, comments=comments ) task.review_history.append(review_comment) # 根据决策更新状态 if decision == ReviewDecision.APPROVE: task.status = TaskStatus.REVIEW_APPROVED elif decision == ReviewDecision.REJECT: task.status = TaskStatus.REVIEW_REJECTED else: task.status = TaskStatus.PAUSED return True def restart_task(self, task_id: str, from_checkpoint: int = -1, target_node: Optional[RDNode] = None) -> bool: """重启任务 - 从检查点或指定节点恢复""" task = self.tasks[task_id] # 确定重启点 if target_node: task.current_node = target_node elif task.checkpoints: checkpoint = task.checkpoints[from_checkpoint] task.current_node = checkpoint.node task.metadata.update(checkpoint.context) task.status = TaskStatus.RUNNING return True
系统提供完整的 RESTful API 用于任务管理和审核流程控制。
完整的测试套件覆盖所有核心功能，确保系统稳定性和正确性。
| 测试类别 | 测试用例数 | 覆盖功能 |
|---|---|---|
| TestTaskStateMachine | 5 | 任务创建、序列化、状态转换 |
| TestPauseResumeFunctionality | 4 | 暂停/恢复、检查点创建、元数据保存 |
| TestReviewWorkflow | 7 | 审核请求、提交决策、审核历史 |
| TestCheckpointAndRestart | 5 | 检查点恢复、指定节点重启、上下文保留 |
| TestWorkflowEngineIntegration | 3 | 工作流引擎初始化、状态报告 |
| TestExportImport | 2 | JSON 导出导入、往返序列化 |
| TestEdgeCases | 4 | 边界条件、错误处理、并发场景 |
from task_state_machine import HumanInLoopManager, ReviewDecision from workflow_engine import WorkflowEngine # 1. 初始化管理器和工作流引擎 manager = HumanInLoopManager() engine = WorkflowEngine(manager) # 2. 创建研发任务 task = manager.create_task( name="电商平台用户中心开发", description="完整的用户中心模块，包括注册、登录、个人中心", start_node=RDNode.REQUIREMENT_ANALYSIS ) # 3. 启动任务执行 task.status = TaskStatus.RUNNING # 4. 执行到第一个审核点（需求分析完成） await engine.execute_task(task.task_id) # 此时任务状态：AWAITING_REVIEW # 5. 人工审核 - 通过 manager.submit_review( task.task_id, reviewer_id="reviewer_001", decision=ReviewDecision.APPROVE, comments="需求清晰，可以进入下一阶段" ) # 6. 继续执行到 PRD 设计节点 task.status = TaskStatus.RUNNING await engine.execute_task(task.task_id) # 7. 假设需要暂停进行修改 manager.pause_task(task.task_id, "等待产品确认") # 8. 审核后拒绝，要求重新设计 manager.submit_review( task.task_id, reviewer_id="reviewer_001", decision=ReviewDecision.REJECT, comments="PRD 需要补充更多细节" ) # 9. 从最新检查点重启 manager.restart_task(task.task_id, from_checkpoint=-1) # 10. 继续执行直到完成 await engine.run_pipeline(task.task_id)