Gitea-Tools/author_duplicate_work_gate.py

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

from __future__ import annotations

import re
from typing import Any

from issue_claim_heartbeat import (
    _linked_open_pr,
    _matching_branch_names,
    classify_issue_claim,
)

STAGES = (
    "claim",
    "lock",
    "worktree",
    "edit",
    "commit",
    "push",
    "create_pr",
)

ELIGIBILITY_OPEN_PR_EXISTS = "OPEN_PR_EXISTS"
ELIGIBILITY_DUPLICATE_BRANCH_EXISTS = "DUPLICATE_BRANCH_EXISTS"
ELIGIBILITY_ACTIVE_CLAIM = "ACTIVE_CLAIM_BY_OTHER"
ELIGIBILITY_CLEAR = "CLEAR"


def assess_author_duplicate_work(
    issue_number: int,
    *,
    stage: str,
    open_prs: list[dict] | None = None,
    branch_names: list[str] | None = None,
    claim_entry: dict | None = None,
    matching_branches: list[str] | None = None,
    allow_stale_takeover: bool = False,
) -> dict[str, Any]:
    """Fail closed when duplicate work is detected before author mutations (#400)."""
    stage_norm = (stage or "").strip().lower()
    if stage_norm not in STAGES:
        return {
            "allowed": False,
            "block": True,
            "eligibility_class": "INVALID_STAGE",
            "stage": stage_norm or None,
            "reasons": [f"unknown duplicate-work stage {stage!r}"],
            "safe_next_action": f"use one of: {', '.join(STAGES)}",
        }

    prs = list(open_prs or [])
    linked_pr = _linked_open_pr(int(issue_number), prs)
    branches = list(
        matching_branches
        if matching_branches is not None
        else _matching_branch_names(int(issue_number), list(branch_names or []))
    )

    reasons: list[str] = []
    eligibility = ELIGIBILITY_CLEAR

    if linked_pr:
        eligibility = ELIGIBILITY_OPEN_PR_EXISTS
        reasons.append(
            f"open PR #{linked_pr.get('number')} already covers issue "
            f"#{issue_number}"
        )

    if branches and stage_norm in {"claim", "lock", "worktree", "edit"}:
        if eligibility == ELIGIBILITY_CLEAR:
            eligibility = ELIGIBILITY_DUPLICATE_BRANCH_EXISTS
        reasons.append(
            f"remote branch(es) already exist for issue #{issue_number}: "
            f"{', '.join(branches)}"
        )

    entry = claim_entry or {}
    claim_status = (entry.get("status") or "").strip()
    if claim_status in {"active", "awaiting_review"} and stage_norm == "claim":
        if entry.get("reclaimable") and allow_stale_takeover:
            pass
        elif claim_status == "active" and not entry.get("reclaimable"):
            if eligibility == ELIGIBILITY_CLEAR:
                eligibility = ELIGIBILITY_ACTIVE_CLAIM
            reasons.append(
                f"issue #{issue_number} has active claim "
                f"(status={claim_status})"
            )
    elif claim_status == "awaiting_review" and stage_norm == "claim":
        if not linked_pr:
            reasons.append(
                f"issue #{issue_number} is awaiting_review but no linked open PR "
                "was supplied for duplicate-work proof"
            )

    allowed = not reasons
    outcome = "duplicate_work_prevented" if not allowed else "clear"
    if stage_norm == "create_pr" and not allowed:
        outcome = "duplicate_pr_prevented"
    elif stage_norm in {"commit", "push"} and not allowed:
        outcome = "duplicate_push_prevented" if stage_norm == "push" else "duplicate_commit_prevented"

    return {
        "allowed": allowed,
        "block": not allowed,
        "eligibility_class": eligibility if not allowed else ELIGIBILITY_CLEAR,
        "stage": stage_norm,
        "linked_open_pr": linked_pr.get("number") if linked_pr else None,
        "matching_branches": branches,
        "claim_status": claim_status or None,
        "outcome": outcome,
        "reasons": reasons,
        "safe_next_action": (
            "stop without edits/commit/push/PR; produce reconciliation handoff "
            "preserving local work only"
            if not allowed
            else "proceed"
        ),
    }


def build_claim_entry_from_classification(classification: dict) -> dict:
    """Map ``classify_issue_claim`` output to gate claim metadata."""
    return {
        "status": classification.get("status"),
        "reclaimable": classification.get("reclaimable"),
        "linked_open_pr": classification.get("linked_open_pr"),
    }


_DUPLICATE_OUTCOME_RE = re.compile(
    r"(duplicate\s+(?:pr|branch|commit|push)\s+prevented|"
    r"duplicate\s+work\s+not\s+prevented|reconciliation\s+handoff)",
    re.IGNORECASE,
)


def assess_work_issue_duplicate_prevention_report(report_text: str) -> dict:
    """#400: work-issue reports must state duplicate-work prevention outcome."""
    text = report_text or ""
    if "duplicate work" not in text.lower() and "duplicate pr" not in text.lower():
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }
    if _DUPLICATE_OUTCOME_RE.search(text):
        return {
            "proven": True,
            "block": False,
            "reasons": [],
            "safe_next_action": "proceed",
        }
    return {
        "proven": False,
        "block": True,
        "reasons": [
            "duplicate-work discussion must name a prevention outcome "
            "(duplicate PR/branch/commit/push prevented, or duplicate work "
            "not prevented, or reconciliation handoff)"
        ],
        "safe_next_action": "state exact duplicate-work prevention class in final report",
    }


def classify_and_assess(
    issue: dict,
    *,
    stage: str,
    comments: list[dict] | None = None,
    open_prs: list[dict] | None = None,
    branch_names: list[str] | None = None,
    allow_stale_takeover: bool = False,
) -> dict[str, Any]:
    """Combine claim classification with duplicate-work gate assessment."""
    issue_number = int(issue.get("number") or 0)
    claim = classify_issue_claim(
        issue=issue,
        comments=comments or [],
        open_prs=open_prs or [],
        branch_names=branch_names or [],
    )
    assessment = assess_author_duplicate_work(
        issue_number,
        stage=stage,
        open_prs=open_prs,
        branch_names=branch_names,
        claim_entry=build_claim_entry_from_classification(claim),
        matching_branches=claim.get("matching_branches"),
        allow_stale_takeover=allow_stale_takeover,
    )
    return {
        "issue_number": issue_number,
        "claim": claim,
        "duplicate_work": assessment,
    }