fix: resolve conflicts for PR #385 against latest master

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()

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 {}

tests/test_llm_workflow_split.py

@@ -150,6 +150,12 @@ def test_merge_simulation_verifier_exported():
     assert callable(assess_merge_simulation_report)
 
 
+def test_validation_integrity_verifier_exported():
+    from review_proofs import assess_validation_integrity_report
+
+    assert callable(assess_validation_integrity_report)
+
+
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 
@@ -168,6 +174,12 @@ def test_validation_worktree_edit_verifier_exported():
     assert callable(assess_validation_worktree_edit_report)
 
 
+def test_mutation_categories_verifier_exported():
+    from review_proofs import assess_mutation_categories_report
+
+    assert callable(assess_mutation_categories_report)
+
+
 def test_worktree_ownership_verifier_exported():
     from review_proofs import assess_worktree_ownership_report
 

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)

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()

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()