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//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 ( "", "", "", "", ): self.assertIn(marker, lines, msg=f"missing marker: {marker}") framework_start = lines.index("") framework_end = lines.index("") agents_start = lines.index("") agents_end = lines.index("") 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_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) rules = rules_text() self.assertNotIn("是否纳入版本控制由项目决定", rules) 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_commits_planning_baseline_before_claim(self): rules = rules_text() main_chain = section(rules, "## 正式工程主链", "## On-ramps 与 detours") ordered_steps = ( "-> to-spec", "-> to-tickets", "-> main_loop.py enqueue", "-> 提交 planning baseline", "-> main_loop.py claim", ) positions = [main_chain.index(step) for step in ordered_steps] self.assertEqual(positions, sorted(positions)) for durable_input in ( "`.scratch//spec.md`", "`.scratch//issues/*.md`", "`.scratch/queue.md`", ): self.assertIn(durable_input, main_chain) self.assertIn("任何 `claim` 之前", main_chain) self.assertIn("不隐式提交", main_chain) def test_agent_rules_commits_final_workflow_state_after_integration(self): rules = rules_text() main_chain = section(rules, "## 正式工程主链", "## On-ramps 与 detours") integration = section(rules, "### Feature 顺序集成", "## Git 与证据门禁") self.assertLess( main_chain.index("-> main_loop.py integrate"), main_chain.index("-> 提交 final workflow state"), ) for durable_path in ( "`.scratch//`", "`.scratch/queue.md`", "`.scratch//.main-loop.json`", ): self.assertIn(durable_path, integration) self.assertIn("不得用\n`git add .scratch`", integration) self.assertIn("不得 amend 或 squash", integration) self.assertIn("`MAIN_INTEGRATION_COMMIT`", integration) 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 \"\"", "--reviewed \"\"", "--verified \"\"", "--main-verified \"\"", "--reviewed \"\"", "把该 `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//spec.md`", "`.scratch//issues/-*.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//spec.md`", startup) self.assertNotIn("当前 `.scratch//issues/.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/.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//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()