Add server-side duplicate detection in gitea_create_issue that re-queries open and recently closed issues at mutation time, blocks normalized title matches unless the operator explicitly approves a split, and validates LLM duplicate-search summaries via review_proofs. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
170 lines
5.3 KiB
Python
170 lines
5.3 KiB
Python
"""Pre-create issue duplicate gate (#207)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
import unicodedata
|
|
|
|
VERDICT_NO_DUPLICATE = "no_duplicate_found"
|
|
VERDICT_DUPLICATE = "duplicate_found"
|
|
VERDICT_AMBIGUOUS = "ambiguous_duplicate_stop"
|
|
|
|
_STOPWORDS = frozenset({"a", "an", "the", "and", "or", "for", "to", "of", "in", "on"})
|
|
|
|
|
|
def normalize_issue_title(title: str) -> str:
|
|
"""Lowercase, punctuation-stripped, whitespace-collapsed title."""
|
|
text = unicodedata.normalize("NFKC", (title or "").strip().lower())
|
|
text = re.sub(r"[^\w\s]", " ", text)
|
|
text = re.sub(r"\s+", " ", text).strip()
|
|
return text
|
|
|
|
|
|
def _title_tokens(title: str) -> set[str]:
|
|
return {
|
|
t for t in normalize_issue_title(title).split()
|
|
if t and t not in _STOPWORDS
|
|
}
|
|
|
|
|
|
def titles_near_duplicate(proposed: str, existing: str) -> bool:
|
|
"""True when normalized titles match or are near-duplicates."""
|
|
norm_a = normalize_issue_title(proposed)
|
|
norm_b = normalize_issue_title(existing)
|
|
if not norm_a or not norm_b:
|
|
return False
|
|
if norm_a == norm_b:
|
|
return True
|
|
if norm_a in norm_b or norm_b in norm_a:
|
|
return True
|
|
tokens_a = _title_tokens(proposed)
|
|
tokens_b = _title_tokens(existing)
|
|
if not tokens_a or not tokens_b:
|
|
return False
|
|
overlap = tokens_a & tokens_b
|
|
union = tokens_a | tokens_b
|
|
ratio = len(overlap) / len(union)
|
|
if ratio >= 0.85:
|
|
return True
|
|
wall_pair = (
|
|
{"hard", "wall"} <= tokens_a and {"wall"} <= tokens_b
|
|
) or (
|
|
{"hard", "wall"} <= tokens_b and {"wall"} <= tokens_a
|
|
)
|
|
return wall_pair
|
|
|
|
|
|
def assess_pre_create_duplicate(
|
|
proposed_title: str,
|
|
existing_issues: list[dict],
|
|
*,
|
|
duplicate_override_reason: str | None = None,
|
|
split_from_issue: int | None = None,
|
|
) -> dict:
|
|
"""Evaluate duplicate risk immediately before issue creation."""
|
|
proposed_title = (proposed_title or "").strip()
|
|
if not proposed_title:
|
|
return {
|
|
"verdict": VERDICT_AMBIGUOUS,
|
|
"performed": False,
|
|
"reasons": ["issue title is required"],
|
|
"matches": [],
|
|
}
|
|
|
|
override = (duplicate_override_reason or "").strip()
|
|
if override and split_from_issue is not None:
|
|
return {
|
|
"verdict": VERDICT_NO_DUPLICATE,
|
|
"performed": True,
|
|
"override_applied": True,
|
|
"split_from_issue": split_from_issue,
|
|
"override_reason": override,
|
|
"reasons": [],
|
|
"matches": [],
|
|
}
|
|
|
|
matches = []
|
|
for issue in existing_issues or []:
|
|
existing_title = (issue.get("title") or "").strip()
|
|
if not existing_title:
|
|
continue
|
|
if titles_near_duplicate(proposed_title, existing_title):
|
|
matches.append({
|
|
"number": issue.get("number"),
|
|
"title": existing_title,
|
|
"state": issue.get("state"),
|
|
})
|
|
|
|
if not matches:
|
|
return {
|
|
"verdict": VERDICT_NO_DUPLICATE,
|
|
"performed": True,
|
|
"normalized_title": normalize_issue_title(proposed_title),
|
|
"reasons": [],
|
|
"matches": [],
|
|
}
|
|
|
|
if len(matches) > 3:
|
|
return {
|
|
"verdict": VERDICT_AMBIGUOUS,
|
|
"performed": False,
|
|
"normalized_title": normalize_issue_title(proposed_title),
|
|
"reasons": [
|
|
"too many near-duplicate title matches; fail closed "
|
|
"until operator clarifies"
|
|
],
|
|
"matches": matches[:5],
|
|
}
|
|
|
|
return {
|
|
"verdict": VERDICT_DUPLICATE,
|
|
"performed": False,
|
|
"normalized_title": normalize_issue_title(proposed_title),
|
|
"reasons": [
|
|
f"duplicate issue title blocked: matches existing "
|
|
f"#{m['number']} ({m['state']})"
|
|
for m in matches
|
|
],
|
|
"matches": matches,
|
|
}
|
|
|
|
|
|
def pre_create_issue_duplicate_gate(
|
|
proposed_title: str,
|
|
existing_issues: list[dict],
|
|
*,
|
|
duplicate_override_reason: str | None = None,
|
|
split_from_issue: int | None = None,
|
|
allow_override: bool = False,
|
|
) -> dict:
|
|
"""Alias for ``assess_pre_create_duplicate`` (#207 suggested name)."""
|
|
reason = (duplicate_override_reason or "").strip()
|
|
if allow_override and not reason:
|
|
reason = "operator-approved split after duplicate review"
|
|
return assess_pre_create_duplicate(
|
|
proposed_title,
|
|
existing_issues,
|
|
duplicate_override_reason=reason or None,
|
|
split_from_issue=split_from_issue,
|
|
)
|
|
|
|
|
|
def assess_duplicate_search_proof(report_text: str, matches: list[dict]) -> dict:
|
|
"""Reject LLM duplicate summaries that omit known exact duplicates (#207)."""
|
|
text = (report_text or "").lower()
|
|
missing = []
|
|
for match in matches or []:
|
|
num = match.get("number")
|
|
title = (match.get("title") or "").lower()
|
|
if num is not None and f"#{num}" not in text and str(num) not in text:
|
|
missing.append(f"issue #{num}")
|
|
if title and title[:40] not in text:
|
|
missing.append(f"title '{match.get('title')}'")
|
|
if missing:
|
|
return {
|
|
"valid": False,
|
|
"reasons": [
|
|
"duplicate-search proof omitted required match: " + ", ".join(missing)
|
|
],
|
|
}
|
|
return {"valid": True, "reasons": []} |