From ec879df4c26069072661e9893d2a20f2fbc40025 Mon Sep 17 00:00:00 2001 From: jcwalker3 Date: Mon, 6 Jul 2026 10:19:46 -0500 Subject: [PATCH 1/3] test: require issue-lock proof in author handoffs (#208) --- review_proofs.py | 1 + tests/test_mcp_server.py | 62 +++++++++++++++++++++++++++++-------- tests/test_review_proofs.py | 52 +++++++++++++++++++++---------- 3 files changed, 85 insertions(+), 30 deletions(-) diff --git a/review_proofs.py b/review_proofs.py index 31b0b65..b8b41a5 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1005,6 +1005,7 @@ 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")), diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index aa90d88..9a80d8d 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -65,6 +65,20 @@ 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 @@ -127,15 +141,19 @@ 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): - mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' + 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_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, [])) @@ -144,13 +162,28 @@ 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): - mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' + 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_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 @@ -2770,8 +2803,8 @@ class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" def tearDown(self): - if os.path.exists("/tmp/gitea_issue_lock.json"): - os.remove("/tmp/gitea_issue_lock.json") + if os.path.exists(ISSUE_LOCK_FILE): + os.remove(ISSUE_LOCK_FILE) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2779,7 +2812,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("/tmp/gitea_issue_lock.json")) + self.assertTrue(os.path.exists(ISSUE_LOCK_FILE)) def test_lock_issue_mismatch_branch_fails(self): with self.assertRaises(ValueError) as ctx: @@ -2816,8 +2849,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("/tmp/gitea_issue_lock.json"): - os.remove("/tmp/gitea_issue_lock.json") + if os.path.exists(ISSUE_LOCK_FILE): + os.remove(ISSUE_LOCK_FILE) 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") @@ -2827,8 +2860,9 @@ 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("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + 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 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") @@ -2838,8 +2872,9 @@ 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("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + 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 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: @@ -2850,8 +2885,9 @@ 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("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + 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 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 22748ce..9f9e57c 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -944,24 +944,46 @@ class TestControllerHandoff(unittest.TestCase): result = assess_controller_handoff(complete, role="review") self.assertEqual(result["verdict"], "complete") - def test_author_role_requires_author_fields(self): - complete = self.BASE_HANDOFF + "\n" + "\n".join([ - "- Selected issue: #182", + 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", - "- PR number opened: #999", + 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([ + "- Selected issue: #182", + "- Claim/comment status: comment-claimed", + "- PR number opened: #999", + "- No review/merge: confirmed", + ]) + result = assess_controller_handoff(without_lock, role="author") + self.assertEqual(result["verdict"], "incomplete") + self.assertIn("Issue lock proof", 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", @@ -973,6 +995,7 @@ 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", @@ -984,6 +1007,7 @@ 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", @@ -1015,23 +1039,17 @@ 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([ - "- Selected issue: #196", - "- Claim/comment status: comment-claimed", - "- PR number opened: #203", - "- No review/merge: confirmed", - ]) + incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join( + self._author_role_fields(issue_number=196, pr_number=203)) 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([ - "- Selected issue: #196", - "- Claim/comment status: comment-claimed", - "- PR number opened: #203", - "- No review/merge: confirmed", - ]) + 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)) res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True) self.assertEqual(res2["verdict"], "complete") From 4bc02a8c7dd8a5137beb7f496c8aa8e6b7386b88 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 11:37:07 -0400 Subject: [PATCH 2/3] fix(reviewer): restore #198 empty-queue helpers for PR #235 (#208) Re-introduce assess_empty_queue_report and parse_trust_gate_status_from_report dropped when #208 was reverted from master. Wire build_final_report grading, inventory handoff trust-gate fields, and tests so capability_stop_terminal imports resolve while issue-lock proof enforcement remains. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 163 ++++++++++++++++++++++++++++++++++++ tests/test_review_proofs.py | 88 +++++++++++++++++++ 2 files changed, 251 insertions(+) diff --git a/review_proofs.py b/review_proofs.py index b8b41a5..73985b2 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -664,6 +664,8 @@ 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")) @@ -787,6 +789,11 @@ 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 @@ -846,6 +853,12 @@ 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"), } @@ -1016,6 +1029,14 @@ 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", @@ -1399,6 +1420,148 @@ 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_review_proofs.py b/tests/test_review_proofs.py index 9f9e57c..e0596c0 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -25,6 +25,7 @@ from review_proofs import ( # noqa: E402 assess_capability_evidence, assess_capability_proof, assess_controller_handoff, + assess_empty_queue_report, assess_inventory_completeness, assess_reviewer_queue_inventory, assess_live_state_recheck, @@ -1020,6 +1021,10 @@ 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", ]) @@ -1280,6 +1285,89 @@ 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.""" From 63a7ba8287a14bc1b1af02b95c26a469e4cba81c Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Mon, 6 Jul 2026 11:40:20 -0400 Subject: [PATCH 3/3] feat(author): add continuation-mode selection wall for open-PR issues (#188) Issues already represented by open PRs must not be selected for fresh work unless the operator explicitly requests continuation. Adds classification, continuation report proofs (old/new head SHA, PR author, branch), contradictory no-PR claim detection, and edited-PR inventory coverage checks. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 284 +++++++++++++++++++++++++++ skills/llm-project-workflow/SKILL.md | 6 + tests/test_review_proofs.py | 123 ++++++++++++ 3 files changed, 413 insertions(+) diff --git a/review_proofs.py b/review_proofs.py index 73985b2..925e554 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1042,6 +1042,16 @@ HANDOFF_ROLE_FIELDS = { ("Inventory completeness", ("inventory complete", "inventory scoped", "inventory completeness")), ), + "continuation": ( + ("Continuation mode", ("continuation mode", "continuation")), + ("Existing PR", ("existing pr", "pr number")), + ("PR author", ("pr author", "existing pr author")), + ("Branch", ("branch", "existing branch")), + ("Old PR head", ("old pr head", "old head")), + ("New PR head", ("new pr head", "new head")), + ("Session authored PR", ("session authored pr", "authored pr")), + ("Why continuation allowed", ("why continuation", "continuation allowed")), + ), } @@ -1562,6 +1572,280 @@ def assess_empty_queue_report( } +# ── Issue selection / continuation mode (#188) ─────────────────────────────── + +ISSUE_SELECTION_UNCLAIMED_NO_PR = "unclaimed_no_pr" +ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR = "represented_by_open_pr" +ISSUE_SELECTION_IN_PROGRESS = "in_progress" +ISSUE_SELECTION_CONTINUATION_EXPLICIT = "continuation_explicit" +ISSUE_SELECTION_EXCLUDED = "excluded" + +_NO_OPEN_PR_CLAIM = re.compile( + r"no duplicate pr|no open pr|no pr open|no eligible pr|" + r"no existing pr|without an open pr", + re.IGNORECASE, +) + + +def classify_issue_for_selection( + issue_number: int, + *, + labels: list[str] | None = None, + open_prs: list[dict] | None = None, + operator_continuation_requested: bool = False, + continuation_issue_numbers: list[int] | None = None, + excluded: bool = False, +) -> dict: + """Classify one issue for author queue selection (#188).""" + label_set = {str(l).lower() for l in (labels or [])} + prs = list(open_prs or []) + continuation_issues = set(continuation_issue_numbers or []) + + if excluded: + status = ISSUE_SELECTION_EXCLUDED + selectable_for_fresh_work = False + reasons = ["issue explicitly excluded from selection"] + elif "status:in-progress" in label_set: + status = ISSUE_SELECTION_IN_PROGRESS + selectable_for_fresh_work = False + reasons = ["issue already marked status:in-progress"] + elif prs and ( + operator_continuation_requested + or issue_number in continuation_issues + ): + status = ISSUE_SELECTION_CONTINUATION_EXPLICIT + selectable_for_fresh_work = False + reasons = ["operator requested continuation for issue with open PR"] + elif prs: + status = ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR + selectable_for_fresh_work = False + reasons = [ + f"issue #{issue_number} already represented by open PR " + f"#{prs[0].get('number')}" + ] + else: + status = ISSUE_SELECTION_UNCLAIMED_NO_PR + selectable_for_fresh_work = True + reasons = [] + + return { + "issue_number": issue_number, + "status": status, + "selectable_for_fresh_work": selectable_for_fresh_work, + "open_prs": prs, + "reasons": reasons, + } + + +def assess_fresh_issue_selection(classifications: list[dict] | None) -> dict: + """Fail closed when fresh selection picks an issue with an open PR.""" + reasons = [] + for item in classifications or []: + if item.get("selectable_for_fresh_work"): + continue + if item.get("status") == ISSUE_SELECTION_CONTINUATION_EXPLICIT: + continue + if item.get("status") == ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR: + reasons.append( + f"issue #{item.get('issue_number')} has open PR and was " + "selected for fresh work without continuation mode" + ) + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_continuation_mode_report( + report_text: str, + *, + pr_number: int | None = None, + pr_author: str | None = None, + branch: str | None = None, + old_head_sha: str | None = None, + new_head_sha: str | None = None, + session_authored_pr: bool | None = None, + continuation_allowed_reason: str | None = None, +) -> dict: + """Issue #188: continuation mode must disclose full PR evidence.""" + text = report_text or "" + lower = text.lower() + reasons = [] + + if not any(p in lower for p in ("continuation", "continue pr", "continue issue")): + reasons.append("report does not declare continuation mode") + + if pr_number is not None: + if f"#{pr_number}" not in lower and f"pr {pr_number}" not in lower: + reasons.append(f"continuation report missing PR #{pr_number}") + if pr_author and pr_author.lower() not in lower: + reasons.append("continuation report missing PR author") + if branch and branch.lower() not in lower: + reasons.append("continuation report missing branch name") + + for label, sha in (("old", old_head_sha), ("new", new_head_sha)): + if not sha: + reasons.append(f"continuation proof missing {label} head SHA") + elif not _FULL_SHA.match(sha.lower()): + reasons.append( + f"continuation {label} head SHA is not a full 40-hex SHA" + ) + elif sha.lower() not in lower: + reasons.append( + f"continuation report missing {label} head SHA in evidence" + ) + + if session_authored_pr is not None: + authored_tokens = ("session authored", "authored pr", "own pr", "my pr") + if not any(t in lower for t in authored_tokens): + reasons.append( + "continuation report missing whether session authored the PR" + ) + + if continuation_allowed_reason: + reason_lower = continuation_allowed_reason.lower() + if ( + reason_lower not in lower + and not any(w in lower for w in reason_lower.split()[:3]) + ): + reasons.append("continuation report missing why continuation is allowed") + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_contradictory_no_pr_claim( + report_text: str, + *, + edited_pr_numbers: list[int] | None = None, + issue_open_pr_map: dict[int, int] | None = None, +) -> dict: + """Downgrade when report claims no open PR but later edits one.""" + text = report_text or "" + lower = text.lower() + reasons = [] + + if not _NO_OPEN_PR_CLAIM.search(lower): + return {"complete": True, "downgraded": False, "reasons": []} + + edited = list(edited_pr_numbers or []) + for pr_num in edited: + if f"#{pr_num}" in lower or f"pr {pr_num}" in lower: + reasons.append( + f"report claims no open PR but edited PR #{pr_num}" + ) + + for issue_num, pr_num in (issue_open_pr_map or {}).items(): + if f"#{issue_num}" in lower or f"issue #{issue_num}" in lower: + if f"#{pr_num}" in lower or f"pr {pr_num}" in lower: + reasons.append( + f"report claims no open PR for issue #{issue_num} but " + f"PR #{pr_num} exists and was referenced" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_edited_pr_inventory_coverage( + report_text: str, + *, + edited_pr_numbers: list[int] | None = None, + inventoried_pr_numbers: list[int] | None = None, +) -> dict: + """Open PR inventory must include PRs the run later edits (#188).""" + text = report_text or "" + lower = text.lower() + reasons = [] + inventoried = set(inventoried_pr_numbers or []) + + for pr_num in edited_pr_numbers or []: + if pr_num in inventoried: + continue + if f"#{pr_num}" not in lower and f"pr {pr_num}" not in lower: + reasons.append( + f"edited PR #{pr_num} missing from open PR inventory" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + +def assess_issue_selection_final_report( + report_text: str, + *, + mode: str = "fresh", + classifications: list[dict] | None = None, + continuation_proof: dict | None = None, + edited_pr_numbers: list[int] | None = None, + inventoried_pr_numbers: list[int] | None = None, + issue_open_pr_map: dict[int, int] | None = None, +) -> dict: + """Issue #188: composite A-bar for author issue-selection runs.""" + handoff_role = "continuation" if mode == "continuation" else "author" + checks = { + "controller_handoff": assess_controller_handoff( + report_text, role=handoff_role + ), + "contradictory_no_pr": assess_contradictory_no_pr_claim( + report_text, + edited_pr_numbers=edited_pr_numbers, + issue_open_pr_map=issue_open_pr_map, + ), + "edited_pr_inventory": assess_edited_pr_inventory_coverage( + report_text, + edited_pr_numbers=edited_pr_numbers, + inventoried_pr_numbers=inventoried_pr_numbers, + ), + } + + if mode == "fresh": + checks["fresh_selection"] = assess_fresh_issue_selection(classifications) + else: + proof = continuation_proof or {} + checks["continuation_mode"] = assess_continuation_mode_report( + report_text, + pr_number=proof.get("pr_number"), + pr_author=proof.get("pr_author"), + branch=proof.get("branch"), + old_head_sha=proof.get("old_head_sha"), + new_head_sha=proof.get("new_head_sha"), + session_authored_pr=proof.get("session_authored_pr"), + continuation_allowed_reason=proof.get("continuation_allowed_reason"), + ) + + reasons = [] + downgraded = False + for name, result in checks.items(): + verdict = result.get("verdict") + if verdict in ("missing", "incomplete"): + downgraded = True + reasons.extend(result.get("reasons") or []) + elif result.get("downgraded") or not result.get("complete", True): + downgraded = True + reasons.extend( + f"{name}: {r}" for r in (result.get("reasons") or []) + ) + + return { + "grade": "A" if not downgraded else "downgraded", + "downgraded": downgraded, + "checks": checks, + "reasons": reasons, + "complete": not downgraded, + } + + 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/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 48c96a1..1145bcd 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -379,6 +379,12 @@ Role-specific fields (append to the compact block): `Linked issue status:`, `Cleanup status:` - author tasks: `Selected issue:`, `Claim/comment status:`, `PR number opened:`, `No review/merge:` (explicit confirmation) +- continuation tasks (#188): `Continuation mode:`, `Existing PR:`, + `PR author:`, `Branch:`, `Old PR head:`, `New PR head:`, + `Session authored PR:`, `Why continuation allowed:` — issues with open + PRs are excluded from fresh selection unless operator explicitly requests + continuation (`review_proofs.classify_issue_for_selection`, + `assess_issue_selection_final_report`) - queue/inventory tasks: `Repositories checked:`, `Open PR counts:`, `Selected PR or reason none selected:`, `Inventory completeness:` diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index e0596c0..8ce6298 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -21,12 +21,19 @@ import unittest sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) from review_proofs import ( # noqa: E402 + ISSUE_SELECTION_CONTINUATION_EXPLICIT, + ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR, assess_author_pr_report, assess_capability_evidence, assess_capability_proof, + assess_contradictory_no_pr_claim, + assess_continuation_mode_report, assess_controller_handoff, + assess_edited_pr_inventory_coverage, assess_empty_queue_report, + assess_fresh_issue_selection, assess_inventory_completeness, + assess_issue_selection_final_report, assess_reviewer_queue_inventory, assess_live_state_recheck, assess_review_mutation_final_report, @@ -36,6 +43,7 @@ from review_proofs import ( # noqa: E402 assess_sweep_evidence, assess_validation_report, build_final_report, + classify_issue_for_selection, pr_inventory_trust_gate, resolve_repos_from_user_reference, verify_pinned_head_checkout, @@ -1662,5 +1670,120 @@ class TestAuthorReporting(unittest.TestCase): self.assertFalse(result["complete"]) +class TestIssueSelectionContinuation(unittest.TestCase): + """Issue #188: continuation mode wall for issues with open PRs.""" + + OLD_SHA = PINNED + NEW_SHA = OTHER + OPEN_PR = [{"number": 187, "head": {"ref": "feat/issue-183-harden-author-run-reporting"}}] + + def test_open_pr_issue_excluded_from_fresh_selection(self): + classified = classify_issue_for_selection( + 183, open_prs=self.OPEN_PR, + ) + self.assertEqual(classified["status"], ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR) + self.assertFalse(classified["selectable_for_fresh_work"]) + blocked = assess_fresh_issue_selection([classified]) + self.assertTrue(blocked["downgraded"]) + + def test_explicit_continuation_allows_represented_issue(self): + classified = classify_issue_for_selection( + 183, + open_prs=self.OPEN_PR, + operator_continuation_requested=True, + ) + self.assertEqual(classified["status"], ISSUE_SELECTION_CONTINUATION_EXPLICIT) + blocked = assess_fresh_issue_selection([classified]) + self.assertFalse(blocked["downgraded"]) + + def test_contradictory_no_pr_claim_downgrades(self): + report = ( + "Selected issue #183; no duplicate PR open. " + "Updated PR #187 on branch feat/issue-183-harden-author-run-reporting." + ) + result = assess_contradictory_no_pr_claim( + report, edited_pr_numbers=[187], issue_open_pr_map={183: 187}, + ) + self.assertTrue(result["downgraded"]) + + def test_edited_pr_must_appear_in_inventory(self): + report = "Open PR inventory: PR #195 only." + result = assess_edited_pr_inventory_coverage( + report, + edited_pr_numbers=[187], + inventoried_pr_numbers=[195], + ) + self.assertTrue(result["downgraded"]) + + def test_continuation_report_requires_old_and_new_head(self): + report = ( + "Issue #182 continuation mode. PR #186. " + f"old head {self.OLD_SHA} -> new head {self.NEW_SHA}. " + "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff. " + "Session authored PR: yes. Continuation allowed: operator requested." + ) + result = assess_continuation_mode_report( + report, + pr_number=186, + pr_author="jcwalker3", + branch="feat/issue-182-controller-handoff", + old_head_sha=self.OLD_SHA, + new_head_sha=self.NEW_SHA, + session_authored_pr=True, + continuation_allowed_reason="operator requested continuation", + ) + self.assertTrue(result["complete"]) + + def test_issue_selection_final_report_continuation_earns_a(self): + report = "\n".join([ + "Issue #182 continuation mode — no new issue claimed.", + f"PR #186 updated: old head {self.OLD_SHA}, " + f"new head {self.NEW_SHA}.", + "PR author: jcwalker3. Branch: feat/issue-182-controller-handoff.", + "Session authored PR: yes.", + "Continuation allowed: operator requested rebase.", + "Open PR inventory included PR #186.", + "## Controller Handoff", + "- Task: continuation", + "- Repo: Scaled-Tech-Consulting/Gitea-Tools", + "- Role: author", + "- Identity: prgs-author", + "- Issue/PR: #182 / PR #186", + "- Branch/SHA: feat/issue-182-controller-handoff", + "- Files changed: review_proofs.py", + "- Validation: tests passed", + "- Mutations: push_branch", + "- Workspace mutations: none", + "- Current status: PR mergeable", + "- Blockers: none", + "- Next: review", + "- Safety: no review/merge", + "- Continuation mode: issue #182 continuation", + "- Existing PR: #186", + "- PR author: jcwalker3", + "- Branch: feat/issue-182-controller-handoff", + f"- Old PR head: {self.OLD_SHA}", + f"- New PR head: {self.NEW_SHA}", + "- Session authored PR: yes", + "- Why continuation allowed: operator requested rebase", + ]) + result = assess_issue_selection_final_report( + report, + mode="continuation", + continuation_proof={ + "pr_number": 186, + "pr_author": "jcwalker3", + "branch": "feat/issue-182-controller-handoff", + "old_head_sha": self.OLD_SHA, + "new_head_sha": self.NEW_SHA, + "session_authored_pr": True, + "continuation_allowed_reason": "operator requested rebase", + }, + edited_pr_numbers=[186], + inventoried_pr_numbers=[186], + ) + self.assertEqual(result["grade"], "A") + + if __name__ == "__main__": unittest.main()