"""#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()