--- a/taskcluster/docker/recipes/run-task
+++ b/taskcluster/docker/recipes/run-task
@@ -32,16 +32,30 @@ import urllib2
 
 FINGERPRINT_URL = 'http://taskcluster/secrets/v1/secret/project/taskcluster/gecko/hgfingerprint'
 FALLBACK_FINGERPRINT = {
     'fingerprints':
         "sha256:8e:ad:f7:6a:eb:44:06:15:ed:f3:e4:69:a6:64:60:37:2d:ff:98:88:37"
         ":bf:d7:b8:40:84:01:48:9c:26:ce:d9"}
 
 
+NON_EMPTY_VOLUME = '''
+error: volume %s is not empty
+
+Our Docker image policy requires volumes to be empty.
+
+The volume was likely populated as part of building the Docker image.
+Change the Dockerfile and anything run from it to not create files in
+any VOLUME.
+
+A lesser possibility is that you stumbled upon a TaskCluster platform bug
+where it fails to use new volumes for tasks.
+'''
+
+
 def print_line(prefix, m):
     now = datetime.datetime.utcnow()
     print(b'[%s %sZ] %s' % (prefix, now.isoformat(), m), end=b'')
 
 
 def run_and_prefix_output(prefix, args, extra_env=None):
     """Runs a process and prefixes its output with the time.
 
@@ -329,16 +343,52 @@ def main(args):
         # happen because run-task should be the first thing that touches a
         # cache.
         else:
             print('cache %s is not empty and is missing a .cacherequires '
                   'file; the cache names for this task are likely '
                   'mis-configured')
             return 1
 
+    if 'TASKCLUSTER_VOLUMES' in os.environ:
+        volumes = os.environ['TASKCLUSTER_VOLUMES'].split(';')
+        del os.environ['TASKCLUSTER_VOLUMES']
+    else:
+        volumes = []
+
+    # Sanitize volumes.
+    for volume in volumes:
+        # If a volume is a cache, it was dealt with above.
+        if volume in caches:
+            print_line(b'volume', b'volume %s is a cache\n' % volume)
+            continue
+
+        # The only time we should see files in the volume is if the Docker
+        # image build put files there.
+        #
+        # For the sake of simplicity, our policy is that volumes should be
+        # empty. This also has the advantage that an empty volume looks
+        # a lot like an empty cache. Tasks can rely on caches being
+        # swapped in and out on any volume without any noticeable change
+        # of behavior.
+        volume_files = os.listdir(volume)
+        if volume_files:
+            print(NON_EMPTY_VOLUME % volume)
+            print('entries in root directory: %s' %
+                  ' '.join(sorted(volume_files)))
+            return 1
+
+        # The volume is almost certainly owned by root:root. Chown it so it
+        # is writable.
+
+        if running_as_root:
+            print_line(b'volume', b'changing ownership of volume %s '
+                                  b'to %d:%d\n' % (volume, uid, gid))
+            set_dir_permissions(volume, uid, gid)
+
     # Change ownership of requested paths.
     # FUTURE: parse argument values for user/group if we don't want to
     # use --user/--group.
     for path in args.chown or []:
         if not running_as_root:
             print_line(b'set_dir_permissions', b'--chown not allowed when not running as root')
             return 1
 

--- a/taskcluster/taskgraph/transforms/task.py
+++ b/taskcluster/taskgraph/transforms/task.py
@@ -692,29 +692,29 @@ def build_docker_worker_payload(config, 
         for artifact in worker['artifacts']:
             artifacts[artifact['name']] = {
                 'path': artifact['path'],
                 'type': artifact['type'],
                 'expires': task_def['expires'],  # always expire with the task
             }
         payload['artifacts'] = artifacts
 
+    run_task = payload.get('command', [''])[0].endswith('run-task')
+
     if 'caches' in worker:
         caches = {}
 
         # run-task knows how to validate caches.
         #
         # To help ensure new run-task features and bug fixes don't interfere
         # with existing caches, we seed the hash of run-task into cache names.
         # So, any time run-task changes, we should get a fresh set of caches.
         # This means run-task can make changes to cache interaction at any time
         # without regards for backwards or future compatibility.
 
-        run_task = payload.get('command', [''])[0].endswith('run-task')
-
         if run_task:
             suffix = '-%s' % _run_task_suffix()
         else:
             suffix = ''
 
         skip_untrusted = config.params['project'] == 'try' or level == 1
 
         for cache in worker['caches']:
@@ -729,16 +729,21 @@ def build_docker_worker_payload(config, 
 
         # Assertion: only run-task is interested in this.
         if run_task:
             payload['env']['TASKCLUSTER_CACHES'] = ';'.join(sorted(
                 caches.values()))
 
         payload['cache'] = caches
 
+    # And send down volumes information to run-task as well.
+    if run_task and worker.get('volumes'):
+        payload['env']['TASKCLUSTER_VOLUMES'] = ';'.join(
+            sorted(worker['volumes']))
+
     if features:
         payload['features'] = features
     if capabilities:
         payload['capabilities'] = capabilities
 
     # coalesce / superseding
     if 'coalesce-name' in task and level > 1:
         key = COALESCE_KEY.format(