"""Tests for blind PR queue review proofs (Issue #173). Issue #173 requires the reviewer-side workflow to *prove* — not assume — each precondition of a blind queue review: 1. Pinned PR head vs local checkout HEAD (checkout proof). 2. Complete open-PR inventory across all configured repositories. 3. Evidence-strength of validation reporting (exact commands, counts, justified ignores, output actually read). 4. Evidence-backed self-review contamination assessment (never assumed). 5. A final report that distinguishes each proof and can only claim an "A"-grade run when every proof is present; otherwise it downgrades or blocks instead of merging confidently. These are the harness assertions from the issue's Required behavior 7. """ import io import sys 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_already_landed_report_state, assess_already_landed_review_gate, assess_author_pr_report, assess_partial_reconciliation_report, resolve_partial_reconciliation_plan, assess_validation_environment_proof, assess_full_suite_failure_approval_gate, assess_queue_ordering_proof, assess_reviewer_baseline_validation_proof, assess_capability_evidence, assess_capability_proof, assess_contradictory_no_pr_claim, assess_continuation_mode_report, assess_controller_handoff, assess_create_issue_final_report, assess_create_issue_mode_isolation, assess_create_issue_workflow_source, assess_work_issue_final_report, assess_work_issue_mode_isolation, assess_work_issue_workflow_source, assess_edited_pr_inventory_coverage, assess_empty_queue_report, assess_fresh_issue_selection, assess_inventory_completeness, assess_issue_filing_final_report, assess_issue_filing_mutation_capability, assess_issue_filing_sha_evidence, assess_issue_selection_final_report, assess_queue_target_final_report, assess_reviewer_queue_inventory, assess_live_state_recheck, assess_review_mutation_final_report, assess_role_boundary, assess_secret_sweep, assess_self_review_contamination, assess_sweep_evidence, assess_validation_report, build_final_report, classify_issue_for_selection, pr_inventory_trust_gate, reconcile_queue_target, resolve_repos_from_user_reference, verify_pinned_head_checkout, ) PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9" def _good_checkout(): return verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="prgs/master", ) def _good_inventory(): return assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, "open_pr_count": 2, }, { "repo": "Scaled-Tech-Consulting/mcp-control-plane", "state_filter": "open", "pagination_complete": True, "open_pr_count": 1, }, ], required_repos=[ "Scaled-Tech-Consulting/Gitea-Tools", "Scaled-Tech-Consulting/mcp-control-plane", ], ) def _good_validation(**overrides): report = { "command": "venv/bin/pytest tests/ --ignore=branches", "output_read": True, "result": "pass", "passed": 421, "failed": 0, "skipped": 13, "ignored_paths": [ { "path": "branches", "justification": ( "review worktrees under branches/ duplicate test module " "names and break collection; they are not part of the " "repository's own suite" ), } ], "canonical_command": "venv/bin/pytest tests/ --ignore=branches", } report.update(overrides) return assess_validation_report(report) def _good_contamination(): return assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=False, evidence_source="session transcript: branch never checked out or pushed here", ) def _good_role_boundary(): return assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": False, "author_mutations": [], "review_mutations": [], "operator_authorized_author_work": False, "scratch_evidence_claimed": False, } ) def _good_capability_evidence(): return assess_capability_evidence([ { "task": "review_pr", "allowed": True, "evidence_source": ( "gitea_resolve_task_capability(review_pr) output: " "allowed_in_current_session=true, profile prgs-reviewer" ), }, { "task": "merge_pr", "allowed": True, "evidence_source": ( "gitea_resolve_task_capability(merge_pr) output: " "allowed_in_current_session=true, profile prgs-reviewer" ), }, ]) def _good_sweep(**overrides): sweep = { "command": ( "git diff prgs/master...HEAD | grep -inE " "'password|token|secret|api[_-]?key|authorization|bearer|https?://'" ), "scope": "full PR diff against prgs/master", "clean": True, } sweep.update(overrides) return assess_sweep_evidence(sweep) def _good_live_state(**overrides): recheck = { "pr_state": "open", "pinned_head_sha": PINNED, "live_head_sha": PINNED, "pinned_base_ref": "master", "live_base_ref": "master", "blocking_change_requests": False, } recheck.update(overrides) return assess_live_state_recheck(recheck) def _good_review_mutation(): report = ( "Review complete on PR #203.\n" "Review mutations: one request_changes on #203.\n" "Review decision: request_changes." ) lock = { "live_mutations": [{"pr_number": 203, "action": "request_changes"}], "correction_authorized": False, } return assess_review_mutation_final_report(report, lock) def _good_worktree(**overrides): proof = { "worktree_path": "/repo/branches/review-feat-issue-224", "porcelain_status": "", "pr_scope_files": ["docs/wiki/Repositories.md"], "scratch_used": True, "scratch_path": "/repo/branches/review-feat-issue-224", "git_commands": [ "git fetch prgs master feat/issue-224-wiki-proof-refresh", "git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh", ], } proof.update(overrides) from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402 return assess_reviewer_worktree_proof(proof) def _good_role_boundary_179(**overrides): kwargs = { "task_role": "reviewer", "namespaces_used": ["gitea-reviewer"], "justification": None, } kwargs.update(overrides) return assess_role_boundary(**kwargs) GOOD_HANDOFF = "\n".join([ "## Controller Handoff", "", "- Task: review PR #999", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: reviewer", "- Identity: sysadmin / prgs-reviewer", "- Issue/PR: #182 / PR #999", "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Files changed: review_proofs.py", "- Validation: 700 passed, 6 skipped", "- Mutations: review only", "- Workspace mutations: none", "- Current status: PR open", "- Blockers: none", "- Next: merge if approved", "- Safety: no self-review; no self-merge; no secrets", ]) def _good_handoff(): return assess_controller_handoff(GOOD_HANDOFF) def _good_capability_proof(): return assess_capability_proof({ "comment_issue": { "requested_task": "comment_issue", "required_operation_permission": "gitea.issue.comment", "allowed_in_current_session": True, } }) def _good_secret_sweep(**overrides): sweep = { "method": "git diff prgs/master...HEAD | grep -iE 'token|secret'", "scope": "full PR diff relative to prgs/master", "clean": True, } sweep.update(overrides) return assess_secret_sweep(sweep) class TestCheckoutProof(unittest.TestCase): """Required behavior 1 + 2: prove HEAD == pinned PR head or stop.""" def test_matching_pinned_and_local_head_is_proven(self): proof = _good_checkout() self.assertTrue(proof["proven"]) self.assertFalse(proof["block"]) def test_fetched_sha_but_unchecked_out_head_blocks(self): # Harness assertion (behavior 7, bullet 1): fetching/pinning a SHA # without checking it out must block review and merge. proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=OTHER, pr_base_ref="master", diff_base_ref="prgs/master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) self.assertTrue(any("HEAD" in r for r in proof["reasons"])) def test_missing_pinned_sha_fails_closed(self): proof = verify_pinned_head_checkout( pinned_head_sha="", local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) def test_missing_local_head_fails_closed(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=None, pr_base_ref="master", diff_base_ref="master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) def test_abbreviated_sha_is_not_proof(self): # A prefix match is not proof of the exact pinned head. proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED[:12], pr_base_ref="master", diff_base_ref="master", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) def test_wrong_diff_base_blocks(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="develop", ) self.assertFalse(proof["proven"]) self.assertTrue(proof["block"]) self.assertTrue(any("base" in r.lower() for r in proof["reasons"])) def test_remote_tracking_diff_base_matches_pr_base(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=PINNED, pr_base_ref="master", diff_base_ref="refs/remotes/prgs/master", ) self.assertTrue(proof["proven"]) class TestInventoryCompleteness(unittest.TestCase): """Required behavior 3: inventory must prove completeness.""" def test_complete_multi_repo_inventory(self): # Harness assertion (behavior 7, bullet 2): complete queue inventory # with multiple PRs across repos. result = _good_inventory() self.assertTrue(result["complete"]) self.assertTrue(result["can_claim_exhaustive"]) self.assertEqual( result["total_open_prs"], { "Scaled-Tech-Consulting/Gitea-Tools": 2, "Scaled-Tech-Consulting/mcp-control-plane": 1, }, ) def test_missing_repo_makes_inventory_incomplete(self): result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, "open_pr_count": 1, } ], required_repos=[ "Scaled-Tech-Consulting/Gitea-Tools", "Scaled-Tech-Consulting/mcp-control-plane", ], ) self.assertFalse(result["complete"]) self.assertFalse(result["can_claim_exhaustive"]) self.assertTrue( any("mcp-control-plane" in r for r in result["reasons"]) ) def test_unfinished_pagination_blocks_exhaustive_claim(self): # Harness assertion (behavior 7, bullet 3): pagination / multi-page # PR listing must be handled or the claim is refused. result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": False, "open_pr_count": 50, } ], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) self.assertFalse(result["can_claim_exhaustive"]) self.assertTrue(any("pagination" in r.lower() for r in result["reasons"])) def test_missing_state_filter_blocks_exhaustive_claim(self): result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "pagination_complete": True, "open_pr_count": 1, } ], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) def test_missing_count_blocks_exhaustive_claim(self): result = assess_inventory_completeness( repo_reports=[ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, } ], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) def test_empty_inventory_fails_closed(self): result = assess_inventory_completeness( repo_reports=[], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) self.assertFalse(result["complete"]) self.assertFalse(result["can_claim_exhaustive"]) class TestValidationReporting(unittest.TestCase): """Required behavior 4: exact commands and exact results.""" def test_full_report_is_strong(self): result = _good_validation() self.assertEqual(result["verdict"], "strong") self.assertTrue(result["claimable"]) def test_missing_test_counts_is_weak(self): # Harness assertion (behavior 7, bullet 4): validation result missing # test counts is flagged as weak/incomplete. result = _good_validation(passed=None, failed=None, skipped=None) self.assertEqual(result["verdict"], "weak") self.assertTrue(any("count" in r.lower() for r in result["reasons"])) def test_output_not_read_cannot_claim_result(self): # Harness assertion (behavior 7, bullet 6): a report cannot say # validation passed unless the command output was actually read. result = _good_validation(output_read=False) self.assertEqual(result["verdict"], "invalid") self.assertFalse(result["claimable"]) def test_missing_command_cannot_claim_result(self): result = _good_validation(command="") self.assertEqual(result["verdict"], "invalid") self.assertFalse(result["claimable"]) def test_unjustified_ignored_path_is_weak(self): # Harness assertion (behavior 7, bullet 7): ignored paths in # validation must be explicitly justified. result = _good_validation( ignored_paths=[{"path": "branches", "justification": ""}] ) self.assertEqual(result["verdict"], "weak") self.assertTrue(any("branches" in r for r in result["reasons"])) def test_non_canonical_command_without_note_is_weak(self): result = _good_validation( command="venv/bin/pytest tests/test_prs.py", canonical_command="venv/bin/pytest tests/ --ignore=branches", ) self.assertEqual(result["verdict"], "weak") self.assertTrue(any("canonical" in r.lower() for r in result["reasons"])) def test_non_canonical_command_with_justification_is_strong(self): result = _good_validation( command="venv/bin/pytest tests/test_prs.py", canonical_command="venv/bin/pytest tests/ --ignore=branches", deviation_justification="targeted re-run of the only failing file", ) self.assertEqual(result["verdict"], "strong") class TestSelfReviewContamination(unittest.TestCase): """Required behavior 5: contamination claims need evidence.""" def test_identity_match_is_contaminated(self): result = assess_self_review_contamination( reviewer_identity="jcwalker3", pr_author="jcwalker3", ) self.assertEqual(result["status"], "contaminated") def test_session_authorship_claim_without_evidence_is_unknown(self): # Harness assertion (behavior 7, bullet 5): self-review contamination # requires evidence, not assumption. result = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=True, evidence_source=None, ) self.assertEqual(result["status"], "unknown") self.assertIn("choose another PR or stop", result["safe_next_action"]) def test_session_authorship_with_evidence_is_contaminated(self): result = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=True, evidence_source="this session created branch feat/x and opened the PR", ) self.assertEqual(result["status"], "contaminated") def test_clean_requires_evidence_too(self): result = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=False, evidence_source=None, ) self.assertEqual(result["status"], "unknown") def test_distinct_identities_with_evidence_is_clean(self): result = _good_contamination() self.assertEqual(result["status"], "clean") def test_missing_identities_are_unknown(self): result = assess_self_review_contamination( reviewer_identity=None, pr_author="jcwalker3", ) self.assertEqual(result["status"], "unknown") class TestRoleBoundary(unittest.TestCase): """Issue #175: reviewer queue tasks must not pivot into author work.""" def test_reviewer_queue_without_author_mutations_is_clean(self): result = _good_role_boundary() self.assertEqual(result["status"], "clean") self.assertEqual(result["violations"], []) def test_reviewer_queue_author_mutation_without_authorization_violates(self): result = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": True, "author_mutations": ["claim issue #171", "push branch"], "operator_authorized_author_work": False, "mixed_namespace_justification": ( "author namespace was used for implementation" ), } ) self.assertEqual(result["status"], "violation") self.assertTrue(any("pivot" in r for r in result["violations"])) def test_mixed_namespace_use_without_justification_is_warning(self): result = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": True, "author_mutations": [], } ) self.assertEqual(result["status"], "warning") self.assertTrue( any("mixed" in r.lower() for r in result["reasons"]) ) def test_author_task_cannot_perform_review_mutations(self): result = assess_role_boundary( { "task_role": "author", "reviewer_namespace_used": False, "author_namespace_used": True, "review_mutations": ["approve PR"], } ) self.assertEqual(result["status"], "violation") self.assertTrue( any("reviewer-only" in r for r in result["violations"]) ) def test_scratch_only_notes_are_not_durable_evidence(self): result = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "scratch_evidence_claimed": True, "scratch_evidence_durable": False, } ) self.assertEqual(result["status"], "warning") self.assertTrue(any("scratch-only" in r for r in result["reasons"])) class TestFinalReport(unittest.TestCase): """Required behavior 6 + acceptance criteria: the report must distinguish each proof, and only a fully proven run earns an "A".""" def _report(self, **overrides): kwargs = { "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-feat-issue-224", "porcelain_status": "", "pr_scope_files": ["docs/wiki/Repositories.md"], "scratch_used": True, "scratch_path": "/repo/branches/review-feat-issue-224", }, } kwargs.update(overrides) return build_final_report(**kwargs) def test_fully_proven_run_is_grade_a(self): report = self._report() self.assertEqual(report["grade"], "A") self.assertEqual(report["violations"], []) # Behavior 6: each distinction is a distinct field. self.assertTrue(report["identity_eligible"]) self.assertTrue(report["pr_author_distinct_from_reviewer"]) self.assertEqual(report["session_contamination"], "clean") self.assertEqual(report["role_boundary"], "clean") self.assertTrue(report["validated_on_pinned_head"]) self.assertFalse(report["merge_performed"]) self.assertTrue(report["issue_status_verified"]) def test_unproven_checkout_downgrades_and_blocks_merge(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=OTHER, pr_base_ref="master", diff_base_ref="master", ) report = self._report(checkout_proof=proof) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertFalse(report["validated_on_pinned_head"]) def test_merge_claim_without_checkout_proof_is_a_violation(self): proof = verify_pinned_head_checkout( pinned_head_sha=PINNED, local_head_sha=OTHER, pr_base_ref="master", diff_base_ref="master", ) report = self._report(checkout_proof=proof, merge_performed=True) self.assertEqual(report["grade"], "blocked") self.assertTrue(report["violations"]) def test_unread_validation_output_cannot_be_reported_as_passed(self): report = self._report(validation=_good_validation(output_read=False)) self.assertFalse(report["validation_passed"]) self.assertNotEqual(report["grade"], "A") def test_weak_validation_downgrades(self): report = self._report( validation=_good_validation(passed=None, failed=None, skipped=None) ) self.assertNotEqual(report["grade"], "A") def test_incomplete_inventory_downgrades(self): inventory = assess_inventory_completeness( repo_reports=[], required_repos=["Scaled-Tech-Consulting/Gitea-Tools"], ) report = self._report(inventory=inventory) self.assertNotEqual(report["grade"], "A") def test_unknown_contamination_downgrades_and_blocks_merge(self): contamination = assess_self_review_contamination( reviewer_identity="sysadmin", pr_author="jcwalker3", session_authored=True, evidence_source=None, ) report = self._report(contamination=contamination) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertEqual(report["session_contamination"], "unknown") def test_merge_by_contaminated_reviewer_is_a_violation(self): contamination = assess_self_review_contamination( reviewer_identity="jcwalker3", pr_author="jcwalker3", ) report = self._report(contamination=contamination, merge_performed=True) self.assertEqual(report["grade"], "blocked") self.assertTrue(report["violations"]) def test_ineligible_identity_downgrades_and_blocks_merge(self): report = self._report(identity_eligible=False) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) def test_missing_role_boundary_downgrades_and_blocks_merge(self): kwargs = { "checkout_proof": _good_checkout(), "inventory": _good_inventory(), "validation": _good_validation(), "contamination": _good_contamination(), "identity_eligible": True, "merge_performed": False, "issue_status_verified": True, "review_mutation": _good_review_mutation(), } report = build_final_report(**kwargs) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertEqual(report["role_boundary"], "warning") def test_role_boundary_violation_blocks_report(self): boundary = assess_role_boundary( { "task_role": "reviewer", "task_kind": "blind_pr_queue_review", "reviewer_namespace_used": True, "author_namespace_used": True, "author_mutations": ["create PR"], "operator_authorized_author_work": False, "mixed_namespace_justification": "implementation pivot", } ) report = self._report(role_boundary=boundary) self.assertEqual(report["grade"], "blocked") self.assertFalse(report["merge_allowed"]) def test_failed_capability_proof_downgrades(self): report = self._report( capability_proof={ "proven": False, "reasons": ["task mark_issue not allowed"], } ) self.assertNotEqual(report["grade"], "A") self.assertTrue( any("capability" in r for r in report["downgrade_reasons"]) ) def test_failed_sweep_proof_downgrades(self): report = self._report( sweep_proof={ "proven": False, "reasons": ["method missing"], } ) self.assertNotEqual(report["grade"], "A") self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"])) class TestStdoutIsolation(unittest.TestCase): """Regression test for #178: tests must not close or corrupt stdout/stderr (prevents need for junitxml workaround in full suite runs and review validation). """ def test_stdout_remains_usable(self): """After typical test activity (mocks, redirects, prints from mains), stdout should be usable.""" # Verify not closed self.assertFalse(getattr(sys.stdout, "closed", False)) # Should be able to write (even if captured by pytest) try: sys.stdout.write("") sys.stdout.flush() except Exception as exc: self.fail(f"stdout write failed after test activity: {exc}") def test_stderr_remains_usable(self): self.assertFalse(getattr(sys.stderr, "closed", False)) try: sys.stderr.write("") sys.stderr.flush() except Exception as exc: self.fail(f"stderr write failed: {exc}") class TestRepoNameDisambiguation(unittest.TestCase): """Harness assertions for repo name disambiguation (new blind-review hardening). "MCP Gitea tool" etc. must resolve to Gitea-Tools, not silently default to mcp-control-plane. "open PRs" or ambiguous must check both. Single-repo zero-result must not hide PRs in the other configured repo. """ CONFIGURED = [ "Scaled-Tech-Consulting/Gitea-Tools", "Scaled-Tech-Consulting/mcp-control-plane", ] def test_gitea_tools_aliases_resolve_to_gitea_tools_only(self): for ref in [ "MCP Gitea tool", "gitea tool", "Gitea-Tools", "gitea mcp tool", "gitea-tools repo", ]: result = resolve_repos_from_user_reference(ref, self.CONFIGURED) self.assertEqual(result, ["Scaled-Tech-Consulting/Gitea-Tools"]) def test_mcp_control_plane_alias_resolves_only_to_it(self): result = resolve_repos_from_user_reference( "mcp-control-plane", self.CONFIGURED ) self.assertEqual(result, ["Scaled-Tech-Consulting/mcp-control-plane"]) def test_ambiguous_or_empty_defaults_to_both(self): self.assertEqual( resolve_repos_from_user_reference("open PRs", self.CONFIGURED), self.CONFIGURED, ) self.assertEqual( resolve_repos_from_user_reference("", self.CONFIGURED), self.CONFIGURED, ) self.assertEqual( resolve_repos_from_user_reference("review the queue", self.CONFIGURED), self.CONFIGURED, ) def test_single_repo_zero_result_does_not_hide_other(self): # Simulate a run that only inventoried mcp because of bad alias resolution. # The inventory must still require Gitea-Tools to claim "no open PRs". mcp_only_reports = [ { "repo": "Scaled-Tech-Consulting/mcp-control-plane", "state_filter": "open", "pagination_complete": True, "open_pr_count": 0, } ] result = assess_inventory_completeness( repo_reports=mcp_only_reports, required_repos=self.CONFIGURED, ) self.assertFalse(result["complete"]) self.assertTrue( any("Gitea-Tools" in r for r in result["reasons"]) ) def test_full_inventory_of_both_is_required_for_exhaustive_claim(self): both_reports = [ { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, "open_pr_count": 1, }, { "repo": "Scaled-Tech-Consulting/mcp-control-plane", "state_filter": "open", "pagination_complete": True, "open_pr_count": 0, }, ] result = assess_inventory_completeness( repo_reports=both_reports, required_repos=self.CONFIGURED ) self.assertTrue(result["complete"]) self.assertTrue(result["can_claim_exhaustive"]) class TestControllerHandoff(unittest.TestCase): """Issue #182: final reports must end with a Controller Handoff.""" BASE_HANDOFF = "\n".join([ "## Controller Handoff", "", "- Task: implement issue #182", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: author", "- Identity: jcwalker3 / prgs-author", "- Issue/PR: #182 / PR #999", "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Files changed: review_proofs.py", "- Validation: 700 passed, 6 skipped", "- Mutations: one PR opened", "- Workspace mutations: none", "- Current status: PR open", "- Blockers: none", "- Next: review PR #999", "- Safety: no self-review; no self-merge; no secrets", ]) def test_report_without_handoff_is_downgraded(self): result = assess_controller_handoff("long report text, no handoff") self.assertEqual(result["verdict"], "missing") self.assertTrue(result["downgraded"]) def test_wrong_title_is_downgraded(self): text = self.BASE_HANDOFF.replace( "## Controller Handoff", "## Handoff Summary") result = assess_controller_handoff(text) self.assertEqual(result["verdict"], "missing") def test_complete_base_handoff_passes(self): result = assess_controller_handoff( "full report body...\n\n" + self.BASE_HANDOFF) self.assertEqual(result["verdict"], "complete") self.assertFalse(result["downgraded"]) def test_missing_base_fields_are_listed(self): text = "\n".join( line for line in self.BASE_HANDOFF.splitlines() if not line.startswith(("- Mutations:", "- Safety:"))) result = assess_controller_handoff(text) self.assertEqual(result["verdict"], "incomplete") self.assertTrue(result["downgraded"]) self.assertIn("Mutations", result["missing_fields"]) self.assertIn("Safety", result["missing_fields"]) def test_review_role_requires_review_fields(self): # Issue #320: review handoffs must not carry the legacy # 'Workspace mutations' line, so strip it from the shared base. review_base = "\n".join( line for line in self.BASE_HANDOFF.splitlines() if not line.startswith("- Workspace mutations:")) result = assess_controller_handoff(review_base, role="review") self.assertEqual(result["verdict"], "incomplete") self.assertIn("Pinned reviewed head", result["missing_fields"]) self.assertIn("Worktree path", result["missing_fields"]) self.assertIn("Merge result", result["missing_fields"]) complete = review_base + "\n" + "\n".join([ "- Selected PR: #999", "- Reviewer eligibility: passed", "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Worktree path: /repo/branches/review-pr-999", "- Worktree dirty: no", "- Scratch worktree used: yes (/repo/branches/review-pr-999)", "- Unrelated local mutations: none", "- Review decision: approve", "- Merge result: merged", "- Linked issue status: closed", "- Cleanup status: branch deleted", "- File edits by reviewer: none", "- Worktree/index mutations: created and removed review worktree", "- Git ref mutations: git fetch prgs", "- MCP/Gitea mutations: none", "- Review mutations: one approve on #999", "- Merge mutations: PR #999 merged", "- Cleanup mutations: removed review worktree", "- External-state mutations: none", "- Read-only diagnostics: git log, git diff", ]) 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([ "- 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", ]) res = assess_controller_handoff(incomplete_eq, role="author") self.assertEqual(res["verdict"], "incomplete") self.assertIn("Selected issue", res["missing_fields"]) # 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", ]) res = assess_controller_handoff(incomplete_multi, role="author") self.assertEqual(res["verdict"], "incomplete") self.assertIn("Selected issue", res["missing_fields"]) 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", ]) res = assess_controller_handoff(incomplete_pr, role="author") self.assertEqual(res["verdict"], "incomplete") self.assertIn("PR number opened", res["missing_fields"]) def test_inventory_role_requires_inventory_fields(self): 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", ]) result = assess_controller_handoff(complete, role="inventory") self.assertEqual(result["verdict"], "complete") def test_skill_doc_declares_handoff_requirement(self): # Doc-contract: SKILL.md must keep requiring the exact section and # naming this validator, or the convention silently rots. skill = ( __import__("pathlib").Path(__file__).resolve().parent.parent / "skills" / "llm-project-workflow" / "SKILL.md" ).read_text(encoding="utf-8") self.assertIn("## Controller Handoff", skill) self.assertIn("assess_controller_handoff", skill) self.assertIn("issue #182", skill) self.assertIn("## Work Selection Rule for LLMs", skill) self.assertIn("work already claimed", skill) self.assertIn("## Global LLM Worktree Rule", skill) self.assertIn("branches/", skill) 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)) 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)) res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True) self.assertEqual(res2["verdict"], "complete") class TestReviewHandoffPreciseMutationCategories(unittest.TestCase): """Issue #320: reviewer handoffs drop legacy 'Workspace mutations'.""" REVIEW_BASE = "\n".join([ "## Controller Handoff", "", "- Task: review PR #999", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: reviewer", "- Identity: jcwalker3 / prgs-reviewer", "- Issue/PR: PR #999", "- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Files changed: review_proofs.py", "- Validation: 700 passed, 6 skipped", "- Mutations: one review submitted", "- Current status: review complete", "- Blockers: none", "- Next: controller decides", "- Safety: no self-review; no self-merge; no secrets", "- Selected PR: #999", "- Reviewer eligibility: passed", "- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9", "- Worktree path: /repo/branches/review-pr-999", "- Worktree dirty: no", "- Scratch worktree used: yes (/repo/branches/review-pr-999)", "- Unrelated local mutations: none", "- Review decision: approve", "- Merge result: none", "- Linked issue status: open", "- Cleanup status: worktree removed", ]) PRECISE_CATEGORIES = "\n".join([ "- File edits by reviewer: none", "- Worktree/index mutations: created and removed review worktree", "- Git ref mutations: git fetch prgs", "- MCP/Gitea mutations: none", "- Review mutations: one approve on #999", "- Merge mutations: none", "- Cleanup mutations: removed review worktree", "- External-state mutations: none", "- Read-only diagnostics: git log, git diff", ]) def _complete_handoff(self, **overrides): text = self.REVIEW_BASE + "\n" + self.PRECISE_CATEGORIES for old, new in overrides.items(): text = text.replace(old, new) return text def test_review_handoff_with_precise_categories_is_complete(self): result = assess_controller_handoff( self._complete_handoff(), role="review") self.assertEqual(result["verdict"], "complete") self.assertFalse(result["downgraded"]) def test_review_handoff_rejects_legacy_workspace_mutations_none(self): text = self._complete_handoff() + "\n- Workspace mutations: none" result = assess_controller_handoff(text, role="review") self.assertEqual(result["verdict"], "incomplete") self.assertTrue(result["downgraded"]) self.assertTrue(any( "workspace mutations" in reason.lower() for reason in result["reasons"] )) def test_review_handoff_rejects_legacy_workspace_mutations_any_value(self): text = (self._complete_handoff() + "\n- Workspace mutations: edited review_proofs.py") result = assess_controller_handoff(text, role="review") self.assertEqual(result["verdict"], "incomplete") self.assertTrue(any( "workspace mutations" in reason.lower() for reason in result["reasons"] )) def test_review_handoff_missing_precise_categories_is_incomplete(self): result = assess_controller_handoff(self.REVIEW_BASE, role="review") self.assertEqual(result["verdict"], "incomplete") self.assertIn("File edits by reviewer", result["missing_fields"]) self.assertIn("Worktree/index mutations", result["missing_fields"]) self.assertIn("Read-only diagnostics", result["missing_fields"]) def test_review_handoff_does_not_require_workspace_mutations(self): result = assess_controller_handoff( self._complete_handoff(), role="review") self.assertNotIn("Workspace mutations", result["missing_fields"]) def test_request_changes_handoff_with_precise_categories_is_complete(self): text = self._complete_handoff(**{ "- Review decision: approve": "- Review decision: request-changes", "- Review mutations: one approve on #999": "- Review mutations: one request_changes on #999", }) result = assess_controller_handoff(text, role="review") self.assertEqual(result["verdict"], "complete") def test_merge_handoff_with_precise_categories_is_complete(self): text = self._complete_handoff(**{ "- Merge result: none": "- Merge result: merged", "- Merge mutations: none": "- Merge mutations: PR #999 merged", }) result = assess_controller_handoff(text, role="review") self.assertEqual(result["verdict"], "complete") def test_fetch_only_handoff_with_precise_categories_is_complete(self): text = self._complete_handoff(**{ "- Review decision: approve": "- Review decision: none", "- Review mutations: one approve on #999": "- Review mutations: none", "- Worktree/index mutations: created and removed review worktree": "- Worktree/index mutations: none", "- Cleanup mutations: removed review worktree": "- Cleanup mutations: none", "- Mutations: one review submitted": "- Mutations: fetch only", }) result = assess_controller_handoff(text, role="review") self.assertEqual(result["verdict"], "complete") def test_blocked_handoff_with_precise_categories_is_complete(self): text = self._complete_handoff(**{ "- Blockers: none": "- Blockers: infra_stop (registry unreachable)", "- Review decision: approve": "- Review decision: none", "- Review mutations: one approve on #999": "- Review mutations: none", }) result = assess_controller_handoff(text, role="review") self.assertEqual(result["verdict"], "complete") def test_author_role_still_requires_workspace_mutations(self): # Non-review roles keep the legacy contract until their own issues # migrate them. author = TestControllerHandoff.BASE_HANDOFF stripped = "\n".join( line for line in author.splitlines() if not line.startswith("- Workspace mutations:")) result = assess_controller_handoff(stripped, role="author") self.assertEqual(result["verdict"], "incomplete") self.assertIn("Workspace mutations", result["missing_fields"]) class TestReviewMutationFinalReport(unittest.TestCase): """Final reports must list exactly one live review mutation.""" LOCK_ONE = { "live_mutations": [{"pr_number": 203, "action": "request_changes"}], "correction_authorized": False, } def test_single_mutation_report_complete(self): report = ( "Review complete on PR #203.\n" "Review mutations: one request_changes on #203.\n" "Review decision: request_changes." ) result = assess_review_mutation_final_report(report, self.LOCK_ONE) self.assertTrue(result["complete"]) self.assertFalse(result["downgraded"]) def test_missing_mutation_details_downgraded(self): result = assess_review_mutation_final_report( "Review complete. No details.", self.LOCK_ONE ) self.assertFalse(result["complete"]) self.assertTrue(result["downgraded"]) def test_two_mutations_require_correction_explanation(self): lock = { "live_mutations": [ {"pr_number": 203, "action": "approve"}, {"pr_number": 203, "action": "request_changes"}, ], "correction_authorized": True, "correction_reason": "operator approved correcting mistaken approve", } incomplete = assess_review_mutation_final_report( "Submitted approve then request_changes on #203.", lock ) self.assertFalse(incomplete["complete"]) complete = assess_review_mutation_final_report( "\n".join([ "Review mutations: approve (mistake), then request_changes (final).", "Correction flow: operator approved correcting mistaken approve on PR #203.", ]), lock, ) self.assertTrue(complete["complete"]) def test_build_final_report_wires_review_mutation_from_report_text(self): report_text = ( "Review complete on PR #203.\n" "Review mutations: one request_changes on #203." ) lock = { "live_mutations": [{"pr_number": 203, "action": "request_changes"}], "correction_authorized": False, } 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(), report_text=report_text, review_decision_lock=lock, ) self.assertTrue(final["review_mutation_complete"]) class TestQueueTargetReconciliation(unittest.TestCase): """Queue target lock: reconcile operator-supplied backlog before inventory (#200).""" CONFIGURED = [ "Scaled-Tech-Consulting/Gitea-Tools", "Scaled-Tech-Consulting/mcp-control-plane", ] GITEA_TOOLS = "Scaled-Tech-Consulting/Gitea-Tools" MCP = "Scaled-Tech-Consulting/mcp-control-plane" OPERATOR_CONTEXT = ( "six open PRs in Scaled-Tech-Consulting/Gitea-Tools including " "#195, #193, #192, #190, #187, and #181" ) PROFILE = { "profile_name": "prgs-reviewer", "allowed_operations": ["read", "gitea.read"], } def test_supplied_gitea_tools_prs_but_inventoried_mcp_is_mismatch(self): lock = reconcile_queue_target( operator_context=self.OPERATOR_CONTEXT, inventoried_repo=self.MCP, configured_repos=self.CONFIGURED, ) self.assertEqual(lock["status"], "target_repo_mismatch") self.assertEqual(lock["resolved_repo"], self.GITEA_TOOLS) self.assertEqual(lock["resolution_source"], "operator_context") self.assertIn(195, lock["supplied_pr_numbers"]) self.assertFalse(lock["allow_clean_stop"]) self.assertFalse(lock["allow_trusted_empty"]) def test_wrong_repo_zero_open_cannot_stop_cleanly(self): lock = reconcile_queue_target( operator_context=self.OPERATOR_CONTEXT, inventoried_repo=self.MCP, configured_repos=self.CONFIGURED, ) gate = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="mcp-control-plane", state="open", authenticated_profile=self.PROFILE, local_remote_url=( "https://gitea.prgs.cc/Scaled-Tech-Consulting/" "mcp-control-plane.git" ), corroboration_open_pr_counter=0, queue_target_lock=lock, ) self.assertEqual(gate["status"], "target_repo_mismatch") self.assertNotEqual(gate["status"], "trusted_empty") def test_supplied_pr_numbers_reconciled_before_empty_stop(self): lock = reconcile_queue_target( operator_context=( "currently we have 6 open PRs: #195, #193, #192, " "#190, #187, #181 in Gitea-Tools" ), inventoried_repo=self.GITEA_TOOLS, configured_repos=self.CONFIGURED, ) self.assertEqual(lock["status"], "resolved") self.assertEqual(len(lock["supplied_pr_numbers"]), 6) self.assertTrue(all(item["matches_inventoried_repo"] for item in lock["reconciliation"])) def test_ambiguous_repo_context_fails_closed(self): lock = reconcile_queue_target( operator_context=( "open PRs in mcp-control-plane and gitea-tools including PR #195" ), inventoried_repo=self.MCP, configured_repos=self.CONFIGURED, ) self.assertEqual(lock["status"], "unresolved") self.assertFalse(lock["allow_trusted_empty"]) self.assertTrue( any("could not be resolved" in r for r in lock["reasons"]) ) def test_conflicting_directive_vs_backlog_fails_closed(self): lock = reconcile_queue_target( operator_context="Repository: Scaled-Tech-Consulting/mcp-control-plane", supplied_pr_backlog=[ {"number": 195, "repo": self.GITEA_TOOLS}, ], inventoried_repo=self.MCP, configured_repos=self.CONFIGURED, ) self.assertEqual(lock["status"], "unresolved") self.assertIn("conflicts", " ".join(lock["reasons"]).lower()) def test_final_report_must_document_reconciliation(self): lock = reconcile_queue_target( operator_context=self.OPERATOR_CONTEXT, inventoried_repo=self.GITEA_TOOLS, configured_repos=self.CONFIGURED, ) incomplete = assess_queue_target_final_report( "Open PRs: 0. Stopping.", lock ) self.assertFalse(incomplete["complete"]) self.assertTrue(incomplete["downgraded"]) complete_report = "\n".join([ "Queue inventory complete.", f"Resolved repo: {self.GITEA_TOOLS}", "Resolution source: operator_context", "queue_target_lock.status: resolved", "Supplied PR reconciliation: #195, #193, #192, #190, #187, #181", ]) complete = assess_queue_target_final_report(complete_report, lock) self.assertTrue(complete["complete"]) self.assertFalse(complete["downgraded"]) def test_assess_reviewer_queue_inventory_blocks_mismatch(self): result = assess_reviewer_queue_inventory( [{ "repo": self.MCP, "state_filter": "open", "pagination_complete": True, "open_pr_count": 0, "list_prs_response": [], "remote": "prgs", "authenticated_profile": self.PROFILE, "local_remote_url": ( "https://gitea.prgs.cc/Scaled-Tech-Consulting/" "mcp-control-plane.git" ), }], operator_context=self.OPERATOR_CONTEXT, ) self.assertFalse(result["can_claim_empty_queue"]) self.assertIn("target_repo_mismatch", str(result["trust_gates"])) class TestPRInventoryTrustGate(unittest.TestCase): """Issue #194: unit tests for the PR inventory trust gate.""" def setUp(self): self.profile = { "profile_name": "prgs-reviewer", "allowed_operations": ["read", "gitea.read", "gitea.pr.approve"], } self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" def test_trusted_nonempty(self): res = pr_inventory_trust_gate([{"number": 1}]) self.assertEqual(res["status"], "trusted_nonempty") self.assertFalse(res["corroborated"]) def test_inventory_error_none_or_not_list(self): self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error") self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error") def test_untrusted_empty_no_pagination_or_corroboration(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=None, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"]) def test_untrusted_empty_profile_permission_mismatch(self): bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]} res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=bad_profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertIn("authenticated profile lacks read permissions", res["reasons"]) def test_untrusted_empty_remote_url_mismatch(self): bad_url = "https://gitea.prgs.cc/other-org/other-repo.git" res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=bad_url, user_context=None, corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"]) def test_untrusted_empty_user_context_indicates_prs(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context="please check open PR #181", corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "untrusted_empty") self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"])) def test_trusted_empty_with_corroboration(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=0, has_finality_metadata=False ) self.assertEqual(res["status"], "trusted_empty") self.assertTrue(res["corroborated"]) def test_trusted_empty_with_finality_metadata(self): res = pr_inventory_trust_gate( [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", state="open", authenticated_profile=self.profile, local_remote_url=self.local_url, user_context=None, corroboration_open_pr_counter=None, has_finality_metadata=True ) self.assertEqual(res["status"], "trusted_empty") self.assertTrue(res["corroborated"]) class TestAssessReviewerQueueInventory(unittest.TestCase): """Issue #196: trust gate wired into canonical queue inventory.""" def _repo_report(self, **overrides): report = { "repo": "Scaled-Tech-Consulting/Gitea-Tools", "state_filter": "open", "pagination_complete": True, "open_pr_count": 0, "list_prs_response": [], "remote": "prgs", "authenticated_profile": { "profile_name": "prgs-reviewer", "allowed_operations": ["read", "gitea.read"], }, "local_remote_url": ( "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" ), } report.update(overrides) return report def test_empty_without_corroboration_blocks_empty_queue_claim(self): result = assess_reviewer_queue_inventory([ self._repo_report(pagination_complete=False), self._repo_report( repo="Scaled-Tech-Consulting/mcp-control-plane", open_pr_count=0, pagination_complete=False, ), ]) self.assertFalse(result["can_claim_empty_queue"]) self.assertIn("untrusted_empty", str(result["trust_gates"])) def test_trusted_empty_with_finality_allows_empty_queue_claim(self): result = assess_reviewer_queue_inventory([ self._repo_report(), self._repo_report( repo="Scaled-Tech-Consulting/mcp-control-plane", local_remote_url=( "https://gitea.prgs.cc/Scaled-Tech-Consulting/" "mcp-control-plane.git" ), ), ]) self.assertTrue(result["can_claim_empty_queue"]) self.assertTrue(result["can_claim_exhaustive"]) def test_user_context_indicating_open_prs_blocks_empty_claim(self): result = assess_reviewer_queue_inventory( [self._repo_report()], user_context="please review open PR #195 in the queue", ) self.assertFalse(result["can_claim_empty_queue"]) self.assertTrue(result["blockers"]) def test_nonempty_inventory_skips_empty_trust_gate_block(self): result = assess_reviewer_queue_inventory([ self._repo_report(open_pr_count=2), self._repo_report( repo="Scaled-Tech-Consulting/mcp-control-plane", open_pr_count=1, ), ]) self.assertTrue(result["complete"]) 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.""" def test_evidence_backed_claims_are_proven(self): result = _good_capability_evidence() self.assertTrue(result["proven"]) self.assertEqual(result["reasons"], []) def test_claim_without_evidence_source_is_not_proven(self): result = assess_capability_evidence([ {"task": "review_pr", "allowed": True, "evidence_source": ""}, ]) self.assertFalse(result["proven"]) self.assertTrue(any("evidence" in r.lower() for r in result["reasons"])) def test_no_claims_at_all_fails_closed(self): result = assess_capability_evidence([]) self.assertFalse(result["proven"]) def test_disallowed_task_is_not_proven(self): result = assess_capability_evidence([ { "task": "merge_pr", "allowed": False, "evidence_source": "gitea_resolve_task_capability output", }, ]) self.assertFalse(result["proven"]) class TestSweepEvidence(unittest.TestCase): """#179 gap 2: secret/provenance sweep must be exact.""" def test_exact_sweep_is_proven(self): result = _good_sweep() self.assertEqual(result["verdict"], "exact") self.assertTrue(result["proven"]) def test_vague_sweep_without_command_is_downgraded(self): result = _good_sweep(command="") self.assertEqual(result["verdict"], "vague") self.assertFalse(result["proven"]) def test_sweep_without_scope_is_downgraded(self): result = _good_sweep(scope="") self.assertEqual(result["verdict"], "vague") self.assertFalse(result["proven"]) def test_missing_sweep_fails_closed(self): result = assess_sweep_evidence(None) self.assertEqual(result["verdict"], "missing") self.assertFalse(result["proven"]) def test_unstated_result_is_downgraded(self): result = _good_sweep(clean=None) self.assertFalse(result["proven"]) class TestLiveStateRecheck(unittest.TestCase): """#179 gap 3: explicit pre-mutation live-state recheck.""" def test_clean_recheck_is_proven(self): result = _good_live_state() self.assertTrue(result["proven"]) self.assertFalse(result["block"]) def test_missing_recheck_fails_closed(self): result = assess_live_state_recheck(None) self.assertFalse(result["proven"]) self.assertTrue(result["block"]) def test_closed_pr_blocks(self): result = _good_live_state(pr_state="closed") self.assertFalse(result["proven"]) self.assertTrue(result["block"]) def test_moved_head_blocks(self): result = _good_live_state(live_head_sha=OTHER) self.assertFalse(result["proven"]) self.assertTrue(any("head" in r.lower() for r in result["reasons"])) def test_changed_base_blocks(self): result = _good_live_state(live_base_ref="develop") self.assertFalse(result["proven"]) def test_unresolved_blocking_reviews_block(self): result = _good_live_state(blocking_change_requests=True) self.assertFalse(result["proven"]) def test_unchecked_blocking_state_fails_closed(self): result = _good_live_state(blocking_change_requests=None) self.assertFalse(result["proven"]) class TestRoleBoundary179(unittest.TestCase): """#179 gap 4: reviewer flows avoid unjustified author-namespace use.""" def test_native_namespace_only_is_clean(self): result = _good_role_boundary_179() self.assertTrue(result["proven"]) def test_foreign_namespace_without_justification_is_downgraded(self): result = _good_role_boundary_179( namespaces_used=["gitea-reviewer", "gitea-author"] ) self.assertFalse(result["proven"]) self.assertTrue(any("justif" in r.lower() for r in result["reasons"])) def test_foreign_namespace_with_justification_is_clean(self): result = _good_role_boundary_179( namespaces_used=["gitea-reviewer", "gitea-author"], justification=( "author namespace read-only whoami used to evidence " "self-review contamination status" ), ) self.assertTrue(result["proven"]) def test_unreported_namespaces_fail_closed(self): result = _good_role_boundary_179(namespaces_used=None) self.assertFalse(result["proven"]) class TestFinalReport179Bar(unittest.TestCase): """#179 acceptance adds capability, sweep, live-state, and role proofs.""" def _report(self, **overrides): kwargs = { "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-feat-issue-224", "porcelain_status": "", "pr_scope_files": ["docs/wiki/Repositories.md"], "scratch_used": True, "scratch_path": "/repo/branches/review-feat-issue-224", }, } kwargs.update(overrides) return build_final_report(**kwargs) def test_all_179_proofs_present_is_grade_a(self): report = self._report() self.assertEqual(report["grade"], "A") self.assertTrue(report["capability_evidence_proven"]) self.assertEqual(report["sweep_verdict"], "exact") self.assertTrue(report["live_state_recheck_proven"]) self.assertTrue(report["role_boundary_clean"]) def test_missing_capability_evidence_downgrades(self): report = self._report(capability_evidence=None) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["capability_evidence_proven"]) def test_unevidenced_capability_claim_downgrades(self): report = self._report( capability_evidence=assess_capability_evidence([ {"task": "review_pr", "allowed": True, "evidence_source": ""}, ]) ) self.assertNotEqual(report["grade"], "A") def test_vague_sweep_downgrades(self): report = self._report(sweep=_good_sweep(command="")) self.assertNotEqual(report["grade"], "A") self.assertEqual(report["sweep_verdict"], "vague") def test_missing_sweep_downgrades(self): report = self._report(sweep=None) self.assertNotEqual(report["grade"], "A") def test_missing_live_state_recheck_downgrades_and_blocks_merge(self): report = self._report(live_state=None) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) self.assertFalse(report["live_state_recheck_proven"]) def test_stale_live_state_blocks_merge(self): report = self._report(live_state=_good_live_state(live_head_sha=OTHER)) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) def test_merge_claim_without_live_recheck_is_a_violation(self): report = self._report(live_state=None, merge_performed=True) self.assertEqual(report["grade"], "blocked") self.assertTrue(report["violations"]) def test_unjustified_author_namespace_downgrades(self): report = self._report( role_boundary=_good_role_boundary_179( namespaces_used=["gitea-reviewer", "gitea-author"] ) ) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["role_boundary_clean"]) def test_positive_baseline_from_173_still_holds(self): report = self._report() self.assertTrue(report["inventory_complete"]) self.assertTrue(report["validated_on_pinned_head"]) def test_missing_review_mutation_downgrades(self): report = self._report(review_mutation=None) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["review_mutation_complete"]) def test_incomplete_review_mutation_downgrades(self): report = self._report(review_mutation={"complete": False, "downgraded": True, "reasons": []}) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["review_mutation_complete"]) class TestAuthorReporting(unittest.TestCase): """Harness assertions for author reporting and capability/sweep proofs.""" def test_good_capability_proof_passes(self): result = _good_capability_proof() self.assertTrue(result["proven"]) def test_missing_capability_proof_fails_closed(self): result = assess_capability_proof({}) self.assertFalse(result["proven"]) self.assertTrue(any("fail closed" in r for r in result["reasons"])) def test_unresolved_capability_proof_fails_closed(self): result = assess_capability_proof({ "mark_issue": { "requested_task": "unknown_task", "required_operation_permission": None, "allowed_in_current_session": None, } }) self.assertFalse(result["proven"]) self.assertTrue( any("could not be resolved" in r for r in result["reasons"]) ) def test_good_secret_sweep_passes(self): result = _good_secret_sweep() self.assertTrue(result["proven"]) self.assertEqual(result["verdict"], "strong") def test_vague_secret_sweep_without_method_is_weak(self): result = _good_secret_sweep(method="") self.assertFalse(result["proven"]) self.assertEqual(result["verdict"], "weak") self.assertTrue( any("method" in r or "scan" in r for r in result["reasons"]) ) def test_unconfirmed_secret_sweep_is_weak(self): result = _good_secret_sweep(clean=False) self.assertFalse(result["proven"]) self.assertEqual(result["verdict"], "weak") def test_good_author_pr_report_passes(self): result = assess_author_pr_report({ "pr_number": 185, "branch": "feat/issue-184-repo-name-disambiguation", "head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae", }) self.assertTrue(result["complete"]) def test_author_pr_report_without_head_sha_is_incomplete(self): result = assess_author_pr_report({ "pr_number": 185, "branch": "feat/issue-184-repo-name-disambiguation", "head_sha": "", }) self.assertFalse(result["complete"]) self.assertTrue(any("head SHA" in r for r in result["reasons"])) def test_author_pr_report_rejects_short_sha(self): result = assess_author_pr_report({ "pr_number": 185, "branch": "feat/issue-184-repo-name-disambiguation", "head_sha": "e2bccba", }) 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") class TestIssueFilingFinalReport(unittest.TestCase): """Issue #191: issue-filing runs need A-bar final report proofs.""" ISSUE_TITLE = ( "Implement fail-closed continuation mode for issues already " "represented by open PRs" ) FULL_SHA = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9" CAPABILITIES = { "create_issue": { "requested_task": "create_issue", "required_operation_permission": "gitea.issue.create", "allowed_in_current_session": True, }, } CLOSEST = [{"number": 183, "title": "Harden author-run reporting", "state": "open"}] def _good_report(self, *, sha_line=None): lines = [ "Created Gitea-Tools #189 — " f"`{self.ISSUE_TITLE}`", "Duplicate check: searched 12 open issues.", "Closest existing: #183 — Harden author-run reporting " "(update rejected — different scope).", "Why new issue justified: continuation mode is distinct from #183.", "Only mutation: issue creation (gitea.issue.create).", "Confirm no labels, comments, PRs, reviews, merges, or closes.", "## Controller Handoff", "- Task: file issue", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: author", "- Identity: prgs-author", "- Issue/PR: #189", "- Branch/SHA: n/a", "- Files changed: none", "- Validation: duplicate gate + create_issue capability resolved", "- Mutations: create_issue via gitea.issue.create", "- Workspace mutations: none", "- Current status: issue created", "- Blockers: none", "- Next: implementation", "- Safety: no review/merge", "- Issue created or updated: created #189", "- Related issues: #183, #188", ] if sha_line: lines.insert(4, sha_line) return "\n".join(lines) def test_complete_issue_filing_report_earns_a(self): result = assess_issue_filing_final_report( self._good_report(), issue_number=189, issue_title=self.ISSUE_TITLE, action="created", mutations=["create_issue"], resolved_capabilities=self.CAPABILITIES, issues_searched=12, closest_matches=self.CLOSEST, performed_mutations=["create_issue"], ) self.assertEqual(result["grade"], "A") self.assertFalse(result["downgraded"]) def test_missing_controller_handoff_downgrades(self): report = self._good_report().replace("## Controller Handoff", "") result = assess_issue_filing_final_report( report, issue_number=189, issue_title=self.ISSUE_TITLE, mutations=["create_issue"], resolved_capabilities=self.CAPABILITIES, issues_searched=12, performed_mutations=["create_issue"], ) self.assertTrue(result["downgraded"]) def test_abbreviated_sha_downgrades(self): result = assess_issue_filing_sha_evidence( f"old head: {self.FULL_SHA[:7]}" ) self.assertTrue(result["downgraded"]) def test_mutation_without_capability_proof_downgrades(self): report = self._good_report().replace("gitea.issue.create", "allowed") result = assess_issue_filing_mutation_capability( report, ["create_issue"], self.CAPABILITIES, ) self.assertTrue(result["downgraded"]) def test_label_mutation_without_label_proof_downgrades(self): caps = { "set_issue_labels": { "requested_task": "set_issue_labels", "required_operation_permission": "gitea.issue.comment", "allowed_in_current_session": True, }, } report = "\n".join([ "Updated labels with gitea.issue.comment capability resolved.", "Only mutations: set_issue_labels", ]) result = assess_issue_filing_mutation_capability( report, ["set_issue_labels"], caps ) self.assertTrue(result["downgraded"]) class TestCreateIssueFinalReport(unittest.TestCase): """#337: create-issue final-report verifier coverage.""" ISSUE_TITLE = "Extract create-issue workflow from monolithic LLM project skill" CAPABILITIES = { "create_issue": { "requested_task": "create_issue", "required_operation_permission": "gitea.issue.create", "allowed_in_current_session": True, }, } CLOSEST = [{"number": 333, "title": "Split monolithic workflow", "state": "closed"}] def _good_report(self): return "\n".join([ "Task mode: create-issue", "Workflow source: skills/llm-project-workflow/workflows/create-issue.md", f"Updated issue #337 — `{self.ISSUE_TITLE}` (remaining verifier AC).", "Duplicate check: searched 54 open issues.", "Closest existing: #333 — Split monolithic workflow (partial overlap).", "Why new issue justified: dedicated create-issue extraction.", "Only mutation: issue creation (gitea.issue.create).", "Confirm no labels, comments, PRs, reviews, merges, or closes.", "## Controller Handoff", "- Task: create-issue", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: author", "- Identity: jcwalker3 / prgs-author", "- Active profile: prgs-author", "- Runtime context: prgs-author on prgs", "- Requested issue task: extract create-issue workflow verifier", "- Workflow source: workflows/create-issue.md @ fd396df", "- Capability proof: create_issue allowed (gitea.issue.create)", "- Duplicate search terms: create-issue, workflow extract", "- Duplicate search pagination proof: 54 open issues < limit 500", "- Duplicates found: #333 partial only", "- Issues created: none (implementation PR for remaining AC)", "- Issues commented: none", "- Issues edited: none", "- Issues skipped as duplicates: none", "- File edits by issue creator: review_proofs.py, tests", "- Worktree/index mutations: feat/issue-337 worktree created", "- Git ref mutations: fetch prgs/master", "- MCP/Gitea mutations: gitea_mark_issue start on #337", "- Issue mutations: claimed #337", "- Label/assignment/milestone mutations: none", "- External-state mutations: none", "- Read-only diagnostics: whoami, list issues", "- Blockers: none", "- Current status: implementing verifier", "- Safe next action: open PR", "- Safety statement: no review/merge/repo main-checkout edits", "- Issue/PR: #337", "- Branch/SHA: feat/issue-337-create-issue-verifier", "- Files changed: review_proofs.py, tests", "- Validation: pytest tests/test_review_proofs.py", "- Mutations: create_issue capability resolved; mark_issue only", "- Next: open PR", "- Safety: no review/merge", ]) def test_complete_create_issue_report_earns_a(self): result = assess_create_issue_final_report( self._good_report(), issue_number=337, issue_title=self.ISSUE_TITLE, action="updated", mutations=["create_issue"], resolved_capabilities=self.CAPABILITIES, issues_searched=54, closest_matches=self.CLOSEST, performed_mutations=["create_issue"], ) self.assertEqual(result["grade"], "A") self.assertFalse(result["downgraded"]) def test_missing_workflow_source_downgrades(self): report = self._good_report().replace("workflows/create-issue.md", "SKILL.md only") result = assess_create_issue_workflow_source(report) self.assertTrue(result["downgraded"]) def test_work_issue_field_in_handoff_downgrades(self): report = self._good_report().replace( "- Safety statement:", "- Selected issue: #123\n- Safety statement:", ) result = assess_create_issue_mode_isolation(report) self.assertTrue(result["downgraded"]) self.assertTrue( any("Selected issue" in r for r in result["reasons"]) ) def test_review_merge_field_in_handoff_downgrades(self): report = self._good_report().replace( "- Safety statement:", "- Review decision: APPROVE\n- Safety statement:", ) result = assess_create_issue_mode_isolation(report) self.assertTrue(result["downgraded"]) def test_legacy_workspace_mutations_without_precise_categories_downgrades(self): report = self._good_report().replace( "- File edits by issue creator:", "- Workspace mutations: none\n- File edits by issue creator:", ) lines = report.splitlines() cleaned = [ line for line in lines if not line.strip().startswith("- File edits by issue creator:") ] report = "\n".join(cleaned) result = assess_create_issue_mode_isolation(report) self.assertTrue(result["downgraded"]) class TestWorkIssueFinalReport(unittest.TestCase): """#139: work-issue final-report verifier coverage.""" def _good_report(self): return "\n".join([ "Task mode: work-issue", "Workflow source: skills/llm-project-workflow/workflows/work-issue.md", "## Controller Handoff", "- Task: work-issue", "- Repo: Scaled-Tech-Consulting/Gitea-Tools", "- Role: author", "- Identity: jcwalker3 / prgs-author", "- Active profile: prgs-author", "- Runtime context: prgs-author on prgs", "- Selected issue: #139", "- Branch name: feat/issue-139-role-aware-work-issue-routing", "- PR number: not created in this session", "- File edits by author: role_session_router.py", "- Blockers: none", "- Current status: implementing routing", "- Safe next action: open PR", "- Next: open PR", "- Safety statement: no review/merge", ]) def test_complete_work_issue_report_earns_a(self): result = assess_work_issue_final_report(self._good_report()) self.assertEqual(result["grade"], "A") self.assertFalse(result["downgraded"]) def test_missing_workflow_source_downgrades(self): report = self._good_report().replace( "workflows/work-issue.md", "SKILL.md only" ) result = assess_work_issue_workflow_source(report) self.assertTrue(result["downgraded"]) def test_review_field_in_handoff_downgrades(self): report = self._good_report().replace( "- Safety statement:", "- Review decision: APPROVE\n- Safety statement:", ) result = assess_work_issue_mode_isolation(report) self.assertTrue(result["downgraded"]) class TestValidationEnvironmentProof(unittest.TestCase): """Issue #311: exact validation command and execution environment.""" ROOT = "/repo/Gitea-Tools/branches/review-pr-280" def test_venv_pytest_with_full_report_passes(self): result = assess_validation_environment_proof( report_text=( "Validation command: venv/bin/pytest tests/ --ignore=branches\n" f"Working directory: {self.ROOT}\n" "Result: 1014 passed, 6 skipped" ), ) self.assertTrue(result["proven"]) def test_bare_pytest_without_path_proof_fails(self): result = assess_validation_environment_proof( report_text=( "Validation command: pytest\n" f"Working directory: {self.ROOT}\n" "1014 passed, 6 skipped" ), ) self.assertFalse(result["proven"]) def test_bare_pytest_with_path_and_version_proof_passes(self): result = assess_validation_environment_proof( report_text=( "Validation command: pytest tests/\n" f"Working directory: {self.ROOT}\n" "which pytest: /repo/Gitea-Tools/venv/bin/pytest\n" "pytest --version: pytest 8.3.4\n" "Resolves to the project venv: yes\n" "1014 passed, 6 skipped" ), ) self.assertTrue(result["proven"]) def test_vague_pytest_summary_without_command_fails(self): result = assess_validation_environment_proof( report_text=( f"Working directory: {self.ROOT}\n" "pytest executed inside the worktree: 1014 passed, 6 skipped" ), ) self.assertFalse(result["proven"]) def test_structured_environment_proof_passes(self): result = assess_validation_environment_proof( validation_environment={ "command": "pytest tests/", "working_directory": self.ROOT, "which_pytest": "/repo/Gitea-Tools/venv/bin/pytest", "pytest_version": "pytest 8.3.4", "venv_resolved": True, "passed": 1014, "skipped": 6, }, ) self.assertTrue(result["proven"]) def test_missing_working_directory_fails(self): result = assess_validation_environment_proof( report_text=( "Validation command: venv/bin/pytest tests/\n" "1014 passed, 6 skipped" ), ) self.assertFalse(result["proven"]) def test_build_final_report_downgrades_missing_environment_proof(self): report = 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-feat-issue-224", "porcelain_status": "", "pr_scope_files": ["docs/wiki/Repositories.md"], "scratch_used": True, }, report_text=( f"Working directory: {self.ROOT}\n" "pytest executed: 1014 passed, 6 skipped" ), ) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["validation_environment_proven"]) result = assess_queue_ordering_proof( report_text="Selected the next eligible PR: PR #286.", ) self.assertFalse(result["proven"]) self.assertTrue(result["claims_selection"]) def test_next_eligible_claim_with_policy_passes(self): result = assess_queue_ordering_proof( report_text=( "Queue ordering policy: oldest PR number first\n" "Selected the next eligible PR: PR #286." ), ) self.assertTrue(result["proven"]) self.assertEqual(result["policy"], "oldest PR number first") def test_newest_first_api_with_oldest_first_selection_passes(self): result = assess_queue_ordering_proof( report_text=( "Queue ordering policy: oldest PR number first\n" "API order was newest-first (PR #291 before PR #286).\n" "Selected the next eligible PR: PR #286." ), queue_ordering={ "policy": "oldest PR number first", "selected_pr_number": 286, "api_pr_numbers": [291, 286], }, ) self.assertTrue(result["proven"]) def test_earlier_actionable_pr_without_skip_reason_fails(self): result = assess_queue_ordering_proof( report_text=( "Queue ordering policy: oldest PR number first\n" "Selected the next eligible PR: PR #300." ), queue_ordering={ "policy": "oldest PR number first", "selected_pr_number": 300, "api_pr_numbers": [300, 286], }, ) self.assertFalse(result["proven"]) self.assertTrue(result["block"]) def test_priority_override_requires_explanation(self): result = assess_queue_ordering_proof( report_text="Selected the next eligible PR: PR #300.", queue_ordering={ "policy": "priority label", "selected_pr_number": 300, "api_pr_numbers": [286, 300], }, ) self.assertFalse(result["proven"]) def test_priority_override_with_explanation_passes(self): result = assess_queue_ordering_proof( report_text=( "Queue ordering policy: priority label\n" "Priority override: security label on PR #300.\n" "Selected the next eligible PR: PR #300." ), queue_ordering={ "policy": "priority label", "selected_pr_number": 300, "api_pr_numbers": [286, 300], "priority_override": "security label on PR #300", }, ) self.assertTrue(result["proven"]) def test_build_final_report_downgrades_missing_ordering_proof(self): report = 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-feat-issue-224", "porcelain_status": "", "pr_scope_files": ["docs/wiki/Repositories.md"], "scratch_used": True, }, report_text="Selected the next eligible PR: PR #286.", ) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["queue_ordering_proven"]) class TestReviewerBaselineValidationProof(unittest.TestCase): """Issue #325: baseline validation must use branches/ worktrees.""" ROOT = "/repo/Gitea-Tools" def _good_baseline_proof(self, **overrides): proof = { "worktree_path": f"{self.ROOT}/branches/baseline-master-pr280", "baseline_target_sha": PINNED, "pr_head_sha": OTHER, "baseline_failures": ["tests/test_foo.py::test_bar"], "pr_failures": ["tests/test_foo.py::test_bar"], "failure_signatures_match": True, "clean_before": True, "clean_after": True, } proof.update(overrides) return proof def test_main_checkout_test_run_blocks(self): result = assess_reviewer_baseline_validation_proof( validation_runs=[ { "command": "venv/bin/pytest tests/", "working_directory": self.ROOT, "project_root": self.ROOT, } ], project_root=self.ROOT, ) self.assertFalse(result["proven"]) self.assertTrue(result["block"]) self.assertTrue(result["violations"]) def test_branches_worktree_test_run_passes(self): result = assess_reviewer_baseline_validation_proof( validation_runs=[ { "command": "venv/bin/pytest tests/", "working_directory": f"{self.ROOT}/branches/review-pr-280", "project_root": self.ROOT, } ], project_root=self.ROOT, ) self.assertTrue(result["proven"]) def test_preexisting_claim_without_baseline_proof_fails(self): result = assess_reviewer_baseline_validation_proof( report_text=( "Validation: full-suite failures are pre-existing on master." ), project_root=self.ROOT, ) self.assertFalse(result["proven"]) self.assertTrue(result["claims_preexisting"]) def test_preexisting_claim_with_complete_baseline_proof_passes(self): result = assess_reviewer_baseline_validation_proof( report_text="Failures are pre-existing on master.", baseline_proof=self._good_baseline_proof(), project_root=self.ROOT, ) self.assertTrue(result["proven"]) def test_mismatched_failures_block_preexisting_claim(self): result = assess_reviewer_baseline_validation_proof( report_text="Same as master.", baseline_proof=self._good_baseline_proof( failure_signatures_match=False, pr_failures=["tests/test_other.py::test_other"], ), project_root=self.ROOT, ) self.assertFalse(result["proven"]) def test_report_parsed_main_checkout_cwd_blocks(self): result = assess_reviewer_baseline_validation_proof( report_text=( "- Validation command: venv/bin/pytest tests/\n" f"- Working directory: {self.ROOT}\n" ), project_root=self.ROOT, ) self.assertFalse(result["proven"]) self.assertTrue(result["block"]) def test_build_final_report_downgrades_main_checkout_validation(self): report = 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-feat-issue-224", "porcelain_status": "", "pr_scope_files": ["docs/wiki/Repositories.md"], "scratch_used": True, }, report_text=( "- Validation command: venv/bin/pytest tests/\n" f"- Working directory: {self.ROOT}\n" ), project_root=self.ROOT, ) self.assertNotEqual(report["grade"], "A") self.assertFalse(report["baseline_validation_proven"]) class TestPartialReconciliationPlan(unittest.TestCase): """Issue #302: policy when PR-close capability is missing (Option B).""" def _plan(self, **overrides): kwargs = { "ancestor_proven": True, "capabilities": { "close_pr": False, "comment_pr": True, "close_issue": True, "comment_issue": True, }, } kwargs.update(overrides) return resolve_partial_reconciliation_plan(**kwargs) def test_close_capability_present_allows_full_reconcile(self): plan = self._plan(capabilities={ "close_pr": True, "comment_pr": True, "close_issue": True, "comment_issue": True, }) self.assertEqual(plan["outcome"], "FULL_RECONCILE_CLOSE_ALLOWED") self.assertIn("close_pr", plan["allowed_mutations"]) def test_close_missing_with_comment_posts_comment_then_stops(self): plan = self._plan() self.assertEqual( plan["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP") self.assertEqual(plan["allowed_mutations"], ("comment_pr",)) for field in ( "PR head SHA", "target branch SHA", "ancestor proof", "linked issue status", "required missing capability", ): self.assertIn(field, plan["required_comment_fields"]) def test_comment_capability_missing_produces_handoff_only(self): plan = self._plan(capabilities={ "close_pr": False, "comment_pr": False, "close_issue": True, "comment_issue": True, }) self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") self.assertEqual(plan["allowed_mutations"], ()) def test_all_capabilities_missing_produces_handoff_only(self): plan = self._plan(capabilities={ "close_pr": False, "comment_pr": False, "close_issue": False, "comment_issue": False, }) self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") self.assertEqual(plan["allowed_mutations"], ()) def test_unproven_ancestry_blocks_all_mutations(self): plan = self._plan(ancestor_proven=False, capabilities={ "close_pr": True, "comment_pr": True, "close_issue": True, "comment_issue": True, }) self.assertEqual(plan["outcome"], "GATE_NOT_PROVEN") self.assertEqual(plan["allowed_mutations"], ()) def test_missing_capability_dict_fails_closed(self): plan = self._plan(capabilities=None) self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") self.assertEqual(plan["allowed_mutations"], ()) class TestPartialReconciliationReport(unittest.TestCase): """Issue #302: reports must explain why comments were or were not posted.""" def _plan(self, outcome): capabilities = { "FULL_RECONCILE_CLOSE_ALLOWED": { "close_pr": True, "comment_pr": True, "close_issue": True, "comment_issue": True, }, "PARTIAL_RECONCILE_COMMENT_THEN_STOP": { "close_pr": False, "comment_pr": True, "close_issue": True, "comment_issue": True, }, "RECOVERY_HANDOFF_ONLY": { "close_pr": False, "comment_pr": False, "close_issue": False, "comment_issue": False, }, }[outcome] return resolve_partial_reconciliation_plan( ancestor_proven=True, capabilities=capabilities) def test_partial_report_requires_comment_and_missing_capability(self): plan = self._plan("PARTIAL_RECONCILE_COMMENT_THEN_STOP") good = ( "Reconciliation comment posted on PR #278 with ancestor proof. " "PR close skipped: close capability gitea.pr.close missing; " "stopped for authorized close." ) result = assess_partial_reconciliation_report(good, plan=plan) self.assertTrue(result["complete"]) bad = "Stopped. Nothing done." result = assess_partial_reconciliation_report(bad, plan=plan) self.assertTrue(result["block"]) def test_handoff_only_report_requires_no_comment_explanation(self): plan = self._plan("RECOVERY_HANDOFF_ONLY") good = ( "No reconciliation comment posted: comment capability missing. " "No Gitea mutation performed; recovery handoff produced." ) result = assess_partial_reconciliation_report(good, plan=plan) self.assertTrue(result["complete"]) bad = "Recovery handoff produced." result = assess_partial_reconciliation_report(bad, plan=plan) self.assertTrue(result["block"]) def test_full_reconcile_report_requires_close_result(self): plan = self._plan("FULL_RECONCILE_CLOSE_ALLOWED") good = "PR close result: closed PR #278 after ancestor proof." result = assess_partial_reconciliation_report(good, plan=plan) self.assertTrue(result["complete"]) bad = "Reconciliation done." result = assess_partial_reconciliation_report(bad, plan=plan) self.assertTrue(result["block"]) class TestAlreadyLandedReviewGate(unittest.TestCase): """Issue #292: PR head already on target blocks approval and merge.""" def _gate(self, **overrides): kwargs = { "pr_number": 278, "candidate_head_sha": PINNED, "target_branch": "master", "target_branch_sha": OTHER, "head_is_ancestor_of_target": True, "live_head_sha": PINNED, } kwargs.update(overrides) return assess_already_landed_review_gate(**kwargs) def test_already_landed_blocks_approval_and_merge(self): result = self._gate() self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") self.assertFalse(result["approve_allowed"]) self.assertFalse(result["merge_api_allowed"]) self.assertFalse(result["request_changes_allowed"]) self.assertTrue(result["reconciliation_handoff_required"]) def test_already_landed_with_real_blocker_allows_request_changes(self): result = self._gate(real_blocker=True) self.assertEqual(result["state"], "ALREADY_LANDED_RECONCILE_REQUIRED") self.assertFalse(result["approve_allowed"]) self.assertFalse(result["merge_api_allowed"]) self.assertTrue(result["request_changes_allowed"]) def test_normal_candidate_passes_gate(self): result = self._gate(head_is_ancestor_of_target=False) self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") self.assertTrue(result["approve_allowed"]) self.assertTrue(result["merge_api_allowed"]) self.assertFalse(result["reconciliation_handoff_required"]) def test_non_mergeable_normal_pr_still_passes_this_gate(self): # Mergeability/conflicts are separate gates; ancestry gate only # decides already-landed vs normal candidate. result = self._gate( head_is_ancestor_of_target=False, mergeable=False) self.assertEqual(result["state"], "NORMAL_REVIEW_CANDIDATE") self.assertTrue(result["request_changes_allowed"]) def test_unchecked_ancestry_blocks_review_mutations(self): result = self._gate(head_is_ancestor_of_target=None) self.assertEqual(result["state"], "GATE_NOT_PROVEN") self.assertFalse(result["approve_allowed"]) self.assertFalse(result["merge_api_allowed"]) def test_changed_head_since_pin_blocks_until_recheck(self): result = self._gate( head_is_ancestor_of_target=False, live_head_sha=OTHER) self.assertEqual(result["state"], "GATE_NOT_PROVEN") self.assertFalse(result["approve_allowed"]) self.assertTrue(any( "head" in reason.lower() for reason in result["reasons"] )) def test_invalid_shas_block_gate(self): result = self._gate(candidate_head_sha="deadbeef") self.assertEqual(result["state"], "GATE_NOT_PROVEN") self.assertFalse(result["approve_allowed"]) result = self._gate(target_branch_sha=None) self.assertEqual(result["state"], "GATE_NOT_PROVEN") def test_already_landed_handoff_lists_required_fields(self): result = self._gate() for field in ( "PR number/title", "candidate head SHA", "target branch", "target branch SHA", "ancestor proof", "linked issue status", "recommended reconciliation action", "capability proof for close/comment mutations", ): self.assertIn(field, result["required_handoff_fields"]) class TestAlreadyLandedReportState(unittest.TestCase): """Issue #292: reports cannot say APPROVED after the gate fired.""" GOOD_REPORT = ( "Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED\n" "Candidate head SHA: " + PINNED + "\n" "Review decision: none\n" "Merge result: none\n" ) def test_gate_fired_with_reconcile_state_passes(self): result = assess_already_landed_report_state( self.GOOD_REPORT, gate_fired=True) self.assertTrue(result["complete"]) self.assertFalse(result["block"]) def test_gate_fired_with_approved_state_blocks(self): report = self.GOOD_REPORT.replace( "Review decision: none", "Review decision: APPROVED") result = assess_already_landed_report_state(report, gate_fired=True) self.assertTrue(result["block"]) def test_gate_fired_with_merged_state_blocks(self): report = self.GOOD_REPORT.replace( "Merge result: none", "Merge result: MERGED") result = assess_already_landed_report_state(report, gate_fired=True) self.assertTrue(result["block"]) def test_gate_fired_with_ready_to_merge_blocks(self): result = assess_already_landed_report_state( self.GOOD_REPORT + "Current status: READY_TO_MERGE\n", gate_fired=True) self.assertTrue(result["block"]) def test_gate_fired_without_reconcile_state_blocks(self): result = assess_already_landed_report_state( "Current status: review complete.\n", gate_fired=True) self.assertTrue(result["block"]) self.assertTrue(any( "already_landed_reconcile_required" in reason.lower() for reason in result["reasons"] )) def test_gate_not_fired_reports_pass_untouched(self): result = assess_already_landed_report_state( "Review decision: APPROVED\nMerge result: MERGED\n", gate_fired=False) self.assertTrue(result["complete"]) class TestFullSuiteFailureApprovalGate(unittest.TestCase): """Issue #323: full-suite failure blocks approval without baseline proof.""" ROOT = "/repo/Gitea-Tools" def _good_proof(self, **overrides): proof = { "worktree_path": f"{self.ROOT}/branches/baseline-master-pr280", "baseline_target_sha": PINNED, "pr_head_sha": OTHER, "baseline_failures": ["tests/test_foo.py::test_bar"], "pr_failures": ["tests/test_foo.py::test_bar"], "failure_signatures_match": True, "clean_before": True, "clean_after": True, "pr_suite_command": "venv/bin/pytest tests/", "baseline_suite_command": "venv/bin/pytest tests/", "new_tests_passed": True, "unrelated_explanation": ( "failures touch agent_temp_artifacts only; PR changes " "review_proofs handoff fields" ), } proof.update(overrides) return proof def _good_report(self): return ( "Full suite failed on PR worktree and on baseline.\n" "- Validation command: venv/bin/pytest tests/\n" f"- Working directory: {self.ROOT}/branches/review-pr-280\n" "Failures are pre-existing on master.\n" ) def test_full_suite_pass_allows_approval_without_baseline(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=True, project_root=self.ROOT, ) self.assertTrue(result["approve_allowed"]) self.assertFalse(result["block"]) self.assertEqual(result["default_action"], "proceed") def test_full_suite_failure_defaults_to_request_changes(self): result = assess_full_suite_failure_approval_gate( review_decision="request_changes", full_suite_passed=False, project_root=self.ROOT, ) self.assertFalse(result["block"]) self.assertEqual(result["default_action"], "request_changes") def test_approval_with_failure_and_no_proof_blocks(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) self.assertTrue(result["block"]) self.assertEqual(result["default_action"], "request_changes") def test_approval_with_matching_baseline_proof_allowed(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, baseline_proof=self._good_proof(), report_text=self._good_report(), project_root=self.ROOT, ) self.assertTrue(result["approve_allowed"]) self.assertFalse(result["block"]) def test_mismatched_failure_signatures_block_approval(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, baseline_proof=self._good_proof( failure_signatures_match=False, pr_failures=["tests/test_other.py::test_other"], ), report_text=self._good_report(), project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) self.assertTrue(result["block"]) def test_main_checkout_baseline_blocks_approval(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, baseline_proof=self._good_proof(worktree_path=self.ROOT), report_text=self._good_report(), project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) self.assertTrue(result["block"]) def test_missing_new_tests_proof_blocks_approval(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, baseline_proof=self._good_proof(new_tests_passed=None), report_text=self._good_report(), project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) self.assertTrue(any( "new" in reason.lower() and "test" in reason.lower() for reason in result["reasons"] )) def test_missing_unrelated_explanation_blocks_approval(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, baseline_proof=self._good_proof(unrelated_explanation=""), report_text=self._good_report(), project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) def test_missing_suite_commands_block_approval(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, baseline_proof=self._good_proof( pr_suite_command="", baseline_suite_command=""), report_text=self._good_report(), project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) self.assertTrue(any( "command" in reason.lower() for reason in result["reasons"] )) def test_unknown_suite_result_blocks_approval(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=None, project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) self.assertTrue(result["block"]) def test_vague_same_as_master_claim_without_proof_blocks(self): result = assess_full_suite_failure_approval_gate( review_decision="approve", full_suite_passed=False, report_text="Validation: failures same as master; approving.", project_root=self.ROOT, ) self.assertFalse(result["approve_allowed"]) self.assertTrue(result["block"]) def test_non_approve_decision_never_blocks(self): result = assess_full_suite_failure_approval_gate( review_decision="comment", full_suite_passed=None, project_root=self.ROOT, ) self.assertFalse(result["block"]) if __name__ == "__main__": unittest.main() if __name__ == "__main__": unittest.main()