Files
playbook/test/test_template_contracts.py
T

598 lines
23 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_SPEC = importlib.util.spec_from_file_location(
"playbook_main_loop_contracts", ROOT / "scripts" / "main_loop.py"
)
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 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 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"))
)
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.assertIn(".scratch/<feature>/spec.md", templates_readme)
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_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",
},
)
rules = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
stable_paths = (
"memory-bank/project-brief.md",
"memory-bank/tech-context.md",
"memory-bank/system-patterns.md",
)
for path in stable_paths:
self.assertIn(path, rules)
self.assertIn("进入 `grill-with-docs` 或本地 ticket 执行协议前", rules)
normalized_rules = " ".join(rules.split())
for required in (
"只记录下一 session 仍需要的稳定知识",
"写入 `tech-context.md` 的命令和环境事实必须已经验证",
"关键取舍及理由写入 `docs/adr/`",
"不把聊天流水、未验证猜测或短期进度写入 `CONTEXT.md`",
"没有长期价值的信息时不更新这些文件",
):
self.assertIn(required, normalized_rules)
def test_prompt_templates_are_not_part_of_the_workflow(self):
self.assertFalse(TEMPLATES.joinpath("prompts").exists())
def test_agent_rules_template_defines_ticket_and_integration_contracts(self):
rules = rules_text()
normalized = " ".join(rules.split())
self.assertIn("{{PLAYBOOK_ROOT}}", rules)
self.assertIn("{{PLAYBOOK_SCRIPTS}}", rules)
for heading in (
"## 任务入口",
"## 正式工程主链",
"## On-ramps 与 detours",
"## Phase boundaries",
"## 本地 Ticket 执行协议",
"## 文档职责",
"## 调度语义",
"## 执行隔离",
"## 主循环命令",
"## Git 与证据门禁",
"## 辅助能力",
"## Session 收尾",
):
self.assertIn(heading, rules, msg=f"missing section: {heading}")
for invariant in (
"`.scratch/queue.md`",
"多个 frontier tickets 可在 worktree 模式并发执行",
"只有 `reclaim` 可以接管",
"Standards/Spec 双轴审查",
"三个独立门禁,不能互相替代",
"跨机器或独立 clone",
"共享同一文件系统",
):
self.assertIn(invariant, normalized, msg=f"missing invariant: {invariant}")
for legacy in LEGACY_FLOW_TERMS:
self.assertNotIn(legacy, rules)
def test_agent_rules_states_whether_scratch_is_version_controlled(self):
rules = rules_text()
responsibilities = section(rules, "## 文档职责", "## 稳定知识维护")
self.assertIn("AGENT_RULES.local.md", responsibilities)
self.assertIn(
"纳入版本控制",
responsibilities,
msg="the review contract reads spec sources out of .scratch/, so "
"whether it is tracked has to be an explicit project decision",
)
def test_agent_rules_routes_each_current_matt_on_ramp_to_its_destination(self):
rules = rules_text()
on_ramps = section(rules, "## On-ramps 与 detours", "## Phase boundaries")
self.assertIn("`wayfinder`", on_ramps)
self.assertIn("`to-spec -> to-tickets`", on_ramps)
self.assertIn("`research`", on_ramps)
self.assertIn("先进入 `grill-with-docs`", on_ramps)
self.assertNotIn("`prototype`", on_ramps)
bug_route = section(rules, "### 入口 3", "### 入口 4")
self.assertIn("`diagnosing-bugs` 完整执行 Phase 1-6", bug_route)
self.assertNotIn("to-spec", bug_route)
self.assertIn(
"improve-codebase-architecture",
bug_route,
msg="diagnosing-bugs hands off to improve-codebase-architecture after "
"the fix lands, not to a design session before it",
)
self.assertNotIn(
"grill-with-docs",
bug_route,
msg="stopping a half-fixed defect to run a design session contradicts "
"the skill's own phase order",
)
def test_agent_rules_defines_a_local_ticket_execution_adapter(self):
rules = rules_text()
main_flow = section(rules, "## 正式工程主链", "## On-ramps 与 detours")
adapter = section(rules, "## 本地 Ticket 执行协议", "## 文档职责")
self.assertIn("本地 ticket 执行协议", main_flow)
ordered_steps = (
"读取已领取 ticket 的 spec",
"按 `tdd`",
"提交全部实现",
"运行 `code-review`",
"调用 `main_loop.py finish`",
)
positions = [adapter.index(step) for step in ordered_steps]
self.assertEqual(positions, sorted(positions))
def test_agent_rules_binds_state_and_evidence_to_claimed_git_context(self):
rules = rules_text()
claim = section(rules, "### 领取", "### 心跳和接管")
commands = " ".join(section(rules, "## 主循环命令", "## Git 与证据门禁").split())
# claim's contract is its output keys and what each one addresses.
for key in (
"FEATURE",
"TICKET",
"CONTROL_ROOT",
"STATE_ROOT",
"WORKSPACE",
"BRANCH",
"BASE",
"ISOLATION",
):
self.assertIn(f"`{key}`", claim, msg=f"claim output key undocumented: {key}")
self.assertNotIn(
"--state-root .scratch",
claim.split("stdout 返回", 1)[1],
msg="after a claim, state must be addressed by the absolute STATE_ROOT",
)
# Evidence has to be bound to the claimed commits, not to free text.
for binding in (
"--review-base <同一feature HEAD>",
"--verified \"<ticket-verification.json>\"",
"--reviewed \"<ticket-review.json>\"",
"--verified \"<feature-verification.json>\"",
"--main-verified \"<main-candidate-verification.json>\"",
"--reviewed \"<feature-review.json>\"",
"把该 `FEATURE_HEAD` 合入 ticket branch",
):
self.assertIn(binding, commands, msg=f"missing evidence binding: {binding}")
def test_agent_rules_defines_mechanical_review_inputs_and_pass_mapping(self):
rules = rules_text()
review = " ".join(
section(rules, "### Review 适配契约", "### Ticket 完成或状态转换").split()
)
for required in (
"fixed point",
"`.scratch/<feature>/spec.md`",
"`.scratch/<feature>/issues/<ticket>-*.md`",
"零个未解决的硬 finding",
"不得据此填写 `standards=pass` 或 `spec=pass`",
):
self.assertIn(required, review)
self.assertIn(
"不给 pass/fail 判定",
review,
msg="code-review emits findings only; the pass mapping is this "
"adapter's own layer and must not be presented as the skill's verdict",
)
def test_agent_rules_reads_claimed_context_after_claim_and_defines_lease_policy(self):
rules = rules_text()
startup = section(rules, "## 会话启动", "## 任务入口")
claim = section(rules, "### 领取", "### 心跳和接管")
lease = section(rules, "### 心跳和接管", "### Review 适配契约")
self.assertNotIn("当前 `.scratch/<feature>/spec.md`", startup)
self.assertNotIn("当前 `.scratch/<feature>/issues/<ticket>.md`", startup)
self.assertIn("领取成功后立即读取", claim)
self.assertIn("全局唯一", claim)
self.assertIn("每 10 分钟", lease)
self.assertIn("固定为 30 分钟", lease)
def test_agent_rules_orders_task_entries_by_cost_with_upgrade_conditions(self):
rules = rules_text()
entries = section(rules, "## 任务入口", "## 正式工程主链")
entry_headings = [h for h in headings(entries) if h.startswith("入口 ")]
self.assertEqual(
entry_headings,
[
"入口 1:直接执行",
"入口 2:单切片改动",
"入口 3:已明确预期行为的 bug",
"入口 4:新 feature 或设计变更",
],
msg="entries must stay ordered cheapest-first so the router can take "
"the first match",
)
self.assertEqual(
entries.count("**升级条件**"),
2,
msg="entry 1 and entry 2 each need an explicit upgrade trigger; "
"without one the router has no defined way out of a light path",
)
entry_two = section(entries, "### 入口 2", "### 入口 3")
entry_four = section(entries, "### 入口 4", "### 非交互模式下的入口 4")
self.assertIn(
"不属于入口 3",
entry_two,
msg="known bugs must reach diagnosing-bugs before the generic slice path",
)
self.assertIn("入口 4", entry_two, msg="entry 2 must name its escalation target")
self.assertIn(
"边界不清时先按入口 2 起步",
entry_four,
msg="an uncertain boundary must start at the single-slice path, not "
"pre-pay the full chain",
)
def test_agent_rules_gives_unattended_sessions_a_pre_ticket_fallback(self):
rules = rules_text()
fallback = section(
rules, "### 非交互模式下的入口 4", "## 正式工程主链"
)
self.assertIn("--result blocked", fallback)
self.assertIn(
"to-questionnaire",
fallback,
msg="grilling needs a user, so a ticketless unattended session must "
"have a defined way to hand questions back",
)
self.assertIn(".scratch/questions/<slug>.md", fallback)
self.assertIn("从 `grill-with-docs` 恢复", " ".join(fallback.split()))
def test_agent_rules_keeps_seam_confirmation_with_to_spec_and_tdd(self):
rules = rules_text()
main_flow = section(rules, "## 正式工程主链", "## On-ramps 与 detours")
self.assertIn("`tdd` 不得在未经确认的 seam 上开始", main_flow)
self.assertIn(
"seam confirmation 的责任在 `to-spec` 与 `tdd`",
main_flow,
msg="the grilling skills never mention seams, so the rules must not "
"route seam confirmation through them",
)
def test_agent_rules_orders_the_phase_boundary_options(self):
rules = rules_text()
phase_boundaries = section(rules, "## Phase boundaries", "## 本地 Ticket")
ordered_options = (
"继续当前 session",
"使用 `clear`",
"使用 `handoff`",
"交给 subagent",
"使用 `compact`",
)
positions = [phase_boundaries.index(option) for option in ordered_options]
self.assertEqual(positions, sorted(positions))
self.assertIn("150k", phase_boundaries)
self.assertIn("`handoff` 解决的是可移植性", phase_boundaries)
def test_agent_rules_documents_every_main_loop_subcommand_and_required_flag(self):
rules = rules_text()
parsers = subcommand_parsers()
for command, parser in parsers.items():
self.assertIn(
f"main_loop.py {command}",
rules,
msg=f"undocumented subcommand: {command}",
)
for flag in required_flags(parser):
self.assertIn(
flag, rules, msg=f"undocumented required flag: {command} {flag}"
)
documented = set(re.findall(r"main_loop\.py ([a-z][a-z-]*)", rules))
self.assertEqual(
documented - set(parsers),
set(),
msg="the rules document subcommands the CLI does not expose",
)
def test_agent_rules_documents_executable_state_transition_commands(self):
rules = rules_text()
lease = section(rules, "### 心跳和接管", "### Review 适配契约")
transitions = section(
rules, "### Ticket 完成或状态转换", "### Feature 顺序集成"
)
self.assertNotIn("finish --result released|blocked", lease)
self.assertIn("--result blocked --reason", transitions)
self.assertIn("--result released", transitions)
self.assertIn("其他 `finish` 转换共用", transitions)
release_ticket = transitions.split("main_loop.py release-ticket", 1)[1]
self.assertNotIn("--repo-root", release_ticket)
self.assertNotIn("--owner", release_ticket)
def test_agent_rules_requires_snapshotted_evidence_artifacts(self):
rules = rules_text()
gate = " ".join(section(rules, "## Git 与证据门禁", "## 辅助能力").split())
for required in (
"UTF-8 JSON artifact",
"output_sha256",
"report_sha256",
".scratch/<feature>/evidence/",
"无法证明命令真的执行过",
):
self.assertIn(required, gate)
def test_agent_rules_only_references_installed_skills(self):
rules = rules_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-]+)`", rules)
if token not in machine_vocabulary
}
self.assertTrue(referenced, msg="expected the rules to reference skills")
self.assertEqual(
sorted(referenced - skills),
[],
msg="the rules reference skills that are not installed under skills/",
)
def test_agent_rules_states_what_the_evidence_gate_cannot_check(self):
rules = rules_text()
gate = section(rules, "## Git 与证据门禁", "## 辅助能力")
self.assertIn("三个独立门禁,不能互相替代", gate)
self.assertIn(
"无法",
gate,
msg="the gate binds evidence to real commits but cannot prove a test "
"run happened; the rules must say so instead of implying enforcement",
)
self.assertIn("--main-verified", gate)
def test_agent_rules_documents_the_recovery_path_for_stuck_tickets(self):
rules = rules_text()
recovery = section(rules, "### 卡死与恢复", "## 执行隔离")
self.assertIn("release-ticket", recovery)
self.assertIn(
"`reclaim` 只接管 `claimed`",
recovery,
msg="reclaim cannot rescue a blocked ticket; the rules must point at "
"the command that can",
)
self.assertIn("BASE", recovery)
def test_agent_rules_limits_main_loop_to_shared_local_markdown_state(self):
rules = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
isolation = rules.split("## 执行隔离", 1)[1].split(
"## 主循环命令", 1
)[0]
normalized = " ".join(isolation.split())
self.assertIn("`main_loop.py` 只支持 local Markdown tracker", normalized)
self.assertIn("没有 远程 tracker adapter", normalized)
self.assertIn("跨机器或独立 clone 的并发不受支持", normalized)
self.assertNotIn("必须改用具备远程", normalized)
if __name__ == "__main__":
unittest.main()