Compare commits

...
Author SHA1 Message Date
sysadminandClaude Opus 4.8 7f97de1ed6 feat: gate author work on early duplicate-work detection (Closes #400)
Add author_duplicate_work_gate and enforce it at claim, lock, and PR
creation. Expose gitea_assess_author_duplicate_work for pre-commit/push
checks and extend work-issue final-report verification.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-07 13:08:10 -04:00
6 changed files with 501 additions and 42 deletions
+198
View File
@@ -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,
}
+158 -41
View File
@@ -503,6 +503,7 @@ import issue_lock_worktree # noqa: E402
import already_landed_reconcile # noqa: E402 import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402 import author_mutation_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402 import issue_claim_heartbeat # noqa: E402
import author_duplicate_work_gate # noqa: E402
import merged_cleanup_reconcile # noqa: E402 import merged_cleanup_reconcile # noqa: E402
import reconciler_profile # noqa: E402 import reconciler_profile # noqa: E402
import reconciliation_workflow # 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")) 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() @mcp.tool()
def gitea_lock_issue( def gitea_lock_issue(
issue_number: int, issue_number: int,
@@ -1111,48 +1182,17 @@ def gitea_lock_issue(
issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment) 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) h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h) auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls?state=open" _enforce_author_duplicate_work_gate(
issue_number,
try: "lock",
prs = api_get_all(url, auth) h=h,
except Exception as e: o=o,
raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}") r=r,
auth=auth,
for pr in prs: exclude_branch_name=branch_name,
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)"
)
work_lease = _build_author_issue_work_lease( work_lease = _build_author_issue_work_lease(
issue_number=issue_number, issue_number=issue_number,
@@ -1273,6 +1313,17 @@ def gitea_create_pr(
f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)" 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 # Check for forbidden terms anywhere in title/body
forbidden_terms = ["equivalent", "related", "same as"] forbidden_terms = ["equivalent", "related", "same as"]
text_to_check = f"{title} {body}".lower() 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)" 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" url = f"{repo_api_url(h, o, r)}/pulls"
payload = {"title": title, "body": body, "head": head, "base": base} payload = {"title": title, "body": body, "head": head, "base": base}
meta = {"title": title, "head": head, "base": base} meta = {"title": title, "head": head, "base": base}
@@ -6003,6 +6053,14 @@ def gitea_mark_issue(
) )
if action == "start": 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, with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
issue_number=issue_number, issue_number=issue_number,
request_metadata={"op": "add", "label": "status:in-progress"}): request_metadata={"op": "add", "label": "status:in-progress"}):
@@ -6084,6 +6142,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() @mcp.tool()
def gitea_reconcile_issue_claims( def gitea_reconcile_issue_claims(
state: str = "open", state: str = "open",
+19 -1
View File
@@ -3622,18 +3622,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: def assess_work_issue_final_report(report_text: str) -> dict:
"""#139: composite verifier for work-issue final reports.""" """#139: composite verifier for work-issue final reports."""
checks = { checks = {
"workflow_source": assess_work_issue_workflow_source(report_text), "workflow_source": assess_work_issue_workflow_source(report_text),
"mode_isolation": assess_work_issue_mode_isolation(report_text), "mode_isolation": assess_work_issue_mode_isolation(report_text),
"duplicate_prevention": assess_work_issue_duplicate_prevention_report(
report_text
),
} }
reasons = [] reasons = []
downgraded = False downgraded = False
for name, result in checks.items(): for name, result in checks.items():
verdict = result.get("verdict") 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 downgraded = True
reasons.extend(result.get("reasons") or []) reasons.extend(result.get("reasons") or [])
elif result.get("downgraded") or not result.get("complete", True): elif result.get("downgraded") or not result.get("complete", True):
@@ -3641,6 +3656,9 @@ def assess_work_issue_final_report(report_text: str) -> dict:
reasons.extend( reasons.extend(
f"{name}: {r}" for r in (result.get("reasons") or []) 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" grade = "A" if not downgraded else "downgraded"
return { return {
@@ -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. 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. If an open PR already exists for the issue, do not implement duplicate work.
Classify the issue as: Classify the issue as:
+102
View File
@@ -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()
+6
View File
@@ -95,6 +95,12 @@ def test_create_issue_workflow_contract():
assert "## 9. Duplicate search before mutation" in text 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(): def test_work_issue_workflow_contract():
text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8") text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
assert "canonical: true" in text assert "canonical: true" in text