✨ feat(workflow): add superpowers planning and execution state tracking

scripts/main_loop.py

@@ -1,12 +1,28 @@
 #!/usr/bin/env python3
+from contextlib import contextmanager
+import os
 import platform
 import re
 import sys
+import threading
+import time
 from pathlib import Path
 from typing import Optional
 
+try:
+    import fcntl
+except ImportError:  # pragma: no cover
+    fcntl = None
+
+try:
+    import msvcrt
+except ImportError:  # pragma: no cover
+    msvcrt = None
+
 PLAN_STATUS_START = "<!-- plan-status:start -->"
 PLAN_STATUS_END = "<!-- plan-status:end -->"
+WORKFLOW_STATE_START = "<!-- workflow-state:start -->"
+WORKFLOW_STATE_END = "<!-- workflow-state:end -->"
 PLAN_FILE_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})-.+\.md$")
 PLAN_LINE_RE = re.compile(
     r"^- \[(?P<check>[ xX])\] `(?P<plan>[^`]+)` "
@@ -15,6 +31,9 @@ PLAN_LINE_RE = re.compile(
 )
 FINISH_STATUSES = {"done", "blocked", "skipped"}
 ENV_BLOCKED_RE = re.compile(r"^env:([^:]+):(.+)$")
+WORKFLOW_PHASES = {"brainstorming", "planning", "executing", "done", "blocked"}
+THREAD_LOCKS: dict[str, threading.Lock] = {}
+THREAD_LOCKS_GUARD = threading.Lock()
 
 
 def usage() -> str:
@@ -23,12 +42,19 @@ def usage() -> str:
         "  python scripts/main_loop.py claim -plans <dir> -progress <file>\n"
         "  python scripts/main_loop.py finish -plan <path> -status <status> "
         "-progress <file> [-note <text>]\n"
+        "  python scripts/main_loop.py record -progress <file> -phase <phase> "
+        "[-spec <path>] [-plan <path>] [-executor <name>] "
+        "[-constraints <csv>]\n"
         "  python scripts/main_loop.py -h\n"
         "Options:\n"
         "  -plans DIR\n"
         "  -plan PATH\n"
         "  -status done|blocked|skipped\n"
         "  -progress FILE\n"
+        "  -phase brainstorming|planning|executing|done|blocked\n"
+        "  -spec PATH\n"
+        "  -executor NAME\n"
+        "  -constraints CSV\n"
         "  -note TEXT\n"
         "  -h, -help Show this help.\n"
     )
@@ -53,9 +79,9 @@ def parse_flags(args: list[str]) -> dict[str, str]:
 def normalize_plan_key(plan_value: str) -> str:
     raw = plan_value.strip().replace("\\", "/")
     raw = raw.lstrip("./")
-    if raw.startswith("docs/plans/"):
-        return raw[len("docs/plans/") :]
-    marker = "/docs/plans/"
+    if raw.startswith("docs/superpowers/plans/"):
+        return raw[len("docs/superpowers/plans/") :]
+    marker = "/docs/superpowers/plans/"
     if marker in raw:
         return raw.split(marker, 1)[1]
     return raw
@@ -98,6 +124,22 @@ def find_block(lines: list[str]) -> Optional[tuple[int, int]]:
     return None
 
 
+def find_named_block(
+    lines: list[str], start_marker: str, end_marker: str
+) -> Optional[tuple[int, int]]:
+    start_idx = None
+    for idx, line in enumerate(lines):
+        if line.strip() == start_marker:
+            start_idx = idx
+            break
+    if start_idx is None:
+        return None
+    for idx in range(start_idx + 1, len(lines)):
+        if lines[idx].strip() == end_marker:
+            return start_idx, idx
+    return None
+
+
 def parse_entries(
     lines: list[str], start_idx: int, end_idx: int
 ) -> list[tuple[str, str, Optional[str], int]]:
