diff --git a/agent/api/container/container_test.go b/agent/api/container/container_test.go
index 1fd96d6a6ff..8ee4f036cbc 100644
--- a/agent/api/container/container_test.go
+++ b/agent/api/container/container_test.go
@@ -96,7 +96,10 @@ func TestIsKnownSteadyState(t *testing.T) {
 func TestGetNextStateProgression(t *testing.T) {
 	// This creates a container with `iota` ContainerStatus (NONE)
 	container := &Container{}
-	// NONE should transition to PULLED
+	// NONE should transition to MANIFEST_PULLED
+	assert.Equal(t, container.GetNextKnownStateProgression(), apicontainerstatus.ContainerManifestPulled)
+	container.SetKnownStatus(apicontainerstatus.ContainerManifestPulled)
+	// MANIFEST_PULLED should transition to PULLED
 	assert.Equal(t, container.GetNextKnownStateProgression(), apicontainerstatus.ContainerPulled)
 	container.SetKnownStatus(apicontainerstatus.ContainerPulled)
 	// PULLED should transition to CREATED
diff --git a/agent/api/task/task.go b/agent/api/task/task.go
index dc2f4dfda55..fc9b09e210b 100644
--- a/agent/api/task/task.go
+++ b/agent/api/task/task.go
@@ -1052,7 +1052,7 @@ func (task *Task) initializeASMAuthResource(credentialsManager credentials.Manag
 		if container.ShouldPullWithASMAuth() {
 			container.BuildResourceDependency(asmAuthResource.GetName(),
 				resourcestatus.ResourceStatus(asmauth.ASMAuthStatusCreated),
-				apicontainerstatus.ContainerPulled)
+				apicontainerstatus.ContainerManifestPulled)
 		}
 	}
 }
diff --git a/agent/api/task/task_test.go b/agent/api/task/task_test.go
index fa7eacbec79..68073928060 100644
--- a/agent/api/task/task_test.go
+++ b/agent/api/task/task_test.go
@@ -2842,8 +2842,18 @@ func TestPostUnmarshalTaskASMDockerAuth(t *testing.T) {
 		},
 	}
 
+	resourceDep := apicontainer.ResourceDependency{
+		Name:           asmauth.ResourceName,
+		RequiredStatus: resourcestatus.ResourceStatus(asmauth.ASMAuthStatusCreated),
+	}
+
 	err := task.PostUnmarshalTask(cfg, credentialsManager, resFields, nil, nil)
