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

feat: move duplicate-work detection before author mutations (Closes #400)

Browse Source 
Extract issue_work_duplicate_gate for open PR, remote branch, and active
claim checks; wire it into lock_issue, commit_files recheck, and create_pr
recheck. Add gitea_assess_work_issue_duplicate read-only preflight, work-issue
report outcome validation, and workflow section 10A.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 13:08:21 -04:00
co-authored by Claude Opus 4.8
parent 1a1e679246
commit cf057d7829
6 changed files with 606 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+173 -40
View File
@@ -503,6 +503,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
@@ -641,6 +642,103 @@ def _branch_entry_name(branch: dict | str) -> str:
return str(branch.get("name") or branch.get("ref") or "")
def _collect_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
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
@@ -1111,48 +1209,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,
@@ -1198,6 +1269,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,
@@ -1289,6 +1393,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}
@@ -2893,6 +3012,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)
issue_work_duplicate_gate.py
+180
View File
@@ -0,0 +1,180 @@
"""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": [],
}
review_proofs.py
+3
View File
@@ -3624,9 +3624,12 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict:
def assess_work_issue_final_report(report_text: str) -> dict:
"""#139: composite verifier for work-issue final reports."""
from issue_work_duplicate_gate import assess_work_issue_duplicate_report
checks = {
"workflow_source": assess_work_issue_workflow_source(report_text),
"mode_isolation": assess_work_issue_mode_isolation(report_text),
"duplicate_work_outcome": assess_work_issue_duplicate_report(report_text),
}
reasons = []
skills/llm-project-workflow/workflows/work-issue.md
+30
View File
@@ -297,6 +297,36 @@ Report:
Do not create another branch/PR for the same issue unless the project explicitly allows taking over or updating existing work and exact capability is proven.
### 10A. Duplicate-work gate phases (#400)
Before any file edits, prove duplicate-work clearance with
`gitea_assess_work_issue_duplicate` or `gitea_lock_issue` (which runs the same
gate). The gate checks live:
* open PRs linked to the issue (head branch or Closes/Fixes reference),
* remote branches matching `issue-<number>`,
* active claim leases from structured heartbeats.
Re-check immediately before:
* `gitea_commit_files` (commit),
* branch push,
* `gitea_create_pr` (PR creation).
If a concurrent open PR appears after work begins:
* before commit/push → stop and preserve local work without pushing,
* after commit but before push → stop without pushing,
* after push but before PR creation → stop and produce a reconciliation
handoff instead of opening a PR.
Final reports must state exactly one duplicate-work outcome:
* `duplicate PR prevented`
* `duplicate branch prevented`
* `duplicate commit prevented`
* `duplicate work not prevented`
## 11. Claim or lock the issue before implementation
Claim/lock the issue before implementation if the project provides a claim/lock mechanism.
tests/test_issue_work_duplicate_gate.py
+219
View File
@@ -0,0 +1,219 @@
"""Tests for early duplicate-work detection (#400)."""
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import issue_work_duplicate_gate as dup_gate
import mcp_server
from issue_work_duplicate_gate import (
OUTCOME_DUPLICATE_BRANCH_PREVENTED,
OUTCOME_DUPLICATE_COMMIT_PREVENTED,
OUTCOME_DUPLICATE_PR_PREVENTED,
OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
PHASE_COMMIT,
PHASE_CREATE_PR,
PHASE_LOCK,
assess_work_issue_duplicate_gate,
assess_work_issue_duplicate_report,
)
class TestDuplicateGateAssessment(unittest.TestCase):
def test_clear_issue_passes(self):
result = assess_work_issue_duplicate_gate(
400,
open_prs=[],
branch_names=["feat/other-issue-99"],
claim_entry={"status": "not_claimed"},
locked_branch="feat/issue-400-duplicate-work-preflight",
phase=PHASE_LOCK,
)
self.assertFalse(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
def test_open_pr_blocks(self):
prs = [{
"number": 397,
"title": "feat: handoff",
"body": "Closes #395",
"head": {"ref": "feat/issue-395-proof-backed-review-handoff"},
}]
result = assess_work_issue_duplicate_gate(
395,
open_prs=prs,
branch_names=[],
phase=PHASE_LOCK,
)
self.assertTrue(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
def test_conflicting_remote_branch_blocks(self):
result = assess_work_issue_duplicate_gate(
395,
open_prs=[],
branch_names=["feat/issue-395-proof-backed-handoff-claims"],
locked_branch="feat/issue-395-new-attempt",
phase=PHASE_LOCK,
)
self.assertTrue(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_BRANCH_PREVENTED)
def test_active_claim_on_other_branch_blocks(self):
result = assess_work_issue_duplicate_gate(
398,
open_prs=[],
branch_names=[],
claim_entry={
"status": "active",
"latest_heartbeat": {
"branch": "feat/issue-398-validation-cwd-proof",
},
},
locked_branch="feat/issue-398-other-branch",
phase=PHASE_LOCK,
)
self.assertTrue(result["block"])
def test_commit_phase_maps_to_commit_outcome(self):
prs = [{
"number": 411,
"title": "x",
"body": "Closes #398",
"head": {"ref": "feat/issue-398-validation-cwd-proof"},
}]
result = assess_work_issue_duplicate_gate(
398,
open_prs=prs,
branch_names=[],
locked_branch="feat/issue-398-alt",
phase=PHASE_COMMIT,
)
self.assertTrue(result["block"])
self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_COMMIT_PREVENTED)
def test_stale_claim_does_not_block_by_status_alone(self):
result = assess_work_issue_duplicate_gate(
400,
open_prs=[],
branch_names=[],
claim_entry={"status": "reclaimable", "reasons": ["stale"]},
locked_branch="feat/issue-400-duplicate-work-preflight",
phase=PHASE_LOCK,
)
self.assertFalse(result["block"])
class TestDuplicateReportOutcome(unittest.TestCase):
def test_requires_exactly_one_outcome(self):
bad = assess_work_issue_duplicate_report("work finished")
self.assertFalse(bad["complete"])
good = assess_work_issue_duplicate_report(
"Duplicate work not prevented for issue #400."
)
self.assertTrue(good["complete"])
self.assertEqual(good["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
class TestMcpDuplicateRecheck(unittest.TestCase):
def setUp(self):
self._dir = tempfile.TemporaryDirectory()
self.lock_path = os.path.join(self._dir.name, "gitea_issue_lock.json")
self._lock_patch = patch.object(
mcp_server, "ISSUE_LOCK_FILE", self.lock_path
)
self._lock_patch.start()
self._remotes = patch.dict(mcp_server.REMOTES, {
"prgs": {"host": "gitea.example.com", "org": "Example-Org",
"repo": "Example-Repo"},
})
self._remotes.start()
mcp_server._IDENTITY_CACHE.clear()
def tearDown(self):
patch.stopall()
self._dir.cleanup()
def _write_lock(self, issue_number=400, branch="feat/issue-400-x"):
with open(self.lock_path, "w", encoding="utf-8") as fh:
json.dump({
"issue_number": issue_number,
"branch_name": branch,
"remote": "prgs",
}, fh)
@patch("mcp_server._assess_issue_duplicate_gate")
@patch("mcp_server.get_profile", return_value={
"profile_name": "test-author",
"allowed_operations": ["gitea.read", "gitea.repo.commit"],
"forbidden_operations": [],
"audit_label": "test-author",
})
@patch("mcp_server.get_auth_header", return_value="token x")
def test_commit_files_blocked_on_recheck(self, _auth, _profile, mock_gate):
self._write_lock()
mock_gate.return_value = {
"block": True,
"reasons": ["open PR #412 already covers issue #400"],
"outcome": OUTCOME_DUPLICATE_COMMIT_PREVENTED,
"safe_next_action": "stop",
}
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with patch(
"mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
):
result = mcp_server.gitea_commit_files(
files=[{
"operation": "create",
"path": "a.txt",
"content_plain": "hi",
}],
message="test",
remote="prgs",
)
self.assertFalse(result["success"])
self.assertIn("duplicate_gate", result)
@patch("mcp_server._assess_issue_duplicate_gate")
@patch("mcp_server.get_profile", return_value={
"profile_name": "test-author",
"allowed_operations": ["gitea.read", "gitea.pr.create"],
"forbidden_operations": [],
"audit_label": "test-author",
})
def test_create_pr_returns_handoff_on_duplicate(self, _profile, mock_gate):
self._write_lock()
mock_gate.return_value = {
"block": True,
"reasons": ["open PR #412 already covers issue #400"],
"outcome": OUTCOME_DUPLICATE_PR_PREVENTED,
"safe_next_action": "reconciliation handoff",
}
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check("capability", resolved_role="author")
with patch(
"mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []),
):
result = mcp_server.gitea_create_pr(
title="feat: x (Closes #400)",
head="feat/issue-400-x",
base="master",
body="Closes #400",
remote="prgs",
)
self.assertFalse(result["success"])
self.assertIsNone(result.get("number"))
self.assertIn("duplicate_gate", result)
if __name__ == "__main__":
unittest.main()
tests/test_review_proofs.py
+1
View File
@@ -2346,6 +2346,7 @@ class TestWorkIssueFinalReport(unittest.TestCase):
"- Safe next action: open PR",
"- Next: open PR",
"- Safety statement: no review/merge",
"- Duplicate work outcome: duplicate work not prevented",
])
def test_complete_work_issue_report_earns_a(self):