Compare commits

..
Author SHA1 Message Date
sysadmin 6e6ca94338 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.
2026-07-25 01:47:35 -04:00
sysadmin 2f4dec8323 Merge pull request 'feat(restart): pre-restart drain proof and hard gate (Closes #661)' (#882) from feat/issue-661-drain-proof-hard-gate into master 2026-07-24 23:40:02 -05:00
2 changed files with 749 additions and 70 deletions
+296 -70
View File
@@ -9552,15 +9552,13 @@ def gitea_edit_pr(
if closing:
gate_reasons = _profile_operation_gate("gitea.pr.close")
if gate_reasons:
return {
"success": False,
"performed": False,
"pr_number": pr_number,
"requested_state": "closed",
"required_permission": "gitea.pr.close",
"reasons": gate_reasons,
"permission_report": _permission_block_report("gitea.pr.close"),
}
return _build_operation_gate_refusal(
"gitea.pr.close",
gate_reasons,
pr_number=pr_number,
requested_state="closed",
required_permission="gitea.pr.close",
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
@@ -13823,13 +13821,17 @@ def gitea_view_issue(
def _permission_block_report(required_operation: str,
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
refusal and never widens any permission, performs network I/O, or
raises (fail-soft: degrades to a minimal fail-closed report). Names
configured profiles only never auth references, tokens, endpoint
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 = {
"requested_operation": required_operation,
@@ -13841,6 +13843,7 @@ def _permission_block_report(required_operation: str,
"matching_configured_profiles": [],
"runtime_switching_supported": False,
"different_mcp_namespace_required": True,
"diagnostic_defect": False,
"exact_safe_next_action": (
"Ask the operator to fix GITEA_MCP_CONFIG/GITEA_MCP_PROFILE; "
"the active profile could not be resolved (fail closed)."),
@@ -13855,6 +13858,32 @@ def _permission_block_report(required_operation: str,
report["active_allowed_operations"] = (
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 = []
try:
config = gitea_config.load_config() or {}
@@ -13903,6 +13932,205 @@ def _permission_block_report(required_operation: str,
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:
# Normalize op first
try:
@@ -14079,7 +14307,7 @@ def _master_parity_block(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.*
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
not the promoted stable control runtime (#615) -- a dev/test or unknown
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)
if stale_reasons:
return stale_reasons
runtime_reasons = _runtime_mode_block(op)
if runtime_reasons:
return runtime_reasons
reasons: list[str] = []
reasons.extend(_master_parity_block(op))
reasons.extend(_runtime_mode_block(op))
try:
profile = get_profile()
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, profile["allowed_operations"], profile["forbidden_operations"])
if op_ok:
return []
return reasons
if _try_auto_switch_for_operation(op):
try:
@@ -14114,17 +14347,26 @@ def _profile_operation_gate(op: str) -> list[str]:
op_ok, op_reason = gitea_config.check_operation(
op, profile["allowed_operations"], profile["forbidden_operations"])
if op_ok:
return []
return reasons
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":
return ["profile has no configured allowed operations (fail closed)"]
if op_reason == "forbidden":
return [f"profile forbids '{op}'"]
if op_reason == "invalid-forbidden-entry":
return ["profile has an unrecognized forbidden operation entry (fail closed)"]
return [f"profile is not allowed to {op}"]
reasons.append(
"profile has no configured allowed operations (fail closed)"
)
elif op_reason == "forbidden":
reasons.append(f"profile forbids '{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:
@@ -14344,10 +14586,14 @@ def _session_context_mutation_block(
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*,
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 (
"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)
if reasons:
blocked = {
"success": False,
"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
return _build_operation_gate_refusal(
required_operation, reasons, **extra_fields
)
auth_block = _mutation_config_authority_block(required_operation)
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)."""
read_block = _profile_operation_gate("gitea.read")
if read_block:
return {
"success": False,
"acquired": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
return _build_operation_gate_refusal(
"gitea.read", read_block, acquired=False
)
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
return {
"success": False,
"acquired": False,
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
return _build_operation_gate_refusal(
"gitea.pr.comment", comment_block, acquired=False
)
# task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
# 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")
if read_block:
return {
"success": False,
"acquired": False,
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
return _build_operation_gate_refusal(
"gitea.read", read_block, acquired=False
)
comment_block = _profile_operation_gate("gitea.pr.comment")
if comment_block:
return {
"success": False,
"acquired": False,
"reasons": comment_block,
"permission_report": _permission_block_report("gitea.pr.comment"),
}
return _build_operation_gate_refusal(
"gitea.pr.comment", comment_block, acquired=False
)
merge_block = _profile_operation_gate("gitea.pr.merge")
if merge_block:
return {
@@ -19375,14 +19603,12 @@ def gitea_update_pr_branch_by_merge(
# Permission: author branch push / PR mutation surface.
push_block = _profile_operation_gate("gitea.branch.push")
if push_block:
return {
"success": False,
"performed": False,
"mutation_allowed": False,
"reasons": push_block,
"permission_report": _permission_block_report("gitea.branch.push"),
"role_kind": role,
}
return _build_operation_gate_refusal(
"gitea.branch.push",
push_block,
mutation_allowed=False,
role_kind=role,
)
if role != "author":
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()