diff --git a/author_duplicate_work_gate.py b/author_duplicate_work_gate.py new file mode 100644 index 0000000..fcea83a --- /dev/null +++ b/author_duplicate_work_gate.py @@ -0,0 +1,198 @@ +"""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, + } \ No newline at end of file diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 40b4b86..5e0990d 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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 author_duplicate_work_gate # noqa: E402 import merged_cleanup_reconcile # noqa: E402 import reconciler_profile # noqa: E402 import reconciliation_workflow # noqa: E402 @@ -1049,6 +1050,76 @@ def gitea_create_issue( return _with_optional_url({"number": data["number"]}, data.get("html_url")) +def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]: + branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit) + return [_branch_entry_name(branch) for branch in branches] + + +def _gather_author_duplicate_work_context( + issue_number: int, + *, + h: str, + o: str, + r: str, + auth: str, + exclude_branch_name: str | None = None, +) -> dict: + base = repo_api_url(h, o, r) + issue = api_request("GET", f"{base}/issues/{issue_number}", auth) + comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or [] + open_prs = api_get_all(f"{base}/pulls?state=open", auth) + branch_names = _list_repo_branch_names(h, o, r, auth) + if exclude_branch_name: + branch_names = [ + name for name in branch_names + if name != exclude_branch_name + ] + return { + "issue": issue, + "comments": comments, + "open_prs": open_prs, + "branch_names": branch_names, + } + + +def _enforce_author_duplicate_work_gate( + issue_number: int, + stage: str, + *, + h: str, + o: str, + r: str, + auth: str, + allow_stale_takeover: bool = False, + exclude_branch_name: str | None = None, +) -> dict: + """Fail closed when duplicate work is detected (#400).""" + ctx = _gather_author_duplicate_work_context( + issue_number, + h=h, + o=o, + r=r, + auth=auth, + exclude_branch_name=exclude_branch_name, + ) + result = author_duplicate_work_gate.classify_and_assess( + ctx["issue"], + stage=stage, + comments=ctx["comments"], + open_prs=ctx["open_prs"], + branch_names=ctx["branch_names"], + allow_stale_takeover=allow_stale_takeover, + ) + assessment = result.get("duplicate_work") or {} + if assessment.get("block"): + reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"]) + raise RuntimeError( + f"Author duplicate-work gate (#400) blocked at stage '{stage}': " + f"{reasons} (fail closed)" + ) + return result + + @mcp.tool() def gitea_lock_issue( issue_number: int, @@ -1111,48 +1182,17 @@ 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)" - ) + _enforce_author_duplicate_work_gate( + issue_number, + "lock", + h=h, + o=o, + r=r, + auth=auth, + exclude_branch_name=branch_name, + ) work_lease = _build_author_issue_work_lease( issue_number=issue_number, @@ -1245,7 +1285,7 @@ def gitea_create_pr( ) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, worktree_path=worktree_path) h, o, r = _resolve(remote, host, org, repo) # ── Issue Lock Validation (Issue #194 / #196) ── @@ -1273,6 +1313,17 @@ def gitea_create_pr( f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)" ) + auth = _auth(h) + _enforce_author_duplicate_work_gate( + int(locked_issue), + "create_pr", + h=h, + o=o, + r=r, + auth=auth, + exclude_branch_name=locked_branch, + ) + # Check for forbidden terms anywhere in title/body forbidden_terms = ["equivalent", "related", "same as"] text_to_check = f"{title} {body}".lower() @@ -1289,7 +1340,6 @@ 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)" ) - auth = _auth(h) url = f"{repo_api_url(h, o, r)}/pulls" payload = {"title": title, "body": body, "head": head, "base": base} meta = {"title": title, "head": head, "base": base} @@ -5985,6 +6035,14 @@ def gitea_mark_issue( ) if action == "start": + _enforce_author_duplicate_work_gate( + issue_number, + "claim", + h=h, + o=o, + r=r, + auth=auth, + ) with _audited("label_issue", host=h, remote=remote, org=o, repo=r, issue_number=issue_number, request_metadata={"op": "add", "label": "status:in-progress"}): @@ -6066,6 +6124,65 @@ def gitea_post_heartbeat( ) +@mcp.tool() +def gitea_assess_author_duplicate_work( + issue_number: int, + stage: str, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, + allow_stale_takeover: bool = False, + exclude_branch_name: str | None = None, +) -> dict: + """Read-only: assess duplicate-work risk before author mutations (#400). + + Call at claim, lock, worktree, edit, commit, push, and create_pr stages. + """ + read_block = _profile_operation_gate("gitea.read") + if read_block: + return { + "success": False, + "reasons": read_block, + "permission_report": _permission_block_report("gitea.read"), + } + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + ctx = _gather_author_duplicate_work_context( + issue_number, + h=h, + o=o, + r=r, + auth=auth, + exclude_branch_name=exclude_branch_name, + ) + result = author_duplicate_work_gate.classify_and_assess( + ctx["issue"], + stage=stage, + comments=ctx["comments"], + open_prs=ctx["open_prs"], + branch_names=ctx["branch_names"], + allow_stale_takeover=allow_stale_takeover, + ) + assessment = result.get("duplicate_work") or {} + return { + "success": True, + "performed": False, + "issue_number": issue_number, + "stage": stage, + "allowed": assessment.get("allowed"), + "block": assessment.get("block"), + "eligibility_class": assessment.get("eligibility_class"), + "outcome": assessment.get("outcome"), + "linked_open_pr": assessment.get("linked_open_pr"), + "matching_branches": assessment.get("matching_branches"), + "claim_status": assessment.get("claim_status"), + "reasons": assessment.get("reasons"), + "safe_next_action": assessment.get("safe_next_action"), + "claim": result.get("claim"), + } + + @mcp.tool() def gitea_reconcile_issue_claims( state: str = "open", diff --git a/review_proofs.py b/review_proofs.py index c7b027e..fe7bf56 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3612,18 +3612,33 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict: } +def assess_work_issue_duplicate_prevention_report(report_text, **kwargs): + """#400: work-issue reports must classify duplicate-work prevention.""" + from author_duplicate_work_gate import ( + assess_work_issue_duplicate_prevention_report as _assess, + ) + + return _assess(report_text, **kwargs) + + def assess_work_issue_final_report(report_text: str) -> dict: """#139: composite verifier for work-issue final reports.""" checks = { "workflow_source": assess_work_issue_workflow_source(report_text), "mode_isolation": assess_work_issue_mode_isolation(report_text), + "duplicate_prevention": assess_work_issue_duplicate_prevention_report( + report_text + ), } reasons = [] downgraded = False for name, result in checks.items(): verdict = result.get("verdict") - if verdict in ("missing", "incomplete"): + if result.get("block"): + downgraded = True + reasons.extend(result.get("reasons") or []) + elif verdict in ("missing", "incomplete"): downgraded = True reasons.extend(result.get("reasons") or []) elif result.get("downgraded") or not result.get("complete", True): @@ -3631,6 +3646,9 @@ def assess_work_issue_final_report(report_text: str) -> dict: reasons.extend( f"{name}: {r}" for r in (result.get("reasons") or []) ) + elif result.get("proven") is False: + downgraded = True + reasons.extend(result.get("reasons") or []) grade = "A" if not downgraded else "downgraded" return { diff --git a/skills/llm-project-workflow/workflows/work-issue.md b/skills/llm-project-workflow/workflows/work-issue.md index f450cd7..7cca7f0 100644 --- a/skills/llm-project-workflow/workflows/work-issue.md +++ b/skills/llm-project-workflow/workflows/work-issue.md @@ -277,6 +277,24 @@ Do not select an issue based only on memory from a previous session. Before claiming or working on an issue, check whether there is already an open PR, branch, or active claim for that issue. +Run `gitea_assess_author_duplicate_work` at these stages and stop when `block` is true: + +* `claim` — before `gitea_mark_issue` +* `lock` — before `gitea_lock_issue` +* `worktree` / `edit` — before creating a worktree or editing files +* `commit` — immediately before `git commit` +* `push` — immediately before `git push` +* `create_pr` — immediately before `gitea_create_pr` (also enforced server-side) + +`gitea_mark_issue`, `gitea_lock_issue`, and `gitea_create_pr` enforce the same gate server-side and fail closed. + +If a concurrent open PR appears after work begins: + +* before commit or push — stop and preserve local work without pushing +* after push but before PR creation — produce a reconciliation handoff instead of opening a PR + +Final reports must name the duplicate-work outcome (`duplicate PR prevented`, `duplicate branch prevented`, `duplicate commit prevented`, `duplicate push prevented`, `duplicate work not prevented`, or `reconciliation handoff`). + If an open PR already exists for the issue, do not implement duplicate work. Classify the issue as: diff --git a/tests/test_author_duplicate_work_gate.py b/tests/test_author_duplicate_work_gate.py new file mode 100644 index 0000000..75aeba9 --- /dev/null +++ b/tests/test_author_duplicate_work_gate.py @@ -0,0 +1,102 @@ +"""Tests for early author duplicate-work gate (#400).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from author_duplicate_work_gate import ( # noqa: E402 + ELIGIBILITY_OPEN_PR_EXISTS, + assess_author_duplicate_work, + assess_work_issue_duplicate_prevention_report, +) + + +def _open_pr(number: int = 397, issue: int = 395) -> dict: + return { + "number": number, + "head": {"ref": f"feat/issue-{issue}-example"}, + "title": f"feat: example (Closes #{issue})", + "body": f"Closes #{issue}", + } + + +class TestAuthorDuplicateWorkGate(unittest.TestCase): + def test_clear_when_no_duplicates(self): + result = assess_author_duplicate_work( + 400, + stage="claim", + open_prs=[], + branch_names=["master", "feat/issue-399-other"], + ) + self.assertTrue(result["allowed"]) + self.assertFalse(result["block"]) + + def test_open_pr_blocks_claim(self): + result = assess_author_duplicate_work( + 395, + stage="claim", + open_prs=[_open_pr()], + branch_names=[], + ) + self.assertFalse(result["allowed"]) + self.assertEqual(result["eligibility_class"], ELIGIBILITY_OPEN_PR_EXISTS) + + def test_matching_branch_blocks_lock_not_create_pr(self): + branches = ["feat/issue-400-early-duplicate-work-gate"] + lock = assess_author_duplicate_work( + 400, + stage="lock", + open_prs=[], + branch_names=branches, + ) + self.assertFalse(lock["allowed"]) + + create_pr = assess_author_duplicate_work( + 400, + stage="create_pr", + open_prs=[], + branch_names=branches, + matching_branches=[], + ) + self.assertTrue(create_pr["allowed"]) + + def test_open_pr_blocks_create_pr_stage(self): + result = assess_author_duplicate_work( + 395, + stage="create_pr", + open_prs=[_open_pr()], + branch_names=["feat/issue-395-proof-backed-review-handoff"], + ) + self.assertFalse(result["allowed"]) + self.assertEqual(result["outcome"], "duplicate_pr_prevented") + + def test_push_stage_blocks_on_concurrent_pr(self): + result = assess_author_duplicate_work( + 395, + stage="push", + open_prs=[_open_pr()], + branch_names=[], + ) + self.assertFalse(result["allowed"]) + self.assertEqual(result["outcome"], "duplicate_push_prevented") + + def test_duplicate_prevention_report_requires_outcome(self): + bad = assess_work_issue_duplicate_prevention_report( + "Duplicate work detected for issue #395." + ) + self.assertFalse(bad["proven"]) + + good = assess_work_issue_duplicate_prevention_report( + "Duplicate PR prevented; reconciliation handoff produced." + ) + self.assertTrue(good["proven"]) + + def test_exported_from_review_proofs(self): + from review_proofs import assess_work_issue_duplicate_prevention_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 9ec5b3d..3551305 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -95,6 +95,12 @@ def test_create_issue_workflow_contract(): assert "## 9. Duplicate search before mutation" in text +def test_author_duplicate_work_gate_exported(): + from review_proofs import assess_work_issue_duplicate_prevention_report + + assert callable(assess_work_issue_duplicate_prevention_report) + + def test_work_issue_workflow_contract(): text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8") assert "canonical: true" in text