feat(budpipeline): add deployment actions with internal auth support

## Changes

### BudPipeline Actions
- Add Scale Deployment action to set fixed replica count
- Refactor Deploy Model action with simulation support and smart mode
- Refactor Delete Deployment action with proper cleanup
- Refactor Rate Limit action with multiple algorithm support
- Remove deprecated autoscale action (replaced by scale)
- Remove cluster create/delete actions (not yet implemented)
- Add CREATING_ACTIONS.md documentation guide

### BudApp Internal Authentication
- Add internal auth support to autoscale endpoints for pipeline calls
- Use require_permissions_or_internal decorator pattern
- Add get_current_active_user_or_internal dependency

### BudAdmin UI
- Fix empty endpoint dropdown in pipeline editor
- Add endpoints to dataSources for ActionConfigPanel
- Fetch endpoints when entering edit mode

### ActionContext Enhancement
- Add invoke_service method for Dapr service invocation
- Support dapr-api-token header for internal auth

### Helm Chart
- Add alembic migration to budpipeline startup

### Fixes from Review
- Fix source_topics type consistency (always use list)
- Add force param to delete action
- Update cluster tests for removed actions

Co-Authored-By: Claude Opus 4.5 <[email protected]>

Author: dittops

infra/helm/bud/templates/microservices/budpipeline.yaml

@@ -40,7 +40,7 @@ spec:
       - name: budpipeline
         image: {{ .Values.microservices.budpipeline.image }}
         imagePullPolicy: {{ .Values.imagePullPolicy }}
-        command: ["sh", "-c", "uvicorn budpipeline.main:app --host 0.0.0.0 --port 8010"]
+        command: ["sh", "-c", "alembic upgrade head && uvicorn budpipeline.main:app --host 0.0.0.0 --port 8010"]
         ports:
         - containerPort: 8010
         envFrom:

services/budadmin/src/components/pipelineEditor/components/ActionConfigPanel.tsx

@@ -243,6 +243,7 @@ export interface ActionConfigPanelProps {
     projects?: SelectOption[];
     endpoints?: SelectOption[];
     providers?: SelectOption[];
+    credentials?: SelectOption[];
   };
   /** Whether data sources are loading */
   loadingDataSources?: Set<string>;
