diff --git a/docs/llm-workflow-runbooks.md b/docs/llm-workflow-runbooks.md index 1ab0b14..65279b1 100644 --- a/docs/llm-workflow-runbooks.md +++ b/docs/llm-workflow-runbooks.md @@ -379,6 +379,69 @@ touching anything. files, detected secret, or any production/deploy behavior — **stop, report the blocker, and take no mutating action.** Fail closed; never work around a gate. +## Task/role alignment (#167) + +The **requested task** decides what a session may do — not the credential it +happens to hold. Resolve the task first with +`gitea_resolve_task_capability(task=...)`: it returns `stop_required` and +`task_role_guidance` alongside the permission decision. An LLM asked to +review must never degrade into author work just because it is connected as +an author. + +| Requested task | Required identity/profile | Allowed | Forbidden | Stop when | +|---|---|---|---|---| +| Review PR (`review_pr`) | reviewer (e.g. `sysadmin` / `prgs-reviewer`) | read, gated review verdicts | commits, pushes, file edits, author comments, merge without eligibility | active profile is an author profile — stop immediately; do **not** switch to author-side fixes unless the operator explicitly re-tasks | +| Address PR change requests (`address_pr_change_requests`) | author (e.g. `jcwalker3` / `prgs-author`) | commit/push fixes to the PR branch, PR comment summarizing fixes | review verdicts, approve, request-changes, merge | active profile lacks branch push | +| Merge PR (`merge_pr`) | reviewer/merger | gated merge after eligibility + approval | merging own PR, merging without pinned head match | active profile is an author profile, or any merge gate fails | +| Comment on issue discussion (`comment_issue`) | any profile with `gitea.issue.comment` | issue thread comments | review verdicts, closing via comment | permission missing (`gitea.pr.comment` does **not** imply it) | +| Comment on PR (`comment_pr`) | any profile with `gitea.pr.comment` | PR thread comments | review verdicts | permission missing | +| Author implementation (`create_branch`/`push_branch`/`create_pr`) | author | branch, commit, push, open PR | self-review, self-merge | profile lacks the author permissions | + +If the task is review/merge and the session is an author profile, the only +correct outputs are: the read-only PR queue inventory (#164), the structured +permission report (#142), and a stop. Ask the operator to reconnect to the +reviewer namespace; a credential or profile swap in the same session never +cures same-session authorship. + +## Review feedback discovery (MCP-native) + +Formal review verdicts (APPROVED / REQUEST_CHANGES / COMMENT) live on the +review endpoints, **not** in the issue-comment thread. Never infer review +state from issue comments — use `gitea_get_pr_review_feedback(pr_number=...)` +(read-only, requires `gitea.read`). It reports: + +- every submitted review: reviewer, verdict, redacted body, timestamp, and + the head SHA it reviewed; +- `latest_review_state_by_reviewer` (PENDING drafts never count); +- `has_blocking_change_requests` / `approval_visible` (dismissed reviews do + not block); +- `current_head_sha` vs `latest_reviewed_head_sha`, `review_feedback_stale`, + and `author_pushed_after_request_changes` — so a reviewer can see whether + feedback predates new commits, and an author can see whether fixes have + been pushed since the REQUEST_CHANGES. + +A permission block returns `feedback_not_attempted: true` with a structured +permission report — distinct from a successful "no reviews yet" result, so a +blocked lookup is never misread as "no feedback exists". + +## Validation reporting discipline (#167) + +Validation results in handoffs and PR bodies must state exactly what ran and +what happened. Build the validation section with +`build_validation_report(...)` (in `mcp_server.py`) or follow its contract by +hand — every command is one of: + +- `passed` — the exact command ran and succeeded; +- `failed` — must include the exact command **and its output**; never + paraphrase ("shell-invocation quirks" is not a status); +- `skipped` — deliberately not run; name the reason and any targeted check + that replaced it; +- `not-run` — was required but never executed; say so plainly. + +Never imply full-suite success unless the full-suite command itself passed +(`full_suite_passed: true`). A report that hides a failed or skipped check +is worse than a failing report. + ## Controller Handoff (required, every task) Every task — implementation, review, merge, triage, documentation, diff --git a/mcp_server.py b/mcp_server.py index cb2f9c4..2a6d34a 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -776,6 +776,139 @@ def _redact(text: str) -> str: return out +# Review states that carry a submitted verdict. Gitea also emits PENDING +# (draft) and REQUEST_REVIEW (re-review ask); neither is a verdict and +# neither may drive the blocking/approval summaries. +_VERDICT_STATES = ("APPROVED", "REQUEST_CHANGES", "COMMENT") + + +@mcp.tool() +def gitea_get_pr_review_feedback( + pr_number: int, + remote: str = "dadeschools", + host: str | None = None, + org: str | None = None, + repo: str | None = None, +) -> dict: + """Read-only: discover formal PR review feedback through MCP (#167). + + Formal review verdicts (APPROVED / REQUEST_CHANGES / COMMENT) live on + Gitea's review endpoints, not in the issue-comment thread. This tool is + the MCP-native way to read them, so an LLM never has to infer review + state from issue comments. It never submits reviews and never mutates + anything; the profile must allow ``gitea.read`` (fail closed otherwise, + with a structured #142 permission report and no API call made). + + Endpoints: ``GET /repos/{owner}/{repo}/pulls/{n}`` (current head SHA) + and ``GET /repos/{owner}/{repo}/pulls/{n}/reviews``. + + Normal output is LLM-safe: review bodies pass through credential + redaction and no endpoint URLs appear. Set GITEA_MCP_REVEAL_ENDPOINTS=1 + (admin/debug opt-in) to include each review's web link. + + Args: + pr_number: The pull request index/number. + remote: Known instance — 'dadeschools' or 'prgs'. + host: Override the Gitea host. + org: Override the owner/organization. + repo: Override the repository name. + + Returns: + dict with 'success', 'pr_number', 'pr_state', 'current_head_sha', + 'reviews' (each: reviewer, verdict, body, submitted_at, + reviewed_head_sha, dismissed, stale), + 'latest_review_state_by_reviewer' (latest non-COMMENT verdict per + reviewer; PENDING drafts never count), + 'has_blocking_change_requests' (a reviewer's latest verdict is an + undismissed REQUEST_CHANGES), 'approval_visible', + 'latest_reviewed_head_sha' (head at the most recent + APPROVED/REQUEST_CHANGES review; COMMENT-only reviews never + advance it), 'review_feedback_stale' (that verdict feedback + predates the current head SHA), and + 'author_pushed_after_request_changes' (new commits landed after a + blocking REQUEST_CHANGES). On a permission block: 'success' False, + 'feedback_not_attempted' True, 'reasons', and 'permission_report' — + deliberately distinct from a successful "no reviews yet" result. + """ + reasons = _issue_comment_gate("gitea.read") + if reasons: + return { + "success": False, + "pr_number": pr_number, + "feedback_not_attempted": True, + "reasons": reasons, + "permission_report": _permission_block_report("gitea.read"), + } + h, o, r = _resolve(remote, host, org, repo) + auth = _auth(h) + base = f"{repo_api_url(h, o, r)}/pulls/{pr_number}" + pr = api_request("GET", base, auth) or {} + raw_reviews = api_request("GET", f"{base}/reviews", auth) or [] + current_head = (pr.get("head") or {}).get("sha") + reveal = _reveal_endpoints() + + ordered = sorted( + raw_reviews, + key=lambda rv: ((rv.get("submitted_at") or ""), rv.get("id") or 0), + ) + reviews = [] + latest_by_reviewer = {} + latest_reviewed_head = None + for rv in ordered: + state = (rv.get("state") or "").upper() + reviewer = (rv.get("user") or {}).get("login", "") + commit_id = rv.get("commit_id") + entry = { + "reviewer": reviewer, + "verdict": state, + "body": _redact(rv.get("body") or ""), + "submitted_at": rv.get("submitted_at"), + "reviewed_head_sha": commit_id, + "dismissed": bool(rv.get("dismissed")), + "stale": bool(rv.get("stale")) or bool( + commit_id and current_head and commit_id != current_head), + } + if reveal: + entry["url"] = rv.get("html_url") + reviews.append(entry) + # COMMENT reviews never advance the reviewed-head marker or the + # per-reviewer verdict — otherwise a drive-by comment on the + # current head would mask the staleness of an older undismissed + # REQUEST_CHANGES. + if state in _VERDICT_STATES and state != "COMMENT": + latest_reviewed_head = commit_id or latest_reviewed_head + if reviewer: + latest_by_reviewer[reviewer] = entry + + blocking = [ + e for e in latest_by_reviewer.values() + if e["verdict"] == "REQUEST_CHANGES" and not e["dismissed"] + ] + approvals = [ + e for e in latest_by_reviewer.values() + if e["verdict"] == "APPROVED" and not e["dismissed"] + ] + return { + "success": True, + "pr_number": pr_number, + "pr_state": pr.get("state"), + "current_head_sha": current_head, + "reviews": reviews, + "latest_review_state_by_reviewer": { + login: e["verdict"] for login, e in latest_by_reviewer.items()}, + "has_blocking_change_requests": bool(blocking), + "approval_visible": bool(approvals), + "latest_reviewed_head_sha": latest_reviewed_head, + "review_feedback_stale": bool( + latest_reviewed_head and current_head + and latest_reviewed_head != current_head), + "author_pushed_after_request_changes": any( + e["reviewed_head_sha"] and current_head + and e["reviewed_head_sha"] != current_head + for e in blocking), + } + + @mcp.tool() @_audit_pr_result("submit_pr_review") def gitea_submit_pr_review( @@ -3079,6 +3212,67 @@ def gitea_mirror_refs( "return_code": result.returncode, } +_VALIDATION_STATUSES = ("passed", "failed", "skipped", "not-run") + + +def build_validation_report(commands: list[dict]) -> dict: + """Build an explicit validation report for LLM final summaries (#167). + + Each entry must carry 'command' and a 'status' from 'passed' / + 'failed' / 'skipped' / 'not-run'. A failed entry must carry the exact + 'output'; vague language ("shell-invocation quirks") is rejected as an + unknown status and a missing failure output raises, so a report can + never hide what actually happened. Optional keys: 'reason' (for + skipped/not-run, e.g. what targeted check replaced it) and + 'full_suite': True to mark the full-test-suite command — + 'full_suite_passed' is True only if that entry passed, False if it + failed/was skipped/not run, and None when no entry is marked, so + full-suite success is never implied. + """ + entries = [] + full_suite_passed = None + for raw in commands: + command = (raw.get("command") or "").strip() + if not command: + raise ValueError("validation entry missing 'command' (fail closed)") + status = raw.get("status") + if status not in _VALIDATION_STATUSES: + raise ValueError( + f"unknown validation status {status!r} for '{command}'; " + f"use one of {_VALIDATION_STATUSES} (fail closed)") + output = raw.get("output") + if status == "failed" and not (output or "").strip(): + raise ValueError( + f"failed validation '{command}' must include the exact " + "command output (fail closed)") + entry = {"command": command, "status": status} + if output: + entry["output"] = output + if raw.get("reason"): + entry["reason"] = raw["reason"] + if raw.get("full_suite"): + entry["full_suite"] = True + full_suite_passed = status == "passed" + entries.append(entry) + label = {"passed": "PASSED", "failed": "FAILED", + "skipped": "SKIPPED", "not-run": "NOT-RUN"} + lines = [] + for e in entries: + line = f"{label[e['status']]}: {e['command']}" + if e.get("output"): + line += f" — {e['output']}" + if e.get("reason"): + line += f" ({e['reason']})" + lines.append(line) + return { + "entries": entries, + "all_passed": bool(entries) and all( + e["status"] == "passed" for e in entries), + "full_suite_passed": full_suite_passed, + "summary": "\n".join(lines), + } + + @mcp.tool() def gitea_resolve_task_capability( task: str, @@ -3128,6 +3322,10 @@ def gitea_resolve_task_capability( "permission": "gitea.pr.comment", "role": "author", }, + "address_pr_change_requests": { + "permission": "gitea.branch.push", + "role": "author", + }, "review_pr": { "permission": "gitea.pr.review", "role": "reviewer", @@ -3211,6 +3409,42 @@ def gitea_resolve_task_capability( "or use the corresponding MCP namespace." ) + # Task/role alignment guards (#167): the requested task, not the + # available credential, decides what the session may do. A review/merge + # task under a non-reviewer profile must stop — not silently degrade + # into author-side mutations. + stop_required = not allowed_in_current_session + task_role_guidance = [] + if required_role == "reviewer": + if allowed_in_current_session: + task_role_guidance.append( + "Review/merge task: proceed only through the gated " + "review/merge tools and their eligibility checks; " + "self-review and self-merge remain blocked.") + else: + task_role_guidance.append( + "STOP: the requested task is review/merge but the active " + "profile cannot perform it. Do not fall back to author-side " + "work — do not commit, push, edit files, or comment as " + "author unless the operator explicitly changes the task to " + "author-side remediation (task " + "'address_pr_change_requests').") + elif task == "address_pr_change_requests": + if allowed_in_current_session: + task_role_guidance.append( + "Author-side remediation: you may commit and push fixes to " + "the PR branch, but you must not submit review verdicts and " + "must not merge.") + else: + task_role_guidance.append( + "STOP: author-side remediation requires an author profile " + "with branch push permission; the active profile cannot " + "perform it.") + elif not allowed_in_current_session: + task_role_guidance.append( + "STOP: the active profile cannot perform the requested task; " + "follow exact_safe_next_action instead of improvising.") + return { "requested_task": task, "required_operation_permission": required_permission, @@ -3219,6 +3453,8 @@ def gitea_resolve_task_capability( "active_identity": username, "active_profile_allowed_operations": active_allowed, "allowed_in_current_session": allowed_in_current_session, + "stop_required": stop_required, + "task_role_guidance": task_role_guidance, "matching_configured_profile": matching_profiles, "runtime_switching_supported": switching, "different_mcp_namespace_required": different_namespace_required, diff --git a/tests/test_review_feedback.py b/tests/test_review_feedback.py new file mode 100644 index 0000000..01fff0c --- /dev/null +++ b/tests/test_review_feedback.py @@ -0,0 +1,458 @@ +"""Tests for issue #167: LLM role/task alignment and PR review feedback discovery. + +Covers: +- gitea_get_pr_review_feedback: MCP-native discovery of formal PR review + verdicts (APPROVED / REQUEST_CHANGES / COMMENT) without reading issue + comments, including blocking-state summary, staleness against the current + head SHA, redaction, and fail-closed permission gating. +- gitea_resolve_task_capability role guards: a review/merge task under an + author profile returns explicit stop guidance; author-side remediation is + distinguished as its own task kind. +- build_validation_report: validation reporting distinguishes passed, + failed, skipped, and not-run commands and never hides failures behind + vague language. +""" +import json +import os +import sys +import tempfile +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import gitea_config +import mcp_server +from mcp_server import ( + build_validation_report, + gitea_get_pr_review_feedback, + gitea_resolve_task_capability, +) + +FAKE_AUTH = "Basic dGVzdDp0ZXN0" + + +def _pr_details(head_sha="headsha2", state="open", author="author-user"): + return { + "number": 5, + "title": "PR 5", + "state": state, + "head": {"ref": "feature", "sha": head_sha}, + "base": {"ref": "master"}, + "mergeable": True, + "user": {"login": author}, + } + + +def _review(reviewer, state, commit_id="headsha2", body="", submitted_at="2026-07-05T10:00:00Z", + dismissed=False, review_id=1): + return { + "id": review_id, + "user": {"login": reviewer}, + "state": state, + "body": body, + "submitted_at": submitted_at, + "commit_id": commit_id, + "dismissed": dismissed, + "stale": False, + "html_url": "https://internal.example/pr/5#review", + } + + +class TestPRReviewFeedbackDiscovery(unittest.TestCase): + """Formal review feedback must be discoverable through MCP tools.""" + + def _profile(self, allowed=("gitea.read",)): + return { + "profile_name": "gitea-author", + "allowed_operations": list(allowed), + "forbidden_operations": [], + "base_url": None, + } + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_review_feedback_discoverable_without_issue_comments(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(), + [_review("reviewer1", "REQUEST_CHANGES", body="Please add tests.")], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(result["pr_number"], 5) + self.assertEqual(len(result["reviews"]), 1) + review = result["reviews"][0] + self.assertEqual(review["reviewer"], "reviewer1") + self.assertEqual(review["verdict"], "REQUEST_CHANGES") + self.assertEqual(review["body"], "Please add tests.") + self.assertIn("submitted_at", review) + self.assertTrue(result["has_blocking_change_requests"]) + self.assertFalse(result["approval_visible"]) + self.assertEqual( + result["latest_review_state_by_reviewer"], {"reviewer1": "REQUEST_CHANGES"}) + # only GET calls; discovery never mutates + for call in mock_api.call_args_list: + self.assertEqual(call.args[0], "GET") + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_latest_state_per_reviewer_and_approval_visible(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(), + [ + _review("reviewer1", "REQUEST_CHANGES", + submitted_at="2026-07-05T10:00:00Z", review_id=1), + _review("reviewer1", "APPROVED", + submitted_at="2026-07-05T11:00:00Z", review_id=2), + ], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertEqual( + result["latest_review_state_by_reviewer"], {"reviewer1": "APPROVED"}) + self.assertFalse(result["has_blocking_change_requests"]) + self.assertTrue(result["approval_visible"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_change_request_staleness_after_new_commits(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(head_sha="newhead3"), + [_review("reviewer1", "REQUEST_CHANGES", commit_id="oldhead1")], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertEqual(result["current_head_sha"], "newhead3") + self.assertEqual(result["latest_reviewed_head_sha"], "oldhead1") + self.assertTrue(result["review_feedback_stale"]) + self.assertTrue(result["author_pushed_after_request_changes"]) + self.assertEqual(result["reviews"][0]["reviewed_head_sha"], "oldhead1") + self.assertTrue(result["reviews"][0]["stale"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_feedback_current_when_review_matches_head(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(head_sha="headsha2"), + [_review("reviewer1", "REQUEST_CHANGES", commit_id="headsha2")], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertFalse(result["review_feedback_stale"]) + self.assertFalse(result["author_pushed_after_request_changes"]) + self.assertTrue(result["has_blocking_change_requests"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_dismissed_change_requests_not_blocking(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(), + [_review("reviewer1", "REQUEST_CHANGES", dismissed=True)], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertFalse(result["has_blocking_change_requests"]) + self.assertTrue(result["reviews"][0]["dismissed"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_later_comment_does_not_mask_change_request_staleness(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(head_sha="newhead3"), + [ + _review("reviewer1", "REQUEST_CHANGES", commit_id="oldhead1", + submitted_at="2026-07-05T10:00:00Z", review_id=1), + _review("reviewer2", "COMMENT", commit_id="newhead3", + submitted_at="2026-07-05T11:00:00Z", review_id=2), + ], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertEqual(result["latest_reviewed_head_sha"], "oldhead1") + self.assertTrue(result["review_feedback_stale"]) + self.assertTrue(result["author_pushed_after_request_changes"]) + self.assertTrue(result["has_blocking_change_requests"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_dismissed_approval_not_visible(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(), + [_review("reviewer1", "APPROVED", dismissed=True)], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertFalse(result["approval_visible"]) + self.assertTrue(result["reviews"][0]["dismissed"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_pending_reviews_do_not_set_latest_state(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(), + [ + _review("reviewer1", "APPROVED", + submitted_at="2026-07-05T10:00:00Z", review_id=1), + _review("reviewer1", "PENDING", + submitted_at="2026-07-05T11:00:00Z", review_id=2), + ], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertEqual( + result["latest_review_state_by_reviewer"], {"reviewer1": "APPROVED"}) + self.assertTrue(result["approval_visible"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_review_bodies_are_redacted(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(), + [_review("reviewer1", "COMMENT", + body="failure mentions token supersecret123 in log")], + ] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + dumped = json.dumps(result) + self.assertNotIn("supersecret123", dumped) + self.assertIn("[REDACTED]", result["reviews"][0]["body"]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_no_url_leak_without_reveal_opt_in(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [ + _pr_details(), + [_review("reviewer1", "APPROVED")], + ] + with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": ""}): + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + dumped = json.dumps(result) + self.assertNotIn("internal.example", dumped) + self.assertNotIn("url", result["reviews"][0]) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_permission_block_fails_closed_without_api_calls(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile(allowed=()) + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertFalse(result["success"]) + self.assertTrue(result["feedback_not_attempted"]) + self.assertNotIn("reviews", result) + self.assertIn("permission_report", result) + self.assertTrue(result["reasons"]) + self.assertEqual(mock_api.call_count, 0) + + @patch("mcp_server.api_request") + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("mcp_server.get_profile") + def test_no_reviews_is_distinct_from_not_attempted(self, mock_get_profile, _auth, mock_api): + mock_get_profile.return_value = self._profile() + mock_api.side_effect = [_pr_details(), []] + result = gitea_get_pr_review_feedback(pr_number=5, remote="prgs") + self.assertTrue(result["success"]) + self.assertEqual(result["reviews"], []) + self.assertFalse(result.get("feedback_not_attempted", False)) + self.assertFalse(result["has_blocking_change_requests"]) + self.assertFalse(result["approval_visible"]) + + +CONFIG_GUARD = { + "version": 2, + "contexts": { + "ctx": { + "enabled": True, + "gitea": {"enabled": True, "base_url": "https://gitea.example.com"}, + } + }, + "profiles": { + "author-profile": { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "author-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": [ + "gitea.read", "gitea.issue.create", "gitea.pr.create", + "gitea.branch.push", "gitea.issue.comment", + ], + "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "execution_profile": "author-profile", + }, + "reviewer-profile": { + "enabled": True, + "context": "ctx", + "role": "reviewer", + "username": "reviewer-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"}, + "allowed_operations": [ + "gitea.read", "gitea.pr.review", "gitea.pr.approve", + "gitea.pr.merge", + ], + "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"], + "execution_profile": "reviewer-profile", + }, + }, + "rules": {"allow_runtime_switching": False}, +} + + +class TestTaskRoleGuards(unittest.TestCase): + """Review/merge tasks under an author profile must return stop guidance.""" + + def setUp(self): + self._remotes_patch = patch.dict(mcp_server.REMOTES, { + "prgs": {"host": "gitea.example.com", "org": "Example-Org", + "repo": "Example-Repo"}, + }) + self._remotes_patch.start() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + self._dir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self._dir.name, "profiles.json") + with open(self.config_path, "w", encoding="utf-8") as fh: + fh.write(json.dumps(CONFIG_GUARD)) + + def tearDown(self): + self._remotes_patch.stop() + mcp_server._IDENTITY_CACHE.clear() + gitea_config._active_profile_override = None + self._dir.cleanup() + + def _env(self, profile): + return { + "GITEA_MCP_CONFIG": self.config_path, + "GITEA_MCP_PROFILE": profile, + "GITEA_TOKEN_AUTHOR": "author-pass", + "GITEA_TOKEN_REVIEWER": "reviewer-pass", + } + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_review_task_under_author_profile_returns_stop_guidance(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + res = gitea_resolve_task_capability(task="review_pr", remote="prgs") + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + guidance = " ".join(res["task_role_guidance"]) + self.assertIn("STOP", guidance) + self.assertIn("author", guidance.lower()) + # must forbid author-side mutations while the task is review/merge + self.assertIn("commit", guidance.lower()) + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_merge_task_under_author_profile_returns_stop_guidance(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + res = gitea_resolve_task_capability(task="merge_pr", remote="prgs") + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_author_fix_task_is_distinct_and_allowed_for_author(self, _auth, _api): + with patch.dict(os.environ, self._env("author-profile")): + res = gitea_resolve_task_capability( + task="address_pr_change_requests", remote="prgs") + self.assertEqual(res["required_role_kind"], "author") + self.assertTrue(res["allowed_in_current_session"]) + self.assertFalse(res["stop_required"]) + guidance = " ".join(res["task_role_guidance"]) + self.assertIn("must not submit review verdicts", guidance.lower()) + + @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_review_task_under_reviewer_profile_allowed(self, _auth, _api): + with patch.dict(os.environ, self._env("reviewer-profile")): + res = gitea_resolve_task_capability(task="review_pr", remote="prgs") + self.assertTrue(res["allowed_in_current_session"]) + self.assertFalse(res["stop_required"]) + + @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_author_fix_task_under_reviewer_profile_blocked(self, _auth, _api): + with patch.dict(os.environ, self._env("reviewer-profile")): + res = gitea_resolve_task_capability( + task="address_pr_change_requests", remote="prgs") + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + + +class TestValidationReport(unittest.TestCase): + """Validation reporting must not hide failed or skipped checks.""" + + def test_distinguishes_passed_failed_skipped_not_run(self): + report = build_validation_report([ + {"command": "python3 -m py_compile mcp_server.py", "status": "passed"}, + {"command": "python3 -m unittest discover -s tests", "status": "failed", + "output": "FAILED (failures=1)"}, + {"command": "git diff --check", "status": "skipped", + "reason": "replaced by targeted check"}, + {"command": "secret sweep", "status": "not-run"}, + ]) + self.assertFalse(report["all_passed"]) + statuses = {e["command"]: e["status"] for e in report["entries"]} + self.assertEqual(statuses["python3 -m py_compile mcp_server.py"], "passed") + self.assertEqual(statuses["python3 -m unittest discover -s tests"], "failed") + self.assertEqual(statuses["git diff --check"], "skipped") + self.assertEqual(statuses["secret sweep"], "not-run") + summary = report["summary"] + self.assertIn("FAILED", summary) + self.assertIn("FAILED (failures=1)", summary) + self.assertIn("SKIPPED", summary) + self.assertIn("NOT-RUN", summary) + + def test_failed_entry_requires_exact_output(self): + with self.assertRaises(ValueError): + build_validation_report([ + {"command": "unittest", "status": "failed"}, + ]) + + def test_unknown_status_fails_closed(self): + with self.assertRaises(ValueError): + build_validation_report([ + {"command": "unittest", "status": "shell-invocation quirks"}, + ]) + + def test_missing_command_fails_closed(self): + with self.assertRaises(ValueError): + build_validation_report([{"status": "passed"}]) + + def test_full_suite_success_not_implied(self): + report = build_validation_report([ + {"command": "python3 -m unittest discover -s tests", + "status": "not-run", "full_suite": True}, + {"command": "targeted tests", "status": "passed"}, + ]) + self.assertFalse(report["all_passed"]) + self.assertIs(report["full_suite_passed"], False) + + def test_full_suite_passed_when_run_and_green(self): + report = build_validation_report([ + {"command": "python3 -m unittest discover -s tests", + "status": "passed", "full_suite": True}, + ]) + self.assertTrue(report["all_passed"]) + self.assertIs(report["full_suite_passed"], True) + + def test_full_suite_none_when_no_full_suite_entry(self): + report = build_validation_report([ + {"command": "targeted tests", "status": "passed"}, + ]) + self.assertIsNone(report["full_suite_passed"]) + + +if __name__ == "__main__": + unittest.main()