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) <[email protected]>

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