Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

Merge pull request 'feat: move duplicate-work detection before author mutations (Closes #400)' (#413) from feat/issue-400-duplicate-work-preflight into master

Browse Source
This commit was merged in pull request #413.
This commit is contained in:
sysadmin
2026-07-07 15:40:23 -05:00
parent 89a7d4dbfc e4adccd82a
commit d95ea323d7
10 changed files with 711 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+190 -41
View File
@@ -541,6 +541,7 @@ import issue_lock_worktree # noqa: E402
import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402
import reconciliation_workflow # noqa: E402
@@ -585,7 +586,7 @@ def _load_existing_issue_lock() -> dict | None:
if not os.path.exists(ISSUE_LOCK_FILE):
return None
try:
with open(ISSUE_LOCK_FILE, encoding="utf-8") as f:
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
data = json.load(f)
return data if isinstance(data, dict) else None
except Exception:
@@ -679,6 +680,119 @@ def _branch_entry_name(branch: dict | str) -> str:
return str(branch.get("name") or branch.get("ref") or "")
def _live_fetch_issue_duplicate_context(
h: str,
o: str,
r: str,
auth: str,
issue_number: int,
) -> tuple[list[dict], list[str], dict]:
"""Live open PRs, remote branch names, and claim state for one issue."""
base = repo_api_url(h, o, r)
open_prs = api_get_all(f"{base}/pulls?state=open", auth)
branches = api_get_all(f"{base}/branches", auth)
branch_names = [_branch_entry_name(b) for b in branches]
issue = api_request("GET", f"{base}/issues/{issue_number}", auth) or {}
comments = api_request(
"GET", f"{base}/issues/{issue_number}/comments", auth
) or []
claim_entry = issue_claim_heartbeat.classify_issue_claim(
issue=issue,
comments=comments,
open_prs=open_prs,
branch_names=branch_names,
)
return open_prs, branch_names, claim_entry
# Injectable duplicate-work context fetcher (#400). Production uses the live
# Gitea API path above; unit tests patch this symbol instead of hitting the
# network.
issue_duplicate_context_fetcher = _live_fetch_issue_duplicate_context
def _collect_issue_duplicate_context(
h: str,
o: str,
r: str,
auth: str,
issue_number: int,
) -> tuple[list[dict], list[str], dict]:
return issue_duplicate_context_fetcher(h, o, r, auth, issue_number)
def _assess_issue_duplicate_gate(
issue_number: int,
*,
h: str,
o: str,
r: str,
auth: str,
locked_branch: str | None = None,
phase: str,
) -> dict:
open_prs, branch_names, claim_entry = _collect_issue_duplicate_context(
h, o, r, auth, issue_number
)
return issue_work_duplicate_gate.assess_work_issue_duplicate_gate(
issue_number,
open_prs=open_prs,
branch_names=branch_names,
claim_entry=claim_entry,
locked_branch=locked_branch,
phase=phase,
)
def _duplicate_gate_block_response(gate: dict, **extra) -> dict:
out = {
"success": False,
"performed": False,
"reasons": list(gate.get("reasons") or []),
"duplicate_gate": gate,
"safe_next_action": gate.get("safe_next_action"),
}
out.update(extra)
return out
def _enforce_locked_issue_duplicate_recheck(
remote: str,
phase: str,
*,
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict | None:
"""Re-check duplicate-work gates for the locked issue (#400)."""
lock_data = _load_existing_issue_lock()
if not lock_data:
return None
issue_number = int(lock_data.get("issue_number") or 0)
locked_branch = lock_data.get("branch_name")
if not issue_number:
return None
h, o, r = _resolve(
remote or lock_data.get("remote") or "dadeschools",
host or lock_data.get("host"),
org or lock_data.get("org"),
repo or lock_data.get("repo"),
)
auth = _auth(h)
gate = _assess_issue_duplicate_gate(
issue_number,
h=h,
o=o,
r=r,
auth=auth,
locked_branch=locked_branch,
phase=phase,
)
if gate.get("block"):
return gate
return None
def _reveal_endpoints() -> bool:
"""Admin/debug opt-in (#120): include endpoint URLs and token source
names in tool output. Off by default so normal LLM-facing responses
@@ -1151,48 +1265,21 @@ def gitea_lock_issue(
issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
)
# 2. Check if the issue already has an open PR (reuse protection)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls?state=open"
try:
prs = api_get_all(url, auth)
except Exception as e:
raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
for pr in prs:
pr_head = pr.get("head", {}).get("ref", "")
pr_title = pr.get("title", "")
pr_body = pr.get("body", "")
if expected_pattern in pr_head:
raise ValueError(
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
)
patterns = [
f"closes #{issue_number}",
f"fixes #{issue_number}",
]
text_to_check = f"{pr_title} {pr_body}".lower()
if any(p in text_to_check for p in patterns):
raise ValueError(
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
)
branch_url = f"{repo_api_url(h, o, r)}/branches"
try:
branches = api_get_all(branch_url, auth)
except Exception as e:
raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
for branch in branches:
name = _branch_entry_name(branch)
if expected_pattern in name:
raise ValueError(
f"Issue #{issue_number} already has matching branch '{name}' "
"(fail closed)"
)
duplicate_gate = _assess_issue_duplicate_gate(
issue_number,
h=h,
o=o,
r=r,
auth=auth,
locked_branch=branch_name,
phase=issue_work_duplicate_gate.PHASE_LOCK,
)
if duplicate_gate.get("block"):
raise ValueError("; ".join(duplicate_gate.get("reasons") or [
f"duplicate work gate blocked issue #{issue_number} (fail closed)"
]))
work_lease = _build_author_issue_work_lease(
issue_number=issue_number,
@@ -1238,6 +1325,39 @@ def gitea_lock_issue(
return result
@mcp.tool()
def gitea_assess_work_issue_duplicate(
issue_number: int,
branch_name: str | None = None,
phase: str = issue_work_duplicate_gate.PHASE_LOCK,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Read-only duplicate-work gate for author sessions before mutations (#400)."""
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"),
}
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
gate = _assess_issue_duplicate_gate(
issue_number,
h=h,
o=o,
r=r,
auth=auth,
locked_branch=branch_name,
phase=phase,
)
return {"success": not gate.get("block"), **gate}
@mcp.tool()
def gitea_create_pr(
title: str,
@@ -1329,6 +1449,21 @@ def gitea_create_pr(
f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
)
duplicate_block = _enforce_locked_issue_duplicate_recheck(
remote,
issue_work_duplicate_gate.PHASE_CREATE_PR,
host=host,
org=org,
repo=repo,
)
if duplicate_block:
return _duplicate_gate_block_response(
duplicate_block,
number=None,
issue_number=locked_issue,
branch_name=locked_branch,
)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls"
payload = {"title": title, "body": body, "head": head, "base": base}
@@ -2933,6 +3068,20 @@ def gitea_commit_files(
if blocked:
return blocked
duplicate_block = _enforce_locked_issue_duplicate_recheck(
remote,
issue_work_duplicate_gate.PHASE_COMMIT,
host=host,
org=org,
repo=repo,
)
if duplicate_block:
return _duplicate_gate_block_response(
duplicate_block,
commit="",
branch="",
)
verify_preflight_purity(remote)
processed_files, source_proofs = _prepare_commit_payload_files(files)