-	assert.NoError(t, err)
+	require.NoError(t, err)
+	assert.Equal(t,
+		resourceDep,
+		task.Containers[0].
+			TransitionDependenciesMap[apicontainerstatus.ContainerManifestPulled].
+			ResourceDependencies[0])
 }
 
 func TestPostUnmarshalTaskSecret(t *testing.T) {
diff --git a/agent/engine/dependencygraph/graph.go b/agent/engine/dependencygraph/graph.go
index 88515b23a29..bb33b9b60c1 100644
--- a/agent/engine/dependencygraph/graph.go
+++ b/agent/engine/dependencygraph/graph.go
@@ -415,6 +415,12 @@ func containerOrderingDependenciesIsResolved(target *apicontainer.Container,
 	targetContainerKnownStatus := target.GetKnownStatus()
 	dependsOnContainerKnownStatus := dependsOnContainer.GetKnownStatus()
 
+	// Containers are allowed to transition to MANIFEST_PULLED irrespective of any container
+	// ordering dependencies.
+	if targetContainerKnownStatus < apicontainerstatus.ContainerManifestPulled {
+		return true
+	}
+
 	// The 'target' container desires to be moved to 'Created' or the 'steady' state.
 	// Allow this only if the environment variable ECS_PULL_DEPENDENT_CONTAINERS_UPFRONT is enabled and
 	// known status of the `target` container state has not reached to 'Pulled' state;
diff --git a/agent/engine/dependencygraph/graph_test.go b/agent/engine/dependencygraph/graph_test.go
index 94fa4a7aea6..4563202de36 100644
--- a/agent/engine/dependencygraph/graph_test.go
+++ b/agent/engine/dependencygraph/graph_test.go
@@ -28,9 +28,12 @@ import (
 	mock_taskresource "github.com/aws/amazon-ecs-agent/agent/taskresource/mocks"
 	resourcestatus "github.com/aws/amazon-ecs-agent/agent/taskresource/status"
 	apicontainerstatus "github.com/aws/amazon-ecs-agent/ecs-agent/api/container/status"
+	"github.com/aws/amazon-ecs-agent/ecs-agent/credentials"
+	mock_credentials "github.com/aws/amazon-ecs-agent/ecs-agent/credentials/mocks"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func volumeStrToVol(vols []string) []apicontainer.VolumeFrom {
@@ -49,6 +52,16 @@ func steadyStateContainer(name string, dependsOn []apicontainer.DependsOn, desir
 	return container
 }
 
+// Creates a container with supplied properties and MANIFEST_PULLED as its known state.
+func steadyStateManifestPulledContainer(name string, dependsOn []apicontainer.DependsOn, desiredState apicontainerstatus.ContainerStatus, steadyState apicontainerstatus.ContainerStatus) *apicontainer.Container {
+	container := apicontainer.NewContainerWithSteadyState(steadyState)
+	container.Name = name
+	container.DependsOnUnsafe = dependsOn
+	container.DesiredStatusUnsafe = desiredState
+	container.KnownStatusUnsafe = apicontainerstatus.ContainerManifestPulled
+	return container
+}
+
 func createdContainer(name string, dependsOn []apicontainer.DependsOn, steadyState apicontainerstatus.ContainerStatus) *apicontainer.Container {
 	container := apicontainer.NewContainerWithSteadyState(steadyState)
 	container.Name = name
@@ -132,11 +145,11 @@ func TestDependenciesAreResolvedWhenSteadyStateIsRunning(t *testing.T) {
 	assert.NoError(t, err, "One container should resolve trivially")
 
 	// Webserver stack
-	php := steadyStateContainer("php", []apicontainer.DependsOn{{ContainerName: "db", Condition: startCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
-	db := steadyStateContainer("db", []apicontainer.DependsOn{{ContainerName: "dbdatavolume", Condition: createCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
+	php := steadyStateManifestPulledContainer("php", []apicontainer.DependsOn{{ContainerName: "db", Condition: startCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
+	db := steadyStateManifestPulledContainer("db", []apicontainer.DependsOn{{ContainerName: "dbdatavolume", Condition: createCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
 	dbdata := createdContainer("dbdatavolume", []apicontainer.DependsOn{}, apicontainerstatus.ContainerRunning)
-	webserver := steadyStateContainer("webserver", []apicontainer.DependsOn{{ContainerName: "php", Condition: startCondition}, {ContainerName: "htmldata", Condition: createCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
-	htmldata := steadyStateContainer("htmldata", []apicontainer.DependsOn{{ContainerName: "sharedcssfiles", Condition: createCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
+	webserver := steadyStateManifestPulledContainer("webserver", []apicontainer.DependsOn{{ContainerName: "php", Condition: startCondition}, {ContainerName: "htmldata", Condition: createCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
+	htmldata := steadyStateManifestPulledContainer("htmldata", []apicontainer.DependsOn{{ContainerName: "sharedcssfiles", Condition: createCondition}}, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning)
 	sharedcssfiles := createdContainer("sharedcssfiles", []apicontainer.DependsOn{}, apicontainerstatus.ContainerRunning)
 
 	task = &apitask.Task{
@@ -170,6 +183,106 @@ func TestDependenciesAreResolvedWhenSteadyStateIsRunning(t *testing.T) {
 	assert.NoError(t, err, "Php should resolve")
 }
 
+// Tests container dependency on execution role credentials.
+func TestExecutionRoleAuthDependencyResolution(t *testing.T) {
+	tcs := []struct {
+		name                     string
+		targetContainer          *apicontainer.Container
+		id                       string
+		cfg                      *config.Config
+		credsManagerExpectations func(*mock_credentials.MockManager)
+		expectedError            error
+	}{
+		{
+			name: "container does not depend on execution role",
+			targetContainer: &apicontainer.Container{
+				KnownStatusUnsafe: apicontainerstatus.ContainerStatusNone,
+			},
+		},
+		{
+			name: "execution role dependent container is before manifest pulled state and creds are not resolved yet",
+			targetContainer: &apicontainer.Container{
+				KnownStatusUnsafe: apicontainerstatus.ContainerStatusNone,
+				RegistryAuthentication: &apicontainer.RegistryAuthenticationData{
+					Type:        apicontainer.AuthTypeECR,
+					ECRAuthData: &apicontainer.ECRAuthData{UseExecutionRole: true},
+				},
+			},
+			id: "id",
+			credsManagerExpectations: func(mm *mock_credentials.MockManager) {
+				mm.EXPECT().
+					GetTaskCredentials("id").
+					Return(credentials.TaskIAMRoleCredentials{}, false)
+			},
+			expectedError: CredentialsNotResolvedErr,
+		},
+		{
+			name: "execution role dependent container is at manifest pulled state and creds are not resolved yet",
+			targetContainer: &apicontainer.Container{
+				KnownStatusUnsafe: apicontainerstatus.ContainerManifestPulled,
+				RegistryAuthentication: &apicontainer.RegistryAuthenticationData{
+					Type:        apicontainer.AuthTypeECR,
+					ECRAuthData: &apicontainer.ECRAuthData{UseExecutionRole: true},
+				},
+			},
+			id: "id",
+			credsManagerExpectations: func(mm *mock_credentials.MockManager) {
+				mm.EXPECT().
+					GetTaskCredentials("id").
+					Return(credentials.TaskIAMRoleCredentials{}, false)
+			},
+			expectedError: CredentialsNotResolvedErr,
+		},
+		{
+			name: "execution role dependent container is at pulled state so creds not required anymore",
+			targetContainer: &apicontainer.Container{
+				KnownStatusUnsafe: apicontainerstatus.ContainerPulled,
+				RegistryAuthentication: &apicontainer.RegistryAuthenticationData{
+					Type:        apicontainer.AuthTypeECR,
+					ECRAuthData: &apicontainer.ECRAuthData{UseExecutionRole: true},
+				},
+			},
+		},
+		{
+			name: "execution role dependent container is before manifest pulled state and creds are resolved",
+			targetContainer: &apicontainer.Container{
+				KnownStatusUnsafe: apicontainerstatus.ContainerStatusNone,
+				RegistryAuthentication: &apicontainer.RegistryAuthenticationData{
+					Type:        apicontainer.AuthTypeECR,
+					ECRAuthData: &apicontainer.ECRAuthData{UseExecutionRole: true},
+				},
+			},
+			id: "id",
+			credsManagerExpectations: func(mm *mock_credentials.MockManager) {
+				mm.EXPECT().
+					GetTaskCredentials("id").
+					Return(credentials.TaskIAMRoleCredentials{}, true)
+			},
+			expectedError: nil,
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			ctrl := gomock.NewController(t)
+			defer ctrl.Finish()
+
+			credsManager := mock_credentials.NewMockManager(ctrl)
+			if tc.credsManagerExpectations != nil {
+				tc.credsManagerExpectations(credsManager)
+			}
+
+			dep, err := DependenciesAreResolved(
+				tc.targetContainer, nil, tc.id, credsManager, nil, tc.cfg)
+			if tc.expectedError == nil {
+				require.NoError(t, err)
+				assert.Nil(t, dep) // no dependency is returned for execution role
+			} else {
+				require.Equal(t, tc.expectedError, err)
+			}
+		})
+	}
+}
+
 func TestRunDependencies(t *testing.T) {
 	c1 := &apicontainer.Container{
 		Name:              "a",
@@ -313,20 +426,39 @@ func TestVerifyTransitionDependenciesResolved(t *testing.T) {
 		ResolvedErr     error
 	}{
 		{
-			Name:            "Nothing running, pull depends on running",
+			Name:            "Nothing running, manifest_pull depends on running",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
 			TargetDesired:   apicontainerstatus.ContainerRunning,
-			TargetNext:      apicontainerstatus.ContainerPulled,
+			TargetNext:      apicontainerstatus.ContainerManifestPulled,
 			DependencyName:  "container",
 			DependencyKnown: apicontainerstatus.ContainerStatusNone,
 			SatisfiedStatus: apicontainerstatus.ContainerRunning,
 			ResolvedErr:     ErrContainerDependencyNotResolved,
 		},
-
 		{
-			Name:            "Nothing running, pull depends on resources provisioned",
+			Name:            "Nothing running, manifest_pull depends on resources provisioned",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
 			TargetDesired:   apicontainerstatus.ContainerRunning,
+			TargetNext:      apicontainerstatus.ContainerManifestPulled,
+			DependencyName:  "container",
+			DependencyKnown: apicontainerstatus.ContainerStatusNone,
+			SatisfiedStatus: apicontainerstatus.ContainerResourcesProvisioned,
+			ResolvedErr:     ErrContainerDependencyNotResolved,
+		},
+		{
+			Name:            "Dependency is at None, current is manifest_pulled, pull depends on running",
+			TargetKnown:     apicontainerstatus.ContainerManifestPulled,
+			TargetDesired:   apicontainerstatus.ContainerRunning,
+			TargetNext:      apicontainerstatus.ContainerPulled,
+			DependencyName:  "container",
+			DependencyKnown: apicontainerstatus.ContainerStatusNone,
+			SatisfiedStatus: apicontainerstatus.ContainerRunning,
+			ResolvedErr:     ErrContainerDependencyNotResolved,
+		},
+		{
+			Name:            "Dependency is at None, current is manifest_pulled, pull depends on resources provisioned",
+			TargetKnown:     apicontainerstatus.ContainerManifestPulled,
+			TargetDesired:   apicontainerstatus.ContainerRunning,
 			TargetNext:      apicontainerstatus.ContainerPulled,
 			DependencyName:  "container",
 			DependencyKnown: apicontainerstatus.ContainerStatusNone,
@@ -343,49 +475,58 @@ func TestVerifyTransitionDependenciesResolved(t *testing.T) {
 			SatisfiedStatus: apicontainerstatus.ContainerRunning,
 		},
 		{
-			Name:            "Dependency created, pull depends on running",
+			Name:            "Dependency created, manifest_pull depends on running",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
 			TargetDesired:   apicontainerstatus.ContainerRunning,
-			TargetNext:      apicontainerstatus.ContainerPulled,
+			TargetNext:      apicontainerstatus.ContainerManifestPulled,
 			DependencyName:  "container",
 			DependencyKnown: apicontainerstatus.ContainerCreated,
 			SatisfiedStatus: apicontainerstatus.ContainerRunning,
 			ResolvedErr:     ErrContainerDependencyNotResolved,
 		},
 		{
-			Name:            "Dependency created, pull depends on resources provisioned",
+			Name:            "Dependency created, manifest_pull depends on resources provisioned",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
 			TargetDesired:   apicontainerstatus.ContainerRunning,
-			TargetNext:      apicontainerstatus.ContainerPulled,
+			TargetNext:      apicontainerstatus.ContainerManifestPulled,
 			DependencyName:  "container",
 			DependencyKnown: apicontainerstatus.ContainerCreated,
 			SatisfiedStatus: apicontainerstatus.ContainerResourcesProvisioned,
 			ResolvedErr:     ErrContainerDependencyNotResolved,
 		},
 		{
-			Name:            "Dependency running, pull depends on running",
+			Name:            "Dependency running, manifest_pull depends on running",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
 			TargetDesired:   apicontainerstatus.ContainerRunning,
+			TargetNext:      apicontainerstatus.ContainerManifestPulled,
+			DependencyName:  "container",
+			DependencyKnown: apicontainerstatus.ContainerRunning,
+			SatisfiedStatus: apicontainerstatus.ContainerRunning,
+		},
+		{
+			Name:            "Dependency running, current is manifest_pulled, pull depends on running",
+			TargetKnown:     apicontainerstatus.ContainerManifestPulled,
+			TargetDesired:   apicontainerstatus.ContainerRunning,
 			TargetNext:      apicontainerstatus.ContainerPulled,
 			DependencyName:  "container",
 			DependencyKnown: apicontainerstatus.ContainerRunning,
 			SatisfiedStatus: apicontainerstatus.ContainerRunning,
 		},
 		{
-			Name:            "Dependency running, pull depends on resources provisioned",
+			Name:            "Dependency running, manifest_pull depends on resources provisioned",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
 			TargetDesired:   apicontainerstatus.ContainerRunning,
-			TargetNext:      apicontainerstatus.ContainerPulled,
+			TargetNext:      apicontainerstatus.ContainerManifestPulled,
 			DependencyName:  "container",
 			DependencyKnown: apicontainerstatus.ContainerRunning,
 			SatisfiedStatus: apicontainerstatus.ContainerResourcesProvisioned,
 			ResolvedErr:     ErrContainerDependencyNotResolved,
 		},
 		{
-			Name:            "Dependency resources provisioned, pull depends on resources provisioned",
+			Name:            "Dependency resources provisioned, manifest_pull depends on resources provisioned",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
 			TargetDesired:   apicontainerstatus.ContainerRunning,
-			TargetNext:      apicontainerstatus.ContainerPulled,
+			TargetNext:      apicontainerstatus.ContainerManifestPulled,
 			DependencyName:  "container",
 			DependencyKnown: apicontainerstatus.ContainerResourcesProvisioned,
 			SatisfiedStatus: apicontainerstatus.ContainerResourcesProvisioned,
@@ -452,13 +593,29 @@ func TestVerifyResourceDependenciesResolved(t *testing.T) {
 		ExpectedResolved bool
 	}{
 		{
-			Name:             "resource none,container pull depends on resource created",
+			Name:             "resource none,container manifest_pull depends on resource created",
 			TargetKnown:      apicontainerstatus.ContainerStatusNone,
+			TargetDep:        apicontainerstatus.ContainerManifestPulled,
+			DependencyKnown:  resourcestatus.ResourceStatus(0),
+			RequiredStatus:   resourcestatus.ResourceStatus(1),
+			ExpectedResolved: false,
+		},
+		{
+			Name:             "resource none,container pull depends on resource created",
+			TargetKnown:      apicontainerstatus.ContainerManifestPulled,
 			TargetDep:        apicontainerstatus.ContainerPulled,
 			DependencyKnown:  resourcestatus.ResourceStatus(0),
 			RequiredStatus:   resourcestatus.ResourceStatus(1),
 			ExpectedResolved: false,
 		},
+		{
+			Name:             "resource created,container manifest_pull depends on resource created",
+			TargetKnown:      apicontainerstatus.ContainerStatusNone,
+			TargetDep:        apicontainerstatus.ContainerManifestPulled,
+			DependencyKnown:  resourcestatus.ResourceStatus(1),
+			RequiredStatus:   resourcestatus.ResourceStatus(1),
+			ExpectedResolved: true,
+		},
 		{
 			Name:             "resource created,container pull depends on resource created",
 			TargetKnown:      apicontainerstatus.ContainerStatusNone,
@@ -510,13 +667,28 @@ func TestVerifyTransitionResourceDependenciesResolved(t *testing.T) {
 		ResolvedErr     error
 	}{
 		{
-			Name:            "resource none,container pull depends on resource created",
+			Name:            "resource none,container manifest_pull depends on resource created",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
+			TargetDep:       apicontainerstatus.ContainerManifestPulled,
+			DependencyKnown: resourcestatus.ResourceStatus(0),
+			RequiredStatus:  resourcestatus.ResourceStatus(1),
+			ResolvedErr:     ErrResourceDependencyNotResolved,
+		},
+		{
+			Name:            "resource none,container pull depends on resource created",
+			TargetKnown:     apicontainerstatus.ContainerManifestPulled,
 			TargetDep:       apicontainerstatus.ContainerPulled,
 			DependencyKnown: resourcestatus.ResourceStatus(0),
 			RequiredStatus:  resourcestatus.ResourceStatus(1),
 			ResolvedErr:     ErrResourceDependencyNotResolved,
 		},
+		{
+			Name:            "resource created, manifest_pull depends on resource created",
+			TargetKnown:     apicontainerstatus.ContainerStatusNone,
+			TargetDep:       apicontainerstatus.ContainerManifestPulled,
+			DependencyKnown: resourcestatus.ResourceStatus(1),
+			RequiredStatus:  resourcestatus.ResourceStatus(1),
+		},
 		{
 			Name:            "resource created,container pull depends on resource created",
 			TargetKnown:     apicontainerstatus.ContainerStatusNone,
@@ -570,7 +742,7 @@ func TestTransitionDependencyResourceNotFound(t *testing.T) {
 		KnownStatusUnsafe:         apicontainerstatus.ContainerStatusNone,
 		TransitionDependenciesMap: make(map[apicontainerstatus.ContainerStatus]apicontainer.TransitionDependencySet),
 	}
-	target.BuildResourceDependency("resource", resourcestatus.ResourceStatus(1), apicontainerstatus.ContainerPulled)
+	target.BuildResourceDependency("resource", resourcestatus.ResourceStatus(1), apicontainerstatus.ContainerManifestPulled)
 
 	resources := make(map[string]taskresource.TaskResource)
 	resources["resource1"] = mockResource // different resource name
@@ -1120,6 +1292,47 @@ func TestContainerOrderingIsResolvedWithDependentContainersPullUpfront(t *testin
 	}
 }
 
+// Tests that a dependent container is always able to transition to MANIFEST_PULLED state
+// regardless of any container dependencies.
+func TestContainerOrderingDependenciesIsResolvedManifetPulled(t *testing.T) {
+	dependencyKnownStatuses := []apicontainerstatus.ContainerStatus{
+		apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerManifestPulled,
+		apicontainerstatus.ContainerPulled, apicontainerstatus.ContainerCreated,
+		apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerResourcesProvisioned,
+		apicontainerstatus.ContainerStopped,
+	}
+	pullUpfrontSettings := []config.BooleanDefaultFalse{
+		{Value: config.ExplicitlyEnabled},
+		{Value: config.ExplicitlyDisabled},
+		{Value: config.NotSet},
+	}
+	dependencyConditions := []string{
+		createCondition, startCondition, successCondition, completeCondition, healthyCondition,
+	}
+	for _, dependencyKnownStatus := range dependencyKnownStatuses {
+		for _, pullUpfrontSetting := range pullUpfrontSettings {
+			for _, dependencyCondition := range dependencyConditions {
+				t.Run(
+					fmt.Sprintf("%v - %v - %v",
+						dependencyKnownStatus.String(), pullUpfrontSetting, dependencyCondition,
+					),
+					func(t *testing.T) {
+						target := &apicontainer.Container{
+							KnownStatusUnsafe: apicontainerstatus.ContainerStatusNone,
+						}
+						dep := &apicontainer.Container{
+							KnownStatusUnsafe: dependencyKnownStatus,
+						}
+						cfg := &config.Config{DependentContainersPullUpfront: pullUpfrontSetting}
+						isResolved := containerOrderingDependenciesIsResolved(
+							target, dep, dependencyCondition, cfg)
+						assert.True(t, isResolved)
+					})
+			}
+		}
+	}
+}
+
 func assertContainerTargetOrderingResolved(f func(target *apicontainer.Container, dep *apicontainer.Container, depCond string, cfg *config.Config) bool, targetKnown, depKnown apicontainerstatus.ContainerStatus, depCond string, exitcode int, expectedResolvable bool, cfg *config.Config) func(t *testing.T) {
 	return func(t *testing.T) {
 		target := &apicontainer.Container{
@@ -1153,6 +1366,9 @@ func assertContainerOrderingCanResolve(f func(target *apicontainer.Container, de
 func assertContainerOrderingResolved(f func(target *apicontainer.Container, dep *apicontainer.Container, depCond string, cfg *config.Config) bool, targetDesired, depKnown apicontainerstatus.ContainerStatus, depCond string, exitcode int, expectedResolved bool, cfg *config.Config) func(t *testing.T) {
 	return func(t *testing.T) {
 		target := &apicontainer.Container{
+			// Transition to MANIFEST_PULLED is always allowed, so set current known state to
+			// MANIFEST_PULLED for testing ordering
+			KnownStatusUnsafe:   apicontainerstatus.ContainerManifestPulled,
 			DesiredStatusUnsafe: targetDesired,
 		}
 		dep := &apicontainer.Container{
@@ -1175,6 +1391,7 @@ func TestContainerOrderingHealthyConditionIsResolved(t *testing.T) {
 		Resolved                      bool
 	}{
 		{
+			TargetKnown:                 apicontainerstatus.ContainerManifestPulled,
 			TargetDesired:               apicontainerstatus.ContainerCreated,
 			DependencyKnownHealthStatus: apicontainerstatus.ContainerHealthy,
 			HealthCheckType:             "docker",
@@ -1182,6 +1399,7 @@ func TestContainerOrderingHealthyConditionIsResolved(t *testing.T) {
 			Resolved:                    true,
 		},
 		{
+			TargetKnown:                   apicontainerstatus.ContainerManifestPulled,
 			TargetDesired:                 apicontainerstatus.ContainerCreated,
 			DependencyKnownHealthStatus:   apicontainerstatus.ContainerUnhealthy,
 			HealthCheckType:               "docker",
@@ -1190,6 +1408,7 @@ func TestContainerOrderingHealthyConditionIsResolved(t *testing.T) {
 			Resolved:                      false,
 		},
 		{
+			TargetKnown:   apicontainerstatus.ContainerManifestPulled,
 			TargetDesired: apicontainerstatus.ContainerCreated,
 			Resolved:      false,
 		},
diff --git a/agent/engine/dependencygraph/graph_unix_test.go b/agent/engine/dependencygraph/graph_unix_test.go
index 474d442d0a1..bc92d2857f3 100644
--- a/agent/engine/dependencygraph/graph_unix_test.go
+++ b/agent/engine/dependencygraph/graph_unix_test.go
@@ -44,6 +44,14 @@ func TestVerifyCgroupDependenciesResolved(t *testing.T) {
 			DependencyKnown: resourcestatus.ResourceStatus(cgroup.CgroupStatusNone),
 			RequiredStatus:  resourcestatus.ResourceStatus(cgroup.CgroupCreated),
 
+			ExpectedResolved: true,
+		},
+		{
+			Name:             "resource none,container pull depends on resource created",
+			TargetKnown:      apicontainerstatus.ContainerManifestPulled,
+			TargetDep:        apicontainerstatus.ContainerPulled,
+			DependencyKnown:  resourcestatus.ResourceStatus(cgroup.CgroupStatusNone),
+			RequiredStatus:   resourcestatus.ResourceStatus(cgroup.CgroupCreated),
 			ExpectedResolved: false,
 		},
 		{
diff --git a/agent/engine/docker_task_engine.go b/agent/engine/docker_task_engine.go
index d2c9ecf6d92..3fabe006772 100644
--- a/agent/engine/docker_task_engine.go
+++ b/agent/engine/docker_task_engine.go
@@ -289,6 +289,7 @@ func (engine *DockerTaskEngine) reconcileHostResources() {
 
 func (engine *DockerTaskEngine) initializeContainerStatusToTransitionFunction() {
 	containerStatusToTransitionFunction := map[apicontainerstatus.ContainerStatus]transitionApplyFunc{
+		apicontainerstatus.ContainerManifestPulled:       engine.pullContainerManifest,
 		apicontainerstatus.ContainerPulled:               engine.pullContainer,
 		apicontainerstatus.ContainerCreated:              engine.createContainer,
 		apicontainerstatus.ContainerRunning:              engine.startContainer,
@@ -1231,6 +1232,19 @@ func (engine *DockerTaskEngine) GetDaemonManagers() map[string]dm.DaemonManager
 	return engine.daemonManagers
 }
 
+// Pulls the manifest of the container image.
+func (engine *DockerTaskEngine) pullContainerManifest(
+	task *apitask.Task, container *apicontainer.Container,
+) dockerapi.DockerContainerMetadata {
+	// Currently a no-op
+	logger.Debug("Manifest pull is currently a no-op", logger.Fields{
+		field.TaskID:    task.GetID(),
+		field.Container: container.Name,
+		field.Image:     container.Image,
+	})
+	return dockerapi.DockerContainerMetadata{}
+}
+
 func (engine *DockerTaskEngine) pullContainer(task *apitask.Task, container *apicontainer.Container) dockerapi.DockerContainerMetadata {
 	switch container.Type {
 	case apicontainer.ContainerCNIPause, apicontainer.ContainerNamespacePause, apicontainer.ContainerServiceConnectRelay, apicontainer.ContainerManagedDaemon:
diff --git a/agent/engine/task_manager.go b/agent/engine/task_manager.go
index 07b7cf4ac43..9f4703c6b5a 100644
--- a/agent/engine/task_manager.go
+++ b/agent/engine/task_manager.go
@@ -620,8 +620,9 @@ func (mtask *managedTask) emitTaskEvent(task *apitask.Task, reason string) {
 	}
 	if !taskKnownStatus.BackendRecognized() {
 		logger.Debug("Skipping event emission for task", logger.Fields{
-			field.TaskID: mtask.GetID(),
-			field.Error:  "status not recognized by ECS",
+			field.TaskID:      mtask.GetID(),
+			field.Error:       "status not recognized by ECS",
+			field.KnownStatus: taskKnownStatus.String(),
 		})
 		return
 	}
@@ -862,18 +863,19 @@ func (mtask *managedTask) handleEventError(containerChange dockerContainerChange
 	switch event.Status {
 	// event.Status is the desired container transition from container's known status
 	// (* -> event.Status)
-	case apicontainerstatus.ContainerPulled:
+	case apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerPulled:
 		// If the agent pull behavior is always or once, we receive the error because
-		// the image pull fails, the task should fail. If we don't fail task here,
+		// the image or manifest pull fails, the task should fail. If we don't fail task here,
 		// then the cached image will probably be used for creating container, and we
 		// don't want to use cached image for both cases.
 		if mtask.cfg.ImagePullBehavior == config.ImagePullAlwaysBehavior ||
 			mtask.cfg.ImagePullBehavior == config.ImagePullOnceBehavior {
-			logger.Error("Error while pulling image; moving task to STOPPED", logger.Fields{
+			logger.Error("Error while pulling image or its manifest; moving task to STOPPED", logger.Fields{
 				field.TaskID:    mtask.GetID(),
 				field.Image:     container.Image,
 				field.Container: container.Name,
 				field.Error:     event.Error,
+				field.Status:    event.Status,
 			})
 			// The task should be stopped regardless of whether this container is
 			// essential or non-essential.
@@ -881,15 +883,16 @@ func (mtask *managedTask) handleEventError(containerChange dockerContainerChange
 			return false
 		}
 		// If the agent pull behavior is prefer-cached, we receive the error because
-		// the image pull fails and there is no cached image in local, we don't make
+		// the image or manifest pull fails and there is no cached image in local, we don't make
 		// the task fail here, will let create container handle it instead.
 		// If the agent pull behavior is default, use local image cache directly,
 		// assuming it exists.
-		logger.Error("Error while pulling image; will try to run anyway", logger.Fields{
+		logger.Error("Error while pulling image or its manifest; will try to run anyway", logger.Fields{
 			field.TaskID:    mtask.GetID(),
 			field.Image:     container.Image,
 			field.Container: container.Name,
 			field.Error:     event.Error,
+			field.Status:    event.Status,
 		})
 		// proceed anyway
 		return true
diff --git a/agent/engine/task_manager_test.go b/agent/engine/task_manager_test.go
index bd538a4ea24..36a81fd71c9 100644
--- a/agent/engine/task_manager_test.go
+++ b/agent/engine/task_manager_test.go
@@ -105,6 +105,36 @@ func TestHandleEventError(t *testing.T) {
 			ExpectedContainerDesiredStatusStopped: true,
 			ExpectedOK:                            true,
 		},
+		{
+			Name:              "Manifest pull failed - default pull behavior",
+			EventStatus:       apicontainerstatus.ContainerManifestPulled,
+			ImagePullBehavior: config.ImagePullDefaultBehavior,
+			Error:             &dockerapi.DockerTimeoutError{},
+			ExpectedOK:        true,
+		},
+		{
+			Name:                             "Manifest pull failed - always pull behavior",
+			EventStatus:                      apicontainerstatus.ContainerManifestPulled,
+			ImagePullBehavior:                config.ImagePullAlwaysBehavior,
+			Error:                            &dockerapi.DockerTimeoutError{},
+			ExpectedTaskDesiredStatusStopped: true,
+			ExpectedOK:                       false,
+		},
+		{
+			Name:                             "Manifest pull failed - once pull behavior",
+			EventStatus:                      apicontainerstatus.ContainerManifestPulled,
+			ImagePullBehavior:                config.ImagePullOnceBehavior,
+			Error:                            &dockerapi.DockerTimeoutError{},
+			ExpectedTaskDesiredStatusStopped: true,
+			ExpectedOK:                       false,
+		},
+		{
+			Name:              "Manifest pull failed - prefer-cached pull behavior",
+			EventStatus:       apicontainerstatus.ContainerManifestPulled,
+			ImagePullBehavior: config.ImagePullPreferCachedBehavior,
+			Error:             &dockerapi.DockerTimeoutError{},
+			ExpectedOK:        true,
+		},
 		{
 			Name:                            "Pull failed",
 			Error:                           &dockerapi.DockerTimeoutError{},
@@ -249,16 +279,26 @@ func TestContainerNextState(t *testing.T) {
 	}{
 		// NONE -> RUNNING transition is allowed and actionable, when desired is Running
 		// The expected next status is Pulled
-		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerPulled, true, nil},
+		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerManifestPulled, true, nil},
 		// NONE -> RESOURCES_PROVISIONED transition is allowed and actionable, when desired
 		// is Running. The exptected next status is Pulled
-		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerResourcesProvisioned, apicontainerstatus.ContainerPulled, true, nil},
+		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerResourcesProvisioned, apicontainerstatus.ContainerManifestPulled, true, nil},
 		// NONE -> NONE transition is not be allowed and is not actionable,
 		// when desired is Running
 		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerStatusNone, false, dependencygraph.ContainerPastDesiredStatusErr},
 		// NONE -> STOPPED transition will result in STOPPED and is allowed, but not
 		// actionable, when desired is STOPPED
 		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerStopped, apicontainerstatus.ContainerStopped, false, nil},
+		// MANIFEST_PULLED -> PULLED transition is allowed and actionable, when desired is Running
+		{apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerPulled, true, nil},
+		// MANIFEST_PULLED -> PULLED transition is allowed and actionable, when desired is RESOURCES_PROVISIONED
+		{apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerResourcesProvisioned, apicontainerstatus.ContainerPulled, true, nil},
+		// MANIFEST_PULLED -> MANIFEST_PULLED transition is not allowed and not actionable
+		{apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerStatusNone, false, dependencygraph.ContainerPastDesiredStatusErr},
+		// MANIFEST_PULLED -> NONE transition is not allowed and not actionable
+		{apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerStatusNone, false, dependencygraph.ContainerPastDesiredStatusErr},
+		// MANIFEST_PULLED -> STOPPED transition will result in STOPPED and is allowed, but not actionable
+		{apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerStopped, apicontainerstatus.ContainerStopped, false, nil},
 		// PULLED -> RUNNING transition is allowed and actionable, when desired is Running
 		// The exptected next status is Created
 		{apicontainerstatus.ContainerPulled, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerCreated, true, nil},
@@ -376,12 +416,12 @@ func TestContainerNextStateWithTransitionDependencies(t *testing.T) {
 		expectedTransitionActionable bool
 		reason                       error
 	}{
-		// NONE -> RUNNING transition is not allowed and not actionable, when pull depends on create and dependency is None
+		// NONE -> RUNNING transition is not allowed and not actionable, when manifest_pull depends on create and dependency is None
 		{
-			name:                         "pull depends on created, dependency is none",
+			name:                         "manifest_pull depends on created, dependency is none",
 			containerCurrentStatus:       apicontainerstatus.ContainerStatusNone,
 			containerDesiredStatus:       apicontainerstatus.ContainerRunning,
-			containerDependentStatus:     apicontainerstatus.ContainerPulled,
+			containerDependentStatus:     apicontainerstatus.ContainerManifestPulled,
 			dependencyCurrentStatus:      apicontainerstatus.ContainerStatusNone,
 			dependencySatisfiedStatus:    apicontainerstatus.ContainerCreated,
 			expectedContainerStatus:      apicontainerstatus.ContainerStatusNone,
@@ -390,9 +430,33 @@ func TestContainerNextStateWithTransitionDependencies(t *testing.T) {
 		},
 		// NONE -> RUNNING transition is not allowed and not actionable, when desired is Running and dependency is Created
 		{
-			name:                         "pull depends on running, dependency is created",
+			name:                         "manifest_pull depends on running, dependency is created",
 			containerCurrentStatus:       apicontainerstatus.ContainerStatusNone,
 			containerDesiredStatus:       apicontainerstatus.ContainerRunning,
+			containerDependentStatus:     apicontainerstatus.ContainerManifestPulled,
+			dependencyCurrentStatus:      apicontainerstatus.ContainerCreated,
+			dependencySatisfiedStatus:    apicontainerstatus.ContainerRunning,
+			expectedContainerStatus:      apicontainerstatus.ContainerStatusNone,
+			expectedTransitionActionable: false,
+			reason:                       dependencygraph.ErrContainerDependencyNotResolved,
+		},
+		// NONE -> RUNNING transition is not allowed and not actionable, when pull depends on create and dependency is None
+		{
+			name:                         "pull depends on created, current is manifest_pulled, dependency is none",
+			containerCurrentStatus:       apicontainerstatus.ContainerManifestPulled,
+			containerDesiredStatus:       apicontainerstatus.ContainerRunning,
+			containerDependentStatus:     apicontainerstatus.ContainerPulled,
+			dependencyCurrentStatus:      apicontainerstatus.ContainerStatusNone,
+			dependencySatisfiedStatus:    apicontainerstatus.ContainerCreated,
+			expectedContainerStatus:      apicontainerstatus.ContainerStatusNone,
+			expectedTransitionActionable: false,
+			reason:                       dependencygraph.ErrContainerDependencyNotResolved,
+		},
+		// NONE -> RUNNING transition is not allowed and not actionable, when desired is Running and dependency is Created
+		{
+			name:                         "pull depends on running, current is manifest_pulled, dependency is created",
+			containerCurrentStatus:       apicontainerstatus.ContainerManifestPulled,
+			containerDesiredStatus:       apicontainerstatus.ContainerRunning,
 			containerDependentStatus:     apicontainerstatus.ContainerPulled,
 			dependencyCurrentStatus:      apicontainerstatus.ContainerCreated,
 			dependencySatisfiedStatus:    apicontainerstatus.ContainerRunning,
@@ -404,7 +468,7 @@ func TestContainerNextStateWithTransitionDependencies(t *testing.T) {
 		// The expected next status is Pulled
 		{
 			name:                         "pull depends on running, dependency is running, next status is pulled",
-			containerCurrentStatus:       apicontainerstatus.ContainerStatusNone,
+			containerCurrentStatus:       apicontainerstatus.ContainerManifestPulled,
 			containerDesiredStatus:       apicontainerstatus.ContainerRunning,
 			containerDependentStatus:     apicontainerstatus.ContainerPulled,
 			dependencyCurrentStatus:      apicontainerstatus.ContainerRunning,
@@ -416,7 +480,7 @@ func TestContainerNextStateWithTransitionDependencies(t *testing.T) {
 		// The expected next status is Pulled
 		{
 			name:                         "pull depends on running, dependency is stopped, next status is pulled",
-			containerCurrentStatus:       apicontainerstatus.ContainerStatusNone,
+			containerCurrentStatus:       apicontainerstatus.ContainerManifestPulled,
 			containerDesiredStatus:       apicontainerstatus.ContainerRunning,
 			containerDependentStatus:     apicontainerstatus.ContainerPulled,
 			dependencyCurrentStatus:      apicontainerstatus.ContainerStopped,
@@ -427,13 +491,13 @@ func TestContainerNextStateWithTransitionDependencies(t *testing.T) {
 		// NONE -> RUNNING transition is allowed and actionable, when desired is Running and dependency is None and
 		// dependent status is Running
 		{
-			name:                         "create depends on running, dependency is none, next status is pulled",
+			name:                         "create depends on running, dependency is none, next status is manifest_pulled",
 			containerCurrentStatus:       apicontainerstatus.ContainerStatusNone,
 			containerDesiredStatus:       apicontainerstatus.ContainerRunning,
 			containerDependentStatus:     apicontainerstatus.ContainerCreated,
 			dependencyCurrentStatus:      apicontainerstatus.ContainerStatusNone,
 			dependencySatisfiedStatus:    apicontainerstatus.ContainerRunning,
-			expectedContainerStatus:      apicontainerstatus.ContainerPulled,
+			expectedContainerStatus:      apicontainerstatus.ContainerManifestPulled,
 			expectedTransitionActionable: true,
 		},
 	}
@@ -485,11 +549,11 @@ func TestContainerNextStateWithDependencies(t *testing.T) {
 		// NONE -> RUNNING transition is not allowed and not actionable, when desired is Running and dependency is Created
 		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerCreated, apicontainerstatus.ContainerStatusNone, false, dependencygraph.DependentContainerNotResolvedErr},
 		// NONE -> RUNNING transition is allowed and actionable, when desired is Running and dependency is Running
-		// The expected next status is Pulled
-		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerPulled, true, nil},
+		// The expected next status is MANIFEST_PULLED
+		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerManifestPulled, true, nil},
 		// NONE -> RUNNING transition is allowed and actionable, when desired is Running and dependency is Stopped
-		// The expected next status is Pulled
-		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerStopped, apicontainerstatus.ContainerPulled, true, nil},
+		// The expected next status is MANIFEST_PULLED
+		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerStopped, apicontainerstatus.ContainerManifestPulled, true, nil},
 	}
 
 	for _, tc := range testCases {
@@ -538,13 +602,15 @@ func TestContainerNextStateWithPullCredentials(t *testing.T) {
 		// NONE -> RUNNING transition is not allowed when container is waiting for credentials
 		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerStatusNone, "not_existed", true, false, dependencygraph.CredentialsNotResolvedErr},
 		// NONE -> RUNNING transition is allowed when the required execution credentials existed
-		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerPulled, "existed", true, true, nil},
+		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerManifestPulled, "existed", true, true, nil},
+		// MANIFEST_PULLED -> RUNNING transition is not allowed when the credentials don't exist
+		{apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerStatusNone, "not_existed", true, false, dependencygraph.CredentialsNotResolvedErr},
 		// PULLED -> RUNNING transition is allowed even the credentials is required
 		{apicontainerstatus.ContainerPulled, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerCreated, "not_existed", true, true, nil},
 		// NONE -> STOPPED transition is allowed even the credentials is required
 		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerStopped, apicontainerstatus.ContainerStopped, "not_existed", true, false, nil},
 		// NONE -> RUNNING transition is allowed when the container doesn't use execution credentials
-		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerPulled, "not_existed", false, true, nil},
+		{apicontainerstatus.ContainerStatusNone, apicontainerstatus.ContainerRunning, apicontainerstatus.ContainerManifestPulled, "not_existed", false, true, nil},
 	}
 
 	taskEngine := &DockerTaskEngine{
diff --git a/agent/taskresource/asmauth/asmauth_test.go b/agent/taskresource/asmauth/asmauth_test.go
index 78a29838d9c..e9d18af056a 100644
--- a/agent/taskresource/asmauth/asmauth_test.go
+++ b/agent/taskresource/asmauth/asmauth_test.go
@@ -18,6 +18,7 @@ package asmauth
 
 import (
 	"encoding/json"
+	"fmt"
 	"testing"
 	"time"
 
@@ -126,22 +127,42 @@ func TestMarshalUnmarshalJSON(t *testing.T) {
 }
 
 func TestInitialize(t *testing.T) {
-	ctrl := gomock.NewController(t)
-	defer ctrl.Finish()
-
-	credentialsManager := mock_credentials.NewMockManager(ctrl)
-	asmClientCreator := mock_factory.NewMockClientCreator(ctrl)
-	asmRes := &ASMAuthResource{
-		knownStatusUnsafe:   resourcestatus.ResourceCreated,
-		desiredStatusUnsafe: resourcestatus.ResourceCreated,
+	tcs := []struct {
+		taskKnownStatus   apitaskstatus.TaskStatus
+		taskDesiredStatus apitaskstatus.TaskStatus
+		expectReset       bool
+	}{
+		{apitaskstatus.TaskStatusNone, apitaskstatus.TaskRunning, true},
+		{apitaskstatus.TaskManifestPulled, apitaskstatus.TaskRunning, true},
+		{apitaskstatus.TaskPulled, apitaskstatus.TaskRunning, false},
+		{apitaskstatus.TaskStatusNone, apitaskstatus.TaskStopped, false},
+		{apitaskstatus.TaskManifestPulled, apitaskstatus.TaskStopped, false},
+	}
+	for _, tc := range tcs {
+		t.Run(
+			fmt.Sprintf("Known: %s; Desired: %s", tc.taskKnownStatus, tc.taskDesiredStatus),
+			func(t *testing.T) {
+				ctrl := gomock.NewController(t)
+				defer ctrl.Finish()
+
+				credentialsManager := mock_credentials.NewMockManager(ctrl)
+				asmClientCreator := mock_factory.NewMockClientCreator(ctrl)
+				asmRes := &ASMAuthResource{
+					knownStatusUnsafe:   resourcestatus.ResourceCreated,
+					desiredStatusUnsafe: resourcestatus.ResourceCreated,
+				}
+				asmRes.Initialize(&taskresource.ResourceFields{
+					ResourceFieldsCommon: &taskresource.ResourceFieldsCommon{
+						ASMClientCreator:   asmClientCreator,
+						CredentialsManager: credentialsManager,
+					},
+				}, tc.taskKnownStatus, tc.taskDesiredStatus)
+				if tc.expectReset {
+					assert.Equal(t, resourcestatus.ResourceStatusNone, asmRes.GetKnownStatus())
+				} else {
+					assert.Equal(t, resourcestatus.ResourceCreated, asmRes.GetKnownStatus())
+				}
+				assert.Equal(t, resourcestatus.ResourceCreated, asmRes.GetDesiredStatus())
+			})
 	}
-	asmRes.Initialize(&taskresource.ResourceFields{
-		ResourceFieldsCommon: &taskresource.ResourceFieldsCommon{
-			ASMClientCreator:   asmClientCreator,
-			CredentialsManager: credentialsManager,
-		},
-	}, apitaskstatus.TaskStatusNone, apitaskstatus.TaskRunning)
-	assert.Equal(t, resourcestatus.ResourceStatusNone, asmRes.GetKnownStatus())
-	assert.Equal(t, resourcestatus.ResourceCreated, asmRes.GetDesiredStatus())
-
 }
diff --git a/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/container/status/containerstatus.go b/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/container/status/containerstatus.go
index b832b39b704..b1b1ed85779 100644
--- a/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/container/status/containerstatus.go
+++ b/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/container/status/containerstatus.go
@@ -22,6 +22,8 @@ import (
 const (
 	// ContainerStatusNone is the zero state of a container; this container has not completed pull
 	ContainerStatusNone ContainerStatus = iota
+	// ContainerManifestPulled represents a container which has had its image manifest pulled
+	ContainerManifestPulled
 	// ContainerPulled represents a container which has had the image pulled
 	ContainerPulled
 	// ContainerCreated represents a container that has been created
@@ -58,6 +60,7 @@ type ContainerHealthStatus int32
 
 var containerStatusMap = map[string]ContainerStatus{
 	"NONE":                  ContainerStatusNone,
+	"MANIFEST_PULLED":       ContainerManifestPulled,
 	"PULLED":                ContainerPulled,
 	"CREATED":               ContainerCreated,
 	"RUNNING":               ContainerRunning,
diff --git a/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/statusmapping.go b/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/statusmapping.go
index 2c8b95f2147..202ce5c7e31 100644
--- a/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/statusmapping.go
+++ b/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/statusmapping.go
@@ -20,15 +20,17 @@ import (
 // MapContainerToTaskStatus maps the container status to the corresponding task status. The
 // transition map is illustrated below.
 //
-// Container: None -> Pulled -> Created -> Running -> Provisioned -> Stopped -> Zombie
+// Container: None -> ManifestPulled -> Pulled -> Created -> Running -> Provisioned -> Stopped -> Zombie
 //
-// Task     : None ->     Created       ->         Running        -> Stopped
+// Task     : None -> ManifestPulled ->           Created -> Running ->                Stopped
 func MapContainerToTaskStatus(knownState apicontainerstatus.ContainerStatus, steadyState apicontainerstatus.ContainerStatus) TaskStatus {
 	switch knownState {
 	case apicontainerstatus.ContainerStatusNone:
 		return TaskStatusNone
 	case steadyState:
 		return TaskRunning
+	case apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerPulled:
+		return TaskManifestPulled
 	case apicontainerstatus.ContainerCreated:
 		return TaskCreated
 	case apicontainerstatus.ContainerStopped:
diff --git a/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/taskstatus.go b/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/taskstatus.go
index 04f44ed30fa..6037a335606 100644
--- a/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/taskstatus.go
+++ b/agent/vendor/github.com/aws/amazon-ecs-agent/ecs-agent/api/task/status/taskstatus.go
@@ -21,6 +21,8 @@ import (
 const (
 	// TaskStatusNone is the zero state of a task; this task has been received but no further progress has completed
 	TaskStatusNone TaskStatus = iota
+	// TaskManifestPulled represents a task which has had all its container image manifests pulled
+	TaskManifestPulled
 	// TaskPulled represents a task which has had all its container images pulled, but not all have yet progressed passed pull
 	TaskPulled
 	// TaskCreated represents a task which has had all its containers created
@@ -37,6 +39,8 @@ const (
 	TaskRunningString = "RUNNING"
 	//TaskCreatedString represents task created status string
 	TaskCreatedString = "CREATED"
+	// TaskManifestPulledString represents task manifest_pulled status string
+	TaskManifestPulledString = "MANIFEST_PULLED"
 	// TaskNoneString represents task none status string
 	TaskNoneString = "NONE"
 )
@@ -45,10 +49,11 @@ const (
 type TaskStatus int32
 
 var taskStatusMap = map[string]TaskStatus{
-	TaskNoneString:    TaskStatusNone,
-	TaskCreatedString: TaskCreated,
-	TaskRunningString: TaskRunning,
-	TaskStoppedString: TaskStopped,
+	TaskNoneString:           TaskStatusNone,
+	TaskManifestPulledString: TaskManifestPulled,
+	TaskCreatedString:        TaskCreated,
+	TaskRunningString:        TaskRunning,
+	TaskStoppedString:        TaskStopped,
 }
 
 // String returns a human readable string representation of this object
diff --git a/ecs-agent/api/container/status/containerstatus.go b/ecs-agent/api/container/status/containerstatus.go
index b832b39b704..b1b1ed85779 100644
--- a/ecs-agent/api/container/status/containerstatus.go
+++ b/ecs-agent/api/container/status/containerstatus.go
@@ -22,6 +22,8 @@ import (
 const (
 	// ContainerStatusNone is the zero state of a container; this container has not completed pull
 	ContainerStatusNone ContainerStatus = iota
+	// ContainerManifestPulled represents a container which has had its image manifest pulled
+	ContainerManifestPulled
 	// ContainerPulled represents a container which has had the image pulled
 	ContainerPulled
 	// ContainerCreated represents a container that has been created
@@ -58,6 +60,7 @@ type ContainerHealthStatus int32
 
 var containerStatusMap = map[string]ContainerStatus{
 	"NONE":                  ContainerStatusNone,
+	"MANIFEST_PULLED":       ContainerManifestPulled,
 	"PULLED":                ContainerPulled,
 	"CREATED":               ContainerCreated,
 	"RUNNING":               ContainerRunning,
diff --git a/ecs-agent/api/container/status/containerstatus_test.go b/ecs-agent/api/container/status/containerstatus_test.go
index fe0485a3243..8fbdee94385 100644
--- a/ecs-agent/api/container/status/containerstatus_test.go
+++ b/ecs-agent/api/container/status/containerstatus_test.go
@@ -31,6 +31,11 @@ func TestShouldReportToBackend(t *testing.T) {
 	assert.False(t, containerStatus.ShouldReportToBackend(ContainerRunning))
 	assert.False(t, containerStatus.ShouldReportToBackend(ContainerResourcesProvisioned))
 
+	// ContainerManifestPulled is not reported to backend
+	containerStatus = ContainerManifestPulled
+	assert.False(t, containerStatus.ShouldReportToBackend(ContainerRunning))
+	assert.False(t, containerStatus.ShouldReportToBackend(ContainerResourcesProvisioned))
+
 	// ContainerPulled is not reported to backend
 	containerStatus = ContainerPulled
 	assert.False(t, containerStatus.ShouldReportToBackend(ContainerRunning))
@@ -65,6 +70,11 @@ func TestBackendStatus(t *testing.T) {
 	assert.Equal(t, containerStatus.BackendStatus(ContainerRunning), ContainerStatusNone)
 	assert.Equal(t, containerStatus.BackendStatus(ContainerResourcesProvisioned), ContainerStatusNone)
 
+	// BackendStatus is still ContainerStatusNone when container status is ContainerManifestPulled
+	containerStatus = ContainerManifestPulled
+	assert.Equal(t, containerStatus.BackendStatus(ContainerRunning), ContainerStatusNone)
+	assert.Equal(t, containerStatus.BackendStatus(ContainerResourcesProvisioned), ContainerStatusNone)
+
 	// BackendStatus is still ContainerStatusNone when container status is ContainerPulled
 	containerStatus = ContainerPulled
 	assert.Equal(t, containerStatus.BackendStatus(ContainerRunning), ContainerStatusNone)
@@ -98,19 +108,55 @@ type testContainerStatus struct {
 	SomeStatus ContainerStatus `json:"status"`
 }
 
-func TestUnmarshalContainerStatus(t *testing.T) {
-	status := ContainerStatusNone
+// Tests that pointers to a struct that contains ContainerStatus are marshaled correctly.
+func TestMarshalContainerStatusNestedPointer(t *testing.T) {
+	tcs := []struct {
+		status   ContainerStatus
+		expected string
+	}{
+		{status: ContainerStatusNone, expected: "NONE"},
+		{status: ContainerManifestPulled, expected: "MANIFEST_PULLED"},
+		{status: ContainerPulled, expected: "PULLED"},
+		{status: ContainerCreated, expected: "CREATED"},
+		{status: ContainerRunning, expected: "RUNNING"},
+		{status: ContainerResourcesProvisioned, expected: "RESOURCES_PROVISIONED"},
+		{status: ContainerStopped, expected: "STOPPED"},
+	}
+	for _, tc := range tcs {
+		marshaled, err := json.Marshal(&testContainerStatus{SomeStatus: tc.status})
+		require.NoError(t, err)
+		assert.Equal(t, fmt.Sprintf(`{"status":%q}`, tc.status), string(marshaled))
+	}
+}
 
-	err := json.Unmarshal([]byte(`"RUNNING"`), &status)
-	if err != nil {
-		t.Error(err)
+// Tests that ContainerStatus is unmarshaled correctly.
+func TestUnmarshalContainerStatusSimple(t *testing.T) {
+	tcs := []struct {
+		statusString string
+		expected     ContainerStatus
+	}{
+		{statusString: "NONE", expected: ContainerStatusNone},
+		{statusString: "MANIFEST_PULLED", expected: ContainerManifestPulled},
+		{statusString: "PULLED", expected: ContainerPulled},
+		{statusString: "CREATED", expected: ContainerCreated},
+		{statusString: "RUNNING", expected: ContainerRunning},
+		{statusString: "RESOURCES_PROVISIONED", expected: ContainerResourcesProvisioned},
+		{statusString: "STOPPED", expected: ContainerStopped},
 	}
-	if status != ContainerRunning {
-		t.Error("RUNNING should unmarshal to RUNNING, not " + status.String())
+	for _, tc := range tcs {
+		t.Run(tc.statusString, func(t *testing.T) {
+			status := ContainerStatusNone
+			err := json.Unmarshal([]byte(fmt.Sprintf("%q", tc.statusString)), &status)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expected, status)
+		})
 	}
+}
 
+// Tests that structs that have a ContainerStatus field are unmarshaled correctly.
+func TestUnmarshalContainerStatusNested(t *testing.T) {
 	var test testContainerStatus
-	err = json.Unmarshal([]byte(`{"status":"STOPPED"}`), &test)
+	err := json.Unmarshal([]byte(`{"status":"STOPPED"}`), &test)
 	if err != nil {
 		t.Error(err)
 	}
@@ -276,10 +322,3 @@ func TestContainerStatusJSONUnmarshalInt(t *testing.T) {
 		})
 	}
 }
-
-func TestTemporary(t *testing.T) {
-	marshaled := `{"1": "ok"}`
-	unmarshaled := map[ContainerStatus]string{}
-	err := json.Unmarshal([]byte(marshaled), &unmarshaled)
-	require.NoError(t, err)
-}
diff --git a/ecs-agent/api/task/status/statusmapping.go b/ecs-agent/api/task/status/statusmapping.go
index 2c8b95f2147..202ce5c7e31 100644
--- a/ecs-agent/api/task/status/statusmapping.go
+++ b/ecs-agent/api/task/status/statusmapping.go
@@ -20,15 +20,17 @@ import (
 // MapContainerToTaskStatus maps the container status to the corresponding task status. The
 // transition map is illustrated below.
 //
-// Container: None -> Pulled -> Created -> Running -> Provisioned -> Stopped -> Zombie
+// Container: None -> ManifestPulled -> Pulled -> Created -> Running -> Provisioned -> Stopped -> Zombie
 //
-// Task     : None ->     Created       ->         Running        -> Stopped
+// Task     : None -> ManifestPulled ->           Created -> Running ->                Stopped
 func MapContainerToTaskStatus(knownState apicontainerstatus.ContainerStatus, steadyState apicontainerstatus.ContainerStatus) TaskStatus {
 	switch knownState {
 	case apicontainerstatus.ContainerStatusNone:
 		return TaskStatusNone
 	case steadyState:
 		return TaskRunning
+	case apicontainerstatus.ContainerManifestPulled, apicontainerstatus.ContainerPulled:
+		return TaskManifestPulled
 	case apicontainerstatus.ContainerCreated:
 		return TaskCreated
 	case apicontainerstatus.ContainerStopped:
diff --git a/ecs-agent/api/task/status/statusmapping_test.go b/ecs-agent/api/task/status/statusmapping_test.go
index 77d06314338..2e1d8328e3f 100644
--- a/ecs-agent/api/task/status/statusmapping_test.go
+++ b/ecs-agent/api/task/status/statusmapping_test.go
@@ -30,10 +30,19 @@ func TestTaskStatus(t *testing.T) {
 	assert.Equal(t, MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerRunning), TaskStatusNone)
 	assert.Equal(t, MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerResourcesProvisioned), TaskStatusNone)
 
-	// When container state is PULLED, Task state is still NONE
+	// When container state is MANIFEST_PULLED, Task state is MANIFEST_PULLED as well
+	containerStatus = apicontainerstatus.ContainerManifestPulled
+	assert.Equal(t,
+		MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerRunning),
+		TaskManifestPulled)
+	assert.Equal(t,
+		MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerResourcesProvisioned),
+		TaskManifestPulled)
+
+	// When container state is PULLED, Task state is still MANIFEST_PULLED
 	containerStatus = apicontainerstatus.ContainerPulled
-	assert.Equal(t, MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerRunning), TaskStatusNone)
-	assert.Equal(t, MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerResourcesProvisioned), TaskStatusNone)
+	assert.Equal(t, MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerRunning), TaskManifestPulled)
+	assert.Equal(t, MapContainerToTaskStatus(containerStatus, apicontainerstatus.ContainerResourcesProvisioned), TaskManifestPulled)
 
 	// When container state is CREATED, Task state is CREATED as well
 	containerStatus = apicontainerstatus.ContainerCreated
diff --git a/ecs-agent/api/task/status/taskstatus.go b/ecs-agent/api/task/status/taskstatus.go
index 04f44ed30fa..6037a335606 100644
--- a/ecs-agent/api/task/status/taskstatus.go
+++ b/ecs-agent/api/task/status/taskstatus.go
@@ -21,6 +21,8 @@ import (
 const (
 	// TaskStatusNone is the zero state of a task; this task has been received but no further progress has completed
 	TaskStatusNone TaskStatus = iota
+	// TaskManifestPulled represents a task which has had all its container image manifests pulled
+	TaskManifestPulled
 	// TaskPulled represents a task which has had all its container images pulled, but not all have yet progressed passed pull
 	TaskPulled
 	// TaskCreated represents a task which has had all its containers created
@@ -37,6 +39,8 @@ const (
 	TaskRunningString = "RUNNING"
 	//TaskCreatedString represents task created status string
 	TaskCreatedString = "CREATED"
+	// TaskManifestPulledString represents task manifest_pulled status string
+	TaskManifestPulledString = "MANIFEST_PULLED"
 	// TaskNoneString represents task none status string
 	TaskNoneString = "NONE"
 )
@@ -45,10 +49,11 @@ const (
 type TaskStatus int32
 
 var taskStatusMap = map[string]TaskStatus{
-	TaskNoneString:    TaskStatusNone,
-	TaskCreatedString: TaskCreated,
-	TaskRunningString: TaskRunning,
-	TaskStoppedString: TaskStopped,
+	TaskNoneString:           TaskStatusNone,
+	TaskManifestPulledString: TaskManifestPulled,
+	TaskCreatedString:        TaskCreated,
+	TaskRunningString:        TaskRunning,
+	TaskStoppedString:        TaskStopped,
 }
 
 // String returns a human readable string representation of this object
diff --git a/ecs-agent/api/task/status/taskstatus_test.go b/ecs-agent/api/task/status/taskstatus_test.go
index 4a9938581d6..0276979fb49 100644
--- a/ecs-agent/api/task/status/taskstatus_test.go
+++ b/ecs-agent/api/task/status/taskstatus_test.go
@@ -18,26 +18,61 @@ package status
 
 import (
 	"encoding/json"
+	"fmt"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 type testTaskStatus struct {
 	SomeStatus TaskStatus `json:"status"`
 }
 
-func TestUnmarshalTaskStatus(t *testing.T) {
-	status := TaskStatusNone
+func TestUnmarshalTaskStatusSimple(t *testing.T) {
+	tcs := []struct {
+		statusString string
+		expected     TaskStatus
+	}{
+		{"NONE", TaskStatusNone},
+		{"MANIFEST_PULLED", TaskManifestPulled},
+		{"CREATED", TaskCreated},
+		{"RUNNING", TaskRunning},
+		{"STOPPED", TaskStopped},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.statusString, func(t *testing.T) {
+			status := TaskStatusNone
+			err := json.Unmarshal([]byte(fmt.Sprintf("%q", tc.statusString)), &status)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expected, status)
+		})
+	}
+}
 
-	err := json.Unmarshal([]byte(`"RUNNING"`), &status)
-	if err != nil {
-		t.Error(err)
+func TestMarshaTaskStatusPointer(t *testing.T) {
+	tcs := []struct {
+		status   TaskStatus
+		expected string
+	}{
+		{status: TaskStatusNone, expected: `"NONE"`},
+		{status: TaskManifestPulled, expected: `"MANIFEST_PULLED"`},
+		{status: TaskCreated, expected: `"CREATED"`},
+		{status: TaskRunning, expected: `"RUNNING"`},
+		{status: TaskStopped, expected: `"STOPPED"`},
 	}
-	if status != TaskRunning {
-		t.Error("RUNNING should unmarshal to RUNNING, not " + status.String())
+	for _, tc := range tcs {
+		t.Run(tc.expected, func(t *testing.T) {
+			marshled, err := json.Marshal(&tc.status)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expected, string(marshled))
+		})
 	}
+}
 
+func TestUnmarshalTaskStatusNested(t *testing.T) {
 	var test testTaskStatus
-	err = json.Unmarshal([]byte(`{"status":"STOPPED"}`), &test)
+	err := json.Unmarshal([]byte(`{"status":"STOPPED"}`), &test)
 	if err != nil {
 		t.Error(err)
 	}
@@ -45,3 +80,24 @@ func TestUnmarshalTaskStatus(t *testing.T) {
 		t.Error("STOPPED should unmarshal to STOPPED, not " + test.SomeStatus.String())
 	}
 }
+
+// Tests that a pointer to a struct that has a TaskStatus is marshaled correctly.
+func TestMarshalTaskStatusNestedPointer(t *testing.T) {
+	tcs := []struct {
+		status   TaskStatus
+		expected string
+	}{
+		{status: TaskStatusNone, expected: "NONE"},
+		{status: TaskManifestPulled, expected: "MANIFEST_PULLED"},
+		{status: TaskCreated, expected: "CREATED"},
+		{status: TaskRunning, expected: "RUNNING"},
+		{status: TaskStopped, expected: "STOPPED"},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.status.String(), func(t *testing.T) {
+			marshaled, err := json.Marshal(&testTaskStatus{tc.status})
+			require.NoError(t, err)
+			assert.Equal(t, fmt.Sprintf(`{"status":%q}`, tc.expected), string(marshaled))
+		})
+	}
+}
diff --git a/ecs-agent/daemonimages/csidriver/bin/ebs-csi-driver b/ecs-agent/daemonimages/csidriver/bin/ebs-csi-driver
new file mode 100755
index 00000000000..906758057ae
Binary files /dev/null and b/ecs-agent/daemonimages/csidriver/bin/ebs-csi-driver differ