Merge pull request 'feat: add issue claim heartbeat leases and stale-claim reconciliation (Closes #268)' (#370) from feat/issue-268-heartbeat-leases into master

This commit was merged in pull request #370.
This commit is contained in:
2026-07-07 08:36:55 -05:00
6 changed files with 702 additions and 2 deletions
+274 -1
View File
@@ -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",
+295
View File
@@ -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 = "<!-- gitea-issue-claim-heartbeat:v1 -->"
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
+12
View File
@@ -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.
+116
View File
@@ -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()
+2
View File
@@ -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 {}
+3 -1
View File
@@ -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)