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

Merge remote-tracking branch 'prgs/master' into feat/issue-299-already-landed-handoff

Browse Source
This commit is contained in:
sysadmin
2026-07-07 09:57:54 -04:00
parent b53e3de7cc 0bb6cae95a
commit fbfcc08640
7 changed files with 952 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+73
View File
@@ -478,6 +478,7 @@ import agent_temp_artifacts
import issue_lock_worktree # noqa: E402
import issue_claim_heartbeat # noqa: E402
import merged_cleanup_reconcile # noqa: E402
import native_mcp_preference # noqa: E402
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -4083,6 +4084,14 @@ _GUIDE_RULES = {
"inventory, summarize) needs no explicit authorization. Enforced "
"fail closed via subagent_gate.assess_subagent_delegation and "
"validate_subagent_report (#266)."),
"native_mcp_preference": (
"Gitea mutations must use native MCP tools first. When MCP is "
"available, block shell scripts, direct API calls, WebFetch, "
"Playwright, helper encoders, and profile-secret reads unless "
"explicit recovery-mode proof is complete. If MCP transport is "
"broken or the shell circuit breaker trips, emit a terminal recovery "
"report instead of improvising unsafe fallback. Reviewers must not "
"touch or restart MCP server files/processes (#270)."),
}
_COMMON_WORKFLOWS = [
@@ -4787,6 +4796,69 @@ def _build_runtime_task_capabilities(
}
@mcp.tool()
@mcp.tool()
def gitea_assess_gitea_operation_path(
task: str,
path_kind: str | None = None,
command_or_detail: str | None = None,
mcp_available: bool = True,
mcp_tool_visible: bool = True,
recovery_mode: bool = False,
recovery_proof_complete: bool = False,
remote: str = "dadeschools",
host: str | None = None,
) -> dict:
"""Read-only: assess whether a proposed Gitea mutation path is allowed (#270).
Native MCP must be preferred when available. Shell scripts, direct API
calls, WebFetch, Playwright, and helper encoders are blocked unless
explicit recovery-mode proof is supplied.
"""
profile = get_profile()
role = _role_kind(
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
return native_mcp_preference.assess_gitea_operation_path(
task=task,
path_kind=path_kind,
command_or_detail=command_or_detail,
mcp_available=mcp_available,
mcp_tool_visible=mcp_tool_visible,
recovery_mode=recovery_mode,
recovery_proof_complete=recovery_proof_complete,
role=role,
active_profile=profile.get("profile_name"),
)
@mcp.tool()
def gitea_record_shell_spawn_outcome(
exit_code: int | None = None,
stdout: str = "",
stderr: str = "",
spawn_failure: bool | None = None,
probe_attempted: bool = False,
probe_succeeded: bool | None = None,
) -> dict:
"""Record shell spawn outcome and update the session circuit breaker (#270)."""
return native_mcp_preference.record_shell_spawn_outcome(
exit_code=exit_code,
stdout=stdout,
stderr=stderr,
spawn_failure=spawn_failure,
probe_attempted=probe_attempted,
probe_succeeded=probe_succeeded,
)
@mcp.tool()
def gitea_get_shell_health() -> dict:
"""Read-only: return shell spawn circuit-breaker state for this MCP session (#270)."""
return native_mcp_preference.shell_health_status()
@mcp.tool()
def gitea_get_runtime_context(
remote: str = "dadeschools",
@@ -4905,6 +4977,7 @@ def gitea_get_runtime_context(
"preflight_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"),
"session_capabilities": session_capabilities,
"shell_health": native_mcp_preference.shell_health_status(),
}
if reveal and h:
native_mcp_preference.py
+369
View File
@@ -0,0 +1,369 @@
"""Native MCP preference gate and shell health circuit breaker (#270).
Gitea mutations must use native MCP tools first. Shell scripts, direct API
calls, browser helpers, and improvised encoders are blocked when MCP is
available unless explicit recovery-mode proof is supplied.
"""
from __future__ import annotations
import re
from typing import Any
from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES
# Tasks that must prefer native MCP over shell/API/helper fallbacks.
GITEA_MUTATION_TASKS = frozenset({
"comment_issue",
"mark_issue",
"lock_issue",
"set_issue_labels",
"create_issue",
"close_issue",
"create_branch",
"push_branch",
"create_pr",
"close_pr",
"comment_pr",
"review_pr",
"merge_pr",
"approve_pr",
"request_changes_pr",
"commit_files",
"gitea_commit_files",
"delete_branch",
"address_pr_change_requests",
})
ALLOWED_PATH_KINDS = frozenset({
"mcp_native",
"recovery_fallback",
})
BLOCKED_PATH_KINDS = frozenset({
"shell_script",
"direct_api",
"webfetch",
"playwright",
"helper_script",
"unsafe_helper",
"mcp_server_touch",
})
SHELL_SPAWN_FAILURE_THRESHOLD = 2
TERMINAL_REPORT_HEADING = (
"MCP transport unavailable or shell circuit breaker tripped. "
"No unsafe Gitea fallback performed."
)
_RECOVERY_MODE_RE = re.compile(
r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|"
r"mcp tools unavailable|no mcp path)\b",
re.IGNORECASE,
)
_LOCAL_SCRIPT_RE = re.compile(
r"(?:^|[\s\"'`/])(?:python3?|bash|sh)?\s*(?:"
+ "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES)
+ r")",
re.IGNORECASE,
)
_WEBFETCH_RE = re.compile(r"\b(?:webfetch|mcp_web_fetch|fetch\s+url)\b", re.IGNORECASE)
_PLAYWRIGHT_RE = re.compile(r"\b(?:playwright|browser_navigate|browser_click)\b", re.IGNORECASE)
_HELPER_SCRIPT_RE = re.compile(
r"(?:^|[\s\"'`/])(?:_encode_|_emit_|_inline_)[\w-]+\.py",
re.IGNORECASE,
)
_MANUAL_BASE64_RE = re.compile(
r"\b(?:manual\s+base64|llm-generated\s+base64|base64-encode\s+in\s+chat)\b",
re.IGNORECASE,
)
_DIRECT_API_RE = re.compile(
r"\b(?:api_request|urllib\.request|requests\.(?:post|patch|put)|curl\s+-X\s+POST)\b",
re.IGNORECASE,
)
_MCP_SERVER_TOUCH_RE = re.compile(
r"\b(?:kill|pkill|restart|reload|touch|edit|modify|write)\b.{0,40}\b(?:mcp_server|mcp-server|gitea_mcp_server)\b",
re.IGNORECASE,
)
_CLI_AUTH_DIVERGENCE_RE = re.compile(
r"GITEA_MCP_PROFILE\s*=\s*['\"]?([^\s'\";]+)",
re.IGNORECASE,
)
_session_shell: dict[str, Any] = {
"consecutive_spawn_failures": 0,
"shell_unavailable": False,
"hard_stopped": False,
"last_exit_code": None,
}
def _clean(value: str | None) -> str:
return (value or "").strip()
def is_spawn_failure(
*,
exit_code: int | None = None,
stdout: str | None = None,
stderr: str | None = None,
spawn_failure: bool | None = None,
) -> bool:
"""True for executor spawn failures (exit_code -1, empty output)."""
if spawn_failure is True:
return True
if spawn_failure is False:
return False
if exit_code != -1:
return False
return not (_clean(stdout) or _clean(stderr))
def record_shell_spawn_outcome(
*,
exit_code: int | None = None,
stdout: str | None = None,
stderr: str | None = None,
spawn_failure: bool | None = None,
probe_attempted: bool = False,
probe_succeeded: bool | None = None,
) -> dict[str, Any]:
"""Track shell spawn outcomes and trip the session circuit breaker (#270 AC4)."""
global _session_shell
failed = is_spawn_failure(
exit_code=exit_code,
stdout=stdout,
stderr=stderr,
spawn_failure=spawn_failure,
)
if failed:
_session_shell["consecutive_spawn_failures"] = (
int(_session_shell.get("consecutive_spawn_failures") or 0) + 1
)
_session_shell["last_exit_code"] = exit_code
if probe_attempted and probe_succeeded is False:
_session_shell["shell_unavailable"] = True
else:
_session_shell["consecutive_spawn_failures"] = 0
_session_shell["shell_unavailable"] = False
_session_shell["hard_stopped"] = False
_session_shell["last_exit_code"] = exit_code
failures = int(_session_shell["consecutive_spawn_failures"])
if failures >= SHELL_SPAWN_FAILURE_THRESHOLD:
_session_shell["shell_unavailable"] = True
_session_shell["hard_stopped"] = True
return shell_health_status()
def shell_health_status() -> dict[str, Any]:
"""Return the current shell health / circuit-breaker state."""
failures = int(_session_shell.get("consecutive_spawn_failures") or 0)
hard_stopped = bool(_session_shell.get("hard_stopped"))
shell_unavailable = bool(_session_shell.get("shell_unavailable"))
return {
"consecutive_spawn_failures": failures,
"shell_unavailable": shell_unavailable,
"hard_stopped": hard_stopped,
"threshold": SHELL_SPAWN_FAILURE_THRESHOLD,
"shell_use_allowed": not hard_stopped,
"last_exit_code": _session_shell.get("last_exit_code"),
"safe_next_action": (
"emit terminal recovery report; prefer native MCP for remaining Gitea mutations"
if hard_stopped
else (
"probe shell once with echo/pwd; if probe fails mark shell unavailable"
if failures == 1
else "shell healthy"
)
),
}
def clear_shell_health_for_tests() -> None:
"""Reset session shell state (tests only)."""
global _session_shell
_session_shell = {
"consecutive_spawn_failures": 0,
"shell_unavailable": False,
"hard_stopped": False,
"last_exit_code": None,
}
def classify_command_path(command_or_detail: str | None) -> str:
"""Classify a proposed command/detail into a path kind."""
text = command_or_detail or ""
if _MCP_SERVER_TOUCH_RE.search(text):
return "mcp_server_touch"
if _WEBFETCH_RE.search(text):
return "webfetch"
if _PLAYWRIGHT_RE.search(text):
return "playwright"
if _HELPER_SCRIPT_RE.search(text) or _MANUAL_BASE64_RE.search(text):
return "unsafe_helper"
if _LOCAL_SCRIPT_RE.search(text):
return "shell_script"
if _DIRECT_API_RE.search(text):
return "direct_api"
return "mcp_native"
def detect_cli_auth_divergence(
command_or_detail: str | None,
*,
active_profile: str | None,
) -> list[str]:
"""Flag shell commands that override GITEA_MCP_PROFILE away from the session."""
reasons: list[str] = []
text = command_or_detail or ""
active = _clean(active_profile)
match = _CLI_AUTH_DIVERGENCE_RE.search(text)
if not match or not active:
return reasons
requested = _clean(match.group(1))
if requested and requested != active:
reasons.append(
f"CLI auth divergence: command sets GITEA_MCP_PROFILE='{requested}' "
f"but active session profile is '{active}' (fail closed)"
)
return reasons
def format_mcp_unavailable_terminal_report(
*,
task: str | None = None,
reasons: list[str] | None = None,
) -> str:
"""Terminal report when MCP is broken and unsafe recovery is forbidden (#270 AC3)."""
lines = [TERMINAL_REPORT_HEADING, ""]
if task:
lines.append(f"Blocked task: {task}")
if reasons:
lines.extend(["", "Reasons:"])
lines.extend(f"- {reason}" for reason in reasons)
lines.extend([
"",
"Required recovery (no improvised fallback):",
"- Restart the MCP session",
"- Kill hung background terminals holding the shell executor",
"- Retry the native MCP tool once after reconnect",
"- If MCP remains unavailable, stop and hand off to the operator",
"- Do not run local Gitea scripts, WebFetch, Playwright, or manual base64 encoders",
])
return "\n".join(lines)
def assess_gitea_operation_path(
*,
task: str,
path_kind: str | None = None,
command_or_detail: str | None = None,
mcp_available: bool = True,
mcp_tool_visible: bool = True,
recovery_mode: bool = False,
recovery_proof_complete: bool = False,
role: str | None = None,
active_profile: str | None = None,
) -> dict[str, Any]:
"""Fail closed when a Gitea mutation would bypass native MCP (#270)."""
task_name = _clean(task)
resolved_kind = _clean(path_kind) or classify_command_path(command_or_detail)
reasons: list[str] = []
if task_name and task_name not in GITEA_MUTATION_TASKS:
reasons.append(
f"unknown or non-mutation Gitea task '{task_name}'; "
"native MCP preference gate applies only to Gitea mutations"
)
shell = shell_health_status()
if shell["hard_stopped"] and resolved_kind != "mcp_native":
reasons.append(
"shell circuit breaker is hard-stopped after consecutive spawn failures; "
"use native MCP or emit terminal recovery report"
)
recovery_declared = recovery_mode or bool(
_RECOVERY_MODE_RE.search(command_or_detail or "")
)
recovery_allowed = (
recovery_declared
and recovery_proof_complete
and not mcp_available
and resolved_kind in BLOCKED_PATH_KINDS - {"mcp_server_touch"}
)
if resolved_kind in BLOCKED_PATH_KINDS and not recovery_allowed:
reasons.append(f"path kind '{resolved_kind}' is not an approved Gitea mutation path")
if resolved_kind == "mcp_server_touch":
if _clean(role) == "reviewer" or (
active_profile and "reviewer" in active_profile.lower()
):
reasons.append(
"reviewer workflows must not touch, restart, or kill MCP server files/processes"
)
else:
reasons.append(
"MCP server files/processes must not be modified during normal Gitea workflows"
)
reasons.extend(
detect_cli_auth_divergence(command_or_detail, active_profile=active_profile)
)
if (
mcp_available
and mcp_tool_visible
and resolved_kind != "mcp_native"
and not recovery_allowed
):
if not recovery_declared:
reasons.append(
"native MCP tools are available; shell/API/helper fallback is forbidden"
)
elif not recovery_proof_complete:
reasons.append(
"recovery-mode fallback requires complete recovery proof before proceeding"
)
if not mcp_available and resolved_kind != "mcp_native" and not recovery_allowed:
if not recovery_declared:
reasons.append(
"MCP transport unavailable; produce terminal recovery report instead of "
"improvising unsafe fallback"
)
block = bool(reasons)
terminal_report = None
if block and (not mcp_available or shell["hard_stopped"]):
terminal_report = format_mcp_unavailable_terminal_report(
task=task_name or None,
reasons=reasons,
)
return {
"task": task_name or None,
"path_kind": resolved_kind,
"mcp_available": mcp_available,
"mcp_tool_visible": mcp_tool_visible,
"recovery_mode": recovery_declared,
"shell_health": shell,
"block": block,
"allowed": not block,
"reasons": reasons,
"terminal_report": terminal_report,
"safe_next_action": (
"use the native MCP tool for this task"
if mcp_available and mcp_tool_visible and block
else (
terminal_report or "proceed with native MCP"
if block
else "proceed with native MCP"
)
),
}
review_proofs.py
+163
View File
@@ -4545,6 +4545,169 @@ def assess_full_suite_failure_approval_gate(
}
# ---------------------------------------------------------------------------
# Reconciler close gate (#309)
# ---------------------------------------------------------------------------
RECONCILER_CLOSE_REPORT_FIELDS = (
"identity/profile",
"close capability proof",
"PR live state",
"candidate head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"PR close result",
"issue close result",
"no review/merge confirmation",
)
def assess_reconciler_close_gate(
*,
pr_number: int | None,
pr_state: str | None,
candidate_head_sha: str | None,
live_head_sha: str | None,
target_branch: str | None,
target_branch_sha: str | None,
head_is_ancestor_of_target: bool | None,
close_pr_capability: bool,
close_issue_capability: bool = False,
linked_issue_state: str | None = None,
issue_resolved_by_landing: bool = False,
) -> dict:
"""#309: gate the dedicated reconciler close path for landed PRs.
The reconciler path may close a PR only with exact ``gitea.pr.close``
capability and full already-landed proof: live open PR, pinned head,
freshly fetched target branch SHA, and ancestry of the head in the
target. Non-landed PRs are never closable here, and the gate never
grants review, approval, request-changes, or merge.
Linked-issue closure: skipped when the issue is already closed;
allowed only when the issue is open, resolved by the landed commits,
and exact ``gitea.issue.close`` capability is proven.
"""
reasons: list[str] = []
if not isinstance(pr_number, int) or pr_number <= 0:
reasons.append("PR number missing or invalid (#309)")
if (pr_state or "").strip().lower() != "open":
reasons.append(
"PR is not live-verified open at mutation time; re-fetch the "
"PR before any reconciler close (#309)"
)
if not _FULL_SHA.match((candidate_head_sha or "").strip()):
reasons.append(
"candidate head SHA is not a full 40-hex commit SHA (#309)"
)
if live_head_sha is not None and candidate_head_sha and (
live_head_sha.strip() != candidate_head_sha.strip()
):
reasons.append(
"PR head changed since the ancestry proof was pinned; re-run "
"the already-landed check (#309)"
)
if not (target_branch or "").strip():
reasons.append("target branch missing (#309)")
if not _FULL_SHA.match((target_branch_sha or "").strip()):
reasons.append(
"target branch SHA missing or not a full 40-hex SHA; fetch the "
"target branch freshly before the ancestry check (#309)"
)
if head_is_ancestor_of_target is None:
reasons.append(
"ancestry of the PR head against the target branch was not "
"checked (#309)"
)
never_allowed = {
"review_allowed": False,
"approve_allowed": False,
"request_changes_allowed": False,
"merge_allowed": False,
}
if reasons:
return {
"outcome": "GATE_NOT_PROVEN",
"pr_close_allowed": False,
"issue_close_allowed": False,
"reasons": reasons,
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"re-fetch the PR and target branch, pin SHAs, run the "
"ancestry check, then re-run this gate"
),
**never_allowed,
}
if head_is_ancestor_of_target is False:
return {
"outcome": "NOT_LANDED_CLOSE_BLOCKED",
"pr_close_allowed": False,
"issue_close_allowed": False,
"reasons": [
f"PR #{pr_number} head is not an ancestor of "
f"{target_branch}; non-landed PRs cannot be closed through "
"the reconciler path (#309)"
],
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"route the PR through the normal review workflow; the "
"reconciler path only closes already-landed PRs"
),
**never_allowed,
}
if close_pr_capability is not True:
return {
"outcome": "RECOVERY_HANDOFF_REQUIRED",
"pr_close_allowed": False,
"issue_close_allowed": False,
"reasons": [
"exact gitea.pr.close capability not proven; produce a "
"recovery handoff instead of closing (#309)"
],
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"produce a recovery handoff naming the missing "
"gitea.pr.close capability and the completed ancestor proof"
),
**never_allowed,
}
issue_close_allowed = (
(linked_issue_state or "").strip().lower() == "open"
and issue_resolved_by_landing is True
and close_issue_capability is True
)
issue_close_reasons = []
if (linked_issue_state or "").strip().lower() == "closed":
issue_close_reasons.append(
"linked issue already closed; no issue close attempted (#309)"
)
elif not issue_close_allowed:
issue_close_reasons.append(
"issue close requires an open linked issue resolved by the "
"landed commits and exact gitea.issue.close capability (#309)"
)
return {
"outcome": "CLOSE_ALLOWED",
"pr_close_allowed": True,
"issue_close_allowed": issue_close_allowed,
"reasons": issue_close_reasons,
"required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
"safe_next_action": (
"close the already-landed PR, report the close result, and "
"handle the linked issue per the proven capability"
),
**never_allowed,
}
# ---------------------------------------------------------------------------
# Identity disclosure (#305)
# ---------------------------------------------------------------------------
role_session_router.py
+28
View File
@@ -90,12 +90,20 @@ TASK_REQUIRED_ROLE = {
"blind_pr_queue_review": "reviewer",
"request_changes_pr": "reviewer",
"approve_pr": "reviewer",
# #309: reconciler tasks close already-landed PRs/issues only.
"reconcile_close_landed_pr": "reconciler",
"reconcile_close_landed_issue": "reconciler",
}
WRONG_ROLE_REVIEWER_MSG = (
"Wrong role/session for reviewer task. Launch reviewer MCP namespace."
)
WRONG_ROLE_RECONCILER_MSG = (
"Wrong role/session for reconciler task. Launch a reconciler-capable "
"MCP namespace/profile with exact close capability."
)
_session_last_route: dict | None = None
@@ -191,6 +199,26 @@ def route_task_session(
_record_route(result)
return result
if required_role == "reconciler":
result = {
"task_type": task_type,
"required_role": required_role,
"active_role": active_role_kind,
"active_profile": active_profile,
"route_result": ROUTE_WRONG_ROLE,
"downstream_allowed": False,
"reasons": [
WRONG_ROLE_RECONCILER_MSG,
"Reconciler tasks cannot run in author or reviewer "
"sessions without exact close capability.",
],
"message": WRONG_ROLE_RECONCILER_MSG,
"runtime_switching_supported": runtime_switching_supported,
"profile_switch_blocked": not runtime_switching_supported,
}
_record_route(result)
return result
if required_role == "author":
route = ROUTE_TO_AUTHOR
message = (
task_capability_map.py
+10
View File
@@ -96,6 +96,16 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
"permission": "gitea.read",
"role": "author",
},
# #309: dedicated reconciler path for already-landed open PRs. Exact
# close capabilities only — never review/approve/request_changes/merge.
"reconcile_close_landed_pr": {
"permission": "gitea.pr.close",
"role": "reconciler",
},
"reconcile_close_landed_issue": {
"permission": "gitea.issue.close",
"role": "reconciler",
},
"post_heartbeat": {
"permission": "gitea.issue.comment",
"role": "author",
tests/test_native_mcp_preference.py
+121
View File
@@ -0,0 +1,121 @@
"""Tests for native MCP preference gate and shell circuit breaker (#270)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import native_mcp_preference as nmp # noqa: E402
class TestShellHealthCircuitBreaker(unittest.TestCase):
def setUp(self):
nmp.clear_shell_health_for_tests()
def test_two_spawn_failures_hard_stop(self):
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
self.assertTrue(status["hard_stopped"])
self.assertFalse(status["shell_use_allowed"])
self.assertEqual(status["consecutive_spawn_failures"], 2)
def test_success_resets_breaker(self):
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="")
self.assertFalse(status["hard_stopped"])
self.assertEqual(status["consecutive_spawn_failures"], 0)
class TestNativeMcpPreferenceGate(unittest.TestCase):
def setUp(self):
nmp.clear_shell_health_for_tests()
def test_mcp_available_blocks_local_script(self):
result = nmp.assess_gitea_operation_path(
task="create_pr",
command_or_detail="python create_pr.py --remote prgs --title T --head h",
mcp_available=True,
mcp_tool_visible=True,
)
self.assertTrue(result["block"])
self.assertEqual(result["path_kind"], "shell_script")
def test_mcp_native_path_allowed(self):
result = nmp.assess_gitea_operation_path(
task="comment_issue",
path_kind="mcp_native",
mcp_available=True,
mcp_tool_visible=True,
)
self.assertFalse(result["block"])
self.assertTrue(result["allowed"])
def test_mcp_unavailable_requires_terminal_report(self):
result = nmp.assess_gitea_operation_path(
task="merge_pr",
path_kind="shell_script",
mcp_available=False,
mcp_tool_visible=False,
)
self.assertTrue(result["block"])
self.assertTrue(
any("terminal recovery report" in r for r in result["reasons"])
)
self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"])
def test_recovery_fallback_allowed_with_proof(self):
result = nmp.assess_gitea_operation_path(
task="merge_pr",
path_kind="shell_script",
command_or_detail="Recovery mode: MCP tools unavailable in this session.",
mcp_available=False,
recovery_mode=True,
recovery_proof_complete=True,
)
self.assertFalse(result["block"])
def test_cli_auth_divergence_blocked(self):
result = nmp.assess_gitea_operation_path(
task="create_pr",
command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py",
mcp_available=True,
active_profile="prgs-author",
)
self.assertTrue(result["block"])
self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"]))
def test_unsafe_helper_fallback_denied(self):
result = nmp.assess_gitea_operation_path(
task="commit_files",
command_or_detail="python _encode_payload.py",
mcp_available=True,
mcp_tool_visible=True,
)
self.assertTrue(result["block"])
self.assertEqual(result["path_kind"], "unsafe_helper")
def test_reviewer_mcp_server_touch_blocked(self):
result = nmp.assess_gitea_operation_path(
task="review_pr",
command_or_detail="pkill -f gitea_mcp_server.py",
mcp_available=True,
role="reviewer",
active_profile="prgs-reviewer",
)
self.assertTrue(result["block"])
self.assertEqual(result["path_kind"], "mcp_server_touch")
def test_hard_stopped_shell_blocks_non_mcp_path(self):
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="")
result = nmp.assess_gitea_operation_path(
task="lock_issue",
path_kind="shell_script",
mcp_available=True,
)
self.assertTrue(result["block"])
self.assertTrue(any("circuit breaker" in r for r in result["reasons"]))
if __name__ == "__main__":
unittest.main()
tests/test_reconciler_close_gate.py
+188
View File
@@ -0,0 +1,188 @@
"""Issue #309: dedicated reconciler capability to close already-landed PRs.
The reconciler path must prove exact ``gitea.pr.close`` capability, must
never imply review/approve/request-changes/merge capability, and may close
a PR only after full already-landed proof.
"""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import role_session_router # noqa: E402
import task_capability_map # noqa: E402
from review_proofs import assess_reconciler_close_gate # noqa: E402
PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
class TestReconcilerCapabilityMap(unittest.TestCase):
def test_reconcile_close_pr_task_maps_to_pr_close(self):
self.assertEqual(
task_capability_map.required_permission(
"reconcile_close_landed_pr"),
"gitea.pr.close",
)
self.assertEqual(
task_capability_map.required_role("reconcile_close_landed_pr"),
"reconciler",
)
def test_reconcile_close_issue_task_maps_to_issue_close(self):
self.assertEqual(
task_capability_map.required_permission(
"reconcile_close_landed_issue"),
"gitea.issue.close",
)
self.assertEqual(
task_capability_map.required_role(
"reconcile_close_landed_issue"),
"reconciler",
)
def test_reconciler_tasks_do_not_grant_review_permissions(self):
for task in ("reconcile_close_landed_pr",
"reconcile_close_landed_issue"):
permission = task_capability_map.required_permission(task)
self.assertNotIn("review", permission)
self.assertNotIn("approve", permission)
self.assertNotIn("merge", permission)
self.assertNotIn("request_changes", permission)
class TestReconcilerRouting(unittest.TestCase):
def test_reconciler_task_requires_reconciler_role(self):
self.assertEqual(
role_session_router.required_role_for_task(
"reconcile_close_landed_pr"),
"reconciler",
)
def test_reconciler_task_blocked_in_author_session(self):
result = role_session_router.route_task_session(
"reconcile_close_landed_pr",
active_profile="prgs-author",
active_role_kind="author",
allowed_in_current_session=False,
)
self.assertEqual(result["route_result"], "wrong_role_stop")
self.assertFalse(result["downstream_allowed"])
def test_reconciler_task_allowed_in_matching_session(self):
result = role_session_router.route_task_session(
"reconcile_close_landed_pr",
active_profile="prgs-reconciler",
active_role_kind="reconciler",
allowed_in_current_session=True,
)
self.assertEqual(result["route_result"], "allowed_current_session")
self.assertTrue(result["downstream_allowed"])
class TestReconcilerCloseGate(unittest.TestCase):
def _gate(self, **overrides):
kwargs = {
"pr_number": 278,
"pr_state": "open",
"candidate_head_sha": PINNED,
"live_head_sha": PINNED,
"target_branch": "master",
"target_branch_sha": OTHER,
"head_is_ancestor_of_target": True,
"close_pr_capability": True,
"close_issue_capability": False,
"linked_issue_state": "closed",
}
kwargs.update(overrides)
return assess_reconciler_close_gate(**kwargs)
def test_already_landed_open_pr_close_allowed(self):
result = self._gate()
self.assertEqual(result["outcome"], "CLOSE_ALLOWED")
self.assertTrue(result["pr_close_allowed"])
def test_not_landed_pr_cannot_be_closed(self):
result = self._gate(head_is_ancestor_of_target=False)
self.assertEqual(result["outcome"], "NOT_LANDED_CLOSE_BLOCKED")
self.assertFalse(result["pr_close_allowed"])
def test_unchecked_ancestry_blocks_close(self):
result = self._gate(head_is_ancestor_of_target=None)
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
self.assertFalse(result["pr_close_allowed"])
def test_stale_target_branch_sha_blocks_close(self):
result = self._gate(target_branch_sha=None)
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
self.assertFalse(result["pr_close_allowed"])
def test_closed_pr_state_blocks_close(self):
result = self._gate(pr_state="closed")
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
self.assertFalse(result["pr_close_allowed"])
def test_changed_head_blocks_close(self):
result = self._gate(live_head_sha=OTHER)
self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
self.assertFalse(result["pr_close_allowed"])
def test_missing_close_capability_produces_recovery_handoff(self):
result = self._gate(close_pr_capability=False)
self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED")
self.assertFalse(result["pr_close_allowed"])
self.assertTrue(any(
"gitea.pr.close" in reason for reason in result["reasons"]
))
def test_linked_issue_already_closed_skips_issue_close(self):
result = self._gate(linked_issue_state="closed",
close_issue_capability=True)
self.assertFalse(result["issue_close_allowed"])
def test_linked_open_issue_close_requires_capability(self):
allowed = self._gate(
linked_issue_state="open",
issue_resolved_by_landing=True,
close_issue_capability=True,
)
self.assertTrue(allowed["issue_close_allowed"])
blocked = self._gate(
linked_issue_state="open",
issue_resolved_by_landing=True,
close_issue_capability=False,
)
self.assertFalse(blocked["issue_close_allowed"])
def test_gate_never_allows_review_mutations(self):
for result in (
self._gate(),
self._gate(head_is_ancestor_of_target=False),
self._gate(close_pr_capability=False),
):
self.assertFalse(result["review_allowed"])
self.assertFalse(result["approve_allowed"])
self.assertFalse(result["request_changes_allowed"])
self.assertFalse(result["merge_allowed"])
def test_report_fields_listed_for_close(self):
result = self._gate()
for field in (
"identity/profile",
"close capability proof",
"PR live state",
"candidate head SHA",
"target branch SHA",
"ancestor proof",
"linked issue status",
"PR close result",
"issue close result",
"no review/merge confirmation",
):
self.assertIn(field, result["required_report_fields"])
if __name__ == "__main__":
unittest.main()