From 822661ec81268304b54a69197494d2bad53f208f Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Sun, 9 Aug 2026 03:22:04 +0800 Subject: [PATCH] fix(governance): bind acceptance gates to clean commits Add the lightweight high-risk closeout helper and its regression coverage. Reject tracked, staged, or untracked public changes before and after full local validation so accepted results map to a reproducible commit. --- scripts/high_risk_acceptance.py | 258 +++++++++++++++++++ scripts/pre_push_checks.sh | 19 ++ scripts/run_full_tests_linux.sh | 19 ++ scripts/run_full_tests_windows.ps1 | 15 ++ tests/architecture_dependency_policy.json | 1 + tests/test_architecture_dependency_policy.py | 2 +- tests/test_closeout_worktree_cleanliness.py | 55 ++++ tests/test_high_risk_acceptance.py | 258 +++++++++++++++++++ 8 files changed, 626 insertions(+), 1 deletion(-) create mode 100644 scripts/high_risk_acceptance.py create mode 100644 tests/test_closeout_worktree_cleanliness.py create mode 100644 tests/test_high_risk_acceptance.py diff --git a/scripts/high_risk_acceptance.py b/scripts/high_risk_acceptance.py new file mode 100644 index 0000000..4ea738b --- /dev/null +++ b/scripts/high_risk_acceptance.py @@ -0,0 +1,258 @@ +"""Lightweight closeout evidence for changes classified as high risk. + +This pilot deliberately reuses the adversarial gate's path classifier. Standard- +risk and empty diffs remain outside this workflow and produce no receipt. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +from collections.abc import Sequence +from datetime import datetime, timezone +from pathlib import Path + +from run_adversarial_gate import ( + DEFAULT_HIGH_RISK_PATTERNS, + _filter_high_risk_files, + _run_git_diff, +) + +SCHEMA = "openclaw-high-risk-receipt/1" +EXACT_COMMIT_RE = re.compile(r"[0-9a-fA-F]{40}\Z") +ITEM_RE = re.compile(r"[A-Z][A-Z0-9-]{0,31}\Z") + + +class AcceptanceError(RuntimeError): + """A safe, user-actionable closeout validation failure.""" + + +def _run_git(repo_root: Path, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["git", *args], + cwd=repo_root, + capture_output=True, + text=True, + check=False, + ) + + +def _git_root() -> Path: + result = subprocess.run( + ["git", "rev-parse", "--show-toplevel"], + capture_output=True, + text=True, + check=False, + ) + if result.returncode != 0 or not result.stdout.strip(): + raise AcceptanceError("current directory is not inside a Git worktree") + return Path(result.stdout.strip()).resolve() + + +def _resolve_commit(repo_root: Path, reference: str, label: str) -> str: + if not reference or reference.startswith("-"): + raise AcceptanceError(f"{label} must be a valid Git revision") + result = _run_git(repo_root, "rev-parse", "--verify", f"{reference}^{{commit}}") + commit = result.stdout.strip().lower() + if result.returncode != 0 or not EXACT_COMMIT_RE.fullmatch(commit): + raise AcceptanceError(f"{label} does not resolve to a commit") + return commit + + +def _require_ancestor(repo_root: Path, base_commit: str, candidate_commit: str) -> None: + result = _run_git( + repo_root, + "merge-base", + "--is-ancestor", + base_commit, + candidate_commit, + ) + if result.returncode != 0: + raise AcceptanceError("base commit is not an ancestor of candidate commit") + + +def _changed_files( + repo_root: Path, base_commit: str, candidate_commit: str +) -> list[str]: + previous_cwd = Path.cwd() + try: + os.chdir(repo_root) + return [str(path) for path in _run_git_diff(base_commit, candidate_commit)] + finally: + os.chdir(previous_cwd) + + +def _require_closeout_state( + repo_root: Path, + candidate_argument: str, + candidate_commit: str, +) -> str: + if not EXACT_COMMIT_RE.fullmatch(candidate_argument): + raise AcceptanceError( + "high-risk candidate must be an exact 40-character commit SHA" + ) + + head_commit = _resolve_commit(repo_root, "HEAD", "HEAD") + if candidate_commit != head_commit: + raise AcceptanceError("candidate commit must equal current HEAD") + + branch_result = _run_git(repo_root, "branch", "--show-current") + branch = branch_result.stdout.strip() + if branch_result.returncode != 0 or branch != "dev": + raise AcceptanceError("high-risk closeout must run on branch dev") + + status_result = _run_git( + repo_root, + "status", + "--porcelain", + "--untracked-files=no", + ) + if status_result.returncode != 0 or status_result.stdout.strip(): + raise AcceptanceError("tracked worktree and index must be clean") + return branch + + +def _require_identity(value: str | None, label: str) -> str: + if value is None or not value.strip(): + raise AcceptanceError(f"{label} is required for high-risk closeout") + normalized = value.strip() + if len(normalized) > 128 or any(ord(character) < 32 for character in normalized): + raise AcceptanceError(f"{label} contains invalid characters") + return normalized + + +def _validate_closeout_arguments(args: argparse.Namespace) -> tuple[str, str, str]: + if args.item is None or not ITEM_RE.fullmatch(args.item): + raise AcceptanceError("item must be an uppercase roadmap identifier") + + implementer = _require_identity(args.implementer, "implementer") + reviewer = _require_identity(args.reviewer, "reviewer") + if implementer.casefold() == reviewer.casefold(): + raise AcceptanceError("reviewer must be distinct from implementer") + if args.review_verdict != "APPROVED": + raise AcceptanceError("review verdict must be APPROVED") + if args.full_gate_status != "PASS": + raise AcceptanceError("full TEST_SOP gate must be PASS") + return args.item, implementer, reviewer + + +def _resolve_output(repo_root: Path, output: str | None) -> tuple[Path, str]: + if output is None or not output.strip(): + raise AcceptanceError("output is required for high-risk closeout") + + candidate = Path(output) + output_path = ( + (repo_root / candidate).resolve() + if not candidate.is_absolute() + else candidate.resolve() + ) + planning_root = (repo_root / ".planning").resolve() + try: + relative_to_planning = output_path.relative_to(planning_root) + relative_to_repo = output_path.relative_to(repo_root) + except ValueError as exc: + raise AcceptanceError( + "output must be under the repository .planning directory" + ) from exc + if relative_to_planning == Path("."): + raise AcceptanceError("output must name a file under .planning") + if output_path.exists(): + raise AcceptanceError("output already exists") + + relative_posix = relative_to_repo.as_posix() + ignored = _run_git( + repo_root, + "check-ignore", + "-v", + "--no-index", + "--", + relative_posix, + ) + if ignored.returncode != 0: + raise AcceptanceError("output must be ignored by repository Git rules") + return output_path, relative_posix + + +def _write_receipt(output_path: Path, receipt: dict[str, object]) -> None: + output_path.parent.mkdir(parents=True, exist_ok=True) + try: + with output_path.open("x", encoding="utf-8", newline="\n") as handle: + json.dump(receipt, handle, indent=2, sort_keys=True) + handle.write("\n") + except FileExistsError as exc: + raise AcceptanceError("output already exists") from exc + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="Validate and record the lightweight high-risk closeout pilot." + ) + parser.add_argument("--base", required=True, help="Base Git revision.") + parser.add_argument("--candidate", default="HEAD", help="Candidate Git revision.") + parser.add_argument("--item") + parser.add_argument("--implementer") + parser.add_argument("--reviewer") + parser.add_argument("--review-verdict") + parser.add_argument("--full-gate-status") + parser.add_argument("--output") + return parser + + +def run(argv: Sequence[str] | None = None) -> int: + args = _parser().parse_args(argv) + try: + repo_root = _git_root() + base_commit = _resolve_commit(repo_root, args.base, "base") + candidate_commit = _resolve_commit(repo_root, args.candidate, "candidate") + + # IMPORTANT: ancestry is validated before classification so unrelated + # histories cannot be mistaken for a standard-risk, non-applicable diff. + _require_ancestor(repo_root, base_commit, candidate_commit) + changed_files = _changed_files(repo_root, base_commit, candidate_commit) + high_risk_changed = _filter_high_risk_files( + changed_files, DEFAULT_HIGH_RISK_PATTERNS + ) + if not high_risk_changed: + print("HIGH_RISK_ACCEPTANCE: NOT_APPLICABLE") + return 0 + + branch = _require_closeout_state(repo_root, args.candidate, candidate_commit) + item, implementer, reviewer = _validate_closeout_arguments(args) + output_path, output_label = _resolve_output(repo_root, args.output) + receipt: dict[str, object] = { + "schema": SCHEMA, + "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"), + "item": item, + "branch": branch, + "base_commit": base_commit, + "candidate_commit": candidate_commit, + "changed_files": changed_files, + "high_risk_changed_files": high_risk_changed, + "review": { + "implementer": implementer, + "reviewer": reviewer, + "verdict": args.review_verdict, + }, + "gates": {"full_test_sop": args.full_gate_status}, + "limitations": ( + "Pilot receipt records declared review and gate results; it is not " + "identity authentication or a cryptographic attestation." + ), + } + _write_receipt(output_path, receipt) + print(f"HIGH_RISK_ACCEPTANCE: PASS ({output_label})") + return 0 + except AcceptanceError as exc: + print(f"HIGH_RISK_ACCEPTANCE: FAIL: {exc}") + return 1 + except (OSError, subprocess.SubprocessError): + print("HIGH_RISK_ACCEPTANCE: FAIL: repository validation could not complete") + return 1 + + +if __name__ == "__main__": + sys.exit(run()) diff --git a/scripts/pre_push_checks.sh b/scripts/pre_push_checks.sh index 5db07ab..6d6d4f9 100644 --- a/scripts/pre_push_checks.sh +++ b/scripts/pre_push_checks.sh @@ -80,6 +80,24 @@ report_precommit_repo_drift_and_exit() { exit 1 } +assert_clean_public_worktree() { + # CRITICAL: tracked diff snapshots do not include untracked deliverables. A push + # gate must validate one clean committed candidate, not local-only files. + local status + if ! status="$(git status --porcelain --untracked-files=all)"; then + echo "[pre-push] ERROR: unable to inspect Git worktree state" >&2 + exit 1 + fi + if [ -n "$status" ]; then + echo "[pre-push] ERROR: validation requires a clean committed candidate." >&2 + echo "[pre-push] Commit or remove public changes, then retry." >&2 + printf '%s\n' "$status" >&2 + exit 1 + fi +} + +assert_clean_public_worktree + is_wsl() { grep -qiE "(microsoft|wsl)" /proc/version 2>/dev/null } @@ -430,4 +448,5 @@ MOLTBOT_STATE_DIR="$ROOT_DIR/moltbot_state/_pre_push_adversarial" \ echo "[pre-push] 9/9 npm test (Playwright)" npm test +assert_clean_public_worktree echo "[pre-push] PASS" diff --git a/scripts/run_full_tests_linux.sh b/scripts/run_full_tests_linux.sh index 0c075cd..02c87c7 100644 --- a/scripts/run_full_tests_linux.sh +++ b/scripts/run_full_tests_linux.sh @@ -83,6 +83,24 @@ report_precommit_repo_drift_and_exit() { exit 1 } +assert_clean_public_worktree() { + # CRITICAL: tracked diff snapshots do not include untracked deliverables. A full + # acceptance gate must bind to one clean committed candidate, not local-only files. + local status + if ! status="$(git status --porcelain --untracked-files=all)"; then + echo "[tests] ERROR: unable to inspect Git worktree state" >&2 + exit 1 + fi + if [ -n "$status" ]; then + echo "[tests] ERROR: acceptance requires a clean committed candidate." >&2 + echo "[tests] Commit or remove public changes, then rerun." >&2 + printf '%s\n' "$status" >&2 + exit 1 + fi +} + +assert_clean_public_worktree + require_cmd node require_cmd npm @@ -265,4 +283,5 @@ echo "[tests] 10/10 frontend E2E" # not assume a warmed local browser cache when running on fresh WSL/Linux hosts. OPENCLAW_PLAYWRIGHT_INSTALL=1 OPENCLAW_PLAYWRIGHT_BROWSERS=chromium npm test +assert_clean_public_worktree echo "[tests] PASS" diff --git a/scripts/run_full_tests_windows.ps1 b/scripts/run_full_tests_windows.ps1 index 363eaee..b2a606b 100644 --- a/scripts/run_full_tests_windows.ps1 +++ b/scripts/run_full_tests_windows.ps1 @@ -54,6 +54,20 @@ function Assert-PreCommitDidNotMutateRepo { } } +function Assert-CleanPublicWorktree { + # CRITICAL: tracked diff snapshots do not include untracked deliverables. A full + # acceptance gate must bind to one clean committed candidate, not local-only files. + $statusLines = @(& git status --porcelain --untracked-files=all) + if ($LASTEXITCODE -ne 0) { + throw "[tests] ERROR: unable to inspect Git worktree state" + } + if ($statusLines.Count -gt 0) { + throw "[tests] ERROR: acceptance requires a clean committed candidate. Commit or remove public changes, then rerun.`n$($statusLines -join [Environment]::NewLine)" + } +} + +Assert-CleanPublicWorktree + Require-Cmd node Require-Cmd npm @@ -363,4 +377,5 @@ $env:OPENCLAW_PLAYWRIGHT_BROWSERS = "chromium" # not assume a warmed local browser cache when running on fresh Windows hosts. Invoke-Checked "frontend E2E" { npm test } +Assert-CleanPublicWorktree Write-Host "[tests] PASS" diff --git a/tests/architecture_dependency_policy.json b/tests/architecture_dependency_policy.json index 51bf342..33319a6 100644 --- a/tests/architecture_dependency_policy.json +++ b/tests/architecture_dependency_policy.json @@ -124,6 +124,7 @@ "scripts/devtools/verify_s30_doctor.py", "scripts/generate_openapi_spec.py", "scripts/generate_provenance.py", + "scripts/high_risk_acceptance.py", "scripts/lint_implementation_record.py", "scripts/openclaw_smoke_import.py", "scripts/operator_doctor.py", diff --git a/tests/test_architecture_dependency_policy.py b/tests/test_architecture_dependency_policy.py index 22ca86e..227aaea 100644 --- a/tests/test_architecture_dependency_policy.py +++ b/tests/test_architecture_dependency_policy.py @@ -375,7 +375,7 @@ class RepositoryArchitecturePolicyTests(unittest.TestCase): analysis = dependency_policy.analyze_repository(self.repo_root, policy) self.assertEqual(analysis.findings, ()) - self.assertEqual(len(analysis.owned_paths), 305) + self.assertEqual(len(analysis.owned_paths), 306) self.assertEqual(len(policy["accepted_cycles"]), 2) self.assertEqual(len(policy["dynamic_imports"]), 8) self.assertEqual(len(policy["compatibility_exceptions"]), 9) diff --git a/tests/test_closeout_worktree_cleanliness.py b/tests/test_closeout_worktree_cleanliness.py new file mode 100644 index 0000000..ee6b9bb --- /dev/null +++ b/tests/test_closeout_worktree_cleanliness.py @@ -0,0 +1,55 @@ +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +WINDOWS_FULL_GATE = ROOT / "scripts" / "run_full_tests_windows.ps1" +LINUX_FULL_GATE = ROOT / "scripts" / "run_full_tests_linux.sh" +PRE_PUSH_GATE = ROOT / "scripts" / "pre_push_checks.sh" + + +def _assert_ordered(test: unittest.TestCase, content: str, *needles: str) -> None: + positions = [content.index(needle) for needle in needles] + test.assertEqual(positions, sorted(positions)) + + +class CloseoutWorktreeCleanlinessContractTests(unittest.TestCase): + def test_windows_full_gate_checks_all_public_state_before_and_after_validation( + self, + ): + content = WINDOWS_FULL_GATE.read_text(encoding="utf-8") + + self.assertIn("function Assert-CleanPublicWorktree", content) + self.assertIn("& git status --porcelain --untracked-files=all", content) + self.assertEqual(content.splitlines().count("Assert-CleanPublicWorktree"), 2) + _assert_ordered( + self, + content, + "\nAssert-CleanPublicWorktree\n", + 'Invoke-Checked "npm ci" { npm ci }', + '\nAssert-CleanPublicWorktree\nWrite-Host "[tests] PASS"', + ) + + def test_bash_full_gates_check_all_public_state_before_and_after_validation(self): + for path, label in ( + (LINUX_FULL_GATE, "tests"), + (PRE_PUSH_GATE, "pre-push"), + ): + with self.subTest(path=path.name): + content = path.read_text(encoding="utf-8") + + self.assertIn("assert_clean_public_worktree()", content) + self.assertIn("git status --porcelain --untracked-files=all", content) + self.assertEqual( + content.splitlines().count("assert_clean_public_worktree"), 2 + ) + _assert_ordered( + self, + content, + "\nassert_clean_public_worktree\n", + "npm ci", + f'\nassert_clean_public_worktree\necho "[{label}] PASS"', + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_high_risk_acceptance.py b/tests/test_high_risk_acceptance.py new file mode 100644 index 0000000..36b9a37 --- /dev/null +++ b/tests/test_high_risk_acceptance.py @@ -0,0 +1,258 @@ +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "high_risk_acceptance.py" + + +def _git( + repo: Path, *args: str, check: bool = True +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["git", *args], + cwd=repo, + capture_output=True, + text=True, + check=check, + ) + + +def _commit(repo: Path, message: str) -> str: + _git(repo, "add", "--all") + _git( + repo, + "-c", + "user.name=Governance Test", + "-c", + "user.email=governance@example.invalid", + "commit", + "-m", + message, + ) + return _git(repo, "rev-parse", "HEAD").stdout.strip() + + +def _make_repo( + root: Path, changed_path: str, ignore_rule: str = ".planning/\n" +) -> tuple[Path, str, str]: + repo = root / "repo" + repo.mkdir() + _git(repo, "init", "-b", "dev") + (repo / ".gitignore").write_text(ignore_rule, encoding="utf-8") + initial = repo / changed_path + initial.parent.mkdir(parents=True, exist_ok=True) + initial.write_text("before\n", encoding="utf-8") + base = _commit(repo, "initial") + initial.write_text("after\n", encoding="utf-8") + candidate = _commit(repo, "candidate") + return repo, base, candidate + + +def _invoke( + repo: Path, + base: str, + candidate: str, + *, + include_closeout: bool = True, + output: str = ".planning/acceptance/R999.json", + item: str = "R999", + implementer: str = "implementer-a", + reviewer: str = "reviewer-b", + verdict: str = "APPROVED", + full_gate: str = "PASS", +) -> subprocess.CompletedProcess[str]: + command = [ + sys.executable, + str(SCRIPT), + "--base", + base, + "--candidate", + candidate, + ] + if include_closeout: + command.extend( + [ + "--item", + item, + "--implementer", + implementer, + "--reviewer", + reviewer, + "--review-verdict", + verdict, + "--full-gate-status", + full_gate, + "--output", + output, + ] + ) + return subprocess.run( + command, + cwd=repo, + capture_output=True, + text=True, + check=False, + ) + + +class HighRiskAcceptancePilotTests(unittest.TestCase): + def test_valid_high_risk_closeout_writes_lightweight_receipt(self): + with tempfile.TemporaryDirectory() as tmpdir: + repo, base, candidate = _make_repo(Path(tmpdir), "services/safe_io.py") + result = _invoke(repo, base, candidate) + + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn("HIGH_RISK_ACCEPTANCE: PASS", result.stdout) + receipt_path = repo / ".planning/acceptance/R999.json" + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + self.assertEqual( + set(receipt), + { + "schema", + "generated_at", + "item", + "branch", + "base_commit", + "candidate_commit", + "changed_files", + "high_risk_changed_files", + "review", + "gates", + "limitations", + }, + ) + self.assertEqual(receipt["schema"], "openclaw-high-risk-receipt/1") + self.assertEqual(receipt["item"], "R999") + self.assertEqual(receipt["branch"], "dev") + self.assertEqual(receipt["base_commit"], base) + self.assertEqual(receipt["candidate_commit"], candidate) + self.assertEqual(receipt["changed_files"], ["services/safe_io.py"]) + self.assertEqual( + receipt["high_risk_changed_files"], ["services/safe_io.py"] + ) + self.assertEqual( + receipt["review"], + { + "implementer": "implementer-a", + "reviewer": "reviewer-b", + "verdict": "APPROVED", + }, + ) + self.assertEqual(receipt["gates"], {"full_test_sop": "PASS"}) + self.assertIn("not identity authentication", receipt["limitations"]) + self.assertNotIn(str(repo), receipt_path.read_text(encoding="utf-8")) + + def test_standard_risk_and_empty_diffs_are_not_applicable_without_receipts(self): + with tempfile.TemporaryDirectory() as tmpdir: + repo, base, candidate = _make_repo(Path(tmpdir), "docs/readme.md") + result = _invoke(repo, base, candidate, include_closeout=False, output="") + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn("HIGH_RISK_ACCEPTANCE: NOT_APPLICABLE", result.stdout) + self.assertFalse((repo / ".planning").exists()) + + empty = _invoke( + repo, candidate, candidate, include_closeout=False, output="" + ) + self.assertEqual(empty.returncode, 0, empty.stdout + empty.stderr) + self.assertIn("HIGH_RISK_ACCEPTANCE: NOT_APPLICABLE", empty.stdout) + self.assertFalse((repo / ".planning").exists()) + + def test_high_risk_closeout_rejects_dirty_or_mismatched_repository_state(self): + scenarios = ( + "dirty", + "non_head", + "symbolic_candidate", + "wrong_branch", + "non_ancestor", + ) + for scenario in scenarios: + with ( + self.subTest(scenario=scenario), + tempfile.TemporaryDirectory() as tmpdir, + ): + repo, base, candidate = _make_repo(Path(tmpdir), "services/safe_io.py") + if scenario == "dirty": + (repo / "services/safe_io.py").write_text( + "uncommitted\n", encoding="utf-8" + ) + elif scenario == "non_head": + extra = repo / "docs/extra.md" + extra.parent.mkdir(parents=True, exist_ok=True) + extra.write_text("extra\n", encoding="utf-8") + _commit(repo, "later") + elif scenario == "symbolic_candidate": + candidate = "HEAD" + elif scenario == "wrong_branch": + _git(repo, "branch", "-m", "main") + else: + _git(repo, "checkout", "--orphan", "other") + other = repo / "other.txt" + other.write_text("other\n", encoding="utf-8") + _git(repo, "rm", "-r", "--cached", ".") + candidate = _commit(repo, "unrelated") + _git(repo, "branch", "-D", "dev") + _git(repo, "branch", "-m", "dev") + + result = _invoke(repo, base, candidate) + self.assertNotEqual(result.returncode, 0) + self.assertIn("HIGH_RISK_ACCEPTANCE: FAIL", result.stdout) + self.assertFalse((repo / ".planning/acceptance/R999.json").exists()) + + def test_high_risk_closeout_rejects_incomplete_or_non_independent_review(self): + cases = ( + {"item": "r999"}, + {"reviewer": "IMPLEMENTER-A"}, + {"verdict": "CHANGES_REQUESTED"}, + {"full_gate": "FAIL"}, + {"reviewer": ""}, + ) + for overrides in cases: + with ( + self.subTest(overrides=overrides), + tempfile.TemporaryDirectory() as tmpdir, + ): + repo, base, candidate = _make_repo(Path(tmpdir), "services/safe_io.py") + result = _invoke(repo, base, candidate, **overrides) + self.assertNotEqual(result.returncode, 0) + self.assertIn("HIGH_RISK_ACCEPTANCE: FAIL", result.stdout) + self.assertFalse((repo / ".planning/acceptance/R999.json").exists()) + + def test_receipt_output_must_be_new_ignored_path_under_planning(self): + with tempfile.TemporaryDirectory() as tmpdir: + repo, base, candidate = _make_repo(Path(tmpdir), "services/safe_io.py") + outside = _invoke(repo, base, candidate, output="receipt.json") + self.assertNotEqual(outside.returncode, 0) + self.assertFalse((repo / "receipt.json").exists()) + + first = _invoke(repo, base, candidate) + self.assertEqual(first.returncode, 0, first.stdout + first.stderr) + second = _invoke(repo, base, candidate) + self.assertNotEqual(second.returncode, 0) + self.assertIn("already exists", second.stdout) + + with tempfile.TemporaryDirectory() as tmpdir: + repo, base, candidate = _make_repo( + Path(tmpdir), + "services/safe_io.py", + ignore_rule=".planning/accepted/\n", + ) + unignored = _invoke(repo, base, candidate) + self.assertNotEqual(unignored.returncode, 0) + self.assertIn("must be ignored", unignored.stdout) + self.assertFalse((repo / ".planning/acceptance/R999.json").exists()) + + def test_high_risk_path_policy_is_reused_not_duplicated(self): + source = SCRIPT.read_text(encoding="utf-8") + self.assertIn("run_adversarial_gate", source) + self.assertIn("DEFAULT_HIGH_RISK_PATTERNS", source) + self.assertIn("_filter_high_risk_files", source) + self.assertNotIn('"services/safe_io.py"', source) + self.assertNotIn("DEFAULT_HIGH_RISK_PATTERNS =", source) + + +if __name__ == "__main__": + unittest.main()