feat: support model deployment in parallel (#166)

* feat: support model deployment in parallel

* feat: update status for parallel execution handling

* fix: update typer version

* fix: code cleanup

Author: yanbasic

pyproject.toml

@@ -19,7 +19,7 @@ exclude = [".venv"]
 
 [tool.poetry.dependencies]
 python = "^3.9"
-typer = {extras = ["all"], version = "0.15.3"}
+typer = "0.15.3"
 click = "8.0.4"
 rich = "14.0.0"
 boto3 = "^1.35.0"

src/emd/cfn/codepipeline/template.yaml

@@ -276,7 +276,7 @@ Resources:
         - Name: CreateTime
         - Name: Region
       PipelineType: V2
-      ExecutionMode: QUEUED
+      ExecutionMode: PARALLEL
       Stages:
         - Name: Source
           Actions:

src/emd/commands/destroy.py

@@ -14,18 +14,42 @@
 layout = make_layout()
 
 
-#@app.callback(invoke_without_command=True)(invoke_without_command=True)
 @app.callback(invoke_without_command=True)
 @catch_aws_credential_errors
 @check_emd_env_exist
 @load_aws_profile
 def destroy(
-    model_id: Annotated[
-        str, typer.Argument(help="Model ID"),
-    ],
-    model_tag: Annotated[
-        str, typer.Argument(help="Model tag")
-    ] = MODEL_DEFAULT_TAG
-    ):
-    # console.print("[bold blue]Checking AWS environment...[/bold blue]")
-    sdk_destroy(model_id,model_tag=model_tag,waiting_until_complete=True)
+    model_identifier: Annotated[
+        str,
+        typer.Argument(
+            help="Model identifier in format 'model_id/model_tag' (e.g., 'Qwen2.5-0.5B-Instruct/d2')"
+        )
+    ]
+):
+    """
+    Destroy a model deployment.
+
+    Examples:
+        emd destroy Qwen2.5-0.5B-Instruct/d2
+        emd destroy Qwen2.5-VL-32B-Instruct/twopath
+        emd destroy DeepSeek-R1-0528-Qwen3-8B/dev
+    """
+    try:
+        console.print(f"[yellow]Destroying model deployment: {model_identifier}[/yellow]")
+
+        # Use the new SDK format
+        sdk_destroy(model_identifier=model_identifier, waiting_until_complete=True)
+
+        console.print(f"[green]✅ Model deployment '{model_identifier}' has been successfully deleted[/green]")
+        console.print("[dim]The model stack and all associated resources have been removed[/dim]")
+
+    except ValueError as e:
+        console.print(f"[red]❌ Invalid format: {e}[/red]")
+        console.print("[yellow]Expected format: 'model_id/model_tag'[/yellow]")
+        console.print("[yellow]Examples:[/yellow]")
+        console.print("  [cyan]emd destroy Qwen2.5-0.5B-Instruct/d2[/cyan]")
+        console.print("  [cyan]emd destroy Qwen2.5-VL-32B-Instruct/twopath[/cyan]")
+        raise typer.Exit(1)
+    except Exception as e:
+        console.print(f"[red]❌ Failed to destroy model deployment: {e}[/red]")
+        raise typer.Exit(1)

src/emd/commands/status.py

@@ -8,6 +8,8 @@
 from emd.utils.logger_utils import make_layout
 from rich.console import Console
 from rich.table import Table
+from rich.spinner import Spinner
+from rich.live import Live
 
 app = typer.Typer(pretty_exceptions_enable=False)
 console = Console()
@@ -26,24 +28,29 @@ def status(
         str, typer.Argument(help="Model tag")
     ] = MODEL_DEFAULT_TAG,
 ):
-    ret = get_model_status(model_id, model_tag=model_tag)
+    # Show loading indicator while fetching model status
+    with console.status("[bold green]Fetching model status...", spinner="dots"):
+        ret = get_model_status(model_id, model_tag=model_tag)
+
     inprogress = ret['inprogress']
     completed = ret['completed']
 
     data = []
+    # Process all in-progress executions (now includes ALL parallel executions)
     for d in inprogress:
         if d['status'] == "Stopped":
             continue
         data.append({
             "model_id": d['model_id'],
             "model_tag": d['model_tag'],
-            "status": f"{d['status']} ({d['stage_name']})",
-            "service_type": d['service_type'],
-            "instance_type": d['instance_type'],
-            "create_time": d['create_time'],
-            "outputs": d['outputs'],
+            "status": f"{d['status']} ({d['stage_name']})" if d.get('stage_name') else d['status'],
+            "service_type": d.get('service_type', ''),
+            "instance_type": d.get('instance_type', ''),
+            "create_time": d.get('create_time', ''),
+            "outputs": d.get('outputs', ''),  # Use .get() to handle missing outputs field
         })
 
