diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index fa63126..91f1b3a 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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( diff --git a/tests/test_issue_897_permission_stale_runtime_classification.py b/tests/test_issue_897_permission_stale_runtime_classification.py new file mode 100644 index 0000000..cea4620 --- /dev/null +++ b/tests/test_issue_897_permission_stale_runtime_classification.py @@ -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()