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)