Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat(issue-filing): enforce pre-create duplicate title gate (#207)

Browse Source 
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) <[email protected]>
This commit is contained in:
sysadmin
2026-07-05 17:24:15 -04:00
co-authored by Claude Opus 4.8
parent 34a26d1c14
commit 75c176991e
7 changed files with 422 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_audit.py
+25 -7
View File
@@ -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"},
tests/test_issue_duplicate_gate.py
+173
View File
@@ -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()
tests/test_mcp_server.py
+12 -3
View File
@@ -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)
tests/test_role_session_router.py
+2 -1
View File
@@ -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(