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
9 changed files with 495 additions and 603 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,
}
+152 -168
View File
@@ -503,7 +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 issue_work_duplicate_gate # 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
@@ -642,103 +642,6 @@ def _branch_entry_name(branch: dict | str) -> str:
return str(branch.get("name") or branch.get("ref") or "") 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: def _reveal_endpoints() -> bool:
"""Admin/debug opt-in (#120): include endpoint URLs and token source """Admin/debug opt-in (#120): include endpoint URLs and token source
names in tool output. Off by default so normal LLM-facing responses names in tool output. Off by default so normal LLM-facing responses
@@ -1147,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,
@@ -1211,19 +1184,15 @@ def gitea_lock_issue(
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h) auth = _auth(h)
duplicate_gate = _assess_issue_duplicate_gate( _enforce_author_duplicate_work_gate(
issue_number, issue_number,
"lock",
h=h, h=h,
o=o, o=o,
r=r, r=r,
auth=auth, auth=auth,
locked_branch=branch_name, exclude_branch_name=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( work_lease = _build_author_issue_work_lease(
issue_number=issue_number, issue_number=issue_number,
@@ -1269,39 +1238,6 @@ def gitea_lock_issue(
return result 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() @mcp.tool()
def gitea_create_pr( def gitea_create_pr(
title: str, title: str,
@@ -1377,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()
@@ -1393,22 +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)"
) )
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" 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}
@@ -3012,20 +2943,6 @@ def gitea_commit_files(
if blocked: if blocked:
return 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) verify_preflight_purity(remote)
processed_files, source_proofs = _prepare_commit_payload_files(files) processed_files, source_proofs = _prepare_commit_payload_files(files)
@@ -6136,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"}):
@@ -6217,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",
-180
View File
@@ -1,180 +0,0 @@
"""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": [],
}
+19 -4
View File
@@ -3622,21 +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."""
from issue_work_duplicate_gate import assess_work_issue_duplicate_report
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_work_outcome": assess_work_issue_duplicate_report(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):
@@ -3644,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:
@@ -297,36 +315,6 @@ 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. 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 ## 11. Claim or lock the issue before implementation
Claim/lock the issue before implementation if the project provides a claim/lock mechanism. Claim/lock the issue before implementation if the project provides a claim/lock mechanism.
+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()
-220
View File
@@ -1,220 +0,0 @@
"""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",
})
@patch("mcp_server.get_auth_header", return_value="token x")
def test_create_pr_returns_handoff_on_duplicate(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_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()
+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
-1
View File
@@ -2346,7 +2346,6 @@ class TestWorkIssueFinalReport(unittest.TestCase):
"- Safe next action: open PR", "- Safe next action: open PR",
"- Next: open PR", "- Next: open PR",
"- Safety statement: no review/merge", "- Safety statement: no review/merge",
"- Duplicate work outcome: duplicate work not prevented",
]) ])
def test_complete_work_issue_report_earns_a(self): def test_complete_work_issue_report_earns_a(self):