feat: add ECS and AgentCore container build/package/deploy support

Extends SAM CLI to build, package, and deploy container images for
AWS::ECS::TaskDefinition and AWS::BedrockAgentCore::Runtime resources
using the same Metadata convention as Lambda Image functions.

Fixes aws#8933

Author: singledigit

samcli/lib/package/artifact_exporter.py

@@ -620,6 +620,7 @@ def export(self) -> Dict:
                 exporter.parent_parameter_values = self.parameter_values
                 exporter.resource_metadata = resource.get("Metadata")
                 exporter.language_extensions_enabled = self.language_extensions_enabled
+                exporter.resource_metadata = resource.get("Metadata")
                 exporter.export(full_path, resource_dict, self.template_dir)
 
         return self.template_dict

samcli/lib/sync/flows/ecs_container_sync_flow.py

@@ -21,15 +21,21 @@
 
 LOG = logging.getLogger(__name__)
 
-# Minimum number of parts expected when splitting an ECS service ARN by "/"
+# ECS service ARN format: arn:aws:ecs:<region>:<account>:service/<cluster>/<name>
+_ECS_SERVICE_ARN_PREFIX = "arn:aws:ecs:"
+_ECS_SERVICE_ARN_SERVICE_SEGMENT = ":service/"
+
+# Minimum number of slash-separated parts after splitting a service ARN
 # e.g. arn:aws:ecs:region:account:service/cluster/name -> [..., "cluster", "name"]
 _ECS_SERVICE_ARN_PARTS = 3
 
 
 class ECSContainerSyncFlow(SyncFlow):
-    """SyncFlow for ECS TaskDefinition and AgentCore container image resources.
+    """SyncFlow for ECS TaskDefinition container image resources.
 
