diff --git a/review_proofs.py b/review_proofs.py index 73985b2..31b0b65 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -664,8 +664,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination, report_text, review_decision_lock ) - empty_queue_report = assess_empty_queue_report(report_text) - contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) validation_claimable = bool(validation.get("claimable")) @@ -789,11 +787,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "reviewer worktree safety proof missing or failed (#233)" ) downgrade_reasons.extend(worktree.get("reasons", [])) - if empty_queue_report.get("claimed") and not empty_queue_report.get("proven"): - downgrade_reasons.append( - "empty-queue report missing or failed trust-gate proof (#198)" - ) - downgrade_reasons.extend(empty_queue_report.get("reasons", [])) merge_allowed = ( identity_eligible @@ -853,12 +846,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "unrelated_mutations_avoided": bool( worktree.get("unrelated_mutations_avoided") ), - "empty_queue_trust_gate_proven": ( - empty_queue_report.get("proven") - if empty_queue_report.get("claimed") - else True - ), - "empty_queue_trust_gate_status": empty_queue_report.get("status"), } @@ -1018,7 +1005,6 @@ HANDOFF_ROLE_FIELDS = { ), "author": ( ("Selected issue", ("selected issue",)), - ("Issue lock proof", ("issue lock proof", "lock before diff")), ("Claim/comment status", ("claim/comment status", "claim status", "claim")), ("PR number opened", ("pr number opened", "pr opened", "pr number")), @@ -1029,14 +1015,6 @@ HANDOFF_ROLE_FIELDS = { ("Repositories checked", ("repositories checked", "repos checked")), ("Open PR counts", ("open pr counts", "open pr count", "open prs per repo")), - ("PR inventory trust gate", ("pr inventory trust gate", - "pr_inventory_trust_gate.status", - "trust gate status")), - ("Trust gate reasons", ("trust gate reasons", - "pr_inventory_trust_gate.reason")), - ("Trust gate corroborated", ("trust gate corroborated", - "pr_inventory_trust_gate.corroborated")), - ("Inventory profile", ("inventory profile", "inventory mcp profile")), ("Selected PR or reason", ("selected pr", "none selected", "reason none selected")), ("Inventory completeness", ("inventory complete", "inventory scoped", @@ -1420,148 +1398,6 @@ def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]: return lines -_EMPTY_QUEUE_CLAIM = re.compile( - r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|" - r"nothing to review|queue cleared|inventory empty|" - r"open pr count:\s*0|workflow correctly stops with nothing", - re.I, -) - -_WEAK_EMPTY_QUEUE_CORROBORATION = re.compile( - r"latest commit.*(?:merge|pr #)|merge of pr #|" - r"master latest commit|recent merge proves", - re.I, -) - -_TRUST_GATE_STATUS_LINE = re.compile( - r"pr_inventory_trust_gate\.status:\s*(\S+)", - re.I, -) - - -def parse_trust_gate_status_from_report(report_text: str | None) -> str | None: - """Extract ``pr_inventory_trust_gate.status`` from report text, if present.""" - match = _TRUST_GATE_STATUS_LINE.search(report_text or "") - return match.group(1).strip().lower() if match else None - - -def assess_empty_queue_report( - report_text: str | None, - *, - trust_gate: dict | None = None, - task_role: str | None = None, - inventory_profile: str | None = None, -) -> dict: - """Issue #198: empty-queue reports must cite the formal trust-gate result. - - Blocks reports that claim an empty queue without - ``pr_inventory_trust_gate.status == trusted_empty``, required inventory - metadata, or that rely on weak corroboration (e.g. a recent merge commit). - """ - text = report_text or "" - lower = text.lower() - reasons: list[str] = [] - missing: list[str] = [] - - if not _EMPTY_QUEUE_CLAIM.search(text): - return { - "claimed": False, - "proven": True, - "block": False, - "status": None, - "missing_fields": [], - "reasons": [], - } - - status = ( - (trust_gate or {}).get("status") - or parse_trust_gate_status_from_report(text) - ) - status_norm = (status or "").strip().lower() or None - - if not status_norm: - missing.append("pr_inventory_trust_gate.status") - reasons.append( - "empty-queue report missing pr_inventory_trust_gate.status; " - "fail closed" - ) - elif status_norm != "trusted_empty": - reasons.append( - f"empty-queue report has trust-gate status '{status_norm}', " - "not trusted_empty" - ) - - has_gate_reasons = ( - "pr_inventory_trust_gate.reason" in lower - or bool((trust_gate or {}).get("reasons")) - ) - if status_norm and status_norm != "trusted_empty" and not has_gate_reasons: - missing.append("pr_inventory_trust_gate.reasons") - - if status_norm == "trusted_empty": - if "pr_inventory_trust_gate.corroborated" not in lower and ( - trust_gate or {} - ).get("corroborated") is not True: - missing.append("pr_inventory_trust_gate.corroborated") - - inventory_markers = ( - "repository:", - "remote:", - "owner:", - "state filter:", - "state_filter:", - ) - if not any(marker in lower for marker in inventory_markers): - missing.append("inventory remote/owner/repo/state filter") - - profile_markers = ( - "mcp profile:", - "mcp-profile:", - "inventory profile:", - "active profile:", - ) - has_profile = ( - any(marker in lower for marker in profile_markers) - or bool((inventory_profile or "").strip()) - ) - if not has_profile: - missing.append("inventory MCP profile") - - if _WEAK_EMPTY_QUEUE_CORROBORATION.search(text): - if "pr_inventory_trust_gate.status: trusted_empty" not in lower: - reasons.append( - "weak corroboration (recent merge commit) cannot substitute " - "for pr_inventory_trust_gate.status == trusted_empty" - ) - - role = (task_role or "").strip().lower() - if role == "author" and re.search( - r"reviewer queue|nothing to review|review backlog empty", - text, - re.I, - ): - reasons.append( - "author-bound session presented reviewer queue inventory as a " - "reviewer decision" - ) - - if missing: - reasons.extend( - f"empty-queue report missing required field: {field}" - for field in missing - ) - - proven = not reasons and not missing - return { - "claimed": True, - "proven": proven, - "block": not proven, - "status": status_norm, - "missing_fields": missing, - "reasons": reasons, - } - - 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( diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 9a80d8d..aa90d88 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -65,20 +65,6 @@ CREATE_PR_ENV = { ), } -ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" - - -def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides): - record = { - "issue_number": issue_number, - "branch_name": branch_name, - "remote": "dadeschools", - "org": "Scaled-Tech-Consulting", - "repo": "Gitea-Tools", - } - record.update(overrides) - return record - # --------------------------------------------------------------------------- # Create Issue @@ -141,19 +127,15 @@ class TestCreatePR(unittest.TestCase): @patch("os.path.exists", return_value=True) @patch("builtins.open") def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role): - lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) - mock_open.return_value.__enter__.return_value.read.return_value = lock_json + mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} with patch.dict(os.environ, CREATE_PR_ENV, clear=True): result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertEqual(result["number"], 3) self.assertNotIn("url", result) - mock_exists.assert_called_with(ISSUE_LOCK_FILE) - mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8") payload = mock_api.call_args[0][3] self.assertEqual(payload["head"], "feat/x") self.assertEqual(payload["base"], "main") - self.assertIn("Closes #123", payload["title"]) @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @@ -162,28 +144,13 @@ class TestCreatePR(unittest.TestCase): @patch("os.path.exists", return_value=True) @patch("builtins.open") def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role): - lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) - mock_open.return_value.__enter__.return_value.read.return_value = lock_json + mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"} with patch.dict(os.environ, env, clear=True): result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertIn("pulls/3", result["url"]) - @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", - return_value=(True, [])) - @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - @patch("os.path.exists", return_value=True) - @patch("builtins.open") - def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role): - lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) - mock_open.return_value.__enter__.return_value.read.return_value = lock_json - with patch.dict(os.environ, CREATE_PR_ENV, clear=True): - with self.assertRaises(ValueError) as ctx: - gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main") - self.assertIn("Closes #123", str(ctx.exception)) - mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8") - # --------------------------------------------------------------------------- # Close Issue @@ -2803,8 +2770,8 @@ class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" def tearDown(self): - if os.path.exists(ISSUE_LOCK_FILE): - os.remove(ISSUE_LOCK_FILE) + if os.path.exists("/tmp/gitea_issue_lock.json"): + os.remove("/tmp/gitea_issue_lock.json") @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2812,7 +2779,7 @@ class TestIssueLocking(unittest.TestCase): mock_api.return_value = [] # no open PRs res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertTrue(res["success"]) - self.assertTrue(os.path.exists(ISSUE_LOCK_FILE)) + self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json")) def test_lock_issue_mismatch_branch_fails(self): with self.assertRaises(ValueError) as ctx: @@ -2849,8 +2816,8 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_missing_lock_fails(self, _auth, _role): - if os.path.exists(ISSUE_LOCK_FILE): - os.remove(ISSUE_LOCK_FILE) + if os.path.exists("/tmp/gitea_issue_lock.json"): + os.remove("/tmp/gitea_issue_lock.json") with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(RuntimeError) as ctx: gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs") @@ -2860,9 +2827,8 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_branch_mismatch_fails(self, _auth, _role): - with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: - json.dump(_sample_issue_lock( - issue_number=196, branch_name="feat/issue-196-mutations"), f) + with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: + json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(ValueError) as ctx: gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs") @@ -2872,9 +2838,8 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_forbidden_terms_fails(self, _auth, _role): - with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: - json.dump(_sample_issue_lock( - issue_number=196, branch_name="feat/issue-196-mutations"), f) + with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: + json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): for term in ("equivalent to #196", "related to #196", "same as #196"): with self.assertRaises(ValueError) as ctx: @@ -2885,9 +2850,8 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_missing_closes_ref_fails(self, _auth, _role): - with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: - json.dump(_sample_issue_lock( - issue_number=196, branch_name="feat/issue-196-mutations"), f) + with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: + json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(ValueError) as ctx: gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs") diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index f794c42..22748ce 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -27,7 +27,6 @@ from review_proofs import ( # noqa: E402 assess_controller_handoff, assess_inventory_completeness, assess_reviewer_queue_inventory, - assess_empty_queue_report, assess_live_state_recheck, assess_review_mutation_final_report, assess_role_boundary, @@ -945,46 +944,24 @@ class TestControllerHandoff(unittest.TestCase): result = assess_controller_handoff(complete, role="review") self.assertEqual(result["verdict"], "complete") - def _author_role_fields(self, issue_number=182, pr_number=999): - return [ - f"- Selected issue: #{issue_number}", - "- Issue lock proof: lock before diff on feat/x @ master", - "- Claim/comment status: comment-claimed", - f"- PR number opened: #{pr_number}", - "- No review/merge: confirmed", - ] - - def test_handoff_role_fields_author_includes_issue_lock_proof(self): - from review_proofs import HANDOFF_ROLE_FIELDS - names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]] - self.assertIn("Issue lock proof", names) - def test_author_role_requires_author_fields(self): - complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields()) - result = assess_controller_handoff(complete, role="author") - self.assertEqual(result["verdict"], "complete") - - result = assess_controller_handoff(self.BASE_HANDOFF, role="author") - self.assertEqual(result["verdict"], "incomplete") - self.assertIn("Issue lock proof", result["missing_fields"]) - self.assertIn("No review/merge confirmation", result["missing_fields"]) - - def test_author_role_requires_issue_lock_proof(self): - without_lock = self.BASE_HANDOFF + "\n" + "\n".join([ + complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #182", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", ]) - result = assess_controller_handoff(without_lock, role="author") + result = assess_controller_handoff(complete, role="author") + self.assertEqual(result["verdict"], "complete") + + result = assess_controller_handoff(self.BASE_HANDOFF, role="author") self.assertEqual(result["verdict"], "incomplete") - self.assertIn("Issue lock proof", result["missing_fields"]) + self.assertIn("No review/merge confirmation", result["missing_fields"]) def test_author_role_rejects_equivalent_or_multiple_issues(self): # 1. equivalent reference blocked incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: Issue #194 / #196 equivalent", - "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", @@ -996,7 +973,6 @@ class TestControllerHandoff(unittest.TestCase): # 2. multiple issues blocked incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #194, #196", - "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", @@ -1008,7 +984,6 @@ class TestControllerHandoff(unittest.TestCase): def test_author_role_rejects_fuzzy_pr_number(self): incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #196", - "- Issue lock proof: lock before diff on feat/issue-196 @ master", "- Claim/comment status: comment-claimed", "- PR number opened: PR #203 / #204 equivalent", "- No review/merge: confirmed", @@ -1021,10 +996,6 @@ class TestControllerHandoff(unittest.TestCase): complete = self.BASE_HANDOFF + "\n" + "\n".join([ "- Repositories checked: Gitea-Tools, mcp-control-plane", "- Open PR counts: 2 / 0", - "- PR inventory trust gate: trusted_nonempty / trusted_empty", - "- Trust gate reasons: none", - "- Trust gate corroborated: true", - "- Inventory profile: prgs-reviewer", "- Selected PR or reason: none eligible (self-authored)", "- Inventory completeness: complete, no pagination needed", ]) @@ -1044,17 +1015,23 @@ class TestControllerHandoff(unittest.TestCase): def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self): # 1. Workspace mutations: none is rejected when local_edits is True - incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join( - self._author_role_fields(issue_number=196, pr_number=203)) + incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ + "- Selected issue: #196", + "- Claim/comment status: comment-claimed", + "- PR number opened: #203", + "- No review/merge: confirmed", + ]) res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True) self.assertEqual(res["verdict"], "incomplete") self.assertIn("Workspace mutations", res["missing_fields"]) # 2. Workspace mutations: edited files is allowed when local_edits is True - complete_eq = self.BASE_HANDOFF.replace( - "- Workspace mutations: none", - "- Workspace mutations: edited review_proofs.py", - ) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203)) + complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([ + "- Selected issue: #196", + "- Claim/comment status: comment-claimed", + "- PR number opened: #203", + "- No review/merge: confirmed", + ]) res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True) self.assertEqual(res2["verdict"], "complete") @@ -1285,89 +1262,6 @@ class TestAssessReviewerQueueInventory(unittest.TestCase): self.assertEqual(result["trust_gates"], {}) -class TestAssessEmptyQueueReport(unittest.TestCase): - """Issue #198: empty-queue reports require formal trust-gate proof.""" - - def _trusted_report(self, **extra): - lines = [ - "Queue inventory complete.", - "Repository: Scaled-Tech-Consulting/Gitea-Tools", - "Open PR count: 0", - "pr_inventory_trust_gate.status: trusted_empty", - "pr_inventory_trust_gate.corroborated: true", - "Inventory profile: prgs-reviewer", - "Workflow correctly stops with nothing to review.", - ] - lines.extend(extra) - return "\n".join(lines) - - def test_non_empty_report_not_claimed(self): - result = assess_empty_queue_report("Reviewed PR #236 and merged.") - self.assertFalse(result["claimed"]) - self.assertTrue(result["proven"]) - - def test_empty_claim_without_trust_gate_blocked(self): - report = ( - "Open PR count: 0\n" - "Pagination complete: yes\n" - "Queue cleared." - ) - result = assess_empty_queue_report(report) - self.assertTrue(result["claimed"]) - self.assertFalse(result["proven"]) - self.assertTrue(result["block"]) - - def test_trusted_empty_report_with_required_fields_passes(self): - result = assess_empty_queue_report(self._trusted_report()) - self.assertTrue(result["proven"]) - - def test_weak_merge_commit_corroboration_blocked(self): - report = ( - "Open PR count: 0\n" - "Master latest commit is merge of PR #79 so queue is empty." - ) - result = assess_empty_queue_report(report) - self.assertFalse(result["proven"]) - self.assertTrue( - any("weak corroboration" in r for r in result["reasons"]) - ) - - def test_author_session_reviewer_queue_wording_blocked(self): - result = assess_empty_queue_report( - self._trusted_report(), - task_role="author", - ) - self.assertFalse(result["proven"]) - - def test_build_final_report_downgrades_weak_empty_queue(self): - final = build_final_report( - checkout_proof=_good_checkout(), - inventory=_good_inventory(), - validation=_good_validation(), - contamination=_good_contamination(), - identity_eligible=True, - merge_performed=False, - issue_status_verified=True, - capability_evidence=_good_capability_evidence(), - sweep=_good_sweep(), - live_state=_good_live_state(), - role_boundary=_good_role_boundary(), - review_mutation=_good_review_mutation(), - controller_handoff=_good_handoff(), - capability_proof=_good_capability_proof(), - sweep_proof=_good_secret_sweep(), - worktree_proof={ - "worktree_path": "/repo/branches/review-pr-1", - "porcelain_status": "", - "scratch_used": True, - "scratch_path": "/repo/branches/review-pr-1", - }, - report_text="Open PR count: 0. Queue cleared.", - ) - self.assertNotEqual(final["grade"], "A") - self.assertFalse(final["empty_queue_trust_gate_proven"]) - - class TestCapabilityEvidence(unittest.TestCase): """#179 gap 1: capability claims need exact evidence."""