Gitea-Tools/issue_work_duplicate_gate.py

"""Early duplicate-work detection for author work-issue sessions (#400)."""

from __future__ import annotations

from typing import Any

import issue_claim_heartbeat as claim_hb

PHASE_LOCK = "lock_issue"
PHASE_COMMIT = "commit"
PHASE_PUSH = "push"
PHASE_CREATE_PR = "create_pr"

OUTCOME_DUPLICATE_PR_PREVENTED = "duplicate_pr_prevented"
OUTCOME_DUPLICATE_BRANCH_PREVENTED = "duplicate_branch_prevented"
OUTCOME_DUPLICATE_COMMIT_PREVENTED = "duplicate_commit_prevented"
OUTCOME_DUPLICATE_WORK_NOT_PREVENTED = "duplicate_work_not_prevented"

_ACTIVE_CLAIM_STATUSES = frozenset({"active", "awaiting_review"})


def _issue_pattern(issue_number: int) -> str:
    return f"issue-{int(issue_number)}"


def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
    return claim_hb._linked_open_pr(issue_number, open_prs)


def _matching_branches(
    issue_number: int,
    branch_names: list[str],
    *,
    locked_branch: str | None = None,
) -> list[str]:
    pattern = _issue_pattern(issue_number)
    matches = [
        name for name in (branch_names or [])
        if pattern in (name or "").lower()
    ]
    if locked_branch:
        locked = locked_branch.strip()
        matches = [name for name in matches if name != locked]
    return matches


def assess_work_issue_duplicate_gate(
    issue_number: int,
    *,
    open_prs: list[dict] | None = None,
    branch_names: list[str] | None = None,
    claim_entry: dict | None = None,
    locked_branch: str | None = None,
    phase: str = PHASE_LOCK,
) -> dict[str, Any]:
    """Fail closed when duplicate work is already in flight for an issue."""
    reasons: list[str] = []
    outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED
    prs = list(open_prs or [])
    branches = list(branch_names or [])
    pattern = _issue_pattern(issue_number)

    linked = _linked_open_pr(issue_number, prs)
    if linked:
        reasons.append(
            f"open PR #{linked.get('number')} already covers issue "
            f"#{issue_number} (fail closed)"
        )
        outcome = OUTCOME_DUPLICATE_PR_PREVENTED

    conflicting_branches = _matching_branches(
        issue_number, branches, locked_branch=locked_branch
    )
    if conflicting_branches:
        names = ", ".join(conflicting_branches[:5])
        reasons.append(
            f"remote branch(es) already match issue pattern '{pattern}': "
            f"{names} (fail closed)"
        )
        if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
            outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED

    entry = claim_entry or {}
    if entry.get("linked_open_pr") and not linked:
        reasons.append(
            f"claim inventory reports open PR #{entry['linked_open_pr']} "
            f"for issue #{issue_number} (fail closed)"
        )
        outcome = OUTCOME_DUPLICATE_PR_PREVENTED

    status = (entry.get("status") or "").strip().lower()
    if status in _ACTIVE_CLAIM_STATUSES and not linked:
        heartbeat = entry.get("latest_heartbeat") or {}
        claim_branch = (heartbeat.get("branch") or "").strip()
        if locked_branch and claim_branch and claim_branch != locked_branch:
            reasons.append(
                f"active claim lease on branch '{claim_branch}' blocks "
                f"work on '{locked_branch}' for issue #{issue_number} "
                "(fail closed)"
            )
            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED
        elif not locked_branch and status == "active":
            reasons.append(
                f"issue #{issue_number} has an active claim lease "
                "(fail closed)"
            )
            if outcome == OUTCOME_DUPLICATE_WORK_NOT_PREVENTED:
                outcome = OUTCOME_DUPLICATE_BRANCH_PREVENTED

    if phase in {PHASE_COMMIT, PHASE_PUSH} and reasons:
        if outcome == OUTCOME_DUPLICATE_PR_PREVENTED:
            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED
        elif outcome == OUTCOME_DUPLICATE_BRANCH_PREVENTED:
            outcome = OUTCOME_DUPLICATE_COMMIT_PREVENTED

    block = bool(reasons)
    return {
        "block": block,
        "performed": not block,
        "issue_number": issue_number,
        "phase": phase,
        "outcome": outcome,
        "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"),
        "conflicting_branches": conflicting_branches,
        "claim_status": status or None,
        "reasons": reasons,
        "safe_next_action": (
            "stop before mutating; preserve local work and produce a "
            "reconciliation handoff if a concurrent PR appeared after push"
            if block and phase == PHASE_CREATE_PR
            else "stop before mutating; do not commit or push duplicate work"
            if block
            else "proceed"
        ),
    }


def assess_work_issue_duplicate_report(report_text: str) -> dict[str, Any]:
    """Require explicit duplicate-work outcome wording in work-issue reports."""
    text = (report_text or "").lower()
    markers = {
        OUTCOME_DUPLICATE_PR_PREVENTED: (
            "duplicate pr prevented",
            "duplicate_pr_prevented",
        ),
        OUTCOME_DUPLICATE_BRANCH_PREVENTED: (
            "duplicate branch prevented",
            "duplicate_branch_prevented",
        ),
        OUTCOME_DUPLICATE_COMMIT_PREVENTED: (
            "duplicate commit prevented",
            "duplicate_commit_prevented",
        ),
        OUTCOME_DUPLICATE_WORK_NOT_PREVENTED: (
            "duplicate work not prevented",
            "duplicate_work_not_prevented",
            "no duplicate work",
        ),
    }
    matched = [
        key for key, phrases in markers.items()
        if any(phrase in text for phrase in phrases)
    ]
    if len(matched) != 1:
        return {
            "complete": False,
            "downgraded": True,
            "reasons": [
                "work-issue report must state exactly one duplicate-work "
                "outcome (duplicate PR/branch/commit prevented, or "
                "duplicate work not prevented)"
            ],
        }
    return {
        "complete": True,
        "downgraded": False,
        "outcome": matched[0],
        "reasons": [],
    }