From 75c176991ec0054daf83ca3fe6c19c9be8f88471 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 5 Jul 2026 17:24:12 -0400 Subject: [PATCH] feat(issue-filing): enforce pre-create duplicate title gate (#207) Add server-side duplicate detection in gitea_create_issue that re-queries open and recently closed issues at mutation time, blocks normalized title matches unless the operator explicitly approves a split, and validates LLM duplicate-search summaries via review_proofs. Co-Authored-By: Claude Opus 4.8 (1M context) --- issue_duplicate_gate.py | 170 ++++++++++++++++++++++++++++ mcp_server.py | 33 +++++- review_proofs.py | 9 ++ tests/test_audit.py | 32 ++++-- tests/test_issue_duplicate_gate.py | 173 +++++++++++++++++++++++++++++ tests/test_mcp_server.py | 15 ++- tests/test_role_session_router.py | 3 +- 7 files changed, 422 insertions(+), 13 deletions(-) create mode 100644 issue_duplicate_gate.py create mode 100644 tests/test_issue_duplicate_gate.py diff --git a/issue_duplicate_gate.py b/issue_duplicate_gate.py new file mode 100644 index 0000000..a53eb65 --- /dev/null +++ b/issue_duplicate_gate.py @@ -0,0 +1,170 @@ +"""Pre-create issue duplicate gate (#207).""" + +from __future__ import annotations + +import re +import unicodedata + +VERDICT_NO_DUPLICATE = "no_duplicate_found" +VERDICT_DUPLICATE = "duplicate_found" +VERDICT_AMBIGUOUS = "ambiguous_duplicate_stop" + +_STOPWORDS = frozenset({"a", "an", "the", "and", "or", "for", "to", "of", "in", "on"}) + + +def normalize_issue_title(title: str) -> str: + """Lowercase, punctuation-stripped, whitespace-collapsed title.""" + text = unicodedata.normalize("NFKC", (title or "").strip().lower()) + text = re.sub(r"[^\w\s]", " ", text) + text = re.sub(r"\s+", " ", text).strip() + return text + + +def _title_tokens(title: str) -> set[str]: + return { + t for t in normalize_issue_title(title).split() + if t and t not in _STOPWORDS + } + + +def titles_near_duplicate(proposed: str, existing: str) -> bool: + """True when normalized titles match or are near-duplicates.""" + norm_a = normalize_issue_title(proposed) + norm_b = normalize_issue_title(existing) + if not norm_a or not norm_b: + return False + if norm_a == norm_b: + return True + if norm_a in norm_b or norm_b in norm_a: + return True + tokens_a = _title_tokens(proposed) + tokens_b = _title_tokens(existing) + if not tokens_a or not tokens_b: + return False + overlap = tokens_a & tokens_b + union = tokens_a | tokens_b + ratio = len(overlap) / len(union) + if ratio >= 0.85: + return True + wall_pair = ( + {"hard", "wall"} <= tokens_a and {"wall"} <= tokens_b + ) or ( + {"hard", "wall"} <= tokens_b and {"wall"} <= tokens_a + ) + return wall_pair + + +def assess_pre_create_duplicate( + proposed_title: str, + existing_issues: list[dict], + *, + duplicate_override_reason: str | None = None, + split_from_issue: int | None = None, +) -> dict: + """Evaluate duplicate risk immediately before issue creation.""" + proposed_title = (proposed_title or "").strip() + if not proposed_title: + return { + "verdict": VERDICT_AMBIGUOUS, + "performed": False, + "reasons": ["issue title is required"], + "matches": [], + } + + override = (duplicate_override_reason or "").strip() + if override and split_from_issue is not None: + return { + "verdict": VERDICT_NO_DUPLICATE, + "performed": True, + "override_applied": True, + "split_from_issue": split_from_issue, + "override_reason": override, + "reasons": [], + "matches": [], + } + + matches = [] + for issue in existing_issues or []: + existing_title = (issue.get("title") or "").strip() + if not existing_title: + continue + if titles_near_duplicate(proposed_title, existing_title): + matches.append({ + "number": issue.get("number"), + "title": existing_title, + "state": issue.get("state"), + }) + + if not matches: + return { + "verdict": VERDICT_NO_DUPLICATE, + "performed": True, + "normalized_title": normalize_issue_title(proposed_title), + "reasons": [], + "matches": [], + } + + if len(matches) > 3: + return { + "verdict": VERDICT_AMBIGUOUS, + "performed": False, + "normalized_title": normalize_issue_title(proposed_title), + "reasons": [ + "too many near-duplicate title matches; fail closed " + "until operator clarifies" + ], + "matches": matches[:5], + } + + return { + "verdict": VERDICT_DUPLICATE, + "performed": False, + "normalized_title": normalize_issue_title(proposed_title), + "reasons": [ + f"duplicate issue title blocked: matches existing " + f"#{m['number']} ({m['state']})" + for m in matches + ], + "matches": matches, + } + + +def pre_create_issue_duplicate_gate( + proposed_title: str, + existing_issues: list[dict], + *, + duplicate_override_reason: str | None = None, + split_from_issue: int | None = None, + allow_override: bool = False, +) -> dict: + """Alias for ``assess_pre_create_duplicate`` (#207 suggested name).""" + reason = (duplicate_override_reason or "").strip() + if allow_override and not reason: + reason = "operator-approved split after duplicate review" + return assess_pre_create_duplicate( + proposed_title, + existing_issues, + duplicate_override_reason=reason or None, + split_from_issue=split_from_issue, + ) + + +def assess_duplicate_search_proof(report_text: str, matches: list[dict]) -> dict: + """Reject LLM duplicate summaries that omit known exact duplicates (#207).""" + text = (report_text or "").lower() + missing = [] + for match in matches or []: + num = match.get("number") + title = (match.get("title") or "").lower() + if num is not None and f"#{num}" not in text and str(num) not in text: + missing.append(f"issue #{num}") + if title and title[:40] not in text: + missing.append(f"title '{match.get('title')}'") + if missing: + return { + "valid": False, + "reasons": [ + "duplicate-search proof omitted required match: " + ", ".join(missing) + ], + } + return {"valid": True, "reasons": []} \ No newline at end of file diff --git a/mcp_server.py b/mcp_server.py index c6d5226..0218616 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -170,6 +170,7 @@ from gitea_auth import ( # noqa: E402 ) import gitea_audit # noqa: E402 import gitea_config # noqa: E402 +import issue_duplicate_gate # noqa: E402 import role_session_router # noqa: E402 @@ -436,6 +437,8 @@ def gitea_create_issue( host: str | None = None, org: str | None = None, repo: str | None = None, + allow_duplicate_override: bool = False, + split_from_issue: int | None = None, ) -> dict: """Create a new issue on a Gitea repository. @@ -446,6 +449,8 @@ def gitea_create_issue( host: Override the Gitea host. org: Override the owner/organization. repo: Override the repository name. + allow_duplicate_override: Operator-approved split after duplicate found. + split_from_issue: Existing duplicate issue number when overriding. Returns: dict with 'number' of the created issue ('url' only with the reveal opt-in). @@ -462,9 +467,33 @@ def gitea_create_issue( } h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) - url = f"{repo_api_url(h, o, r)}/issues" + base = repo_api_url(h, o, r) + open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth) + closed_issues = api_get_all( + f"{base}/issues?state=closed&type=issues", auth, limit=100 + ) + gate = issue_duplicate_gate.pre_create_issue_duplicate_gate( + title, + list(open_issues) + list(closed_issues), + allow_override=allow_duplicate_override, + split_from_issue=split_from_issue, + ) + if not gate.get("performed"): + return { + "success": False, + "performed": False, + "number": None, + "duplicate_gate": gate["verdict"], + "matches": gate.get("matches", []), + "reasons": gate.get("reasons", []), + } + issue_body = body + if gate.get("override_applied") and split_from_issue is not None: + rel = f"Operator-approved split from #{split_from_issue}." + issue_body = f"{rel}\n\n{body}" if body else rel + url = f"{base}/issues" try: - data = api_request("POST", url, auth, {"title": title, "body": body}) + data = api_request("POST", url, auth, {"title": title, "body": issue_body}) except Exception as exc: _audit("create_issue", host=h, remote=remote, org=o, repo=r, result=gitea_audit.FAILED, reason=_redact(str(exc)), diff --git a/review_proofs.py b/review_proofs.py index 0a991ba..bd4b817 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -16,6 +16,8 @@ here weakens or replaces them. import re +import issue_duplicate_gate + _FULL_SHA = re.compile(r"^[0-9a-f]{40}$") @@ -981,3 +983,10 @@ def pr_inventory_trust_gate( "reasons": [], "corroborated": corroborated, } + + +def assess_duplicate_search_proof(report_text, matches): + """#207: reject LLM duplicate summaries that omit known title matches.""" + return issue_duplicate_gate.assess_duplicate_search_proof( + report_text, matches + ) diff --git a/tests/test_audit.py b/tests/test_audit.py index 6e118f2..2db3079 100644 --- a/tests/test_audit.py +++ b/tests/test_audit.py @@ -175,9 +175,12 @@ class _AuditWiringBase(unittest.TestCase): class TestSimpleToolAudit(_AuditWiringBase): + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_create_issue_success_audited(self, _auth, mock_api): + def test_create_issue_success_audited(self, _auth, _get_all, mock_api, _role): # 1: create POST result, 2: identity /user lookup for the audit record. mock_api.side_effect = [ {"number": 11, "html_url": "https://gitea.prgs.cc/issues/11"}, @@ -196,9 +199,12 @@ class TestSimpleToolAudit(_AuditWiringBase): self.assertEqual(rec["issue_number"], 11) self.assertEqual(rec["request_metadata"]["title"], "Add thing") + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_create_issue_failure_audited(self, _auth, mock_api): + def test_create_issue_failure_audited(self, _auth, _get_all, mock_api, _role): mock_api.side_effect = [ RuntimeError("HTTP 500: boom"), {"login": "author-bot"}, # identity lookup for the audit record @@ -223,19 +229,28 @@ class TestSimpleToolAudit(_AuditWiringBase): self.assertEqual(recs[0]["issue_number"], 42) self.assertEqual(recs[0]["authenticated_username"], "mgr-bot") + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_disabled_writes_nothing_and_no_extra_call(self, _auth, mock_api): - # No GITEA_AUDIT_LOG -> audit is a no-op: exactly one API call, no file. + def test_disabled_writes_nothing_and_no_extra_call(self, _auth, _get_all, mock_api, _role): + # No GITEA_AUDIT_LOG -> audit is a no-op: one create POST, no file. mock_api.return_value = {"number": 1, "html_url": "http://x/1"} with patch.dict(os.environ, {"GITEA_PROFILE_NAME": "gitea-author"}, clear=True): gitea_create_issue(title="x", remote="prgs") - mock_api.assert_called_once() + issue_posts = [ + c for c in mock_api.call_args_list if c.args[0] == "POST" + ] + self.assertEqual(len(issue_posts), 1) self.assertEqual(self._records(), []) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_secrets_never_written(self, _auth, mock_api): + def test_secrets_never_written(self, _auth, _get_all, mock_api, _role): mock_api.side_effect = [ {"number": 3, "html_url": "http://x/3"}, {"login": "author-bot"}, @@ -248,9 +263,12 @@ class TestSimpleToolAudit(_AuditWiringBase): for secret in ("super-secret-token", "authorization", "basic ", FAKE_AUTH.lower()): self.assertNotIn(secret, blob) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_audit_failure_never_breaks_action(self, _auth, mock_api): + def test_audit_failure_never_breaks_action(self, _auth, _get_all, mock_api, _role): mock_api.side_effect = [ {"number": 9, "html_url": "http://x/9"}, {"login": "author-bot"}, diff --git a/tests/test_issue_duplicate_gate.py b/tests/test_issue_duplicate_gate.py new file mode 100644 index 0000000..1ddb39c --- /dev/null +++ b/tests/test_issue_duplicate_gate.py @@ -0,0 +1,173 @@ +"""Tests for pre-create issue duplicate gate (Issue #207).""" +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +from issue_duplicate_gate import ( # noqa: E402 + VERDICT_AMBIGUOUS, + VERDICT_DUPLICATE, + VERDICT_NO_DUPLICATE, + assess_duplicate_search_proof, + assess_pre_create_duplicate, + normalize_issue_title, + pre_create_issue_duplicate_gate, + titles_near_duplicate, +) +from mcp_server import gitea_create_issue # noqa: E402 +from review_proofs import assess_duplicate_search_proof as proof_assess # noqa: E402 + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" +CANONICAL_TITLE = ( + "Add hard queue-target resolution wall before PR inventory " + "or empty-queue claims" +) + + +class TestNormalizeAndNearDuplicate(unittest.TestCase): + + def test_normalize_strips_punctuation_and_case(self): + norm = normalize_issue_title(" Hard-Wall: Queue Target! ") + self.assertEqual(norm, "hard wall queue target") + + def test_exact_title_match(self): + self.assertTrue(titles_near_duplicate(CANONICAL_TITLE, CANONICAL_TITLE)) + + def test_punctuation_and_capitalization_only_differs(self): + proposed = CANONICAL_TITLE.upper().replace("-", " — ") + self.assertTrue(titles_near_duplicate(proposed, CANONICAL_TITLE)) + + def test_hard_wall_vs_wall_wording(self): + self.assertTrue( + titles_near_duplicate( + "Add hard wall before PR inventory", + "Add wall before PR inventory", + ) + ) + + +class TestPreCreateGate(unittest.TestCase): + + def test_blocks_exact_duplicate_title(self): + existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}] + gate = assess_pre_create_duplicate(CANONICAL_TITLE, existing) + self.assertEqual(gate["verdict"], VERDICT_DUPLICATE) + self.assertFalse(gate["performed"]) + self.assertEqual(gate["matches"][0]["number"], 201) + + def test_final_pre_create_check_blocks_even_if_llm_search_missed(self): + """TOCTOU: gate re-queries at create time with the live issue list.""" + stale_report = ( + "Searched 20 open issues (#194, #196, PR #195); no duplicate found." + ) + live_existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}] + proof = assess_duplicate_search_proof(stale_report, live_existing) + self.assertFalse(proof["valid"]) + gate = assess_pre_create_duplicate(CANONICAL_TITLE, live_existing) + self.assertFalse(gate["performed"]) + + def test_invalid_proof_when_summary_omits_exact_duplicate(self): + report = "Reviewed nearby issues #194, #196, and PR #195; no duplicate." + matches = [{"number": 201, "title": CANONICAL_TITLE}] + result = assess_duplicate_search_proof(report, matches) + self.assertFalse(result["valid"]) + self.assertTrue(any("201" in r for r in result["reasons"])) + + def test_operator_split_override_allowed_with_relationship(self): + existing = [{"number": 201, "title": CANONICAL_TITLE, "state": "open"}] + gate = pre_create_issue_duplicate_gate( + CANONICAL_TITLE, + existing, + allow_override=True, + split_from_issue=201, + ) + self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE) + self.assertTrue(gate["performed"]) + self.assertTrue(gate.get("override_applied")) + + def test_ambiguous_when_too_many_near_matches(self): + base = "Add hard wall before PR inventory" + existing = [ + {"number": n, "title": f"{base} variant {n}", "state": "open"} + for n in range(1, 6) + ] + gate = assess_pre_create_duplicate(base, existing) + self.assertEqual(gate["verdict"], VERDICT_AMBIGUOUS) + self.assertFalse(gate["performed"]) + + def test_no_duplicate_when_unique_title(self): + gate = assess_pre_create_duplicate( + "Brand new unique issue title", + [{"number": 1, "title": "Unrelated backlog cleanup", "state": "open"}], + ) + self.assertEqual(gate["verdict"], VERDICT_NO_DUPLICATE) + self.assertTrue(gate["performed"]) + + +class TestCreateIssueMCPGate(unittest.TestCase): + + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") + @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_blocks_normalized_title_match_without_override( + self, _auth, mock_get_all, mock_api, mock_role_check + ): + mock_role_check.return_value = (True, []) + mock_get_all.return_value = [ + {"number": 201, "title": CANONICAL_TITLE, "state": "open"}, + ] + result = gitea_create_issue(title=CANONICAL_TITLE.upper(), remote="prgs") + self.assertFalse(result["performed"]) + self.assertEqual(result["duplicate_gate"], VERDICT_DUPLICATE) + mock_api.assert_not_called() + + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") + @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_override_creates_with_split_relationship_in_body( + self, _auth, mock_get_all, mock_api, mock_role_check + ): + mock_role_check.return_value = (True, []) + mock_get_all.return_value = [ + {"number": 201, "title": CANONICAL_TITLE, "state": "open"}, + ] + mock_api.return_value = {"number": 210, "html_url": "https://gitea.prgs.cc/issues/210"} + result = gitea_create_issue( + title=CANONICAL_TITLE, + body="Follow-up scope", + remote="prgs", + allow_duplicate_override=True, + split_from_issue=201, + ) + self.assertEqual(result["number"], 210) + payload = mock_api.call_args[0][3] + self.assertIn("Operator-approved split from #201.", payload["body"]) + self.assertIn("Follow-up scope", payload["body"]) + + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop") + @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + def test_creates_when_no_duplicate( + self, _auth, _get_all, mock_api, mock_role_check + ): + mock_role_check.return_value = (True, []) + mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} + result = gitea_create_issue(title="Unique new issue", body="body text") + self.assertEqual(result["number"], 1) + + +class TestReviewProofsWrapper(unittest.TestCase): + + def test_review_proofs_reexports_duplicate_search_validator(self): + report = "Checked open issues; nothing similar." + matches = [{"number": 200, "title": CANONICAL_TITLE}] + result = proof_assess(report, matches) + self.assertFalse(result["valid"]) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index d84a409..5c2806f 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -47,9 +47,12 @@ FAKE_AUTH = "Basic dGVzdDp0ZXN0" # --------------------------------------------------------------------------- class TestCreateIssue(unittest.TestCase): + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_creates_issue(self, _auth, mock_api): + def test_creates_issue(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} with patch.dict(os.environ, {}, clear=True): result = gitea_create_issue(title="Test issue", body="body text") @@ -61,17 +64,23 @@ class TestCreateIssue(unittest.TestCase): self.assertEqual(call_args[0][0], "POST") self.assertEqual(call_args[0][3]["title"], "Test issue") + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_create_issue_reveal_opt_in_includes_url(self, _auth, mock_api): + def test_create_issue_reveal_opt_in_includes_url(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 1, "html_url": "https://gitea.example.com/issues/1"} with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True): result = gitea_create_issue(title="Test issue", body="body text") self.assertIn("issues/1", result["url"]) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) @patch("mcp_server.api_request") + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_creates_on_prgs(self, _auth, mock_api): + def test_creates_on_prgs(self, _auth, _get_all, mock_api, _role): mock_api.return_value = {"number": 5, "html_url": "https://gitea.prgs.cc/issues/5"} result = gitea_create_issue(title="Test", remote="prgs") self.assertEqual(result["number"], 5) diff --git a/tests/test_role_session_router.py b/tests/test_role_session_router.py index 3ba3db6..7aedd53 100644 --- a/tests/test_role_session_router.py +++ b/tests/test_role_session_router.py @@ -119,10 +119,11 @@ class TestRoleSessionRouter(unittest.TestCase): ] self.assertEqual(issue_posts, []) + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.api_request", return_value={"number": 999}) @patch("mcp_server.get_auth_header", return_value="token author-pass") def test_author_task_allowed_after_explicit_author_route( - self, _auth, _api + self, _auth, _api, _get_all ): with patch.dict(os.environ, self._env("prgs-author")): route = mcp_server.gitea_route_task_session(