973 lines
35 KiB
Python
973 lines
35 KiB
Python
import importlib.util
|
|
import os
|
|
import platform
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SCRIPT = ROOT / "scripts" / "main_loop.py"
|
|
|
|
_SPEC = importlib.util.spec_from_file_location("playbook_main_loop", SCRIPT)
|
|
assert _SPEC and _SPEC.loader
|
|
MAIN_LOOP = importlib.util.module_from_spec(_SPEC)
|
|
_SPEC.loader.exec_module(MAIN_LOOP)
|
|
|
|
|
|
def run_cli(*args, cwd=None):
|
|
return subprocess.run(
|
|
[sys.executable, str(SCRIPT), *args],
|
|
capture_output=True,
|
|
text=True,
|
|
cwd=cwd,
|
|
)
|
|
|
|
|
|
def valid_plan_text(title: str = "Demo Plan") -> str:
|
|
return "\n".join(
|
|
[
|
|
f"# {title}",
|
|
"",
|
|
"## Plan Meta",
|
|
"",
|
|
"- **Verification Scope**: `unit`",
|
|
"- **Verification Gate**: `python -m unittest tests.test_main_loop_cli`",
|
|
"",
|
|
"## Tasks",
|
|
"",
|
|
"- [ ] Task 1: demo",
|
|
"",
|
|
]
|
|
)
|
|
|
|
|
|
class MainLoopCliTests(unittest.TestCase):
|
|
def _current_env(self) -> str:
|
|
system = platform.system().lower()
|
|
mapping = {"windows": "windows", "linux": "linux", "darwin": "darwin"}
|
|
if system not in mapping:
|
|
self.skipTest(f"Unsupported environment: {system}")
|
|
return mapping[system]
|
|
|
|
def test_claim_seeds_progress_and_marks_first_plan_in_progress(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-01-old.md").write_text(
|
|
valid_plan_text("old"), encoding="utf-8"
|
|
)
|
|
(plans_dir / "2026-01-02-new.md").write_text(
|
|
valid_plan_text("new"), encoding="utf-8"
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
self.assertEqual(
|
|
result.stdout.strip(),
|
|
"PLAN=docs/superpowers/plans/2026-01-01-old.md",
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("<!-- workflow-state:start -->", text)
|
|
self.assertIn("<!-- workflow-state:end -->", text)
|
|
self.assertIn("phase: executing", text)
|
|
self.assertIn("plan: docs/superpowers/plans/2026-01-01-old.md", text)
|
|
self.assertIn("<!-- plan-status:start -->", text)
|
|
self.assertIn("<!-- plan-status:end -->", text)
|
|
self.assertIn("`2026-01-01-old.md` in-progress", text)
|
|
self.assertIn("`2026-01-02-new.md` pending", text)
|
|
|
|
def test_claim_preserves_human_progress_sections_when_plan_block_missing(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-01-demo.md").write_text(
|
|
valid_plan_text(), encoding="utf-8"
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# 当前进展",
|
|
"",
|
|
"## Current Focus",
|
|
"",
|
|
"- keep-this-focus",
|
|
"",
|
|
"## Recent Changes",
|
|
"",
|
|
"- keep-this-change",
|
|
"",
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("- keep-this-focus", text)
|
|
self.assertIn("- keep-this-change", text)
|
|
self.assertIn("<!-- workflow-state:start -->", text)
|
|
self.assertIn("<!-- plan-status:start -->", text)
|
|
self.assertIn("`2026-01-01-demo.md` in-progress", text)
|
|
|
|
def test_claim_returns_existing_in_progress_before_pending(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-01-a.md").write_text(
|
|
valid_plan_text("a"), encoding="utf-8"
|
|
)
|
|
(plans_dir / "2026-01-02-b.md").write_text(
|
|
valid_plan_text("b"), encoding="utf-8"
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# Plan 状态",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-01-02-b.md` pending",
|
|
"- [ ] `2026-01-01-a.md` in-progress",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
self.assertEqual(
|
|
result.stdout.strip(),
|
|
"PLAN=docs/superpowers/plans/2026-01-01-a.md",
|
|
)
|
|
|
|
def test_claim_skips_stale_progress_entries_for_deleted_plans(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-02-live.md").write_text(
|
|
valid_plan_text("live"), encoding="utf-8"
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# Plan 状态",
|
|
"",
|
|
"<!-- workflow-state:start -->",
|
|
"phase: planning",
|
|
"<!-- workflow-state:end -->",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-01-01-deleted.md` pending",
|
|
"- [ ] `2026-01-02-live.md` pending",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
self.assertEqual(
|
|
result.stdout.strip(),
|
|
"PLAN=docs/superpowers/plans/2026-01-02-live.md",
|
|
)
|
|
|
|
def test_claim_resumes_env_blocked_plan_and_preserves_note(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-05-env.md").write_text(
|
|
valid_plan_text("env"), encoding="utf-8"
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
env = self._current_env()
|
|
note = f"env:{env}:Task1,Task3"
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# Plan 状态",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
f"- [ ] `2026-01-05-env.md` blocked: {note}",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
self.assertEqual(
|
|
result.stdout.strip(),
|
|
"\n".join(
|
|
[
|
|
"PLAN=docs/superpowers/plans/2026-01-05-env.md",
|
|
f"NOTE={note}",
|
|
]
|
|
),
|
|
)
|
|
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn(f"`2026-01-05-env.md` in-progress: {note}", text)
|
|
|
|
def test_claim_prefers_earlier_env_blocked_plan_over_later_pending_plan(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-01-env.md").write_text(
|
|
valid_plan_text("env"), encoding="utf-8"
|
|
)
|
|
(plans_dir / "2026-01-02-pending.md").write_text(
|
|
valid_plan_text("pending"), encoding="utf-8"
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
env = self._current_env()
|
|
note = f"env:{env}:Task2"
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# Plan 状态",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
f"- [ ] `2026-01-01-env.md` blocked: {note}",
|
|
"- [ ] `2026-01-02-pending.md` pending",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
self.assertEqual(
|
|
result.stdout.strip(),
|
|
"\n".join(
|
|
[
|
|
"PLAN=docs/superpowers/plans/2026-01-01-env.md",
|
|
f"NOTE={note}",
|
|
]
|
|
),
|
|
)
|
|
|
|
def test_claim_rejects_plan_missing_required_plan_meta(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-01-invalid.md").write_text(
|
|
"# Invalid Plan\n\n- [ ] Task 1: missing metadata\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 2)
|
|
self.assertIn("missing required Plan Meta", result.stderr)
|
|
self.assertIn("2026-01-01-invalid.md", result.stderr)
|
|
|
|
def test_claim_records_claim_metadata_in_workflow_state(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-01-demo.md").write_text(
|
|
valid_plan_text(), encoding="utf-8"
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-owner",
|
|
"codex-test",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = (root / "memory-bank" / "progress.md").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
self.assertIn("claimed_by: codex-test", text)
|
|
self.assertRegex(text, r"claimed_at: \d{4}-\d{2}-\d{2}T")
|
|
|
|
def test_claim_clears_stale_verification_from_workflow_state(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-01-01-demo.md").write_text(
|
|
valid_plan_text(), encoding="utf-8"
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# 当前进展",
|
|
"",
|
|
"<!-- workflow-state:start -->",
|
|
"phase: done",
|
|
"verification: old evidence",
|
|
"<!-- workflow-state:end -->",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-01-01-demo.md` pending",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertNotIn("verification: old evidence", text)
|
|
|
|
def test_finish_updates_line(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# Plan 状态",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-01-03-demo.md` in-progress",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"finish",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-01-03-demo.md",
|
|
"-status",
|
|
"done",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("- [x] `2026-01-03-demo.md` done", text)
|
|
self.assertEqual(
|
|
text.count("- [x] `2026-01-03-demo.md` done"),
|
|
1,
|
|
)
|
|
|
|
def test_finish_done_records_verification_evidence(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# Plan 状态",
|
|
"",
|
|
"<!-- workflow-state:start -->",
|
|
"phase: executing",
|
|
"plan: docs/superpowers/plans/2026-01-03-demo.md",
|
|
"<!-- workflow-state:end -->",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-01-03-demo.md` in-progress",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"finish",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-01-03-demo.md",
|
|
"-status",
|
|
"done",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-verified",
|
|
"python -m unittest tests.test_main_loop_cli",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn(
|
|
"- [x] `2026-01-03-demo.md` done: "
|
|
"verified: python -m unittest tests.test_main_loop_cli",
|
|
text,
|
|
)
|
|
self.assertIn(
|
|
"verification: python -m unittest tests.test_main_loop_cli",
|
|
text,
|
|
)
|
|
|
|
def test_finish_without_verified_clears_stale_verification(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# Plan 状态",
|
|
"",
|
|
"<!-- workflow-state:start -->",
|
|
"phase: executing",
|
|
"plan: docs/superpowers/plans/2026-01-03-demo.md",
|
|
"verification: old evidence",
|
|
"<!-- workflow-state:end -->",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-01-03-demo.md` in-progress",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"finish",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-01-03-demo.md",
|
|
"-status",
|
|
"blocked",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-note",
|
|
"needs confirmation",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertNotIn("verification: old evidence", text)
|
|
self.assertIn("phase: blocked", text)
|
|
|
|
def test_finish_updates_workflow_phase_and_preserves_metadata(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# 当前进展",
|
|
"",
|
|
"## Workflow State",
|
|
"",
|
|
"<!-- workflow-state:start -->",
|
|
"phase: executing",
|
|
"spec: docs/superpowers/specs/2026-05-18-demo-design.md",
|
|
"plan: docs/superpowers/plans/2026-05-18-demo.md",
|
|
"executor: executing-plans",
|
|
"constraints: karpathy-guidelines,.agents,AGENT_RULES",
|
|
"<!-- workflow-state:end -->",
|
|
"",
|
|
"## Plan Status",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-05-18-demo.md` in-progress",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"finish",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-05-18-demo.md",
|
|
"-status",
|
|
"done",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("phase: done", text)
|
|
self.assertIn(
|
|
"spec: docs/superpowers/specs/2026-05-18-demo-design.md", text
|
|
)
|
|
self.assertIn("executor: executing-plans", text)
|
|
self.assertIn(
|
|
"constraints: karpathy-guidelines,.agents,AGENT_RULES",
|
|
text,
|
|
)
|
|
|
|
def test_finish_skipped_updates_workflow_phase_to_skipped(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# 当前进展",
|
|
"",
|
|
"## Workflow State",
|
|
"",
|
|
"<!-- workflow-state:start -->",
|
|
"phase: executing",
|
|
"plan: docs/superpowers/plans/2026-05-18-demo.md",
|
|
"<!-- workflow-state:end -->",
|
|
"",
|
|
"## Plan Status",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-05-18-demo.md` in-progress",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"finish",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-05-18-demo.md",
|
|
"-status",
|
|
"skipped",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-note",
|
|
"obsolete",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("phase: skipped", text)
|
|
self.assertIn("- [ ] `2026-05-18-demo.md` skipped: obsolete", text)
|
|
|
|
def test_record_updates_workflow_state_block(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# 当前进展",
|
|
"",
|
|
"## Plan Status",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"record",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-phase",
|
|
"planning",
|
|
"-spec",
|
|
"docs/superpowers/specs/2026-05-18-demo-design.md",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-05-18-demo.md",
|
|
"-executor",
|
|
"executing-plans",
|
|
"-constraints",
|
|
"karpathy-guidelines,.agents,AGENT_RULES",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("<!-- workflow-state:start -->", text)
|
|
self.assertIn("phase: planning", text)
|
|
self.assertIn(
|
|
"spec: docs/superpowers/specs/2026-05-18-demo-design.md", text
|
|
)
|
|
self.assertIn("plan: docs/superpowers/plans/2026-05-18-demo.md", text)
|
|
self.assertIn("executor: executing-plans", text)
|
|
self.assertIn(
|
|
"constraints: karpathy-guidelines,.agents,AGENT_RULES",
|
|
text,
|
|
)
|
|
|
|
def test_record_claim_finish_workflow_chain(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
(plans_dir / "2026-05-18-demo.md").write_text(
|
|
valid_plan_text(), encoding="utf-8"
|
|
)
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
|
|
result = run_cli(
|
|
"record",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-phase",
|
|
"planning",
|
|
"-spec",
|
|
"docs/superpowers/specs/2026-05-18-demo-design.md",
|
|
cwd=root,
|
|
)
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
|
|
result = run_cli(
|
|
"record",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-phase",
|
|
"planning",
|
|
"-spec",
|
|
"docs/superpowers/specs/2026-05-18-demo-design.md",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-05-18-demo.md",
|
|
"-executor",
|
|
"executing-plans",
|
|
"-constraints",
|
|
"karpathy-guidelines,.agents,AGENT_RULES",
|
|
cwd=root,
|
|
)
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
|
|
result = run_cli(
|
|
"claim",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
|
|
result = run_cli(
|
|
"finish",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-05-18-demo.md",
|
|
"-status",
|
|
"done",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("phase: done", text)
|
|
self.assertIn(
|
|
"spec: docs/superpowers/specs/2026-05-18-demo-design.md", text
|
|
)
|
|
self.assertIn("plan: docs/superpowers/plans/2026-05-18-demo.md", text)
|
|
self.assertIn("executor: executing-plans", text)
|
|
self.assertIn(
|
|
"constraints: karpathy-guidelines,.agents,AGENT_RULES",
|
|
text,
|
|
)
|
|
self.assertIn("- [x] `2026-05-18-demo.md` done", text)
|
|
|
|
def test_status_reports_counts_and_current_workflow_state(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
plans_dir = root / "docs" / "superpowers" / "plans"
|
|
plans_dir.mkdir(parents=True)
|
|
for plan_key in (
|
|
"2026-01-01-a.md",
|
|
"2026-01-02-b.md",
|
|
"2026-01-03-c.md",
|
|
"2026-01-04-d.md",
|
|
"2026-01-05-e.md",
|
|
):
|
|
(plans_dir / plan_key).write_text(valid_plan_text(), encoding="utf-8")
|
|
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
progress.write_text(
|
|
"\n".join(
|
|
[
|
|
"# 当前进展",
|
|
"",
|
|
"<!-- workflow-state:start -->",
|
|
"phase: executing",
|
|
"plan: docs/superpowers/plans/2026-01-02-b.md",
|
|
"claimed_by: codex-test",
|
|
"<!-- workflow-state:end -->",
|
|
"",
|
|
"<!-- plan-status:start -->",
|
|
"- [ ] `2026-01-01-a.md` pending",
|
|
"- [ ] `2026-01-02-b.md` in-progress",
|
|
"- [x] `2026-01-03-c.md` done",
|
|
"- [ ] `2026-01-04-d.md` blocked: env:linux:Task2",
|
|
"- [ ] `2026-01-05-e.md` skipped: obsolete",
|
|
"<!-- plan-status:end -->",
|
|
"",
|
|
]
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = run_cli(
|
|
"status",
|
|
"-plans",
|
|
"docs/superpowers/plans",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
cwd=root,
|
|
)
|
|
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
self.assertIn(
|
|
"STATUS total=5 pending=1 in-progress=1 done=1 blocked=1 skipped=1",
|
|
result.stdout,
|
|
)
|
|
self.assertIn(
|
|
"CURRENT phase=executing "
|
|
"plan=docs/superpowers/plans/2026-01-02-b.md "
|
|
"claimed_by=codex-test",
|
|
result.stdout,
|
|
)
|
|
|
|
def test_concurrent_record_preserves_spec_and_plan_metadata(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
|
|
original_load = MAIN_LOOP.load_progress_lines
|
|
first_load = {"seen": False}
|
|
gate = threading.Lock()
|
|
|
|
def delayed_load(progress_path):
|
|
lines = original_load(progress_path)
|
|
with gate:
|
|
if not first_load["seen"]:
|
|
first_load["seen"] = True
|
|
threading.Event().wait(0.2)
|
|
return lines
|
|
|
|
MAIN_LOOP.load_progress_lines = delayed_load
|
|
try:
|
|
threads = [
|
|
threading.Thread(
|
|
target=MAIN_LOOP.record_workflow_state,
|
|
args=(
|
|
progress,
|
|
"planning",
|
|
"docs/superpowers/specs/2026-05-18-demo-design.md",
|
|
None,
|
|
None,
|
|
None,
|
|
),
|
|
),
|
|
threading.Thread(
|
|
target=MAIN_LOOP.record_workflow_state,
|
|
args=(
|
|
progress,
|
|
"planning",
|
|
None,
|
|
"docs/superpowers/plans/2026-05-18-demo.md",
|
|
"executing-plans",
|
|
"karpathy-guidelines,.agents,AGENT_RULES",
|
|
),
|
|
),
|
|
]
|
|
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
finally:
|
|
MAIN_LOOP.load_progress_lines = original_load
|
|
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("phase: planning", text)
|
|
self.assertIn(
|
|
"spec: docs/superpowers/specs/2026-05-18-demo-design.md", text
|
|
)
|
|
self.assertIn("plan: docs/superpowers/plans/2026-05-18-demo.md", text)
|
|
self.assertIn("executor: executing-plans", text)
|
|
self.assertIn(
|
|
"constraints: karpathy-guidelines,.agents,AGENT_RULES",
|
|
text,
|
|
)
|
|
|
|
def test_cross_process_record_lock_preserves_spec_and_plan_metadata(self):
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
root = Path(tmp_dir)
|
|
progress = root / "memory-bank" / "progress.md"
|
|
progress.parent.mkdir(parents=True)
|
|
|
|
slow_env = dict(os.environ)
|
|
slow_env["PLAYBOOK_MAIN_LOOP_HOLD_LOCK_MS"] = "300"
|
|
|
|
proc = subprocess.Popen(
|
|
[
|
|
sys.executable,
|
|
str(SCRIPT),
|
|
"record",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-phase",
|
|
"planning",
|
|
"-spec",
|
|
"docs/superpowers/specs/2026-05-18-demo-design.md",
|
|
],
|
|
cwd=root,
|
|
env=slow_env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
)
|
|
|
|
time.sleep(0.05)
|
|
result = run_cli(
|
|
"record",
|
|
"-progress",
|
|
"memory-bank/progress.md",
|
|
"-phase",
|
|
"planning",
|
|
"-plan",
|
|
"docs/superpowers/plans/2026-05-18-demo.md",
|
|
"-executor",
|
|
"executing-plans",
|
|
"-constraints",
|
|
"karpathy-guidelines,.agents,AGENT_RULES",
|
|
cwd=root,
|
|
)
|
|
|
|
stdout, stderr = proc.communicate(timeout=5)
|
|
self.assertEqual(proc.returncode, 0, msg=stderr or stdout)
|
|
self.assertEqual(result.returncode, 0, msg=result.stderr)
|
|
|
|
text = progress.read_text(encoding="utf-8")
|
|
self.assertIn("phase: planning", text)
|
|
self.assertIn(
|
|
"spec: docs/superpowers/specs/2026-05-18-demo-design.md", text
|
|
)
|
|
self.assertIn("plan: docs/superpowers/plans/2026-05-18-demo.md", text)
|
|
self.assertIn("executor: executing-plans", text)
|
|
self.assertIn(
|
|
"constraints: karpathy-guidelines,.agents,AGENT_RULES",
|
|
text,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|