feat: add session task capabilities to runtime context (#139)

gitea_get_runtime_context now reports per-task allowed_in_current_session flags, matching configured profiles, and aggregate session_capabilities so LLMs can determine author/comment/review/merge ability without guessing launcher profile config. Refs #139 Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-06 14:32:07 -04:00
parent 183e9f08b8
commit 95d01b07fe
2 changed files with 120 additions and 2 deletions
@@ -4103,6 +4103,90 @@ def gitea_get_profile(
    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()
 def gitea_get_runtime_context(
    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."
        )

+    session_capabilities = _build_runtime_task_capabilities(
+        allowed, forbidden, config
+    )
+
    preflight = assess_preflight_status()
    if not preflight["preflight_ready"]:
        safe_next_action = (
@@ -4214,6 +4302,7 @@ def gitea_get_runtime_context(
        "safe_next_action": safe_next_action,
        "preflight_ready": preflight["preflight_ready"],
        "preflight_block_reasons": preflight["preflight_block_reasons"],
+        "session_capabilities": session_capabilities,
    }

    if reveal and h:
@@ -95,9 +95,13 @@ class TestRuntimeClarity(unittest.TestCase):
    # -------------------------------------------------------------------------
    # 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.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):
            ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
        self.assertEqual(ctx["active_profile"], "author-profile")
@@ -110,10 +114,26 @@ class TestRuntimeClarity(unittest.TestCase):
        self.assertEqual(ctx["suggested_fix"], "reviewer namespace")
        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"])
+        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.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):
            ctx = mcp_server.gitea_get_runtime_context(remote="dadeschools")
        self.assertEqual(ctx["active_profile"], "reviewer-profile")
@@ -121,6 +141,15 @@ class TestRuntimeClarity(unittest.TestCase):
        self.assertTrue(ctx["review_merge_allowed"])
        self.assertEqual(ctx["suggested_fix"], "none")
        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