beige/.claude/skills/clover-loop/scripts/loop_state.py

#!/usr/bin/env python3
"""
Clover Loop State Manager（改造自 ralph-loop state_manager）
六阶段 + 5态 + 机器/人工双层验收 + Pivot 追踪。
"""
import json
import sys
from datetime import datetime
from pathlib import Path

STATE_FILE = ".clover-loop-state.json"
PHASES = {"0": "调研", "1": "约束", "2": "生成", "3": "验证", "4": "校准", "5": "闭环"}


def _path(work_dir=None):
    base = Path(work_dir) if work_dir else Path.cwd()
    return base / STATE_FILE


def load(work_dir=None):
    p = _path(work_dir)
    return json.load(open(p, encoding="utf-8")) if p.exists() else None


def save(state, work_dir=None):
    state["updated_at"] = datetime.now().isoformat()
    json.dump(state, open(_path(work_dir), "w", encoding="utf-8"),
              indent=2, ensure_ascii=False)


def create(task_id, work_dir=None):
    state = {
        "task_id": task_id,
        "created_at": datetime.now().isoformat(),
        "updated_at": datetime.now().isoformat(),
        "current_phase": 0,
        "max_retries_per_layer": 2,   # 单层撞墙2次→Pivot
        "machine_gate": {"lint": None, "test": None, "banned_word": None,
                         "score": None, "dedup": None},
        "human_review": {"status": "pending", "sampled": 0,
                         "approved": 0, "rejected": 0},
        "status": "in_progress",      # in_progress/machine_passed/human_passed/pivoted/blocked
        "layer_attempts": {},         # {层名: 失败次数}
        "pivots": [],
        "logs": [],
    }
    save(state, work_dir)
    return state


def advance(state, work_dir=None):
    cur = state["current_phase"]
    if cur < 5:
        state["current_phase"] = cur + 1
    save(state, work_dir)
    return state


def machine_gate(state, key, result, work_dir=None):
    """记录机器门禁单项结果。全过则置 machine_passed。"""
    state["machine_gate"][key] = result
    g = state["machine_gate"]
    passed = all(v in ("pass", True) or (isinstance(v, (int, float)) and v >= 90)
                 for v in g.values() if v is not None)
    if passed and all(v is not None for v in g.values()):
        state["status"] = "machine_passed"
    save(state, work_dir)
    return state


def record_fail(state, layer, error=None, work_dir=None):
    """记录某层失败。同层2次→强制Pivot。"""
    state["layer_attempts"][layer] = state["layer_attempts"].get(layer, 0) + 1
    state["logs"].append({"ts": datetime.now().isoformat(),
                          "layer": layer, "error": error})
    if state["layer_attempts"][layer] >= state["max_retries_per_layer"]:
        state["status"] = "pivoted"
        state["pivots"].append({"layer": layer, "ts": datetime.now().isoformat(),
                                "reason": f"{layer}层撞墙{state['layer_attempts'][layer]}次"})
    save(state, work_dir)
    return state


def human_review(state, action, work_dir=None):
    """北哥抽检：approve/reject。approve累计达标→human_passed。"""
    hr = state["human_review"]
    hr["sampled"] += 1
    if action == "approve":
        hr["approved"] += 1
        hr["status"] = "approved"
        state["status"] = "human_passed"
    elif action == "reject":
        hr["rejected"] += 1
        hr["status"] = "rejected"
        state["status"] = "blocked"   # 打回=停循环请人介入
    save(state, work_dir)
    return state


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("用法: loop_state.py <create|load|advance|gate|fail|review> [args]")
        print("  create <task_id> / advance / gate <key> <result>")
        print("  fail <layer> [error] / review <approve|reject>")
        sys.exit(1)
    cmd = sys.argv[1]
    if cmd == "create":
        print(json.dumps(create(sys.argv[2]), ensure_ascii=False, indent=2))
    elif cmd == "load":
        s = load()
        print(json.dumps(s, ensure_ascii=False, indent=2) if s else "无状态")
    elif cmd == "advance":
        s = load()
        if s:
            advance(s)
            print(f"进入 P{s['current_phase']} {PHASES[str(s['current_phase'])]}")
    elif cmd == "gate":
        s = load()
        if s:
            key, raw = sys.argv[2], sys.argv[3]
            val = float(raw) if raw.replace(".", "").isdigit() else raw
            machine_gate(s, key, val)
            print(f"门禁 {key}={val}, status={s['status']}")
    elif cmd == "fail":
        s = load()
        if s:
            record_fail(s, sys.argv[2], sys.argv[3] if len(sys.argv) > 3 else None)
            print(f"{sys.argv[2]}层失败{s['layer_attempts'][sys.argv[2]]}次, status={s['status']}")
    elif cmd == "review":
        s = load()
        if s:
            human_review(s, sys.argv[2])
            print(f"抽检{sys.argv[2]}, status={s['status']}")