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]>

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()

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