@@ -547,6 +548,8 @@ export function ActionConfigPanel({
           return dataSources.endpoints || [];
         case 'provider_ref':
           return dataSources.providers || [];
+        case 'credential_ref':
+          return dataSources.credentials || [];
         default:
           return [];
       }
@@ -664,8 +667,8 @@ export function ActionConfigPanel({
             </select>
           )}
 
-          {/* Ref types (model, cluster, provider, etc.) - with toggle for template mode */}
-          {['model_ref', 'cluster_ref', 'project_ref', 'endpoint_ref', 'provider_ref'].includes(param.type) && (
+          {/* Ref types (model, cluster, provider, credential, etc.) - with toggle for template mode */}
+          {['model_ref', 'cluster_ref', 'project_ref', 'endpoint_ref', 'provider_ref', 'credential_ref'].includes(param.type) && (
             <div>
               <div style={{ display: 'flex', gap: '8px', marginBottom: '6px' }}>
                 <button

services/budadmin/src/flows/components/CommonStatus.tsx

@@ -156,9 +156,7 @@ export default function CommonStatus({
     }, [steps, workflowId]);
 
     useEffect(() => {
-        console.log(`socket`, socket)
         if (socket) {
-            console.log(handleNotification)
             socket.on("notification_received", handleNotification);
         }
 

services/budadmin/src/pages/home/budpipelines/[id]/index.tsx

@@ -26,6 +26,8 @@ import { useCluster } from "src/hooks/useCluster";
 import { useModels } from "src/hooks/useModels";
 import { useCloudProviders } from "src/hooks/useCloudProviders";
 import { useProjects } from "src/hooks/useProjects";
+import { useProprietaryCredentials } from "src/stores/useProprietaryCredentials";
+import { useEndPoints } from "src/hooks/useEndPoint";
 import { PrimaryButton } from "@/components/ui/bud/form/Buttons";
 import { useDrawer } from "src/hooks/useDrawer";
 import StepDetailDrawer from "@/components/pipelineEditor/components/StepDetailDrawer";
@@ -121,6 +123,8 @@ const WorkflowDetail = () => {
   const { models, getGlobalModels } = useModels();
   const { providers, getProviders } = useCloudProviders();
   const { projects, getProjects } = useProjects();
+  const { credentials: proprietaryCredentials, getCredentials: getProprietaryCredentials } = useProprietaryCredentials();
+  const { endPoints, getEndPoints } = useEndPoints();
 
   const [activeTab, setActiveTab] = useState("dag");
   const [selectedStep, setSelectedStep] = useState<PipelineStep | null>(null);
@@ -144,11 +148,13 @@ const WorkflowDetail = () => {
     [executions]
   );
 
-  // Transform clusters, models, projects, and providers to SelectOption format for WorkflowEditor
+  // Transform clusters, models, projects, providers, credentials, and endpoints to SelectOption format for WorkflowEditor
+  // Note: clusters use cluster_id (budcluster UUID) as value, not id (budapp UUID),
+  // to match what deploy-workflow endpoint expects
   const dataSources = useMemo(() => ({
     clusters: clusters.map((c) => ({
       label: c.name,
-      value: c.id,
+      value: c.cluster_id,
     })),
     models: models.map((m) => ({
       label: m.name,
@@ -162,9 +168,17 @@ const WorkflowDetail = () => {
       label: p.name,
       value: p.id,
     })),
-  }), [clusters, models, projects, providers]);
+    credentials: proprietaryCredentials.map((c) => ({
+      label: `${c.name} (${c.type})`,
+      value: c.id,
+    })),
+    endpoints: endPoints.map((e) => ({
+      label: e.name || e.id || "",
+      value: e.id || "",
+    })),
+  }), [clusters, models, projects, providers, proprietaryCredentials, endPoints]);
 
-  // Fetch clusters, models, projects, and providers when entering edit mode
+  // Fetch clusters, models, projects, providers, credentials, and endpoints when entering edit mode
   useEffect(() => {
     if (isEditing) {
       const loadingSet = new Set<string>();
@@ -181,6 +195,12 @@ const WorkflowDetail = () => {
       if (providers.length === 0) {
         loadingSet.add("providers");
       }
+      if (proprietaryCredentials.length === 0) {
+        loadingSet.add("credentials");
+      }
+      if (endPoints.length === 0) {
+        loadingSet.add("endpoints");
+      }
 
       if (loadingSet.size > 0) {
         setLoadingDataSources(loadingSet);
@@ -242,6 +262,33 @@ const WorkflowDetail = () => {
           );
         }
 
+        if (proprietaryCredentials.length === 0) {
+          fetchPromises.push(
+            (async () => {
+              await getProprietaryCredentials({ page: 1, limit: 100 });
+              setLoadingDataSources(prev => {
+                const next = new Set(prev);
+                next.delete("credentials");
+                return next;
+              });
+            })()
+          );
+        }
+
+        if (endPoints.length === 0) {
+          fetchPromises.push(
+            (async () => {
+              // Fetch all endpoints (no project filter needed for pipeline actions)
+              await getEndPoints({ page: 1, limit: 100 });
+              setLoadingDataSources(prev => {
+                const next = new Set(prev);
+                next.delete("endpoints");
+                return next;
+              });
+            })()
+          );
+        }
+
         Promise.all(fetchPromises).finally(() => {
           setLoadingDataSources(new Set());
         });

services/budadmin/src/types/actions.ts

@@ -22,7 +22,8 @@ export type ParamType =
   | 'cluster_ref'
   | 'project_ref'
   | 'endpoint_ref'
-  | 'provider_ref';
+  | 'provider_ref'
+  | 'credential_ref';
 
 export interface SelectOption {
   value: string;

services/budapp/budapp/cluster_ops/schemas.py

@@ -605,6 +605,10 @@ class RecommendedClusterRequest(BaseModel):
         default=None,
         description="Comma-separated model endpoint types (e.g., 'EMBEDDING', 'LLM') for engine selection",
     )
+    debug: bool = Field(
+        default=False,
+        description="If True, run simulation synchronously and wait for completion before returning",
+    )
 
 
 class GrafanaDashboardResponse(SuccessResponse):

services/budapp/budapp/endpoint_ops/endpoint_routes.py

@@ -19,21 +19,22 @@
 from typing import List, Optional, Union
 from uuid import UUID
 
-from fastapi import APIRouter, Depends, Header, Query, status
+from fastapi import APIRouter, Body, Depends, Header, Query, status
 from sqlalchemy.orm import Session
 from typing_extensions import Annotated
 
 from budapp.commons import logging
 from budapp.commons.dependencies import (
     get_current_active_user,
+    get_current_active_user_or_internal,
     get_session,
     parse_ordering_fields,
 )
 from budapp.commons.exceptions import ClientException
 from budapp.user_ops.schemas import User
 
 from ..commons.constants import PermissionEnum
-from ..commons.permission_handler import require_permissions
+from ..commons.permission_handler import require_permissions, require_permissions_or_internal
 from ..commons.schemas import ErrorResponse, SuccessResponse
 from ..workflow_ops.schemas import RetrieveWorkflowDataResponse
 from ..workflow_ops.services import WorkflowService
@@ -46,6 +47,7 @@
     DeleteWorkerRequest,
     DeploymentPricingResponse,
     DeploymentSettingsResponse,
+    EndpointDeleteResponse,
     EndpointFilter,
     EndpointPaginatedResponse,
     ModelClusterDetailResponse,
@@ -146,28 +148,37 @@ async def list_all_endpoints(
             "description": "Invalid request parameters",
         },
         status.HTTP_200_OK: {
-            "model": SuccessResponse,
+            "model": EndpointDeleteResponse,
             "description": "Successfully executed delete endpoint workflow",
         },
     },
     description="Delete an endpoint by ID",
 )
 @require_permissions(permissions=[PermissionEnum.ENDPOINT_MANAGE])
 async def delete_endpoint(
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(get_current_active_user_or_internal)],
     session: Annotated[Session, Depends(get_session)],
     endpoint_id: UUID,
     x_resource_type: Annotated[Optional[str], Header()] = None,
     x_entity_id: Annotated[Optional[str], Header()] = None,
-) -> Union[SuccessResponse, ErrorResponse]:
-    """Delete a endpoint by its ID."""
+    callback_topic: Annotated[Optional[str], Body(embed=True)] = None,
+) -> Union[EndpointDeleteResponse, ErrorResponse]:
+    """Delete a endpoint by its ID.
+
+    Args:
+        callback_topic: Optional Dapr pub/sub topic for budpipeline integration.
+                       If provided, completion events will be published to this topic.
+    """
     try:
-        db_workflow = await EndpointService(session).delete_endpoint(endpoint_id, current_user.id)
+        db_workflow = await EndpointService(session).delete_endpoint(
+            endpoint_id, current_user.id, callback_topic=callback_topic
+        )
         logger.debug(f"Endpoint deleting initiated with workflow id: {db_workflow.id}")
-        return SuccessResponse(
+        return EndpointDeleteResponse(
             message="Deployment deleting initiated successfully",
             code=status.HTTP_200_OK,
             object="endpoint.delete",
+            workflow_id=db_workflow.id,
         )
     except ClientException as e:
         logger.exception(f"Failed to delete endpoint: {e}")
@@ -1018,10 +1029,10 @@ async def update_deployment_settings(
     },
     description="Get autoscale configuration for an endpoint",
 )
-@require_permissions(permissions=[PermissionEnum.ENDPOINT_VIEW])
+@require_permissions_or_internal(permissions=[PermissionEnum.ENDPOINT_VIEW])
 async def get_autoscale_config(
     endpoint_id: UUID,
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(get_current_active_user_or_internal)],
     session: Annotated[Session, Depends(get_session)],
     x_resource_type: Annotated[Optional[str], Header()] = None,
     x_entity_id: Annotated[Optional[str], Header()] = None,
@@ -1065,11 +1076,11 @@ async def get_autoscale_config(
     },
     description="Update autoscale configuration for an endpoint",
 )
-@require_permissions(permissions=[PermissionEnum.ENDPOINT_MANAGE])
+@require_permissions_or_internal(permissions=[PermissionEnum.ENDPOINT_MANAGE])
 async def update_autoscale_config(
     endpoint_id: UUID,
     request: UpdateAutoscaleRequest,
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(get_current_active_user_or_internal)],
     session: Annotated[Session, Depends(get_session)],
     x_resource_type: Annotated[Optional[str], Header()] = None,
     x_entity_id: Annotated[Optional[str], Header()] = None,

services/budapp/budapp/endpoint_ops/schemas.py

@@ -100,6 +100,15 @@ class EndpointPaginatedResponse(PaginatedSuccessResponse):
     endpoints: list[EndpointListResponse] = []
 
 
+class EndpointDeleteResponse(SuccessResponse):
+    """Response schema for endpoint delete operation.
+
+    Includes the workflow_id for tracking the async delete workflow.
+    """
+
+    workflow_id: Optional[UUID] = None
+
+
 class WorkerInfoFilter(BaseModel):
     """Filter for worker info."""
 

services/budapp/budapp/endpoint_ops/services.py

@@ -230,8 +230,17 @@ async def get_all_endpoints(
             project_id, offset, limit, filters, order_by, search, status_filter
         )
 
-    async def delete_endpoint(self, endpoint_id: UUID, current_user_id: UUID) -> WorkflowModel:
-        """Delete an endpoint by its ID."""
+    async def delete_endpoint(
+        self, endpoint_id: UUID, current_user_id: UUID, callback_topic: Optional[str] = None
+    ) -> WorkflowModel:
+        """Delete an endpoint by its ID.
+
+        Args:
+            endpoint_id: The endpoint ID to delete.
+            current_user_id: The ID of the user initiating the delete.
+            callback_topic: Optional Dapr pub/sub topic for budpipeline integration.
+                           If provided, completion events will be published to this topic.
+        """
         db_endpoint = await EndpointDataManager(self.session).retrieve_by_fields(
             EndpointModel, {"id": endpoint_id}, exclude_fields={"status": EndpointStatusEnum.DELETED}
         )
@@ -326,7 +335,11 @@ async def delete_endpoint(self, endpoint_id: UUID, current_user_id: UUID) -> Wor
             # Perform delete endpoint request to bud_cluster app
             if has_cluster:
                 bud_cluster_response = await self._perform_bud_cluster_delete_endpoint_request(
-                    db_endpoint.cluster.cluster_id, db_endpoint.namespace, current_user_id, db_workflow.id
+                    db_endpoint.cluster.cluster_id,
+                    db_endpoint.namespace,
+                    current_user_id,
+                    db_workflow.id,
+                    callback_topic=callback_topic,
                 )
             else:
                 # For cloud models without cluster, skip bud_cluster deletion
@@ -380,13 +393,22 @@ async def delete_endpoint(self, endpoint_id: UUID, current_user_id: UUID) -> Wor
         return db_workflow
 
     async def _perform_bud_cluster_delete_endpoint_request(
-        self, bud_cluster_id: Optional[UUID], namespace: str, current_user_id: UUID, workflow_id: UUID
+        self,
+        bud_cluster_id: Optional[UUID],
+        namespace: str,
+        current_user_id: UUID,
+        workflow_id: UUID,
+        callback_topic: Optional[str] = None,
     ) -> Dict:
         """Perform delete endpoint request to bud_cluster app.
 
         Args:
             bud_cluster_id: The ID of the cluster being served by the endpoint to delete.
             namespace: The namespace of the cluster endpoint to delete.
+            current_user_id: The ID of the user initiating the delete.
+            workflow_id: The workflow ID tracking this deletion.
+            callback_topic: Optional Dapr pub/sub topic for budpipeline integration.
+                           If provided, completion events will also be published to this topic.
         """
         if not bud_cluster_id:
             logger.warning(
@@ -397,6 +419,11 @@ async def _perform_bud_cluster_delete_endpoint_request(
             f"{app_settings.dapr_base_url}/v1.0/invoke/{app_settings.bud_cluster_app_id}/method/deployment/delete"
         )
 
+        # Build source_topic - always use list for consistent type
+        source_topics = [app_settings.source_topic]
+        if callback_topic:
+            source_topics.append(callback_topic)
+
         payload = {
             "cluster_id": str(bud_cluster_id),
             "namespace": namespace,
@@ -405,7 +432,7 @@ async def _perform_bud_cluster_delete_endpoint_request(
                 "subscriber_ids": str(current_user_id),
                 "workflow_id": str(workflow_id),
             },
-            "source_topic": f"{app_settings.source_topic}",
+            "source_topic": source_topics,
         }
 
         logger.debug(f"Performing delete endpoint request to budcluster {payload}")

services/budapp/budapp/model_ops/model_routes.py

@@ -914,9 +914,9 @@ async def get_catalog_model_details(
     },
     description="Retrieve details of a model by ID",
 )
-@require_permissions(permissions=[PermissionEnum.MODEL_VIEW])
+@require_permissions_or_internal(permissions=[PermissionEnum.MODEL_VIEW])
 async def retrieve_model(
-    current_user: Annotated[User, Depends(get_current_active_user)],
+    current_user: Annotated[User, Depends(get_current_active_user_or_internal)],
     session: Annotated[Session, Depends(get_session)],
     model_id: UUID,
 ) -> Union[ModelDetailSuccessResponse, ErrorResponse]:
@@ -1231,6 +1231,8 @@ async def deploy_model_by_step(
             chat_template=deploy_request.chat_template,
             supports_lora=deploy_request.supports_lora,
             supports_pipeline_parallelism=deploy_request.supports_pipeline_parallelism,
+            callback_topic=deploy_request.callback_topic,
+            simulator_id=deploy_request.simulator_id,
         )
 
         return await WorkflowService(session).retrieve_workflow_data(db_workflow.id)