Gitea-Tools/issue_claim_heartbeat.py

"""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