diff --git a/review_proofs.py b/review_proofs.py new file mode 100644 index 0000000..7b08213 --- /dev/null +++ b/review_proofs.py @@ -0,0 +1,363 @@ +"""Fail-closed proof helpers for the blind PR queue review workflow (#173). + +A blind queue review must *prove* its preconditions instead of asserting +them. Each helper here evaluates one proof from issue #173's required +behaviors and returns a plain dict verdict; ``build_final_report`` combines +them and only awards an "A" grade when every proof is present. Missing +evidence always downgrades or blocks — never upgrades. + +These helpers are pure (no network, no git calls): the workflow gathers the +raw facts (Gitea PR head SHA, ``git rev-parse HEAD``, listing pagination +state, pytest output, whoami results) and passes them in, so the same logic +is usable from prompts, harness assertions, and tests. The MCP-level gates +(mcp-control-plane #68/#76) remain authoritative for mutations; nothing +here weakens or replaces them. +""" + +import re + +_FULL_SHA = re.compile(r"^[0-9a-f]{40}$") + +SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION = ( + "evidence missing: report contamination as unknown and " + "choose another PR or stop" +) + + +def _normalize_ref(ref): + """Normalize a git ref for base comparison. + + Strips ``refs/heads/`` and ``refs/remotes//`` prefixes and a + plain ``/`` prefix so ``master``, ``prgs/master`` and + ``refs/remotes/prgs/master`` all compare equal. Returns '' for empty + input (which then fails closed in the caller). + """ + ref = (ref or "").strip() + if ref.startswith("refs/heads/"): + ref = ref[len("refs/heads/"):] + elif ref.startswith("refs/remotes/"): + ref = ref[len("refs/remotes/"):] + ref = ref.split("/", 1)[1] if "/" in ref else ref + elif "/" in ref: + # Remote-tracking shorthand like 'prgs/master'. + ref = ref.split("/", 1)[1] + return ref + + +def verify_pinned_head_checkout(pinned_head_sha, local_head_sha, + pr_base_ref, diff_base_ref): + """Required behaviors 1–2: prove HEAD == pinned PR head, correct base. + + Both SHAs must be full 40-hex and identical — an abbreviated or prefix + match is not proof. The diff base must normalize to the PR base branch. + Returns {'proven', 'block', 'reasons', 'pinned_head_sha', + 'local_head_sha'}; ``block`` is True whenever the proof fails, meaning + review and merge must stop before proceeding. + """ + reasons = [] + pinned = (pinned_head_sha or "").strip().lower() + local = (local_head_sha or "").strip().lower() + + if not pinned: + reasons.append("pinned PR head SHA missing; fail closed") + elif not _FULL_SHA.match(pinned): + reasons.append("pinned PR head SHA is not a full 40-hex SHA; fail closed") + + if not local: + reasons.append("local checkout HEAD SHA missing; fail closed") + elif not _FULL_SHA.match(local): + reasons.append( + "local checkout HEAD SHA is not a full 40-hex SHA " + "(abbreviated SHAs are not proof); fail closed" + ) + + if pinned and local and _FULL_SHA.match(pinned) and _FULL_SHA.match(local): + if pinned != local: + reasons.append( + "local checkout HEAD does not match the pinned PR head SHA; " + "stop before review/merge" + ) + + base = _normalize_ref(pr_base_ref) + diff_base = _normalize_ref(diff_base_ref) + if not base: + reasons.append("PR base ref missing; fail closed") + if not diff_base: + reasons.append("diff base ref missing; fail closed") + if base and diff_base and base != diff_base: + reasons.append( + f"diff base '{diff_base}' is not the PR base branch '{base}'" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "pinned_head_sha": pinned or None, + "local_head_sha": local or None, + } + + +def assess_inventory_completeness(repo_reports, required_repos): + """Required behavior 3: the queue inventory must prove completeness. + + *repo_reports* is a list of dicts, one per repository listed: + ``{'repo', 'state_filter', 'pagination_complete', 'open_pr_count'}``. + The inventory is complete only when every required repository has a + report that states its filter, proves pagination was handled, and gives + a total open-PR count. ``can_claim_exhaustive`` — permission to say + "these are the only open PRs" — is granted only for a complete + inventory. + """ + reasons = [] + by_repo = {} + for report in repo_reports or []: + name = (report.get("repo") or "").strip() + if name: + by_repo[name] = report + + totals = {} + for repo in required_repos: + report = by_repo.get(repo) + if report is None: + reasons.append(f"repository '{repo}' was not inventoried") + continue + if not (report.get("state_filter") or "").strip(): + reasons.append(f"repository '{repo}': PR state filter not stated") + if report.get("pagination_complete") is not True: + reasons.append( + f"repository '{repo}': pagination not proven complete; " + "a truncated listing cannot support an exhaustive claim" + ) + count = report.get("open_pr_count") + if not isinstance(count, int) or count < 0: + reasons.append( + f"repository '{repo}': total open PR count not reported" + ) + else: + totals[repo] = count + + complete = not reasons and bool(required_repos) + if not required_repos: + reasons.append("no required repositories configured; fail closed") + return { + "complete": complete, + "can_claim_exhaustive": complete, + "reasons": reasons, + "total_open_prs": totals, + } + + +def assess_validation_report(report): + """Required behavior 4: validation reporting needs exact evidence. + + *report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'), + ``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a + ``justification``), ``canonical_command``, ``deviation_justification``. + + Verdicts: + - 'invalid' — the result may not be claimed at all (no command stated, + or the command output was never read). + - 'weak' — claimable but downgraded (missing counts, unjustified + ignored paths, unexplained deviation from the canonical command). + - 'strong' — full evidence. + """ + reasons = [] + command = (report.get("command") or "").strip() + + if not command: + return { + "verdict": "invalid", + "claimable": False, + "reasons": ["no validation command stated; result cannot be claimed"], + } + if report.get("output_read") is not True: + return { + "verdict": "invalid", + "claimable": False, + "reasons": [ + "command output was not read to completion; a validation " + "result cannot be claimed" + ], + } + + counts = [report.get("passed"), report.get("failed"), report.get("skipped")] + if any(not isinstance(c, int) for c in counts): + reasons.append( + "test counts (passed/failed/skipped) missing; report is incomplete" + ) + + for ignored in report.get("ignored_paths") or []: + path = ignored.get("path") or "" + if not (ignored.get("justification") or "").strip(): + reasons.append( + f"ignored path '{path}' has no justification for why it is " + "safe to ignore" + ) + + canonical = (report.get("canonical_command") or "").strip() + if canonical and command != canonical: + if not (report.get("deviation_justification") or "").strip(): + reasons.append( + "command differs from the repository's canonical validation " + "command and the deviation is not justified" + ) + + verdict = "strong" if not reasons else "weak" + return {"verdict": verdict, "claimable": True, "reasons": reasons} + + +def assess_self_review_contamination(reviewer_identity, pr_author, + session_authored=None, + evidence_source=None): + """Required behavior 5: contamination is evidence-backed, never assumed. + + Distinguishes the two cases the issue calls out: + - reviewer identity == PR author → contaminated (the identities + themselves are the evidence); + - same-session authorship → contaminated only with an explicit + ``evidence_source``; a bare claim (either way) yields 'unknown'. + + 'unknown' is a stop signal: choose another PR or stop — it is never + silently treated as clean or as contaminated. + """ + reviewer = (reviewer_identity or "").strip() + author = (pr_author or "").strip() + evidence = (evidence_source or "").strip() + + if not reviewer or not author: + return { + "status": "unknown", + "reasons": ["reviewer identity or PR author unknown; fail closed"], + "safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION, + } + + if reviewer == author: + return { + "status": "contaminated", + "reasons": [ + f"authenticated reviewer '{reviewer}' is the PR author; " + "self-review is blocked" + ], + "safe_next_action": "choose another PR or stop", + } + + if session_authored is True: + if evidence: + return { + "status": "contaminated", + "reasons": [ + "this session authored/touched the PR branch " + f"(evidence: {evidence})" + ], + "safe_next_action": "choose another PR or stop", + } + return { + "status": "unknown", + "reasons": [ + "same-session authorship was claimed without an evidence " + "source; contamination may not be declared by assumption" + ], + "safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION, + } + + if session_authored is False and evidence: + return { + "status": "clean", + "reasons": [ + f"reviewer '{reviewer}' differs from author '{author}' and " + f"session non-authorship is evidenced ({evidence})" + ], + "safe_next_action": "proceed", + } + + return { + "status": "unknown", + "reasons": [ + "session authorship was not established with evidence; " + "contamination status is unknown" + ], + "safe_next_action": SAFE_NEXT_ACTION_UNKNOWN_CONTAMINATION, + } + + +def build_final_report(checkout_proof, inventory, validation, contamination, + identity_eligible, merge_performed, + issue_status_verified): + """Required behavior 6 + acceptance criteria: one report, distinct proofs. + + Combines the individual proof verdicts into the final-report fields the + issue requires, computes ``merge_allowed`` from the proofs, and grades + the run: + + - 'A' — every proof present (merge itself is optional). + - 'downgraded' — one or more proofs missing/weak; do not merge. + - 'blocked' — a *violation*: a merge was claimed although the proofs + did not allow one. + """ + contamination_status = contamination.get("status", "unknown") + checkout_proven = bool(checkout_proof.get("proven")) + validation_claimable = bool(validation.get("claimable")) + validation_strong = validation.get("verdict") == "strong" + + downgrade_reasons = [] + if not identity_eligible: + downgrade_reasons.append("identity/profile not eligible for review") + if not inventory.get("complete"): + downgrade_reasons.append("open-PR inventory completeness not proven") + downgrade_reasons.extend(inventory.get("reasons", [])) + if not checkout_proven: + downgrade_reasons.append("pinned-head checkout not proven") + downgrade_reasons.extend(checkout_proof.get("reasons", [])) + if not validation_strong: + downgrade_reasons.append( + f"validation evidence is {validation.get('verdict', 'missing')}" + ) + downgrade_reasons.extend(validation.get("reasons", [])) + if contamination_status != "clean": + downgrade_reasons.append( + f"session contamination status is '{contamination_status}'" + ) + if not issue_status_verified: + downgrade_reasons.append("linked issue status not verified") + + merge_allowed = ( + identity_eligible + and checkout_proven + and contamination_status == "clean" + and validation_claimable + and validation.get("verdict") != "invalid" + ) + + violations = [] + if merge_performed and not merge_allowed: + violations.append( + "merge was performed/claimed although the proofs did not allow " + "one; this run is blocked, not graded" + ) + + if violations: + grade = "blocked" + elif downgrade_reasons: + grade = "downgraded" + else: + grade = "A" + + return { + "grade": grade, + "violations": violations, + "downgrade_reasons": downgrade_reasons, + "identity_eligible": bool(identity_eligible), + "pr_author_distinct_from_reviewer": + contamination_status in ("clean",), + "session_contamination": contamination_status, + "inventory_complete": bool(inventory.get("complete")), + "validated_on_pinned_head": checkout_proven and validation_claimable, + "validation_passed": + validation_claimable and validation.get("verdict") != "invalid", + "validation_verdict": validation.get("verdict"), + "merge_allowed": merge_allowed, + "merge_performed": bool(merge_performed), + "issue_status_verified": bool(issue_status_verified), + } diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 0606699..db85a89 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -167,11 +167,38 @@ Worktree folder = branch with `/` replaced by `-` 1. Use a separate review worktree (`scripts/worktree-review `), detached. 2. Verify your authenticated identity. -3. Verify the PR author — **you must not be the author.** +3. Verify the PR author — **you must not be the author.** Self-review + contamination must be *evidence-backed* (#173): state the authenticated + reviewer identity, the PR author identity, whether this session + authored/touched the PR branch, and the evidence source for any + "same-session author" claim. If the evidence is missing, report the + status as **unknown** — never declare contamination by assumption — and + choose another PR or stop (`review_proofs.assess_self_review_contamination`). 4. Verify the worktree is clean. -5. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files. -6. Run the tests. -7. **Do not merge if checks fail. Do not merge if the reviewer is the author.** +5. **Checkout proof (#173):** before reviewing or validating, prove and + state: the selected PR head SHA from Gitea (pinned), the local checkout + SHA (`git rev-parse HEAD`), that `HEAD ==` the pinned PR head SHA, and + that the diff base is the PR base branch. If `HEAD` does not match the + pinned head, **stop before review/merge** + (`review_proofs.verify_pinned_head_checkout`). +6. **Inventory proof (#173):** a blind queue review must prove listing + completeness before claiming "only PRs found": both configured + repositories checked, open-PR filters stated, pagination handled or + explicitly not needed, and the total open PR count per repo reported + (`review_proofs.assess_inventory_completeness`). +7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files. +8. Run the tests. Validation reporting must include the exact command and + exact results: pass/fail, counts of tests passed/skipped/failed, any + ignored paths and why they are safe to ignore, and whether the command + differs from the repository's canonical validation command. Only claim a + validation result after the command has completed and its output has + been read (`review_proofs.assess_validation_report`). +9. **Do not merge if checks fail. Do not merge if the reviewer is the author.** +10. The final report must distinguish (`review_proofs.build_final_report`): + identity eligible; PR author different from reviewer; session + contamination absent (with evidence); validation performed on the pinned + head; merge performed; issue status verified. If any proof is missing, + stop or downgrade the result instead of merging confidently. ## G. Merge / cleanup workflow diff --git a/skills/llm-project-workflow/templates/review-pr.md b/skills/llm-project-workflow/templates/review-pr.md index 64cd42e..3897a57 100644 --- a/skills/llm-project-workflow/templates/review-pr.md +++ b/skills/llm-project-workflow/templates/review-pr.md @@ -20,13 +20,28 @@ Steps: *If the current identity does not match the required role (or is the PR author), STOP. Relaunch/switch to the correct profile first.* 2. Verify your authenticated identity (whoami) and the active profile. 3. Fetch the PR facts: PR author, head SHA, state (must be open), base branch. + Pin the head SHA in your notes; every later step validates THAT SHA. 4. If authenticated user == PR author → STOP (no self-review). -4. scripts/worktree-review # detached, branches/review-* + Session-contamination claims must be evidence-backed (#173): if you + cannot evidence whether this session authored/touched the PR branch, + report contamination as UNKNOWN (not contaminated, not clean) and choose + another PR or stop. +5. scripts/worktree-review # detached, branches/review-* cd branches/review- -5. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches +6. Checkout proof (#173) — prove and state, before any diff review or + validation: + - pinned PR head SHA (from Gitea) + - local checkout SHA (git rev-parse HEAD) + - HEAD == pinned PR head SHA + - diff base == the PR base branch + If HEAD does not match the pinned head → STOP before review/merge. +7. Confirm the worktree is clean. Inspect the FULL diff; confirm scope matches issue #; flag any unrelated files, secrets, or formatting churn. Check that the PR body correctly uses Gitea-closing keywords (`Closes #N` or `Fixes #N`) instead of non-closing ones (`Implements #N`, `Refs #N`). -6. Run the test suite; note results. -7. Post the review verdict: approve only if scope is clean and checks pass; +8. Run the test suite; report the exact command and exact results — pass/fail + plus passed/skipped/failed counts, any ignored paths and why they are safe + to ignore, and whether the command differs from the repository's canonical + validation command. Only claim a result after the output has been read. +9. Post the review verdict: approve only if scope is clean and checks pass; otherwise request changes with specifics. Never merge from this review step. Include a "Review Metadata" block (attribution only — docs/llm-agent-sha.md): diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py new file mode 100644 index 0000000..d67bfe6 --- /dev/null +++ b/tests/test_review_proofs.py @@ -0,0 +1,467 @@ +"""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 sys +import unittest + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +from review_proofs import ( # noqa: E402 + assess_inventory_completeness, + assess_self_review_contamination, + assess_validation_report, + build_final_report, + 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", + ) + + +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 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, + } + 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.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"]) + + +if __name__ == "__main__": + unittest.main()