chore: move workflow/repo/documents/config under assets

assets/workflow/auto-dev-loop/workflow_engine/runner.py

@@ -0,0 +1,218 @@
+#!/usr/bin/env python3
+"""
+workflow_engine/runner.py - 轻量状态机调度器
+用于编排 step1~step5 的全自动开发闭环
+"""
+import json
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+
+BASE_DIR = Path(__file__).parent.parent
+STATE_FILE = BASE_DIR / "workflow_engine/state/current_step.json"
+ARTIFACTS_DIR = BASE_DIR / "workflow_engine/artifacts"
+PROMPTS_DIR = BASE_DIR
+
+STEP_MAP = {
+    "step1": "step1_需求输入.jsonl",
+    "step2": "step2_执行计划.jsonl",
+    "step3": "step3_实施变更.jsonl",
+    "step4": "step4_验证发布.jsonl",
+    "step5": "step5_总控与循环.jsonl",
+}
+
+STEP_FLOW = ["step1", "step2", "step3", "step4", "step5"]
+
+
+def load_state() -> dict:
+    if STATE_FILE.exists():
+        return json.loads(STATE_FILE.read_text(encoding="utf-8"))
+    return {"run_id": None, "step": None, "status": "idle"}
+
+
+def save_state(state: dict):
+    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
+    STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def get_run_id() -> str:
+    return datetime.now().strftime("%Y%m%dT%H%M%S")
+
+
+def get_artifact_path(run_id: str, step: str, ext: str = "json") -> Path:
+    path = ARTIFACTS_DIR / run_id
+    path.mkdir(parents=True, exist_ok=True)
+    return path / f"{step}.{ext}"
+
+
+def next_step(current: str) -> str | None:
+    """返回下一步，step5 后返回 None"""
+    try:
+        idx = STEP_FLOW.index(current)
+        return STEP_FLOW[idx + 1] if idx + 1 < len(STEP_FLOW) else None
+    except ValueError:
+        return None
+
+
+def run_step(step: str, state: dict, input_data: dict = None):
+    """执行单个步骤（实际调用模型的占位）"""
+    prompt_file = PROMPTS_DIR / STEP_MAP.get(step, "")
+    if not prompt_file.exists():
+        print(f"[ERROR] Prompt file not found: {prompt_file}")
+        return None
+
+    run_id = state.get("run_id") or get_run_id()
+    
+    # 更新状态为 running
+    state.update({"run_id": run_id, "step": step, "status": "running"})
+    save_state(state)
+    
+    print(f"[RUN] {step} | run_id={run_id}")
+    print(f"      prompt: {prompt_file.name}")
+    
+    # === 这里是模型调用占位 ===
+    # 实际实现时替换为：
+    # result = call_llm(prompt_file.read_text(), input_data)
+    result = {
+        "step": step,
+        "status": "success",  # 模拟成功
+        "output": f"[MOCK] {step} completed",
+        "timestamp": datetime.now().isoformat()
+    }
+    # ========================
+    
+    # 保存产物
+    artifact_path = get_artifact_path(run_id, step)
+    artifact_path.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(f"      artifact: {artifact_path}")
+    
+    return result
+
+
+def dispatch():
+    """根据当前状态分发到下一步"""
+    state = load_state()
+    target = state.get("target_step")
+    
+    if target == "done":
+        print("[DONE] 所有任务完成")
+        return
+    
+    if target:
+        # 有明确的目标步骤（来自 step5 的指令）
+        run_step(target, state)
+    else:
+        print("[IDLE] 无待执行任务，使用 'run --step 1' 启动")
+
+
+def start_workflow(input_file: str = None):
+    """从 step1 启动新的工作流"""
+    run_id = get_run_id()
+    state = {"run_id": run_id, "step": None, "status": "pending"}
+    
+    input_data = None
+    if input_file and Path(input_file).exists():
+        input_data = json.loads(Path(input_file).read_text(encoding="utf-8"))
+    
+    print(f"[START] 新工作流 run_id={run_id}")
+    
+    # 依次执行 step1 -> step5
+    for step in STEP_FLOW:
+        result = run_step(step, state, input_data)
+        
+        if not result:
+            state["status"] = "error"
+            save_state(state)
+            return
+        
+        # step4 后检查验证结果
+        if step == "step4":
+            verify_status = result.get("verify", {}).get("status", "success")
+            state["verify"] = {"status": verify_status}
+        
+        # step5 决定下一步
+        if step == "step5":
+            # 模拟 step5 的决策逻辑
+            if state.get("verify", {}).get("status") == "failed":
+                state["target_step"] = "step2"  # 失败回跳
+                state["status"] = "retry"
+                print(f"[RETRY] 验证失败，返回 step2 重规划")
+            else:
+                state["target_step"] = "done"
+                state["status"] = "completed"
+                print(f"[COMPLETE] 工作流完成")
+        
+        save_state(state)
+        
+        # 如果需要回跳，递归处理（带熔断）
+        if state.get("target_step") == "step2":
+            retry_count = state.get("retry_count", 0) + 1
+            if retry_count > 3:
+                print(f"[FATAL] 超过最大重试次数")
+                state["status"] = "fatal_error"
+                save_state(state)
+                return
+            state["retry_count"] = retry_count
+            print(f"[RETRY {retry_count}/3] 从 step2 重新执行")
+            
+            # 递归调用自身，复用完整的验证逻辑
+            state["target_step"] = None  # 清除回跳标记
+            save_state(state)
+            
+            for retry_step in STEP_FLOW[1:]:  # step2 onwards
+                result = run_step(retry_step, state, input_data)
+                if not result:
+                    state["status"] = "error"
+                    save_state(state)
+                    return
+                
+                # step4 后检查验证结果
+                if retry_step == "step4":
+                    verify_status = result.get("verify", {}).get("status", "success")
+                    state["verify"] = {"status": verify_status}
+                
+                # step5 决定下一步
+                if retry_step == "step5":
+                    if state.get("verify", {}).get("status") == "failed":
+                        state["target_step"] = "step2"
+                        state["status"] = "retry"
+                        print(f"[RETRY] 验证仍失败，继续重试")
+                        break  # 跳出内层循环，触发外层重试检查
+                    else:
+                        state["target_step"] = "done"
+                        state["status"] = "completed"
+                        print(f"[COMPLETE] 重试成功，工作流完成")
+                
+                save_state(state)
+            
+            # 如果 step5 设置了新的回跳目标，继续递归
+            if state.get("target_step") == "step2":
+                continue  # 继续外层循环的下一次迭代（实际会被 break 跳出）
+            break
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("Usage:")
+        print("  python runner.py start [input.json]  - 启动新工作流")
+        print("  python runner.py dispatch            - 根据状态分发")
+        print("  python runner.py status              - 查看当前状态")
+        return
+    
+    cmd = sys.argv[1]
+    
+    if cmd == "start":
+        input_file = sys.argv[2] if len(sys.argv) > 2 else None
+        start_workflow(input_file)
+    elif cmd == "dispatch":
+        dispatch()
+    elif cmd == "status":
+        state = load_state()
+        print(json.dumps(state, ensure_ascii=False, indent=2))
+    else:
+        print(f"Unknown command: {cmd}")
+
+
+if __name__ == "__main__":
+    main()