csh/playbook
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨ feat(workflow): enforce auditable agent rules

Browse Source
This commit is contained in:
csh
2026-08-12 12:40:31 +08:00
parent 1d9590f6a1
commit 91b6f8453e
7 changed files with 1273 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/test_main_loop_cli.py
+265 -21
View File
@@ -1,4 +1,6 @@
import importlib.util
import hashlib
import json
import os
import subprocess
import sys
@@ -52,12 +54,60 @@ def parse_assignments(output: str) -> dict[str, str]:
return dict(line.split("=", 1) for line in output.splitlines() if "=" in line)
def verification_evidence(commit: str, detail: str) -> str:
return f"commit={commit}; result=pass; {detail}"
EVIDENCE_TMP = tempfile.TemporaryDirectory()
EVIDENCE_ROOT = Path(EVIDENCE_TMP.name)
def review_evidence(commit: str, base: str) -> str:
return f"commit={commit}; base={base}; standards=pass; spec=pass"
def write_evidence_artifact(data: dict[str, object]) -> str:
path = EVIDENCE_ROOT / f"evidence-{len(tuple(EVIDENCE_ROOT.iterdir())):04d}.json"
path.write_text(
json.dumps(data, ensure_ascii=False, sort_keys=True) + "\n",
encoding="utf-8",
newline="\n",
)
return str(path)
def verification_evidence(
commit: str,
detail: str,
*,
overrides: dict[str, object] | None = None,
) -> str:
output = f"{detail}: passed"
data: dict[str, object] = {
"version": 1,
"kind": "verification",
"commit": commit,
"result": "pass",
"command": detail,
"exit_code": 0,
"output": output,
"output_sha256": hashlib.sha256(output.encode("utf-8")).hexdigest(),
}
data.update(overrides or {})
return write_evidence_artifact(data)
def review_evidence(
commit: str,
base: str,
*,
overrides: dict[str, object] | None = None,
) -> str:
report = "Standards and spec have no unresolved hard findings"
data: dict[str, object] = {
"version": 1,
"kind": "review",
"commit": commit,
"base": base,
"standards": "pass",
"spec": "pass",
"report": report,
"report_sha256": hashlib.sha256(report.encode("utf-8")).hexdigest(),
}
data.update(overrides or {})
return write_evidence_artifact(data)
def init_repo(root: Path) -> None:
@@ -131,6 +181,9 @@ class MainLoopCliTests(unittest.TestCase):
"finish",
"heartbeat",
"reclaim",
"block-feature",
"release-feature",
"release-ticket",
"integrate",
):
self.assertIn(command, result.stdout)
@@ -925,6 +978,154 @@ class MainLoopCliTests(unittest.TestCase):
self.assertIn("FEATURE=alpha STATE=active PARTIAL=yes", final_status.stdout)
self.assertIn("FRONTIER=02", final_status.stdout)
def test_release_ticket_recovers_a_blocked_ticket_without_its_owner(self):
with tempfile.TemporaryDirectory() as tmp_dir:
root = Path(tmp_dir)
init_repo(root)
issues = write_feature(root, "alpha")
write_ticket(issues, "01", "first", "First")
enqueue = run_cli(
"enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root
)
self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr)
claim = run_cli(
"claim",
"--state-root",
".scratch",
"--repo-root",
".",
"--owner",
"lost-session",
"--isolation",
"worktree",
cwd=root,
)
self.assertEqual(claim.returncode, 0, msg=claim.stderr)
original = parse_assignments(claim.stdout)
blocked = run_cli(
"finish",
"--state-root",
".scratch",
"--repo-root",
".",
"--feature",
"alpha",
"--ticket",
"01",
"--owner",
"lost-session",
"--result",
"blocked",
"--reason",
"needs a product decision",
cwd=root,
)
self.assertEqual(blocked.returncode, 0, msg=blocked.stderr)
foreign_release = run_cli(
"finish",
"--state-root",
".scratch",
"--repo-root",
".",
"--feature",
"alpha",
"--ticket",
"01",
"--owner",
"new-session",
"--result",
"released",
cwd=root,
)
foreign_reclaim = run_cli(
"reclaim",
"--state-root",
".scratch",
"--repo-root",
".",
"--feature",
"alpha",
"--ticket",
"01",
"--owner",
"new-session",
cwd=root,
)
missing_reason = run_cli(
"release-ticket",
"--state-root",
".scratch",
"--feature",
"alpha",
"--ticket",
"01",
"--reason",
" ",
cwd=root,
)
recovered = run_cli(
"release-ticket",
"--state-root",
".scratch",
"--feature",
"alpha",
"--ticket",
"01",
"--reason",
"original session is gone",
cwd=root,
)
not_blocked_again = run_cli(
"release-ticket",
"--state-root",
".scratch",
"--feature",
"alpha",
"--ticket",
"01",
"--reason",
"already released",
cwd=root,
)
reclaimed = run_cli(
"claim",
"--state-root",
".scratch",
"--repo-root",
".",
"--owner",
"new-session",
"--isolation",
"worktree",
cwd=root,
)
self.assertEqual(foreign_release.returncode, 2)
self.assertIn("owned by another session", foreign_release.stderr)
self.assertEqual(foreign_reclaim.returncode, 2)
self.assertIn("is not claimed", foreign_reclaim.stderr)
self.assertEqual(missing_reason.returncode, 2)
self.assertIn("requires a reason", missing_reason.stderr)
self.assertEqual(recovered.returncode, 0, msg=recovered.stderr)
self.assertEqual(recovered.stdout.strip(), "TICKET_RELEASED=alpha/01")
self.assertEqual(not_blocked_again.returncode, 2)
self.assertIn("is not blocked", not_blocked_again.stderr)
self.assertEqual(reclaimed.returncode, 0, msg=reclaimed.stderr)
resumed = parse_assignments(reclaimed.stdout)
self.assertEqual(resumed["TICKET"], "01")
self.assertEqual(resumed["WORKSPACE"], original["WORKSPACE"])
self.assertEqual(resumed["BRANCH"], original["BRANCH"])
ticket = MAIN_LOOP.load_feature(root / ".scratch", "alpha").tickets["01"]
self.assertEqual(ticket.metadata["last_owner"], "lost-session")
self.assertEqual(
ticket.metadata["released_reason"], "original session is gone"
)
self.assertNotIn("blocked_reason", ticket.metadata)
events = [entry["event"] for entry in ticket.metadata["history"]]
self.assertIn("ticket-released", events)
def test_parallel_processes_claim_distinct_frontier_tickets(self):
if os.name == "nt":
self.assertIsNotNone(MAIN_LOOP.msvcrt)
@@ -1953,7 +2154,11 @@ class MainLoopCliTests(unittest.TestCase):
"--review-base",
context["BASE"],
"--verified",
"result=pass; ticket tests",
verification_evidence(
implementation_commit,
"ticket tests",
overrides={"commit": ""},
),
"--reviewed",
review_evidence(implementation_commit, context["BASE"]),
cwd=root,
@@ -1979,9 +2184,10 @@ class MainLoopCliTests(unittest.TestCase):
"--review-base",
context["BASE"],
"--verified",
(
f"commit={implementation_commit}; result=fail; "
"result=pass; ticket tests"
verification_evidence(
implementation_commit,
"ticket tests",
overrides={"output_sha256": "0" * 64},
),
"--reviewed",
review_evidence(implementation_commit, context["BASE"]),
@@ -1990,10 +2196,13 @@ class MainLoopCliTests(unittest.TestCase):
status = run_cli("status", "--state-root", ".scratch", cwd=root)
self.assertEqual(finish.returncode, 2)
self.assertIn("verification evidence must include commit", finish.stderr)
self.assertIn(
"verification evidence artifact field commit must be a non-empty string",
finish.stderr,
)
self.assertEqual(conflicting.returncode, 2)
self.assertIn(
"verification evidence must include exactly one result",
"verification evidence artifact output_sha256 mismatch",
conflicting.stderr,
)
self.assertIn("CLAIM=alpha/01", status.stdout)
@@ -2349,9 +2558,10 @@ class MainLoopCliTests(unittest.TestCase):
"--verified",
verification_evidence(implementation_commit, "ticket tests"),
"--reviewed",
(
f"commit={implementation_commit}; base={context['BASE']}; "
"standards=pass"
review_evidence(
implementation_commit,
context["BASE"],
overrides={"spec": ""},
),
cwd=root,
)
@@ -2365,12 +2575,23 @@ class MainLoopCliTests(unittest.TestCase):
)
self.assertEqual(missing_ticket_verification.returncode, 2)
self.assertIn(
"verification evidence is required",
"verification evidence artifact is required",
missing_ticket_verification.stderr,
)
self.assertEqual(incomplete_ticket_review.returncode, 2)
self.assertIn("spec=pass", incomplete_ticket_review.stderr)
self.assertIn(
"review evidence artifact field spec must be a non-empty string",
incomplete_ticket_review.stderr,
)
self.assertEqual(resolved.returncode, 0, msg=resolved.stderr)
ticket = MAIN_LOOP.load_feature(root / ".scratch", "alpha").tickets["01"]
ticket_evidence = ticket.metadata["evidence"]
self.assertEqual(set(ticket_evidence), {"verification", "review"})
for record in ticket_evidence.values():
self.assertRegex(record["sha256"], r"^[0-9a-f]{64}$")
self.assertTrue(
(root / ".scratch" / "alpha" / record["path"]).is_file()
)
feature_head = run_git(root, "rev-parse", "feature/alpha").stdout.strip()
main_before = run_git(root, "rev-parse", "main").stdout.strip()
@@ -2412,13 +2633,21 @@ class MainLoopCliTests(unittest.TestCase):
"--main-verified",
verification_evidence(feature_head, "main candidate tests"),
"--reviewed",
f"commit={feature_head}; base={main_before}; standards=pass",
review_evidence(
feature_head,
main_before,
overrides={"spec": ""},
),
cwd=root,
)
unbound_feature_verification = run_cli(
*integrate_args,
"--verified",
"result=pass; feature tests",
verification_evidence(
feature_head,
"feature tests",
overrides={"commit": ""},
),
"--main-verified",
verification_evidence(feature_head, "main candidate tests"),
"--reviewed",
@@ -2437,19 +2666,22 @@ class MainLoopCliTests(unittest.TestCase):
)
self.assertEqual(missing_feature_verification.returncode, 2)
self.assertIn(
"feature verification evidence is required",
"feature verification evidence artifact is required",
missing_feature_verification.stderr,
)
self.assertEqual(missing_main_verification.returncode, 2)
self.assertIn(
"main candidate verification evidence is required",
"main candidate verification evidence artifact is required",
missing_main_verification.stderr,
)
self.assertEqual(incomplete_final_review.returncode, 2)
self.assertIn("spec=pass", incomplete_final_review.stderr)
self.assertIn(
"review evidence artifact field spec must be a non-empty string",
incomplete_final_review.stderr,
)
self.assertEqual(unbound_feature_verification.returncode, 2)
self.assertIn(
"feature verification evidence must include commit",
"feature verification evidence artifact field commit must be a non-empty string",
unbound_feature_verification.stderr,
)
self.assertEqual(wrong_final_review_base.returncode, 2)
@@ -2478,6 +2710,18 @@ class MainLoopCliTests(unittest.TestCase):
self.assertIn("INTEGRATED=alpha", integrated.stdout)
self.assertIn("FEATURE=alpha STATE=integrated PARTIAL=no", status.stdout)
self.assertTrue((root / "feature.txt").is_file())
feature_state = MAIN_LOOP.load_feature_state(
MAIN_LOOP.load_feature(root / ".scratch", "alpha")
)
self.assertEqual(
set(feature_state["evidence"]),
{"feature_verification", "main_candidate_verification", "review"},
)
for record in feature_state["evidence"].values():
self.assertRegex(record["sha256"], r"^[0-9a-f]{64}$")
self.assertTrue(
(root / ".scratch" / "alpha" / record["path"]).is_file()
)
if __name__ == "__main__":
test/test_playbook.py
+75
View File
@@ -400,6 +400,81 @@ no_backup = true
self.assertIn("`.agents/tsl/index.md`", after)
self.assertIn("`.agents/markdown/index.md`", after)
def test_resync_refreshes_the_rules_block_and_keeps_project_additions(self):
with tempfile.TemporaryDirectory() as tmp_dir:
project_root = Path(tmp_dir) / "project"
project_root.mkdir()
playbook_root = MODE_ROOTS["snapshot"]
config = write_config(project_root, "snapshot", playbook_root)
seed = run_playbook(SCRIPT, config, project_root)
self.assertEqual(
seed.returncode, 0, msg=f"{seed.stdout}{seed.stderr}"
)
rules_md = project_root / "AGENT_RULES.md"
seeded = rules_md.read_text(encoding="utf-8")
self.assertIn("<!-- playbook:rules:start -->", seeded)
self.assertIn("<!-- playbook:rules:end -->", seeded)
# A project appendix outside the block, and drift inside it.
appendix = "\n## 项目补充\n\n保留这段项目自己的说明。\n"
drifted = seeded.replace("## 任务入口", "## 任务入口（本地改过）") + appendix
rules_md.write_text(drifted, encoding="utf-8", newline="\n")
resync = run_playbook(SCRIPT, config, project_root)
self.assertEqual(
resync.returncode, 0, msg=f"{resync.stdout}{resync.stderr}"
)
after = rules_md.read_text(encoding="utf-8")
self.assertIn(
"保留这段项目自己的说明。",
after,
msg="content outside the block belongs to the project",
)
self.assertIn(
"## 任务入口\n",
after,
msg="the process itself is playbook-owned and must be refreshed",
)
self.assertNotIn("## 任务入口（本地改过）", after)
self.assertEqual(after.count("<!-- playbook:rules:start -->"), 1)
legacy = project_root / "legacy" / "AGENT_RULES.md"
legacy.parent.mkdir()
legacy.write_text(
"# AGENT_RULES\n\n没有 marker 的旧文件。\n",
encoding="utf-8",
newline="\n",
)
legacy_config = project_root / "playbook-legacy.toml"
legacy_config.write_text(
f"""
[playbook]
project_root = "legacy"
playbook_root = "{playbook_root.as_posix()}"
install_mode = "snapshot"
[sync_rules]
date = "2026-01-01"
no_backup = true
""".lstrip(),
encoding="utf-8",
newline="\n",
)
legacy_run = run_playbook(SCRIPT, legacy_config, project_root)
self.assertEqual(
legacy_run.returncode,
0,
msg=f"{legacy_run.stdout}{legacy_run.stderr}",
)
self.assertEqual(
legacy.read_text(encoding="utf-8"),
"# AGENT_RULES\n\n没有 marker 的旧文件。\n",
msg="a file predating the markers must not be rewritten silently",
)
if __name__ == "__main__":
unittest.main()
test/test_template_contracts.py
+379 -92
View File
@@ -1,9 +1,97 @@
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",
@@ -103,14 +191,17 @@ class TemplateContractsTests(unittest.TestCase):
def test_templates_readme_separates_ownership_from_deployment_config(self):
templates_readme = (TEMPLATES / "README.md").read_text(encoding="utf-8")
classification = templates_readme.split("## 模板分类", 1)[1].split(
"## 模板说明", 1
)[0]
classification = section(templates_readme, "## 模板分类", "## 模板说明")
categories = [h for h in headings(classification) if h[:2] in ("1.", "2.", "3.")]
self.assertIn("### 1. 入口导航", classification)
self.assertIn("### 2. 初始化后由项目维护", classification)
self.assertIn("### 3. 参考模板", classification)
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)
@@ -151,61 +242,85 @@ class TemplateContractsTests(unittest.TestCase):
self.assertFalse(TEMPLATES.joinpath("prompts").exists())
def test_agent_rules_template_defines_ticket_and_integration_contracts(self):
rules = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
self.assertIn("Matt Pocock 工程流程与 ticket 执行约束", rules)
rules = rules_text()
normalized = " ".join(rules.split())
self.assertIn("{{PLAYBOOK_ROOT}}", rules)
self.assertIn("Playbook 模板/供应商目录", rules)
self.assertIn("setup-matt-pocock-skills", rules)
self.assertIn("grill-with-docs", rules)
self.assertIn("to-spec", rules)
self.assertIn("to-tickets", rules)
self.assertIn("一次可以先生成多个 feature", rules)
self.assertIn("`.scratch/queue.md`", rules)
self.assertIn("多个 frontier tickets 可在 worktree 模式并发执行", rules)
self.assertIn("不创建额外 worktree", rules)
self.assertIn("共享同一文件系统", rules)
self.assertIn("跨机器或独立 clone", " ".join(rules.split()))
self.assertIn("只有 `reclaim` 可以接管", rules)
self.assertIn("先提交实现", rules)
self.assertIn("Standards/Spec 双轴审查", rules)
self.assertIn("三个独立门禁，不能互相替代", 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_routes_each_matt_on_ramp_to_its_real_destination(self):
rules = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
on_ramps = rules.split("## On-ramps 与 detours", 1)[1].split(
"## Phase boundaries", 1
)[0]
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.assertIn("`handoff` 到独立目录运行", on_ramps)
self.assertIn("`prototype`", on_ramps)
self.assertIn("再 `handoff` 结论回原设计会话", on_ramps)
self.assertNotIn("`prototype`", on_ramps)
bug_route = rules.split("### 已明确预期行为的 bug", 1)[1].split(
"## 正式工程主链", 1
)[0]
bug_route = section(rules, "### 入口 3", "### 入口 4")
self.assertIn("`diagnosing-bugs` 完整执行 Phase 1-6", bug_route)
self.assertIn("不再进入 `implement` 或重复 `tdd`", 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 = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
main_flow = rules.split("## 正式工程主链", 1)[1].split(
"## On-ramps 与 detours", 1
)[0]
adapter = rules.split("## 本地 Ticket 执行协议", 1)[1].split(
"## 文档职责", 1
)[0]
normalized_adapter = " ".join(adapter.split())
rules = rules_text()
main_flow = section(rules, "## 正式工程主链", "## On-ramps 与 detours")
adapter = section(rules, "## 本地 Ticket 执行协议", "## 文档职责")
self.assertIn("本地 ticket 执行协议", main_flow)
self.assertNotIn("implement + tdd", main_flow)
self.assertIn("不直接调用上游 `implement`", normalized_adapter)
ordered_steps = (
"读取已领取 ticket 的 spec",
"按 `tdd`",
@@ -216,12 +331,46 @@ class TemplateContractsTests(unittest.TestCase):
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 = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
review_contract = rules.split("### Review 适配契约", 1)[1].split(
"### Ticket 完成或状态转换", 1
)[0]
normalized_review_contract = " ".join(review_contract.split())
rules = rules_text()
review = " ".join(
section(rules, "### Review 适配契约", "### Ticket 完成或状态转换").split()
)
for required in (
"fixed point",
@@ -230,15 +379,19 @@ class TemplateContractsTests(unittest.TestCase):
"零个未解决的硬 finding",
"不得据此填写 `standards=pass` 或 `spec=pass`",
):
self.assertIn(required, normalized_review_contract)
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 = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
startup = rules.split("## 会话启动", 1)[1].split("## 任务入口", 1)[0]
claim = rules.split("### 领取", 1)[1].split("### 心跳和接管", 1)[0]
lease = rules.split("### 心跳和接管", 1)[1].split(
"### Review 适配契约", 1
)[0]
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)
@@ -247,18 +400,75 @@ class TemplateContractsTests(unittest.TestCase):
self.assertIn("每 10 分钟", lease)
self.assertIn("固定为 30 分钟", lease)
def test_agent_rules_preserves_to_spec_seam_and_phase_boundary_contracts(self):
rules = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
main_flow = rules.split("## 正式工程主链", 1)[1].split(
"## On-ramps 与 detours", 1
)[0]
self.assertIn("不重新进行已经完成的需求采访", main_flow)
self.assertIn("完成 seam confirmation", main_flow)
self.assertIn("`tdd` 不得在未经确认的 seam 上开始", main_flow)
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("入口 ")]
phase_boundaries = rules.split("## Phase boundaries", 1)[1].split(
"## 文档职责", 1
)[0]
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`",
@@ -267,43 +477,120 @@ class TemplateContractsTests(unittest.TestCase):
"使用 `compact`",
)
positions = [phase_boundaries.index(option) for option in ordered_options]
self.assertEqual(positions, sorted(positions))
self.assertIn("上下文过长不等于必须 `handoff`", phase_boundaries)
self.assertIn("150k", phase_boundaries)
self.assertIn("`handoff` 解决的是可移植性", phase_boundaries)
def test_agent_rules_binds_state_and_evidence_to_claimed_git_context(self):
rules = (TEMPLATES / "AGENT_RULES.template.md").read_text(encoding="utf-8")
execution = rules.split("### 领取", 1)[1]
after_claim_output = execution.split("stdout 返回", 1)[1]
def test_agent_rules_documents_every_main_loop_subcommand_and_required_flag(self):
rules = rules_text()
parsers = subcommand_parsers()
self.assertIn("绝对 `STATE_ROOT`", execution)
self.assertIn("`CONTROL_ROOT` 是共享 Git control checkout", execution)
self.assertIn("`WORKSPACE` 是当前 ticket 的代码工作区", execution)
self.assertNotIn("--state-root .scratch", after_claim_output)
self.assertIn("--review-base <同一feature HEAD>", execution)
self.assertIn(
'commit=<HEAD>; base=<review-base>; standards=pass; spec=pass',
execution,
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",
)
self.assertIn("把该 `FEATURE_HEAD` 合入 ticket branch", execution)
self.assertIn(
"`--feature-head`、`--review-base` 和 review evidence 的 `base`",
execution,
def test_agent_rules_documents_executable_state_transition_commands(self):
rules = rules_text()
lease = section(rules, "### 心跳和接管", "### Review 适配契约")
transitions = section(
rules, "### Ticket 完成或状态转换", "### Feature 顺序集成"
)
self.assertIn(
"base=<最新main HEAD>; standards=pass; spec=pass",
execution,
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", isolation)
self.assertIn("没有远程 tracker adapter", isolation)
self.assertIn("跨机器或独立\nclone 的并发不受支持", isolation)
self.assertNotIn("必须改用具备远程", isolation)
self.assertIn("`main_loop.py` 只支持 local Markdown tracker", normalized)
self.assertIn("没有 远程 tracker adapter", normalized)
self.assertIn("跨机器或独立 clone 的并发不受支持", normalized)
self.assertNotIn("必须改用具备远程", normalized)
if __name__ == "__main__":