From da9c0ba54cc6de37e114ebcac789b764952c2269 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:08:42 -0400 Subject: [PATCH 01/10] feat: enforce validation environment proof in reviewer reports (Closes #311) Add assess_validation_environment_proof requiring exact validation command, working directory, result evidence, and bare-pytest executable/version/venv proof before reviewer reports can claim validation success. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 216 +++++++++++++++++++++++++++++++++++- tests/test_review_proofs.py | 103 +++++++++++++++++ 2 files changed, 318 insertions(+), 1 deletion(-) diff --git a/review_proofs.py b/review_proofs.py index 9cd7e8c..3962e41 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1231,7 +1231,8 @@ def build_final_report(checkout_proof, inventory, validation, contamination, role_boundary=None, review_mutation=None, report_text=None, review_decision_lock=None, controller_handoff=None, capability_proof=None, - sweep_proof=None, worktree_proof=None): + sweep_proof=None, worktree_proof=None, + validation_environment=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. Combines the individual proof verdicts into the final-report fields the @@ -1258,6 +1259,22 @@ def build_final_report(checkout_proof, inventory, validation, contamination, ) empty_queue_report = assess_empty_queue_report(report_text) + if validation_environment is None and report_text: + validation_env_proof = assess_validation_environment_proof( + report_text=report_text, + ) + elif validation_environment is not None: + validation_env_proof = assess_validation_environment_proof( + report_text=report_text, + validation_environment=validation_environment, + ) + else: + validation_env_proof = { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + } contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) @@ -1387,6 +1404,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "empty-queue report missing or failed trust-gate proof (#198)" ) downgrade_reasons.extend(empty_queue_report.get("reasons", [])) + if not validation_env_proof.get("proven"): + downgrade_reasons.append( + "validation environment proof missing or failed (#311)" + ) + downgrade_reasons.extend(validation_env_proof.get("reasons", [])) merge_allowed = ( identity_eligible @@ -1398,9 +1420,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, # #179: no merge without a proven final live-state recheck. and live_state_proven and worktree_proven + and validation_env_proof.get("proven") ) violations = [] + violations.extend(validation_env_proof.get("violations", [])) if merge_performed and not merge_allowed: violations.append( "merge was performed/claimed although the proofs did not allow " @@ -1452,6 +1476,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, else True ), "empty_queue_trust_gate_status": empty_queue_report.get("status"), + "validation_environment_proven": bool( + validation_env_proof.get("proven") + ), + "validation_environment_violations": list( + validation_env_proof.get("violations") or [] + ), } @@ -3020,6 +3050,190 @@ def build_review_mutation_proof(run_log: list[dict]) -> dict: } +# --------------------------------------------------------------------------- +# Validation environment proof (#311) +# --------------------------------------------------------------------------- + +_VALIDATION_CLAIM_RE = re.compile( + r"(?:validation\s+command|pytest\s+(?:executed|passed|failed)|" + r"\d+\s+passed|\d+\s+failed|tests?\s+passed)", + re.IGNORECASE, +) + +_VALIDATION_CMD_LINE_RE = re.compile( + r"validation\s+command\s*:\s*(.+)", + re.IGNORECASE, +) + +_VALIDATION_CWD_LINE_RE = re.compile( + r"working\s+directory\s*:\s*(\S+)", + re.IGNORECASE, +) + +_WHICH_PYTEST_LINE_RE = re.compile( + r"which\s+pytest\s*:\s*(.+)", + re.IGNORECASE, +) + +_PYTEST_VERSION_LINE_RE = re.compile( + r"pytest\s+--version\s*:\s*(.+)", + re.IGNORECASE, +) + +_BARE_PYTEST_CMD_RE = re.compile( + r"^(?:python\s+-m\s+)?pytest(?:\s|$)", + re.IGNORECASE, +) + +_VAGUE_PYTEST_CLAIM_RE = re.compile( + r"pytest\s+(?:executed|passed|failed)\b", + re.IGNORECASE, +) + + +def _parse_validation_command(report_text: str) -> str: + for line in (report_text or "").splitlines(): + match = _VALIDATION_CMD_LINE_RE.search(line) + if match: + return match.group(1).strip().rstrip(".") + return "" + + +def _parse_validation_cwd(report_text: str) -> str: + for line in (report_text or "").splitlines(): + match = _VALIDATION_CWD_LINE_RE.search(line) + if match: + return match.group(1).strip().rstrip(",.;") + return "" + + +def _is_bare_pytest_command(command: str) -> bool: + text = (command or "").strip() + if not text: + return False + if "/" in text or "\\" in text: + return False + return bool(_BARE_PYTEST_CMD_RE.match(text)) + + +def assess_validation_environment_proof( + *, + report_text: str | None = None, + validation_environment: dict | None = None, +) -> dict: + """#311: reviewer reports must include exact validation command/environment.""" + text = report_text or "" + lower = text.lower() + env = validation_environment or {} + reasons: list[str] = [] + violations: list[str] = [] + + claims_validation = bool(_VALIDATION_CLAIM_RE.search(text)) + if not claims_validation and not env.get("command"): + return { + "proven": True, + "block": False, + "claims_validation": False, + "reasons": [], + "violations": [], + "safe_next_action": "proceed", + } + + command = (env.get("command") or _parse_validation_command(text)).strip() + working_directory = ( + env.get("working_directory") or env.get("cwd") + or _parse_validation_cwd(text) + ).strip() + + if not command: + reasons.append( + "validation result claimed without exact validation command (#311)" + ) + elif _VAGUE_PYTEST_CLAIM_RE.search(text) and command == "": + violations.append( + "vague pytest summary without exact validation command (#311)" + ) + + if not working_directory: + reasons.append( + "validation result claimed without working directory (#311)" + ) + + has_result = any( + token in lower + for token in ("passed", "failed", "skipped", "error") + ) or any( + env.get(key) is not None + for key in ("passed", "failed", "skipped", "result") + ) + if not has_result: + reasons.append( + "validation result claimed without pass/fail result evidence (#311)" + ) + + if command and _is_bare_pytest_command(command): + which_pytest = ( + (env.get("which_pytest") or env.get("executable_path") or "") + .strip() + ) + if not which_pytest: + for line in text.splitlines(): + match = _WHICH_PYTEST_LINE_RE.search(line) + if match: + which_pytest = match.group(1).strip() + break + if not which_pytest: + reasons.append( + "bare pytest command requires which-pytest/executable proof (#311)" + ) + + pytest_version = (env.get("pytest_version") or "").strip() + if not pytest_version: + for line in text.splitlines(): + match = _PYTEST_VERSION_LINE_RE.search(line) + if match: + pytest_version = match.group(1).strip() + break + if not pytest_version: + reasons.append( + "bare pytest command requires pytest --version proof (#311)" + ) + + venv_resolved = env.get("venv_resolved") + if venv_resolved is None: + venv_resolved = any( + marker in lower + for marker in ( + "project venv", + "venv/bin/pytest", + "resolves to the project venv", + "venv resolved: yes", + ) + ) + if not venv_resolved and which_pytest and "venv" not in which_pytest: + reasons.append( + "bare pytest command requires proof whether executable " + "resolves to the project venv (#311)" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations), + "claims_validation": True, + "command": command or None, + "working_directory": working_directory or None, + "reasons": reasons, + "violations": violations, + "safe_next_action": ( + "report exact validation command, working directory, result, and " + "for bare pytest also which-pytest, pytest --version, and venv proof" + if not proven + else "proceed" + ), + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index b4683dc..10fd0d3 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -24,6 +24,7 @@ from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, + assess_validation_environment_proof, assess_capability_evidence, assess_capability_proof, assess_contradictory_no_pr_claim, @@ -2160,5 +2161,107 @@ class TestCreateIssueFinalReport(unittest.TestCase): self.assertTrue(result["downgraded"]) +class TestValidationEnvironmentProof(unittest.TestCase): + """Issue #311: exact validation command and execution environment.""" + + ROOT = "/repo/Gitea-Tools/branches/review-pr-280" + + def test_venv_pytest_with_full_report_passes(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: venv/bin/pytest tests/ --ignore=branches\n" + f"Working directory: {self.ROOT}\n" + "Result: 1014 passed, 6 skipped" + ), + ) + self.assertTrue(result["proven"]) + + def test_bare_pytest_without_path_proof_fails(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: pytest\n" + f"Working directory: {self.ROOT}\n" + "1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_bare_pytest_with_path_and_version_proof_passes(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: pytest tests/\n" + f"Working directory: {self.ROOT}\n" + "which pytest: /repo/Gitea-Tools/venv/bin/pytest\n" + "pytest --version: pytest 8.3.4\n" + "Resolves to the project venv: yes\n" + "1014 passed, 6 skipped" + ), + ) + self.assertTrue(result["proven"]) + + def test_vague_pytest_summary_without_command_fails(self): + result = assess_validation_environment_proof( + report_text=( + f"Working directory: {self.ROOT}\n" + "pytest executed inside the worktree: 1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_structured_environment_proof_passes(self): + result = assess_validation_environment_proof( + validation_environment={ + "command": "pytest tests/", + "working_directory": self.ROOT, + "which_pytest": "/repo/Gitea-Tools/venv/bin/pytest", + "pytest_version": "pytest 8.3.4", + "venv_resolved": True, + "passed": 1014, + "skipped": 6, + }, + ) + self.assertTrue(result["proven"]) + + def test_missing_working_directory_fails(self): + result = assess_validation_environment_proof( + report_text=( + "Validation command: venv/bin/pytest tests/\n" + "1014 passed, 6 skipped" + ), + ) + self.assertFalse(result["proven"]) + + def test_build_final_report_downgrades_missing_environment_proof(self): + report = build_final_report( + checkout_proof=_good_checkout(), + inventory=_good_inventory(), + validation=_good_validation(), + contamination=_good_contamination(), + identity_eligible=True, + merge_performed=False, + issue_status_verified=True, + capability_evidence=_good_capability_evidence(), + sweep=_good_sweep(), + live_state=_good_live_state(), + role_boundary=_good_role_boundary(), + review_mutation=_good_review_mutation(), + controller_handoff=_good_handoff(), + capability_proof=_good_capability_proof(), + sweep_proof=_good_secret_sweep(), + worktree_proof={ + "worktree_path": "/repo/branches/review-feat-issue-224", + "porcelain_status": "", + "pr_scope_files": ["docs/wiki/Repositories.md"], + "scratch_used": True, + }, + report_text=( + f"Working directory: {self.ROOT}\n" + "pytest executed: 1014 passed, 6 skipped" + ), + ) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["validation_environment_proven"]) + + if __name__ == "__main__": unittest.main() From 71b8d008972fccac7e2fbd3a44059d2b9930826a Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:13:30 -0400 Subject: [PATCH 02/10] feat: classify merge simulation as worktree/index mutations (Closes #317) Add assess_merge_simulation_report to reject local merge simulations listed as read-only diagnostics and require Worktree/index mutations documentation with worktree path, pre/merge/abort/post-clean proof when simulation commands appear in the session log. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 ++ reviewer_merge_simulation.py | 148 ++++++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_merge_simulation.py | 135 +++++++++++++++++++++ 4 files changed, 297 insertions(+), 1 deletion(-) create mode 100644 reviewer_merge_simulation.py create mode 100644 tests/test_reviewer_merge_simulation.py diff --git a/review_proofs.py b/review_proofs.py index 9cd7e8c..be24c08 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3215,3 +3215,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): "ref_mutating_commands": ref_commands, "fetch_targets": fetch_targets, } + + +def assess_merge_simulation_report(report_text, **kwargs): + """#317: classify local merge simulation as worktree/index mutations.""" + from reviewer_merge_simulation import assess_merge_simulation_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_merge_simulation.py b/reviewer_merge_simulation.py new file mode 100644 index 0000000..5397c4f --- /dev/null +++ b/reviewer_merge_simulation.py @@ -0,0 +1,148 @@ +"""Merge-simulation worktree mutation verifier for reviewer reports (#317).""" + +from __future__ import annotations + +import re +from typing import Any + +_WORKTREE_INDEX_FIELD_RE = re.compile( + r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$", + re.IGNORECASE | re.MULTILINE, +) +_READONLY_DIAG_RE = re.compile( + r"read[- ]only diagnostics\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_WORKTREE_PATH_RE = re.compile( + r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:", + re.IGNORECASE, +) +_PRE_CLEAN_RE = re.compile( + r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)", + re.IGNORECASE, +) +_MERGE_RESULT_RE = re.compile( + r"(?:merge result|simulation result|conflict status|merge conflict)", + re.IGNORECASE, +) +_ABORT_RE = re.compile( + r"(?:merge --abort|abort/reset command|abort command|git merge --abort)", + re.IGNORECASE, +) +_POST_CLEAN_RE = re.compile( + r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)", + re.IGNORECASE, +) + + +def _command_text(entry: Any) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _git_subcommand_line(command: str) -> str | None: + tokens = command.split() + if not tokens or tokens[0] != "git": + return None + return " ".join(tokens[1:]) + + +def merge_simulation_commands(command_log: list | None) -> list[str]: + """Return logged commands that perform local merge simulation (#317).""" + out: list[str] = [] + for entry in command_log or []: + command = _command_text(entry) + rest = _git_subcommand_line(command) + if rest is None: + continue + lower = command.lower() + if rest.startswith("merge"): + if "--no-commit" in lower or ( + "--no-ff" in lower and "merge" in lower and "--commit" not in lower + ): + out.append(command) + elif "--abort" in lower: + out.append(command) + return out + + +def assess_merge_simulation_report( + report_text: str, + *, + command_log: list | None = None, +) -> dict[str, Any]: + """Reject merge simulation classified as read-only; require worktree proof (#317).""" + text = report_text or "" + lower = text.lower() + reasons: list[str] = [] + simulation_commands = merge_simulation_commands(command_log) + has_abort = any("--abort" in c.lower() for c in simulation_commands) + has_simulation = any( + "--no-commit" in c.lower() or ( + "--no-ff" in c.lower() and "--abort" not in c.lower() + ) + for c in simulation_commands + ) + + if not simulation_commands: + return { + "proven": True, + "block": False, + "reasons": [], + "simulation_commands": [], + "safe_next_action": "proceed", + } + + field = _WORKTREE_INDEX_FIELD_RE.search(text) + value = (field.group(1).strip().lower() if field else "") or "" + if not value or value in {"none", "n/a", "no"}: + reasons.append( + "merge simulation commands ran but report lacks " + "'Worktree/index mutations' entry" + ) + elif "merge" not in value and "simulation" not in value: + if not any(cmd.lower() in lower for cmd in simulation_commands): + reasons.append( + "Worktree/index mutations must document merge simulation commands" + ) + + readonly = _READONLY_DIAG_RE.search(text) + if readonly: + readonly_value = readonly.group(1) + for command in simulation_commands: + if command.lower() in readonly_value.lower(): + reasons.append( + "merge simulation may not be listed under read-only diagnostics" + ) + if has_simulation and "merge simulation" in readonly_value.lower(): + reasons.append( + "merge simulation may not be classified as read-only diagnostics" + ) + + if has_simulation: + if not _WORKTREE_PATH_RE.search(text): + reasons.append("merge simulation report missing worktree path") + if not _PRE_CLEAN_RE.search(text): + reasons.append("merge simulation report missing pre-simulation clean status") + if not _MERGE_RESULT_RE.search(text): + reasons.append("merge simulation report missing merge result/conflict status") + if has_abort or has_simulation: + if has_abort and not _ABORT_RE.search(text): + reasons.append("merge simulation report missing abort/reset command proof") + if has_abort and not _POST_CLEAN_RE.search(text): + reasons.append("merge simulation report missing post-abort clean status") + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "simulation_commands": simulation_commands, + "safe_next_action": ( + "classify merge simulation under Worktree/index mutations with full " + "pre/merge/abort/post-clean proof; never list as read-only diagnostics" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index f8d3b04..7f80a40 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow(): def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report - assert callable(assess_create_issue_final_report) \ No newline at end of file + assert callable(assess_create_issue_final_report) + + +def test_merge_simulation_verifier_exported(): + from review_proofs import assess_merge_simulation_report + + assert callable(assess_merge_simulation_report) \ No newline at end of file diff --git a/tests/test_reviewer_merge_simulation.py b/tests/test_reviewer_merge_simulation.py new file mode 100644 index 0000000..e3101f8 --- /dev/null +++ b/tests/test_reviewer_merge_simulation.py @@ -0,0 +1,135 @@ +"""Tests for merge-simulation worktree mutation verifier (#317).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_merge_simulation import ( # noqa: E402 + assess_merge_simulation_report, + merge_simulation_commands, +) + + +def _good_simulation_report(**overrides) -> str: + lines = [ + "Worktree/index mutations: merge simulation in branches/review-pr281", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean (git status --porcelain empty)", + "Merge result: conflicts in review_proofs.py", + "Abort command: git merge --abort", + "Post-abort clean status: clean after abort (git status --porcelain empty)", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + text = text.replace(f"{key}:", f"{key}: {value}") + return text + + +class TestMergeSimulationDetection(unittest.TestCase): + def test_detects_no_commit_merge(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git status", + ]) + self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"]) + + def test_detects_merge_abort(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ]) + self.assertIn("git merge --abort", commands) + + +class TestMergeSimulationReport(unittest.TestCase): + def test_no_simulation_passes(self): + result = assess_merge_simulation_report( + "Review complete. Worktree/index mutations: none", + command_log=["git status --porcelain", "git diff prgs/master...HEAD"], + ) + self.assertTrue(result["proven"]) + + def test_documented_simulation_passes(self): + report = _good_simulation_report() + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_readonly_classification_blocks(self): + report = "\n".join([ + "Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status", + "Worktree/index mutations: none", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("read-only" in r.lower() for r in result["reasons"])) + + def test_missing_worktree_field_blocks(self): + report = "Merge simulation ran but only noted in prose." + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"])) + + def test_conflict_simulation_requires_merge_result(self): + report = _good_simulation_report().replace( + "Merge result: conflicts in review_proofs.py", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("merge result" in r.lower() for r in result["reasons"])) + + def test_aborted_simulation_requires_post_clean_proof(self): + report = _good_simulation_report().replace( + "Post-abort clean status: clean after abort (git status --porcelain empty)", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"])) + + def test_successful_simulation_without_abort_omits_abort_fields(self): + report = "\n".join([ + "Worktree/index mutations: merge simulation completed cleanly", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean before simulation", + "Merge result: clean fast-forward simulation", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_merge_simulation_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 4ea618e8b442e6bbd7d5f93c4b790e4241bc5c0d Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:19:22 -0400 Subject: [PATCH 03/10] Add validation integrity verifier for PR-head vs diagnostic tests (#316) Separate official PR-head validation from post-edit diagnostic runs in reviewer reports. Block treating diagnostic passes as official validation and require contaminated sessions to report recovery-required status. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 + reviewer_validation_integrity.py | 148 ++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_validation_integrity.py | 130 +++++++++++++++++ 4 files changed, 292 insertions(+), 1 deletion(-) create mode 100644 reviewer_validation_integrity.py create mode 100644 tests/test_reviewer_validation_integrity.py diff --git a/review_proofs.py b/review_proofs.py index d6acb01..ef74ad4 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3413,3 +3413,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): "ref_mutating_commands": ref_commands, "fetch_targets": fetch_targets, } + + +def assess_validation_integrity_report(report_text, **kwargs): + """#316: separate official PR-head validation from diagnostic experiments.""" + from reviewer_validation_integrity import assess_validation_integrity_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_validation_integrity.py b/reviewer_validation_integrity.py new file mode 100644 index 0000000..038c457 --- /dev/null +++ b/reviewer_validation_integrity.py @@ -0,0 +1,148 @@ +"""PR-head vs diagnostic validation integrity verifier (#316).""" + +from __future__ import annotations + +import re +from typing import Any + +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only validation", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official validation|pr-head validation|validation integrity)", + re.IGNORECASE, +) +_OFFICIAL_COMMAND_RE = re.compile( + r"(?:official validation command|pr-head validation command|validation command)", + re.IGNORECASE, +) +_OFFICIAL_RESULT_RE = re.compile( + r"(?:official validation result|pr-head validation result|validation result|validation integrity status)", + re.IGNORECASE, +) +_DIRTY_AFTER_RE = re.compile( + r"(?:worktree dirty after (?:official )?validation|dirty (?:state )?after (?:official )?validation|" + r"validation worktree dirty after)", + re.IGNORECASE, +) +_DIAGNOSTIC_EDIT_RE = re.compile( + r"(?:diagnostic edit(?:ed)? path|file edits by reviewer|diagnostic local edit)", + re.IGNORECASE, +) +_DIAGNOSTIC_COMMAND_RE = re.compile( + r"(?:diagnostic (?:validation )?command|diagnostic test run|diagnostic result)", + re.IGNORECASE, +) +_SUGGESTED_FIX_RE = re.compile( + r"(?:diagnostic results? (?:were )?used only for suggested fix|suggested fix only|" + r"not used as official validation)", + re.IGNORECASE, +) +_INTEGRITY_STATUS_RE = re.compile( + r"validation integrity status\s*:\s*(passed|failed|not run|contaminated)", + re.IGNORECASE, +) +_POST_EDIT_AS_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit|" + r"full suite passed after diagnostic edit)", + re.IGNORECASE, +) + + +def _performed_edits(action_log: list[dict] | None) -> list[str]: + paths: list[str] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + path = entry.get("path") + if path and entry.get("kind", "file_edit") in {"file_edit", "edit", "write"}: + paths.append(str(path)) + return paths + + +def assess_validation_integrity_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Separate official PR-head validation from diagnostic edited-worktree tests (#316).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + diagnostic_edits = list(session.get("diagnostic_edits") or []) + if not diagnostic_edits: + diagnostic_edits = _performed_edits(action_log) + diagnostic_ran = bool( + session.get("diagnostic_validation_ran") + or session.get("diagnostic_runs") + ) + official_ran = bool(session.get("official_validation_ran", True)) + official_result = (session.get("official_result") or "").strip().lower() + diagnostic_result = (session.get("diagnostic_result") or "").strip().lower() + dirty_after = session.get("worktree_dirty_after_official") + contaminated = bool(session.get("contaminated")) + + if official_ran: + if not _OFFICIAL_VALIDATION_RE.search(text) and not _OFFICIAL_COMMAND_RE.search(text): + reasons.append("report missing official PR-head validation section") + if not _OFFICIAL_COMMAND_RE.search(text) and "validation:" not in text.lower(): + reasons.append("report missing official validation command") + if not _OFFICIAL_RESULT_RE.search(text): + reasons.append("report missing official validation result or integrity status") + if dirty_after is not None and not _DIRTY_AFTER_RE.search(text): + reasons.append( + "report missing worktree dirty state after official validation" + ) + elif dirty_after is None and official_ran and not _DIRTY_AFTER_RE.search(text): + reasons.append( + "report missing worktree dirty state after official validation" + ) + + if diagnostic_edits or diagnostic_ran: + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "post-edit diagnostic runs must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if diagnostic_edits and not _DIAGNOSTIC_EDIT_RE.search(text): + reasons.append("report missing diagnostic edit path") + if diagnostic_ran and not _DIAGNOSTIC_COMMAND_RE.search(text): + reasons.append("report missing diagnostic command/result") + if not _SUGGESTED_FIX_RE.search(text): + reasons.append( + "report must state diagnostic results were used only for suggested fix" + ) + if _POST_EDIT_AS_OFFICIAL_RE.search(text): + reasons.append( + "post-edit diagnostic results cannot be reported as official PR-head validation" + ) + if diagnostic_result == "pass" and official_result == "fail": + if "request_changes" not in text.lower() and "failed" not in text.lower(): + reasons.append( + "request-changes must cite unmodified PR-head failure, not diagnostic pass" + ) + + if contaminated: + status = _INTEGRITY_STATUS_RE.search(text) + if not status or "contaminated" not in status.group(1).lower(): + reasons.append( + "contaminated validation must report integrity status " + "'contaminated — recovery required'" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "separate official PR-head validation from diagnostic experiments; " + "report dirty-after-validation state" + if reasons + else "proceed" + ), + } diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index f8d3b04..6f13a84 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow(): def test_create_issue_final_report_verifier_exported(): from review_proofs import assess_create_issue_final_report - assert callable(assess_create_issue_final_report) \ No newline at end of file + assert callable(assess_create_issue_final_report) + + +def test_validation_integrity_verifier_exported(): + from review_proofs import assess_validation_integrity_report + + assert callable(assess_validation_integrity_report) \ No newline at end of file diff --git a/tests/test_reviewer_validation_integrity.py b/tests/test_reviewer_validation_integrity.py new file mode 100644 index 0000000..5137057 --- /dev/null +++ b/tests/test_reviewer_validation_integrity.py @@ -0,0 +1,130 @@ +"""Tests for PR-head vs diagnostic validation integrity verifier (#316).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_integrity import assess_validation_integrity_report # noqa: E402 + + +def _official_only_report() -> str: + return "\n".join([ + "Official PR-head validation on unmodified worktree.", + "Official validation command: pytest tests/ -q", + "Official validation result: failed (3 failed)", + "Validation integrity status: failed", + "Worktree dirty after official validation: clean", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + _official_only_report(), + "Diagnostic local experiment — not PR-head validation", + "Diagnostic edit path: tests/test_example.py", + "File edits by reviewer: tests/test_example.py", + "Diagnostic command: pytest tests/test_example.py -q", + "Diagnostic result: passed after local fix", + "Diagnostic results used only for suggested fix; blocker remains PR-head failure.", + ]) + + +class TestValidationIntegrity(unittest.TestCase): + def test_official_only_passes(self): + result = assess_validation_integrity_report( + _official_only_report(), + validation_session={ + "official_validation_ran": True, + "official_result": "fail", + "worktree_dirty_after_official": False, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_failing_pr_head_with_diagnostic_pass_passes(self): + result = assess_validation_integrity_report( + _diagnostic_report(), + validation_session={ + "official_validation_ran": True, + "official_result": "fail", + "diagnostic_validation_ran": True, + "diagnostic_result": "pass", + "diagnostic_edits": ["tests/test_example.py"], + "worktree_dirty_after_official": False, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_post_edit_tests_as_official_blocks(self): + report = "\n".join([ + "Official validation result: passed after diagnostic edit", + "pytest passed after local fix in validation worktree", + ]) + result = assess_validation_integrity_report( + report, + validation_session={ + "official_validation_ran": True, + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_validation_ran": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_missing_dirty_after_validation_blocks(self): + report = _official_only_report().replace( + "Worktree dirty after official validation: clean", + "", + ) + result = assess_validation_integrity_report( + report, + validation_session={"official_validation_ran": True, "official_result": "fail"}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("dirty" in r.lower() for r in result["reasons"])) + + def test_diagnostic_without_label_blocks(self): + report = _diagnostic_report().replace( + "Diagnostic local experiment — not PR-head validation", + "", + ) + result = assess_validation_integrity_report( + report, + validation_session={ + "official_validation_ran": True, + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_validation_ran": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_contaminated_requires_integrity_status(self): + report = "Validation worktree was edited before official validation completed." + result = assess_validation_integrity_report( + report, + validation_session={"contaminated": True, "official_validation_ran": True}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("contaminated" in r.lower() for r in result["reasons"])) + + def test_action_log_edits_trigger_diagnostic_rules(self): + result = assess_validation_integrity_report( + _official_only_report(), + validation_session={"official_validation_ran": True, "official_result": "fail"}, + action_log=[{"path": "tests/test_example.py", "kind": "file_edit", "performed": True}], + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_integrity_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() From ce3bd784c026618a9765a92e03be43b8e68b7430 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:26:37 -0400 Subject: [PATCH 04/10] Add worktree ownership verifier for reviewer reset safety (#312) Require session-owned review worktrees or documented safe-reuse proof before destructive reset/validation. Block contradictory workspace-none claims when git reset --hard occurred. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 + reviewer_worktree_ownership.py | 207 ++++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_worktree_ownership.py | 137 ++++++++++++++ 4 files changed, 358 insertions(+), 1 deletion(-) create mode 100644 reviewer_worktree_ownership.py create mode 100644 tests/test_reviewer_worktree_ownership.py diff --git a/review_proofs.py b/review_proofs.py index 60cca02..d6c6bc6 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3816,3 +3816,10 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_worktree_ownership_report(report_text, **kwargs): + """#312: prove session-owned or safe-reuse worktree before reset/validation.""" + from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_worktree_ownership.py b/reviewer_worktree_ownership.py new file mode 100644 index 0000000..d30c48d --- /dev/null +++ b/reviewer_worktree_ownership.py @@ -0,0 +1,207 @@ +"""Reviewer worktree ownership and safe-reuse proof verifier (#312).""" + +from __future__ import annotations + +import re +from typing import Any + +_SESSION_OWNED_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_WORKTREE_PATH_RE = re.compile( + r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_BRANCHES_PATH_RE = re.compile(r"/branches/|\bbranches/", re.IGNORECASE) +_MAIN_CHECKOUT_RE = re.compile( + r"main checkout|not (?:the )?main checkout|outside branches/", + re.IGNORECASE, +) +_SAFE_REUSE_RE = re.compile(r"safe[- ]reuse proof", re.IGNORECASE) +_REUSE_POLICY_RE = re.compile( + r"(?:project policy|policy) (?:allows|allowing) (?:reuse|reset)", + re.IGNORECASE, +) +_CLEAN_TRACKED_RE = re.compile( + r"(?:clean tracked state|tracked state\s*:\s*clean|no uncommitted tracked)", + re.IGNORECASE, +) +_CLEAN_UNTRACKED_RE = re.compile( + r"(?:clean untracked state|untracked state\s*:\s*clean|no untracked files)", + re.IGNORECASE, +) +_NOT_OTHER_SESSION_RE = re.compile( + r"not owned by (?:another|other) (?:active )?(?:task|session)", + re.IGNORECASE, +) +_BRANCH_BEFORE_RESET_RE = re.compile( + r"(?:branch/head before reset|head before reset|branch before reset)\s*:\s*(\S+)", + re.IGNORECASE, +) +_RESET_TARGET_RE = re.compile( + r"(?:reset target sha|reset target)\s*:\s*([0-9a-f]{7,40})", + re.IGNORECASE, +) +_DESTRUCTIVE_CMD_RE = re.compile( + r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?" + r"(?:reset\s+--hard|clean(?:\s+-[A-Za-z]+)*|checkout\s+(?!HEAD\s+--)|switch\s+)", + re.IGNORECASE, +) +_WORKSPACE_NONE_RE = re.compile(r"workspace mutations\s*:\s*none", re.IGNORECASE) +_WORKTREE_MUTATION_RE = re.compile( + r"(?:worktree(?:/index)? mutations|destructive reset)\s*:\s*(?!none\b).+", + re.IGNORECASE, +) +_RESET_IN_REPORT_RE = re.compile(r"reset\s+--hard", re.IGNORECASE) + + +def _command_text(entry) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _destructive_commands(command_log: list | None) -> list[str]: + return [ + cmd + for cmd in (_command_text(entry) for entry in (command_log or [])) + if cmd and _DESTRUCTIVE_CMD_RE.search(cmd) + ] + + +def _extract_worktree_path(text: str, session: dict) -> str: + path = (session.get("worktree_path") or "").strip() + if path: + return path + match = _WORKTREE_PATH_RE.search(text or "") + return match.group(1).strip() if match else "" + + +def _is_session_owned_path(path: str) -> bool: + normalized = (path or "").replace("\\", "/") + return bool(_SESSION_OWNED_RE.search(normalized)) + + +def _safe_reuse_fields_present(text: str) -> list[str]: + missing: list[str] = [] + if not _WORKTREE_PATH_RE.search(text): + missing.append("exact worktree path") + if not _BRANCHES_PATH_RE.search(text): + missing.append("worktree inside branches/") + if not _MAIN_CHECKOUT_RE.search(text): + missing.append("worktree is not the main checkout") + if not _NOT_OTHER_SESSION_RE.search(text): + missing.append("worktree not owned by another active session") + if not _CLEAN_TRACKED_RE.search(text): + missing.append("clean tracked state") + if not _CLEAN_UNTRACKED_RE.search(text): + missing.append("clean untracked state") + if not _BRANCH_BEFORE_RESET_RE.search(text): + missing.append("branch/head before reset") + if not _RESET_TARGET_RE.search(text): + missing.append("reset target SHA") + if not _REUSE_POLICY_RE.search(text): + missing.append("explicit project policy allowing reuse/reset") + return missing + + +def assess_worktree_ownership_report( + report_text: str, + *, + ownership_session: dict | None = None, + command_log: list | None = None, +) -> dict[str, Any]: + """Prove reviewer worktree ownership before reset or validation (#312).""" + text = report_text or "" + session = dict(ownership_session or {}) + reasons: list[str] = [] + + worktree_path = _extract_worktree_path(text, session) + destructive = _destructive_commands(command_log or session.get("command_log")) + if not destructive and session.get("destructive_reset"): + destructive = ["git reset --hard (session)"] + + session_owned = bool( + session.get("session_owned") + or (worktree_path and _is_session_owned_path(worktree_path)) + ) + safe_reuse = bool(session.get("safe_reuse") or _SAFE_REUSE_RE.search(text)) + main_checkout = bool(session.get("main_checkout")) + dirty_tracked = session.get("dirty_tracked") + dirty_untracked = session.get("dirty_untracked") + other_session_owned = bool(session.get("other_session_owned")) + + if worktree_path or destructive or session.get("validation_ran"): + if not worktree_path: + reasons.append("report missing exact review worktree path") + elif main_checkout or "branches/" not in worktree_path.replace("\\", "/"): + reasons.append("review worktree must be inside branches/, not the main checkout") + elif not _BRANCHES_PATH_RE.search(worktree_path.replace("\\", "/")): + reasons.append("review worktree path must be under branches/") + + if other_session_owned and not safe_reuse: + reasons.append( + "reused worktree appears owned by another active task/session; " + "safe-reuse proof required" + ) + + if dirty_tracked: + reasons.append( + "worktree has uncommitted tracked changes before reset or validation" + ) + if dirty_untracked and safe_reuse: + reasons.append("safe-reuse proof requires clean untracked state") + + if safe_reuse and not session_owned: + reasons.extend( + f"safe-reuse proof missing {field}" + for field in _safe_reuse_fields_present(text) + ) + + if destructive: + if not session_owned and not safe_reuse: + reasons.append( + "destructive worktree commands forbidden without session-owned " + "worktree or safe-reuse proof" + ) + if _WORKSPACE_NONE_RE.search(text) and ( + _RESET_IN_REPORT_RE.search(text) or any("reset" in c.lower() for c in destructive) + ): + reasons.append( + "final report must not claim 'Workspace mutations: none' when " + "git reset --hard occurred" + ) + if not _WORKTREE_MUTATION_RE.search(text) and not _RESET_IN_REPORT_RE.search(text): + reasons.append( + "destructive reset must be reported under Worktree/index mutations " + "or destructive reset operations" + ) + + if ( + worktree_path + and not session_owned + and not safe_reuse + and (destructive or session.get("validation_ran")) + and not _is_session_owned_path(worktree_path) + ): + reasons.append( + "reused branch-named worktree requires safe-reuse proof before reset or validation" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "worktree_path": worktree_path or None, + "session_owned": session_owned, + "safe_reuse": safe_reuse, + "destructive_commands": destructive, + "safe_next_action": ( + "use a fresh session-owned review worktree under branches/review-pr-* " + "or document full safe-reuse proof before destructive reset" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 7e0fe23..443f88b 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -127,4 +127,10 @@ def test_create_issue_final_report_verifier_exported(): def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof - assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file + assert callable(assess_non_mergeable_skip_proof) + + +def test_worktree_ownership_verifier_exported(): + from review_proofs import assess_worktree_ownership_report + + assert callable(assess_worktree_ownership_report) \ No newline at end of file diff --git a/tests/test_reviewer_worktree_ownership.py b/tests/test_reviewer_worktree_ownership.py new file mode 100644 index 0000000..3b9dad3 --- /dev/null +++ b/tests/test_reviewer_worktree_ownership.py @@ -0,0 +1,137 @@ +"""Tests for reviewer worktree ownership verifier (#312).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_worktree_ownership import assess_worktree_ownership_report # noqa: E402 + + +def _fresh_review_report() -> str: + return "\n".join([ + "Review worktree path: branches/review-pr280-feat-issue-275-claim", + "Worktree is inside branches/ and not the main checkout.", + "Tracked state: clean; untracked state: clean.", + "Official PR-head validation on session-owned worktree.", + "Worktree mutations: none", + ]) + + +def _safe_reuse_report() -> str: + return "\n".join([ + "Safe-reuse proof for existing review worktree.", + "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks", + "Worktree inside branches/; not the main checkout.", + "Not owned by another active task/session.", + "Clean tracked state; clean untracked state.", + "Branch/head before reset: feat/issue-260-example", + "Reset target SHA: abc123def4567890abcdef1234567890abcdef12", + "Project policy allows reuse/reset for this clean review worktree.", + "Worktree mutations: git reset --hard FETCH_HEAD", + "Destructive reset operations: git reset --hard FETCH_HEAD", + ]) + + +class TestWorktreeOwnership(unittest.TestCase): + def test_fresh_session_owned_passes(self): + result = assess_worktree_ownership_report( + _fresh_review_report(), + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["session_owned"]) + + def test_safe_reuse_with_reset_passes(self): + result = assess_worktree_ownership_report( + _safe_reuse_report(), + ownership_session={ + "safe_reuse": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_destructive_reset_without_proof_blocks(self): + report = "\n".join([ + "Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks", + "Worktree mutations: git reset --hard FETCH_HEAD", + ]) + result = assess_worktree_ownership_report( + report, + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_reused_dirty_tracked_blocks(self): + result = assess_worktree_ownership_report( + _fresh_review_report(), + ownership_session={ + "validation_ran": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + "dirty_tracked": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("tracked" in r.lower() for r in result["reasons"])) + + def test_safe_reuse_dirty_untracked_blocks(self): + result = assess_worktree_ownership_report( + _safe_reuse_report(), + ownership_session={ + "safe_reuse": True, + "dirty_untracked": True, + "worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("untracked" in r.lower() for r in result["reasons"])) + + def test_workspace_none_with_reset_blocks(self): + report = "\n".join([ + "Review worktree path: branches/review-pr280-feat-issue-275-claim", + "Workspace mutations: none", + "Worktree mutations: git reset --hard FETCH_HEAD", + ]) + result = assess_worktree_ownership_report( + report, + ownership_session={ + "session_owned": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + }, + command_log=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"])) + + def test_main_checkout_blocks(self): + result = assess_worktree_ownership_report( + "Review worktree path: /Users/dev/Gitea-Tools", + ownership_session={ + "validation_ran": True, + "main_checkout": True, + "worktree_path": "/Users/dev/Gitea-Tools", + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_worktree_ownership_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From b842a7f9234c7b1d30899545b00bdfe310906717 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:28:19 -0400 Subject: [PATCH 05/10] Add validation worktree no-edit verifier for reviewer reports (#315) Block file edits in PR validation worktrees, require separate diagnostic scratch worktrees with explicit labeling, and reject contradictory File edits by reviewer: none claims. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 9 + reviewer_validation_worktree_edits.py | 184 ++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- ...test_reviewer_validation_worktree_edits.py | 134 +++++++++++++ 4 files changed, 334 insertions(+), 1 deletion(-) create mode 100644 reviewer_validation_worktree_edits.py create mode 100644 tests/test_reviewer_validation_worktree_edits.py diff --git a/review_proofs.py b/review_proofs.py index 02fa543..043ec35 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3832,3 +3832,12 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_validation_worktree_edit_report(report_text, **kwargs): + """#315: block edits in PR validation worktrees; require diagnostic separation.""" + from reviewer_validation_worktree_edits import ( + assess_validation_worktree_edit_report as _assess, + ) + + return _assess(report_text, **kwargs) diff --git a/reviewer_validation_worktree_edits.py b/reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..c4e904f --- /dev/null +++ b/reviewer_validation_worktree_edits.py @@ -0,0 +1,184 @@ +"""PR validation worktree read-only edit verifier (#315).""" + +from __future__ import annotations + +import re +from typing import Any + +_VALIDATION_WORKTREE_RE = re.compile( + r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_RE = re.compile( + r"branches/(?:diagnostic[\w/-]*|scratch[\w/-]*)", + re.IGNORECASE, +) +_FILE_EDITS_NONE_RE = re.compile( + r"file edits by reviewer\s*:\s*none\b", + re.IGNORECASE, +) +_FILE_EDITS_FIELD_RE = re.compile( + r"file edits by reviewer\s*:\s*(.+)", + re.IGNORECASE, +) +_VALIDATION_WORKTREE_PATH_RE = re.compile( + r"(?:validation worktree path|review worktree path)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_WORKTREE_PATH_RE = re.compile( + r"(?:diagnostic (?:scratch )?worktree path|diagnostic worktree)\s*:\s*(\S+)", + re.IGNORECASE, +) +_DIAGNOSTIC_LABEL_RE = re.compile( + r"diagnostic local experiment|not pr-head validation|diagnostic-only", + re.IGNORECASE, +) +_OFFICIAL_VALIDATION_RE = re.compile( + r"(?:official (?:pr-head )?validation|pr-head validation result)", + re.IGNORECASE, +) +_POST_EDIT_OFFICIAL_RE = re.compile( + r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit)", + re.IGNORECASE, +) + +_PERFORMED_EDIT_KINDS = frozenset({"file_edit", "edit", "write", "edited", "created", "wrote"}) + + +def _normalize_path(path: str) -> str: + return (path or "").replace("\\", "/").strip() + + +def _is_validation_worktree(path: str) -> bool: + return bool(_VALIDATION_WORKTREE_RE.search(_normalize_path(path))) + + +def _is_diagnostic_worktree(path: str) -> bool: + normalized = _normalize_path(path) + return bool( + _DIAGNOSTIC_WORKTREE_RE.search(normalized) + or "diagnostic" in normalized.lower() + ) + + +def _performed_edits(action_log: list[dict] | None) -> list[dict[str, str]]: + edits: list[dict[str, str]] = [] + for entry in action_log or []: + if entry.get("gated_rejected") or entry.get("performed") is False: + continue + kind = (entry.get("kind") or entry.get("action") or "file_edit").strip().lower() + if kind not in _PERFORMED_EDIT_KINDS: + continue + path = (entry.get("path") or "").strip() + if not path: + continue + worktree = _normalize_path(str(entry.get("worktree_path") or entry.get("cwd") or "")) + edits.append({"path": path, "worktree_path": worktree}) + return edits + + +def _extract_validation_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("validation_worktree_path") or "")) + if path: + return path + match = _VALIDATION_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def _extract_diagnostic_path(text: str, session: dict) -> str: + path = _normalize_path(str(session.get("diagnostic_worktree_path") or "")) + if path: + return path + match = _DIAGNOSTIC_WORKTREE_PATH_RE.search(text or "") + return _normalize_path(match.group(1)) if match else "" + + +def assess_validation_worktree_edit_report( + report_text: str, + *, + validation_session: dict | None = None, + action_log: list[dict] | None = None, +) -> dict[str, Any]: + """Block edits in PR validation worktrees; require diagnostic separation (#315).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + + validation_path = _extract_validation_path(text, session) + diagnostic_path = _extract_diagnostic_path(text, session) + validation_edits = list(session.get("validation_worktree_edits") or []) + diagnostic_edits = list(session.get("diagnostic_edits") or []) + edits = _performed_edits(action_log) + + if not validation_edits: + for entry in edits: + wt = entry.get("worktree_path") or validation_path + if wt and _is_validation_worktree(wt) and not _is_diagnostic_worktree(wt): + validation_edits.append(entry["path"]) + + if not diagnostic_edits: + for entry in edits: + wt = entry.get("worktree_path") or "" + if wt and _is_diagnostic_worktree(wt): + diagnostic_edits.append(entry["path"]) + elif entry["path"] and diagnostic_path and not validation_edits: + if _is_diagnostic_worktree(diagnostic_path): + diagnostic_edits.append(entry["path"]) + + claimed_none = bool(_FILE_EDITS_NONE_RE.search(text)) + any_edits = bool(validation_edits or diagnostic_edits or edits) + + if validation_edits: + reasons.append( + "reviewer must not edit files in the PR validation worktree; " + f"observed edits: {', '.join(validation_edits)}" + ) + + if diagnostic_edits or session.get("diagnostic_experiment"): + if not diagnostic_path and not _DIAGNOSTIC_WORKTREE_PATH_RE.search(text): + reasons.append( + "diagnostic experiments require a separate diagnostic scratch worktree path" + ) + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "diagnostic experiments must be labeled " + "'Diagnostic local experiment — not PR-head validation'" + ) + if not _FILE_EDITS_FIELD_RE.search(text) or claimed_none: + reasons.append( + "diagnostic edits must be reported under 'File edits by reviewer'" + ) + + if any_edits and claimed_none: + reasons.append( + "report claims 'File edits by reviewer: none' but reviewer file edits occurred" + ) + + if session.get("official_validation_after_edit") or _POST_EDIT_OFFICIAL_RE.search(text): + if validation_edits or (validation_path and diagnostic_edits): + reasons.append( + "post-edit test results cannot be reported as official PR-head validation" + ) + + if validation_edits and _OFFICIAL_VALIDATION_RE.search(text): + if not _DIAGNOSTIC_LABEL_RE.search(text): + reasons.append( + "validation worktree was edited; official PR-head validation is no longer valid" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "validation_worktree_path": validation_path or None, + "diagnostic_worktree_path": diagnostic_path or None, + "validation_worktree_edits": validation_edits, + "diagnostic_edits": diagnostic_edits, + "safe_next_action": ( + "keep PR validation worktrees read-only; use a separate diagnostic " + "scratch worktree and report all reviewer file edits" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 7e0fe23..ad4524b 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -127,4 +127,10 @@ def test_create_issue_final_report_verifier_exported(): def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof - assert callable(assess_non_mergeable_skip_proof) \ No newline at end of file + assert callable(assess_non_mergeable_skip_proof) + + +def test_validation_worktree_edit_verifier_exported(): + from review_proofs import assess_validation_worktree_edit_report + + assert callable(assess_validation_worktree_edit_report) \ No newline at end of file diff --git a/tests/test_reviewer_validation_worktree_edits.py b/tests/test_reviewer_validation_worktree_edits.py new file mode 100644 index 0000000..adedaca --- /dev/null +++ b/tests/test_reviewer_validation_worktree_edits.py @@ -0,0 +1,134 @@ +"""Tests for PR validation worktree no-edit verifier (#315).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_validation_worktree_edits import ( # noqa: E402 + assess_validation_worktree_edit_report, +) + + +def _read_only_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "File edits by reviewer: none", + "Review decision: request_changes", + ]) + + +def _diagnostic_report() -> str: + return "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official PR-head validation on unmodified worktree.", + "Official validation result: failed (1 failed)", + "Review decision: request_changes", + "Diagnostic local experiment — not PR-head validation", + "Diagnostic scratch worktree path: branches/diagnostic-pr280-fix-test", + "File edits by reviewer: tests/test_agent_temp_artifacts.py (diagnostic only)", + "Diagnostic result: passed after local fix", + ]) + + +class TestValidationWorktreeEdits(unittest.TestCase): + def test_read_only_review_passes(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_validation_worktree_edit_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("validation worktree" in r.lower() for r in result["reasons"])) + + def test_file_edits_none_contradiction_blocks(self): + result = assess_validation_worktree_edit_report( + _read_only_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + }, + action_log=[ + { + "path": "tests/test_agent_temp_artifacts.py", + "kind": "file_edit", + "performed": True, + "worktree_path": "branches/review-pr280-feat-issue-275-claim", + } + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("file edits" in r.lower() for r in result["reasons"])) + + def test_diagnostic_scratch_with_reporting_passes(self): + result = assess_validation_worktree_edit_report( + _diagnostic_report(), + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "diagnostic_worktree_path": "branches/diagnostic-pr280-fix-test", + "diagnostic_edits": ["tests/test_agent_temp_artifacts.py"], + "diagnostic_experiment": True, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_diagnostic_without_scratch_worktree_blocks(self): + result = assess_validation_worktree_edit_report( + "File edits by reviewer: tests/test_example.py", + validation_session={ + "diagnostic_edits": ["tests/test_example.py"], + "diagnostic_experiment": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"])) + + def test_post_edit_official_validation_blocks(self): + report = "\n".join([ + "Validation worktree path: branches/review-pr280-feat-issue-275-claim", + "Official validation result: passed after local edit", + "File edits by reviewer: tests/test_agent_temp_artifacts.py", + ]) + result = assess_validation_worktree_edit_report( + report, + validation_session={ + "validation_worktree_path": ( + "branches/review-pr280-feat-issue-275-claim" + ), + "validation_worktree_edits": ["tests/test_agent_temp_artifacts.py"], + "official_validation_after_edit": True, + }, + ) + self.assertFalse(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_validation_worktree_edit_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From ef7c15f84dbcea85da563add21c5f5ec052a7561 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:28:41 -0400 Subject: [PATCH 06/10] fix: seed profile gate for TestIssueLockArtifactWarning (Closes #359) gitea_lock_issue is profile-gated via task_capability_map (gitea.issue.comment); without GITEA_ALLOWED_OPERATIONS the call fail-closes before the artifact-warning logic runs, so the test asserted success on a permission block. Seed the environment in setUp/tearDown exactly like tests/test_mcp_server.py::TestIssueLocking and complete the mocked git state with base_equivalent, inspected_git_root, and base_branch so the newer lock-worktree base-equivalence gate passes. The test still verifies the artifact warning fires for '?? _emit_payload.py'. Full suite now passes with zero failures. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_agent_temp_artifacts.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/tests/test_agent_temp_artifacts.py b/tests/test_agent_temp_artifacts.py index 2e14385..903eab8 100644 --- a/tests/test_agent_temp_artifacts.py +++ b/tests/test_agent_temp_artifacts.py @@ -62,7 +62,24 @@ class TestPreflightWarnings(unittest.TestCase): self.assertIn("_encode_commit_payload.py", status["preflight_warnings"][0]) +# Issue-write tools are profile-gated (#69); gitea_lock_issue requires +# gitea.issue.comment (see task_capability_map), so the gate must be +# seeded exactly like tests/test_mcp_server.py::TestIssueLocking (#359). +ISSUE_WRITE_ENV = { + "GITEA_ALLOWED_OPERATIONS": ( + "gitea.issue.create,gitea.issue.close,gitea.issue.comment" + ), +} + + class TestIssueLockArtifactWarning(unittest.TestCase): + def setUp(self): + self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True) + self._env_patcher.start() + + def tearDown(self): + self._env_patcher.stop() + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server._auth", return_value="token x") @patch("mcp_server._resolve", return_value=("h", "o", "r")) @@ -72,6 +89,9 @@ class TestIssueLockArtifactWarning(unittest.TestCase): mock_state.return_value = { "current_branch": "master", "porcelain_status": "?? _emit_payload.py\n", + "base_equivalent": True, + "inspected_git_root": "/scratch/wt", + "base_branch": "origin/master", } result = mcp_server.gitea_lock_issue( issue_number=261, From b71d3aeab38b1c05ff74f1c23e0d34f77e2a5ca4 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:31:24 -0400 Subject: [PATCH 07/10] Add precise mutation category verifier for controller handoffs (#319) Reject legacy Workspace mutations wording and require explicit file, worktree/index, and git ref mutation fields in reviewer controller handoffs. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 ++ reviewer_mutation_categories.py | 130 +++++++++++++++++++++ tests/test_llm_workflow_split.py | 6 + tests/test_reviewer_mutation_categories.py | 88 ++++++++++++++ 4 files changed, 231 insertions(+) create mode 100644 reviewer_mutation_categories.py create mode 100644 tests/test_reviewer_mutation_categories.py diff --git a/review_proofs.py b/review_proofs.py index e7c94e2..add5ddf 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3839,3 +3839,10 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess return _assess(report_text, **kwargs) + + +def assess_mutation_categories_report(report_text, **kwargs): + """#319: require precise mutation categories in reviewer controller handoffs.""" + from reviewer_mutation_categories import assess_mutation_categories_report as _assess + + return _assess(report_text, **kwargs) diff --git a/reviewer_mutation_categories.py b/reviewer_mutation_categories.py new file mode 100644 index 0000000..a09e73b --- /dev/null +++ b/reviewer_mutation_categories.py @@ -0,0 +1,130 @@ +"""Precise mutation category verifier for reviewer controller handoffs (#319).""" + +from __future__ import annotations + +import re +from typing import Any + +_CONTROLLER_HANDOFF_RE = re.compile(r"controller handoff", re.IGNORECASE) +_WORKSPACE_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*workspace\s+mutations\s*:", + re.IGNORECASE | re.MULTILINE, +) +_VAGUE_MUTATIONS_NONE_RE = re.compile( + r"^\s*[-*]?\s*mutations\s*:\s*none\s*$", + re.IGNORECASE | re.MULTILINE, +) + +_REQUIRED_CATEGORIES = ( + "file edits by reviewer", + "worktree/index mutations", + "git ref mutations", +) +_WORKTREE_FIELD_ALIASES = ( + "worktree/index mutations", + "worktree mutations", +) + + +def _has_category(text: str, category: str) -> bool: + pattern = re.compile( + rf"^\s*[-*]?\s*{re.escape(category)}\s*:", + re.IGNORECASE | re.MULTILINE, + ) + return bool(pattern.search(text)) + + +def _has_worktree_category(text: str) -> bool: + return any(_has_category(text, alias) for alias in _WORKTREE_FIELD_ALIASES) + + +def _normalize_for_consistency_check(text: str) -> str: + """Map #319 precise labels to #313 field names for consistency delegation.""" + normalized = text + if _has_category(text, "worktree/index mutations") and not _has_category( + text, "worktree mutations" + ): + normalized = re.sub( + r"(worktree/index mutations\s*:)", + "Worktree mutations:", + normalized, + flags=re.IGNORECASE, + ) + return normalized + + +def assess_mutation_categories_report( + report_text: str, + *, + handoff_session: dict | None = None, + observed_commands: list | None = None, +) -> dict[str, Any]: + """Require precise mutation categories in reviewer controller handoffs (#319).""" + text = report_text or "" + session = dict(handoff_session or {}) + reasons: list[str] = [] + lower = text.lower() + + is_controller = bool( + session.get("controller_handoff") + or _CONTROLLER_HANDOFF_RE.search(text) + ) + if not is_controller and not session.get("require_categories"): + return { + "proven": True, + "block": False, + "reasons": [], + "controller_handoff": False, + "safe_next_action": "proceed", + } + + if _WORKSPACE_MUTATIONS_RE.search(text): + reasons.append( + "controller handoffs must not use legacy 'Workspace mutations'; " + "use precise mutation category fields" + ) + + if _VAGUE_MUTATIONS_NONE_RE.search(text): + reasons.append( + "controller handoffs must not use vague 'Mutations: none'; " + "report each precise category separately" + ) + + missing = [ + category + for category in _REQUIRED_CATEGORIES + if category != "worktree/index mutations" and not _has_category(text, category) + ] + if not _has_worktree_category(text): + missing.append("worktree/index mutations") + + if missing: + reasons.append( + "controller handoff missing precise mutation categories: " + + ", ".join(missing) + ) + + commands = observed_commands or session.get("observed_commands") + if commands: + from review_proofs import assess_workspace_mutation_consistency + + consistency = assess_workspace_mutation_consistency( + _normalize_for_consistency_check(text), + commands, + ) + if not consistency.get("complete"): + reasons.extend(consistency.get("reasons") or []) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "controller_handoff": True, + "safe_next_action": ( + "replace Workspace mutations with file edits, worktree/index, git ref, " + "and other precise mutation category fields" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 9a96d17..ba4df08 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -134,3 +134,9 @@ def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof assert callable(assess_non_mergeable_skip_proof) + + +def test_mutation_categories_verifier_exported(): + from review_proofs import assess_mutation_categories_report + + assert callable(assess_mutation_categories_report) diff --git a/tests/test_reviewer_mutation_categories.py b/tests/test_reviewer_mutation_categories.py new file mode 100644 index 0000000..055b174 --- /dev/null +++ b/tests/test_reviewer_mutation_categories.py @@ -0,0 +1,88 @@ +"""Tests for precise mutation category verifier (#319).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_mutation_categories import assess_mutation_categories_report # noqa: E402 + + +def _precise_handoff() -> str: + return "\n".join([ + "Controller Handoff", + "File edits by reviewer: none", + "Worktree/index mutations: git worktree add branches/review-pr1", + "Git ref mutations: git fetch prgs master", + "MCP/Gitea mutations: gitea_view_pr", + "Review mutations: gitea_review_pr request_changes", + "Read-only diagnostics: git status, git diff", + ]) + + +class TestMutationCategories(unittest.TestCase): + def test_precise_categories_pass(self): + result = assess_mutation_categories_report(_precise_handoff()) + self.assertTrue(result["proven"], result["reasons"]) + + def test_legacy_workspace_mutations_blocks(self): + report = "\n".join([ + "Controller Handoff", + "Workspace mutations: none (no local file changes)", + "File edits by reviewer: none", + "Worktree/index mutations: none", + "Git ref mutations: none", + ]) + result = assess_mutation_categories_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"])) + + def test_workspace_none_with_worktree_command_blocks(self): + report = "\n".join([ + "Controller Handoff", + "Workspace mutations: none (no local file changes)", + "File edits by reviewer: none", + "Worktree/index mutations: none", + "Git ref mutations: none", + ]) + result = assess_mutation_categories_report( + report, + observed_commands=["git reset --hard FETCH_HEAD"], + ) + self.assertFalse(result["proven"]) + + def test_file_edits_none_with_worktree_mutation_passes(self): + report = "\n".join([ + "Controller Handoff", + "File edits by reviewer: none", + "Worktree/index mutations: git reset --hard FETCH_HEAD", + "Git ref mutations: git fetch prgs", + ]) + result = assess_mutation_categories_report( + report, + observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_missing_required_categories_blocks(self): + report = "Controller Handoff\nFile edits by reviewer: none\n" + result = assess_mutation_categories_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("missing" in r.lower() for r in result["reasons"])) + + def test_non_controller_handoff_skips(self): + result = assess_mutation_categories_report( + "Workspace mutations: none\n", + ) + self.assertTrue(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_mutation_categories_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 876f1ee3a3914a364f89efa3569b48c1017d68db Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:36:10 -0400 Subject: [PATCH 08/10] feat: hard-stop review mutations on already-landed PR heads (Closes #292) A reviewer workflow approved and attempted to merge PR #278 although its head commit was already an ancestor of master, ending in a Gitea 405 and manual reconciliation. Add the missing pre-mutation gate: - assess_already_landed_review_gate classifies the pinned candidate head against the fetched target branch: NORMAL_REVIEW_CANDIDATE, ALREADY_LANDED_RECONCILE_REQUIRED, or GATE_NOT_PROVEN (fail closed). Already-landed PRs block approval and the merge API, allow request-changes only for a real blocker, and require a reconciliation handoff (PR number/title, candidate head SHA, target branch + SHA, ancestor proof, linked issue status, recommended action, capability proof for close/comment mutations). Unchecked ancestry, invalid or missing SHAs, and a head changed since pinning all fail closed. Mergeability stays a separate gate. - assess_already_landed_report_state rejects APPROVED / MERGED / READY_TO_MERGE wording and missing ALREADY_LANDED_RECONCILE_REQUIRED state when the session's gate fired, complementing the text-marker rules from #327 which need the report to admit the landed state. Tests cover already-landed, normal, non-mergeable, changed-head, unchecked-ancestry, and invalid-SHA gate paths plus all report-state verdicts. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 166 ++++++++++++++++++++++++++++++++++++ tests/test_review_proofs.py | 136 +++++++++++++++++++++++++++++ 2 files changed, 302 insertions(+) diff --git a/review_proofs.py b/review_proofs.py index e7c94e2..3e28113 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3363,6 +3363,172 @@ def assess_reviewer_baseline_validation_proof( } +# --------------------------------------------------------------------------- +# Already-landed review gate (#292) +# --------------------------------------------------------------------------- + +ALREADY_LANDED_STATE = "ALREADY_LANDED_RECONCILE_REQUIRED" + +ALREADY_LANDED_HANDOFF_FIELDS = ( + "PR number/title", + "candidate head SHA", + "target branch", + "target branch SHA", + "ancestor proof", + "linked issue status", + "recommended reconciliation action", + "capability proof for close/comment mutations", +) + +_FORBIDDEN_LANDED_STATES_RE = re.compile( + r"review decision\s*:\s*approved?\b|" + r"merge result\s*:\s*merged\b|" + r"\bready[_ ]to[_ ]merge\b|" + r"\bmerge[d]?\s*:\s*success\b", + re.IGNORECASE, +) + + +def assess_already_landed_review_gate( + *, + pr_number: int | None, + candidate_head_sha: str | None, + target_branch: str | None, + target_branch_sha: str | None, + head_is_ancestor_of_target: bool | None, + live_head_sha: str | None = None, + real_blocker: bool = False, + mergeable: bool | None = None, +) -> dict: + """#292: hard gate before any review mutation on an already-landed PR. + + Runs after PR selection and head pinning, before approval, request- + changes, or the merge API. *head_is_ancestor_of_target* is the result + of an ancestry check of *candidate_head_sha* against the freshly + fetched target branch at *target_branch_sha* (e.g. ``git merge-base + --is-ancestor``). ``None`` means the check was not run — the gate then + fails closed. + + *live_head_sha*, when provided, must equal *candidate_head_sha*; a + changed head invalidates the pinned ancestry result until re-checked. + + *mergeable* is accepted for caller convenience but never consulted: + conflict/mergeability handling is a separate gate. + """ + del mergeable # ancestry gate only; mergeability is a separate gate + reasons: list[str] = [] + + if not isinstance(pr_number, int) or pr_number <= 0: + reasons.append("PR number missing or invalid (#292)") + if not _FULL_SHA.match((candidate_head_sha or "").strip()): + reasons.append( + "candidate head SHA is not a full 40-hex commit SHA (#292)" + ) + if not (target_branch or "").strip(): + reasons.append("target branch missing (#292)") + if not _FULL_SHA.match((target_branch_sha or "").strip()): + reasons.append( + "target branch SHA is not a full 40-hex commit SHA; fetch the " + "target branch and record its SHA before the ancestry check " + "(#292)" + ) + if live_head_sha is not None and candidate_head_sha and ( + live_head_sha.strip() != candidate_head_sha.strip() + ): + reasons.append( + "PR head changed since the ancestry check was pinned; re-fetch " + "and re-run the already-landed gate (#292)" + ) + if head_is_ancestor_of_target is None: + reasons.append( + "ancestry of the PR head against the target branch was not " + "checked; the already-landed gate must run before any review " + "mutation (#292)" + ) + + if reasons: + return { + "state": "GATE_NOT_PROVEN", + "approve_allowed": False, + "request_changes_allowed": False, + "merge_api_allowed": False, + "reconciliation_handoff_required": False, + "required_handoff_fields": (), + "reasons": reasons, + "safe_next_action": ( + "fetch the target branch, pin the live PR head, run the " + "ancestry check, then re-run this gate" + ), + } + + if head_is_ancestor_of_target: + return { + "state": ALREADY_LANDED_STATE, + "approve_allowed": False, + "request_changes_allowed": bool(real_blocker), + "merge_api_allowed": False, + "reconciliation_handoff_required": True, + "required_handoff_fields": ALREADY_LANDED_HANDOFF_FIELDS, + "reasons": [ + f"PR #{pr_number} head {candidate_head_sha} is already an " + f"ancestor of {target_branch} @ {target_branch_sha}; the PR " + "is reconciliation-only (#292)" + ], + "safe_next_action": ( + "stop before review mutation and emit an " + f"{ALREADY_LANDED_STATE} reconciliation handoff; any PR/issue " + "close or comment requires exact capability proof" + ), + } + + return { + "state": "NORMAL_REVIEW_CANDIDATE", + "approve_allowed": True, + "request_changes_allowed": True, + "merge_api_allowed": True, + "reconciliation_handoff_required": False, + "required_handoff_fields": (), + "reasons": [], + "safe_next_action": "proceed with the normal review workflow gates", + } + + +def assess_already_landed_report_state( + report_text: str | None, + *, + gate_fired: bool, +) -> dict: + """#292: reject APPROVED/MERGED/READY_TO_MERGE after the gate fired. + + *gate_fired* comes from the session's own gate result, so a report + that omits the already-landed markers entirely still fails closed — + unlike the text-only #327 wording rules this does not depend on the + report admitting the PR was already landed. + """ + if not gate_fired: + return {"complete": True, "block": False, "reasons": []} + + text = report_text or "" + reasons: list[str] = [] + + if _FORBIDDEN_LANDED_STATES_RE.search(text): + reasons.append( + "report claims APPROVED/MERGED/READY_TO_MERGE although the " + "already-landed gate fired (#292)" + ) + if ALREADY_LANDED_STATE.lower() not in text.lower(): + reasons.append( + f"report must state {ALREADY_LANDED_STATE} when the " + "already-landed gate fired (#292)" + ) + + return { + "complete": not reasons, + "block": bool(reasons), + "reasons": reasons, + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index ec411d2..4eefd47 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -23,6 +23,8 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par from review_proofs import ( # noqa: E402 ISSUE_SELECTION_CONTINUATION_EXPLICIT, ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, + assess_already_landed_report_state, + assess_already_landed_review_gate, assess_author_pr_report, assess_reviewer_baseline_validation_proof, assess_capability_evidence, @@ -2281,5 +2283,139 @@ class TestReviewerBaselineValidationProof(unittest.TestCase): self.assertFalse(report["baseline_validation_proven"]) +class TestAlreadyLandedReviewGate(unittest.TestCase): + """Issue #292: PR head already on target blocks approval and merge.""" + + def _gate(self, **overrides): + kwargs = { + "pr_number": 278, + "candidate_head_sha": PINNED, + "target_branch": "master", + "target_branch_sha": OTHER, + "head_is_ancestor_of_target": True, + "live_head_sha": PINNED, + } + kwargs.update(overrides) + return assess_already_landed_review_gate(**kwargs) + + def test_already_landed_blocks_approval_and_merge(self): + result = self._gate() + self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + self.assertFalse(result["request_changes_allowed"]) + self.assertTrue(result["reconciliation_handoff_required"]) + + def test_already_landed_with_real_blocker_allows_request_changes(self): + result = self._gate(real_blocker=True) + self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + self.assertTrue(result["request_changes_allowed"]) + + def test_normal_candidate_passes_gate(self): + result = self._gate(head_is_ancestor_of_target=False) + self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") + self.assertTrue(result["approve_allowed"]) + self.assertTrue(result["merge_api_allowed"]) + self.assertFalse(result["reconciliation_handoff_required"]) + + def test_non_mergeable_normal_pr_still_passes_this_gate(self): + # Mergeability/conflicts are separate gates; ancestry gate only + # decides already-landed vs normal candidate. + result = self._gate( + head_is_ancestor_of_target=False, mergeable=False) + self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") + self.assertTrue(result["request_changes_allowed"]) + + def test_unchecked_ancestry_blocks_review_mutations(self): + result = self._gate(head_is_ancestor_of_target=None) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["merge_api_allowed"]) + + def test_changed_head_since_pin_blocks_until_recheck(self): + result = self._gate( + head_is_ancestor_of_target=False, live_head_sha=OTHER) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + self.assertTrue(any( + "head" in reason.lower() for reason in result["reasons"] + )) + + def test_invalid_shas_block_gate(self): + result = self._gate(candidate_head_sha="deadbeef") + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + self.assertFalse(result["approve_allowed"]) + + result = self._gate(target_branch_sha=None) + self.assertEqual(result["state"], "GATE_NOT_PROVEN") + + def test_already_landed_handoff_lists_required_fields(self): + result = self._gate() + for field in ( + "PR number/title", + "candidate head SHA", + "target branch", + "target branch SHA", + "ancestor proof", + "linked issue status", + "recommended reconciliation action", + "capability proof for close/comment mutations", + ): + self.assertIn(field, result["required_handoff_fields"]) + + +class TestAlreadyLandedReportState(unittest.TestCase): + """Issue #292: reports cannot say APPROVED after the gate fired.""" + + GOOD_REPORT = ( + "Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" + "Candidate head SHA: " + PINNED + "\n" + "Review decision: none\n" + "Merge result: none\n" + ) + + def test_gate_fired_with_reconcile_state_passes(self): + result = assess_already_landed_report_state( + self.GOOD_REPORT, gate_fired=True) + self.assertTrue(result["complete"]) + self.assertFalse(result["block"]) + + def test_gate_fired_with_approved_state_blocks(self): + report = self.GOOD_REPORT.replace( + "Review decision: none", "Review decision: APPROVED") + result = assess_already_landed_report_state(report, gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_with_merged_state_blocks(self): + report = self.GOOD_REPORT.replace( + "Merge result: none", "Merge result: MERGED") + result = assess_already_landed_report_state(report, gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_with_ready_to_merge_blocks(self): + result = assess_already_landed_report_state( + self.GOOD_REPORT + "Current status: READY_TO_MERGE\n", + gate_fired=True) + self.assertTrue(result["block"]) + + def test_gate_fired_without_reconcile_state_blocks(self): + result = assess_already_landed_report_state( + "Current status: review complete.\n", gate_fired=True) + self.assertTrue(result["block"]) + self.assertTrue(any( + "already_landed_reconcile_required" in reason.lower() + for reason in result["reasons"] + )) + + def test_gate_not_fired_reports_pass_untouched(self): + result = assess_already_landed_report_state( + "Review decision: APPROVED\nMerge result: MERGED\n", + gate_fired=False) + self.assertTrue(result["complete"]) + self.assertFalse(result["block"]) + + if __name__ == "__main__": unittest.main() From be1462dc03ddd65c9e8f85ec8e0b6fd246ec9905 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:38:32 -0400 Subject: [PATCH 09/10] feat: reject eligible/reviewed wording for already-landed PRs (Closes #298) Add assess_already_landed_report_wording to review_proofs: when the already-landed gate fires, final reports and controller handoffs must use the reconciliation wording (Oldest open PR requiring action, Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED, Candidate head SHA, Reviewed head SHA: none, Review worktree used: false) and must not use the legacy fields (oldest/next eligible PR, Pinned reviewed head, Scratch worktree used) or claim a reviewed head SHA. In non-already-landed reports, a populated Pinned reviewed head or Reviewed head SHA fails unless validation and diff review actually passed this session. Fails closed. Tests cover the correct reconciliation wording, each forbidden legacy field, missing required fields, populated reviewed SHA, normal reviewed reports, blocked-infra reports, and validation-failed reports. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 96 +++++++++++++ tests/test_already_landed_report_wording.py | 151 ++++++++++++++++++++ 2 files changed, 247 insertions(+) create mode 100644 tests/test_already_landed_report_wording.py diff --git a/review_proofs.py b/review_proofs.py index 78f815d..9fb5dd7 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1171,6 +1171,102 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None): } +# ── Already-landed report wording (Issue #298) ─────────────────────────────── +# +# An already-landed PR is reconciliation-only: it must never be called +# eligible, its head SHA must never be called reviewed, and the legacy +# eligible/reviewed/scratch-worktree handoff fields must not appear. +# Reviewed-head claims in any report require validation + diff review to +# have actually passed. + +ALREADY_LANDED_FORBIDDEN_PHRASES = ( + "oldest eligible pr", + "next eligible pr", + "pinned reviewed head", + "scratch worktree used", +) + +ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED" + + +def assess_already_landed_report_wording(report_text, *, already_landed, + review_validated=False): + """#298: reject eligible/reviewed wording for already-landed PRs. + + *already_landed* states whether the already-landed gate fired for the + selected PR. *review_validated* states whether validation and diff + review actually passed this session; only then may a reviewed head + SHA be populated in a non-already-landed report. + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + text_lower = (report_text or "").lower() + fields = _report_labeled_fields(report_text) + reasons = [] + + reviewed_head = fields.get("reviewed head sha") + reviewed_head_populated = bool( + reviewed_head and not reviewed_head.strip().lower().startswith("none") + ) + pinned_head_present = "pinned reviewed head" in fields + + if already_landed: + for phrase in ALREADY_LANDED_FORBIDDEN_PHRASES: + if phrase in text_lower: + reasons.append( + f"already-landed report must not use '{phrase}'; the PR " + "is reconciliation-only, not review/merge eligible" + ) + + eligibility = fields.get("eligibility class", "") + if ALREADY_LANDED_ELIGIBILITY_CLASS not in eligibility.upper(): + reasons.append( + "already-landed report missing 'Eligibility class: " + f"{ALREADY_LANDED_ELIGIBILITY_CLASS}'" + ) + + for required in ("oldest open pr requiring action", + "candidate head sha"): + if required not in fields: + reasons.append( + f"already-landed report missing '{required}' field" + ) + + if "reviewed head sha" not in fields: + reasons.append( + "already-landed report must state 'Reviewed head SHA: none'" + ) + elif reviewed_head_populated: + reasons.append( + "already-landed report must not claim a reviewed head SHA; " + "no validation or diff review passed for this PR" + ) + + worktree_used = fields.get("review worktree used", "") + if worktree_used.strip().lower() not in ("false", "no"): + reasons.append( + "already-landed report must state 'Review worktree used: " + "false'" + ) + elif not review_validated: + if pinned_head_present: + reasons.append( + "'Pinned reviewed head' populated before validation and " + "diff review passed" + ) + if reviewed_head_populated: + reasons.append( + "reviewed head SHA populated before validation and diff " + "review passed" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None, justification=None): """Assess reviewer/author role separation for blind queue workflows. diff --git a/tests/test_already_landed_report_wording.py b/tests/test_already_landed_report_wording.py new file mode 100644 index 0000000..ac0379f --- /dev/null +++ b/tests/test_already_landed_report_wording.py @@ -0,0 +1,151 @@ +"""Tests for already-landed final-report wording validator (#298). + +An already-landed PR is reconciliation-only, never review/merge +eligible, and a head SHA may not be called reviewed unless validation +and diff review actually passed. Final reports and controller handoffs +must use the reconciliation wording, not the legacy eligible/reviewed +fields. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_already_landed_report_wording # noqa: E402 + + +GOOD_ALREADY_LANDED = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" +) + + +class TestAlreadyLandedWording(unittest.TestCase): + def test_correct_reconciliation_wording_passes(self): + result = assess_already_landed_report_wording( + GOOD_ALREADY_LANDED, already_landed=True + ) + self.assertTrue(result["complete"]) + self.assertFalse(result["downgraded"]) + self.assertEqual(result["reasons"], []) + + def test_oldest_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- oldest eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("oldest eligible pr" in r.lower() for r in result["reasons"]) + ) + + def test_next_eligible_wording_fails(self): + report = GOOD_ALREADY_LANDED + "- next eligible PR: PR #278\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_pinned_reviewed_head_fails(self): + report = GOOD_ALREADY_LANDED + "- Pinned reviewed head: 2a544b7\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("pinned reviewed head" in r.lower() for r in result["reasons"]) + ) + + def test_scratch_worktree_field_fails(self): + report = GOOD_ALREADY_LANDED + "- Scratch worktree used: False\n" + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + + def test_missing_eligibility_class_fails(self): + report = ( + "Controller Handoff\n" + "- Oldest open PR requiring action: PR #278\n" + "- Candidate head SHA: 2a544b7\n" + "- Reviewed head SHA: none\n" + "- Review worktree used: false\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("eligibility class" in r.lower() for r in result["reasons"]) + ) + + def test_populated_reviewed_head_sha_fails(self): + report = GOOD_ALREADY_LANDED.replace( + "- Reviewed head SHA: none\n", + "- Reviewed head SHA: 2a544b7\n", + ) + result = assess_already_landed_report_wording( + report, already_landed=True + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("reviewed head" in r.lower() for r in result["reasons"]) + ) + + +class TestNonAlreadyLandedReports(unittest.TestCase): + def test_normal_reviewed_report_passes(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Review decision: approve\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=True + ) + self.assertTrue(result["complete"]) + + def test_blocked_infra_report_with_pinned_head_fails(self): + report = ( + "- Selected PR: 281\n" + "- Pinned reviewed head: abc1234\n" + "- Current status: blocked on infra_stop\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("before validation" in r.lower() for r in result["reasons"]) + ) + + def test_validation_failed_report_with_reviewed_sha_fails(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: abc1234\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertFalse(result["complete"]) + + def test_validation_failed_report_with_reviewed_none_passes(self): + report = ( + "- Selected PR: 281\n" + "- Reviewed head SHA: none\n" + "- Validation: full suite failed\n" + ) + result = assess_already_landed_report_wording( + report, already_landed=False, review_validated=False + ) + self.assertTrue(result["complete"]) + + +if __name__ == "__main__": + unittest.main() From 275a362a4650e8c77186eaae70e6eb0cf28519d5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:41:33 -0400 Subject: [PATCH 10/10] feat: add issue claim heartbeat leases and stale-claim reconciliation (Closes #268) Structured claim/progress heartbeats post on issue claim and via gitea_post_heartbeat. Read-only gitea_reconcile_issue_claims inventories active, stale, phantom, and PR-backed claims; gitea_cleanup_stale_claims supports dry-run cleanup of reclaimable labels. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 275 ++++++++++++++++++++++++- issue_claim_heartbeat.py | 295 +++++++++++++++++++++++++++ task_capability_map.py | 12 ++ tests/test_issue_claim_heartbeat.py | 116 +++++++++++ tests/test_issue_write_tool_gates.py | 2 + tests/test_mcp_server.py | 4 +- 6 files changed, 702 insertions(+), 2 deletions(-) create mode 100644 issue_claim_heartbeat.py create mode 100644 tests/test_issue_claim_heartbeat.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index b5415eb..dbe1c2e 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -476,6 +476,7 @@ import task_capability_map # noqa: E402 import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 +import issue_claim_heartbeat # noqa: E402 import merged_cleanup_reconcile # noqa: E402 @@ -5146,6 +5147,41 @@ def gitea_audit_config() -> dict: return report +def _post_structured_issue_comment( + *, + issue_number: int, + body: str, + remote: str, + host: str | None, + org: str | None, + repo: str | None, + audit_op: str = "create_issue_comment", +) -> dict: + """Post an issue-thread comment after permission gates already passed.""" + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments" + with _audited( + audit_op, + host=h, + remote=remote, + org=o, + repo=r, + issue_number=issue_number, + request_metadata={"body_chars": len(body)}, + ): + data = api_request("POST", api, auth, {"body": body}) + result = { + "success": True, + "performed": True, + "comment_id": data["id"], + "issue_number": issue_number, + } + if _reveal_endpoints(): + result["url"] = data.get("html_url") + return result + + @mcp.tool() def gitea_mark_issue( issue_number: int, @@ -5155,6 +5191,8 @@ def gitea_mark_issue( org: str | None = None, repo: str | None = None, worktree_path: str | None = None, + branch_name: str | None = None, + profile: str | None = None, ) -> dict: """Claim or release an issue via the status:in-progress label. @@ -5204,7 +5242,31 @@ def gitea_mark_issue( request_metadata={"op": "add", "label": "status:in-progress"}): api_request("POST", f"{base}/issues/{issue_number}/labels", auth, {"labels": [label_id]}) - return {"success": True, "message": f"Issue #{issue_number} claimed."} + active_profile = (profile or get_profile().get("profile_name") or "unknown") + branch = (branch_name or "pending").strip() or "pending" + heartbeat_body = issue_claim_heartbeat.format_heartbeat_body( + kind="claim", + issue_number=issue_number, + branch=branch, + phase="claimed", + profile=active_profile, + next_action="create worktree and begin implementation", + ) + heartbeat = _post_structured_issue_comment( + issue_number=issue_number, + body=heartbeat_body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="claim_heartbeat", + ) + return { + "success": True, + "message": f"Issue #{issue_number} claimed.", + "heartbeat_posted": heartbeat.get("success", False), + "heartbeat_comment_id": heartbeat.get("comment_id"), + } else: with _audited("unlabel_issue", host=h, remote=remote, org=o, repo=r, issue_number=issue_number, @@ -5214,6 +5276,217 @@ def gitea_mark_issue( return {"success": True, "message": f"Issue #{issue_number} released."} +@mcp.tool() +def gitea_post_heartbeat( + issue_number: int, + branch: str, + phase: str, + pr: str = "none", + next_action: str = "none", + blocker: str = "none", + profile: str | None = None, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Post a structured progress heartbeat on a claimed issue (#268).""" + blocked = _profile_permission_block( + task_capability_map.required_permission("post_heartbeat")) + if blocked: + return blocked + verify_preflight_purity(remote) + active_profile = profile or get_profile().get("profile_name") + body = issue_claim_heartbeat.format_heartbeat_body( + kind="progress", + issue_number=issue_number, + branch=branch, + phase=phase, + profile=active_profile, + pr=pr, + next_action=next_action, + blocker=blocker, + ) + return _post_structured_issue_comment( + issue_number=issue_number, + body=body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="progress_heartbeat", + ) + + +@mcp.tool() +def gitea_reconcile_issue_claims( + state: str = "open", + stale_after_hours: int = 24, + heartbeat_lease_minutes: int = 30, + limit: int = 100, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Read-only inventory of issue claims and heartbeat lease status (#268).""" + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + issues = api_get_all(f"{base}/issues?state={state}&type=issues", auth, limit=limit) + open_prs = api_get_all(f"{base}/pulls?state=open", auth) + branches = api_get_all(f"{base}/branches", auth, limit=limit) + branch_names = [b.get("name") for b in branches if b.get("name")] + + comments_by_issue: dict[int, list[dict]] = {} + for issue in issues: + if not issue_claim_heartbeat.issue_has_in_progress_label(issue): + continue + number = int(issue["number"]) + api = f"{base}/issues/{number}/comments" + comments_by_issue[number] = api_request("GET", api, auth) or [] + + reclaim_after_minutes = max(heartbeat_lease_minutes * 2, 60) + inventory = issue_claim_heartbeat.build_claim_inventory( + issues=issues, + comments_by_issue=comments_by_issue, + open_prs=open_prs, + branch_names=branch_names, + heartbeat_lease_minutes=heartbeat_lease_minutes, + reclaim_after_minutes=reclaim_after_minutes, + ) + inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory) + inventory["success"] = True + inventory["performed"] = False + return inventory + + +@mcp.tool() +def gitea_cleanup_stale_claims( + dry_run: bool = True, + execute_confirmed: bool = False, + heartbeat_lease_minutes: int = 30, + limit: int = 100, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Propose or execute stale-claim cleanup for phantom/reclaimable issues (#268).""" + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "performed": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + + inventory = gitea_reconcile_issue_claims( + heartbeat_lease_minutes=heartbeat_lease_minutes, + limit=limit, + remote=remote, + host=host, + org=org, + repo=repo, + ) + if not inventory.get("success"): + return inventory + + plan = [ + entry + for entry in inventory.get("cleanup_plan") or [] + if entry.get("action") == "remove_status_in_progress_and_comment" + ] + report = { + "success": True, + "dry_run": dry_run, + "performed": False, + "planned_actions": plan, + "inventory_counts": inventory.get("counts"), + } + if dry_run: + return report + + if not execute_confirmed: + raise ValueError( + "execute_confirmed must be True when dry_run=False (fail closed)" + ) + + blocked = _profile_permission_block( + task_capability_map.required_permission("cleanup_stale_claims")) + if blocked: + return blocked + verify_preflight_purity(remote) + + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = repo_api_url(h, o, r) + labels = api_request("GET", f"{base}/labels?limit=100", auth) + label_id = next( + (lb["id"] for lb in labels if lb.get("name") == "status:in-progress"), + None, + ) + if label_id is None: + raise RuntimeError("Label 'status:in-progress' not found") + + actions: list[dict] = [] + active_profile = get_profile().get("profile_name") + for entry in plan: + issue_number = int(entry["issue_number"]) + with _audited( + "unlabel_issue", + host=h, + remote=remote, + org=o, + repo=r, + issue_number=issue_number, + request_metadata={"op": "cleanup_stale_claim", "label": "status:in-progress"}, + ): + api_request( + "DELETE", + f"{base}/issues/{issue_number}/labels/{label_id}", + auth, + ) + cleanup_body = issue_claim_heartbeat.format_heartbeat_body( + kind="cleanup", + issue_number=issue_number, + branch="none", + phase="stale-claim-cleanup", + profile=active_profile, + next_action="issue reclaimable by queue", + blocker=entry.get("status") or "stale", + ) + comment = _post_structured_issue_comment( + issue_number=issue_number, + body=cleanup_body, + remote=remote, + host=host, + org=org, + repo=repo, + audit_op="cleanup_heartbeat", + ) + actions.append( + { + "issue_number": issue_number, + "label_removed": True, + "cleanup_comment_id": comment.get("comment_id"), + } + ) + + report["performed"] = True + report["actions"] = actions + return report + + @mcp.tool() def gitea_list_labels( remote: str = "dadeschools", diff --git a/issue_claim_heartbeat.py b/issue_claim_heartbeat.py new file mode 100644 index 0000000..3d16e1d --- /dev/null +++ b/issue_claim_heartbeat.py @@ -0,0 +1,295 @@ +"""Issue claim heartbeat leases and stale-claim reconciliation (#268). + +Structured issue-thread comments prove live ownership beyond the +``status:in-progress`` label alone. Queue inventory can classify claims as +active, stale, reclaimable, PR-backed, or phantom. +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta, timezone +from typing import Any + +MARKER = "" +IN_PROGRESS_LABEL = "status:in-progress" + +_KIND_CLAIM = "claim" +_KIND_PROGRESS = "progress" +_KIND_CLEANUP = "cleanup" + +_FIELD_RE = re.compile( + r"^\s*-\s*([a-z_]+)\s*:\s*(.+?)\s*$", + re.IGNORECASE | re.MULTILINE, +) +_ISSUE_REF_RE = re.compile(r"issue-(\d+)", re.IGNORECASE) + + +def _parse_timestamp(value: str | None) -> datetime | None: + if not value: + return None + text = value.strip() + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed + + +def format_heartbeat_body( + *, + kind: str, + issue_number: int, + branch: str, + phase: str, + profile: str | None = None, + pr: str = "none", + next_action: str = "none", + blocker: str = "none", +) -> str: + """Return a structured, machine-parseable issue comment body.""" + profile_value = (profile or "unknown").strip() or "unknown" + lines = [ + MARKER, + "**Issue claim heartbeat**", + f"- kind: {kind}", + f"- issue: #{issue_number}", + f"- branch: {branch}", + f"- phase: {phase}", + f"- profile: {profile_value}", + f"- pr: {pr}", + f"- blocker: {blocker}", + f"- next_action: {next_action}", + ] + return "\n".join(lines) + + +def parse_heartbeat_comment(body: str) -> dict[str, Any] | None: + """Parse one structured heartbeat comment, or None when not a heartbeat.""" + text = body or "" + if MARKER not in text: + return None + fields: dict[str, str] = {} + for match in _FIELD_RE.finditer(text): + fields[match.group(1).strip().lower()] = match.group(2).strip() + if not fields: + return None + issue_raw = fields.get("issue", "") + issue_digits = re.sub(r"[^\d]", "", issue_raw) + issue_number = int(issue_digits) if issue_digits.isdigit() else None + return { + "kind": fields.get("kind"), + "issue_number": issue_number, + "branch": fields.get("branch"), + "phase": fields.get("phase"), + "profile": fields.get("profile"), + "pr": fields.get("pr"), + "blocker": fields.get("blocker"), + "next_action": fields.get("next_action"), + "raw_fields": fields, + } + + +def extract_issue_heartbeats( + comments: list[dict], + *, + issue_number: int | None = None, +) -> list[dict]: + """Return parsed heartbeats newest-last, optionally filtered to *issue_number*.""" + heartbeats: list[dict] = [] + for comment in comments or []: + parsed = parse_heartbeat_comment(comment.get("body") or "") + if not parsed: + continue + if issue_number is not None and parsed.get("issue_number") != issue_number: + continue + heartbeats.append( + { + **parsed, + "comment_id": comment.get("id"), + "author": (comment.get("user") or {}).get("login") + or comment.get("author"), + "created_at": comment.get("created_at"), + "updated_at": comment.get("updated_at"), + } + ) + return heartbeats + + +def issue_has_in_progress_label(issue: dict) -> bool: + labels = issue.get("labels") or [] + for label in labels: + name = label if isinstance(label, str) else label.get("name") + if name == IN_PROGRESS_LABEL: + return True + return False + + +def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None: + pattern = f"issue-{issue_number}" + closes = f"closes #{issue_number}" + fixes = f"fixes #{issue_number}" + for pr in open_prs or []: + head = (pr.get("head") or {}).get("ref") or "" + text = f"{pr.get('title', '')} {pr.get('body', '')}".lower() + if pattern in head.lower(): + return pr + if closes in text or fixes in text: + return pr + return None + + +def _matching_branch_names(issue_number: int, branch_names: list[str]) -> list[str]: + pattern = f"issue-{issue_number}" + return [name for name in branch_names if pattern in (name or "").lower()] + + +def classify_issue_claim( + *, + issue: dict, + comments: list[dict], + open_prs: list[dict] | None = None, + branch_names: list[str] | None = None, + now: datetime | None = None, + heartbeat_lease_minutes: int = 30, + reclaim_after_minutes: int = 60, +) -> dict[str, Any]: + """Classify one issue's claim state from labels, heartbeats, PRs, and branches.""" + issue_number = int(issue.get("number") or 0) + open_prs = open_prs or [] + branch_names = branch_names or [] + current = now or datetime.now(timezone.utc) + + has_label = issue_has_in_progress_label(issue) + heartbeats = extract_issue_heartbeats(comments, issue_number=issue_number) + latest = heartbeats[-1] if heartbeats else None + latest_at = _parse_timestamp( + (latest or {}).get("updated_at") or (latest or {}).get("created_at") + ) + age_minutes = None + if latest_at: + age_minutes = int((current - latest_at).total_seconds() // 60) + + linked_pr = _linked_open_pr(issue_number, open_prs) + matching_branches = _matching_branch_names(issue_number, branch_names) + + if not has_label: + status = "not_claimed" + reasons = ["issue lacks status:in-progress label"] + elif linked_pr: + status = "awaiting_review" + reasons = [f"open PR #{linked_pr.get('number')} covers this issue"] + elif not heartbeats: + status = "phantom" + reasons = [ + "status:in-progress label present without structured claim heartbeat" + ] + elif latest_at is None: + status = "phantom" + reasons = ["heartbeat comments present but timestamps could not be parsed"] + elif age_minutes is not None and age_minutes <= heartbeat_lease_minutes: + status = "active" + reasons = [f"heartbeat age {age_minutes}m within {heartbeat_lease_minutes}m lease"] + elif matching_branches and age_minutes is not None and age_minutes <= reclaim_after_minutes: + status = "active" + reasons = [ + f"matching branch(es) {', '.join(matching_branches)} with heartbeat " + f"age {age_minutes}m" + ] + elif age_minutes is not None and age_minutes > reclaim_after_minutes and not matching_branches: + status = "reclaimable" + reasons = [ + f"no heartbeat within {reclaim_after_minutes}m and no matching branch" + ] + elif age_minutes is not None and age_minutes > heartbeat_lease_minutes: + status = "stale" + reasons = [ + f"heartbeat age {age_minutes}m exceeds {heartbeat_lease_minutes}m lease" + ] + else: + status = "active" + reasons = ["claim has structured heartbeat proof"] + + return { + "issue_number": issue_number, + "title": issue.get("title"), + "status": status, + "has_in_progress_label": has_label, + "heartbeat_count": len(heartbeats), + "latest_heartbeat": latest, + "heartbeat_age_minutes": age_minutes, + "linked_open_pr": linked_pr.get("number") if linked_pr else None, + "matching_branches": matching_branches, + "reasons": reasons, + "reclaimable": status == "reclaimable", + "stale": status in {"stale", "phantom", "reclaimable"}, + } + + +def build_claim_inventory( + *, + issues: list[dict], + comments_by_issue: dict[int, list[dict]], + open_prs: list[dict], + branch_names: list[str], + now: datetime | None = None, + heartbeat_lease_minutes: int = 30, + reclaim_after_minutes: int = 60, +) -> dict[str, Any]: + """Build a queue inventory report for in-progress issue claims.""" + entries: list[dict] = [] + for issue in issues or []: + if not issue_has_in_progress_label(issue): + continue + number = int(issue["number"]) + entry = classify_issue_claim( + issue=issue, + comments=comments_by_issue.get(number, []), + open_prs=open_prs, + branch_names=branch_names, + now=now, + heartbeat_lease_minutes=heartbeat_lease_minutes, + reclaim_after_minutes=reclaim_after_minutes, + ) + entries.append(entry) + + counts: dict[str, int] = {} + for entry in entries: + counts[entry["status"]] = counts.get(entry["status"], 0) + 1 + + return { + "entries": entries, + "counts": counts, + "heartbeat_lease_minutes": heartbeat_lease_minutes, + "reclaim_after_minutes": reclaim_after_minutes, + "in_progress_total": len(entries), + } + + +def build_cleanup_plan(inventory: dict[str, Any]) -> list[dict]: + """Return stale/reclaimable/phantom claims that may be cleaned up.""" + plan: list[dict] = [] + for entry in inventory.get("entries") or []: + if entry.get("status") in {"reclaimable", "phantom"}: + plan.append( + { + "issue_number": entry["issue_number"], + "status": entry["status"], + "action": "remove_status_in_progress_and_comment", + "reasons": list(entry.get("reasons") or []), + } + ) + elif entry.get("status") == "stale" and not entry.get("linked_open_pr"): + plan.append( + { + "issue_number": entry["issue_number"], + "status": entry["status"], + "action": "report_only", + "reasons": list(entry.get("reasons") or []), + } + ) + return plan \ No newline at end of file diff --git a/task_capability_map.py b/task_capability_map.py index 9d162c1..2b794e7 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -96,6 +96,18 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + "post_heartbeat": { + "permission": "gitea.issue.comment", + "role": "author", + }, + "reconcile_issue_claims": { + "permission": "gitea.read", + "role": "author", + }, + "cleanup_stale_claims": { + "permission": "gitea.issue.comment", + "role": "author", + }, } # Issue-mutating MCP tools and their resolver task keys. diff --git a/tests/test_issue_claim_heartbeat.py b/tests/test_issue_claim_heartbeat.py new file mode 100644 index 0000000..c8a154e --- /dev/null +++ b/tests/test_issue_claim_heartbeat.py @@ -0,0 +1,116 @@ +"""Tests for issue claim heartbeat leases (#268).""" + +from __future__ import annotations + +import sys +import unittest +from datetime import datetime, timedelta, timezone +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import issue_claim_heartbeat as ich # noqa: E402 + + +class TestHeartbeatFormatting(unittest.TestCase): + def test_format_and_parse_roundtrip(self): + body = ich.format_heartbeat_body( + kind="claim", + issue_number=268, + branch="feat/issue-268-heartbeat-leases", + phase="claimed", + profile="prgs-author", + next_action="implement module", + ) + parsed = ich.parse_heartbeat_comment(body) + self.assertIsNotNone(parsed) + assert parsed is not None + self.assertEqual(parsed["issue_number"], 268) + self.assertEqual(parsed["branch"], "feat/issue-268-heartbeat-leases") + self.assertEqual(parsed["phase"], "claimed") + + +class TestClaimClassification(unittest.TestCase): + NOW = datetime(2026, 7, 7, 12, 0, tzinfo=timezone.utc) + + def _comment(self, *, minutes_ago: int, kind: str = "progress") -> dict: + ts = (self.NOW - timedelta(minutes=minutes_ago)).isoformat() + body = ich.format_heartbeat_body( + kind=kind, + issue_number=268, + branch="feat/issue-268-heartbeat-leases", + phase="implementation", + profile="prgs-author", + ) + return {"id": 1, "body": body, "created_at": ts, "updated_at": ts} + + def test_active_claim_within_lease(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=5)], + now=self.NOW, + ) + self.assertEqual(result["status"], "active") + + def test_stale_claim_without_branch(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=45)], + now=self.NOW, + ) + self.assertEqual(result["status"], "stale") + + def test_reclaimable_after_sixty_minutes_without_branch(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=90)], + now=self.NOW, + ) + self.assertEqual(result["status"], "reclaimable") + + def test_awaiting_review_when_open_pr_exists(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + open_prs = [ + { + "number": 400, + "title": "feat", + "body": "Closes #268", + "head": {"ref": "feat/issue-268-heartbeat-leases"}, + } + ] + result = ich.classify_issue_claim( + issue=issue, + comments=[self._comment(minutes_ago=120)], + open_prs=open_prs, + now=self.NOW, + ) + self.assertEqual(result["status"], "awaiting_review") + + def test_phantom_claim_without_heartbeat(self): + issue = {"number": 268, "title": "t", "labels": [{"name": "status:in-progress"}]} + result = ich.classify_issue_claim( + issue=issue, + comments=[{"id": 1, "body": "working on it"}], + now=self.NOW, + ) + self.assertEqual(result["status"], "phantom") + + +class TestInventory(unittest.TestCase): + def test_build_cleanup_plan_flags_phantom(self): + inventory = { + "entries": [ + {"issue_number": 1, "status": "phantom", "reasons": ["no heartbeat"]}, + {"issue_number": 2, "status": "active", "reasons": []}, + ] + } + plan = ich.build_cleanup_plan(inventory) + self.assertEqual(len(plan), 1) + self.assertEqual(plan[0]["issue_number"], 1) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_issue_write_tool_gates.py b/tests/test_issue_write_tool_gates.py index afe865d..736b295 100644 --- a/tests/test_issue_write_tool_gates.py +++ b/tests/test_issue_write_tool_gates.py @@ -172,6 +172,8 @@ class TestAllowedIssueWritesProceed(IssueWriteGateBase): return [{"id": 10, "name": "status:in-progress"}] if method == "POST" and "/issues/9/labels" in url: return [{"name": "status:in-progress"}] + if method == "POST" and "/issues/9/comments" in url: + return {"id": 501} if method == "GET" and url.endswith("/user"): return {"login": "author-user"} return {} diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 394ab03..99d7d73 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -321,15 +321,17 @@ class TestMarkIssue(unittest.TestCase): @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_start_adds_label(self, _auth, mock_api): - # First call: get labels; second call: add label + # labels, add label, post claim heartbeat comment mock_api.side_effect = [ [{"id": 10, "name": "status:in-progress"}], [{"name": "status:in-progress"}], + {"id": 99}, ] with patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True): result = gitea_mark_issue(issue_number=5, action="start") self.assertTrue(result["success"]) self.assertIn("claimed", result["message"]) + self.assertTrue(result.get("heartbeat_posted")) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)