Files
tsl-devkit/tests/test_main_loop_cli.py
T
csh 3d83740f88 Squashed 'docs/standards/playbook/' changes from c3f8137..25d895d
25d895d 🐛 fix(gitea_workflow): clean up temp repos after job steps
2bc3b11 🐛 fix(gitea_workflow): clean up temporary repo dirs in workflows
98c3f30 📝 docs(agent_rules): allow plan execution on current branch
16c7230 📝 docs(prompts): define custom verify layering
8efc4dd 🐛 fix(skills): quote commit-message description
bc8498f 🐛 fix(ci): install tomli for gitea tests
55cda3b 🐛 fix(tests): report missing toml parser clearly
c0729c7 🐛 fix(playbook): import Optional for cli compatibility
63e24bf 📦 deps(skills): sync thirdparty skills
d2f9356 🐛 fix(ci): isolate gitea workflow repos
588b81d 🐛 fix(ci): inline gitea workflow bootstrap
e0b1c3a ♻️ refactor(skills): standardize first-party skill contracts
2c5050d ♻️ refactor(skills): rename repo skills source dir
f049dfb 📦 deps(skills): drop duplicate first-party superpowers skills
234b335  feat(workflow): add superpowers planning and execution state tracking
c1702a6 📝 docs(markdown): format tracked markdown and drop stale templates
2325409 📝 docs(markdown): clarify optional markdownlint usage
214c44e 🔧 chore(markdown): add markdownlint baseline and lint fixes
a22b324 📝 docs(templates): add execution and memory-bank prompt templates
223a797 📝 docs(templates): update README for Claude Code and current features
4ac8672 📝 docs: simplify README + platform-agnostic tools + auto-create local rules
2431c9d 📝 docs: add claude_md config and use cross-platform paths
d64b248 📝 docs: fix README.md inaccuracies and add Claude Code info
c8d6bf2 🐛 fix(playbook): use relative paths in CLAUDE.md when not at project root
6518f0f  feat(playbook): auto-create CLAUDE.md with path discovery
6ec9a45  feat(skills): add skill_link symlink support + platform-agnostic prompt
9f8b6b5 📝 docs: update README and config example for Claude Code support
79cff6c 📝 docs(skills): add Claude Code platform support
452c6f5  feat(playbook): auto-inject AGENTS.md into CLAUDE.md
e1dbf3c 🐛 fix(skills): remove dual-path from commit-message skill
f3a7259 🔧 chore(ci): use prepare_repo.sh in both workflows
da08212 🔧 chore(ci): extract prepare_repo.sh and clean up workflows
7ade85e 🗑️ remove(tsl): drop syntax_book/, data/ source and build script
f94dba0 ♻️ refactor(skills): update playbook.py and tests for thirdparty/ layout
b3df412 ♻️ refactor(skills): separate thirdparty skills into thirdparty/ subdirectory
64950e7 📦 deps(skills): sync thirdparty skills
a2e3cb0  feat(playbook): add no_backup deploy controls
8609d59 🐛 fix(docs): repair reference catalog source links
956da11 🐛 fix(playbook): publish hidden ci test fixes
3f67754 📦 deps(skills): sync thirdparty skills
08ca87b 📦 deps(skills): add karpathy thirdparty sync
96b705b 📝 docs(tsl): rebuild canonical syntax and routing manual
3ed5052 📦 deps(skills): sync thirdparty skills
60108dd 📦 deps(skills): sync thirdparty skills
da85d4e 🐛 fix(thirdparty): prune nested project snapshots
a2a697e 📦 deps(skills): sync thirdparty skills
9df610a 🐛 fix(thirdparty): exclude duplicated superpowers skills
33dd5bb 🐛 fix(thirdparty): preserve optional manifest fields
91b0ea7 🐛 fix(thirdparty): preserve manifest during snapshot update
2e26f98 🔧 chore(thirdparty): generalize skills sync pipeline
5b9c1e3 📦 deps(skills): sync superpowers
2f2d34a 📝 docs(readme): normalize subtree command spacing
62db7db 🐛 fix(ci): serialize superpowers update and sync
3463223 🐛 fix(ci): use literal superpowers sync paths
48f6de8 📦 deps(skills): sync superpowers
4b23529 🔧 chore(ci): merge superpowers update and sync workflow
a56d75b 📦 deps(skills): sync superpowers
84bcefa 🔧 chore(ci): use ci[bot] commit author name
00a07e5 📦 deps(skills): sync superpowers
7b84daf 🐛 fix(templates): enforce main loop progress tracking
51373d7 🔧 chore(ci): automate superpowers sync workflow
eaaa39c 🐛 fix(ci): prevent stale superpowers sync from restoring skills block
79755c6 📦 deps(skills): sync superpowers
836d878 📦 deps(skills): sync superpowers
8216c9f 📦 deps(skills): sync superpowers
9439505 🐛 fix(playbook): address reported repo issues

git-subtree-dir: docs/standards/playbook
git-subtree-split: 25d895d8b3f56624ccfe99ad7289e9eb49e0f316
2026-05-24 13:04:14 +08:00

585 lines
21 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,
)
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("old", encoding="utf-8")
(plans_dir / "2026-01-02-new.md").write_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("demo", 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("a", encoding="utf-8")
(plans_dir / "2026-01-02-b.md").write_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("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("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_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_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_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("demo", 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_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()