746 lines
30 KiB
Python
746 lines
30 KiB
Python
import argparse
|
||
import importlib.util
|
||
import re
|
||
import sys
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
TEMPLATES = ROOT / "templates"
|
||
MAIN_LOOP_ROOT = ROOT / "skills" / "cook-it-through"
|
||
MAIN_LOOP_SKILL = MAIN_LOOP_ROOT / "SKILL.md"
|
||
MAIN_LOOP_SESSION_BOUNDARY = MAIN_LOOP_ROOT / "rules" / "session-boundary.md"
|
||
MAIN_LOOP_WORKFLOWS = {
|
||
name: MAIN_LOOP_ROOT / "workflows" / f"{name}.md"
|
||
for name in (
|
||
"single-session",
|
||
"feature-planning",
|
||
"ticket-execution",
|
||
"feature-integration",
|
||
)
|
||
}
|
||
MAIN_LOOP_SCRIPTS = MAIN_LOOP_ROOT / "scripts"
|
||
MAIN_LOOP_SCRIPT = MAIN_LOOP_SCRIPTS / "main_loop.py"
|
||
|
||
_MAIN_LOOP_SPEC = importlib.util.spec_from_file_location(
|
||
"playbook_main_loop_contracts", MAIN_LOOP_SCRIPT
|
||
)
|
||
assert _MAIN_LOOP_SPEC and _MAIN_LOOP_SPEC.loader
|
||
MAIN_LOOP = importlib.util.module_from_spec(_MAIN_LOOP_SPEC)
|
||
sys.modules[_MAIN_LOOP_SPEC.name] = MAIN_LOOP
|
||
_MAIN_LOOP_SPEC.loader.exec_module(MAIN_LOOP)
|
||
|
||
|
||
def subcommand_parsers() -> dict[str, argparse.ArgumentParser]:
|
||
parser = MAIN_LOOP.build_parser()
|
||
for action in parser._subparsers._group_actions: # noqa: SLF001
|
||
if isinstance(action, argparse._SubParsersAction): # noqa: SLF001
|
||
return dict(action.choices)
|
||
raise AssertionError("main_loop.py exposes no subcommands")
|
||
|
||
|
||
def required_flags(parser: argparse.ArgumentParser) -> set[str]:
|
||
return {
|
||
option
|
||
for action in parser._actions # noqa: SLF001
|
||
if action.required
|
||
for option in action.option_strings
|
||
if option.startswith("--")
|
||
}
|
||
|
||
|
||
def option_action(
|
||
parser: argparse.ArgumentParser, option: str
|
||
) -> argparse.Action:
|
||
for action in parser._actions: # noqa: SLF001
|
||
if option in action.option_strings:
|
||
return action
|
||
raise AssertionError(f"{parser.prog} exposes no {option}")
|
||
|
||
|
||
def normalized_prose(text: str) -> str:
|
||
return " ".join(text.split())
|
||
|
||
|
||
def isolation_choices() -> set[str]:
|
||
claim = subcommand_parsers()["claim"]
|
||
for action in claim._actions: # noqa: SLF001
|
||
if "--isolation" in action.option_strings:
|
||
return set(action.choices or ())
|
||
raise AssertionError("claim exposes no --isolation choices")
|
||
|
||
|
||
def installed_skills() -> set[str]:
|
||
names: set[str] = set()
|
||
for base in (ROOT / "skills", ROOT / "skills" / "thirdparty"):
|
||
if not base.is_dir():
|
||
continue
|
||
for path in base.iterdir():
|
||
if path.is_dir() and (path / "SKILL.md").is_file():
|
||
names.add(path.name)
|
||
return names
|
||
|
||
|
||
def rules_text() -> str:
|
||
return (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
|
||
|
||
|
||
def main_loop_skill_text() -> str:
|
||
return MAIN_LOOP_SKILL.read_text(encoding="utf-8")
|
||
|
||
|
||
def main_loop_session_boundary_text() -> str:
|
||
return MAIN_LOOP_SESSION_BOUNDARY.read_text(encoding="utf-8")
|
||
|
||
|
||
def main_loop_workflow_text(name: str) -> str:
|
||
return MAIN_LOOP_WORKFLOWS[name].read_text(encoding="utf-8")
|
||
|
||
|
||
def main_loop_instruction_paths() -> tuple[Path, ...]:
|
||
return (
|
||
MAIN_LOOP_SKILL,
|
||
MAIN_LOOP_SESSION_BOUNDARY,
|
||
*MAIN_LOOP_WORKFLOWS.values(),
|
||
)
|
||
|
||
|
||
def main_loop_bundle_text() -> str:
|
||
return "\n".join(
|
||
path.read_text(encoding="utf-8") for path in main_loop_instruction_paths()
|
||
)
|
||
|
||
|
||
def section(text: str, heading: str, until: str) -> str:
|
||
return text.split(heading, 1)[1].split(until, 1)[0]
|
||
|
||
|
||
def headings(text: str) -> list[str]:
|
||
return re.findall(r"^#{2,3} (.+)$", text, re.MULTILINE)
|
||
|
||
|
||
# Tokens written in backticks that name something other than a skill. Machine
|
||
# state, CLI vocabulary and harness commands are derived from the implementation
|
||
# where possible so the allowlist stays small and reviewable.
|
||
NON_SKILL_BACKTICK_TOKENS = frozenset(
|
||
{
|
||
"base",
|
||
"claude",
|
||
"clear",
|
||
"codex",
|
||
"command",
|
||
"commit",
|
||
"compact",
|
||
"main",
|
||
"pass",
|
||
"pass-with-notes",
|
||
"passed",
|
||
"output",
|
||
"result",
|
||
"released",
|
||
"report",
|
||
"spec",
|
||
"standards",
|
||
}
|
||
)
|
||
|
||
LEGACY_FLOW_TERMS = (
|
||
"using-superpowers",
|
||
"brainstorming",
|
||
"writing-plans",
|
||
"executing-plans",
|
||
"docs/superpowers/",
|
||
"plan-status",
|
||
"旧开发流程目录",
|
||
"## 禁止旧流程",
|
||
"旧开发队列",
|
||
"单文件开发计划主链",
|
||
"active-context.md",
|
||
"progress.md",
|
||
"decisions.md",
|
||
)
|
||
|
||
|
||
class TemplateContractsTests(unittest.TestCase):
|
||
def test_templates_define_only_the_matt_ticket_workflow(self):
|
||
combined = "\n".join(
|
||
[
|
||
*(path.read_text(encoding="utf-8") for path in sorted(TEMPLATES.rglob("*.md"))),
|
||
main_loop_bundle_text(),
|
||
]
|
||
)
|
||
|
||
for required in (
|
||
"setup-matt-pocock-skills",
|
||
"grill-with-docs",
|
||
"to-spec",
|
||
"to-tickets",
|
||
".scratch/",
|
||
"main_loop.py claim",
|
||
"main_loop.py integrate",
|
||
):
|
||
self.assertIn(required, combined)
|
||
for legacy in LEGACY_FLOW_TERMS:
|
||
self.assertNotIn(legacy, combined)
|
||
|
||
def test_project_templates_drop_legacy_language_placeholders(self):
|
||
templates_readme = (TEMPLATES / "README.md").read_text(encoding="utf-8")
|
||
self.assertNotIn("{{MAIN_LANGUAGE}}", templates_readme)
|
||
self.assertNotIn("{{LANGUAGE_1}}", templates_readme)
|
||
self.assertNotIn("docs/workflows/", templates_readme)
|
||
self.assertNotIn("templates/workflows/", templates_readme)
|
||
self.assertNotIn("docs/superpowers/", templates_readme)
|
||
self.assertNotIn(".scratch/<feature>/spec.md", templates_readme)
|
||
self.assertIn("`.scratch/<feature>/spec.md`", main_loop_bundle_text())
|
||
self.assertNotIn("docs/prompts/", templates_readme)
|
||
|
||
agents_template = (TEMPLATES / "AGENTS.template.md").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
self.assertNotIn("{{MAIN_LANGUAGE}}", agents_template)
|
||
self.assertIn("AGENT_RULES.md", agents_template)
|
||
self.assertIn("memory-bank/project-brief.md", agents_template)
|
||
self.assertIn("memory-bank/tech-context.md", agents_template)
|
||
self.assertIn("memory-bank/system-patterns.md", agents_template)
|
||
self.assertIn(".scratch/queue.md", agents_template)
|
||
self.assertNotIn("docs/prompts/", agents_template)
|
||
|
||
tech_context_template = (
|
||
TEMPLATES / "memory-bank" / "tech-context.template.md"
|
||
).read_text(encoding="utf-8")
|
||
self.assertNotIn("{{MAIN_LANGUAGE}}", tech_context_template)
|
||
self.assertNotIn("{{LANGUAGE_1}}", tech_context_template)
|
||
self.assertNotIn("**主要语言**", tech_context_template)
|
||
self.assertIn("## 不可假设项", tech_context_template)
|
||
|
||
project_brief_template = (
|
||
TEMPLATES / "memory-bank" / "project-brief.template.md"
|
||
).read_text(encoding="utf-8")
|
||
self.assertIn("## 成功定义", project_brief_template)
|
||
|
||
def test_agents_template_nests_agents_block_inside_framework_block(self):
|
||
agents_template = (TEMPLATES / "AGENTS.template.md").read_text(
|
||
encoding="utf-8"
|
||
)
|
||
lines = [line.strip() for line in agents_template.splitlines()]
|
||
for marker in (
|
||
"<!-- playbook:framework:start -->",
|
||
"<!-- playbook:framework:end -->",
|
||
"<!-- playbook:agents:start -->",
|
||
"<!-- playbook:agents:end -->",
|
||
):
|
||
self.assertIn(marker, lines, msg=f"missing marker: {marker}")
|
||
|
||
framework_start = lines.index("<!-- playbook:framework:start -->")
|
||
framework_end = lines.index("<!-- playbook:framework:end -->")
|
||
agents_start = lines.index("<!-- playbook:agents:start -->")
|
||
agents_end = lines.index("<!-- playbook:agents:end -->")
|
||
nesting_reason = (
|
||
"sync_agents_template replaces the whole framework block; the agents "
|
||
"sub-block must stay inside it so preserve_agents_subblock can carry "
|
||
"the deployed language rules across the replacement"
|
||
)
|
||
self.assertLess(framework_start, agents_start, msg=nesting_reason)
|
||
self.assertLess(agents_start, agents_end, msg=nesting_reason)
|
||
self.assertLess(agents_end, framework_end, msg=nesting_reason)
|
||
|
||
def test_templates_readme_separates_ownership_from_deployment_config(self):
|
||
templates_readme = (TEMPLATES / "README.md").read_text(encoding="utf-8")
|
||
classification = section(templates_readme, "## 模板分类", "## 模板说明")
|
||
categories = [h for h in headings(classification) if h[:2] in ("1.", "2.", "3.")]
|
||
|
||
self.assertEqual(len(categories), 3, msg=f"unexpected categories: {categories}")
|
||
self.assertIn("项目新增的 `memory-bank/*`", classification)
|
||
self.assertIn(
|
||
"playbook:rules:start/end",
|
||
classification,
|
||
msg="ownership of AGENT_RULES.md is split by the marker block; the "
|
||
"classification has to say which side the project owns",
|
||
)
|
||
self.assertNotIn("[sync_", classification)
|
||
self.assertNotIn("force", classification)
|
||
self.assertNotIn("no_backup", classification)
|
||
|
||
def test_templates_readme_separates_state_source_and_protocol_authority(self):
|
||
templates_readme = (TEMPLATES / "README.md").read_text(encoding="utf-8")
|
||
normalized = normalized_prose(templates_readme)
|
||
|
||
self.assertIn("`.scratch/` 是唯一机器状态源", normalized)
|
||
self.assertIn(
|
||
"由第一方 `cook-it-through` Skill 权威定义",
|
||
normalized,
|
||
)
|
||
self.assertNotIn("preserve_agents_subblock()", templates_readme)
|
||
self.assertNotIn("四个入口按成本递增", templates_readme)
|
||
self.assertIsNone(
|
||
re.search(r"\*\*最后更新\*\*：\d{4}-\d{2}-\d{2}", templates_readme),
|
||
msg="templates README must not carry a hand-maintained update date",
|
||
)
|
||
|
||
def test_templates_readme_documents_skill_exclusion_boundary(self):
|
||
templates_readme = (TEMPLATES / "README.md").read_text(encoding="utf-8")
|
||
deployment = normalized_prose(
|
||
section(templates_readme, "## 部署", "## 正式开发流程")
|
||
)
|
||
layout = normalized_prose(
|
||
section(
|
||
templates_readme,
|
||
"## `playbook.py` 部署后结构",
|
||
"## 正式流程运行后按需产生的结构",
|
||
)
|
||
)
|
||
|
||
self.assertIn("启用了 `[sync_rules]`", deployment)
|
||
self.assertIn("安装集合必须包含 `cook-it-through`", deployment)
|
||
self.assertIn('`mode = "all"` 时不得通过 `exclude` 排除', deployment)
|
||
self.assertIn(
|
||
"只有不部署官方 `AGENT_RULES.md` 且不使用正式工程主链的安装场景，才可以排除该 skill",
|
||
deployment,
|
||
)
|
||
self.assertIn("同时启用 `[sync_rules]` 时，必须遵守上文", layout)
|
||
|
||
def test_memory_bank_contains_only_stable_project_knowledge(self):
|
||
memory_templates = {
|
||
path.name for path in (TEMPLATES / "memory-bank").glob("*.template.md")
|
||
}
|
||
self.assertEqual(
|
||
memory_templates,
|
||
{
|
||
"project-brief.template.md",
|
||
"tech-context.template.md",
|
||
"system-patterns.template.md",
|
||
},
|
||
)
|
||
|
||
boundary = normalized_prose(main_loop_session_boundary_text())
|
||
planning = main_loop_workflow_text("feature-planning")
|
||
execution = main_loop_workflow_text("ticket-execution")
|
||
for path in (
|
||
"memory-bank/project-brief.md",
|
||
"memory-bank/tech-context.md",
|
||
"memory-bank/system-patterns.md",
|
||
):
|
||
self.assertIn(path, boundary)
|
||
for required in (
|
||
"已经验证且可复现",
|
||
"重新发现成本高",
|
||
"不能从代码直接看出",
|
||
"下一 session 仍需要",
|
||
"关键取舍及理由写入 `docs/adr/`",
|
||
"`handoff` 产物写入 OS 临时目录",
|
||
):
|
||
self.assertIn(required, boundary)
|
||
self.assertIn("进入 `grill-with-docs` 前", planning)
|
||
self.assertIn("领取后实现前", execution)
|
||
|
||
def test_prompt_templates_are_not_part_of_the_workflow(self):
|
||
self.assertFalse(TEMPLATES.joinpath("prompts").exists())
|
||
|
||
def test_agent_rules_routes_main_loop_work_to_the_firstparty_skill(self):
|
||
rules = rules_text()
|
||
normalized = normalized_prose(rules)
|
||
|
||
self.assertIn("{{PLAYBOOK_ROOT}}", rules)
|
||
self.assertNotIn("{{PLAYBOOK_SCRIPTS}}", rules)
|
||
self.assertEqual(
|
||
headings(rules),
|
||
["优先级", "沟通", "项目边界", "工作流入口"],
|
||
msg="always-loaded rules must remain a thin workflow bootstrap",
|
||
)
|
||
self.assertIn("必须加载 `cook-it-through`", rules)
|
||
self.assertIn(
|
||
"或读取/修改 `.scratch` 中的 queue、ticket、heartbeat、integration 状态前",
|
||
normalized,
|
||
)
|
||
self.assertIn("该 skill 独占", rules)
|
||
self.assertIn("主循环执行引擎随该 skill 安装", normalized)
|
||
self.assertIn("`.agents/index.md`", rules)
|
||
self.assertNotIn("**Blocked by:**", rules)
|
||
self.assertNotIn("## 主循环命令", rules)
|
||
self.assertLessEqual(len(rules.splitlines()), 50)
|
||
self.assertLessEqual(len(rules.encode("utf-8")), 5_000)
|
||
for legacy in LEGACY_FLOW_TERMS:
|
||
self.assertNotIn(legacy, rules)
|
||
|
||
def test_cook_it_through_skill_owns_the_ticket_contract(self):
|
||
rules = rules_text()
|
||
bundle = main_loop_bundle_text()
|
||
planning = main_loop_workflow_text("feature-planning")
|
||
readme = (TEMPLATES / "README.md").read_text(encoding="utf-8")
|
||
skills_readme = (ROOT / "skills/README.md").read_text(encoding="utf-8")
|
||
self.assertFalse((ROOT / "docs/common/main-loop-ticket-contract.md").exists())
|
||
self.assertEqual(
|
||
{
|
||
path.relative_to(MAIN_LOOP_ROOT).as_posix()
|
||
for path in MAIN_LOOP_ROOT.rglob("*")
|
||
if path.is_file() and "__pycache__" not in path.parts
|
||
},
|
||
{
|
||
"SKILL.md",
|
||
"rules/session-boundary.md",
|
||
"workflows/single-session.md",
|
||
"workflows/feature-planning.md",
|
||
"workflows/ticket-execution.md",
|
||
"workflows/feature-integration.md",
|
||
"scripts/main_loop.py",
|
||
"scripts/main_loop_scheduler.py",
|
||
},
|
||
)
|
||
self.assertFalse((ROOT / "scripts/main_loop.py").exists())
|
||
self.assertFalse((ROOT / "scripts/main_loop_scheduler.py").exists())
|
||
for required in (
|
||
"`TicketId`：qualified `feature-slug/NN`",
|
||
"`FeatureIntegrationId`：`feature-slug@integrated`",
|
||
"**Blocked by:** None",
|
||
"**Blocked by:** feature-a/01; feature-b@integrated",
|
||
"同批重复 `--feature`",
|
||
"hard cut",
|
||
):
|
||
self.assertIn(required, planning)
|
||
self.assertIn("第三方 `to-tickets` 只定义通用 tracker 行为", planning)
|
||
self.assertIn("最终机器校验边界", planning)
|
||
self.assertIn("手工修改 ticket `Status`", bundle)
|
||
self.assertIn("<COOK_IT_THROUGH_ROOT>/scripts/main_loop.py", bundle)
|
||
self.assertNotIn("<PLAYBOOK_SCRIPTS>", bundle)
|
||
self.assertIn("`cook-it-through`", rules)
|
||
self.assertNotIn("**Blocked by:**", rules)
|
||
self.assertIn("只在第一方 `skills/cook-it-through/` 定义", readme)
|
||
self.assertNotIn("main-loop-ticket-contract.md", readme)
|
||
self.assertNotIn("**Blocked by:** feature-a/01; feature-b@integrated", readme)
|
||
for public_readme in (readme, skills_readme):
|
||
self.assertNotIn("main_loop.py", public_readme)
|
||
self.assertNotIn("--isolation", readme)
|
||
|
||
def test_cook_it_through_uses_routed_progressive_disclosure(self):
|
||
skill = main_loop_skill_text()
|
||
bundle = main_loop_bundle_text()
|
||
description = next(
|
||
line for line in skill.splitlines() if line.startswith("description:")
|
||
)
|
||
|
||
self.assertLessEqual(len(skill.splitlines()), 85)
|
||
self.assertLessEqual(len(skill.encode("utf-8")), 8_000)
|
||
self.assertLessEqual(sum(len(p.read_text().splitlines()) for p in main_loop_instruction_paths()), 330)
|
||
self.assertIn("main_loop.py <command> --help", skill)
|
||
self.assertNotIn("```bash", bundle)
|
||
self.assertNotIn("入口 1", description)
|
||
for command in subcommand_parsers():
|
||
self.assertNotIn(command, description)
|
||
for negative_boundary in ("纯 TSL 语法/API 查询", "commit message", "远端 Gitea CI"):
|
||
self.assertIn(negative_boundary, description)
|
||
|
||
for path in main_loop_instruction_paths()[1:]:
|
||
relative = path.relative_to(MAIN_LOOP_ROOT).as_posix()
|
||
other_text = "\n".join(
|
||
candidate.read_text(encoding="utf-8")
|
||
for candidate in main_loop_instruction_paths()
|
||
if candidate != path
|
||
)
|
||
self.assertIn(relative, other_text, msg=f"unrouted instruction file: {relative}")
|
||
self.assertNotIn("FILL:", bundle)
|
||
|
||
def test_cook_it_through_keeps_irrecoverable_red_lines_resident(self):
|
||
skill = normalized_prose(main_loop_skill_text())
|
||
rules = rules_text()
|
||
for required in (
|
||
"禁止手工修改 ticket `Status`",
|
||
"禁止伪造或复用证据 artifact",
|
||
"integration dependency 不可见时禁止继续",
|
||
"禁止 stash、reset 或覆盖其他 session 改动",
|
||
"远程 tracker、独立 clone 或跨机器状态",
|
||
):
|
||
self.assertIn(required, skill)
|
||
for migrated_rule in (
|
||
"main-loop:ticket-state",
|
||
"integration frontier",
|
||
"远程 tracker、独立 clone、跨机器状态",
|
||
):
|
||
self.assertNotIn(migrated_rule, rules)
|
||
|
||
def test_gitignore_template_tracks_scratch_and_ignores_only_runtime(self):
|
||
template = (TEMPLATES / "gitignore.template").read_text(encoding="utf-8")
|
||
for durable_rule in ("!/.scratch/", "!/.scratch/**"):
|
||
self.assertIn(durable_rule, template)
|
||
for runtime_rule in (
|
||
"/.scratch/*.lock",
|
||
"/.scratch/worktrees/",
|
||
"/.scratch/**/*.tmp",
|
||
):
|
||
self.assertIn(runtime_rule, template)
|
||
self.assertNotIn("是否纳入版本控制由项目决定", rules_text())
|
||
|
||
def test_cook_it_through_routes_entries_and_blast_radius_floor(self):
|
||
skill = main_loop_skill_text()
|
||
entries = section(skill, "## 任务路由", "## 常驻红线")
|
||
entry_headings = [h for h in headings(entries) if h.startswith("入口 ")]
|
||
self.assertEqual(
|
||
entry_headings,
|
||
[
|
||
"入口 1：直接执行",
|
||
"入口 2：单切片改动",
|
||
"入口 3：已明确预期行为的 bug",
|
||
"入口 4：新 feature 或设计变更",
|
||
],
|
||
)
|
||
self.assertEqual(entries.count("**升级条件**"), 2)
|
||
self.assertIn("AGENT_RULES.local.md", entries)
|
||
self.assertIn("高爆炸半径路径", entries)
|
||
self.assertIn("构建、CI 或分发配置", entries)
|
||
self.assertIn("最低入口 2", entries)
|
||
self.assertIn("入口 1 不加载按需文件", entries)
|
||
entry_two = normalized_prose(section(entries, "### 入口 2", "### 入口 3"))
|
||
entry_four = normalized_prose(entries.split("### 入口 4", 1)[1])
|
||
self.assertIn("不属于入口 3", entry_two)
|
||
self.assertIn("入口 4", entry_two)
|
||
self.assertIn("边界不清时先按入口 2 起步", entry_four)
|
||
|
||
def test_cook_it_through_routes_single_session_work(self):
|
||
workflow = main_loop_workflow_text("single-session")
|
||
for required in (
|
||
"当前 `HEAD` 为 review fixed point",
|
||
"`<fixed-point>` 作为 `code-review` 的 fixed point",
|
||
"仅运行 Standards axis",
|
||
"`diagnosing-bugs` 完整执行 Phase 1-6",
|
||
"improve-codebase-architecture",
|
||
):
|
||
self.assertIn(required, workflow)
|
||
self.assertNotIn("`<fixed-point>...HEAD`", workflow)
|
||
bug = section(workflow, "## 入口 3", "## 完成与升级")
|
||
self.assertNotIn("grill-with-docs", bug)
|
||
|
||
def test_cook_it_through_routes_feature_planning_and_onramps(self):
|
||
planning = main_loop_workflow_text("feature-planning")
|
||
for required in (
|
||
"`wayfinder`",
|
||
"`to-spec -> to-tickets`",
|
||
"`research`",
|
||
"先进入 `grill-with-docs`",
|
||
"首次运行 `setup-matt-pocock-skills`",
|
||
"seam confirmation 在 `to-spec` 与 `tdd`",
|
||
"`tdd` 不得在未经确认的 seam 上开始",
|
||
):
|
||
self.assertIn(required, planning)
|
||
self.assertNotIn("`prototype`", planning)
|
||
|
||
def test_cook_it_through_commits_planning_baseline_before_claim(self):
|
||
planning = main_loop_workflow_text("feature-planning")
|
||
ordered_steps = (
|
||
"-> to-spec",
|
||
"-> to-tickets",
|
||
"-> main_loop.py enqueue",
|
||
"-> 提交 planning baseline",
|
||
"-> main_loop.py claim",
|
||
)
|
||
positions = [planning.index(step) for step in ordered_steps]
|
||
self.assertEqual(positions, sorted(positions))
|
||
for durable_input in (
|
||
"`.scratch/<feature>/spec.md`",
|
||
"`.scratch/<feature>/issues/*.md`",
|
||
"`.scratch/queue.md`",
|
||
):
|
||
self.assertIn(durable_input, planning)
|
||
self.assertIn("任何 claim 前", planning)
|
||
self.assertIn("不隐式提交", planning)
|
||
|
||
def test_cook_it_through_defines_unattended_fallback(self):
|
||
planning = main_loop_workflow_text("feature-planning")
|
||
execution = main_loop_workflow_text("ticket-execution")
|
||
self.assertIn("尚未 claim ticket 时", planning)
|
||
self.assertIn("to-questionnaire", planning)
|
||
self.assertIn(".scratch/questions/<slug>.md", planning)
|
||
self.assertIn("从 `grill-with-docs` 恢复", normalized_prose(planning))
|
||
self.assertIn("finish --result blocked", execution)
|
||
|
||
def test_cook_it_through_binds_state_and_evidence_to_claim(self):
|
||
execution = normalized_prose(main_loop_workflow_text("ticket-execution"))
|
||
for key in (
|
||
"FEATURE",
|
||
"TICKET",
|
||
"CONTROL_ROOT",
|
||
"STATE_ROOT",
|
||
"WORKSPACE",
|
||
"BRANCH",
|
||
"BASE",
|
||
"ISOLATION",
|
||
):
|
||
self.assertIn(f"`{key}`", execution)
|
||
self.assertIn('--state-root "<PROJECT_ROOT>/.scratch"', execution)
|
||
self.assertIn('--repo-root "<PROJECT_ROOT>"', execution)
|
||
self.assertNotIn("--repo-root .", execution)
|
||
self.assertIn("把该 `FEATURE_HEAD` 合入 ticket branch", execution)
|
||
|
||
def test_cook_it_through_defines_ticket_execution_adapter(self):
|
||
execution = main_loop_workflow_text("ticket-execution")
|
||
ordered_steps = (
|
||
"读取已领取 ticket 的 spec",
|
||
"按 `tdd`",
|
||
"提交全部实现",
|
||
"运行 `code-review` 的 Standards/Spec",
|
||
"结构化证据调用 main_loop.py finish",
|
||
)
|
||
positions = [execution.index(step) for step in ordered_steps]
|
||
self.assertEqual(positions, sorted(positions))
|
||
for required in (
|
||
"fixed point",
|
||
"`<STATE_ROOT>/<feature>/spec.md`",
|
||
"`<STATE_ROOT>/<feature>/issues/<NN>-*.md`",
|
||
"零个未解决的硬 finding",
|
||
"不给 pass/fail 判定",
|
||
"不得填写 pass",
|
||
):
|
||
self.assertIn(required, execution)
|
||
|
||
def test_cook_it_through_defines_lease_and_stuck_ticket_recovery(self):
|
||
execution = normalized_prose(main_loop_workflow_text("ticket-execution"))
|
||
for required in (
|
||
"固定为 30 分钟",
|
||
"每 10 分钟",
|
||
"`reclaim` 只接管 stale 的 `claimed`",
|
||
"release-ticket",
|
||
"claim 环境准备失败",
|
||
"会占住该 ticket",
|
||
"原 `BASE`",
|
||
"blocked/skipped 必须给 reason",
|
||
):
|
||
self.assertIn(required, execution)
|
||
|
||
def test_cook_it_through_stops_on_integration_visibility_retry(self):
|
||
execution = main_loop_workflow_text("ticket-execution")
|
||
for key in (
|
||
"TICKET",
|
||
"DEPENDENCY",
|
||
"INTEGRATION_COMMIT",
|
||
"WORKSPACE",
|
||
"BRANCH",
|
||
"BRANCH_HEAD",
|
||
"TICKET_BRANCH",
|
||
"TICKET_BRANCH_HEAD",
|
||
"SYNC_BRANCH",
|
||
"MAIN_BRANCH",
|
||
"MAIN_HEAD",
|
||
"SYNC_COMMAND",
|
||
):
|
||
self.assertIn(f"`{key}`", execution)
|
||
self.assertIn("任一字段缺失", execution)
|
||
self.assertIn("取得正式 assignment 前不得继续", execution)
|
||
|
||
def test_cook_it_through_requires_fresh_evidence_artifacts(self):
|
||
bundle = main_loop_bundle_text()
|
||
execution = main_loop_workflow_text("ticket-execution")
|
||
integration = main_loop_workflow_text("feature-integration")
|
||
for required in (
|
||
"fresh UTF-8 JSON artifact",
|
||
"finish --help",
|
||
".scratch/<feature>/evidence/",
|
||
"无法证明命令真的执行过",
|
||
):
|
||
self.assertIn(required, execution)
|
||
for required in ("三个独立门禁，不能互相替代", "integrate --help"):
|
||
self.assertIn(required, integration)
|
||
self.assertIn("禁止伪造或复用证据 artifact", bundle)
|
||
for required in ("output_sha256", "report_sha256"):
|
||
self.assertIn(required, MAIN_LOOP.EVIDENCE_HELP)
|
||
self.assertIn("--main-verified", required_flags(subcommand_parsers()["integrate"]))
|
||
|
||
def test_cook_it_through_commits_final_state_after_integration(self):
|
||
integration = main_loop_workflow_text("feature-integration")
|
||
self.assertLess(
|
||
integration.index("main_loop.py integrate"),
|
||
integration.index("提交 final workflow state"),
|
||
)
|
||
for durable_path in (
|
||
"`.scratch/<feature>/`",
|
||
"`.scratch/queue.md`",
|
||
"`.main-loop.json`",
|
||
):
|
||
self.assertIn(durable_path, integration)
|
||
self.assertIn("不要运行 `git add .scratch`", integration)
|
||
self.assertIn("不得 amend/squash", integration)
|
||
self.assertIn("`MAIN_INTEGRATION_COMMIT`", integration)
|
||
|
||
def test_cook_it_through_orders_phase_boundary_options_and_reload(self):
|
||
boundary = normalized_prose(main_loop_session_boundary_text())
|
||
ordered_options = (
|
||
"继续当前 session",
|
||
"使用 `clear`",
|
||
"使用 `handoff`",
|
||
"交给 subagent",
|
||
"使用 `compact`",
|
||
)
|
||
positions = [boundary.index(option) for option in ordered_options]
|
||
self.assertEqual(positions, sorted(positions))
|
||
for required in (
|
||
"下一阶段需要当前 session 作为 primary source",
|
||
"约 150k tokens",
|
||
"`handoff` 解决的是可移植性",
|
||
"重新加载 `SKILL.md` 与当前路由文件",
|
||
"不得只依据 `status` 输出继续",
|
||
"`domain-modeling`",
|
||
"`codebase-design` 只作词汇来源",
|
||
):
|
||
self.assertIn(required, boundary)
|
||
|
||
def test_cook_it_through_delegates_command_semantics_to_help(self):
|
||
skill = main_loop_skill_text()
|
||
bundle = main_loop_bundle_text()
|
||
parsers = subcommand_parsers()
|
||
self.assertIn("main_loop.py <command> --help", skill)
|
||
self.assertIsNone(
|
||
re.search(r"\|\s*`?main_loop\.py (?:enqueue|status|claim|finish)", bundle),
|
||
msg="command responsibility tables duplicate argparse help",
|
||
)
|
||
for command, parser in parsers.items():
|
||
self.assertIn(f"main_loop.py {command}", bundle)
|
||
self.assertTrue(parser.description, msg=f"thin help for {command}")
|
||
for flag in required_flags(parser):
|
||
action = option_action(parser, flag)
|
||
self.assertNotIn(action.help, (None, argparse.SUPPRESS))
|
||
documented = set(re.findall(r"main_loop\.py ([a-z][a-z-]*)", bundle))
|
||
self.assertEqual(documented - set(parsers), set())
|
||
|
||
def test_cook_it_through_documents_executable_state_transitions(self):
|
||
execution = normalized_prose(main_loop_workflow_text("ticket-execution"))
|
||
finish = subcommand_parsers()["finish"]
|
||
result = option_action(finish, "--result")
|
||
self.assertEqual(
|
||
set(result.choices or ()),
|
||
{"resolved", "blocked", "released", "skipped"},
|
||
)
|
||
self.assertIn("blocked/skipped 必须给 reason", execution)
|
||
release_ticket = subcommand_parsers()["release-ticket"]
|
||
release_options = {
|
||
option
|
||
for action in release_ticket._actions # noqa: SLF001
|
||
for option in action.option_strings
|
||
}
|
||
self.assertNotIn("--repo-root", release_options)
|
||
self.assertNotIn("--owner", release_options)
|
||
|
||
def test_workflow_instructions_only_reference_installed_skills(self):
|
||
instructions = "\n".join((rules_text(), main_loop_bundle_text()))
|
||
skills = installed_skills()
|
||
machine_vocabulary = (
|
||
set(subcommand_parsers())
|
||
| set(MAIN_LOOP.ALLOWED_STATUSES)
|
||
| isolation_choices()
|
||
| NON_SKILL_BACKTICK_TOKENS
|
||
)
|
||
referenced = {
|
||
token
|
||
for token in re.findall(r"`([a-z][a-z0-9-]+)`", instructions)
|
||
if token not in machine_vocabulary
|
||
}
|
||
self.assertTrue(referenced)
|
||
self.assertEqual(sorted(referenced - skills), [])
|
||
|
||
def test_cook_it_through_limits_state_to_shared_local_markdown(self):
|
||
skill = normalized_prose(main_loop_skill_text())
|
||
self.assertIn("local Markdown tracker", skill)
|
||
self.assertIn("远程 tracker、独立 clone 或跨机器状态", skill)
|
||
self.assertNotIn("必须改用具备远程", skill)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|