@@ -115,13 +157,60 @@ def parse_entries(
 
 
 def render_progress_lines(plans: list[str]) -> list[str]:
-    lines = ["# Plan 状态", "", PLAN_STATUS_START]
+    lines = [
+        "# 当前进展",
+        "",
+        "## Workflow State",
+        "",
+        WORKFLOW_STATE_START,
+        WORKFLOW_STATE_END,
+        "",
+        "## Plan Status",
+        "",
+        PLAN_STATUS_START,
+    ]
     for plan_key in plans:
         lines.append(render_plan_line(plan_key, "pending", None))
     lines.append(PLAN_STATUS_END)
     return lines
 
 
+def render_workflow_state_lines(
+    phase: Optional[str] = None,
+    spec: Optional[str] = None,
+    plan: Optional[str] = None,
+    executor: Optional[str] = None,
+    constraints: Optional[str] = None,
+) -> list[str]:
+    lines = [WORKFLOW_STATE_START]
+    if phase:
+        lines.append(f"phase: {phase}")
+    if spec:
+        lines.append(f"spec: {spec}")
+    if plan:
+        lines.append(f"plan: {plan}")
+    if executor:
+        lines.append(f"executor: {executor}")
+    if constraints:
+        lines.append(f"constraints: {constraints}")
+    lines.append(WORKFLOW_STATE_END)
+    return lines
+
+
+def parse_workflow_state(
+    lines: list[str], start_idx: int, end_idx: int
+) -> dict[str, str]:
+    state: dict[str, str] = {}
+    for idx in range(start_idx + 1, end_idx):
+        line = lines[idx].strip()
+        if ": " not in line:
+            continue
+        key, value = line.split(": ", 1)
+        if key in {"phase", "spec", "plan", "executor", "constraints"}:
+            state[key] = value
+    return state
+
+
 def parse_env_blocked_note(note: Optional[str]) -> Optional[tuple[str, str]]:
     if not note:
         return None
@@ -147,17 +236,132 @@ def write_progress_lines(progress_path: Path, lines: list[str]) -> None:
     progress_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
 
 
+def get_thread_lock(lock_path: Path) -> threading.Lock:
+    key = str(lock_path.resolve())
+    with THREAD_LOCKS_GUARD:
+        lock = THREAD_LOCKS.get(key)
+        if lock is None:
+            lock = threading.Lock()
+            THREAD_LOCKS[key] = lock
+        return lock
+
+
+@contextmanager
+def locked_progress(progress_path: Path):
+    progress_path.parent.mkdir(parents=True, exist_ok=True)
+    lock_path = progress_path.with_name(f"{progress_path.name}.lock")
+    thread_lock = get_thread_lock(lock_path)
+    with thread_lock:
+        with lock_path.open("a+b") as lock_file:
+            if fcntl is not None:
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
+            elif msvcrt is not None:  # pragma: no cover
+                while True:
+                    try:
+                        lock_file.seek(0)
+                        msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
+                        break
+                    except OSError:
+                        time.sleep(0.05)
+            try:
+                hold_ms = os.environ.get("PLAYBOOK_MAIN_LOOP_HOLD_LOCK_MS")
+                if hold_ms:
+                    time.sleep(max(0.0, float(hold_ms) / 1000.0))
+                yield
+            finally:
+                if fcntl is not None:
+                    fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+                elif msvcrt is not None:  # pragma: no cover
+                    lock_file.seek(0)
+                    msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+
+
+def ensure_section(lines: list[str], heading: str) -> list[str]:
+    if any(line.strip() == heading for line in lines):
+        return lines
+    if lines and lines[-1] != "":
+        lines.append("")
+    lines.extend([heading, ""])
+    return lines
+
+
+def ensure_block_with_lines(
+    lines: list[str],
+    start_marker: str,
+    end_marker: str,
+    default_lines: list[str],
+    heading: Optional[str] = None,
+) -> tuple[list[str], int, int]:
+    block = find_named_block(lines, start_marker, end_marker)
+    if block:
+        return lines, block[0], block[1]
+
+    if not lines:
+        lines = ["# 当前进展", ""]
+    if heading:
+        lines = ensure_section(lines, heading)
+    if lines and lines[-1] != "":
+        lines.append("")
+    insert_at = len(lines)
+    lines[insert_at:insert_at] = default_lines
+    return lines, insert_at, insert_at + len(default_lines) - 1
+
+
 def ensure_plan_block(
     lines: list[str], progress_path: Path, plan_keys: list[str]
 ) -> tuple[list[str], int, int]:
-    block = find_block(lines)
-    if not block:
-        lines = render_progress_lines(plan_keys)
-        write_progress_lines(progress_path, lines)
-        block = find_block(lines)
-        if not block:
-            raise ValueError("failed to create plan status block")
-    return lines, block[0], block[1]
+    lines, _, _ = ensure_workflow_state_block(lines)
+    lines, start_idx, end_idx = ensure_block_with_lines(
+        lines,
+        PLAN_STATUS_START,
+        PLAN_STATUS_END,
+        [PLAN_STATUS_START, PLAN_STATUS_END],
+        "## Plan Status",
+    )
+    write_progress_lines(progress_path, lines)
+    return lines, start_idx, end_idx
+
+
+def ensure_workflow_state_block(
+    lines: list[str],
+) -> tuple[list[str], int, int]:
+    return ensure_block_with_lines(
+        lines,
+        WORKFLOW_STATE_START,
+        WORKFLOW_STATE_END,
+        [WORKFLOW_STATE_START, WORKFLOW_STATE_END],
+        "## Workflow State",
+    )
+
+
+def update_workflow_state(
+    lines: list[str],
+    phase: Optional[str] = None,
+    spec: Optional[str] = None,
+    plan: Optional[str] = None,
+    executor: Optional[str] = None,
+    constraints: Optional[str] = None,
+) -> list[str]:
+    lines, start_idx, end_idx = ensure_workflow_state_block(lines)
+    state = parse_workflow_state(lines, start_idx, end_idx)
+    if phase is not None:
+        state["phase"] = phase
+    if spec is not None:
+        state["spec"] = spec
+    if plan is not None:
+        state["plan"] = plan
+    if executor is not None:
+        state["executor"] = executor
+    if constraints is not None:
+        state["constraints"] = constraints
+    lines[start_idx : end_idx + 1] = render_workflow_state_lines(
+        state.get("phase"),
+        state.get("spec"),
+        state.get("plan"),
+        state.get("executor"),
+        state.get("constraints"),
+    )
+    return lines
 
 
 def ensure_all_plans_present(
@@ -175,6 +379,13 @@ def ensure_all_plans_present(
     return entries
 
 
+def filter_existing_entries(
+    entries: list[tuple[str, str, Optional[str], int]], plan_keys: list[str]
+) -> list[tuple[str, str, Optional[str], int]]:
+    available = set(plan_keys)
+    return [entry for entry in entries if entry[0] in available]
+
+
 def choose_claim_entry(
     entries: list[tuple[str, str, Optional[str], int]], current_env: Optional[str]
 ) -> Optional[tuple[str, Optional[str], int]]:
@@ -205,20 +416,27 @@ def claim_plan(plans_dir: Path, progress_path: Path) -> tuple[int, str]:
     if not plan_keys:
         return 2, "ERROR: no plan files found"
 
-    lines = load_progress_lines(progress_path)
-    try:
-        lines, start_idx, end_idx = ensure_plan_block(lines, progress_path, plan_keys)
-    except ValueError as exc:
-        return 2, f"ERROR: {exc}"
+    with locked_progress(progress_path):
+        lines = load_progress_lines(progress_path)
+        try:
+            lines, start_idx, end_idx = ensure_plan_block(lines, progress_path, plan_keys)
+        except ValueError as exc:
+            return 2, f"ERROR: {exc}"
 
-    entries = ensure_all_plans_present(lines, start_idx, end_idx, progress_path, plan_keys)
-    chosen = choose_claim_entry(entries, detect_env())
-    if not chosen:
-        return 2, "ERROR: no claimable plans"
+        entries = ensure_all_plans_present(lines, start_idx, end_idx, progress_path, plan_keys)
+        entries = filter_existing_entries(entries, plan_keys)
+        chosen = choose_claim_entry(entries, detect_env())
+        if not chosen:
+            return 0, "NOOP: no claimable plans"
 
-    plan_key, note, idx = chosen
-    lines[idx] = render_plan_line(plan_key, "in-progress", note)
-    write_progress_lines(progress_path, lines)
+        plan_key, note, idx = chosen
+        lines[idx] = render_plan_line(plan_key, "in-progress", note)
+        lines = update_workflow_state(
+            lines,
+            phase="executing",
+            plan=(plans_dir / plan_key).as_posix(),
+        )
+        write_progress_lines(progress_path, lines)
 
     output = [f"PLAN={(plans_dir / plan_key).as_posix()}"]
     if note:
@@ -234,27 +452,58 @@ def finish_plan(
     if not plan:
         return 2, "ERROR: plan is required"
 
-    lines = load_progress_lines(progress_path)
     plan_key = normalize_plan_key(plan)
+    with locked_progress(progress_path):
+        lines = load_progress_lines(progress_path)
 
-    try:
-        lines, start_idx, end_idx = ensure_plan_block(lines, progress_path, [plan_key])
-    except ValueError as exc:
-        return 2, f"ERROR: {exc}"
+        try:
+            lines, start_idx, end_idx = ensure_plan_block(lines, progress_path, [plan_key])
+        except ValueError as exc:
+            return 2, f"ERROR: {exc}"
 
-    entries = parse_entries(lines, start_idx, end_idx)
-    rendered_note = normalize_note(note) if note else None
-    updated_line = render_plan_line(plan_key, status, rendered_note)
+        entries = parse_entries(lines, start_idx, end_idx)
+        rendered_note = normalize_note(note) if note else None
+        updated_line = render_plan_line(plan_key, status, rendered_note)
 
-    for entry_plan, _, _, idx in entries:
-        if entry_plan == plan_key:
-            lines[idx] = updated_line
-            write_progress_lines(progress_path, lines)
-            return 0, updated_line
+        for entry_plan, _, _, idx in entries:
+            if entry_plan == plan_key:
+                lines[idx] = updated_line
+                workflow_phase = "done" if status == "done" else "blocked"
+                lines = update_workflow_state(
+                    lines,
+                    phase=workflow_phase,
+                    plan=f"docs/superpowers/plans/{plan_key}",
+                )
+                write_progress_lines(progress_path, lines)
+                return 0, updated_line
 
-    lines[end_idx:end_idx] = [updated_line]
-    write_progress_lines(progress_path, lines)
-    return 0, updated_line
+        lines[end_idx:end_idx] = [updated_line]
+        workflow_phase = "done" if status == "done" else "blocked"
+        lines = update_workflow_state(
+            lines,
+            phase=workflow_phase,
+            plan=f"docs/superpowers/plans/{plan_key}",
+        )
+        write_progress_lines(progress_path, lines)
+        return 0, updated_line
+
+
+def record_workflow_state(
+    progress_path: Path,
+    phase: str,
+    spec: Optional[str],
+    plan: Optional[str],
+    executor: Optional[str],
+    constraints: Optional[str],
+) -> tuple[int, str]:
+    if phase not in WORKFLOW_PHASES:
+        return 2, f"ERROR: invalid phase: {phase}"
+
+    with locked_progress(progress_path):
+        lines = load_progress_lines(progress_path)
+        lines = update_workflow_state(lines, phase, spec, plan, executor, constraints)
+        write_progress_lines(progress_path, lines)
+    return 0, "OK"
 
 
 def main(argv: list[str]) -> int:
@@ -266,7 +515,7 @@ def main(argv: list[str]) -> int:
         return 0
 
     mode = argv[0]
-    if mode not in {"claim", "finish"}:
+    if mode not in {"claim", "finish", "record"}:
         print(f"ERROR: unknown mode: {mode}", file=sys.stderr)
         print(usage(), file=sys.stderr)
         return 2
@@ -295,6 +544,26 @@ def main(argv: list[str]) -> int:
         print(message)
         return 0
 
+    if mode == "record":
+        progress = flags.get("-progress")
+        phase = flags.get("-phase")
+        spec = flags.get("-spec")
+        plan = flags.get("-plan")
+        executor = flags.get("-executor")
+        constraints = flags.get("-constraints")
+        if not progress or not phase:
+            print("ERROR: -progress and -phase are required", file=sys.stderr)
+            print(usage(), file=sys.stderr)
+            return 2
+        code, message = record_workflow_state(
+            Path(progress), phase, spec, plan, executor, constraints
+        )
+        if code != 0:
+            print(message, file=sys.stderr)
+            return code
+        print(message)
+        return 0
+
     plan = flags.get("-plan")
     status = flags.get("-status")
     progress = flags.get("-progress")

scripts/playbook.py

@@ -5,6 +5,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from shutil import copy2, copytree, rmtree, which
 import subprocess
+import importlib.util
 
 try:
     import tomllib
@@ -22,6 +23,11 @@ ORDER = [
 ]
 SCRIPT_DIR = Path(__file__).resolve().parent
 PLAYBOOK_ROOT = SCRIPT_DIR.parent
+MAIN_LOOP_SCRIPT = SCRIPT_DIR / "main_loop.py"
+MAIN_LOOP_SPEC = importlib.util.spec_from_file_location("playbook_main_loop", MAIN_LOOP_SCRIPT)
+assert MAIN_LOOP_SPEC and MAIN_LOOP_SPEC.loader
+MAIN_LOOP = importlib.util.module_from_spec(MAIN_LOOP_SPEC)
+MAIN_LOOP_SPEC.loader.exec_module(MAIN_LOOP)
 PATH_CONFIG_KEYS = {"project_root", "deploy_root", "agents_home", "codex_home", "skill_link"}
 DOCS_INDEX_SECTION_HEADINGS = {
     "common": "## 跨语言（common）",
@@ -34,7 +40,23 @@ DOCS_INDEX_SECTION_HEADINGS = {
 
 
 def usage() -> str:
-    return "Usage:\n  python scripts/playbook.py -config <path>\n  python scripts/playbook.py -h"
+    return (
+        "Usage:\n"
+        "  python scripts/playbook.py -config <path>\n"
+        "  python scripts/playbook.py -record-spec <spec_path> -progress <path>\n"
+        "  python scripts/playbook.py -record-plan <plan_path> -progress <path>\n"
+        "  python scripts/playbook.py -h"
+    )
+
+
+def parse_cli_value(argv: list[str], flag: str) -> Optional[str]:
+    if flag not in argv:
+        return None
+    idx = argv.index(flag)
+    if idx + 1 >= len(argv):
+        return None
+    value = argv[idx + 1].strip()
+    return value or None
 
 
 def strip_inline_comment(value: str) -> str:
@@ -1393,6 +1415,47 @@ def main(argv: list[str]) -> int:
     if "-h" in argv or "-help" in argv:
         print(usage())
         return 0
+
+    spec_path = parse_cli_value(argv, "-record-spec")
+    if spec_path is not None:
+        progress_path = parse_cli_value(argv, "-progress")
+        if not progress_path:
+            print("ERROR: -progress is required.\n" + usage(), file=sys.stderr)
+            return 2
+        code, message = MAIN_LOOP.record_workflow_state(
+            Path(progress_path),
+            "planning",
+            spec_path,
+            None,
+            None,
+            None,
+        )
+        if code != 0:
+            print(message, file=sys.stderr)
+            return code
+        print(message)
+        return 0
+
+    plan_path = parse_cli_value(argv, "-record-plan")
+    if plan_path is not None:
+        progress_path = parse_cli_value(argv, "-progress")
+        if not progress_path:
+            print("ERROR: -progress is required.\n" + usage(), file=sys.stderr)
+            return 2
+        code, message = MAIN_LOOP.record_workflow_state(
+            Path(progress_path),
+            "planning",
+            None,
+            plan_path,
+            "executing-plans",
+            "karpathy-guidelines,.agents,AGENT_RULES",
+        )
+        if code != 0:
+            print(message, file=sys.stderr)
+            return code
+        print(message)
+        return 0
+
     if "-config" not in argv:
         print("ERROR: -config is required.\n" + usage(), file=sys.stderr)
         return 2