feat: expose role-aware runtime context capabilities (#139) #257
@@ -4103,6 +4103,90 @@ def gitea_get_profile(
|
|||||||
return result
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
_RUNTIME_CAPABILITY_TASKS = (
|
||||||
|
"create_issue",
|
||||||
|
"comment_issue",
|
||||||
|
"create_pr",
|
||||||
|
"review_pr",
|
||||||
|
"merge_pr",
|
||||||
|
"close_issue",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _matching_configured_profiles(
|
||||||
|
config: dict | None,
|
||||||
|
required_permission: str,
|
||||||
|
) -> list[str]:
|
||||||
|
"""Profile names that allow *required_permission* (redacted metadata only)."""
|
||||||
|
if not config or "profiles" not in config:
|
||||||
|
return []
|
||||||
|
matches: list[str] = []
|
||||||
|
for p_name, p_data in config["profiles"].items():
|
||||||
|
if not p_data.get("enabled", True):
|
||||||
|
continue
|
||||||
|
p_allowed = p_data.get("allowed_operations") or []
|
||||||
|
p_forbidden = p_data.get("forbidden_operations") or []
|
||||||
|
p_allowed_n = []
|
||||||
|
for op in p_allowed:
|
||||||
|
try:
|
||||||
|
p_allowed_n.append(gitea_config.normalize_operation(op))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
p_forbidden_n = []
|
||||||
|
for op in p_forbidden:
|
||||||
|
try:
|
||||||
|
p_forbidden_n.append(gitea_config.normalize_operation(op))
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
ok, _ = gitea_config.check_operation(
|
||||||
|
required_permission, p_allowed_n, p_forbidden_n
|
||||||
|
)
|
||||||
|
if ok:
|
||||||
|
matches.append(p_name)
|
||||||
|
return sorted(matches)
|
||||||
|
|
||||||
|
|
||||||
|
def _build_runtime_task_capabilities(
|
||||||
|
allowed: list[str],
|
||||||
|
forbidden: list[str],
|
||||||
|
config: dict | None,
|
||||||
|
) -> dict:
|
||||||
|
"""Per-task capability summary for role-aware runtime context (#139)."""
|
||||||
|
task_entries = []
|
||||||
|
flags: dict[str, bool] = {}
|
||||||
|
flag_keys = {
|
||||||
|
"create_issue": "can_create_issues",
|
||||||
|
"comment_issue": "can_comment_on_issues",
|
||||||
|
"create_pr": "can_author_prs",
|
||||||
|
"review_pr": "can_review_prs",
|
||||||
|
"merge_pr": "can_merge_prs",
|
||||||
|
"close_issue": "can_close_issues",
|
||||||
|
}
|
||||||
|
for task in _RUNTIME_CAPABILITY_TASKS:
|
||||||
|
permission = task_capability_map.required_permission(task)
|
||||||
|
allowed_here, _ = gitea_config.check_operation(
|
||||||
|
permission, allowed, forbidden
|
||||||
|
)
|
||||||
|
entry = {
|
||||||
|
"task": task,
|
||||||
|
"required_permission": permission,
|
||||||
|
"required_role_kind": task_capability_map.required_role(task),
|
||||||
|
"allowed_in_current_session": allowed_here,
|
||||||
|
"matching_configured_profiles": _matching_configured_profiles(
|
||||||
|
config, permission
|
||||||
|
),
|
||||||
|
}
|
||||||
|
task_entries.append(entry)
|
||||||
|
flag_name = flag_keys.get(task)
|
||||||
|
if flag_name:
|
||||||
|
flags[flag_name] = allowed_here
|
||||||
|
return {
|
||||||
|
**flags,
|
||||||
|
"issue_comment_not_implied_by_pr_comment": True,
|
||||||
|
"task_capabilities": task_entries,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def gitea_get_runtime_context(
|
def gitea_get_runtime_context(
|
||||||
remote: str = "dadeschools",
|
remote: str = "dadeschools",
|
||||||
@@ -4190,6 +4274,10 @@ def gitea_get_runtime_context(
|
|||||||
"or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
|
"or ask the operator to update GITEA_MCP_PROFILE to a reviewer profile."
|
||||||
)
|
)
|
||||||
|
|
||||||
|
session_capabilities = _build_runtime_task_capabilities(
|
||||||
|
allowed, forbidden, config
|
||||||
|
)
|
||||||
|
|
||||||
preflight = assess_preflight_status()
|
preflight = assess_preflight_status()
|
||||||
if not preflight["preflight_ready"]:
|
if not preflight["preflight_ready"]:
|
||||||
safe_next_action = (
|
safe_next_action = (
|
||||||
@@ -4214,6 +4302,7 @@ def gitea_get_runtime_context(
|
|||||||
"safe_next_action": safe_next_action,
|
"safe_next_action": safe_next_action,
|
||||||
"preflight_ready": preflight["preflight_ready"],
|
"preflight_ready": preflight["preflight_ready"],
|
||||||
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
||||||
|
"session_capabilities": session_capabilities,
|
||||||
}
|
}
|
||||||
|
|
||||||
if reveal and h:
|
if reveal and h:
|
||||||
|
|||||||
@@ -95,9 +95,13 @@ class TestRuntimeClarity(unittest.TestCase):
|
|||||||
# -------------------------------------------------------------------------
|
# -------------------------------------------------------------------------
|
||||||
# gitea_get_runtime_context
|
# gitea_get_runtime_context
|
||||||
# -------------------------------------------------------------------------
|
# -------------------------------------------------------------------------
|
||||||
|
@patch(
|
||||||
|
"mcp_server.assess_preflight_status",
|
||||||
|
return_value={"preflight_ready": True, "preflight_block_reasons": []},
|
||||||
|
)
|
||||||
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||||
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||||
def test_get_runtime_context_author(self, _auth, _api):
|
def test_get_runtime_context_author(self, _auth, _api, _preflight):
|
||||||
with patch.dict(os.environ, self._env("author-profile"), clear=True):
|
with patch.dict(os.environ, self._env("author-profile"), clear=True):
|
||||||
ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
|
ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
|
||||||
self.assertEqual(ctx["active_profile"], "author-profile")
|
self.assertEqual(ctx["active_profile"], "author-profile")
|
||||||
@@ -110,10 +114,26 @@ class TestRuntimeClarity(unittest.TestCase):
|
|||||||
self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
|
self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
|
||||||
self.assertIn("does not permit review or merge", ctx["review_merge_blocked_reasons"][0])
|
self.assertIn("does not permit review or merge", ctx["review_merge_blocked_reasons"][0])
|
||||||
self.assertIn("Switch to the reviewer MCP session", ctx["safe_next_action"])
|
self.assertIn("Switch to the reviewer MCP session", ctx["safe_next_action"])
|
||||||
|
caps = ctx["session_capabilities"]
|
||||||
|
self.assertTrue(caps["can_author_prs"])
|
||||||
|
self.assertFalse(caps["can_create_issues"])
|
||||||
|
self.assertFalse(caps["can_comment_on_issues"])
|
||||||
|
self.assertFalse(caps["can_review_prs"])
|
||||||
|
self.assertFalse(caps["can_merge_prs"])
|
||||||
|
self.assertTrue(caps["issue_comment_not_implied_by_pr_comment"])
|
||||||
|
merge_entry = next(
|
||||||
|
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
|
||||||
|
)
|
||||||
|
self.assertFalse(merge_entry["allowed_in_current_session"])
|
||||||
|
self.assertIn("reviewer-profile", merge_entry["matching_configured_profiles"])
|
||||||
|
|
||||||
|
@patch(
|
||||||
|
"mcp_server.assess_preflight_status",
|
||||||
|
return_value={"preflight_ready": True, "preflight_block_reasons": []},
|
||||||
|
)
|
||||||
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
|
||||||
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
|
||||||
def test_get_runtime_context_reviewer(self, _auth, _api):
|
def test_get_runtime_context_reviewer(self, _auth, _api, _preflight):
|
||||||
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
|
||||||
ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
|
ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
|
||||||
self.assertEqual(ctx["active_profile"], "reviewer-profile")
|
self.assertEqual(ctx["active_profile"], "reviewer-profile")
|
||||||
@@ -121,6 +141,15 @@ class TestRuntimeClarity(unittest.TestCase):
|
|||||||
self.assertTrue(ctx["review_merge_allowed"])
|
self.assertTrue(ctx["review_merge_allowed"])
|
||||||
self.assertEqual(ctx["suggested_fix"], "none")
|
self.assertEqual(ctx["suggested_fix"], "none")
|
||||||
self.assertEqual(ctx["safe_next_action"], "None; ready for operations.")
|
self.assertEqual(ctx["safe_next_action"], "None; ready for operations.")
|
||||||
|
caps = ctx["session_capabilities"]
|
||||||
|
self.assertFalse(caps["can_author_prs"])
|
||||||
|
self.assertFalse(caps["can_create_issues"])
|
||||||
|
self.assertFalse(caps["can_review_prs"])
|
||||||
|
self.assertTrue(caps["can_merge_prs"])
|
||||||
|
merge_entry = next(
|
||||||
|
t for t in caps["task_capabilities"] if t["task"] == "merge_pr"
|
||||||
|
)
|
||||||
|
self.assertTrue(merge_entry["allowed_in_current_session"])
|
||||||
|
|
||||||
# -------------------------------------------------------------------------
|
# -------------------------------------------------------------------------
|
||||||
# gitea_list_profiles
|
# gitea_list_profiles
|
||||||
|
|||||||
Reference in New Issue
Block a user