+    # Process completed models
     for d in completed:
         data.append({
             "model_id": d['model_id'],
@@ -79,16 +86,14 @@ def status(
     # Display the Models section
     console.print("\nModels", style="bold")
 
-    # Create a custom box style without vertical lines
-
     # Create a single table for all models with normal horizontal lines but no vertical lines
     models_table = Table(show_header=False, expand=True)
 
     # Add two columns for name/value pairs
     models_table.add_column(justify="left", style="cyan", width=22)
     models_table.add_column(justify="left", overflow="fold")
 
-    # Add each model to the table
+    # Add each model to the table (now shows ALL parallel executions)
     for model_data in data:
         # Add model name as a name/value pair with bold styling
         model_name = f"{model_data['model_id']}/{model_data['model_tag']}"

src/emd/sdk/destroy.py

@@ -1,5 +1,6 @@
 import boto3
 import time
+from typing import Union, Tuple
 from emd.utils.logger_utils import get_logger
 from .status import get_destroy_status
 from emd.constants import (
@@ -18,76 +19,207 @@
 from emd.models.utils.constants import ServiceType
 from emd.models import Model
 from emd.utils.aws_service_utils import get_current_region
+
 logger = get_logger(__name__)
 
 
+def parse_model_identifier(model_identifier: str) -> Tuple[str, str]:
+    """
+    Parse model identifier in format 'model_id/model_tag'
+
+    Args:
+        model_identifier: String in format 'model_id/model_tag' or just 'model_id'
+
+    Returns:
+        Tuple of (model_id, model_tag)
+
+    Raises:
+        ValueError: If format is invalid
+
+    Examples:
+        parse_model_identifier('Qwen2.5-0.5B-Instruct/d2') -> ('Qwen2.5-0.5B-Instruct', 'd2')
+        parse_model_identifier('Qwen2.5-0.5B-Instruct') -> ('Qwen2.5-0.5B-Instruct', 'dev')
+    """
+    if not model_identifier or not model_identifier.strip():
+        raise ValueError("Model identifier cannot be empty")
+
+    model_identifier = model_identifier.strip()
+
+    if '/' not in model_identifier:
+        # Backward compatibility: treat as model_id with default tag
+        return model_identifier, MODEL_DEFAULT_TAG
+
+    parts = model_identifier.split('/')
+    if len(parts) != 2:
+        raise ValueError(
+            f"Invalid format: '{model_identifier}'. "
+            f"Expected format: 'model_id/model_tag' (e.g., 'Qwen2.5-0.5B-Instruct/d2')"
+        )
+
+    model_id, model_tag = parts
+    if not model_id.strip():
+        raise ValueError("Model ID cannot be empty")
+    if not model_tag.strip():
+        raise ValueError("Model tag cannot be empty")
+
+    return model_id.strip(), model_tag.strip()
+
+
 def stop_pipeline_execution(
-        model_id:str,
-        model_tag:str,
-        pipeline_name=CODEPIPELINE_NAME,
-        waiting_until_complete=True
+        model_id: str,
+        model_tag: str,
+        pipeline_name: str = CODEPIPELINE_NAME,
+        waiting_until_complete: bool = True
     ):
+    """
+    Stop an active pipeline execution for a model.
+
+    Args:
+        model_id: Model ID
+        model_tag: Model tag
+        pipeline_name: Name of the CodePipeline
+        waiting_until_complete: Whether to wait for the stop to complete
+    """
+    logger.info(f"Checking for active pipeline executions for model: {model_id}, tag: {model_tag}")
+
     active_executuion_infos = get_pipeline_active_executions(
         pipeline_name=pipeline_name
     )
     active_executuion_infos_d = {
-        Model.get_model_stack_name_prefix(d['model_id'],d['model_tag']):d for d in active_executuion_infos
+        Model.get_model_stack_name_prefix(d['model_id'], d['model_tag']): d
+        for d in active_executuion_infos
     }
-    cur_uuid = Model.get_model_stack_name_prefix(model_id,model_tag)
+
+    cur_uuid = Model.get_model_stack_name_prefix(model_id, model_tag)
+    logger.info(f"Looking for pipeline execution with key: {cur_uuid}")
+
     if cur_uuid in active_executuion_infos_d:
-        pipeline_execution_id = active_executuion_infos_d[cur_uuid]['pipeline_execution_id']
+        execution_info = active_executuion_infos_d[cur_uuid]
+        pipeline_execution_id = execution_info['pipeline_execution_id']
+
+        logger.info(f"Found active pipeline execution: {pipeline_execution_id}")
+        logger.info(f"Current status: {execution_info.get('status', 'Unknown')}")
+
         client = boto3.client('codepipeline', region_name=get_current_region())
         try:
             client.stop_pipeline_execution(
                 pipelineName=pipeline_name,
                 pipelineExecutionId=pipeline_execution_id
             )
+            logger.info(f"Stop request sent for pipeline execution: {pipeline_execution_id}")
         except client.exceptions.DuplicatedStopRequestException as e:
-            logger.error(e)
+            logger.warning(f"Stop request already sent for execution {pipeline_execution_id}: {e}")
+        except Exception as e:
+            logger.error(f"Failed to stop pipeline execution {pipeline_execution_id}: {e}")
+            raise
+
         if waiting_until_complete:
+            logger.info("Waiting for pipeline execution to stop...")
             while True:
                 execution_info = get_pipeline_execution_info(
                     pipeline_name=pipeline_name,
                     pipeline_execution_id=pipeline_execution_id,
                 )
-                logger.info(f"pipeline execution status: {execution_info['status']}")
-                if execution_info['status'] == 'Stopped':
+                current_status = execution_info['status']
+                logger.info(f"Pipeline execution status: {current_status}")
+
+                if current_status == 'Stopped':
+                    logger.info("Pipeline execution stopped successfully")
+                    break
+                elif current_status in ['Succeeded', 'Failed', 'Cancelled']:
+                    logger.info(f"Pipeline execution completed with status: {current_status}")
                     break
+
                 time.sleep(5)
     else:
-        logger.error(f"model: {model_id}, model_tag: {model_tag} not found in pipeline executions.")
+        logger.warning(f"No active pipeline execution found for model: {model_id}, tag: {model_tag}")
+        logger.info(f"Available active executions: {list(active_executuion_infos_d.keys())}")
 
 
 def destroy_ecs(model_id,model_tag,stack_name):
     cf_client = boto3.client('cloudformation', region_name=get_current_region())
     cf_client.delete_stack(StackName=stack_name)
 
-def destroy(model_id:str,model_tag=MODEL_DEFAULT_TAG,waiting_until_complete=True):
+def destroy(
+    model_id: Union[str, None] = None,
+    model_tag: str = MODEL_DEFAULT_TAG,
+    model_identifier: Union[str, None] = None,
+    waiting_until_complete: bool = True
+):
+    """
+    Destroy a model deployment.
+
+    Args:
+        model_id: Model ID (legacy format)
+        model_tag: Model tag (legacy format)
+        model_identifier: Model identifier in 'model_id/model_tag' format (new format)
+        waiting_until_complete: Whether to wait for deletion to complete
+
+    Examples:
+        # New format (recommended)
+        destroy(model_identifier='Qwen2.5-0.5B-Instruct/d2')
+
+        # Legacy format (still supported)
+        destroy(model_id='Qwen2.5-0.5B-Instruct', model_tag='d2')
+
+    Raises:
+        ValueError: If neither format is provided or format is invalid
+    """
     check_env_stack_exist_and_complete()
-    stack_name = Model.get_model_stack_name_prefix(model_id,model_tag=model_tag)
+
+    # Handle different input formats
+    if model_identifier is not None:
+        if model_id is not None:
+            raise ValueError("Cannot specify both model_identifier and model_id. Use either the new format (model_identifier='model_id/model_tag') or legacy format (model_id='model_id', model_tag='model_tag')")
+
+        # Parse new format
+        try:
+            model_id, model_tag = parse_model_identifier(model_identifier)
+            logger.info(f"Parsed model identifier '{model_identifier}' -> model_id='{model_id}', model_tag='{model_tag}'")
+        except ValueError as e:
+            logger.error(f"Invalid model identifier format: {e}")
+            raise
+
+    elif model_id is not None:
+        # Legacy format
+        logger.info(f"Using legacy format -> model_id='{model_id}', model_tag='{model_tag}'")
+    else:
+        raise ValueError("Must specify either model_identifier (new format) or model_id (legacy format)")
+
+    stack_name = Model.get_model_stack_name_prefix(model_id, model_tag=model_tag)
+    logger.info(f"Target stack name: {stack_name}")
+
     if not check_stack_exists(stack_name):
-        stop_pipeline_execution(model_id,model_tag,waiting_until_complete=waiting_until_complete)
+        logger.info(f"Stack {stack_name} does not exist, checking for active pipeline executions...")
+        stop_pipeline_execution(model_id, model_tag, waiting_until_complete=waiting_until_complete)
         return
 
     stack_info = get_stack_info(stack_name)
     parameters = stack_info['parameters']
     if parameters['ServiceType'] == ServiceType.ECS:
-        return destroy_ecs(model_id, model_tag,stack_name)
+        logger.info(f"Destroying ECS service for stack: {stack_name}")
+        return destroy_ecs(model_id, model_tag, stack_name)
 
     cf_client = boto3.client('cloudformation', region_name=get_current_region())
     cf_client.delete_stack(StackName=stack_name)
 
-    logger.info(f"Delete stack initiated: {stack_name}")
+    logger.info(f"CloudFormation stack deletion started: {stack_name}")
+    logger.info("Deleting model infrastructure (compute instances, load balancers, security groups, etc.)")
+
     # check delete status
     if waiting_until_complete:
+        logger.info("Waiting for stack deletion to complete...")
         while True:
             status_info = get_destroy_status(stack_name)
             status = status_info['status']
             status_code = status_info['status_code']
             if status_code == 0:
                 break
-            logger.info(f'stack delete status: {status}')
+            logger.info(f'Stack deletion progress: {status}')
             time.sleep(5)
+
         if status == EMD_STACK_NOT_EXISTS_STATUS:
             status = "DELETE_COMPLETED"
-        logger.info(f'stack delete status: {status}')
+            logger.info("✅ Model deployment successfully deleted - all resources have been removed")
+        else:
+            logger.info(f'Stack deletion completed with status: {status}')