From 544f1877158b123b00cb4d4c9711e5d1558bc21c Mon Sep 17 00:00:00 2001 From: wlvh <30534800+wlvh@users.noreply.github.com> Date: Mon, 20 Jul 2026 15:40:12 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20PASS=20completion=20?= =?UTF-8?q?=E9=94=99=E8=AF=AF=E6=88=90=E5=8A=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- en/scripts/OPERATIONS.md | 6 +- tests/test_sync_coding_workflow.py | 9 +- tests/test_workflow_sync_skill.py | 534 +++++++++++++++--- zh/scripts/OPERATIONS.md | 32 +- zh/skills/workflow-docs-sync/evals/README.md | 6 + .../workflow-docs-sync/scripts/harness.py | 424 ++++++++++++-- 6 files changed, 867 insertions(+), 144 deletions(-) diff --git a/en/scripts/OPERATIONS.md b/en/scripts/OPERATIONS.md index bb27a39..e882b62 100644 --- a/en/scripts/OPERATIONS.md +++ b/en/scripts/OPERATIONS.md @@ -53,6 +53,7 @@ Common rules for every pass: - Use `.coding_workflow/diffs/pr_body_skeleton.md` or current `PR_BODY.md` as the PR body structure authority. - Do not modify sync sentinels or content outside sentinel sections. +- A literal `|` inside a table cell must be written as `\|` or replaced with `
`; only an unescaped `|` separates cells. - Do not fill `pr_test_evidence`; only the PR submission agent owns it. - `agent_execution_evidence` is self-reported read coverage for reviewer spot checks only. - `Full Document Reconcile` must include upstream semantic delta, adopted where, not adopted because, evidence, and downstream impact. @@ -74,8 +75,9 @@ sections. Current task: execute only PASS 1 - Code Facts / Architecture. Must read: 1. `.coding_workflow/diffs/agent_workorder.md` 2. `.coding_workflow/diffs/upstream_full/architecture.md` -3. `PR_BODY.md`; initialize it from `.coding_workflow/diffs/pr_body_skeleton.md` - if missing; stop if the skeleton is missing. +3. `PR_BODY.md` must be supplied as the formal handoff artifact before this + PASS starts; if it is missing, stop and report a workflow-preparation + artifact failure. Do not create it from the skeleton during the PASS. 4. `PR_BODY.md` repo_facts_map and the `architecture.md` row in full_document_reconcile. 5. Current repo entrypoints, module boundaries, data flow, state model, diff --git a/tests/test_sync_coding_workflow.py b/tests/test_sync_coding_workflow.py index 2f8ccda..19c1b3d 100644 --- a/tests/test_sync_coding_workflow.py +++ b/tests/test_sync_coding_workflow.py @@ -472,7 +472,9 @@ def test_each_pass_prompt_contains_common_execution_rules(self) -> None: ".coding_workflow/diffs/agent_workorder.md", ".coding_workflow/diffs/pr_body_skeleton.md", "PR_BODY.md", - "初始化；如果 skeleton 缺失", + "应由 PREPARE 机械创建", + "不得自行复制 skeleton", + "表格 cell 中的字面 `|` 必须写为 `\\|`", SYNC_MODULE.SYNC_AUTO_START, SYNC_MODULE.SYNC_AUTO_END, "任何 sync sentinel、sentinel 外内容", @@ -517,10 +519,11 @@ def test_each_pass_prompt_contains_common_execution_rules(self) -> None: )[0] self.assertNotIn("本文档", pass_text) self.assertEqual( - pass_text.count("如果不存在，先用"), + pass_text.count("应由 PREPARE 机械创建"), 1, - msg=f"{title} prompt should define PR_BODY init once", + msg=f"{title} prompt should bind PR_BODY to PREPARE once", ) + self.assertNotIn("如果不存在，先用", pass_text) self.assertNotIn("回报修改了哪些文件", pass_text) self.assertNotIn("回报是否写入 downstream impact", pass_text) self.assertNotIn("PR body sections", pass_text) diff --git a/tests/test_workflow_sync_skill.py b/tests/test_workflow_sync_skill.py index 2e42f41..ca72293 100644 --- a/tests/test_workflow_sync_skill.py +++ b/tests/test_workflow_sync_skill.py @@ -60,21 +60,21 @@ ) -def load_sync_module() -> Any: - """加载 sync 实现，返回机械 PASS ownership 真相源。""" - module_path = REPO_ROOT / "scripts/sync_coding_workflow.py" - spec = importlib.util.spec_from_file_location( - "sync_coding_workflow_skill_test", - module_path, - ) +def load_module(*, module_path: Path, module_name: str) -> Any: + """加载指定 Python 文件，返回测试所需的真实生产模块。""" + spec = importlib.util.spec_from_file_location(module_name, module_path) if spec is None or spec.loader is None: - raise AssertionError("无法加载 sync_coding_workflow.py") + raise AssertionError(f"无法加载 {module_path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module -SYNC_MODULE = load_sync_module() +SYNC_MODULE = load_module( + module_path=REPO_ROOT / "scripts/sync_coding_workflow.py", + module_name="sync_test", +) +HARNESS_MODULE = load_module(module_path=HARNESS, module_name="harness_test") def run_command( @@ -196,6 +196,9 @@ def create_upstream( "printf 'workorder\\n' > " ".coding_workflow/diffs/agent_workorder.md\n" "printf '{}\\n' > .coding_workflow/diffs/sync_state.json\n" + "[ ! -f PR_BODY.md ] || " + "[ -f .coding_workflow/diffs/pr_body_skeleton.md ] || " + "cp PR_BODY.md .coding_workflow/diffs/pr_body_skeleton.md\n" "if [ \"$#\" -eq 1 ] && [ \"$1\" = \"--final\" ]; then\n" f"{final_lines}" "fi\n" @@ -207,19 +210,46 @@ def create_upstream( def pr_body_text() -> str: - """返回包含全部 stable sentinel 的最小测试 PR body。""" - sections: list[str] = [""] - for section_name in AGENT_SECTIONS: - sections.extend( - ( - f"", - f"## {section_name}", - "", - f"{section_name} baseline", - f"", - ) + """返回保留真实 heading、表头和全部逐 PASS 行的 sentinel body。""" + reconcile = [ + "## Full Document Reconcile", + "", + "| " + " | ".join(SYNC_MODULE.FULL_RECONCILE_COLUMNS) + " |", + "| " + + " | ".join("---" for _ in SYNC_MODULE.FULL_RECONCILE_COLUMNS) + + " |", + ] + for sync_pass in SYNC_MODULE.SYNC_PASSES: + for relative_path in sync_pass["files"]: + reconcile.append( + f"| {sync_pass['title']} | `{relative_path}` | `fixture` | " + "待补充 | 待补充 | 待补充 | 待补充 | 待补充 |" + ) + contents = ( + SYNC_MODULE.render_repo_facts_template(), + "\n".join(reconcile), + SYNC_MODULE.render_pr_test_evidence_template(), + SYNC_MODULE.render_upstream_drift_log_template(), + SYNC_MODULE.render_agent_execution_evidence_template(), + SYNC_MODULE.render_remaining_human_decisions_template().replace( + "- none", "- 待判断: fixture\n- concrete fixture decision"), + ) + body = [ + SYNC_MODULE.SYNC_PR_BODY_MARKER, + SYNC_MODULE.wrap_agent_section( + section_name="repo_facts_map", + content=contents[0], + ), + "\nfixture\n", + ] + body.extend( + SYNC_MODULE.wrap_agent_section( + section_name=section_name, + content=content, ) - return "\n".join(sections) + "\n" + for section_name, content in zip(AGENT_SECTIONS[1:], contents[1:]) + ) + return "\n\n".join(body) + "\n" def initialize_core_target( @@ -361,6 +391,17 @@ def run_harness( ) +def finish_one(*, target: Path, upstream: Path, sha: str) -> Any: + """运行 PASS_1 finish，返回可供连续失败重试断言的结果。""" + return run_harness( + command="finish-pass", + target=target, + upstream=upstream, + upstream_sha=sha, + mode="PASS_1", + ) + + def append_text(*, path: Path, text: str) -> None: """向 UTF-8 文件末尾追加测试内容。""" path.write_text( @@ -391,6 +432,77 @@ def replace_section( ) +def section_text(*, path: Path, section_name: str) -> str: + """读取指定 sentinel section，返回不含边界标记的原始文本。""" + content = path.read_text(encoding="utf-8") + start = f"" + end = f"" + return content[content.index(start) + len(start):content.index(end)] + + +def complete_pass_body(*, path: Path, mode: str) -> None: + """只完成当前 PASS 的既有 heading/owned rows/evidence row。""" + ownership = json.loads(OWNERSHIP.read_text(encoding="utf-8")) + entry = ownership[mode] + if mode == "PASS_1": + repo_facts = section_text( + path=path, + section_name="repo_facts_map", + ) + replace_section( + path=path, + section_name="repo_facts_map", + text=repo_facts.replace("证据: 待补充", "证据: fixture evidence"), + ) + + # 逐行替换能保留表头、未来 PASS 行和完整 handoff 结构。 + reconcile = section_text( + path=path, + section_name="full_document_reconcile", + ).splitlines() + for relative_path in entry["owned"]: + matches = [ + index + for index, line in enumerate(reconcile) + if f"`{relative_path}`" in line + ] + if len(matches) != 1: + raise AssertionError(f"reconcile row mismatch: {relative_path}") + values = ( + "## Full Document Reconcile", "none", "待判断", r"literal \| pipe", + "none", + ) + for value in values: + reconcile[matches[0]] = reconcile[matches[0]].replace( + "待补充", value, 1, + ) + replace_section( + path=path, + section_name="full_document_reconcile", + text="\n".join(reconcile), + ) + title = entry["title"].split(maxsplit=1)[1] + evidence = section_text( + path=path, + section_name="agent_execution_evidence", + ).splitlines() + matches = [ + index for index, line in enumerate(evidence) + if f"| {title} |" in line + ] + if len(matches) != 1: + raise AssertionError(f"execution row mismatch: {mode}") + for value in ("## Agent Execution Evidence", "none", r"x \| y"): + evidence[matches[0]] = evidence[matches[0]].replace( + "待补充", value, 1, + ) + replace_section( + path=path, + section_name="agent_execution_evidence", + text="\n".join(evidence), + ) + + def complete_prepare( *, target: Path, @@ -414,9 +526,9 @@ def complete_pass( upstream: Path, upstream_sha: str, mode: str, - relative_path: str, + relative_paths: list[str], ) -> None: - """按 start/semantic edit/finish 顺序完成一个 PASS。""" + """按 start、逐行 body/owned docs 修改、finish 完成一个 PASS。""" start = run_harness( command="start-pass", target=target, @@ -426,10 +538,12 @@ def complete_pass( ) if start.returncode != 0: raise AssertionError(start.stdout + start.stderr) - append_text( - path=target / relative_path, - text=f"\n{mode} semantic edit\n", - ) + complete_pass_body(path=target / "PR_BODY.md", mode=mode) + for relative_path in relative_paths: + append_text( + path=target / relative_path, + text=f"\n{mode} semantic edit\n", + ) finish = run_harness( command="finish-pass", target=target, @@ -454,18 +568,22 @@ def complete_prepare_and_passes( upstream_sha=upstream_sha, ) edits = { - "PASS_1": "architecture.md", - "PASS_2": "interact.md", - "PASS_3": "TESTING.md", - "PASS_4": "AGENTS.md", + "PASS_1": ["architecture.md"], + "PASS_2": [ + "capability_contract.json", + "interact.md", + "docs/business_user_guide.md", + ], + "PASS_3": ["TESTING.md"], + "PASS_4": ["PR_Checklist.md", "SOP.md", "AGENTS.md"], } - for mode, relative_path in edits.items(): + for mode, relative_paths in edits.items(): complete_pass( target=target, upstream=upstream, upstream_sha=upstream_sha, mode=mode, - relative_path=relative_path, + relative_paths=relative_paths, ) @@ -475,56 +593,24 @@ def complete_real_prepare_and_passes( upstream: Path, upstream_sha: str, ) -> None: - """用真实 skeleton/sync 完成 PREPARE 和四个 ownership PASS。""" + """用正式 body 保留全表结构并逐 PASS 完成已有责任行。""" complete_prepare( target=target, upstream=upstream, upstream_sha=upstream_sha, ) - start = run_harness( - command="start-pass", - target=target, - upstream=upstream, - upstream_sha=upstream_sha, - mode="PASS_1", - ) - if start.returncode != 0: - raise AssertionError(start.stdout + start.stderr) - shutil.copy2( - target / ".coding_workflow/diffs/pr_body_skeleton.md", - target / "PR_BODY.md", - ) - for section_name in ( - "repo_facts_map", - "full_document_reconcile", - "agent_execution_evidence", - ): - replace_section( - path=target / "PR_BODY.md", - section_name=section_name, - text=f"## {section_name}\n\nreal PASS evidence", - ) - append_text(path=target / "architecture.md", text="\nPASS_1 edit\n") - finish = run_harness( - command="finish-pass", - target=target, - upstream=upstream, - upstream_sha=upstream_sha, - mode="PASS_1", - ) - if finish.returncode != 0: - raise AssertionError(finish.stdout + finish.stderr) - for mode, relative_path in { - "PASS_2": "interact.md", - "PASS_3": "TESTING.md", - "PASS_4": "AGENTS.md", + for mode, relative_paths in { + "PASS_1": ["architecture.md"], + "PASS_2": ["interact.md"], + "PASS_3": ["TESTING.md"], + "PASS_4": ["AGENTS.md"], }.items(): complete_pass( target=target, upstream=upstream, upstream_sha=upstream_sha, mode=mode, - relative_path=relative_path, + relative_paths=relative_paths, ) @@ -653,6 +739,10 @@ def test_titles_ownership_and_transport_alignment(self) -> None: ) } self.assertEqual(set(ownership), set(expected)) + self.assertEqual( + HARNESS_MODULE.PERMITTED_INHERIT_PATHS, + set(SYNC_MODULE.PERMITTED_INHERIT_FILES), + ) operations = OPERATIONS.read_text(encoding="utf-8") for mode in PASS_MODES: self.assertEqual(ownership[mode]["title"], expected[mode]["title"]) @@ -745,16 +835,13 @@ def test_prepare_requires_clean_pinned_upstream(self) -> None: ) self.assertNotEqual(dirty_upstream.returncode, 0) - def test_prepare_installs_missing_templates(self) -> None: - """真实 sync 安装模板，PASS 可从 skeleton 初始化 PR body。""" + def test_prepare_creates_body_and_pass_one_retries_in_place(self) -> None: + """Case A 各失败点保留 active PASS，补全责任项后原地成功。""" with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) upstream = root / "upstream" target = root / "target" - sha = create_upstream( - repo_root=upstream, - fake_sync=False, - ) + sha = create_upstream(repo_root=upstream, fake_sync=False) initialize_repo(repo_root=target) result = run_harness( command="prepare", @@ -771,6 +858,18 @@ def test_prepare_installs_missing_templates(self) -> None: / ".coding_workflow/skill_results/PREPARE.json" ).is_file() ) + body_path = target / "PR_BODY.md" + skeleton = target / ".coding_workflow/diffs/pr_body_skeleton.md" + self.assertTrue(body_path.is_file()) + self.assertEqual( + body_path.read_text(encoding="utf-8"), + skeleton.read_text(encoding="utf-8"), + ) + body = body_path.read_text(encoding="utf-8") + self.assertTrue(all( + body.count(f"") == 1 + for name in AGENT_SECTIONS + )) start = run_harness( command="start-pass", target=target, @@ -779,27 +878,274 @@ def test_prepare_installs_missing_templates(self) -> None: mode="PASS_1", ) self.assertEqual(start.returncode, 0, msg=start.stdout) - shutil.copy2( - target / ".coding_workflow/diffs/pr_body_skeleton.md", - target / "PR_BODY.md", + append_text( + path=target / "architecture.md", + text="\nPASS_1 semantic edit\n", ) + incomplete = finish_one(target=target, upstream=upstream, sha=sha) + self.assertNotEqual(incomplete.returncode, 0) + self.assertEqual( + parse_json(result=incomplete)["error"], + "PASS PR_BODY 未完成", + ) + run_path = target / ".coding_workflow/skill_runtime/run.json" + state = json.loads(run_path.read_text(encoding="utf-8")) + self.assertEqual(state["active_mode"], "PASS_1") + body_path.unlink() + deleted = finish_one(target=target, upstream=upstream, sha=sha) + self.assertNotEqual(deleted.returncode, 0) + self.assertEqual( + parse_json(result=deleted)["error"], + "PR_BODY scope 无法校验", + ) + baseline = json.loads( + ( + target + / ".coding_workflow/skill_runtime/baselines/PASS_1.json" + ).read_text(encoding="utf-8") + ) + body_path.write_text(baseline["pr_body_text"], encoding="utf-8") + pending_execution = section_text( + path=body_path, + section_name="agent_execution_evidence", + ) + complete_pass_body(path=body_path, mode="PASS_1") replace_section( - path=target / "PR_BODY.md", - section_name="repo_facts_map", - text="PASS_1 smoke", + path=body_path, + section_name="agent_execution_evidence", + text=pending_execution, + ) + pending = finish_one(target=target, upstream=upstream, sha=sha) + self.assertNotEqual(pending.returncode, 0) + self.assertIn("PASS 1", parse_json(result=pending)["detail"]) + current_state = json.loads(run_path.read_text(encoding="utf-8")) + self.assertEqual(current_state["active_mode"], "PASS_1") + self.assertFalse( + ( + target / ".coding_workflow/skill_results/PASS_1.json" + ).exists() + ) + + complete_pass_body(path=body_path, mode="PASS_1") + done = body_path.read_text(encoding="utf-8") + fact_evidence = "证据: fixture evidence" + reconcile_values = ( + r"## Full Document Reconcile | none | 待判断 | " + r"literal \| pipe | none" + ) + execution_values = r"## Agent Execution Evidence | none | x \| y" + execution_header = ( + "| pass | required files read | key facts observed | " + "skipped files / reason |" + ) + invalid_bodies = { + "repo_empty": done.replace(f"{fact_evidence}\n", "", 1), + "evidence_empty": done.replace(fact_evidence, "证据:", 1), + "reconcile_empty": done.replace( + reconcile_values, " | | | | ", 1, + ), + "execution_empty": done.replace( + execution_values, " | | ", 1, + ), + "literal_pipe": done.replace( + r"literal \| pipe", "literal | pipe", 1, + ), + "execution_header": done.replace( + execution_header, "| bad | bad | bad | bad | bad |", 1, + ), + } + for scenario, invalid_body in invalid_bodies.items(): + with self.subTest(completion=scenario): + body_path.write_text(invalid_body, encoding="utf-8") + invalid = finish_one( + target=target, upstream=upstream, sha=sha, + ) + self.assertNotEqual(invalid.returncode, 0) + payload = parse_json(result=invalid) + self.assertEqual(payload["error"], "PASS PR_BODY 未完成") + if scenario == "literal_pipe": + self.assertIn( + "表格 cell 中的字面 | 必须写为 \\| 或改用
", + payload["detail"], + ) + with self.assertRaises(HARNESS_MODULE.HarnessError): + HARNESS_MODULE.assert_remaining_decisions_complete( + section_text="## Remaining Human Decisions\n", + ) + body_path.write_text(done, encoding="utf-8") + upstream_architecture = ( + upstream / "zh/architecture.md" + ).read_text(encoding="utf-8") + (target / "architecture.md").write_bytes( + upstream_architecture.replace("\n", "\r\n").encode("utf-8") + ) + template_copy = finish_one( + target=target, upstream=upstream, sha=sha + ) + self.assertNotEqual(template_copy.returncode, 0) + self.assertEqual( + parse_json(result=template_copy)["error"], + "PASS owned 文档未项目化", + ) + self.assertIn( + "architecture.md", + parse_json(result=template_copy)["detail"], ) append_text( path=target / "architecture.md", - text="\nPASS_1 semantic edit\n", + text="\nproject architecture evidence\n", ) - finish = run_harness( - command="finish-pass", - target=target, - upstream=upstream, - upstream_sha=sha, - mode="PASS_1", + self.assertIn( + "| PASS 2 - Capability / User Behavior | " + "`capability_contract.json` | `installed_template` | 待补充", + body_path.read_text(encoding="utf-8"), ) + finish = finish_one(target=target, upstream=upstream, sha=sha) self.assertEqual(finish.returncode, 0, msg=finish.stdout) + state = json.loads(run_path.read_text(encoding="utf-8")) + self.assertEqual(state["completed_modes"], ["PREPARE", "PASS_1"]) + self.assertIsNone(state["active_mode"]) + pass_result = json.loads( + ( + target / ".coding_workflow/skill_results/PASS_1.json" + ).read_text(encoding="utf-8") + ) + self.assertEqual(pass_result["status"], "passed") + self.assertEqual( + pass_result["changed_paths"], + ["PR_BODY.md", "architecture.md"], + ) + + def test_prepare_preserves_valid_body_and_rejects_damage(self) -> None: + """PREPARE 迁移合法旧 schema，并在 sync 前拒绝损坏 body。""" + with tempfile.TemporaryDirectory() as temp_dir: + root = Path(temp_dir) + upstream = root / "upstream" + sha = create_upstream(repo_root=upstream, fake_sync=False) + valid = pr_body_text().replace( + "证据: 待补充", "证据: legacy retained", 1, + ) + sections = SYNC_MODULE.preserved_agent_sections( + text=valid, state={}, + ) + blocks = { + name: SYNC_MODULE.wrap_agent_section( + section_name=name, content=sections[name], + ) + for name in AGENT_SECTIONS + } + reordered = valid.replace( + blocks["repo_facts_map"], "", 1, + ).replace( + blocks["full_document_reconcile"], + blocks["repo_facts_map"], 1, + ).replace("", blocks["full_document_reconcile"], 1) + last_heading = SYNC_MODULE.REPO_FACTS_HEADINGS[-1] + old_repo = valid.replace( + f"{last_heading}\n证据: 待补充", "", 1, + ) + successes = { + "valid": valid, + "missing_one": old_repo.replace( + blocks["pr_test_evidence"], "", + ), + "missing_two": valid.replace( + blocks["pr_test_evidence"], "", + ).replace(blocks["upstream_drift_log"], ""), + "reordered": reordered, + } + for scenario, old_body in successes.items(): + with self.subTest(migration=scenario): + target = root / scenario + initialize_core_target(target=target, upstream=upstream) + body_path = target / "PR_BODY.md" + body_path.write_text(old_body, encoding="utf-8") + result = run_harness( + command="prepare", target=target, + upstream=upstream, upstream_sha=sha, + ) + self.assertEqual(result.returncode, 0, msg=result.stdout) + actual = body_path.read_text(encoding="utf-8") + HARNESS_MODULE.parse_agent_sections(text=actual) + preserve = SYNC_MODULE.preserved_agent_sections + self.assertEqual( + preserve(text=old_body, state={}), + preserve(text=actual, state={}), + ) + + heading_target = root / "missing_one" + body_path = heading_target / "PR_BODY.md" + heading_body = body_path.read_text(encoding="utf-8") + self.assertNotIn(last_heading, heading_body) + started = run_harness( + command="start-pass", target=heading_target, + upstream=upstream, upstream_sha=sha, mode="PASS_1", + ) + self.assertEqual(started.returncode, 0, msg=started.stdout) + diffs = heading_target / ".coding_workflow/diffs" + schema = section_text( + path=diffs / "pr_body_skeleton.md", + section_name="repo_facts_map", + ) + repo_facts = section_text( + path=body_path, section_name="repo_facts_map", + ) + addition = schema[schema.index(last_heading):] + replace_section( + path=body_path, section_name="repo_facts_map", + text=f"{repo_facts.rstrip()}\n\n{addition}", + ) + complete_pass_body(path=body_path, mode="PASS_1") + append_text( + path=heading_target / "architecture.md", + text="\nproject architecture evidence\n", + ) + finished = finish_one( + target=heading_target, upstream=upstream, sha=sha, + ) + self.assertEqual(finished.returncode, 0, msg=finished.stdout) + repo_start = "" + repo_end = "" + full_start = "" + full_end = "" + missing = valid.replace(blocks["pr_test_evidence"], "", 1) + marker = SYNC_MODULE.SYNC_PR_BODY_MARKER + auto_end = SYNC_MODULE.SYNC_AUTO_END + damage = { + "non_sentinel": "legacy PR body\n", + "marker_duplicate": valid.replace(marker, marker * 2, 1), + "auto_partial": valid.replace(auto_end, "", 1), + "only_start": valid.replace(repo_end, "", 1), + "only_end": valid.replace(repo_start, "", 1), + "missing_duplicate": missing.replace( + blocks["full_document_reconcile"], + blocks["full_document_reconcile"] * 2, 1, + ), + "missing_partial": missing.replace(full_end, "", 1), + "overlap": valid.replace(repo_end, "", 1).replace( + full_start, f"{full_start}\n{repo_end}", 1, + ), + "outside": f"outside\n{valid}", + } + for scenario, damaged_body in damage.items(): + with self.subTest(damage=scenario): + target = root / f"damage_{scenario}" + initialize_core_target(target=target, upstream=upstream) + body_path = target / "PR_BODY.md" + body_path.write_text(damaged_body, encoding="utf-8") + result = run_harness( + command="prepare", target=target, + upstream=upstream, upstream_sha=sha, + ) + self.assertNotEqual(result.returncode, 0) + self.assertEqual( + parse_json(result=result)["error"], "已有 PR_BODY.md 无效", + ) + self.assertEqual( + body_path.read_text(encoding="utf-8"), damaged_body, + ) + run_path = target / HARNESS_MODULE.RUN_PATH + self.assertFalse(run_path.exists()) def test_complete_mode_sequence_handoff(self) -> None: """PREPARE 至 SUBMIT 起点必须允许合法累计 dirty handoff。""" @@ -824,7 +1170,11 @@ def test_complete_mode_sequence_handoff(self) -> None: payload = parse_json(result=submit) self.assertEqual( payload["allowed_commit_paths"], - ["AGENTS.md", "TESTING.md", "architecture.md", "interact.md"], + [ + "AGENTS.md", "PR_Checklist.md", "SOP.md", "TESTING.md", + "architecture.md", "capability_contract.json", + "docs/business_user_guide.md", "interact.md", + ], ) def test_start_pass_rejects_inter_mode_edit(self) -> None: diff --git a/zh/scripts/OPERATIONS.md b/zh/scripts/OPERATIONS.md index 8b4d595..a1db771 100644 --- a/zh/scripts/OPERATIONS.md +++ b/zh/scripts/OPERATIONS.md @@ -62,14 +62,26 @@ Studio / 自动编排直接从 clean pinned upstream 加载 canonical skill， Skill 模式由调用方运行单一 `harness.py`： -- `prepare`：校验仓库和 clean pinned upstream，运行普通 sync 并初始化 run。 +- `prepare`：校验仓库和 clean pinned upstream，运行普通 sync；若根 + `PR_BODY.md` 缺失，则从本轮 skeleton 机械复制正式 body，再初始化 run。已有合法 + sentinel body 保留；旧草稿或损坏 sentinel body 会失败且不会被覆盖。 - `start-pass`：校验 mode 间没有新编辑，保存本 PASS baseline，输出 prompt 位置。 -- `finish-pass`：检查 owned 文件和 PR body section，运行 pinned 普通 sync并推进 mode。 +- `finish-pass`：先检查 owned 路径、PR body 授权范围、当前 PASS 既有责任行和文档 + readiness，再运行 pinned 普通 sync 并推进 mode。 - `prepare-submit`：建立 active、unsealed SUBMIT baseline，不运行 final gate。 - `seal-submit`：限制本阶段只能填写提交测试证据，运行 pinned final gate并封存内容。 - `finish-submit`：把 sealed snapshot/body/path set 与 commit、工作区和远端 PR 绑定。 - `status`：输出 completed modes、active mode 和结果路径。 +PREPARE 成功后，根 `PR_BODY.md` 始终是正式交接产物； +`.coding_workflow/diffs/pr_body_skeleton.md` 只是它的机械来源，不能替代正式 body。 +PASS Agent 只能填写 ownership 声明的 section，并且只能更新匹配当前 owned path 和 +当前 PASS 的既有表格行。不得用一段摘要替换整个表格，也不得删除、改写其他 PASS 行。 + +机械 PASS completion 只证明：1. 强制产物存在；2. sentinel / heading / row 结构 +存在；3. 当前责任字段非空；4. owned 文档达到 readiness。内容真实性与证据质量仍由 +独立 reviewer 根据仓库事实判断。 + 该 harness 防善意执行者漏步骤和顺手越权，不试图抵御与它拥有相同 shell、Git 和 工作区写权限的主动恶意执行者。 @@ -97,6 +109,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 共用执行规则： - 完整 PR body 结构、sync sentinel、Repo Facts heading 和表头，以 `.coding_workflow/diffs/pr_body_skeleton.md` 或当前 `PR_BODY.md` 为准。 +- 表格 cell 中的字面 `|` 必须写为 `\|` 或改用 `
`；只有未转义 `|` 才作为列分隔符。 - 不得手改 `` 到 `` 区域、 任何 sync sentinel、sentinel 外内容。 - 只修改本 pass 允许的文件和 agent-owned section 内容；本 pass 负责的 @@ -123,8 +136,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 必须读取： 1. `.coding_workflow/diffs/agent_workorder.md` 2. `.coding_workflow/diffs/upstream_full/architecture.md` -3. `PR_BODY.md`；如果不存在，先用 `.coding_workflow/diffs/pr_body_skeleton.md` - 初始化；如果 skeleton 缺失，停止并回报普通 sync 未正确生成 +3. `PR_BODY.md` 应由 PREPARE 机械创建；如缺失立即停止并回报 PREPARE 产物异常，不得自行复制 skeleton。 4. `PR_BODY.md` 的 `repo_facts_map` 和 `full_document_reconcile` 中 `architecture.md` 行 5. 当前仓库的代码入口、模块边界、数据流、状态模型、外部依赖和代码层不变量 @@ -169,6 +181,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 共用执行规则： - 完整 PR body 结构、sync sentinel、Repo Facts heading 和表头，以 `.coding_workflow/diffs/pr_body_skeleton.md` 或当前 `PR_BODY.md` 为准。 +- 表格 cell 中的字面 `|` 必须写为 `\|` 或改用 `
`；只有未转义 `|` 才作为列分隔符。 - 不得手改 `` 到 `` 区域、 任何 sync sentinel、sentinel 外内容。 - 只修改本 pass 允许的文件和 agent-owned section 内容；本 pass 负责的 @@ -206,8 +219,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 2. `.coding_workflow/diffs/upstream_full/capability_contract.json` 3. `.coding_workflow/diffs/upstream_full/interact.md` 4. `.coding_workflow/diffs/upstream_full/docs/business_user_guide.md` -5. `PR_BODY.md`；如果不存在，先用 `.coding_workflow/diffs/pr_body_skeleton.md` - 初始化；如果 skeleton 缺失，停止并回报普通 sync 未正确生成 +5. `PR_BODY.md` 应由 PREPARE 机械创建；如缺失立即停止并回报 PREPARE 产物异常，不得自行复制 skeleton。 6. `PR_BODY.md` 的 `repo_facts_map` 中能力相关事实和 `full_document_reconcile` 中本 pass 三个 owned docs 行 7. `architecture.md`、`capability_contract.json`、`interact.md`、 @@ -261,6 +273,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 共用执行规则： - 完整 PR body 结构、sync sentinel、Repo Facts heading 和表头，以 `.coding_workflow/diffs/pr_body_skeleton.md` 或当前 `PR_BODY.md` 为准。 +- 表格 cell 中的字面 `|` 必须写为 `\|` 或改用 `
`；只有未转义 `|` 才作为列分隔符。 - 不得手改 `` 到 `` 区域、 任何 sync sentinel、sentinel 外内容。 - 只修改本 pass 允许的文件和 agent-owned section 内容；本 pass 负责的 @@ -286,8 +299,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 必须读取： 1. `.coding_workflow/diffs/agent_workorder.md` 2. `.coding_workflow/diffs/upstream_full/TESTING.md` -3. `PR_BODY.md`；如果不存在，先用 `.coding_workflow/diffs/pr_body_skeleton.md` - 初始化；如果 skeleton 缺失，停止并回报普通 sync 未正确生成 +3. `PR_BODY.md` 应由 PREPARE 机械创建；如缺失立即停止并回报 PREPARE 产物异常，不得自行复制 skeleton。 4. `PR_BODY.md` 的 `repo_facts_map` 中测试现状相关事实和 `full_document_reconcile` 中 `TESTING.md` 行 5. `TESTING.md`、目标项目测试入口、`tests/` 清单和机械信号命令输出；只打开机械信号 @@ -342,6 +354,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 共用执行规则： - 完整 PR body 结构、sync sentinel、Repo Facts heading 和表头，以 `.coding_workflow/diffs/pr_body_skeleton.md` 或当前 `PR_BODY.md` 为准。 +- 表格 cell 中的字面 `|` 必须写为 `\|` 或改用 `
`；只有未转义 `|` 才作为列分隔符。 - 不得手改 `` 到 `` 区域、 任何 sync sentinel、sentinel 外内容。 - 只修改本 pass 允许的文件和 agent-owned section 内容；本 pass 负责的 @@ -384,8 +397,7 @@ owned docs，并把结论写入 `PR_BODY.md` 的 agent-owned 区。 3. `.coding_workflow/diffs/upstream_full/SOP.md` 4. `.coding_workflow/diffs/upstream_full/AGENTS.md` 5. `.coding_workflow/diffs/upstream_full/.github/pull_request_template.md` -6. `PR_BODY.md`；如果不存在，先用 `.coding_workflow/diffs/pr_body_skeleton.md` - 初始化；如果 skeleton 缺失，停止并回报普通 sync 未正确生成 +6. `PR_BODY.md` 应由 PREPARE 机械创建；如缺失立即停止并回报 PREPARE 产物异常，不得自行复制 skeleton。 7. `PR_BODY.md` 的 `full_document_reconcile` 全表和 `remaining_human_decisions` 8. `PR_Checklist.md`、`SOP.md`、`AGENTS.md` 和 `.github/pull_request_template.md` 9. PASS 1/2/3 owned docs 仅在验证 downstream impact 闭合时定向读取相关段落； diff --git a/zh/skills/workflow-docs-sync/evals/README.md b/zh/skills/workflow-docs-sync/evals/README.md index ec0d3e4..0207f0b 100644 --- a/zh/skills/workflow-docs-sync/evals/README.md +++ b/zh/skills/workflow-docs-sync/evals/README.md @@ -32,6 +32,12 @@ 选择已有可运行代码、但缺少一个或多个核心 workflow 文档的仓库。验证两条轨道能否用 代码证据补齐缺失文档，同时避免把模板占位、无证据能力或无关治理文件写入结果。 +Shadow Case A 的 Skill 轨道还必须证明：PREPARE 后根 `PR_BODY.md` 已作为正式交接 +产物存在；每个 PASS 的成功结果同时对应正式 body 中当前 PASS 的 sentinel、heading、 +表格宽度、非空责任 cell 和 owned 文档 readiness。删除 `待补充` 但留下空 block / cell +必须失败；表格字面 `|` 必须转义为 `\|` 或改用 `
`。这些机械条件不裁定证据真实性， +独立 reviewer 仍负责语义 verdict。仅有 skeleton 不算正式交接或 PASS success。 + ## Case B：存在 class-2 上游漂移 选择核心文档已项目化、但未吸收当前 upstream 语义更新的仓库。验证两条轨道是否都能 diff --git a/zh/skills/workflow-docs-sync/scripts/harness.py b/zh/skills/workflow-docs-sync/scripts/harness.py index f6ee886..763c431 100644 --- a/zh/skills/workflow-docs-sync/scripts/harness.py +++ b/zh/skills/workflow-docs-sync/scripts/harness.py @@ -13,6 +13,7 @@ import json import os import re +import shutil import subprocess import sys from datetime import datetime, timezone @@ -33,6 +34,12 @@ SKILL_ROOT = Path(__file__).resolve().parents[1] SOURCE_METADATA_PATH = SKILL_ROOT / ".source.json" WORKFLOW_EXTRA_PATHS = {".gitignore", "PR_BODY.md"} +PERMITTED_INHERIT_PATHS = {".github/pull_request_template.md"} +SYNC_PR_BODY_MARKER = "" +SYNC_AUTO_START = "" +SYNC_AUTO_END = "" +UNFILLED_PLACEHOLDER = "待补充" +LITERAL_PIPE_GUIDANCE = "表格 cell 中的字面 | 必须写为 \\| 或改用
" AGENT_SECTIONS = ( "repo_facts_map", "full_document_reconcile", @@ -663,32 +670,100 @@ def require_boundary_unchanged( ) -def parse_agent_sections(*, text: str) -> tuple[str, dict[str, str]]: - """参数为 PR body；预期返回占位骨架和 agent section 内容。""" - cursor = 0 - skeleton_parts: list[str] = [] - sections: dict[str, str] = {} - for section_name in AGENT_SECTIONS: - start_marker = f"" - end_marker = f"" - start_index = text.find(start_marker, cursor) - if start_index < 0: +def validated_sentinel_spans( + *, text: str, require_all: bool, canonical_order: bool +) -> dict[str, tuple[int, int]]: + """参数为 body 与 schema 强度；预期返回合法且不重叠的 sentinel spans。""" + count = text.count(SYNC_PR_BODY_MARKER) + if count != 1: + raise HarnessError( + error="PR_BODY sync sentinel 无效", + detail=f"{SYNC_PR_BODY_MARKER} 匹配次数：{count}", + ) + marker_start = text.index(SYNC_PR_BODY_MARKER) + spans = { + "marker": (marker_start, marker_start + len(SYNC_PR_BODY_MARKER)) + } + sentinels = [("auto", SYNC_AUTO_START, SYNC_AUTO_END, True)] + sentinels.extend( + ( + name, + f"", + f"", + require_all, + ) + for name in AGENT_SECTIONS + ) + for name, start, end, required in sentinels: + start_count, end_count = text.count(start), text.count(end) + if start_count == 0 and end_count == 0 and not required: + continue + if start_count != 1 or end_count != 1: + raise HarnessError( + error="PR_BODY sentinel 无效", + detail=f"{name}: start={start_count}, end={end_count}", + ) + start_index, end_index = text.index(start), text.index(end) + if end_index < start_index: raise HarnessError( - error="PR_BODY agent section 缺失", - detail=section_name, + error="PR_BODY sentinel 未闭合", + detail=name, ) - body_start = start_index + len(start_marker) - end_index = text.find(end_marker, body_start) - if end_index < 0: + spans[name] = (start_index, end_index + len(end)) + + # 旧 schema 可改变 section 顺序，但任何 span 都不能覆盖另一 span。 + ordered = sorted(spans.items(), key=lambda item: item[1]) + cursor = 0 + outside_parts: list[str] = [] + for name, (start_index, end_index) in ordered: + if start_index < cursor: raise HarnessError( - error="PR_BODY agent section 未闭合", - detail=section_name, + error="PR_BODY sentinel 重叠", + detail=name, ) - skeleton_parts.append(text[cursor:body_start]) - skeleton_parts.append(f"") - skeleton_parts.append(end_marker) - sections[section_name] = text[body_start:end_index] - cursor = end_index + len(end_marker) + outside_parts.append(text[cursor:start_index]) + cursor = end_index + outside_parts.append(text[cursor:]) + if "".join(outside_parts).strip(): + raise HarnessError( + error="PR_BODY sentinel 外存在内容", + detail="请把人工内容迁移到对应 agent-owned section", + ) + expected = ["marker", "repo_facts_map", "auto", *AGENT_SECTIONS[1:]] + if canonical_order and [name for name, _ in ordered] != expected: + raise HarnessError( + error="PR_BODY sentinel 顺序非法", + detail="必须使用当前 skeleton canonical 顺序", + ) + return spans + + +def validate_refreshable_pr_body(*, text: str) -> None: + """参数为 sync 前 body；预期允许完整缺失的旧 agent section。""" + validated_sentinel_spans( + text=text, require_all=False, canonical_order=False, + ) + + +def parse_agent_sections(*, text: str) -> tuple[str, dict[str, str]]: + """参数为 current body；预期严格校验并返回骨架和 section。""" + spans = validated_sentinel_spans( + text=text, require_all=True, canonical_order=True, + ) + cursor = 0 + skeleton_parts: list[str] = [] + sections: dict[str, str] = {} + for name in AGENT_SECTIONS: + start = f"" + end = f"" + start_index, span_end = spans[name] + body_start = start_index + len(start) + body_end = span_end - len(end) + skeleton_parts.extend(( + text[cursor:body_start], f"", end, + )) + sections[name] = text[body_start:body_end] + cursor = span_end skeleton_parts.append(text[cursor:]) return "".join(skeleton_parts), sections @@ -698,8 +773,8 @@ def assert_pr_body_scope( baseline_text: str | None, current_text: str | None, allowed_sections: set[str], -) -> None: - """参数为起止 body 和允许 section；预期其他区域完全不变。""" +) -> tuple[dict[str, str], dict[str, str]]: + """参数为起止 body 和允许 section；预期返回已校验的 section。""" if baseline_text is None or current_text is None: raise HarnessError( error="PR_BODY scope 无法校验", @@ -728,6 +803,261 @@ def assert_pr_body_scope( detail=", ".join(unauthorized), extra={"unauthorized_sections": unauthorized}, ) + return baseline_sections, current_sections + + +def ensure_prepared_pr_body(*, repo_root: Path, refreshable: bool) -> None: + """参数为目标仓库；预期保留合法 body 或从 sync skeleton 复制。""" + body_path = repo_root / "PR_BODY.md" + skeleton_path = repo_root / ".coding_workflow/diffs/pr_body_skeleton.md" + if body_path.is_file(): + try: + validator = ( + validate_refreshable_pr_body if refreshable + else parse_agent_sections + ) + validator(text=body_path.read_text(encoding="utf-8")) + except HarnessError as exc: + raise HarnessError( + error="已有 PR_BODY.md 无效", + detail=( + f"{exc.error}: {exc.detail}；请移开现有文件，或人工迁移" + "旧内容到完整 sentinel body；原文件未覆盖" + ), + ) from exc + return + if body_path.exists(): + raise HarnessError( + error="已有 PR_BODY.md 无效", + detail="路径不是普通文件；请移开后重试，原路径未覆盖", + ) + if not skeleton_path.is_file(): + raise HarnessError( + error="PREPARE PR body skeleton 缺失", + detail=str(skeleton_path), + ) + skeleton_text = skeleton_path.read_text(encoding="utf-8") + parse_agent_sections(text=skeleton_text) + shutil.copy2(src=skeleton_path, dst=body_path) + + +def matching_table_rows( + *, text: str, key: str, prefer_code: bool +) -> list[str]: + """参数为 section、稳定行键和代码格式偏好；预期返回候选表格行。""" + table_lines = [line for line in text.splitlines() if line.startswith("|")] + if prefer_code: + code_rows = [line for line in table_lines if f"| `{key}` |" in line] + if code_rows: + return code_rows + return [line for line in table_lines if f"| {key} |" in line] + + +def split_table_row(*, row: str) -> list[str]: + """参数为 Markdown 表格行；预期仅按未转义竖线拆分 cells。""" + if not row.startswith("|") or not row.endswith("|") or row.endswith("\\|"): + return [] + cells: list[str] = [] + cell_start = 1 + for index, character in enumerate(row[1:-1], start=1): + if character == "|" and row[index - 1] != "\\": + cells.append(row[cell_start:index]) + cell_start = index + 1 + cells.append(row[cell_start:-1]) + return cells + + +def fail_completion(*, detail: str) -> None: + """参数为机械 completion 失败证据；预期抛出稳定 harness 错误。""" + raise HarnessError(error="PASS PR_BODY 未完成", detail=detail) + + +def require_current_table( + *, schema_text: str, current_text: str, heading: str, label: str +) -> int: + """参数为 pinned/current section；预期 current 表头保持 canonical。""" + schema_lines = schema_text.splitlines() + current_lines = current_text.splitlines() + schema_rows = [line for line in schema_lines if line.startswith("|")] + if schema_lines.count(heading) != 1 or not schema_rows: + fail_completion(detail=f"pinned {label} heading 或表头缺失") + header = schema_rows[0] + if current_lines.count(heading) != 1 or current_lines.count(header) != 1: + fail_completion(detail=f"{label} 表头不符；{LITERAL_PIPE_GUIDANCE}") + return len(split_table_row(row=header)) + + +def require_completed_row( + *, section_text: str, key: str, label: str, prefer_code: bool, + expected_cells: int, owned_cells: int +) -> None: + """参数为稳定行键与 schema 宽度；预期责任 cells 非空且已填写。""" + matches = matching_table_rows( + text=section_text, + key=key, + prefer_code=prefer_code, + ) + if len(matches) != 1: + fail_completion( + detail=f"{label} 匹配行数必须为 1，实际 {len(matches)}", + ) + cells = split_table_row(row=matches[0]) + if len(cells) != expected_cells: + fail_completion( + detail=( + f"{label} cell 数应为 {expected_cells}，实际 {len(cells)}；" + f"{LITERAL_PIPE_GUIDANCE}" + ), + ) + if any(not cell.strip() for cell in cells[-owned_cells:]): + fail_completion(detail=f"{label} 责任 cell 不得为空") + if UNFILLED_PLACEHOLDER in matches[0]: + fail_completion(detail=f"{label} 仍含 {UNFILLED_PLACEHOLDER}") + + +def assert_repo_facts_complete( + *, schema_text: str, current_text: str +) -> None: + """参数为 current skeleton 与 body；预期固定事实 block 证据非空。""" + headings = [ + line for line in schema_text.splitlines() if line.startswith("### ") + ] + if len(headings) != 10 or len(set(headings)) != 10: + fail_completion( + detail=f"current Repo Facts heading 数应为 10，实际 {len(headings)}", + ) + current_lines = current_text.splitlines() + if current_lines.count("## Repo Facts Map") != 1: + fail_completion(detail="Repo Facts Map heading 匹配次数必须为 1") + for heading in headings: + matches = [ + index + for index, line in enumerate(current_lines) + if line == heading + ] + if len(matches) != 1: + fail_completion( + detail=f"{heading} 匹配次数必须为 1，实际 {len(matches)}", + ) + block_end = next(( + index for index in range(matches[0] + 1, len(current_lines)) + if current_lines[index].startswith("### ") + ), len(current_lines)) + block = current_lines[matches[0] + 1:block_end] + evidence = [line for line in block if line.startswith("证据:")] + if len(evidence) != 1: + fail_completion( + detail=f"{heading} 的 证据: 行必须恰好一行", + ) + if not evidence[0].partition(":")[2].strip(): + fail_completion(detail=f"{heading} 的 证据: 值不得为空") + if UNFILLED_PLACEHOLDER in "\n".join(block): + fail_completion(detail=f"{heading} block 仍含 待补充") + + +def assert_remaining_decisions_complete(*, section_text: str) -> None: + """参数为 Remaining Human Decisions；预期 heading 后有非空 payload。""" + heading = "## Remaining Human Decisions" + lines = [line.strip() for line in section_text.splitlines()] + payload = [line for line in lines if line and line != heading] + if lines.count(heading) != 1 or not payload: + fail_completion(detail="Remaining Human Decisions 缺少非空 payload") + if UNFILLED_PLACEHOLDER in "\n".join(payload): + fail_completion(detail="Remaining Human Decisions 仍含 待补充") + + +def assert_pass_body_completion( + *, + mode: str, + ownership_entry: dict[str, Any], + schema_sections: dict[str, str], + current_sections: dict[str, str], +) -> None: + """参数为 PASS ownership 与 body sections；预期当前责任项机械完成。""" + if mode == "PASS_1": + assert_repo_facts_complete( + schema_text=schema_sections["repo_facts_map"], + current_text=current_sections["repo_facts_map"], + ) + reconcile_width = require_current_table( + schema_text=schema_sections["full_document_reconcile"], + current_text=current_sections["full_document_reconcile"], + heading="## Full Document Reconcile", + label="Full Document Reconcile", + ) + for relative_path in ownership_entry["owned"]: + require_completed_row( + section_text=current_sections["full_document_reconcile"], + key=relative_path, + label=f"{mode} reconcile `{relative_path}`", + prefer_code=True, + expected_cells=reconcile_width, + owned_cells=5, + ) + execution_title = re.sub( + pattern=r"^\d+\.\d+\s+", + repl="", + string=ownership_entry["title"], + ) + execution_width = require_current_table( + schema_text=schema_sections["agent_execution_evidence"], + current_text=current_sections["agent_execution_evidence"], + heading="## Agent Execution Evidence", + label="Agent Execution Evidence", + ) + require_completed_row( + section_text=current_sections["agent_execution_evidence"], + key=execution_title, + label=f"{execution_title} execution evidence", + prefer_code=False, + expected_cells=execution_width, + owned_cells=3, + ) + if mode == "PASS_4": + assert_remaining_decisions_complete( + section_text=current_sections["remaining_human_decisions"], + ) + + +def assert_owned_docs_ready( + *, + mode: str, + target_repo: Path, + upstream_dir: Path, + ownership_entry: dict[str, Any], +) -> None: + """参数为 PASS 与 pinned upstream；预期 owned 文档存在且已项目化。""" + for relative_path in ownership_entry["owned"]: + target_path = target_repo / relative_path + if not target_path.is_file(): + raise HarnessError( + error="PASS owned 文档缺失", + detail=relative_path, + extra={"owned_path": relative_path}, + ) + if relative_path in PERMITTED_INHERIT_PATHS: + continue + upstream_path = upstream_dir / "zh" / relative_path + if not upstream_path.is_file(): + raise HarnessError( + error="pinned owned 模板缺失", + detail=relative_path, + extra={"owned_path": relative_path}, + ) + current_text = target_path.read_text(encoding="utf-8") + upstream_text = upstream_path.read_text(encoding="utf-8") + normalized_current = current_text.replace("\r\n", "\n").replace( + "\r", "\n" + ) + normalized_upstream = upstream_text.replace("\r\n", "\n").replace( + "\r", "\n" + ) + if normalized_current == normalized_upstream: + raise HarnessError( + error="PASS owned 文档未项目化", + detail=f"{mode}: {relative_path} 仍与 pinned zh 模板完全相同", + extra={"owned_path": relative_path}, + ) def git_paths_between( @@ -853,6 +1183,10 @@ def prepare(*, args: Any) -> dict[str, Any]: detail="先完成当前 run，或显式删除 ignored runtime 后重启", ) + # sync 允许旧 schema migration；先只读拒绝损坏 body，避免它被自动补写。 + if (target_repo / "PR_BODY.md").exists(): + ensure_prepared_pr_body(repo_root=target_repo, refreshable=True) + # PREPARE 之前不接受混入的业务或核心文档改动。 before_dirty = assert_dirty_allowed( repo_root=target_repo, @@ -869,6 +1203,9 @@ def prepare(*, args: Any) -> dict[str, Any]: final=False, ) + # 正式 handoff body 必须在状态快照前落盘，后续 PASS 不再依赖 skeleton fallback。 + ensure_prepared_pr_body(repo_root=target_repo, refreshable=False) + # pinned sync 只能产生 workflow 管理路径。 after_dirty = assert_dirty_allowed( repo_root=target_repo, @@ -1055,21 +1392,34 @@ def finish_pass(*, args: Any) -> dict[str, Any]: detail=", ".join(unauthorized), extra={"unauthorized_paths": unauthorized}, ) - if "PR_BODY.md" in changed: - assert_pr_body_scope( - baseline_text=( - baseline["pr_body_text"] - or ( - target_repo / ".coding_workflow/diffs/pr_body_skeleton.md" - ).read_text(encoding="utf-8") - ), - current_text=read_pr_body(repo_root=target_repo), - allowed_sections=set( - ownership[args.mode]["pr_body_sections"] - ), + _, current_sections = assert_pr_body_scope( + baseline_text=baseline["pr_body_text"], + current_text=read_pr_body(repo_root=target_repo), + allowed_sections=set(ownership[args.mode]["pr_body_sections"]), + ) + schema_path = target_repo / ".coding_workflow/diffs/pr_body_skeleton.md" + if not schema_path.is_file(): + raise HarnessError( + error="current PR body skeleton 缺失", + detail=str(schema_path), ) + _, schema_sections = parse_agent_sections( + text=schema_path.read_text(encoding="utf-8"), + ) + assert_pass_body_completion( + mode=args.mode, + ownership_entry=ownership[args.mode], + schema_sections=schema_sections, + current_sections=current_sections, + ) + assert_owned_docs_ready( + mode=args.mode, + target_repo=target_repo, + upstream_dir=upstream_dir, + ownership_entry=ownership[args.mode], + ) - # ownership 通过后才运行 pinned sync，防止 sync 掩盖 Agent 越权。 + # ownership 与 completion 全通过后才 sync，失败不会污染 auto 区或推进状态。 sync = run_pinned_sync( target_repo=target_repo, upstream_dir=upstream_dir,