Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

Merge pull request 'Harden LLM role/task alignment and PR review feedback discovery' (#169) from feat/issue-167-review-feedback-discovery into master

Browse Source
This commit was merged in pull request #169.
This commit is contained in:
sysadmin
2026-07-05 12:17:43 -05:00
parent 0b6733a7e8 a4060c59cc
commit 29f04d003a
3 changed files with 757 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp_server.py
+236
View File
@@ -785,6 +785,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(
@@ -3108,6 +3241,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,
@@ -3157,6 +3351,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",
@@ -3240,6 +3438,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,
@@ -3248,6 +3482,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,