fix(gate): stop classifying stale-runtime blocks as permission denials (Closes #897)

Stale-runtime and runtime-mode mutation refusals previously shared the
permission-denial channel, so permission_report claimed a missing op the
active profile already held and recommended gitea_activate_profile.
Typed blocker_kind payloads report reconnect-only recovery for staleness,
omit permission_report for non-permission gates, and fail closed when a
permission_report would invent a missing permission the profile holds.
This commit is contained in:
2026-07-25 01:47:35 -04:00
parent 2f4dec8323
commit 6e6ca94338
2 changed files with 749 additions and 70 deletions
+296 -70
View File
@@ -9552,15 +9552,13 @@ def gitea_edit_pr(
if closing: if closing:
gate_reasons = _profile_operation_gate("gitea.pr.close") gate_reasons = _profile_operation_gate("gitea.pr.close")
if gate_reasons: if gate_reasons:
return { return _build_operation_gate_refusal(
"success": False, "gitea.pr.close",
"performed": False, gate_reasons,
"pr_number": pr_number, pr_number=pr_number,
"requested_state": "closed", requested_state="closed",
"required_permission": "gitea.pr.close", required_permission="gitea.pr.close",
"reasons": gate_reasons, )
"permission_report": _permission_block_report("gitea.pr.close"),
}
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h) auth = _auth(h)
@@ -13823,13 +13821,17 @@ def gitea_view_issue(
def _permission_block_report(required_operation: str, def _permission_block_report(required_operation: str,
identity: str | None = None) -> dict: identity: str | None = None) -> dict:
"""Structured, LLM-safe explanation of a permission denial (#142). """Structured, LLM-safe explanation of a permission denial (#142, #897).
Built only after a gate has already refused; it adds guidance to the Built only after a gate has already refused; it adds guidance to the
refusal and never widens any permission, performs network I/O, or refusal and never widens any permission, performs network I/O, or
raises (fail-soft: degrades to a minimal fail-closed report). Names raises (fail-soft: degrades to a minimal fail-closed report). Names
configured profiles only never auth references, tokens, endpoint configured profiles only never auth references, tokens, endpoint
URLs, or keychain IDs. URLs, or keychain IDs.
#897: never fabricate a missing permission when the active profile
already allows the operation. That path is a diagnostic defect (the
refusal was not a permission denial), not a cue to switch profiles.
""" """
report = { report = {
"requested_operation": required_operation, "requested_operation": required_operation,
@@ -13841,6 +13843,7 @@ def _permission_block_report(required_operation: str,
"matching_configured_profiles": [], "matching_configured_profiles": [],
"runtime_switching_supported": False, "runtime_switching_supported": False,
"different_mcp_namespace_required": True, "different_mcp_namespace_required": True,
"diagnostic_defect": False,
"exact_safe_next_action": ( "exact_safe_next_action": (
"Ask the operator to fix GITEA_MCP_CONFIG/GITEA_MCP_PROFILE; " "Ask the operator to fix GITEA_MCP_CONFIG/GITEA_MCP_PROFILE; "
"the active profile could not be resolved (fail closed)."), "the active profile could not be resolved (fail closed)."),
@@ -13855,6 +13858,32 @@ def _permission_block_report(required_operation: str,
report["active_allowed_operations"] = ( report["active_allowed_operations"] = (
profile.get("allowed_operations") or []) profile.get("allowed_operations") or [])
# #897: fail closed as a diagnostic defect when the active profile
# already holds the operation — callers must not invent "missing".
try:
holds, _hold_reason = gitea_config.check_operation(
required_operation,
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
except Exception:
holds = False
if holds:
report["missing_permission"] = None
report["required_permission"] = required_operation
report["diagnostic_defect"] = True
report["different_mcp_namespace_required"] = False
report["exact_safe_next_action"] = (
"Diagnostic defect: the active profile already allows "
f"{required_operation}. This is not a permission denial — "
"inspect blocker_kind / reasons (stale-runtime or runtime-mode). "
"Do not call gitea_activate_profile or switch MCP sessions."
)
report["matching_configured_profiles"] = [
p for p in [profile.get("profile_name")] if p
]
return report
matching = [] matching = []
try: try:
config = gitea_config.load_config() or {} config = gitea_config.load_config() or {}
@@ -13903,6 +13932,205 @@ def _permission_block_report(required_operation: str,
return report return report
def _reason_is_stale_runtime(reason: str) -> bool:
"""True when *reason* is a master-parity / stale-daemon refusal (#897)."""
r = (reason or "").lower()
if not r:
return False
if "stale relative to live master" in r:
return True
if "server code is stale" in r:
return True
if "daemon is stale" in r:
return True
if "started at commit" in r and "workspace master is now" in r:
return True
if "mcp server started at" in r and "stale" in r:
return True
if "restart the server to load the current capability gates" in r:
return True
if "restart/reconnect before mutating" in r:
return True
return False
def _reason_is_runtime_mode(reason: str) -> bool:
"""True when *reason* is a stable-control / runtime-mode refusal (#897)."""
r = (reason or "").lower()
if not r:
return False
if _reason_is_stale_runtime(reason):
return False
if "runtime mode could not be assessed" in r:
return True
if "runtime mode is" in r:
return True
if "stable control runtime" in r:
return True
if "dev-test" in r and ("runtime" in r or "production" in r):
return True
if "development worktree" in r or "dev worktree" in r:
return True
if "launched from a 'branches/" in r or "launched from a \"branches/" in r:
return True
if "process-root / active-workspace alignment" in r:
return True
if "namespace" in r and "reproof" in r:
return True
return False
def _reason_is_permission(reason: str) -> bool:
"""True when *reason* is a genuine profile-permission denial (#897)."""
r = (reason or "").lower()
if not r:
return False
if _reason_is_stale_runtime(reason) or _reason_is_runtime_mode(reason):
return False
if "profile could not be resolved" in r:
return True
if "profile has no configured allowed operations" in r:
return True
if "profile forbids" in r:
return True
if "profile is not allowed to" in r:
return True
if "unrecognized forbidden operation" in r:
return True
return False
def _classify_operation_gate_reasons(reasons: list[str]) -> dict:
"""Partition gate reasons into stale / runtime-mode / permission (#897)."""
stale: list[str] = []
runtime_mode: list[str] = []
permission: list[str] = []
other: list[str] = []
for reason in reasons or []:
if _reason_is_stale_runtime(reason):
stale.append(reason)
elif _reason_is_runtime_mode(reason):
runtime_mode.append(reason)
elif _reason_is_permission(reason):
permission.append(reason)
else:
other.append(reason)
return {
"stale_runtime": stale,
"runtime_mode": runtime_mode,
"permission": permission,
"other": other,
}
def _stale_runtime_reconnect_action() -> str:
"""Sanctioned recovery for a stale daemon — reconnect only (#685/#897)."""
return (
"Reconnect the IDE/client MCP session so the server reloads at the "
"current master head. Do not call gitea_activate_profile or switch "
"MCP role sessions — profile switching does not clear a stale daemon."
)
def _build_operation_gate_refusal(
required_operation: str,
reasons: list[str],
**extra_fields,
) -> dict:
"""Structured gate refusal with typed blockers (#897).
Stale-runtime and runtime-mode refusals never attach a
``permission_report`` and never recommend profile switching. True
permission denials still get ``permission_report``. When both apply,
causes are reported separately under distinct fields.
"""
classified = _classify_operation_gate_reasons(reasons)
stale = classified["stale_runtime"]
runtime_mode = classified["runtime_mode"]
permission = classified["permission"]
other = classified["other"]
blocked: dict = {
"success": False,
"performed": False,
"reasons": list(reasons),
"mutation_performed": False,
"session_context_audit": session_ctx.mutation_context_audit_fields(),
"gate_reason_classes": {
"stale_runtime": list(stale),
"runtime_mode": list(runtime_mode),
"permission": list(permission),
"other": list(other),
},
}
if stale:
parity = _current_master_parity()
blocked["blocker_kind"] = "runtime_reconnect_required"
blocked["restart_required"] = True
blocked["stop_required"] = True
blocked["startup_head"] = parity.get("startup_head")
blocked["current_head"] = parity.get("current_head")
blocked["daemon_start_head"] = (
parity.get("daemon_start_head") or parity.get("startup_head")
)
blocked["local_head"] = (
parity.get("local_head") or parity.get("current_head")
)
blocked["live_remote_head"] = parity.get("live_remote_head")
blocked["live_stale"] = bool(parity.get("live_stale"))
blocked["live_known"] = bool(parity.get("live_known"))
blocked["exact_safe_next_action"] = _stale_runtime_reconnect_action()
if permission or other:
blocked["permission_block_reasons"] = list(permission) + list(other)
blocked["stale_runtime_reasons"] = list(stale)
# Never attach permission_report for a staleness refusal.
blocked.update(extra_fields)
return blocked
if runtime_mode:
blocked["blocker_kind"] = "runtime_mode_blocked"
blocked["restart_required"] = False
blocked["stop_required"] = True
blocked["exact_safe_next_action"] = (
"Real workflow mutations run only on the promoted stable control "
"runtime. Promote/reload the stable runtime; do not call "
"gitea_activate_profile or switch MCP role sessions to clear a "
"runtime-mode block."
)
if permission or other:
blocked["permission_block_reasons"] = list(permission) + list(other)
blocked["runtime_mode_reasons"] = list(runtime_mode)
blocked.update(extra_fields)
return blocked
# Pure permission (or unclassified-as-permission) denial.
blocked["blocker_kind"] = "permission_denied"
blocked["permission_report"] = _permission_block_report(required_operation)
blocked.update(extra_fields)
return blocked
def _permission_report_for_gate_reasons(
required_operation: str,
reasons: list[str] | None,
) -> dict | None:
"""Attach ``permission_report`` only for true permission denials (#897).
Call sites that historically always attached a permission report after
``_profile_operation_gate`` should use this so stale/runtime refusals
do not emit a fabricated missing-permission payload.
"""
if not reasons:
return None
classified = _classify_operation_gate_reasons(reasons)
if classified["stale_runtime"] or classified["runtime_mode"]:
return None
if not (classified["permission"] or classified["other"]):
return None
return _permission_block_report(required_operation)
def _role_for_operation(op: str) -> str | None: def _role_for_operation(op: str) -> str | None:
# Normalize op first # Normalize op first
try: try:
@@ -14079,7 +14307,7 @@ def _master_parity_block(op: str) -> list[str]:
def _profile_operation_gate(op: str) -> list[str]: def _profile_operation_gate(op: str) -> list[str]:
"""Profile permission check for a single gated operation (#126, #216, #420). """Profile permission check for a single gated operation (#126, #216, #420, #897).
Issue discussion comments are gated separately from the gitea.pr.* Issue discussion comments are gated separately from the gitea.pr.*
review/merge family: listing requires ``gitea.read``, creating requires review/merge family: listing requires ``gitea.read``, creating requires
@@ -14092,21 +14320,26 @@ def _profile_operation_gate(op: str) -> list[str]:
capability gate that has since been merged, and when the runtime itself is capability gate that has since been merged, and when the runtime itself is
not the promoted stable control runtime (#615) -- a dev/test or unknown not the promoted stable control runtime (#615) -- a dev/test or unknown
runtime holds production credentials but has not been promoted. runtime holds production credentials but has not been promoted.
#897: collect *all* independent refusal classes (stale, runtime-mode,
permission) rather than short-circuiting after the first. Callers that
only need a boolean still treat any non-empty list as blocked; typed
consumers (``_build_operation_gate_refusal``) can separate causes.
""" """
stale_reasons = _master_parity_block(op) reasons: list[str] = []
if stale_reasons: reasons.extend(_master_parity_block(op))
return stale_reasons reasons.extend(_runtime_mode_block(op))
runtime_reasons = _runtime_mode_block(op)
if runtime_reasons:
return runtime_reasons
try: try:
profile = get_profile() profile = get_profile()
except Exception as exc: except Exception as exc:
return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"] reasons.append(
f"profile could not be resolved (fail closed): {_redact(str(exc))}"
)
return reasons
op_ok, op_reason = gitea_config.check_operation( op_ok, op_reason = gitea_config.check_operation(
op, profile["allowed_operations"], profile["forbidden_operations"]) op, profile["allowed_operations"], profile["forbidden_operations"])
if op_ok: if op_ok:
return [] return reasons
if _try_auto_switch_for_operation(op): if _try_auto_switch_for_operation(op):
try: try:
@@ -14114,17 +14347,26 @@ def _profile_operation_gate(op: str) -> list[str]:
op_ok, op_reason = gitea_config.check_operation( op_ok, op_reason = gitea_config.check_operation(
op, profile["allowed_operations"], profile["forbidden_operations"]) op, profile["allowed_operations"], profile["forbidden_operations"])
if op_ok: if op_ok:
return [] return reasons
except Exception as exc: except Exception as exc:
return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"] reasons.append(
f"profile could not be resolved (fail closed): {_redact(str(exc))}"
)
return reasons
if op_reason == "no-allowed-operations": if op_reason == "no-allowed-operations":
return ["profile has no configured allowed operations (fail closed)"] reasons.append(
if op_reason == "forbidden": "profile has no configured allowed operations (fail closed)"
return [f"profile forbids '{op}'"] )
if op_reason == "invalid-forbidden-entry": elif op_reason == "forbidden":
return ["profile has an unrecognized forbidden operation entry (fail closed)"] reasons.append(f"profile forbids '{op}'")
return [f"profile is not allowed to {op}"] elif op_reason == "invalid-forbidden-entry":
reasons.append(
"profile has an unrecognized forbidden operation entry (fail closed)"
)
else:
reasons.append(f"profile is not allowed to {op}")
return reasons
def _mutation_config_authority_block(required_operation: str) -> dict | None: def _mutation_config_authority_block(required_operation: str) -> dict | None:
@@ -14344,10 +14586,14 @@ def _session_context_mutation_block(
def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None: def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
"""Structured permission denial for gated tools (#69, #142). """Structured operation-gate denial for gated tools (#69, #142, #897).
Returns a block dict when the active profile forbids *required_operation*, Returns a block dict when the active profile forbids *required_operation*,
or ``None`` when the gate passes. Never performs network I/O. the daemon is stale, or the runtime mode is not mutation-safe or
``None`` when the gate passes. Never performs network I/O.
#897: stale-runtime and runtime-mode refusals are typed
(``blocker_kind``) and never carry a ``permission_report``.
""" """
req_role = "reviewer" if any(required_operation.startswith(p) for p in ( req_role = "reviewer" if any(required_operation.startswith(p) for p in (
"gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review" "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes", "gitea.pr.review"
@@ -14357,15 +14603,9 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
reasons = _profile_operation_gate(required_operation) reasons = _profile_operation_gate(required_operation)
if reasons: if reasons:
blocked = { return _build_operation_gate_refusal(
"success": False, required_operation, reasons, **extra_fields
"performed": False, )
"reasons": reasons,
"permission_report": _permission_block_report(required_operation),
"session_context_audit": session_ctx.mutation_context_audit_fields(),
}
blocked.update(extra_fields)
return blocked
auth_block = _mutation_config_authority_block(required_operation) auth_block = _mutation_config_authority_block(required_operation)
if auth_block is not None: if auth_block is not None:
@@ -14494,20 +14734,14 @@ def gitea_acquire_reviewer_pr_lease(
"""Acquire a per-PR reviewer lease before review/merge mutations (#407).""" """Acquire a per-PR reviewer lease before review/merge mutations (#407)."""
read_block = _profile_operation_gate("gitea.read") read_block = _profile_operation_gate("gitea.read")
if read_block: if read_block:
return { return _build_operation_gate_refusal(
"success": False, "gitea.read", read_block, acquired=False
"acquired": False, )
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comment_block = _profile_operation_gate("gitea.pr.comment") comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block: if comment_block:
return { return _build_operation_gate_refusal(
"success": False, "gitea.pr.comment", comment_block, acquired=False
"acquired": False, )
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
# task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604 # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
# anti-stomp for the declared lease-acquire mutation inventory entry. # anti-stomp for the declared lease-acquire mutation inventory entry.
@@ -14623,20 +14857,14 @@ def gitea_acquire_merger_pr_lease(
""" """
read_block = _profile_operation_gate("gitea.read") read_block = _profile_operation_gate("gitea.read")
if read_block: if read_block:
return { return _build_operation_gate_refusal(
"success": False, "gitea.read", read_block, acquired=False
"acquired": False, )
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comment_block = _profile_operation_gate("gitea.pr.comment") comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block: if comment_block:
return { return _build_operation_gate_refusal(
"success": False, "gitea.pr.comment", comment_block, acquired=False
"acquired": False, )
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
merge_block = _profile_operation_gate("gitea.pr.merge") merge_block = _profile_operation_gate("gitea.pr.merge")
if merge_block: if merge_block:
return { return {
@@ -19375,14 +19603,12 @@ def gitea_update_pr_branch_by_merge(
# Permission: author branch push / PR mutation surface. # Permission: author branch push / PR mutation surface.
push_block = _profile_operation_gate("gitea.branch.push") push_block = _profile_operation_gate("gitea.branch.push")
if push_block: if push_block:
return { return _build_operation_gate_refusal(
"success": False, "gitea.branch.push",
"performed": False, push_block,
"mutation_allowed": False, mutation_allowed=False,
"reasons": push_block, role_kind=role,
"permission_report": _permission_block_report("gitea.branch.push"), )
"role_kind": role,
}
if role != "author": if role != "author":
pre = pr_sync_status.assess_update_pr_branch_preflight( pre = pr_sync_status.assess_update_pr_branch_preflight(
@@ -0,0 +1,453 @@
"""#897: stale-runtime / runtime-mode refusals must not look like permission denials.
Acceptance criteria (issue #897):
* Stale-runtime and runtime-mode refusals are typed distinctly from
profile-permission refusals (distinct ``blocker_kind``).
* A refusal caused by staleness or runtime mode never emits a
``permission_report`` and never names a permission the active profile holds.
* ``_permission_block_report`` verifies the active profile actually lacks the
operation before reporting it missing.
* A stale-runtime refusal reports reconnect-only recovery and never recommends
``gitea_activate_profile`` or an MCP session switch.
* The blocker payload states the observed heads (parity fields).
* Matrix across author / reviewer / merger / reconciler profiles.
* Regression: ``gitea_create_issue`` on a stale daemon under ``prgs-author``
never returns ``missing_permission: gitea.issue.create``.
"""
from __future__ import annotations
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_config # noqa: E402
import gitea_mcp_server as mcp_server # noqa: E402
SHA_START = "7af40fb5ff7debd5e9165fe97d9c7c279358e175"
SHA_LIVE = "2f4dec832327513118f2fe92b74da25d124a01cb"
ROLE_MATRIX = (
(
"prgs-author",
"author",
"gitea.issue.create",
[
"gitea.read",
"gitea.issue.create",
"gitea.issue.comment",
"gitea.issue.close",
"gitea.branch.create",
"gitea.branch.push",
"gitea.pr.create",
"gitea.pr.comment",
"gitea.repo.commit",
],
["gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes"],
"gitea.pr.merge", # forbidden op for pure-permission case
),
(
"prgs-reviewer",
"reviewer",
"gitea.pr.review",
[
"gitea.read",
"gitea.pr.review",
"gitea.pr.approve",
"gitea.pr.request_changes",
"gitea.pr.comment",
"gitea.issue.comment",
],
["gitea.branch.push", "gitea.pr.create"],
"gitea.branch.push",
),
(
"prgs-merger",
"merger",
"gitea.pr.merge",
[
"gitea.read",
"gitea.pr.merge",
"gitea.pr.comment",
"gitea.issue.comment",
],
["gitea.pr.approve", "gitea.branch.push", "gitea.pr.create"],
"gitea.branch.push",
),
(
"prgs-reconciler",
"reconciler",
"gitea.branch.delete",
[
"gitea.read",
"gitea.branch.delete",
"gitea.pr.comment",
"gitea.issue.comment",
"gitea.pr.close",
"gitea.issue.close",
],
["gitea.pr.approve", "gitea.pr.merge"],
"gitea.pr.merge",
),
)
def _profile(name: str, role: str, allowed: list[str], forbidden: list[str]) -> dict:
return {
"profile_name": name,
"role": role,
"role_kind": role,
"allowed_operations": list(allowed),
"forbidden_operations": list(forbidden),
"identity": "test-user",
}
def _config(profiles: dict) -> dict:
return {
"version": 2,
"profiles": {
name: {
"role": p["role"],
"allowed_operations": p["allowed_operations"],
"forbidden_operations": p["forbidden_operations"],
}
for name, p in profiles.items()
},
"rules": {"allow_runtime_switching": True},
}
class Issue897Helpers(unittest.TestCase):
def test_classify_stale_reason_strings(self):
stale = (
f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
f"at {SHA_START[:12]}; the daemon is stale relative to live master "
"-- restart/reconnect before mutating"
)
classified = mcp_server._classify_operation_gate_reasons([stale])
self.assertEqual(classified["stale_runtime"], [stale])
self.assertEqual(classified["permission"], [])
self.assertEqual(classified["runtime_mode"], [])
def test_classify_permission_reason(self):
reason = "profile is not allowed to gitea.pr.merge"
classified = mcp_server._classify_operation_gate_reasons([reason])
self.assertEqual(classified["permission"], [reason])
self.assertEqual(classified["stale_runtime"], [])
def test_classify_runtime_mode_reason(self):
reason = (
"runtime mode is 'dev-test' and the mutation targets the "
"production repository; dev/test runtimes must not mutate real "
"issues or PRs (ADR: stable control runtime vs dev runtime)"
)
classified = mcp_server._classify_operation_gate_reasons([reason])
self.assertEqual(classified["runtime_mode"], [reason])
self.assertEqual(classified["stale_runtime"], [])
class Issue897PermissionBlockReport(unittest.TestCase):
def test_holds_op_is_diagnostic_defect_not_missing_permission(self):
profile = _profile(
"prgs-author",
"author",
["gitea.read", "gitea.issue.create", "gitea.issue.comment"],
[],
)
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
mcp_server.gitea_config, "load_config", return_value=_config({"prgs-author": profile})
), patch.object(
mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
):
report = mcp_server._permission_block_report("gitea.issue.create")
self.assertTrue(report.get("diagnostic_defect"), report)
self.assertIsNone(report.get("missing_permission"), report)
action = (report.get("exact_safe_next_action") or "").lower()
# Must not *recommend* profile switching; mentioning the forbidden
# action in a "do not call" instruction is fine.
self.assertNotIn("call gitea_activate_profile with", action)
self.assertNotIn("switch to the author mcp session", action)
self.assertNotIn("switch to the reviewer mcp session", action)
self.assertIn("diagnostic defect", action)
def test_true_missing_permission_still_reports(self):
profile = _profile(
"prgs-author",
"author",
["gitea.read", "gitea.issue.create"],
["gitea.pr.merge"],
)
reviewer = _profile(
"prgs-reviewer",
"reviewer",
["gitea.read", "gitea.pr.merge", "gitea.pr.approve"],
[],
)
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
mcp_server.gitea_config,
"load_config",
return_value=_config({"prgs-author": profile, "prgs-reviewer": reviewer}),
), patch.object(
mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=True
):
report = mcp_server._permission_block_report("gitea.pr.merge")
self.assertFalse(report.get("diagnostic_defect"), report)
self.assertEqual(report.get("missing_permission"), "gitea.pr.merge")
self.assertIn("prgs-reviewer", report.get("matching_configured_profiles") or [])
class Issue897GateRefusalMatrix(unittest.TestCase):
def _stale_parity(self) -> dict:
return {
"in_parity": True,
"stale": False,
"restart_required": True,
"determinable": True,
"startup_head": SHA_START,
"current_head": SHA_START,
"daemon_start_head": SHA_START,
"local_head": SHA_START,
"live_remote_head": SHA_LIVE,
"live_known": True,
"live_stale": True,
"mutation_safe": False,
"reasons": [
f"live remote master is {SHA_LIVE[:12]} but the MCP server "
f"started at {SHA_START[:12]}; the daemon is stale relative "
"to live master -- restart/reconnect before mutating"
],
}
def test_stale_plus_permitted_op_all_roles(self):
for name, role, permitted_op, allowed, forbidden, _forbidden_op in ROLE_MATRIX:
with self.subTest(profile=name, op=permitted_op):
profile = _profile(name, role, allowed, forbidden)
parity = self._stale_parity()
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
mcp_server, "_current_master_parity", return_value=parity
), patch.object(
mcp_server, "_master_parity_block", return_value=list(parity["reasons"])
), patch.object(
mcp_server, "_runtime_mode_block", return_value=[]
), patch.object(
mcp_server, "_ensure_matching_profile", return_value=None
), patch.object(
mcp_server.session_ctx,
"mutation_context_audit_fields",
return_value={"session_profile": name},
):
blocked = mcp_server._profile_permission_block(permitted_op)
self.assertIsNotNone(blocked, name)
assert blocked is not None
self.assertEqual(
blocked.get("blocker_kind"),
"runtime_reconnect_required",
blocked,
)
self.assertNotIn("permission_report", blocked, blocked)
self.assertTrue(blocked.get("restart_required"), blocked)
self.assertEqual(blocked.get("startup_head"), SHA_START, blocked)
self.assertEqual(blocked.get("live_remote_head"), SHA_LIVE, blocked)
action = (blocked.get("exact_safe_next_action") or "").lower()
self.assertIn("reconnect", action)
self.assertNotIn("call gitea_activate_profile with", action)
self.assertNotIn("switch to the author mcp session", action)
self.assertNotIn("switch to the reviewer mcp session", action)
def test_fresh_plus_forbidden_op_all_roles(self):
for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
with self.subTest(profile=name, op=forbidden_op):
profile = _profile(name, role, allowed, forbidden)
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
mcp_server, "_master_parity_block", return_value=[]
), patch.object(
mcp_server, "_runtime_mode_block", return_value=[]
), patch.object(
mcp_server, "_ensure_matching_profile", return_value=None
), patch.object(
mcp_server.session_ctx,
"mutation_context_audit_fields",
return_value={"session_profile": name},
), patch.object(
mcp_server.gitea_config,
"load_config",
return_value=_config({name: profile}),
), patch.object(
mcp_server.gitea_config, "is_runtime_switching_enabled", return_value=False
):
blocked = mcp_server._profile_permission_block(forbidden_op)
self.assertIsNotNone(blocked, name)
assert blocked is not None
self.assertEqual(blocked.get("blocker_kind"), "permission_denied", blocked)
self.assertIn("permission_report", blocked, blocked)
report = blocked["permission_report"]
self.assertEqual(report.get("missing_permission"), forbidden_op, report)
self.assertFalse(report.get("diagnostic_defect"), report)
# No runtime reconnect fields for pure permission denial
self.assertNotEqual(
blocked.get("blocker_kind"), "runtime_reconnect_required"
)
def test_stale_plus_forbidden_op_both_causes_separated(self):
for name, role, _permitted, allowed, forbidden, forbidden_op in ROLE_MATRIX:
with self.subTest(profile=name, op=forbidden_op):
profile = _profile(name, role, allowed, forbidden)
parity = self._stale_parity()
stale_reason = parity["reasons"][0]
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
mcp_server, "_current_master_parity", return_value=parity
), patch.object(
mcp_server, "_master_parity_block", return_value=[stale_reason]
), patch.object(
mcp_server, "_runtime_mode_block", return_value=[]
), patch.object(
mcp_server, "_ensure_matching_profile", return_value=None
), patch.object(
mcp_server.session_ctx,
"mutation_context_audit_fields",
return_value={"session_profile": name},
):
# Gate collects both classes; force permission reason too.
with patch.object(
mcp_server,
"_profile_operation_gate",
return_value=[
stale_reason,
f"profile is not allowed to {forbidden_op}",
],
):
blocked = mcp_server._profile_permission_block(forbidden_op)
self.assertIsNotNone(blocked)
assert blocked is not None
self.assertEqual(
blocked.get("blocker_kind"), "runtime_reconnect_required", blocked
)
self.assertNotIn("permission_report", blocked, blocked)
self.assertIn("permission_block_reasons", blocked, blocked)
self.assertIn("stale_runtime_reasons", blocked, blocked)
classes = blocked.get("gate_reason_classes") or {}
self.assertTrue(classes.get("stale_runtime"), classes)
self.assertTrue(classes.get("permission"), classes)
def test_runtime_mode_block_no_permission_report(self):
profile = _profile(
"prgs-author",
"author",
["gitea.read", "gitea.issue.create"],
[],
)
runtime_reason = (
"runtime mode is 'dev-test' and the mutation targets the "
"production repository; dev/test runtimes must not mutate real "
"issues or PRs (ADR: stable control runtime vs dev runtime)"
)
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
mcp_server, "_master_parity_block", return_value=[]
), patch.object(
mcp_server, "_runtime_mode_block", return_value=[runtime_reason]
), patch.object(
mcp_server, "_ensure_matching_profile", return_value=None
), patch.object(
mcp_server.session_ctx,
"mutation_context_audit_fields",
return_value={"session_profile": "prgs-author"},
):
blocked = mcp_server._profile_permission_block("gitea.issue.create")
self.assertIsNotNone(blocked)
assert blocked is not None
self.assertEqual(blocked.get("blocker_kind"), "runtime_mode_blocked", blocked)
self.assertNotIn("permission_report", blocked, blocked)
action = (blocked.get("exact_safe_next_action") or "").lower()
self.assertNotIn("call gitea_activate_profile with", action)
self.assertIn("stable control runtime", action)
class Issue897CreateIssueRegression(unittest.TestCase):
def test_create_issue_stale_daemon_never_missing_issue_create(self):
"""Regression AC: stale prgs-author create_issue must not claim missing create."""
profile = _profile(
"prgs-author",
"author",
[
"gitea.read",
"gitea.issue.create",
"gitea.issue.comment",
"gitea.branch.create",
"gitea.branch.push",
"gitea.pr.create",
"gitea.pr.comment",
"gitea.repo.commit",
],
[],
)
stale_reason = (
f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
f"at {SHA_START[:12]}; the daemon is stale relative to live master "
"-- restart/reconnect before mutating"
)
parity = {
"in_parity": True,
"stale": False,
"restart_required": True,
"determinable": True,
"startup_head": SHA_START,
"current_head": SHA_START,
"daemon_start_head": SHA_START,
"local_head": SHA_START,
"live_remote_head": SHA_LIVE,
"live_known": True,
"live_stale": True,
"mutation_safe": False,
"reasons": [stale_reason],
}
with patch.object(mcp_server, "get_profile", return_value=profile), patch.object(
mcp_server, "_current_master_parity", return_value=parity
), patch.object(
mcp_server, "_master_parity_block", return_value=[stale_reason]
), patch.object(
mcp_server, "_runtime_mode_block", return_value=[]
), patch.object(
mcp_server, "_ensure_matching_profile", return_value=None
), patch.object(
mcp_server.session_ctx,
"mutation_context_audit_fields",
return_value={"session_profile": "prgs-author"},
), patch.object(
mcp_server, "_mutation_config_authority_block", return_value=None
), patch.object(
mcp_server, "_session_context_mutation_block", return_value=None
):
blocked = mcp_server._profile_permission_block(
"gitea.issue.create", remote="prgs"
)
self.assertIsNotNone(blocked)
assert blocked is not None
self.assertEqual(blocked.get("blocker_kind"), "runtime_reconnect_required")
self.assertNotIn("permission_report", blocked)
# Even if a caller still built a raw report, holds-check must not claim missing.
with patch.object(mcp_server, "get_profile", return_value=profile):
raw = mcp_server._permission_block_report("gitea.issue.create")
self.assertIsNone(raw.get("missing_permission"), raw)
self.assertNotEqual(raw.get("missing_permission"), "gitea.issue.create")
def test_permission_report_for_gate_reasons_skips_stale(self):
stale = (
f"live remote master is {SHA_LIVE[:12]} but the MCP server started "
f"at {SHA_START[:12]}; the daemon is stale relative to live master "
"-- restart/reconnect before mutating"
)
self.assertIsNone(
mcp_server._permission_report_for_gate_reasons(
"gitea.issue.comment", [stale]
)
)
if __name__ == "__main__":
unittest.main()