FEAT - TASKER CODEX: manage agent project grants

plane-src/apps/api/plane/app/views/__init__.py

@@ -177,6 +177,7 @@ from .api import ApiTokenEndpoint
 from .codex_agents import (
     CodexAgentDetailEndpoint,
     CodexAgentGrantListEndpoint,
+    CodexAgentGrantReplaceEndpoint,
     CodexAgentListEndpoint,
     CodexAgentRevokeEndpoint,
     CodexAgentSetupEndpoint,

plane-src/apps/api/plane/app/views/codex_agents.py

@@ -139,6 +139,41 @@ def validate_project_in_workspace(workspace, project_id, user):
     return None
 
 
+def validate_projects_in_workspace(workspace, project_ids, user):
+    if not isinstance(project_ids, list) or len(project_ids) == 0:
+        return Response(
+            {
+                "ok": False,
+                "error": "project_ids_required",
+                "message": "Select at least one project for Codex Agent grants.",
+            },
+            status=status.HTTP_400_BAD_REQUEST,
+        )
+
+    normalized_project_ids = []
+    for project_id in project_ids:
+        project_id = str(project_id or "").strip()
+        if not project_id or project_id in normalized_project_ids:
+            continue
+
+        project_error = validate_project_in_workspace(workspace, project_id, user)
+        if project_error is not None:
+            return project_error
+        normalized_project_ids.append(project_id)
+
+    if not normalized_project_ids:
+        return Response(
+            {
+                "ok": False,
+                "error": "project_ids_required",
+                "message": "Select at least one project for Codex Agent grants.",
+            },
+            status=status.HTTP_400_BAD_REQUEST,
+        )
+
+    return normalized_project_ids
+
+
 def gateway_request(method, path, payload=None):
     config, error_response = require_gateway_config()
     if error_response is not None:
@@ -284,6 +319,34 @@ class CodexAgentGrantListEndpoint(CodexAgentEntitledEndpoint):
         return gateway_request("POST", f"/api/internal/v1/owners/{owner_path(request.user)}/agents/{agent_path(agent_id)}/grants", payload)
 
 
+class CodexAgentGrantReplaceEndpoint(CodexAgentEntitledEndpoint):
+    @allow_permission(allowed_roles=[ROLE.ADMIN, ROLE.MEMBER], level="WORKSPACE")
+    def post(self, request, slug, agent_id):
+        entitlement_error = self.require_entitlement(request, slug)
+        if entitlement_error is not None:
+            return entitlement_error
+
+        workspace, workspace_error = require_workspace(slug)
+        if workspace_error is not None:
+            return workspace_error
+
+        project_ids_or_error = validate_projects_in_workspace(workspace, request.data.get("project_ids"), request.user)
+        if isinstance(project_ids_or_error, Response):
+            return project_ids_or_error
+
+        payload = {
+            "workspace_slug": slug,
+            "project_ids": project_ids_or_error,
+            "scopes": request.data.get("scopes") or [],
+            "mode": request.data.get("mode") or "voluntary",
+        }
+        return gateway_request(
+            "POST",
+            f"/api/internal/v1/owners/{owner_path(request.user)}/agents/{agent_path(agent_id)}/grants/replace",
+            payload,
+        )
+
+
 class CodexAgentTokenListEndpoint(CodexAgentEntitledEndpoint):
     @allow_permission(allowed_roles=[ROLE.ADMIN, ROLE.MEMBER], level="WORKSPACE")
     def get(self, request, slug, agent_id):