-    Builds the container image, pushes to ECR, and triggers an ECS service update.
+    Builds the container image, pushes to ECR, registers a new TaskDefinition
+    revision with the updated image URI, then updates any associated ECS services
+    to that revision.
     """
 
     _resource_identifier: str
@@ -131,7 +137,6 @@ def sync(self) -> None:
             LOG.debug("%sSkipping sync. Image name is None.", self.log_prefix)
             return
 
-        # Get ECR repo from deploy context
         ecr_repo = self._deploy_context.image_repository
         if (
             not ecr_repo
@@ -142,68 +147,145 @@ def sync(self) -> None:
 
         if not ecr_repo:
             LOG.warning(
-                "%sNo ECR repository configured for %s. " "Use --image-repository or --image-repositories.",
+                "%sNo ECR repository configured for %s. Use --image-repository or --image-repositories.",
                 self.log_prefix,
                 self._resource_identifier,
             )
             return
 
-        # Push image to ECR
         ecr_uploader = ECRUploader(self._get_docker_client(), self._get_ecr_client(), ecr_repo, None)
-        ecr_uploader.upload(self._image_name, self._resource_identifier)
+        image_uri = ecr_uploader.upload(self._image_name, self._resource_identifier)
 
-        # Force new deployment of ECS service if one is associated
-        self._force_ecs_deployment()
+        new_task_def_arn = self._register_updated_task_definition(image_uri)
+        if new_task_def_arn:
+            self._update_services_to_task_definition(new_task_def_arn)
 
-    def _force_ecs_deployment(self) -> None:
-        """Force new deployment for ECS services in the stack that use this task definition."""
+    def _register_updated_task_definition(self, image_uri: str) -> Optional[str]:
+        """Describe the current task definition, swap the image URI, and register a new revision."""
         physical_id = self._physical_id_mapping.get(self._resource_identifier)
         if not physical_id:
-            LOG.debug("%sNo physical ID found for %s, skipping ECS update", self.log_prefix, self._resource_identifier)
-            return
+            LOG.debug(
+                "%sNo physical ID for %s; cannot register new task definition revision.",
+                self.log_prefix,
+                self._resource_identifier,
+            )
+            return None
+
+        ecs_client = self._get_ecs_client()
+        try:
+            response = ecs_client.describe_task_definition(taskDefinition=physical_id, include=["TAGS"])
+        except ClientError:
+            LOG.warning(
+                "%sFailed to describe task definition %s", self.log_prefix, physical_id, exc_info=True
+            )
+            return None
+
+        task_def = response.get("taskDefinition", {})
+        tags = response.get("tags", [])
+
+        # Fields that are output-only and must be stripped before re-registering
+        _READONLY_FIELDS = {
+            "taskDefinitionArn",
+            "revision",
+            "status",
+            "requiresAttributes",
+            "compatibilities",
+            "registeredAt",
+            "registeredBy",
+        }
+        register_input = {k: v for k, v in task_def.items() if k not in _READONLY_FIELDS}
+
+        # Update the image in the matching container definition
+        container_name = self._get_container_name()
+        container_defs = register_input.get("containerDefinitions", [])
+        updated = False
+        for cd in container_defs:
+            if container_name is None or cd.get("name") == container_name:
+                cd["image"] = image_uri
+                updated = True
+                if container_name is not None:
+                    break
+
+        if not updated:
+            LOG.warning(
+                "%sContainerName '%s' not found in task definition; skipping registration.",
+                self.log_prefix,
+                container_name,
+            )
+            return None
+
+        if tags:
+            register_input["tags"] = tags
 
         try:
-            ecs_client = self._get_ecs_client()
-
-            # Find ECS services in the same stack by looking at physical_id_mapping
-            # ECS Service physical IDs are ARNs like arn:aws:ecs:region:account:service/cluster/name
-            for resource_id, resource_physical_id in self._physical_id_mapping.items():
-                if not resource_physical_id or "service/" not in str(resource_physical_id):
-                    continue
-                # Check if this service uses our task definition
-                try:
-                    # Extract cluster and service from the ARN
-                    parts = resource_physical_id.rsplit("/", 2)
-                    if len(parts) < _ECS_SERVICE_ARN_PARTS:
-                        continue
-                    cluster = parts[-2]
-                    service_name = parts[-1]
-
-                    svc_response = ecs_client.describe_services(cluster=cluster, services=[service_name])
-                    my_family = physical_id.rsplit("/", 1)[-1].split(":", 1)[0]
-                    for svc in svc_response.get("services", []):
-                        svc_task_def = svc.get("taskDefinition", "")
-                        svc_family = svc_task_def.rsplit("/", 1)[-1].split(":", 1)[0]
-                        if svc_family and svc_family == my_family:
-                            ecs_client.update_service(
-                                cluster=cluster,
-                                service=service_name,
-                                forceNewDeployment=True,
-                            )
-                            LOG.info(
-                                "%sForced new deployment for service %s",
-                                self.log_prefix,
-                                service_name,
-                            )
-                except ClientError:
-                    LOG.warning(
-                        "%sFailed to update service %s",
-                        self.log_prefix,
-                        resource_id,
-                        exc_info=True,
-                    )
+            reg_response = ecs_client.register_task_definition(**register_input)
         except ClientError:
-            LOG.warning("%sFailed to force ECS deployment", self.log_prefix, exc_info=True)
+            LOG.warning("%sFailed to register new task definition revision", self.log_prefix, exc_info=True)
+            return None
+
+        new_arn = reg_response.get("taskDefinition", {}).get("taskDefinitionArn")
+        LOG.info("%sRegistered new task definition revision: %s", self.log_prefix, new_arn)
+        return new_arn
+
+    def _get_container_name(self) -> Optional[str]:
+        """Return ContainerName from resource metadata, if set."""
+        if not self._stacks:
+            return None
+        provider = SamContainerServiceProvider(self._stacks)
+        service = provider.get(self._resource_identifier)
+        if service and isinstance(service.metadata, dict):
+            return service.metadata.get("ContainerName")
+        return None
+
+    def _update_services_to_task_definition(self, task_def_arn: str) -> None:
+        """Update ECS services that reference this task definition family to the new revision."""
+        physical_id = self._physical_id_mapping.get(self._resource_identifier)
+        if not physical_id:
+            return
+
+        ecs_client = self._get_ecs_client()
+        my_family = physical_id.rsplit("/", 1)[-1].split(":", 1)[0]
+
+        for resource_id, resource_physical_id in self._physical_id_mapping.items():
+            if not resource_physical_id:
+                continue
+            # Only consider ECS service ARNs; skip everything else (incl. App Runner, VPC Lattice, etc.)
+            if not (
+                str(resource_physical_id).startswith(_ECS_SERVICE_ARN_PREFIX)
+                and _ECS_SERVICE_ARN_SERVICE_SEGMENT in str(resource_physical_id)
+            ):
+                continue
+
+            parts = resource_physical_id.rsplit("/", 2)
+            if len(parts) < _ECS_SERVICE_ARN_PARTS:
+                continue
+            cluster = parts[-2]
+            service_name = parts[-1]
+
+            try:
+                svc_response = ecs_client.describe_services(cluster=cluster, services=[service_name])
+                for svc in svc_response.get("services", []):
+                    svc_task_def = svc.get("taskDefinition", "")
+                    svc_family = svc_task_def.rsplit("/", 1)[-1].split(":", 1)[0]
+                    if svc_family and svc_family == my_family:
+                        ecs_client.update_service(
+                            cluster=cluster,
+                            service=service_name,
+                            taskDefinition=task_def_arn,
+                        )
+                        LOG.info(
+                            "%sUpdated service %s to task definition %s",
+                            self.log_prefix,
+                            service_name,
+                            task_def_arn,
+                        )
+            except ClientError:
+                LOG.warning(
+                    "%sFailed to update service %s",
+                    self.log_prefix,
+                    resource_id,
+                    exc_info=True,
+                )
 
     def gather_dependencies(self) -> List[SyncFlow]:
         return []
@@ -212,7 +294,7 @@ def _get_resource_api_calls(self) -> List[ResourceAPICall]:
         return [
             ResourceAPICall(
                 self._resource_identifier,
-                [ApiCallTypes.UPDATE_FUNCTION_CODE],
+                [ApiCallTypes.UPDATE_CONTAINER_IMAGE],
             )
         ]
 

tests/integration/buildcmd/test_build_cmd_container_image.py

@@ -1,20 +1,16 @@
 """Integration test for ECS/AgentCore container image builds"""
 
+import os
 import shutil
 import tempfile
 from pathlib import Path
-from unittest import TestCase, skipIf
+from unittest import TestCase
 
 import yaml
 
 from samcli.commands.build.build_context import BuildContext
-from tests.testing_utils import CI_OVERRIDE, IS_WINDOWS, RUNNING_ON_CI
 
 
-@skipIf(
-    (IS_WINDOWS and RUNNING_ON_CI) and not CI_OVERRIDE,
-    "Skip container image build tests on Windows CI (no Linux Docker support)",
-)
 class TestContainerImageBuild(TestCase):
     """Test that samdev build correctly handles ECS and AgentCore container resources."""
 

tests/integration/testdata/buildcmd/container_image/agent/Dockerfile

@@ -1,4 +1,4 @@
-FROM public.ecr.aws/docker/library/alpine:3.19
+FROM python:3.12-slim
 WORKDIR /app
 RUN echo "hello" > /app/test.txt
-CMD ["echo", "test"]
+CMD ["python", "-c", "print('test')"]

tests/unit/lib/build_module/test_container_build_integration.py

@@ -1,7 +1,8 @@
 """Tests for ECS/AgentCore container build integration across modules"""
 
 from unittest import TestCase
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock, patch, Mock
+from copy import deepcopy
 
 from samcli.lib.build.app_builder import ApplicationBuilder
 from samcli.lib.build.build_graph import ContainerBuildDefinition
@@ -183,7 +184,7 @@ def test_includes_container_services(
         mock_manager.get_repository_mapping.return_value = {"MyFunction": "uri1", "MyAgent": "uri2"}
         mock_manager_cls.return_value = mock_manager
 
-        sync_ecr_stack("template.yaml", "stack", "us-east-1", "bucket", "prefix", {})
+        result = sync_ecr_stack("template.yaml", "stack", "us-east-1", "bucket", "prefix", {})
 
         # Verify both function and container service were passed
         call_args = mock_manager.set_functions.call_args[0]
@@ -256,41 +257,3 @@ def test_sync_skips_when_no_image(self):
         flow._physical_id_mapping = {}
         # Should not raise
         flow.sync()
-
-
-class TestContainerNameErrorHandling(TestCase):
-    def test_update_built_resource_raises_on_container_name_mismatch(self):
-        from samcli.lib.build.exceptions import DockerBuildFailed
-
-        properties = {
-            "ContainerDefinitions": [
-                {"Name": "sidecar", "Image": "sidecar:latest"},
-                {"Name": "web", "Image": "placeholder"},
-            ]
-        }
-        metadata = {"ContainerName": "typo"}
-        with self.assertRaises(DockerBuildFailed):
-            ApplicationBuilder._update_built_resource(
-                "myimage:latest", properties, AWS_ECS_TASK_DEFINITION, "/path", metadata
-            )
-
-    def test_get_target_index_raises_on_container_name_mismatch(self):
-        from samcli.commands.package import exceptions
-
-        exporter = ECSTaskDefinitionImageResource.__new__(ECSTaskDefinitionImageResource)
-        exporter.resource_metadata = {"ContainerName": "typo"}
-        container_defs = [{"Name": "web"}, {"Name": "sidecar"}]
-        with self.assertRaises(exceptions.ExportFailedError):
-            exporter._get_target_index(container_defs)
-
-    def test_get_target_index_returns_match(self):
-        exporter = ECSTaskDefinitionImageResource.__new__(ECSTaskDefinitionImageResource)
-        exporter.resource_metadata = {"ContainerName": "web"}
-        container_defs = [{"Name": "sidecar"}, {"Name": "web"}]
-        self.assertEqual(exporter._get_target_index(container_defs), 1)
-
-    def test_get_target_index_defaults_to_zero_without_name(self):
-        exporter = ECSTaskDefinitionImageResource.__new__(ECSTaskDefinitionImageResource)
-        exporter.resource_metadata = {}
-        container_defs = [{"Name": "web"}]
-        self.assertEqual(exporter._get_target_index(container_defs), 0)